#!/usr/bin/env python3 """Confront emulator profiles with their upstream source. Anchors every source_ref of a profile against the commit the profile was written at and against upstream HEAD, separating what is mechanically recalable from what needs the code read again. """ from __future__ import annotations import argparse import difflib import json import os import posixpath import re import subprocess import sys from dataclasses import asdict, dataclass from datetime import date from pathlib import Path import yaml sys.path.insert(0, os.path.dirname(__file__)) import upstream from common import load_emulator_profiles from upstream import CompareResult, find_renamed DEFAULT_CACHE = ".cache/upstream" ANON_QUOTA = 60 TRIAGE_PATH_SAMPLE = 5 STATUS_ORDER = ( "ANCHORED", "EXTERNAL", "SHIFTED", "RENAMED", "MOVED", "AMBIGUOUS", "CHANGED", "GONE", ) REVIEW_STATUSES = ("CHANGED", "GONE", "AMBIGUOUS") REBASE_STATUSES = ("SHIFTED", "RENAMED", "MOVED") WIDEN_STEPS = (0, 3, 6, 12, 25, 50) MAX_MATCH_LINES = 20000 PIN = "pin" HEAD = "head" _REF_RE = re.compile(r"^(?P[^:]+?)(?::(?P\d+)(?:-(?P\d+))?)?$") _BARE_RANGE_RE = re.compile(r"^:?(\d+(?:-\d+)?)$") HEADER_SUFFIXES = (".h", ".hpp", ".hh", ".hxx", ".inc", ".inl") _SPLIT_RE = re.compile(r"[,;]") _ANNOTATION_RE = re.compile(r"\s*\([^)]*\)?") _LOCATED_RE = re.compile(r"^[^\s:]+:\d+(?:-\d+)?$") @dataclass(frozen=True) class RefPart: path: str start: int | None end: int | None raw: str = "" @dataclass(frozen=True) class AnchorResult: status: str start: int | None end: int | None candidates: list[int] reason: str | None = None @dataclass(frozen=True) class PartResult: part: RefPart status: str new_path: str | None start: int | None end: int | None candidates: list[int] reason: str | None = None repo: str | None = None head_url: str | None = None @dataclass(frozen=True) class RepoView: """One repository a profile cites, at the two revisions under comparison.""" repo: object pin: str head: str origin: str field: str def parse_source_ref(ref: str) -> tuple[str, int | None, int | None]: """Split 'path/file.cpp:123-129' into (path, start, end).""" m = _REF_RE.match(ref.strip()) if not m: return ref.strip(), None, None start = int(m.group("start")) if m.group("start") else None end = int(m.group("end")) if m.group("end") else start return m.group("path"), start, end def source_ref_values(value) -> list[tuple[str, str]]: """Flatten a source_ref into (label, refs) pairs. Most entries hold a plain string. `ymir` keys its refs by mode instead, since its standalone and libretro builds read different files. """ if isinstance(value, dict): return [(str(k), str(v)) for k, v in value.items() if v] return [("", str(value))] if value else [] def split_source_ref(ref: str) -> list[RefPart]: """A source_ref may carry several references. Separators are the comma and the semicolon. A part reduced to a line or a line range continues the previous part's file: `geo.c:234-243, 273-285` cites two ranges of the same file, a form used by 430 parts across 69 profiles. Parenthetical annotations are prose and are dropped: 285 entries across 50 profiles carry them. """ parts: list[RefPart] = [] for raw in split_outside_parentheses(str(ref or "")): chunk = _trim_prose(_ANNOTATION_RE.sub("", raw).strip()) if not chunk: continue bare = _BARE_RANGE_RE.match(chunk) if bare and parts: start, _, end = bare.group(1).partition("-") parts.append( RefPart(parts[-1].path, int(start), int(end or start), raw) ) continue path, start_line, end_line = parse_source_ref(chunk) parts.append(RefPart(path, start_line, end_line, raw)) return parts def split_outside_parentheses(text: str) -> list[str]: """Split on commas and semicolons, ignoring those inside an annotation. Each chunk keeps its annotation, so a rewrite can put the prose back. """ chunks: list[str] = [] current: list[str] = [] depth = 0 for char in text: if char == "(": depth += 1 elif char == ")": depth = max(0, depth - 1) if char in ",;" and depth == 0: chunks.append("".join(current)) current = [] else: current.append(char) chunks.append("".join(current)) return [c for c in (chunk.strip() for chunk in chunks) if c] def _trim_prose(chunk: str) -> str: """Drop a trailing comment written without parentheses. `libretro.cpp:1520-1521 candidates_a1200` cites a location followed by a note. The leading token is kept only when it already reads as a reference, so `munt ROMInfo.cpp`, which names an external project, stays intact and is reported rather than silently truncated to `munt`. """ if " " not in chunk: return chunk head = chunk.split(" ", 1)[0] if _LOCATED_RE.match(head) or _BARE_RANGE_RE.match(head): return head return chunk def collect_tokens(entry: dict) -> list[str]: """Values declared by one file entry: hashes, then name as fallback.""" tokens: list[str] = [] for field in ("sha1", "md5", "crc32", "sha256", "known_hash_adler32"): val = entry.get(field) vals = val if isinstance(val, list) else [val] if val else [] for v in vals: v = str(v).lower().removeprefix("0x") if v: tokens.append(v) if not tokens: name = entry.get("name", "") if name: tokens.append(os.path.basename(name).lower()) return tokens def is_external_citation(path: str) -> bool: """True when a ref names a project rather than a path in a known repo. `munt ROMInfo.cpp` and `Nuked-SC55-CLAP rom_io.cpp` cite where a ROM is identified inside a dependency the profile does not declare, so no revision of the declared repositories can confirm or deny them. Reporting them as broken would be wrong: they are simply out of reach. """ if " " not in path: return False head = path.split(" ", 1)[0] return "/" not in head and "." not in head def _anchor_tokens(entry: dict) -> list[str]: """Values worth searching for when a cited line misses its subject. Archive entries are named `stvbios.zip` while the driver source writes the set name alone, so the stem is searched as well as the full name. """ tokens = collect_tokens(entry) name = os.path.basename(str(entry.get("name", ""))).lower() stem = name.rsplit(".", 1)[0] if stem and stem not in tokens and len(stem) > 2: tokens.append(stem) return tokens def worst_status(statuses) -> str: """Severity of an entry is the worst severity among its parts.""" worst = "ANCHORED" for status in statuses: if STATUS_ORDER.index(status) > STATUS_ORDER.index(worst): worst = status return worst def _normalize(lines: list[str]) -> list[str]: return [line.strip() for line in lines] def _find_all(haystack: list[str], needle: list[str]) -> list[int]: """Zero-based offsets where the needle sequence occurs.""" size = len(needle) if not size or size > len(haystack): return [] first = needle[0] hits = [] for i in range(len(haystack) - size + 1): if haystack[i] == first and haystack[i : i + size] == needle: hits.append(i) return hits def _map_index(opcodes, index: int) -> int | None: for tag, i1, i2, j1, j2 in opcodes: if i1 <= index < i2: return j1 + (index - i1) if tag == "equal" else j1 return None def _map_changed(pin: list[str], head: list[str], lo: int, hi: int) -> AnchorResult: """Map a pinned range onto HEAD once exact anchoring has failed.""" if len(pin) > MAX_MATCH_LINES or len(head) > MAX_MATCH_LINES: return AnchorResult( "CHANGED", None, None, [], f"file over {MAX_MATCH_LINES} lines" ) matcher = difflib.SequenceMatcher(None, pin, head, autojunk=False) opcodes = matcher.get_opcodes() if not any(tag == "equal" for tag, *_ in opcodes): return AnchorResult("GONE", None, None, []) new_lo = _map_index(opcodes, lo) new_hi = _map_index(opcodes, max(lo, hi - 1)) if new_lo is None or new_hi is None: return AnchorResult("GONE", None, None, []) return AnchorResult("CHANGED", new_lo + 1, max(new_lo, new_hi) + 1, []) def anchor_block( pin_lines: list[str], head_lines: list[str], start: int, end: int ) -> AnchorResult: """Locate the pinned line range inside the HEAD revision of a file.""" pin = _normalize(pin_lines) head = _normalize(head_lines) lo, hi = start - 1, end if lo < 0 or lo >= len(pin): return AnchorResult("GONE", None, None, []) if not any(pin[lo:hi]): return AnchorResult("CHANGED", None, None, [], "cited range is blank") candidates: list[int] = [] for pad in WIDEN_STEPS: a = max(0, lo - pad) b = min(len(pin), hi + pad) hits = _find_all(head, pin[a:b]) offset = lo - a if len(hits) == 1: new_start = hits[0] + offset + 1 new_end = new_start + (hi - lo) - 1 status = "ANCHORED" if new_start == start else "SHIFTED" return AnchorResult(status, new_start, new_end, []) if not hits: break candidates = [h + offset + 1 for h in hits] else: return AnchorResult("AMBIGUOUS", None, None, candidates) if candidates: return AnchorResult("AMBIGUOUS", None, None, candidates) return _map_changed(pin, head, lo, hi) NUDGE_WINDOW = 60 def nudge_to_declared( pin_lines: list[str], start: int, end: int, tokens ) -> tuple[int, int] | None: """Move a cited range onto the nearest line carrying a declared value. Some refs sit a few lines off their subject, and a few land on a blank line, which anchors nothing. When the entry declares a hash or a name and that value appears close by, the ref plainly meant that line. """ if not tokens: return None lowered = [line.lower() for line in pin_lines] hits = [ i + 1 for i, line in enumerate(lowered) if any(token in line for token in tokens) ] if not hits: return None nearest = min(hits, key=lambda n: abs(n - start)) if abs(nearest - start) > NUDGE_WINDOW or nearest == start: return None if sum(1 for h in hits if abs(h - nearest) <= 1) > 1: return None return nearest, nearest + (end - start) def resolve_rename( result: CompareResult, path: str, head_paths=() ) -> tuple[str | None, list[str]]: """New path of a moved file, or the candidates when several match. The comparison is authoritative when it reports the rename. Otherwise the HEAD tree is searched by basename, then by stem: a C++ file rewritten in C keeps its stem and loses its basename. """ renamed = find_renamed(result, path) if renamed: return renamed, [] pool = list(head_paths) or [change.path for change in result.files] base = posixpath.basename(path) matches = _narrow( [p for p in pool if p != path and posixpath.basename(p) == base], path ) if len(matches) == 1: return matches[0], [] if not matches: stem = base.rsplit(".", 1)[0] matches = _narrow( [ p for p in pool if p != path and posixpath.basename(p).rsplit(".", 1)[0] == stem ], path, ) if len(matches) == 1: return matches[0], [] return None, matches def _is_header(path: str) -> bool: return path.lower().endswith(HEADER_SUFFIXES) def _narrow(matches: list[str], path: str) -> list[str]: """Prefer candidates of the same kind, then of the same directory. A stem search on `ss.cpp` matches both `ss.c` and `ss.h`, and only one of them is the implementation the ref meant. """ if len(matches) < 2: return matches kind = [p for p in matches if _is_header(p) == _is_header(path)] if len(kind) == 1: return kind pool = kind or matches directory = posixpath.dirname(path) same = [p for p in pool if posixpath.dirname(p) == directory] return same if len(same) == 1 else pool def locate_by_declared_value( head_lines: list[str], anchored: AnchorResult, tokens ) -> tuple[int, int, str] | None: """Follow a ref whose subject can still be pinpointed at HEAD. Two cases carry enough evidence to move a ref without judging the edit itself: the mapped range already contains the declared value, or the value occurs exactly once in the whole file. A value occurring several times is left alone, since nothing says which one the ref meant. """ lowered = [line.lower() for line in head_lines] if anchored.start is not None: end = anchored.end or anchored.start block = "\n".join(lowered[anchored.start - 1 : end]) if any(token in block for token in tokens): return ( anchored.start, end, "content edited, declared value present in the new range", ) hits = [ i + 1 for i, line in enumerate(lowered) if any(token in line for token in tokens) ] if len(hits) == 1: return hits[0], hits[0], "declared value occurs once at HEAD" return None def anchor_part( part: RefPart, fetch, rename_getter, describe=None, tokens=() ) -> PartResult: """Locate one reference part at HEAD, following a rename when needed. `describe(path)` returns (repo slug, raw URL at HEAD) for the repository that owns the path, or (None, None) when the caller does not track it. """ if is_external_citation(part.path): return PartResult( part, "EXTERNAL", None, None, None, [], "names a project the profile does not declare", ) slug, url, actual = ( describe(part.path) if describe else (None, None, part.path) ) head_lines = fetch(HEAD, part.path) path = actual renamed = actual != part.path if head_lines is None: moved, candidates = rename_getter(part.path) if moved is None: if candidates: return PartResult( part, "AMBIGUOUS", None, None, None, [], f"{len(candidates)} candidates: {', '.join(candidates[:5])}", slug, url, ) return PartResult( part, "GONE", None, None, None, [], "absent at HEAD, no rename found", slug, url, ) path = moved head_lines = fetch(HEAD, path) if head_lines is None: return PartResult( part, "GONE", None, None, None, [], None, slug, url ) renamed = True if describe: slug, url, _ = describe(path) if part.start is None: status = "RENAMED" if renamed else "ANCHORED" return PartResult( part, status, path if renamed else None, None, None, [], None, slug, url ) pin_lines = fetch(PIN, part.path) if pin_lines is None and path != part.path: # A ref may cite a bare filename the tree search resolved at HEAD; the # same resolved path usually holds at the pin. A genuine rename keeps # the old path at the pin, which is why that one is tried first. pin_lines = fetch(PIN, path) if pin_lines is None: return PartResult( part, "GONE", None, None, None, [], "pin revision missing", slug, url ) end = part.end or part.start anchored = anchor_block(pin_lines, head_lines, part.start, end) note = anchored.reason if anchored.status in REVIEW_STATUSES: nudged = nudge_to_declared(pin_lines, part.start, end, tokens) if nudged is not None: retry = anchor_block(pin_lines, head_lines, *nudged) if retry.status not in REVIEW_STATUSES: anchored = retry note = f"cited line was {nudged[0] - part.start:+d} off its subject" status = anchored.status if status in ("CHANGED", "GONE") and tokens: located = locate_by_declared_value(head_lines, anchored, tokens) if located is not None: start, end, why = located anchored = AnchorResult("MOVED", start, end, []) status, note = "MOVED", why if renamed and status in ("ANCHORED", "SHIFTED"): status = "RENAMED" return PartResult( part, status, path if renamed else None, anchored.start, anchored.end, anchored.candidates, note, slug, url, ) SHIFT_SUPPORT = 3 def dominant_shifts(parts, support: int = SHIFT_SUPPORT) -> dict[str, int]: """Most common line shift per file, taken from the unambiguous anchors. A file whose refs mostly moved by the same amount gives strong evidence for the ones that stayed ambiguous: repetitive tables, as in the NeoGeo driver, match a cited line in several places, and only one of them sits where the rest of the file moved to. """ by_file: dict[str, dict[int, int]] = {} for part in parts: if part.status not in ("ANCHORED", "SHIFTED"): continue if part.start is None or part.part.start is None: continue counts = by_file.setdefault(part.part.path, {}) delta = part.start - part.part.start counts[delta] = counts.get(delta, 0) + 1 shifts = {} for path, counts in by_file.items(): delta, seen = max(counts.items(), key=lambda kv: kv[1]) if seen >= support: shifts[path] = delta return shifts def resolve_by_shift(part: PartResult, shifts: dict[str, int]) -> PartResult: """Settle an ambiguous part when one candidate matches the file's shift.""" if part.status != "AMBIGUOUS" or not part.candidates: return part if part.part.start is None or part.part.path not in shifts: return part wanted = part.part.start + shifts[part.part.path] if part.candidates.count(wanted) != 1: return part span = (part.part.end or part.part.start) - part.part.start return PartResult( part.part, "ANCHORED" if wanted == part.part.start else "SHIFTED", part.new_path, wanted, wanted + span, [], f"settled by the {shifts[part.part.path]:+d} line shift of its file", part.repo, part.head_url, ) @dataclass class EntryReport: name: str source_ref: str status: str parts: list[PartResult] @dataclass class ProfileReport: name: str repo: str | None = None repos: list[str] = None pinned_tag: str | None = None host: str | None = None pin: str | None = None pin_origin: str | None = None head: str | None = None entries: list[EntryReport] = None skipped: str | None = None counts: dict[str, int] = None def needs_review(self) -> int: counts = self.counts or {} return sum(counts.get(status, 0) for status in REVIEW_STATUSES) def select_repo(profile: dict) -> upstream.Repo | None: """Repository the profile was read from: source first, then upstream.""" for field in ("source", "upstream"): repo = upstream.parse_repo(str(profile.get(field) or "")) if repo is not None: return repo return None def resolve_pin( profile: dict, repo, cache_dir: str, offline: bool, field: str = "source" ) -> tuple[str | None, str | None]: """Commit the profile was written at, and how it was obtained. `field` selects which declared pin applies: `source_commit` for the port the profile was read from, `upstream_commit` for the original project. """ pinned = profile.get(f"{field}_commit") if pinned: return str(pinned), f"{field}_commit" date = str(profile.get("profiled_date") or "") if not date: return None, None sha = upstream.resolve_commit_at(repo, date, cache_dir, offline) return sha, (f"profiled_date {date}" if sha else None) def select_views( profile: dict, cache_dir: str, offline: bool ) -> list[RepoView]: """Every repository the profile cites, source before upstream. 241 profiles declare a `source` distinct from their `upstream`, and 168 of those have both on a supported forge. Their refs mix paths from the two, so a single repository cannot resolve them all. """ views: list[RepoView] = [] seen: set[tuple[str, str]] = set() for field in ("source", "upstream"): repo = upstream.parse_repo(str(profile.get(field) or "")) if repo is None or (repo.host, repo.slug) in seen: continue pin, origin = resolve_pin(profile, repo, cache_dir, offline, field) if not pin: continue head = upstream.resolve_head(repo, cache_dir, offline) if not head: continue seen.add((repo.host, repo.slug)) views.append(RepoView(repo, pin, head, origin, field)) return views SELF_CHECK_CONTEXT = 2 def verify_at_pin(part: RefPart, pin_lines, tokens) -> PartResult: """Check a ref against its own revision instead of against HEAD. A profile pinned to a superseded tag documents a program HEAD no longer contains, so comparing the two says nothing. What can still be checked is self-consistency: does the cited range carry the value the entry declares? """ if pin_lines is None: return PartResult( part, "GONE", None, None, None, [], "absent at the pinned revision" ) if part.start is None: return PartResult(part, "ANCHORED", None, None, None, []) if part.start > len(pin_lines): return PartResult( part, "GONE", None, None, None, [], "beyond the end of the file" ) if not tokens: return PartResult(part, "ANCHORED", None, None, None, []) lo = max(0, part.start - 1 - SELF_CHECK_CONTEXT) hi = min(len(pin_lines), (part.end or part.start) + SELF_CHECK_CONTEXT) window = "\n".join(pin_lines[lo:hi]).lower() if any(token in window for token in tokens): return PartResult(part, "ANCHORED", None, None, None, []) return PartResult( part, "CHANGED", None, None, None, [], "declared value not found at the pinned revision", ) def version_tag_candidates(core_version: str) -> list[str]: """Tag spellings a declared core_version might use.""" version = str(core_version or "").strip() if not version or " " in version: return [] bare = version.lstrip("vV") return list(dict.fromkeys([version, f"v{bare}", bare])) def detect_pinned_tag( profile: dict, repo, pin: str, head: str, cache_dir: str, offline: bool ) -> str | None: """Tag the profile is pinned to, when the repository has moved past it. A profile can document a frozen release line that lives as a tag inside a still-developed repository. HEAD is then a different program: the refs describe the tag and must never be recaled onto modern code. `pcsx2-legacy` is pinned to v1.6.0 of PCSX2, where a plugin-era DEV9 ref resolves onto the modern built-in DEV9, a different code path. The declared `core_version` names the tag, which is looked up by name so that a repository publishing nightly tags cannot hide an old release behind pagination. """ if pin == head: return None for tag in version_tag_candidates(profile.get("core_version")): if upstream.tag_commit(repo, tag, cache_dir, offline) == pin: return tag return None def build_report( name: str, profile: dict, cache_dir: str, offline: bool = False ) -> ProfileReport: """Confront one profile with its upstream.""" report = ProfileReport(name=name, entries=[], counts={}) refs = [ ( entry.get("name", "") + (f" [{label}]" if label else ""), value, _anchor_tokens(entry), ) for entry in (profile.get("files") or []) if isinstance(entry, dict) and entry.get("source_ref") for label, value in source_ref_values(entry.get("source_ref")) ] if select_repo(profile) is None: declared = str(profile.get("source") or profile.get("upstream") or "") report.skipped = f"unsupported host: {declared or 'none declared'}" return report views = select_views(profile, cache_dir, offline) if not views: report.skipped = ( "no source_commit and no resolvable profiled_date " "on any declared repository" ) return report primary = views[0] report.repo, report.host = primary.repo.slug, primary.repo.host report.repos = [v.repo.slug for v in views] report.pin, report.pin_origin = primary.pin, primary.origin report.head = primary.head report.pinned_tag = detect_pinned_tag( profile, primary.repo, primary.pin, primary.head, cache_dir, offline ) # A profile carrying no source_ref still has a pin worth writing and a # version worth checking, so the revisions above are resolved first. if not refs: report.skipped = "no source_ref" return report owners: dict[str, RepoView] = {} context: dict[str, object] = {} def _locate(path: str) -> tuple[RepoView, str]: """Repository and real path carrying a cited path, HEAD before pin. A ref may prefix the path with the repository directory name, as it appears in a parent folder holding both clones: 270 parts across 21 profiles do. That prefix is stripped only as a last resort, once the path as written has failed against every repository and revision. """ candidates = [path] head, _, tail = path.partition("/") if tail and any(head == v.repo.name for v in views): candidates.append(tail) for candidate in candidates: for sha_of in (lambda v: v.head, lambda v: v.pin): for view in views: found = upstream.fetch_file( view.repo, sha_of(view), candidate, cache_dir, offline ) if found is not None: return view, candidate return primary, path def resolve_path(path: str) -> tuple[RepoView, str]: if path not in owners: owners[path] = _locate(path) return owners[path] def _context_for(view: RepoView): key = view.repo.slug if key not in context: comparison = upstream.compare( view.repo, view.pin, view.head, cache_dir, offline ) tree, truncated = upstream.list_tree( view.repo, view.head, cache_dir, offline ) context[key] = (comparison, tree) if truncated and tree: print( f"{name}: {key} HEAD tree truncated by the forge, " "rename search is partial", file=sys.stderr, ) return context[key] def rename_getter(path: str) -> tuple[str | None, list[str]]: """Resolved only when a cited file has vanished, then memoised. Every declared repository is searched, not just the one that owns the path: a profile citing both a port and its upstream may point at a file that only the other one carries. """ owner, actual = resolve_path(path) ordered = [owner] + [v for v in views if v is not owner] candidates: list[str] = [] for view in ordered: comparison, tree = _context_for(view) found, near = resolve_rename(comparison, actual, tree) if found: return found, [] # Only the owning repository reports near misses: pooling them # across repositories turns a resolvable path into an ambiguous one. if view is owner: candidates = near return None, candidates def fetch(which: str, path: str): view, actual = resolve_path(path) sha = view.pin if which == PIN else view.head return upstream.fetch_file(view.repo, sha, actual, cache_dir, offline) def describe(path: str) -> tuple[str | None, str | None, str]: view, actual = resolve_path(path) slug = view.repo.slug if view is not primary else None return slug, upstream.raw_url(view.repo, view.head, actual), actual if report.pinned_tag: staged = [ (entry_name, ref, [ verify_at_pin(part, fetch(PIN, part.path), tokens) for part in split_source_ref(ref) ]) for entry_name, ref, tokens in refs ] else: staged = [ (entry_name, ref, [ anchor_part(part, fetch, rename_getter, describe, tokens) for part in split_source_ref(ref) ]) for entry_name, ref, tokens in refs ] shifts = dominant_shifts([p for _, _, parts in staged for p in parts]) for entry_name, ref, parts in staged: parts = [resolve_by_shift(p, shifts) for p in parts] status = worst_status([p.status for p in parts]) report.entries.append(EntryReport(entry_name, ref, status, parts)) report.counts[status] = report.counts.get(status, 0) + 1 return report def _render_span(path: str, start: int | None, end: int | None) -> str: if start is None: return path if end and end != start: return f"{path}:{start}-{end}" return f"{path}:{start}" def _original_part(part: PartResult) -> str: """The part exactly as the profile writes it today.""" return part.part.raw or _render_span( part.part.path, part.part.start, part.part.end ) def _rendered_part(part: PartResult) -> str: """The part rewritten onto its new location, annotation preserved.""" span = _render_span(part.new_path or part.part.path, part.start, part.end) raw = part.part.raw if not raw: return span located = _trim_prose(_ANNOTATION_RE.sub("", raw).strip()) if located and located in raw: return raw.replace(located, span, 1) return span def _part_line(part: PartResult) -> str: detail = part.status if part.new_path: detail += f" -> {part.new_path}" if part.start is not None and part.status != "ANCHORED": span = f"{part.start}" if part.end and part.end != part.start: span += f"-{part.end}" detail += f" -> {span}" if part.candidates: detail += f" candidats: {', '.join(str(c) for c in part.candidates)}" if part.reason: detail += f" ({part.reason})" if part.repo: detail += f" [{part.repo}]" return f" {_original_part(part)} {detail}" def format_report(report: ProfileReport, changed_only: bool = False) -> str: """Human-readable report for one profile.""" slugs = " + ".join(report.repos) if report.repos else (report.repo or "") lines = [f"{report.name} {slugs}".rstrip()] if report.pin and report.head: lines.append( f" {report.pin[:7]} -> {report.head[:7]} pin: {report.pin_origin}" ) if report.pinned_tag: lines.append( f" pinned to tag {report.pinned_tag}: checked against its own " "revision, not against HEAD" ) if report.skipped: lines.append(f" skipped: {report.skipped}") return "\n".join(lines) for entry in report.entries: if changed_only and entry.status not in REVIEW_STATUSES: continue lines.append(f" {entry.name} {entry.source_ref}") for part in entry.parts: lines.append(_part_line(part)) summary = ", ".join( f"{count} {status.lower()}" for status, count in sorted(report.counts.items()) ) lines.append(f" {len(report.entries)} refs: {summary}") lines.append(f" {report.needs_review()} demandent une relecture") return "\n".join(lines) def report_to_dict(report: ProfileReport) -> dict: """Serialisable form of one report.""" payload = asdict(report) payload["needs_review"] = report.needs_review() return payload def format_markdown(reports: list[ProfileReport]) -> str: """Report set as a markdown document.""" lines = [ "# profile-sync", "", "| profil | depot | refs | relecture |", "|---|---|---|---|", ] for report in reports: if report.skipped: lines.append( f"| {report.name} | {report.repo or ''} | skipped | {report.skipped} |" ) continue lines.append( f"| {report.name} | {report.repo} | {len(report.entries)} " f"| {report.needs_review()} |" ) for report in reports: lines.extend(["", "```", format_report(report), "```"]) return "\n".join(lines) def fetch_plan(report: ProfileReport) -> list[str]: """Raw URLs at HEAD for the refs that need the code read again.""" if report.skipped or not report.head or not report.host: return [] owner, _, name = (report.repo or "").partition("/") fallback = upstream.make_repo(report.host, owner, name) urls = [] for entry in report.entries: if entry.status not in REVIEW_STATUSES: continue for part in entry.parts: url = part.head_url if url is None: if fallback is None: continue path = part.new_path or part.part.path url = upstream.raw_url(fallback, report.head, path) if part.part.start: url += f"#L{part.part.start}" urls.append(url) return urls @dataclass(frozen=True) class VersionReport: declared: str latest_tag: str | None latest_release: str | None release_date: str | None tag_matches_declared: bool tag_commit: str | None def check_version( profile: dict, repo, cache_dir: str, offline: bool ) -> VersionReport | None: """Declared core_version against the latest upstream tag and release.""" declared = str(profile.get("core_version") or "") if not declared: return None tags = upstream.list_tags(repo, cache_dir, offline) release = upstream.latest_release(repo, cache_dir, offline) matches = declared in tags commit = ( upstream.resolve_tag_commit(repo, declared, cache_dir, offline) if matches else None ) return VersionReport( declared, tags[0] if tags else None, release.tag if release else None, release.date if release else None, matches, commit, ) def version_warning(report: VersionReport | None, pin: str | None) -> str | None: """Flag a declared version that names a tag away from the resolved pin.""" if report is None or not report.tag_commit or not pin: return None if report.tag_commit == pin: return None return ( f"core_version {report.declared} names tag commit {report.tag_commit[:7]}, " f"the pin resolved to {pin[:7]}" ) FILE_EXTENSIONS = ( "bin", "rom", "zip", "dat", "bios", "img", "chd", "nvram", "sav", "ips", "pal", "fnt", ) _FILE_RE = re.compile( r"[\w./+-]+\.(?:" + "|".join(FILE_EXTENSIONS) + r")\b", re.IGNORECASE ) _SHA1_RE = re.compile(r"\b[0-9a-fA-F]{40}\b") _MD5_RE = re.compile(r"\b[0-9a-fA-F]{32}\b") _CRC_RE = re.compile(r"\b(?:0x)?([0-9a-fA-F]{8})\b") def declared_names(profile: dict) -> set[str]: """Every filename the profile covers, aliases included.""" names = set() for entry in profile.get("files") or []: if not isinstance(entry, dict): continue for value in [entry.get("name", "")] + list(entry.get("aliases") or []): if value: names.add(os.path.basename(str(value)).casefold()) return names def declared_hashes(profile: dict) -> set[str]: """Every hash the profile declares.""" hashes = set() for entry in profile.get("files") or []: if isinstance(entry, dict): hashes.update(collect_tokens(entry)) return hashes def detect_new_files( head_lines: list[str], known: set[str], cap: int = 25 ) -> tuple[list[tuple[int, str]], int]: """Filename literals at HEAD the profile does not declare.""" found: list[tuple[int, str]] = [] seen: set[str] = set() elided = 0 for number, line in enumerate(head_lines, start=1): for match in _FILE_RE.findall(line): name = os.path.basename(match).casefold() if name in known or name in seen: continue seen.add(name) if len(found) >= cap: elided += 1 continue found.append((number, os.path.basename(match))) return found, elided def watch_hashes(added_lines: list[str], known: set[str]) -> list[tuple[str, str]]: """Hash literals in added lines that no profile entry declares.""" found: list[tuple[str, str]] = [] for line in added_lines: strong = bool(_SHA1_RE.search(line) or _MD5_RE.search(line)) for pattern, kind in ((_SHA1_RE, "sha1"), (_MD5_RE, "md5")): for value in pattern.findall(line): if value.lower() not in known: found.append((kind, value.lower())) if strong or not _FILE_RE.search(line): continue for value in _CRC_RE.findall(line): if value.lower() not in known: found.append(("crc32", value.lower())) return found def unified_for_path( pin_lines: list[str], head_lines: list[str], path: str, context: int ) -> str: """Unified diff of one cited file between the two revisions.""" diff = difflib.unified_diff( pin_lines, head_lines, fromfile=f"a/{path}", tofile=f"b/{path}", n=context, lineterm="", ) return "\n".join(diff) def tree_diff(result: CompareResult, ref_dirs: set[str]) -> list[str]: """Tree changes restricted to the directories the refs point at.""" lines = [] for change in result.files: directory = posixpath.dirname(change.path) if ref_dirs and not any( directory == d or directory.startswith(d + "/") for d in ref_dirs ): continue if change.status == "renamed" and change.previous_path: lines.append(f"renamed {change.previous_path} -> {change.path}") else: lines.append(f"{change.status} {change.path}") if result.truncated: lines.append("comparison truncated by the forge, list is partial") return lines DRIFT_WEIGHTS = {"GONE": 40, "CHANGED": 30, "AMBIGUOUS": 15, "SHIFTED": 1} def drift_score( report: ProfileReport, version: VersionReport | None, commits: int ) -> int: """Rank profiles by how far they have drifted from their pin.""" counts = report.counts or {} score = sum(DRIFT_WEIGHTS.get(k, 0) * v for k, v in counts.items()) if version is not None and not version.tag_matches_declared: if version.latest_release and version.latest_release != version.declared: score += 10 score += min(commits, 20) // 4 return score class YamlWriteError(Exception): """A profile edit did not land exactly as intended.""" def _top_level_indices(lines: list[str], key: str) -> list[int]: """Indices of lines declaring a key at the top level of the document.""" prefix = f"{key}:" return [i for i, line in enumerate(lines) if line.startswith(prefix)] def insert_after_line(text: str, key: str, new_line: str) -> str: """Insert a line right after a top-level key.""" lines = text.splitlines() indices = _top_level_indices(lines, key) if len(indices) != 1: raise YamlWriteError( f"{key}: expected one top-level line, found {len(indices)}" ) lines.insert(indices[0] + 1, new_line) return "\n".join(lines) + ("\n" if text.endswith("\n") else "") def _field_value(line: str) -> str: return line.split(":", 1)[1].strip().strip('"').strip("'") def find_field_line(text: str, key: str, expected: str, start: int = 0) -> int: """Index of the first `key:` line at or after `start` holding `expected`. Positional counting is not safe here: `source_ref` also appears outside `files:` in some profiles, under `data_directories:` for instance, which shifts every ordinal past it. """ lines = text.splitlines() marker = f"{key}:" for index in range(start, len(lines)): line = lines[index] if line.strip().startswith(marker) and _field_value(line) == expected: return index raise YamlWriteError( f"{key}: no line holding {expected!r} at or after line {start + 1}" ) def replace_field_line(text: str, key: str, expected: str, value: str, start: int = 0) -> tuple[str, int]: """Rewrite the first `key:` line holding `expected`. Returns text and index.""" index = find_field_line(text, key, expected, start) lines = text.splitlines() line = lines[index] indent = line[: len(line) - len(line.lstrip())] lines[index] = f'{indent}{key}: "{value}"' joined = "\n".join(lines) + ("\n" if text.endswith("\n") else "") return joined, index def apply_edit(path: Path, new_text: str, expected: dict) -> None: """Write, then verify the parsed document matches what was intended.""" original = path.read_text(encoding="utf-8") path.write_text(new_text, encoding="utf-8") try: written = yaml.safe_load(new_text) except yaml.YAMLError as exc: path.write_text(original, encoding="utf-8") raise YamlWriteError(f"{path}: parse failed after write: {exc}") from exc if written != expected: path.write_text(original, encoding="utf-8") raise YamlWriteError(f"{path}: structure changed beyond the intended field") def backfill_commit(path: Path, sha: str) -> bool: """Insert source_commit after profiled_date. False when already present.""" text = path.read_text(encoding="utf-8") document = yaml.safe_load(text) if document.get("source_commit"): return False new_text = insert_after_line(text, "profiled_date", f'source_commit: "{sha}"') expected = dict(document) expected["source_commit"] = sha apply_edit(path, new_text, expected) return True def _is_rewritable(ref: str) -> bool: """Every ref whose parts can be put back with their prose intact.""" parts = split_source_ref(ref) if not parts: return False return ", ".join(p.raw for p in parts) == ", ".join( chunk for chunk in split_outside_parentheses(ref) ) def rebase_refs( path: Path, report: ProfileReport, accept_changed: bool = False ) -> list[str]: """Recale the line ranges of parts whose content is unchanged. All or nothing per profile. `source_commit` names the revision the refs are written against, so moving some refs to HEAD while others still describe the pinned revision would leave the profile self-contradictory, and the next comparison would read the pinned revision at line numbers that only make sense at HEAD. """ if report.pinned_tag: return [] statuses = REBASE_STATUSES + (("CHANGED",) if accept_changed else ()) blocking = [s for s in REVIEW_STATUSES if s not in statuses] if any((report.counts or {}).get(s) for s in blocking): return [] text = path.read_text(encoding="utf-8") document = yaml.safe_load(text) carriers = [ entry for entry in (document.get("files") or []) if isinstance(entry, dict) and entry.get("source_ref") ] applied: list[str] = [] cursor = 0 for position, entry in enumerate(report.entries): if not _is_rewritable(entry.source_ref) or entry.name.endswith("]"): # Rewriting would drop the author's annotations, or the refs live # under a mode key rather than on a source_ref line. continue rendered = [] touched = False for part in entry.parts: if part.status in statuses and part.start is not None: recaled = _rendered_part(part) if recaled != _original_part(part): touched = True rendered.append(recaled) else: rendered.append(_original_part(part)) if not touched: # Still consume this entry's line so a later entry carrying the # same ref cannot match it. cursor = find_field_line(text, "source_ref", entry.source_ref, cursor) + 1 continue new_ref = ", ".join(rendered) text, index = replace_field_line( text, "source_ref", entry.source_ref, new_ref, cursor ) cursor = index + 1 carriers[position]["source_ref"] = new_ref applied.append(f"{entry.source_ref} -> {new_ref}") if applied: apply_edit(path, text, document) return applied def pending_recale(report: ProfileReport, accept_changed: bool = False) -> int: """Parts that ought to move but sit in a ref the writer will not touch. An annotated ref cannot be regenerated without losing its prose, so its parts stay on the pinned line numbers. Advancing the pin while they do would leave the profile describing two revisions at once. """ movable = REBASE_STATUSES + (("CHANGED",) if accept_changed else ()) pending = 0 for entry in report.entries or []: if _is_rewritable(entry.source_ref) and not entry.name.endswith("]"): continue pending += sum( 1 for part in entry.parts if part.status in movable and part.start is not None and _rendered_part(part) != _original_part(part) ) return pending def bump_commit( path: Path, report: ProfileReport, accept_changed: bool = False ) -> bool: """Advance source_commit to HEAD when nothing needs a read again.""" if report.skipped or not report.head or report.pinned_tag: return False blocking = REVIEW_STATUSES if not accept_changed else ( s for s in REVIEW_STATUSES if s != "CHANGED" ) if any((report.counts or {}).get(s) for s in blocking): return False if pending_recale(report, accept_changed): return False text = path.read_text(encoding="utf-8") document = yaml.safe_load(text) expected = dict(document) expected["source_commit"] = report.head if document.get("source_commit"): new_text, _ = replace_field_line( text, "source_commit", str(document["source_commit"]), report.head ) else: new_text = insert_after_line( text, "profiled_date", f'source_commit: "{report.head}"' ) apply_edit(path, new_text, expected) return True def emulators_dir_is_dirty(emulators_dir: str) -> bool: """True when the profile directory carries uncommitted changes.""" result = subprocess.run( ["git", "status", "--porcelain", "--", emulators_dir], capture_output=True, text=True, check=False, ) return bool(result.stdout.strip()) def build_parser() -> argparse.ArgumentParser: """Command line: selection, output, detection, writes, network.""" parser = argparse.ArgumentParser( description="Confront emulator profiles with their upstream source" ) group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--emulator", help="a single profile") group.add_argument("--all", action="store_true", help="every profile") parser.add_argument("--emulators-dir", default="emulators") parser.add_argument("--system", help="keep profiles covering this system") parser.add_argument("--type", dest="type_", help="keep profiles of this type") parser.add_argument("--stale-before", help="keep profiles profiled before a date") parser.add_argument("--limit", type=int, help="stop after N profiles") parser.add_argument("--changed-only", action="store_true") parser.add_argument("--json", action="store_true", dest="as_json") parser.add_argument("--markdown", action="store_true") parser.add_argument("--fetch-plan", action="store_true") parser.add_argument("--full-diff", action="store_true") parser.add_argument("--ref", help="restrict --full-diff to one cited path") parser.add_argument("--context", type=int, default=3) parser.add_argument("--check-version", action="store_true") parser.add_argument("--detect-new-files", action="store_true") parser.add_argument("--watch-hashes", action="store_true") parser.add_argument("--tree-diff", action="store_true") parser.add_argument("--triage", action="store_true") parser.add_argument("--backfill-commits", action="store_true") parser.add_argument("--rebase-refs", action="store_true") parser.add_argument("--bump-commit", action="store_true") parser.add_argument( "--accept-changed", action="store_true", help="recale CHANGED refs too, after their diff has been read", ) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--force", action="store_true") parser.add_argument("--offline", action="store_true") parser.add_argument("--cache-dir", default=DEFAULT_CACHE) return parser def select_profiles(profiles: dict, args) -> dict: """Profiles matching the selection flags.""" if args.emulator: if args.emulator not in profiles: print(f"unknown profile: {args.emulator}", file=sys.stderr) raise SystemExit(1) return {args.emulator: profiles[args.emulator]} selected = {} for name, profile in sorted(profiles.items()): if profile.get("type") in ("alias", "test"): continue if args.system and args.system not in (profile.get("systems") or []): continue if args.type_ and profile.get("type") != args.type_: continue if args.stale_before: profiled = str(profile.get("profiled_date") or "") if not profiled or profiled >= args.stale_before: continue selected[name] = profile if args.limit and len(selected) >= args.limit: break return selected def _check_quota(count: int, offline: bool) -> None: """Refuse a run the anonymous quota cannot carry to completion.""" if offline or os.environ.get("GITHUB_TOKEN"): return if count * 4 > ANON_QUOTA: print( f"{count} profiles need roughly {count * 4} API calls, the anonymous " f"quota is {ANON_QUOTA} per hour. Set GITHUB_TOKEN or use --offline.", file=sys.stderr, ) raise SystemExit(1) def _apply_writes(args, name: str, profile: dict, report: ProfileReport) -> None: path = Path(args.emulators_dir) / f"{name}.yml" if not path.is_file(): return if args.backfill_commits and report.pin and not profile.get("source_commit"): if args.dry_run: print(f"{name}: would write source_commit {report.pin[:7]}") elif backfill_commit(path, report.pin): print(f"{name}: source_commit {report.pin[:7]}") if args.rebase_refs and not args.dry_run and not report.skipped: for line in rebase_refs(path, report, args.accept_changed): print(f"{name}: {line}") if ( args.bump_commit and not args.dry_run and bump_commit(path, report, args.accept_changed) ): print(f"{name}: source_commit -> {report.head[:7]}") def _cited_paths(report: ProfileReport) -> list[str]: return sorted( {part.new_path or part.part.path for e in report.entries for part in e.parts} ) def _print_version(args, profile: dict, report: ProfileReport, repo) -> None: version = check_version(profile, repo, args.cache_dir, args.offline) if version is None: return print( f" version: declared {version.declared}, " f"latest tag {version.latest_tag}, " f"latest release {version.latest_release} ({version.release_date})" ) warning = version_warning(version, report.pin) if warning: print(f" {warning}") def _print_detection(args, profile: dict, report: ProfileReport, repo) -> None: names, hashes = declared_names(profile), declared_hashes(profile) for path in _cited_paths(report): head_lines = upstream.fetch_file( repo, report.head, path, args.cache_dir, args.offline ) if head_lines is None: continue if args.detect_new_files: found, elided = detect_new_files(head_lines, names) for number, name in found: print(f" new file: {path}:{number} {name}") if elided: print(f" new file: {elided} further matches not shown") if args.watch_hashes: pin_lines = ( upstream.fetch_file( repo, report.pin, path, args.cache_dir, args.offline ) or [] ) added = [ line for line in difflib.unified_diff(pin_lines, head_lines, n=0) if line.startswith("+") and not line.startswith("+++") ] for kind, value in watch_hashes(added, hashes): print(f" new {kind}: {path} {value}") def _print_extras(args, profile: dict, report: ProfileReport) -> None: """Optional per-profile sections beyond the ref report.""" wants = ( args.check_version or args.detect_new_files or args.watch_hashes or args.full_diff or args.tree_diff ) if not wants or not report.head: return repo = select_repo(profile) if repo is None: return if args.check_version: _print_version(args, profile, report, repo) if args.detect_new_files or args.watch_hashes: _print_detection(args, profile, report, repo) if args.full_diff: for path in _cited_paths(report): if args.ref and args.ref not in path: continue pin_lines = ( upstream.fetch_file( repo, report.pin, path, args.cache_dir, args.offline ) or [] ) head_lines = ( upstream.fetch_file( repo, report.head, path, args.cache_dir, args.offline ) or [] ) text = unified_for_path(pin_lines, head_lines, path, args.context) if text: print(text) if args.tree_diff: comparison = upstream.compare( repo, report.pin, report.head, args.cache_dir, args.offline ) # The repository root is a directory like any other: dropping the # empty dirname would turn the filter off and list the whole tree. ref_dirs = {posixpath.dirname(p) for p in _cited_paths(report)} for line in tree_diff(comparison, ref_dirs): print(f" {line}") def _print_triage(args, selected: dict, reports: list[ProfileReport]) -> None: """Profiles ranked by drift, worst first.""" ranked = [] for report in reports: profile = selected[report.name] version = None commits = 0 repo = select_repo(profile) if report.head and repo is not None: version = check_version(profile, repo, args.cache_dir, args.offline) paths = _cited_paths(report) for path in paths[:TRIAGE_PATH_SAMPLE]: commits += upstream.commits_touching( repo, path, report.pin, args.cache_dir, args.offline ) if len(paths) > TRIAGE_PATH_SAMPLE: print( f"{report.name}: commit count sampled on " f"{TRIAGE_PATH_SAMPLE} of {len(paths)} paths", file=sys.stderr, ) ranked.append((drift_score(report, version, commits), report)) for score, report in sorted(ranked, key=lambda item: -item[0]): state = report.skipped or f"{report.needs_review()} to review" print(f"{score:5d} {report.name:30s} {state}") def main() -> None: """Build one report per selected profile, then apply writes and output.""" args = build_parser().parse_args() if args.accept_changed and not args.emulator: print( "--accept-changed applies to one profile at a time: its diff has " "to be read before its CHANGED refs can be recaled.", file=sys.stderr, ) raise SystemExit(1) profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False) selected = select_profiles(profiles, args) _check_quota(len(selected), args.offline) writes = args.backfill_commits or args.rebase_refs or args.bump_commit if ( writes and not args.dry_run and not args.force and emulators_dir_is_dirty(args.emulators_dir) ): print( f"{args.emulators_dir} carries uncommitted changes. " "Commit them first or pass --force.", file=sys.stderr, ) raise SystemExit(1) reports = [] for name, profile in selected.items(): try: report = build_report(name, profile, args.cache_dir, args.offline) except upstream.RateLimitError: raise except upstream.UpstreamError as exc: # One unreachable forge must not abandon the other profiles. print(f"{name}: {exc}", file=sys.stderr) report = ProfileReport( name=name, entries=[], counts={}, skipped=f"upstream error: {exc}" ) reports.append(report) if writes: try: _apply_writes(args, name, profile, report) except YamlWriteError as exc: # The guard already left the file untouched; keep going. print(f"{name}: write refused: {exc}", file=sys.stderr) if args.as_json: print(json.dumps([report_to_dict(r) for r in reports], indent=2)) return if args.markdown: target = Path("claudedocs") / f"profile-sync-{date.today().isoformat()}.md" target.parent.mkdir(parents=True, exist_ok=True) target.write_text(format_markdown(reports), encoding="utf-8") print(f"written: {target}") return if args.fetch_plan: for report in reports: for url in fetch_plan(report): print(url) return if args.triage: _print_triage(args, selected, reports) return printed = 0 for report in reports: if args.changed_only and not report.needs_review(): continue printed += 1 print(format_report(report, args.changed_only)) _print_extras(args, selected[report.name], report) print() if args.changed_only: _print_elided(reports, printed) def _print_elided(reports: list[ProfileReport], printed: int) -> None: """Account for what --changed-only left out, grouped by reason.""" reasons: dict[str, int] = {} for report in reports: if report.needs_review(): continue key = report.skipped or "nothing to review" reasons[key] = reasons.get(key, 0) + 1 if not reasons: return total = sum(reasons.values()) print(f"{printed} profils listes, {total} non listes:") for reason, count in sorted(reasons.items(), key=lambda item: -item[1]): print(f" {count:4d} {reason}") if __name__ == "__main__": main()