evalstate/diffusers-pr-api

1from __future__ import annotations2 3from dataclasses import dataclass4from pathlib import Path5from typing import Any, Literal6 7from slop_farmer.data.parquet_io import read_json, read_parquet_rows8from slop_farmer.data.snapshot_paths import (9    CURRENT_ANALYSIS_MANIFEST_PATH,10    load_current_analysis_manifest,11    repo_relative_path_to_local,12)13 14AnalysisVariant = Literal["auto", "hybrid", "deterministic"]15 16 17@dataclass(slots=True, frozen=True)18class _SnapshotMetadata:19    repo: str20    snapshot_id: str21 22 23@dataclass(slots=True, frozen=True)24class _AnalysisSelection:25    path: Path26    payload: dict[str, Any]27    variant_used: str28    llm_enrichment: bool29 30 31def get_snapshot_surfaces(snapshot_dir: Path) -> dict[str, Any]:32    issue_status = get_issue_cluster_status(snapshot_dir, variant="auto")33    contributor_status = get_contributor_status(snapshot_dir)34    return {35        "issues": {36            "available": issue_status["available"],37            "variant_used": issue_status.get("variant_used"),38            "llm_enrichment": issue_status.get("llm_enrichment"),39            "generated_at": issue_status.get("generated_at"),40            "cluster_count": (issue_status.get("counts") or {}).get("meta_bugs", 0),41            "duplicate_pr_count": (issue_status.get("counts") or {}).get("duplicate_prs", 0),42            "available_variants": issue_status.get("available_variants") or [],43        },44        "contributors": {45            "available": contributor_status["available"],46            "generated_at": contributor_status.get("generated_at"),47            "contributor_count": contributor_status.get("contributor_count", 0),48        },49    }50 51 52def get_issue_cluster_status(snapshot_dir: Path, *, variant: AnalysisVariant) -> dict[str, Any]:53    metadata = _snapshot_metadata(snapshot_dir)54    candidates = _analysis_candidates(snapshot_dir)55    selection = _select_analysis_report(candidates, variant=variant)56    status = {57        "repo": metadata.repo,58        "snapshot_id": metadata.snapshot_id,59        "variant_requested": variant,60        "available": selection is not None,61        "available_variants": sorted({candidate["variant"] for candidate in candidates}),62    }63    if selection is None:64        return {65            **status,66            "variant_used": None,67            "llm_enrichment": False,68            "generated_at": None,69            "report_path": None,70            "counts": {"meta_bugs": 0, "duplicate_issues": 0, "duplicate_prs": 0},71        }72    payload = selection.payload73    return {74        **status,75        "variant_used": selection.variant_used,76        "llm_enrichment": selection.llm_enrichment,77        "generated_at": payload.get("generated_at"),78        "report_path": selection.path.name,79        "counts": _analysis_counts(payload),80    }81 82 83def list_issue_clusters(84    snapshot_dir: Path,85    *,86    limit: int | None,87    variant: AnalysisVariant,88) -> dict[str, Any]:89    metadata, selection, issue_map, pr_map = _analysis_context(snapshot_dir, variant=variant)90    base = _analysis_base_payload(metadata, selection, variant=variant)91    if selection is None:92        return {**base, "clusters": [], "cluster_count": 0}93    clusters = [94        _issue_cluster_summary(cluster, issue_map, pr_map, rank=index)95        for index, cluster in enumerate(selection.payload.get("meta_bugs") or [], start=1)96    ]97    total = len(clusters)98    return {99        **base,100        "clusters": clusters[:limit] if limit is not None else clusters,101        "cluster_count": total,102    }103 104 105def get_issue_cluster(106    snapshot_dir: Path,107    *,108    cluster_id: str,109    variant: AnalysisVariant,110) -> dict[str, Any]:111    metadata, selection, issue_map, pr_map = _analysis_context(snapshot_dir, variant=variant)112    base = _analysis_base_payload(metadata, selection, variant=variant)113    if selection is None:114        return {115            **base,116            "cluster_id": cluster_id,117            "found": False,118            "cluster": None,119            "issues": [],120            "pull_requests": [],121        }122    cluster = next(123        (124            row125            for row in selection.payload.get("meta_bugs") or []126            if str(row.get("cluster_id") or "") == cluster_id127        ),128        None,129    )130    if cluster is None:131        return {132            **base,133            "cluster_id": cluster_id,134            "found": False,135            "cluster": None,136            "issues": [],137            "pull_requests": [],138        }139    issue_numbers = _ordered_ints(cluster.get("issue_numbers"))140    pr_numbers = _ordered_ints(cluster.get("pr_numbers"))141    canonical_pr_number = _coerce_int(cluster.get("canonical_pr_number"))142    return {143        **base,144        "cluster_id": cluster_id,145        "found": True,146        "cluster": {147            **_issue_cluster_summary(cluster, issue_map, pr_map),148            "canonical_issue_reason": cluster.get("canonical_issue_reason"),149            "canonical_pr_reason": cluster.get("canonical_pr_reason"),150            "best_issue_reason": cluster.get("best_issue_reason"),151            "best_pr_reason": cluster.get("best_pr_reason"),152        },153        "issues": [_issue_member_row(number, issue_map.get(number)) for number in issue_numbers],154        "pull_requests": [155            _pr_member_row(156                number,157                pr_map.get(number),158                role="canonical" if canonical_pr_number == number else "member",159            )160            for number in pr_numbers161        ],162    }163 164 165def get_issue_clusters_for_pr(166    snapshot_dir: Path,167    *,168    pr_number: int,169    variant: AnalysisVariant,170) -> dict[str, Any]:171    metadata, selection, issue_map, pr_map = _analysis_context(snapshot_dir, variant=variant)172    base = _analysis_base_payload(metadata, selection, variant=variant)173    if selection is None:174        return {**base, "pr_number": pr_number, "found": False, "clusters": [], "cluster_count": 0}175    matches = []176    for index, cluster in enumerate(selection.payload.get("meta_bugs") or [], start=1):177        pr_numbers = _ordered_ints(cluster.get("pr_numbers"))178        if pr_number not in pr_numbers:179            continue180        canonical_pr_number = _coerce_int(cluster.get("canonical_pr_number"))181        matches.append(182            {183                **_issue_cluster_summary(cluster, issue_map, pr_map, rank=index),184                "membership_role": "canonical" if canonical_pr_number == pr_number else "member",185            }186        )187    return {188        **base,189        "pr_number": pr_number,190        "found": bool(matches),191        "clusters": matches,192        "cluster_count": len(matches),193    }194 195 196def check_issue_cluster_membership(197    snapshot_dir: Path,198    *,199    pr_number: int,200    cluster_id: str | None,201    variant: AnalysisVariant,202) -> dict[str, Any]:203    lookup = get_issue_clusters_for_pr(snapshot_dir, pr_number=pr_number, variant=variant)204    matches = list(lookup.get("clusters") or [])205    matching_cluster_ids = [str(row.get("cluster_id")) for row in matches if row.get("cluster_id")]206    if cluster_id is None:207        return {208            **lookup,209            "cluster_id": None,210            "matched": bool(matching_cluster_ids),211            "matching_cluster_ids": matching_cluster_ids,212        }213    match = next((row for row in matches if row.get("cluster_id") == cluster_id), None)214    return {215        **lookup,216        "cluster_id": cluster_id,217        "matched": match is not None,218        "matching_cluster_ids": matching_cluster_ids,219        "membership": match,220    }221 222 223def list_issue_duplicate_prs(224    snapshot_dir: Path,225    *,226    limit: int | None,227    variant: AnalysisVariant,228) -> dict[str, Any]:229    metadata, selection, issue_map, pr_map = _analysis_context(snapshot_dir, variant=variant)230    base = _analysis_base_payload(metadata, selection, variant=variant)231    if selection is None:232        return {**base, "duplicate_prs": [], "duplicate_pr_count": 0}233    rows = [234        _duplicate_pr_summary(entry, issue_map, pr_map, rank=index)235        for index, entry in enumerate(selection.payload.get("duplicate_prs") or [], start=1)236    ]237    total = len(rows)238    return {239        **base,240        "duplicate_prs": rows[:limit] if limit is not None else rows,241        "duplicate_pr_count": total,242    }243 244 245def get_issue_best(snapshot_dir: Path, *, variant: AnalysisVariant) -> dict[str, Any]:246    metadata, selection, issue_map, pr_map = _analysis_context(snapshot_dir, variant=variant)247    base = _analysis_base_payload(metadata, selection, variant=variant)248    if selection is None:249        return {**base, "best_issue": None, "best_pr": None}250    return {251        **base,252        "best_issue": _best_issue_summary(selection.payload.get("best_issue"), issue_map),253        "best_pr": _best_pr_summary(selection.payload.get("best_pr"), pr_map),254    }255 256 257def get_contributor_status(snapshot_dir: Path) -> dict[str, Any]:258    metadata = _snapshot_metadata(snapshot_dir)259    report = _read_optional_json(snapshot_dir / "new-contributors-report.json")260    raw_contributors = report.get("contributors")261    contributors: list[Any] = raw_contributors if isinstance(raw_contributors, list) else []262    return {263        "repo": str(report.get("repo") or metadata.repo),264        "snapshot_id": str(report.get("snapshot_id") or metadata.snapshot_id),265        "available": bool(report),266        "generated_at": report.get("generated_at"),267        "window_days": _coerce_int(report.get("window_days")),268        "contributor_count": len(contributors),269    }270 271 272def list_contributors(snapshot_dir: Path, *, limit: int | None) -> dict[str, Any]:273    status = get_contributor_status(snapshot_dir)274    report = _read_optional_json(snapshot_dir / "new-contributors-report.json")275    rows = [276        _contributor_summary(entry, rank=index)277        for index, entry in enumerate(report.get("contributors") or [], start=1)278        if isinstance(entry, dict)279    ]280    total = len(rows)281    return {282        **status,283        "contributors": rows[:limit] if limit is not None else rows,284        "contributor_count": total,285    }286 287 288def get_contributor(snapshot_dir: Path, *, author_login: str) -> dict[str, Any]:289    status = get_contributor_status(snapshot_dir)290    report = _read_optional_json(snapshot_dir / "new-contributors-report.json")291    contributor = _find_contributor(report.get("contributors") or [], author_login)292    if contributor is None:293        return {294            **status,295            "author_login": author_login,296            "found": False,297            "summary": None,298            "risk": None,299            "contributor": None,300        }301    return {302        **status,303        "author_login": str(contributor.get("author_login") or author_login),304        "found": True,305        "summary": _contributor_summary(contributor),306        "risk": _contributor_risk(contributor),307        "contributor": contributor,308    }309 310 311def get_contributor_risk(snapshot_dir: Path, *, author_login: str) -> dict[str, Any]:312    contributor = get_contributor(snapshot_dir, author_login=author_login)313    risk = contributor.get("risk")314    return {315        "repo": contributor.get("repo"),316        "snapshot_id": contributor.get("snapshot_id"),317        "available": contributor.get("available"),318        "generated_at": contributor.get("generated_at"),319        "author_login": contributor.get("author_login"),320        "found": contributor.get("found"),321        "risk_available": risk is not None,322        "risk": risk,323    }324 325 326def _analysis_context(327    snapshot_dir: Path,328    *,329    variant: AnalysisVariant,330) -> tuple[331    _SnapshotMetadata,332    _AnalysisSelection | None,333    dict[int, dict[str, Any]],334    dict[int, dict[str, Any]],335]:336    metadata = _snapshot_metadata(snapshot_dir)337    selection = _select_analysis_report(_analysis_candidates(snapshot_dir), variant=variant)338    issue_map, pr_map = _artifact_maps(snapshot_dir)339    return metadata, selection, issue_map, pr_map340 341 342def _analysis_base_payload(343    metadata: _SnapshotMetadata,344    selection: _AnalysisSelection | None,345    *,346    variant: AnalysisVariant,347) -> dict[str, Any]:348    base = {349        "repo": metadata.repo,350        "snapshot_id": metadata.snapshot_id,351        "variant_requested": variant,352        "available": selection is not None,353        "variant_used": None,354        "llm_enrichment": False,355        "generated_at": None,356    }357    if selection is None:358        return base359    return {360        **base,361        "variant_used": selection.variant_used,362        "llm_enrichment": selection.llm_enrichment,363        "generated_at": selection.payload.get("generated_at"),364    }365 366 367def _analysis_candidates(snapshot_dir: Path) -> list[dict[str, Any]]:368    candidates: list[dict[str, Any]] = []369    for path in _analysis_report_paths(snapshot_dir):370        payload = _read_optional_json(path)371        if not payload:372            continue373        llm_enrichment = bool(payload.get("llm_enrichment"))374        candidates.append(375            {376                "path": path,377                "payload": payload,378                "variant": _analysis_variant(path.name, payload, llm_enrichment=llm_enrichment),379                "llm_enrichment": llm_enrichment,380            }381        )382    return candidates383 384 385def _select_analysis_report(386    candidates: list[dict[str, Any]],387    *,388    variant: AnalysisVariant,389) -> _AnalysisSelection | None:390    if not candidates:391        return None392    if variant == "auto":393        ordered = sorted(candidates, key=_analysis_auto_priority)394    else:395        ordered = [candidate for candidate in candidates if candidate["variant"] == variant]396        ordered.sort(key=_analysis_specific_priority)397    if not ordered:398        return None399    winner = ordered[0]400    return _AnalysisSelection(401        path=Path(winner["path"]),402        payload=dict(winner["payload"]),403        variant_used=str(winner["variant"]),404        llm_enrichment=bool(winner["llm_enrichment"]),405    )406 407 408def _analysis_report_paths(snapshot_dir: Path) -> list[Path]:409    ordered: list[Path] = []410    current_manifest_path = repo_relative_path_to_local(411        snapshot_dir, CURRENT_ANALYSIS_MANIFEST_PATH412    )413    if current_manifest_path.exists():414        try:415            current_manifest = load_current_analysis_manifest(current_manifest_path)416        except ValueError:417            current_manifest = None418        if current_manifest is not None:419            for artifact_path in (current_manifest.get("artifacts") or {}).values():420                if not isinstance(artifact_path, str):421                    continue422                ordered.append(repo_relative_path_to_local(snapshot_dir, artifact_path))423    ordered.extend(424        [425            snapshot_dir / "analysis-report-hybrid.json",426            snapshot_dir / "analysis-report-deterministic.json",427            snapshot_dir / "analysis-report.json",428        ]429    )430    seen: set[Path] = set()431    deduped: list[Path] = []432    for path in ordered:433        if path in seen:434            continue435        seen.add(path)436        deduped.append(path)437    deduped.extend(438        path for path in sorted(snapshot_dir.glob("analysis-report*.json")) if path not in seen439    )440    return [path for path in deduped if path.exists()]441 442 443def _analysis_auto_priority(candidate: dict[str, Any]) -> tuple[int, str]:444    path = Path(candidate["path"])445    if path.name == "analysis-report-hybrid.json":446        return (0, path.name)447    if bool(candidate.get("llm_enrichment")):448        return (1, path.name)449    if path.name == "analysis-report.json":450        return (2, path.name)451    return (3, path.name)452 453 454def _analysis_specific_priority(candidate: dict[str, Any]) -> tuple[int, str]:455    path = Path(candidate["path"])456    if path.name.endswith(f"-{candidate['variant']}.json"):457        return (0, path.name)458    if path.name == "analysis-report.json":459        return (1, path.name)460    return (2, path.name)461 462 463def _analysis_variant(path_name: str, payload: dict[str, Any], *, llm_enrichment: bool) -> str:464    lowered = path_name.lower()465    if "hybrid" in lowered:466        return "hybrid"467    if "deterministic" in lowered:468        return "deterministic"469    if isinstance(payload.get("variant_used"), str):470        variant_used = str(payload["variant_used"]).strip().lower()471        if variant_used in {"hybrid", "deterministic"}:472            return variant_used473    return "hybrid" if llm_enrichment else "deterministic"474 475 476def _analysis_counts(payload: dict[str, Any]) -> dict[str, int]:477    return {478        "meta_bugs": len(payload.get("meta_bugs") or []),479        "duplicate_issues": len(payload.get("duplicate_issues") or []),480        "duplicate_prs": len(payload.get("duplicate_prs") or []),481    }482 483 484def _artifact_maps(485    snapshot_dir: Path,486) -> tuple[dict[int, dict[str, Any]], dict[int, dict[str, Any]]]:487    issue_rows = (488        read_parquet_rows(snapshot_dir / "issues.parquet")489        if (snapshot_dir / "issues.parquet").exists()490        else []491    )492    pr_rows = (493        read_parquet_rows(snapshot_dir / "pull_requests.parquet")494        if (snapshot_dir / "pull_requests.parquet").exists()495        else []496    )497    issue_map = {498        int(row["number"]): row for row in issue_rows if _coerce_int(row.get("number")) is not None499    }500    pr_map = {501        int(row["number"]): row for row in pr_rows if _coerce_int(row.get("number")) is not None502    }503    return issue_map, pr_map504 505 506def _issue_cluster_summary(507    cluster: dict[str, Any],508    issue_map: dict[int, dict[str, Any]],509    pr_map: dict[int, dict[str, Any]],510    *,511    rank: int | None = None,512) -> dict[str, Any]:513    canonical_issue_number = _coerce_int(cluster.get("canonical_issue_number"))514    canonical_pr_number = _coerce_int(cluster.get("canonical_pr_number"))515    issue_numbers = _ordered_ints(cluster.get("issue_numbers"))516    pr_numbers = _ordered_ints(cluster.get("pr_numbers"))517    return {518        "rank": rank,519        "cluster_id": str(cluster.get("cluster_id") or f"cluster-{rank or 0}"),520        "title": _cluster_title(521            cluster, issue_map, pr_map, canonical_issue_number, canonical_pr_number522        ),523        "summary": cluster.get("summary"),524        "status": cluster.get("status"),525        "confidence": _coerce_float(cluster.get("confidence")),526        "canonical_issue_number": canonical_issue_number,527        "canonical_issue_title": _title_for_issue(canonical_issue_number, issue_map),528        "canonical_issue_url": _url_for_issue(canonical_issue_number, issue_map),529        "canonical_pr_number": canonical_pr_number,530        "canonical_pr_title": _title_for_pr(canonical_pr_number, pr_map),531        "canonical_pr_url": _url_for_pr(canonical_pr_number, pr_map),532        "issue_numbers": issue_numbers,533        "issue_count": len(issue_numbers),534        "pr_numbers": pr_numbers,535        "pr_count": len(pr_numbers),536        "evidence_types": [str(value) for value in (cluster.get("evidence_types") or []) if value],537        "github_url": _cluster_url(canonical_issue_number, canonical_pr_number, issue_map, pr_map),538    }539 540 541def _cluster_title(542    cluster: dict[str, Any],543    issue_map: dict[int, dict[str, Any]],544    pr_map: dict[int, dict[str, Any]],545    canonical_issue_number: int | None,546    canonical_pr_number: int | None,547) -> str:548    issue_title = _title_for_issue(canonical_issue_number, issue_map)549    if issue_title:550        return issue_title551    pr_title = _title_for_pr(canonical_pr_number, pr_map)552    if pr_title:553        return pr_title554    summary = str(cluster.get("summary") or "").strip()555    if summary:556        return summary557    return str(cluster.get("cluster_id") or "cluster")558 559 560def _cluster_url(561    canonical_issue_number: int | None,562    canonical_pr_number: int | None,563    issue_map: dict[int, dict[str, Any]],564    pr_map: dict[int, dict[str, Any]],565) -> str | None:566    return _url_for_issue(canonical_issue_number, issue_map) or _url_for_pr(567        canonical_pr_number, pr_map568    )569 570 571def _duplicate_pr_summary(572    entry: dict[str, Any],573    issue_map: dict[int, dict[str, Any]],574    pr_map: dict[int, dict[str, Any]],575    *,576    rank: int,577) -> dict[str, Any]:578    canonical_pr_number = _coerce_int(entry.get("canonical_pr_number"))579    target_issue_number = _coerce_int(entry.get("target_issue_number"))580    duplicates = _ordered_ints(entry.get("duplicate_pr_numbers"))581    return {582        "rank": rank,583        "cluster_id": str(entry.get("cluster_id") or f"duplicate-pr-{rank}"),584        "canonical_pr_number": canonical_pr_number,585        "canonical_pr_title": _title_for_pr(canonical_pr_number, pr_map),586        "canonical_pr_url": _url_for_pr(canonical_pr_number, pr_map),587        "target_issue_number": target_issue_number,588        "target_issue_title": _title_for_issue(target_issue_number, issue_map),589        "target_issue_url": _url_for_issue(target_issue_number, issue_map),590        "duplicate_pr_numbers": duplicates,591        "duplicate_pr_count": len(duplicates),592        "reason": entry.get("reason"),593    }594 595 596def _best_issue_summary(entry: Any, issue_map: dict[int, dict[str, Any]]) -> dict[str, Any] | None:597    if not isinstance(entry, dict):598        return None599    issue_number = _coerce_int(entry.get("issue_number"))600    return {601        "cluster_id": entry.get("cluster_id"),602        "issue_number": issue_number,603        "title": _title_for_issue(issue_number, issue_map),604        "url": _url_for_issue(issue_number, issue_map),605        "reason": entry.get("reason"),606        "score": _coerce_float(entry.get("score")),607    }608 609 610def _best_pr_summary(entry: Any, pr_map: dict[int, dict[str, Any]]) -> dict[str, Any] | None:611    if not isinstance(entry, dict):612        return None613    pr_number = _coerce_int(entry.get("pr_number"))614    return {615        "cluster_id": entry.get("cluster_id"),616        "pr_number": pr_number,617        "title": _title_for_pr(pr_number, pr_map),618        "url": _url_for_pr(pr_number, pr_map),619        "reason": entry.get("reason"),620        "score": _coerce_float(entry.get("score")),621    }622 623 624def _issue_member_row(number: int, row: dict[str, Any] | None) -> dict[str, Any]:625    row = row or {}626    return {627        "number": number,628        "title": row.get("title"),629        "state": row.get("state"),630        "author_login": row.get("author_login"),631        "created_at": row.get("created_at"),632        "updated_at": row.get("updated_at"),633        "html_url": row.get("html_url"),634    }635 636 637def _pr_member_row(number: int, row: dict[str, Any] | None, *, role: str) -> dict[str, Any]:638    row = row or {}639    return {640        "number": number,641        "role": role,642        "title": row.get("title"),643        "author_login": row.get("author_login"),644        "state": row.get("state"),645        "draft": bool(row.get("draft")),646        "merged": bool(row.get("merged")),647        "author_association": row.get("author_association"),648        "created_at": row.get("created_at"),649        "updated_at": row.get("updated_at"),650        "html_url": row.get("html_url"),651    }652 653 654def _contributor_summary(contributor: dict[str, Any], *, rank: int | None = None) -> dict[str, Any]:655    raw_activity = contributor.get("activity")656    activity: dict[str, Any] = raw_activity if isinstance(raw_activity, dict) else {}657    return {658        "rank": rank,659        "author_login": contributor.get("author_login"),660        "name": contributor.get("name"),661        "profile_url": contributor.get("profile_url"),662        "repo_association": contributor.get("repo_association"),663        "first_seen_in_snapshot": contributor.get("first_seen_in_snapshot"),664        "new_to_repo": contributor.get("new_to_repo"),665        "snapshot_pr_count": _coerce_int(contributor.get("snapshot_pr_count")) or 0,666        "snapshot_issue_count": _coerce_int(contributor.get("snapshot_issue_count")) or 0,667        "follow_through_score": contributor.get("follow_through_score"),668        "breadth_score": contributor.get("breadth_score"),669        "automation_risk_signal": contributor.get("automation_risk_signal"),670        "heuristic_note": contributor.get("heuristic_note"),671        "account_age_days": _coerce_int(contributor.get("account_age_days")),672        "public_pr_count_42d": _coerce_int(activity.get("visible_authored_pr_count")),673        "public_repo_count_42d": _coerce_int(activity.get("distinct_repos_with_authored_prs")),674        "repo_pull_requests_url": contributor.get("repo_pull_requests_url"),675        "repo_issues_url": contributor.get("repo_issues_url"),676    }677 678 679def _contributor_risk(contributor: dict[str, Any]) -> dict[str, Any]:680    raw_activity = contributor.get("activity")681    activity: dict[str, Any] = raw_activity if isinstance(raw_activity, dict) else {}682    return {683        "automation_risk_signal": contributor.get("automation_risk_signal"),684        "heuristic_note": contributor.get("heuristic_note"),685        "follow_through_score": contributor.get("follow_through_score"),686        "breadth_score": contributor.get("breadth_score"),687        "account_age_days": _coerce_int(contributor.get("account_age_days")),688        "public_pr_count_42d": _coerce_int(activity.get("visible_authored_pr_count")),689        "public_repo_count_42d": _coerce_int(activity.get("distinct_repos_with_authored_prs")),690        "report_reason": contributor.get("report_reason"),691    }692 693 694def _find_contributor(entries: list[Any], author_login: str) -> dict[str, Any] | None:695    lowered = author_login.casefold()696    for entry in entries:697        if not isinstance(entry, dict):698            continue699        login = str(entry.get("author_login") or "")700        if login.casefold() == lowered:701            return entry702    return None703 704 705def _snapshot_metadata(snapshot_dir: Path) -> _SnapshotMetadata:706    manifest = _read_optional_json(snapshot_dir / "manifest.json")707    repo = str(manifest.get("repo") or _infer_repo(snapshot_dir) or "")708    snapshot_id = str(manifest.get("snapshot_id") or snapshot_dir.name)709    return _SnapshotMetadata(repo=repo, snapshot_id=snapshot_id)710 711 712def _infer_repo(snapshot_dir: Path) -> str | None:713    for filename in ("pull_requests.parquet", "issues.parquet"):714        path = snapshot_dir / filename715        if not path.exists():716            continue717        rows = read_parquet_rows(path)718        if rows and rows[0].get("repo"):719            return str(rows[0]["repo"])720    for filename in _analysis_report_paths(snapshot_dir):721        payload = _read_optional_json(filename)722        if payload.get("repo"):723            return str(payload["repo"])724    report = _read_optional_json(snapshot_dir / "new-contributors-report.json")725    if report.get("repo"):726        return str(report["repo"])727    return None728 729 730def _title_for_issue(number: int | None, issue_map: dict[int, dict[str, Any]]) -> str | None:731    if number is None or number not in issue_map:732        return None733    title = issue_map[number].get("title")734    return str(title) if title else None735 736 737def _url_for_issue(number: int | None, issue_map: dict[int, dict[str, Any]]) -> str | None:738    if number is None or number not in issue_map:739        return None740    value = issue_map[number].get("html_url")741    return str(value) if value else None742 743 744def _title_for_pr(number: int | None, pr_map: dict[int, dict[str, Any]]) -> str | None:745    if number is None or number not in pr_map:746        return None747    title = pr_map[number].get("title")748    return str(title) if title else None749 750 751def _url_for_pr(number: int | None, pr_map: dict[int, dict[str, Any]]) -> str | None:752    if number is None or number not in pr_map:753        return None754    value = pr_map[number].get("html_url")755    return str(value) if value else None756 757 758def _ordered_ints(values: Any) -> list[int]:759    if not isinstance(values, list):760        return []761    ordered: list[int] = []762    for value in values:763        number = _coerce_int(value)764        if number is not None:765            ordered.append(number)766    return ordered767 768 769def _coerce_int(value: Any) -> int | None:770    if value is None:771        return None772    try:773        return int(value)774    except (TypeError, ValueError):775        return None776 777 778def _coerce_float(value: Any) -> float | None:779    if value is None:780        return None781    try:782        return float(value)783    except (TypeError, ValueError):784        return None785 786 787def _read_optional_json(path: Path) -> dict[str, Any]:788    if not path.exists():789        return {}790    payload = read_json(path)791    return payload if isinstance(payload, dict) else {}792