chemenu/tools/chemenu/corpus_diff.py

"""Compare two revisions of `kb/` on the invariants a content migration must
not change.

**Why this is not `lint`.** `lint` asks whether the corpus is currently
consistent: does every reference resolve, is every page schema-valid. It reads
one revision and cannot, even in principle, notice that something *went
missing* - a page that used to cite a source and no longer does is perfectly
consistent. That is the failure mode of a bulk rewrite, and it needs a
comparison against where the corpus came from.

The check is modelled on the one the German translation ran by hand across 248
pages. It found four defects: a dropped citation that silently unsourced a
claim, a dropped wikilink, an invented one, and a translated H1. **Three of the
four had unchanged link/cite *sets* and only changed counts**, which is why
every multiset here is a `Counter` and never a `set` - and why
`kb_scan.extract_wikilinks` (a set, correct for `lint`) must not be used.

What is deliberately *not* compared: the prose. A migration is expected to
rewrite bodies; flagging that would make the tool useless. Only the structural
skeleton is held fixed - plus one bit in the opposite direction, `body_changed`,
so a unit that silently did nothing is visible too.
"""
from __future__ import annotations

from collections import Counter
from dataclasses import dataclass, field
from typing import Any, Optional

from chemenu import blocks, kb_scan, provenance
from chemenu.page import Page
from chemenu.type_resolver import resolver

# Frontmatter fields compared by value on every page. Page-reference arrays are
# added per page from the type-spec's own `page_ref_fields:`, so a new type
# needs no change here.
#
# `modified:` and `summary:` are deliberately absent: a migration is supposed to
# bump the one and rewrite the other. `date:` is present because it is the raw
# material's publication date, which nothing may move (see `touch`).
STRUCTURAL_FIELDS = (
    "type",
    "created",
    "date",
    "provenance",
    "source_type",
    "source_language",
    "raw_files",
)


@dataclass(frozen=True)
class PageShape:
    """Everything about a page that a content migration must preserve."""

    title: str
    path: str
    h1: Optional[str]
    wikilinks: Counter
    cite_refs: Counter
    cite_defs: dict[str, str]
    fields: dict[str, Any]
    markers: dict[str, int]
    body: str

    @classmethod
    def of(cls, page: Page) -> "PageShape":
        # Cite references are counted on the body *without* the footnote block:
        # a definition line contains its own `[^id]`, so counting the raw body
        # would double every citation and mask a dropped one. This mirrors what
        # every other caller of CITE_REF_RE does (see provenance.py).
        head, definitions = provenance.split_cite_block(page.body)
        fields = {name: page.frontmatter.get(name) for name in STRUCTURAL_FIELDS}
        for name in _page_ref_fields(page):
            fields[name] = page.frontmatter.get(name)
        subtype_field = _subtype_field(page)
        if subtype_field:
            fields[subtype_field] = page.frontmatter.get(subtype_field)
        return cls(
            title=page.title,
            path=str(page.path),
            h1=page.h1_title,
            wikilinks=kb_scan.count_wikilinks(head),
            cite_refs=Counter(m.group(1) for m in provenance.CITE_REF_RE.finditer(head)),
            cite_defs={cite_id: source for cite_id, (source, _) in definitions.items()},
            fields=fields,
            markers=blocks.marker_pairs(page.body),
            body=page.body,
        )


def _page_ref_fields(page: Page) -> list[str]:
    raw = page.frontmatter.get("type")
    if not raw:
        return []
    try:
        return resolver.get_page_ref_fields(raw, page.path)
    except (ValueError, KeyError):
        return []


def _subtype_field(page: Page) -> Optional[str]:
    raw = page.frontmatter.get("type")
    if not raw:
        return None
    try:
        return resolver.get_subtype_field(raw, page.path)
    except (ValueError, KeyError):
        return None


@dataclass
class PageFinding:
    path: str  # the page's title (`compare()`'s key), kept named `path` for JSON stability
    kind: str  # "h1" | "wikilinks" | "cite-refs" | "cite-defs" | "frontmatter" | "unchanged"
    detail: str

    def __str__(self) -> str:  # noqa: D105 - report line
        return f"{self.path}: {self.kind} - {self.detail}"


@dataclass
class CorpusDiff:
    findings: list[PageFinding] = field(default_factory=list)
    added: list[str] = field(default_factory=list)
    removed: list[str] = field(default_factory=list)
    moved: list[tuple[str, str, str]] = field(default_factory=list)
    compared: int = 0

    @property
    def ok(self) -> bool:
        """Added and removed pages are reported but are not failures: creating
        or retiring a page is a legitimate thing for a migration to do, and
        `lint` already checks that nothing dangles afterwards. A changed
        invariant on a page that exists in both revisions is the failure."""
        return not self.findings


def _counter_delta(before: Counter, after: Counter) -> str:
    """A readable description of how two multisets differ, counts included."""
    parts = []
    for key in sorted(set(before) | set(after)):
        was, now = before.get(key, 0), after.get(key, 0)
        if was != now:
            parts.append(f"{key!r} {was}->{now}")
    return ", ".join(parts)


def compare_page(title: str, before: PageShape, after: PageShape) -> list[PageFinding]:
    findings: list[PageFinding] = []

    if before.h1 != after.h1:
        findings.append(
            PageFinding(title, "h1", f"{before.h1!r} -> {after.h1!r} (the title is the page's only identifier)")
        )

    if before.wikilinks != after.wikilinks:
        findings.append(PageFinding(title, "wikilinks", _counter_delta(before.wikilinks, after.wikilinks)))

    if before.cite_refs != after.cite_refs:
        findings.append(PageFinding(title, "cite-refs", _counter_delta(before.cite_refs, after.cite_refs)))

    if before.cite_defs != after.cite_defs:
        changed = []
        for cite_id in sorted(set(before.cite_defs) | set(after.cite_defs)):
            was, now = before.cite_defs.get(cite_id), after.cite_defs.get(cite_id)
            if was != now:
                changed.append(f"[^{cite_id}] {was!r} -> {now!r}")
        findings.append(PageFinding(title, "cite-defs", ", ".join(changed)))

    # A generated region that lost or gained a marker is the failure mode the
    # delimiters were introduced against, and it is silent: a lost opening
    # marker turns the region into ordinary prose, and the next write appends a
    # second region beside it. An agent rewriting prose at the boundary is
    # exactly how that happens, which is what makes it a migration invariant
    # rather than a lint nicety.
    #
    # Counts, not presence - the same reasoning as the wikilink counter. A page
    # that goes from one links region to two has the same *set* of region names.
    if before.markers != after.markers:
        changed_regions = []
        for name in sorted(set(before.markers) | set(after.markers)):
            was, now = before.markers.get(name, 0), after.markers.get(name, 0)
            if was != now:
                changed_regions.append(f"{name}: {was} -> {now}")
        findings.append(PageFinding(title, "markers", ", ".join(changed_regions)))

    changed_fields = []
    for name in sorted(set(before.fields) | set(after.fields)):
        was, now = before.fields.get(name), after.fields.get(name)
        if was != now:
            changed_fields.append(f"{name}: {was!r} -> {now!r}")
    if changed_fields:
        findings.append(PageFinding(title, "frontmatter", "; ".join(changed_fields)))

    return findings


def compare(
    before: dict[str, PageShape],
    after: dict[str, PageShape],
    expect_body_change: bool = False,
) -> CorpusDiff:
    """Compare two revisions' page shapes, keyed by page title (the wiki's
    only identity for a page - see AGENTS.md invariant 2), not by path. A page
    that only changed directory therefore compares as itself rather than as a
    removed-and-added pair; its path change is reported separately, in
    `moved`, and is never a finding on its own - a migration is allowed to
    relocate a page, only not to change what it says.

    `expect_body_change` turns the opposite question on: report a page whose
    body is byte-identical. A migration unit that reports no such page did
    something to every page it claimed to touch.
    """
    diff = CorpusDiff()
    diff.added = sorted(set(after) - set(before))
    diff.removed = sorted(set(before) - set(after))

    for title in sorted(set(before) & set(after)):
        diff.compared += 1
        diff.findings.extend(compare_page(title, before[title], after[title]))
        if before[title].path != after[title].path:
            diff.moved.append((title, before[title].path, after[title].path))
        if expect_body_change and before[title].body == after[title].body:
            diff.findings.append(
                PageFinding(title, "unchanged", "body is byte-identical, but this unit claimed to rewrite it")
            )

    return diff


def render_report(diff: CorpusDiff, from_rev: str) -> str:
    lines = [f"# Corpus diff against {from_rev}", ""]
    lines.append(
        f"{diff.compared} page(s) compared, {len(diff.added)} added, "
        f"{len(diff.removed)} removed, {len(diff.moved)} moved, {len(diff.findings)} finding(s)."
    )
    lines.append("")

    if diff.findings:
        lines.append("## Invariant violations")
        lines.append("")
        lines += [f"- {finding}" for finding in diff.findings]
        lines.append("")
    else:
        lines.append("No invariant changed on any page present in both revisions.")
        lines.append("")

    for label, paths in (("Added pages", diff.added), ("Removed pages", diff.removed)):
        if paths:
            lines.append(f"## {label}")
            lines.append("")
            lines += [f"- {path}" for path in paths]
            lines.append("")

    if diff.moved:
        lines.append("## Moved pages")
        lines.append("")
        lines += [f"- {title}: `{old}` -> `{new}`" for title, old, new in diff.moved]
        lines.append("")

    return "\n".join(lines)