FabledGrimoire/scripts/measure_duplication.py

#!/usr/bin/env python3
"""Measure near-verbatim duplication in a tree, and what a change added to it.

WHY THIS EXISTS

A DRY audit made of several passes ends with a question no single pass can
answer: did the audit as a whole reduce duplication, and did it create any? The
second half is the one that matters. A pass that shortens two statements can
leave them byte-identical to a third, and the new copy is invisible to that
pass's own scan because it did not exist when the scan ran. This script is
the close-out measure the DRY Pass process names, so the measure is a tool and
not something each audit improvises.

WHAT IT MEASURES

Each file is reduced to its significant lines: blank lines, comment-only lines
and lines of bare punctuation (`}`, `});`, `],`) are dropped; whitespace is
collapsed; string literals are folded to "S", so two copies that differ only in
a message or a key still match. A *window* is N consecutive significant lines
(default 6). A window whose text occurs at two or more places in the same group
is duplicated, and every line it covers is a duplicated line.

Per group it reports significant lines, duplicated windows (distinct texts),
duplicated lines and their share. With `--before REV` it measures that revision
too and lists the duplicated windows that exist ONLY after — not duplicated
before, either because the text was absent or because it occurred once. Those
are grouped by the set of files they appear in, which is the list to read.

WHAT IT DOES NOT SEE

Structural or semantic copies — two components with the same shape and
different names, the same predicate written two ways. It catches near-verbatim
text only, so a fold of structural copies shows here less than it counts. Read
the numbers as a floor and the after-only list as the finding.

USAGE

    measure_duplication.py [--repo DIR] [--before REV] [--after REV]
                           [--group NAME=GLOB[,GLOB...]]... [--exclude GLOB]...
                           [--window N] [--show N] [--json]

The after tree is the working tree's tracked files unless `--after REV` names a
revision. Revisions are read through `git archive`, so nothing is checked out.
Globs are fnmatch patterns over repo-relative paths, where `*` crosses `/`.
Groups are tried in order and the first match claims a file, so list
`--group 'go tests=*_test.go'` before `--group 'go=*.go'`. With no `--group`,
files are grouped by extension among common source types.

Stdlib only, so it runs from a scratch copy in any project:
`get_snippet` the recorded snippet, write its code to a file, run it there.
"""
from __future__ import annotations

import argparse
import fnmatch
import hashlib
import io
import json
import re
import subprocess
import sys
import tarfile
from collections import defaultdict
from pathlib import Path

DEFAULT_EXTS = {
    "c", "cc", "cpp", "cs", "css", "dart", "go", "h", "hpp", "java", "js",
    "jsx", "kt", "kts", "php", "py", "rb", "rs", "scss", "sh", "sql",
    "svelte", "swift", "ts", "tsx", "vue",
}

_COMMENT = re.compile(r"^(//|/\*|\*|--|<!--|#(\s|!|$))")
_PUNCT_ONLY = re.compile(r"^[\s{}()\[\];,]*$")
_STRING = re.compile(r'"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|`(?:\\.|[^`\\])*`')
_SPACE = re.compile(r"\s+")


def significant_lines(text: str) -> list[tuple[int, str]]:
    """(1-based line number, normalized text) for each line that carries code."""
    out = []
    for number, raw in enumerate(text.splitlines(), 1):
        line = raw.strip()
        if not line or _COMMENT.match(line) or _PUNCT_ONLY.match(line):
            continue
        out.append((number, _SPACE.sub(" ", _STRING.sub('"S"', line))))
    return out


def _git(repo: Path, *args: str) -> bytes:
    return subprocess.run(
        ["git", "-C", str(repo), *args], check=True, capture_output=True,
    ).stdout


def read_working_tree(repo: Path):
    """Tracked files as they are on disk — what the next commit would hold."""
    for rel in _git(repo, "ls-files", "-z").decode().split("\0"):
        path = repo / rel
        if rel and path.is_file():
            yield rel, path.read_bytes()


def read_revision(repo: Path, rev: str):
    """Every file at `rev`, streamed out of `git archive` without a checkout."""
    stream = io.BytesIO(_git(repo, "archive", "--format=tar", rev))
    with tarfile.open(fileobj=stream, mode="r:") as archive:
        for member in archive:
            if member.isfile():
                yield member.name, archive.extractfile(member).read()


def group_of(rel: str, groups: list[tuple[str, list[str]]]) -> str | None:
    if not groups:
        ext = rel.rsplit(".", 1)[-1] if "." in Path(rel).name else ""
        return ext if ext in DEFAULT_EXTS else None
    for name, patterns in groups:
        if any(fnmatch.fnmatch(rel, p) for p in patterns):
            return name
    return None


def measure(files, groups, excludes, window: int) -> dict:
    """Per group: line counts, and every duplicated window with where it occurs."""
    by_group: dict[str, dict] = defaultdict(
        lambda: {"files": {}, "index": defaultdict(list)},
    )
    for rel, data in files:
        if any(fnmatch.fnmatch(rel, p) for p in excludes):
            continue
        name = group_of(rel, groups)
        if name is None:
            continue
        try:
            text = data.decode("utf-8")
        except UnicodeDecodeError:
            continue
        lines = significant_lines(text)
        g = by_group[name]
        g["files"][rel] = lines
        for i in range(len(lines) - window + 1):
            body = "\n".join(t for _, t in lines[i:i + window])
            key = hashlib.sha1(body.encode()).hexdigest()
            g["index"][key].append((rel, i))

    result = {}
    for name, g in sorted(by_group.items()):
        dup = {k: occ for k, occ in g["index"].items() if len(occ) > 1}
        covered = set()
        for occ in dup.values():
            for rel, i in occ:
                covered.update((rel, j) for j in range(i, i + window))
        total = sum(len(lines) for lines in g["files"].values())
        result[name] = {
            "lines": total,
            "dup_windows": len(dup),
            "dup_lines": len(covered),
            "share": len(covered) / total if total else 0.0,
            "_dup": dup,
            "_files": g["files"],
        }
    return result


def only_after(before: dict, after: dict) -> dict[str, list[dict]]:
    """Duplicated windows in `after` that were not duplicated in `before`,
    grouped by the set of files they occur in."""
    out = {}
    for name, a in after.items():
        was = before.get(name, {}).get("_dup", {})
        sets: dict[tuple, dict] = {}
        for key, occ in a["_dup"].items():
            if key in was:
                continue
            fileset = tuple(sorted({rel for rel, _ in occ}))
            entry = sets.setdefault(fileset, {"files": list(fileset), "windows": 0, "at": []})
            entry["windows"] += 1
            if not entry["at"]:
                entry["at"] = [
                    f"{rel}:{a['_files'][rel][i][0]}" for rel, i in sorted(occ)
                ]
        if sets:
            out[name] = sorted(sets.values(), key=lambda e: -e["windows"])
    return out


def _public(result: dict) -> dict:
    return {
        name: {k: v for k, v in g.items() if not k.startswith("_")}
        for name, g in result.items()
    }


def _pct(share: float) -> str:
    return f"{share * 100:.1f}%"


def report(before: dict | None, after: dict, fresh: dict, show: int) -> str:
    rows = []
    for name in sorted(set(after) | set(before or {})):
        a = after.get(name, {"lines": 0, "dup_windows": 0, "dup_lines": 0, "share": 0.0})
        if before is None:
            rows.append((name, f"{a['lines']:,}", str(a["dup_windows"]),
                         str(a["dup_lines"]), _pct(a["share"])))
            continue
        b = before.get(name, {"lines": 0, "dup_windows": 0, "dup_lines": 0, "share": 0.0})
        rows.append((
            name,
            f"{b['lines']:,} → {a['lines']:,}",
            f"{b['dup_windows']} → {a['dup_windows']}",
            f"{b['dup_lines']} → {a['dup_lines']}",
            f"{_pct(b['share'])} → {_pct(a['share'])}",
        ))
    header = ("group", "lines", "dup windows", "dup lines", "share")
    widths = [max(len(r[c]) for r in [header, *rows]) for c in range(len(header))]
    lines = ["  ".join(cell.ljust(w) for cell, w in zip(r, widths)).rstrip()
             for r in [header, *rows]]
    if before is not None:
        total = sum(len(v) for v in fresh.values())
        lines += ["", f"Duplicated only after: {total} file set(s). Read each one —"
                  " a copy a pass's own shortening exposed, or a deliberate repeat."]
        for name, sets in fresh.items():
            for entry in sets[:show]:
                lines.append(f"  [{name}] {entry['windows']} window(s): "
                             + ", ".join(entry["at"]))
            if len(sets) > show:
                lines.append(f"  [{name}] … {len(sets) - show} more (--show)")
    return "\n".join(lines)


def _parse_group(spec: str) -> tuple[str, list[str]]:
    name, sep, globs = spec.partition("=")
    if not sep or not name or not globs:
        raise argparse.ArgumentTypeError(f"--group wants NAME=GLOB[,GLOB...], got {spec!r}")
    return name, [g for g in globs.split(",") if g]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--repo", type=Path, default=Path("."))
    parser.add_argument("--before", help="revision to compare against (e.g. the sha before the first pass)")
    parser.add_argument("--after", help="revision to measure (default: the working tree's tracked files)")
    parser.add_argument("--group", action="append", type=_parse_group, default=[])
    parser.add_argument("--exclude", action="append", default=[])
    parser.add_argument("--window", type=int, default=6)
    parser.add_argument("--show", type=int, default=20)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)

    repo = args.repo.resolve()
    after_files = read_revision(repo, args.after) if args.after else read_working_tree(repo)
    after = measure(after_files, args.group, args.exclude, args.window)
    before = fresh = None
    if args.before:
        before = measure(read_revision(repo, args.before), args.group, args.exclude, args.window)
        fresh = only_after(before, after)

    if args.json:
        json.dump({
            "window": args.window,
            "before": _public(before) if before is not None else None,
            "after": _public(after),
            "only_after": fresh,
        }, sys.stdout, indent=2)
        print()
    else:
        print(report(before, after, fresh or {}, args.show))
    return 0


if __name__ == "__main__":
    sys.exit(main())