feat(scripts): measure_duplication - the DRY close-out measure: before/after duplicate share, and the copies that exist only after (#4745)

A multi-pass DRY audit could not answer whether it created duplication:
a pass's own shortening can leave two statements identical to a third,
and no per-pass scan sees a copy that did not exist when it ran. The
Librarian retrospective improvised this measure in a scratchpad; this is
it as a tool the DRY Pass process can name.

6-line windows over significant lines (comments, blanks and bare
punctuation dropped, strings folded), grouped by glob or extension.
Revisions are read through git archive, so nothing is checked out.
Stdlib only, so any project can run a scratch copy.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

scripts/measure_duplication.py

@@ -0,0 +1,272 @@
+#!/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 (#4745: a retro over
+a 17-pass audit found exactly this, and folded it into a view). 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())

tests/test_measure_duplication.py

@@ -0,0 +1,128 @@
+"""The DRY close-out measure (#4745).
+
+Stdlib-only and kept in scripts/ so any project can run a scratch copy of it,
+so these tests import it by path rather than as a package.
+
+What matters is the after-only list: it is the part of the measure that finds
+something a pass could not, so it must name a copy the change CREATED and stay
+quiet about one that was already there. A list that repeats old duplication
+every time is one nobody reads.
+"""
+import importlib.util
+import pathlib
+import shutil
+import subprocess
+
+import pytest
+
+_PATH = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "measure_duplication.py"
+_spec = importlib.util.spec_from_file_location("measure_duplication", _PATH)
+dup = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(dup)
+
+
+BODY = """
+def load(user_id, note_id):
+    # a comment the measure ignores
+    if not allowed(user_id, note_id):
+        raise ValueError("note {} not found".format(note_id))
+    with session() as s:
+        row = s.get(Row, note_id)
+        if row is None:
+            raise ValueError("not a row")
+        return row
+"""
+
+
+def _files(**texts):
+    return [(name, text.encode()) for name, text in texts.items()]
+
+
+def test_comments_blanks_and_bare_punctuation_are_not_code():
+    lines = dup.significant_lines('x = 1\n\n// note\n# note\n  });\n-- sql note\ny = "two"\n')
+    assert [t for _, t in lines] == ["x = 1", 'y = "S"']
+    assert [n for n, _ in lines] == [1, 7]
+
+
+def test_a_preprocessor_line_is_code_and_a_hash_comment_is_not():
+    lines = dup.significant_lines("#include <x.h>\n# a comment\n#!/bin/sh\n")
+    assert [t for _, t in lines] == ["#include <x.h>"]
+
+
+def test_copies_that_differ_only_in_their_strings_are_one_copy():
+    other = BODY.replace('"not a row"', '"no such row"')
+    result = dup.measure(_files(**{"a.py": BODY, "b.py": other}), [], [], window=6)
+    py = result["py"]
+    assert py["dup_windows"] > 0
+    assert py["dup_lines"] == py["lines"]
+    assert py["share"] == 1.0
+
+
+def test_unrelated_files_measure_zero():
+    other = "\n".join(f"v{i} = compute({i})" for i in range(20))
+    result = dup.measure(_files(**{"a.py": BODY, "b.py": other}), [], [], window=6)
+    assert result["py"]["dup_lines"] == 0
+
+
+def test_the_first_matching_group_claims_a_file():
+    groups = [("tests", ["tests/*"]), ("py", ["*.py"])]
+    assert dup.group_of("tests/test_x.py", groups) == "tests"
+    assert dup.group_of("src/x.py", groups) == "py"
+    assert dup.group_of("README.md", groups) is None
+
+
+def test_without_groups_only_source_extensions_are_measured():
+    assert dup.group_of("src/x.go", []) == "go"
+    assert dup.group_of("docs/x.md", []) is None
+    assert dup.group_of("Makefile", []) is None
+
+
+def test_excluded_paths_are_not_measured():
+    result = dup.measure(
+        _files(**{"a.py": BODY, "gen/b.py": BODY}), [], ["gen/*"], window=6,
+    )
+    assert result["py"]["dup_lines"] == 0
+
+
+def test_a_copy_the_change_created_is_listed_after_only():
+    before = dup.measure(_files(**{"a.py": BODY, "c.py": "z = 1\n"}), [], [], window=6)
+    after = dup.measure(_files(**{"a.py": BODY, "c.py": BODY}), [], [], window=6)
+    fresh = dup.only_after(before, after)
+    assert [e["files"] for e in fresh["py"]] == [["a.py", "c.py"]]
+    assert fresh["py"][0]["at"] == ["a.py:2", "c.py:2"]
+
+
+def test_a_copy_that_was_already_there_is_not_listed():
+    both = _files(**{"a.py": BODY, "b.py": BODY})
+    before = dup.measure(both, [], [], window=6)
+    after = dup.measure(both, [], [], window=6)
+    assert dup.only_after(before, after) == {}
+
+
+def test_the_report_shows_before_and_after_side_by_side():
+    before = dup.measure(_files(**{"a.py": BODY}), [], [], window=6)
+    after = dup.measure(_files(**{"a.py": BODY, "b.py": BODY}), [], [], window=6)
+    text = dup.report(before, after, dup.only_after(before, after), show=5)
+    assert "0.0% → 100.0%" in text
+    assert "a.py:2, b.py:2" in text
+
+
+@pytest.mark.skipif(shutil.which("git") is None, reason="needs git")
+def test_a_revision_is_read_without_a_checkout(tmp_path, capsys):
+    def git(*args):
+        subprocess.run(
+            ["git", "-C", str(tmp_path), "-c", "user.name=t", "-c", "user.email=t@t", *args],
+            check=True, capture_output=True,
+        )
+
+    git("init", "-q")
+    (tmp_path / "a.py").write_text(BODY)
+    git("add", "a.py")
+    git("commit", "-qm", "one copy")
+    (tmp_path / "b.py").write_text(BODY)
+    git("add", "b.py")
+
+    assert dup.main(["--repo", str(tmp_path), "--before", "HEAD"]) == 0
+    out = capsys.readouterr().out
+    assert "Duplicated only after: 1 file set(s)" in out
+    assert "a.py:2, b.py:2" in out