evalstate/diffusers-pr-api

1from __future__ import annotations2 3import re4from collections import defaultdict5from typing import Any6 7REFERENCE_PATTERN = re.compile(8    r"(?:(?P<owner>[A-Za-z0-9_.-]+)/(?P<repo>[A-Za-z0-9_.-]+))?#(?P<number>\d+)"9)10CLOSING_KEYWORD_PATTERN = re.compile(11    r"\b(?P<verb>close[sd]?|fix(?:e[sd])?|resolve[sd]?)\s+(?:(?P<owner>[A-Za-z0-9_.-]+)/(?P<repo>[A-Za-z0-9_.-]+))?#(?P<number>\d+)\b",12    flags=re.IGNORECASE,13)14DUPLICATE_REFERENCE_PATTERN = re.compile(15    r"\bduplicate\s+of\s+(?:(?P<owner>[A-Za-z0-9_.-]+)/(?P<repo>[A-Za-z0-9_.-]+))?#(?P<number>\d+)\b",16    flags=re.IGNORECASE,17)18 19 20def extract_references(21    text: str | None, default_owner: str, default_repo: str22) -> list[dict[str, Any]]:23    if not text:24        return []25    refs: list[dict[str, Any]] = []26    for match in REFERENCE_PATTERN.finditer(text):27        refs.append(28            {29                "target_owner": match.group("owner") or default_owner,30                "target_repo": match.group("repo") or default_repo,31                "target_number": int(match.group("number")),32                "reference_kind": "mention",33            }34        )35    for match in CLOSING_KEYWORD_PATTERN.finditer(text):36        refs.append(37            {38                "target_owner": match.group("owner") or default_owner,39                "target_repo": match.group("repo") or default_repo,40                "target_number": int(match.group("number")),41                "reference_kind": "closing_reference",42                "verb": match.group("verb").lower(),43            }44        )45    for match in DUPLICATE_REFERENCE_PATTERN.finditer(text):46        refs.append(47            {48                "target_owner": match.group("owner") or default_owner,49                "target_repo": match.group("repo") or default_repo,50                "target_number": int(match.group("number")),51                "reference_kind": "duplicate_reference",52            }53        )54    return refs55 56 57def build_text_link_rows(58    *,59    repo: str,60    owner: str,61    repo_name: str,62    source_type: str,63    source_number: int,64    source_id: int | None,65    body: str | None,66    snapshot_id: str,67    extracted_at: str,68) -> list[dict[str, Any]]:69    rows: list[dict[str, Any]] = []70    seen: set[tuple[str, str, int, str]] = set()71    for ref in extract_references(body, owner, repo_name):72        key = (ref["target_owner"], ref["target_repo"], ref["target_number"], ref["reference_kind"])73        if key in seen:74            continue75        seen.add(key)76        rows.append(77            {78                "repo": repo,79                "source_type": source_type,80                "source_number": source_number,81                "source_github_id": source_id,82                "target_owner": ref["target_owner"],83                "target_repo": ref["target_repo"],84                "target_number": ref["target_number"],85                "link_type": ref["reference_kind"],86                "link_origin": "text",87                "snapshot_id": snapshot_id,88                "extracted_at": extracted_at,89            }90        )91    return rows92 93 94def build_pr_duplicate_candidate_rows(95    *,96    repo: str,97    pull_requests: list[dict[str, Any]],98    link_rows: list[dict[str, Any]],99    snapshot_id: str,100    extracted_at: str,101) -> list[dict[str, Any]]:102    pr_targets: dict[int, set[int]] = defaultdict(set)103    for row in link_rows:104        if row["source_type"] != "pull_request":105            continue106        if (107            row["target_owner"] != repo.split("/", 1)[0]108            or row["target_repo"] != repo.split("/", 1)[1]109        ):110            continue111        pr_targets[row["source_number"]].add(row["target_number"])112 113    target_to_prs: dict[int, list[int]] = defaultdict(list)114    for pr_row in pull_requests:115        number = pr_row["number"]116        for target in sorted(pr_targets.get(number, set())):117            target_to_prs[target].append(number)118 119    rows: list[dict[str, Any]] = []120    for target_number, prs in target_to_prs.items():121        if len(prs) < 2:122            continue123        for pr_number in prs:124            rows.append(125                {126                    "repo": repo,127                    "source_type": "pull_request",128                    "source_number": pr_number,129                    "source_github_id": None,130                    "target_owner": repo.split("/", 1)[0],131                    "target_repo": repo.split("/", 1)[1],132                    "target_number": target_number,133                    "link_type": "shared_issue_target",134                    "link_origin": "derived",135                    "snapshot_id": snapshot_id,136                    "extracted_at": extracted_at,137                }138            )139    return rows140