"""Holistic pull-request scope clustering.

This module answers a different question than duplicate-PR detection:

    "Which open PRs look similar in code scope and touched areas?"

The algorithm intentionally uses the whole open-PR queue at once instead of
only raw pairwise overlap:

1. Build a sparse "scope vector" for each open PR from:
   - exact filenames
   - directory prefixes
   - coarse file chunks derived from unified-diff hunk ranges
2. Weight every feature by inverse PR frequency so very common files and
   directories contribute less than rare, discriminative ones.
3. Blend content similarity with simple "shape" features:
   - total changed-line size
   - changed-file count
   - touched-directory count
   - concentration in the dominant leaf directory
4. Build a mutual k-nearest-neighbor graph from those scores.
5. Keep only strong or locally-supported edges so broad PRs do not bridge
   unrelated groups.
6. Split graph components greedily around local medoids to avoid weak chains.

The result is a conservative "similar scope" clustering intended for queue
analysis and review routing, not a strict duplicate classifier.
"""

from __future__ import annotations

import json
import re
from collections import Counter, defaultdict
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import UTC, datetime
from math import log, sqrt
from pathlib import Path, PurePosixPath
from typing import Any

from pydantic import BaseModel, Field

from slop_farmer.data.parquet_io import read_json, read_parquet_rows
from slop_farmer.data.snapshot_source import resolve_snapshot_source_dir
from slop_farmer.reports.pr_heuristics import (
    compile_cluster_suppression_rules,
    suppressed_pull_request_reasons,
)

HUNK_HEADER_PATTERN = re.compile(r"^@@ -\d+(?:,\d+)? \+(?P<start>\d+)(?:,(?P<count>\d+))? @@")
__all__ = [
    "PrScopeCluster",
    "PrScopeClusterOptions",
    "PrScopePair",
    "build_pr_scope_clusters",
    "run_pr_scope_report",
]


@dataclass(slots=True)
class PrScopeClusterOptions:
    """Tuning knobs for holistic PR scope clustering.

    The defaults aim to be conservative:
    - only open, non-draft PRs are clustered
    - exact files matter most
    - directory and chunk features let nearby-but-not-identical changes group
    - cluster edges need either high similarity or mutual-neighbor support
    """

    include_closed: bool = False
    include_drafts: bool = False
    chunk_size: int = 80
    max_neighbors: int = 5
    min_similarity: float = 0.30
    pair_similarity: float = 0.40
    strong_similarity: float = 0.55
    min_shared_neighbors: int = 1
    expansion_similarity: float = 0.34
    min_cluster_average_similarity: float = 0.32
    min_cluster_size: int = 2
    max_feature_df_ratio: float = 0.90
    file_weight: float = 1.0
    directory_weight: float = 0.60
    chunk_weight: float = 0.75
    content_weight: float = 0.70
    size_weight: float = 0.15
    breadth_weight: float = 0.10
    concentration_weight: float = 0.05
    max_shared_features: int = 8


class PrScopePair(BaseModel):
    left_pr_number: int
    right_pr_number: int
    similarity: float
    content_similarity: float
    size_similarity: float
    breadth_similarity: float
    concentration_similarity: float
    shared_filenames: list[str] = Field(default_factory=list)
    shared_directories: list[str] = Field(default_factory=list)


class PrScopeCluster(BaseModel):
    cluster_id: str
    pr_numbers: list[int]
    representative_pr_number: int
    average_similarity: float
    summary: str
    shared_filenames: list[str] = Field(default_factory=list)
    shared_directories: list[str] = Field(default_factory=list)
    pairwise: list[PrScopePair] = Field(default_factory=list)


@dataclass(slots=True)
class _ScopeProfile:
    number: int
    total_changed_lines: int
    file_count: int
    directory_count: int
    dominant_dir_share: float
    filenames: set[str]
    directories: set[str]
    raw_vector: dict[str, float]
    vector: dict[str, float]


@dataclass(slots=True)
class _ScopeComparison:
    left: int
    right: int
    similarity: float
    content_similarity: float
    size_similarity: float
    breadth_similarity: float
    concentration_similarity: float
    shared_filenames: list[str]
    shared_directories: list[str]


def build_pr_scope_clusters(
    pull_requests: Sequence[Mapping[str, Any]],
    pr_files: Sequence[Mapping[str, Any]],
    *,
    options: PrScopeClusterOptions | None = None,
    suppression_rules: Sequence[Mapping[str, Any]] = (),
) -> list[PrScopeCluster]:
    """Cluster open PRs by weighted file-scope similarity.

    This is intentionally holistic:
    - feature weights depend on the full open-PR set
    - similarity blends exact-file, directory, and chunk overlap
    - graph edges depend on each PR's neighborhood, not only raw pair scores
    """

    settings = options or PrScopeClusterOptions()
    suppressed_prs = suppressed_pull_request_reasons(
        pull_requests,
        pr_files,
        compile_cluster_suppression_rules(suppression_rules),
    )
    active_prs = [
        row
        for row in pull_requests
        if _include_pull_request(row, settings) and int(row["number"]) not in suppressed_prs
    ]
    if len(active_prs) < 2:
        return []

    files_by_pr: defaultdict[int, list[Mapping[str, Any]]] = defaultdict(list)
    active_numbers = {int(row["number"]) for row in active_prs if row.get("number") is not None}
    for row in pr_files:
        pr_number = row.get("pull_request_number")
        if pr_number is None:
            continue
        number = int(pr_number)
        if number in active_numbers:
            files_by_pr[number].append(row)

    profiles = [
        _build_scope_profile(row, files_by_pr.get(int(row["number"]), []), settings)
        for row in active_prs
    ]
    profiles_by_number = {profile.number: profile for profile in profiles}
    feature_idf = _feature_idf(profiles, settings)
    for profile in profiles:
        profile.vector = _normalize_vector(
            {
                feature: weight * feature_idf[feature]
                for feature, weight in profile.raw_vector.items()
                if feature in feature_idf
            }
        )

    comparisons = _pairwise_comparisons(profiles, settings)
    if not comparisons:
        return []
    comparison_map = {(entry.left, entry.right): entry for entry in comparisons}

    top_neighbors = _top_neighbors(comparisons, settings)
    edges = _cluster_edges(comparisons, top_neighbors, settings)
    if not edges:
        return []

    clusters: list[PrScopeCluster] = []
    for component in _connected_components(sorted(active_numbers), edges):
        if len(component) < settings.min_cluster_size:
            continue
        member_sets = _refine_component(component, comparison_map, settings)
        for members in member_sets:
            clusters.append(
                _cluster_entry(
                    members=members,
                    profiles_by_number=profiles_by_number,
                    comparison_map=comparison_map,
                    feature_idf=feature_idf,
                    settings=settings,
                )
            )
    return sorted(
        clusters,
        key=lambda cluster: (
            -len(cluster.pr_numbers),
            -cluster.average_similarity,
            cluster.cluster_id,
        ),
    )


def run_pr_scope_report(options: Any) -> Path:
    """Resolve a snapshot, cluster open PR scopes, and write a JSON report."""

    snapshot_dir = _resolve_snapshot_dir(options)
    snapshot = _load_snapshot_context(snapshot_dir)
    clusters = build_pr_scope_clusters(
        snapshot["pull_requests"],
        snapshot["pr_files"],
        suppression_rules=options.cluster_suppression_rules,
    )
    output_path = (options.output or (snapshot_dir / "pr-scope-clusters.json")).resolve()
    output_path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "repo": snapshot["repo"],
        "snapshot_id": snapshot["snapshot_id"],
        "generated_at": datetime.now(tz=UTC)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z"),
        "cluster_count": len(clusters),
        "pr_scope_clusters": [cluster.model_dump(mode="json") for cluster in clusters],
    }
    output_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    return output_path


def _resolve_snapshot_dir(options: Any) -> Path:
    return resolve_snapshot_source_dir(
        snapshot_dir=options.snapshot_dir,
        local_snapshots_root=options.output_dir.resolve() / "snapshots",
        hf_repo_id=options.hf_repo_id,
        hf_revision=options.hf_revision,
        hf_materialize_dir=options.hf_materialize_dir,
        hf_output_dir=options.output_dir,
    )


def _load_snapshot_context(snapshot_dir: Path) -> dict[str, Any]:
    manifest_path = snapshot_dir / "manifest.json"
    manifest = read_json(manifest_path) if manifest_path.exists() else {}
    pull_requests = read_parquet_rows(snapshot_dir / "pull_requests.parquet")
    pr_files = read_parquet_rows(snapshot_dir / "pr_files.parquet")
    repo = manifest.get("repo") or (pull_requests[0]["repo"] if pull_requests else None) or ""
    snapshot_id = manifest.get("snapshot_id") or snapshot_dir.name
    return {
        "repo": repo,
        "snapshot_id": snapshot_id,
        "pull_requests": pull_requests,
        "pr_files": pr_files,
    }


def _include_pull_request(row: Mapping[str, Any], options: PrScopeClusterOptions) -> bool:
    if row.get("number") is None:
        return False
    if not options.include_closed and str(row.get("state") or "").lower() != "open":
        return False
    return options.include_drafts or not bool(row.get("draft"))


def _build_scope_profile(
    pr_row: Mapping[str, Any],
    file_rows: Sequence[Mapping[str, Any]],
    options: PrScopeClusterOptions,
) -> _ScopeProfile:
    number = int(pr_row["number"])
    file_lines: defaultdict[str, int] = defaultdict(int)
    prefix_directory_lines: defaultdict[str, int] = defaultdict(int)
    leaf_directory_lines: defaultdict[str, int] = defaultdict(int)
    chunk_lines: defaultdict[str, int] = defaultdict(int)

    for row in file_rows:
        filename = str(row.get("filename") or "").strip()
        if not filename:
            continue
        changed_lines = _file_changed_lines(row)
        file_lines[filename] += changed_lines
        leaf_directory = _leaf_directory(filename)
        leaf_directory_lines[leaf_directory] += changed_lines
        for directory in _directory_prefixes(filename):
            prefix_directory_lines[directory] += changed_lines
        patch = row.get("patch")
        if patch:
            for chunk_key, chunk_size in _chunk_line_weights(
                filename, str(patch), options.chunk_size
            ).items():
                chunk_lines[chunk_key] += chunk_size

    if not file_lines:
        fallback_total = max(
            1, int(pr_row.get("additions") or 0) + int(pr_row.get("deletions") or 0)
        )
        file_count = max(1, int(pr_row.get("changed_files") or 0))
        return _ScopeProfile(
            number=number,
            total_changed_lines=fallback_total,
            file_count=file_count,
            directory_count=0,
            dominant_dir_share=0.0,
            filenames=set(),
            directories=set(),
            raw_vector={},
            vector={},
        )

    raw_vector: dict[str, float] = {}
    for filename, changed_lines in file_lines.items():
        raw_vector[f"file:{filename}"] = sqrt(changed_lines) * options.file_weight
    for directory, changed_lines in prefix_directory_lines.items():
        raw_vector[f"dir:{directory}"] = (
            raw_vector.get(f"dir:{directory}", 0.0) + sqrt(changed_lines) * options.directory_weight
        )
    for chunk_key, changed_lines in chunk_lines.items():
        raw_vector[chunk_key] = (
            raw_vector.get(chunk_key, 0.0) + sqrt(changed_lines) * options.chunk_weight
        )

    total_changed_lines = sum(file_lines.values())
    dominant_dir_share = 0.0
    if leaf_directory_lines and total_changed_lines > 0:
        dominant_dir_share = max(leaf_directory_lines.values()) / total_changed_lines
    return _ScopeProfile(
        number=number,
        total_changed_lines=total_changed_lines,
        file_count=len(file_lines),
        directory_count=len(leaf_directory_lines),
        dominant_dir_share=dominant_dir_share,
        filenames=set(file_lines),
        directories=set(prefix_directory_lines),
        raw_vector=raw_vector,
        vector={},
    )


def _file_changed_lines(row: Mapping[str, Any]) -> int:
    additions = int(row.get("additions") or 0)
    deletions = int(row.get("deletions") or 0)
    changes = int(row.get("changes") or 0)
    total = additions + deletions
    if total > 0:
        return total
    if changes > 0:
        return changes
    return 1


def _directory_prefixes(filename: str) -> list[str]:
    parts = PurePosixPath(filename).parts[:-1]
    prefixes: list[str] = []
    current: list[str] = []
    for part in parts:
        current.append(part)
        prefixes.append("/".join(current))
    return prefixes


def _leaf_directory(filename: str) -> str:
    parts = PurePosixPath(filename).parts[:-1]
    return "/".join(parts) if parts else "."


def _chunk_line_weights(filename: str, patch: str, chunk_size: int) -> dict[str, int]:
    weights: defaultdict[str, int] = defaultdict(int)
    for start, end in _patch_ranges(patch):
        current = start
        while current <= end:
            chunk_index = (current - 1) // chunk_size
            chunk_start = chunk_index * chunk_size + 1
            chunk_end = chunk_start + chunk_size - 1
            overlap_end = min(end, chunk_end)
            weights[f"chunk:{filename}:{chunk_start}-{chunk_end}"] += overlap_end - current + 1
            current = overlap_end + 1
    return weights


def _patch_ranges(patch: str) -> list[tuple[int, int]]:
    ranges: list[tuple[int, int]] = []
    for line in patch.splitlines():
        match = HUNK_HEADER_PATTERN.match(line)
        if match is None:
            continue
        start = int(match.group("start"))
        count = int(match.group("count") or "1")
        if count <= 0:
            continue
        ranges.append((start, start + count - 1))
    return ranges


def _feature_idf(
    profiles: Sequence[_ScopeProfile], options: PrScopeClusterOptions
) -> dict[str, float]:
    document_frequency: Counter[str] = Counter()
    profile_count = len(profiles)
    for profile in profiles:
        document_frequency.update(profile.raw_vector.keys())

    idf: dict[str, float] = {}
    for feature, frequency in document_frequency.items():
        if frequency / profile_count > options.max_feature_df_ratio:
            continue
        idf[feature] = log((profile_count + 1) / (frequency + 1)) + 1.0
    return idf


def _normalize_vector(vector: Mapping[str, float]) -> dict[str, float]:
    norm = sqrt(sum(weight * weight for weight in vector.values()))
    if norm <= 0.0:
        return {}
    return {feature: weight / norm for feature, weight in vector.items() if weight > 0.0}


def _pairwise_comparisons(
    profiles: Sequence[_ScopeProfile],
    options: PrScopeClusterOptions,
) -> list[_ScopeComparison]:
    comparisons: list[_ScopeComparison] = []
    weight_total = (
        options.content_weight
        + options.size_weight
        + options.breadth_weight
        + options.concentration_weight
    )
    if weight_total <= 0:
        raise ValueError("PR scope similarity weights must sum to a positive value.")

    ordered = sorted(profiles, key=lambda profile: profile.number)
    for index, left in enumerate(ordered):
        for right in ordered[index + 1 :]:
            content_similarity = _cosine_similarity(left.vector, right.vector)
            if (
                content_similarity <= 0.0
                and not left.filenames.intersection(right.filenames)
                and not left.directories.intersection(right.directories)
            ):
                continue
            size_similarity = _ratio_similarity(left.total_changed_lines, right.total_changed_lines)
            breadth_similarity = (
                _ratio_similarity(left.file_count, right.file_count)
                + _ratio_similarity(left.directory_count, right.directory_count)
            ) / 2.0
            concentration_similarity = max(
                0.0, 1.0 - abs(left.dominant_dir_share - right.dominant_dir_share)
            )
            similarity = (
                content_similarity * options.content_weight
                + size_similarity * options.size_weight
                + breadth_similarity * options.breadth_weight
                + concentration_similarity * options.concentration_weight
            ) / weight_total
            comparisons.append(
                _ScopeComparison(
                    left=left.number,
                    right=right.number,
                    similarity=similarity,
                    content_similarity=content_similarity,
                    size_similarity=size_similarity,
                    breadth_similarity=breadth_similarity,
                    concentration_similarity=concentration_similarity,
                    shared_filenames=sorted(left.filenames & right.filenames)[:10],
                    shared_directories=sorted(
                        left.directories & right.directories,
                        key=lambda value: (-value.count("/"), value),
                    )[:10],
                )
            )
    return comparisons


def _cosine_similarity(left: Mapping[str, float], right: Mapping[str, float]) -> float:
    if not left or not right:
        return 0.0
    if len(left) > len(right):
        left, right = right, left
    return sum(weight * right.get(feature, 0.0) for feature, weight in left.items())


def _ratio_similarity(left: int, right: int) -> float:
    largest = max(left, right)
    if largest <= 0:
        return 1.0
    return min(left, right) / largest


def _top_neighbors(
    comparisons: Sequence[_ScopeComparison],
    options: PrScopeClusterOptions,
) -> dict[int, set[int]]:
    ranked: defaultdict[int, list[tuple[float, int]]] = defaultdict(list)
    for entry in comparisons:
        if entry.similarity < options.min_similarity:
            continue
        ranked[entry.left].append((entry.similarity, entry.right))
        ranked[entry.right].append((entry.similarity, entry.left))

    neighbors: dict[int, set[int]] = {}
    for number, items in ranked.items():
        sorted_items = sorted(items, key=lambda item: (-item[0], item[1]))
        neighbors[number] = {other for _, other in sorted_items[: options.max_neighbors]}
    return neighbors


def _cluster_edges(
    comparisons: Sequence[_ScopeComparison],
    top_neighbors: Mapping[int, set[int]],
    options: PrScopeClusterOptions,
) -> set[tuple[int, int]]:
    edges: set[tuple[int, int]] = set()
    for entry in comparisons:
        if entry.similarity < options.min_similarity:
            continue
        left_neighbors = top_neighbors.get(entry.left, set())
        right_neighbors = top_neighbors.get(entry.right, set())
        if entry.similarity >= options.strong_similarity:
            edges.add((entry.left, entry.right))
            continue
        if entry.right not in left_neighbors or entry.left not in right_neighbors:
            continue
        shared_neighbor_count = len(left_neighbors & right_neighbors)
        if (
            shared_neighbor_count >= options.min_shared_neighbors
            or entry.similarity >= options.pair_similarity
        ):
            edges.add((entry.left, entry.right))
    return edges


def _connected_components(numbers: Sequence[int], edges: set[tuple[int, int]]) -> list[list[int]]:
    adjacency: defaultdict[int, set[int]] = defaultdict(set)
    for left, right in edges:
        adjacency[left].add(right)
        adjacency[right].add(left)

    components: list[list[int]] = []
    seen: set[int] = set()
    for number in sorted(numbers):
        if number in seen or number not in adjacency:
            continue
        stack = [number]
        component: list[int] = []
        while stack:
            current = stack.pop()
            if current in seen:
                continue
            seen.add(current)
            component.append(current)
            stack.extend(sorted(adjacency[current] - seen, reverse=True))
        if component:
            components.append(sorted(component))
    return components


def _refine_component(
    members: Sequence[int],
    comparison_map: Mapping[tuple[int, int], _ScopeComparison],
    options: PrScopeClusterOptions,
) -> list[list[int]]:
    if len(members) <= options.min_cluster_size:
        average_similarity = _cluster_average_similarity(members, comparison_map)
        if average_similarity >= options.min_cluster_average_similarity:
            return [sorted(members)]
        return []

    remaining = set(members)
    refined: list[list[int]] = []
    while len(remaining) >= options.min_cluster_size:
        seed = max(
            sorted(remaining),
            key=lambda number: (
                _mean_similarity(number, remaining - {number}, comparison_map),
                -number,
            ),
        )
        cluster = [seed]
        candidates = sorted(
            remaining - {seed},
            key=lambda number: (_similarity(seed, number, comparison_map), -number),
            reverse=True,
        )
        for candidate in candidates:
            mean_to_cluster = _mean_similarity(candidate, set(cluster), comparison_map)
            if mean_to_cluster >= options.expansion_similarity:
                cluster.append(candidate)
        cluster = sorted(cluster)
        average_similarity = _cluster_average_similarity(cluster, comparison_map)
        if (
            len(cluster) >= options.min_cluster_size
            and average_similarity >= options.min_cluster_average_similarity
        ):
            refined.append(cluster)
            remaining.difference_update(cluster)
        else:
            remaining.remove(seed)
    return refined


def _mean_similarity(
    number: int,
    others: set[int],
    comparison_map: Mapping[tuple[int, int], _ScopeComparison],
) -> float:
    if not others:
        return 0.0
    return sum(_similarity(number, other, comparison_map) for other in others) / len(others)


def _cluster_average_similarity(
    members: Sequence[int],
    comparison_map: Mapping[tuple[int, int], _ScopeComparison],
) -> float:
    if len(members) < 2:
        return 0.0
    total = 0.0
    comparisons = 0
    ordered = sorted(members)
    for index, left in enumerate(ordered):
        for right in ordered[index + 1 :]:
            total += _similarity(left, right, comparison_map)
            comparisons += 1
    if comparisons == 0:
        return 0.0
    return total / comparisons


def _similarity(
    left: int,
    right: int,
    comparison_map: Mapping[tuple[int, int], _ScopeComparison],
) -> float:
    entry = comparison_map.get((left, right)) or comparison_map.get((right, left))
    return entry.similarity if entry is not None else 0.0


def _cluster_entry(
    *,
    members: Sequence[int],
    profiles_by_number: Mapping[int, _ScopeProfile],
    comparison_map: Mapping[tuple[int, int], _ScopeComparison],
    feature_idf: Mapping[str, float],
    settings: PrScopeClusterOptions,
) -> PrScopeCluster:
    ordered = sorted(members)
    pairwise: list[PrScopePair] = []
    for index, left in enumerate(ordered):
        for right in ordered[index + 1 :]:
            entry = comparison_map.get((left, right)) or comparison_map.get((right, left))
            if entry is None:
                continue
            pairwise.append(
                PrScopePair(
                    left_pr_number=entry.left,
                    right_pr_number=entry.right,
                    similarity=round(entry.similarity, 3),
                    content_similarity=round(entry.content_similarity, 3),
                    size_similarity=round(entry.size_similarity, 3),
                    breadth_similarity=round(entry.breadth_similarity, 3),
                    concentration_similarity=round(entry.concentration_similarity, 3),
                    shared_filenames=entry.shared_filenames,
                    shared_directories=entry.shared_directories,
                )
            )
    pairwise.sort(
        key=lambda entry: (-entry.similarity, entry.left_pr_number, entry.right_pr_number)
    )
    average_similarity = _cluster_average_similarity(ordered, comparison_map)
    representative_pr_number = _representative_pr_number(ordered, comparison_map)
    shared_filenames = _shared_exact_features(
        ordered,
        profiles_by_number,
        feature_idf,
        prefix="file:",
        limit=settings.max_shared_features,
    )
    shared_directories = _shared_exact_features(
        ordered,
        profiles_by_number,
        feature_idf,
        prefix="dir:",
        limit=settings.max_shared_features,
    )
    return PrScopeCluster(
        cluster_id=f"pr-scope-{ordered[0]}-{len(ordered)}",
        pr_numbers=ordered,
        representative_pr_number=representative_pr_number,
        average_similarity=round(average_similarity, 3),
        summary=_cluster_summary(
            ordered, representative_pr_number, shared_filenames, shared_directories
        ),
        shared_filenames=shared_filenames,
        shared_directories=shared_directories,
        pairwise=pairwise,
    )


def _representative_pr_number(
    members: Sequence[int],
    comparison_map: Mapping[tuple[int, int], _ScopeComparison],
) -> int:
    return max(
        sorted(members),
        key=lambda number: (
            _mean_similarity(number, set(members) - {number}, comparison_map),
            -number,
        ),
    )


def _shared_exact_features(
    members: Sequence[int],
    profiles_by_number: Mapping[int, _ScopeProfile],
    feature_idf: Mapping[str, float],
    *,
    prefix: str,
    limit: int,
) -> list[str]:
    counts: Counter[str] = Counter()
    for number in members:
        profile = profiles_by_number[number]
        values = profile.filenames if prefix == "file:" else profile.directories
        counts.update(values)
    minimum_count = 2 if len(members) > 1 else 1
    shared = [
        value
        for value, count in counts.items()
        if count >= minimum_count and f"{prefix}{value}" in feature_idf
    ]
    if prefix == "dir:":
        shared.sort(
            key=lambda value: (
                -counts[value],
                -value.count("/"),
                -feature_idf.get(f"{prefix}{value}", 0.0),
                value,
            )
        )
    else:
        shared.sort(
            key=lambda value: (-counts[value], -feature_idf.get(f"{prefix}{value}", 0.0), value)
        )
    return shared[:limit]


def _cluster_summary(
    members: Sequence[int],
    representative_pr_number: int,
    shared_filenames: Sequence[str],
    shared_directories: Sequence[str],
) -> str:
    count = len(members)
    if shared_filenames:
        preview = ", ".join(f"`{value}`" for value in shared_filenames[:3])
        return (
            f"{count} open PRs share weighted file overlap around {preview}; "
            f"representative PR #{representative_pr_number}."
        )
    if shared_directories:
        preview = ", ".join(f"`{value}`" for value in shared_directories[:3])
        return (
            f"{count} open PRs cluster in {preview} with similar change breadth; "
            f"representative PR #{representative_pr_number}."
        )
    return f"{count} open PRs have similar weighted scope; representative PR #{representative_pr_number}."