improvise/scripts/repo-stats

#!/usr/bin/env python3
"""Churn, coupling and doc-growth report over a window of history.

code-maat's analyses and this script's per-file churn go into one directory of
CSVs; improvise imports them into one model and computes the report's tables.
See context/code/tooling.md section 7.
"""

import argparse
import csv
import datetime
import json
import os
import re
import subprocess
import sys
import tempfile

# First match wins, so the order is part of the definition.
AREAS = [
    ("generated", re.compile(r"^(\.beads/|site/|Cargo\.nix$|roadmap\.org$)"
                             r"|\.(lock|cast)$|(^|/)package-lock\.json$")),
    ("repo-map", re.compile(r"^context/repo-map\.md$")),
    ("kb.pl", re.compile(r"^prolog-sessions/kb\.pl$")),
    ("context/code", re.compile(r"^context/code/")),
    ("context/test-evidence", re.compile(r"^context/test-evidence/")),
    ("other context", re.compile(r"^context/")),
    ("other docs", re.compile(r"^(docs/.*\.md|README\.md|RELEASING\.md)$")),
    ("agent docs and skills", re.compile(r"^(AGENTS\.md|CLAUDE\.md|\.agents/|\.claude/|\.codex/)")),
    ("scripts/CI", re.compile(r"^(scripts/|\.github/|nix/|flake\.nix$)")),
    ("web and harness", re.compile(r"^(web|harness)/")),
    ("rust tests", re.compile(r"(^|/)(tests|benches)/.*\.rs$|(^|/|_)tests?\.rs$")),
    ("rust src", re.compile(r"\.(rs|pest)$")),
]
OTHER = "other"
PROSE = {"context/code", "context/test-evidence", "other context", "other docs", "repo-map",
         "agent docs and skills"}
CODE = {"rust src", "rust tests", "scripts/CI", "web and harness"}
AGENT_RULES = "agent docs and skills"

# One CSV per code-maat analysis, plus churn.csv. Each stem becomes an item of
# the model's File category, which the .improvise script pages on.
ANALYSES = ["coupling", "abs-churn", "age", "soc"]
CATEGORIES = ["commit", "date", "week", "entity", "area", "kind", "coupled",
              "crosses-crate", "agent-rules"]
CHURN_HEADER = ["commit", "date", "week", "entity", "area", "kind", "added", "deleted"]
MEASURES = ["added", "deleted", "commits", "degree", "average-revs", "age-months", "soc"]
TABLES = ["by-area", "by-kind", "area-by-week", "by-day"]
HERE = os.path.dirname(os.path.abspath(__file__))


class ImproviseFailed(Exception):
    """improvise refused the pipeline's input: a product regression, not a statistic."""


def area_of(path):
    for name, rx in AREAS:
        if rx.search(path):
            return name
    return OTHER


def kind_of(area):
    return "prose" if area in PROSE else "code" if area in CODE else "other"


def crate_of(path):
    """The crate a path belongs to, or None for a path in no crate."""
    m = re.match(r"^crates/([^/]+)/", path)
    if m:
        return m.group(1)
    if path.startswith("harness/opfs-contract/"):
        return "opfs-contract"
    if re.match(r"^(src|tests|benches|examples)/", path) or path == "Cargo.toml":
        return "improvise"
    return None


def git(repo, *args):
    return subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True,
                          check=True).stdout


def resolve_ref(repo, ref):
    """The ref asked for; the default gh/main falls back to HEAD where it is absent."""
    for r in ([ref] if ref else ["gh/main", "HEAD"]):
        out = subprocess.run(["git", "-C", repo, "rev-parse", "--verify", "--quiet",
                              r + "^{commit}"], capture_output=True, text=True)
        if out.returncode == 0:
            return r, out.stdout.strip()
    sys.exit(f"repo-stats: cannot resolve {ref or 'gh/main or HEAD'}")


def git2_log(repo, sha, since):
    """The log in code-maat's git2 format: the ref's history, without merges."""
    return git(repo, "log", sha, "--no-merges", f"--since={since}", "--numstat",
               "--date=short", "--pretty=format:--%h--%ad--%aN", "--no-renames")


def is_header(line):
    return line.startswith("--") and line.count("--") >= 3


def churn_rows(log):
    """One row per (commit, file). A commit is written `g<sha>`, because improvise
    reads an all-digit or exponent-shaped sha as a number (tooling.md section 7)."""
    rows, commit, date = [], None, None
    for line in log.split("\n"):
        if is_header(line):
            _, commit, date, _ = line.split("--", 3)
        elif "\t" in line:
            a, d, path = line.split("\t", 2)
            year, week, _ = datetime.date.fromisoformat(date).isocalendar()
            area = area_of(path)
            rows.append({"commit": "g" + commit, "date": date,
                         "week": f"{year}-W{week:02d}", "entity": path, "area": area,
                         "kind": kind_of(area), "added": 0 if a == "-" else int(a),
                         "deleted": 0 if d == "-" else int(d)})
    return rows


def write_csv(path, header, rows):
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=header)
        w.writeheader()
        w.writerows(rows)


def read_csv(path):
    with open(path, newline="") as f:
        return list(csv.DictReader(f))


def code_maat(logfile, analysis, out, *extra):
    subprocess.run(["code-maat", "-l", logfile, "-c", "git2", "-a", analysis,
                    "-o", out, *extra], capture_output=True, text=True, check=True)


def flag_coupling(path):
    """code-maat's coupling.csv, plus the two flags code-maat has no notion of."""
    rows = read_csv(path)
    for r in rows:
        a, b = crate_of(r["entity"]), crate_of(r["coupled"])
        r["crosses-crate"] = "yes" if a and b and a != b else "no"
        both = area_of(r["entity"]) == area_of(r["coupled"]) == AGENT_RULES
        r["agent-rules"] = "yes" if both else "no"
    write_csv(path, ["entity", "coupled", "degree", "average-revs", "crosses-crate",
                     "agent-rules"], rows)


def build_dir(out, log, rows, min_revs, today):
    """The directory improvise imports: churn.csv and one CSV per analysis."""
    with tempfile.NamedTemporaryFile("w", suffix=".log", delete=False) as f:
        f.write(log + "\n")
    try:
        revs = ["--min-revs", str(min_revs), "--min-shared-revs", str(min_revs)]
        extra = {"coupling": revs, "age": ["--age-time-now", today]}
        for a in ANALYSES:
            code_maat(f.name, a, os.path.join(out, f"{a}.csv"), *extra.get(a, []))
    finally:
        os.unlink(f.name)
    flag_coupling(os.path.join(out, "coupling.csv"))


def improvise(out, *args):
    exe = os.environ.get("IMPROVISE", "improvise")
    p = subprocess.run([exe, *args], cwd=out, capture_output=True, text=True)
    if p.returncode != 0 or "warning:" in p.stderr:
        raise ImproviseFailed(f"improvise {' '.join(args)} exited {p.returncode}:\n"
                              f"{p.stderr}")


def run_improvise(out):
    """Import the directory into one model, a view per file included, then build
    the report's views and export them."""
    flags = [f"--category={c}" for c in CATEGORIES] + [f"--measure={m}" for m in MEASURES]
    improvise(out, "import", ".", "--no-wizard", *flags, "--name=Repo stats",
              "--formula=DeletionRatio = deleted / added FORMAT .3",
              "-o", "repo-stats.improv")
    os.makedirs(os.path.join(out, "report"), exist_ok=True)
    improvise(out, "script", os.path.join(HERE, "repo-stats.improvise"),
              "-f", "repo-stats.improv")
    return {t: read_rows(os.path.join(out, "report", f"{t}.csv")) for t in TABLES}


def read_rows(path):
    with open(path, newline="") as f:
        return list(csv.reader(f))


def gotchas(text):
    """(number, lines, chars) for each numbered item of context/gotchas.md."""
    out, cur, inside = [], None, True
    for line in text.split("\n"):
        if inside and line.startswith("---"):
            inside = False
        if not inside:
            continue
        m = re.match(r"^(\d+)\. ", line)
        if m:
            cur = [int(m.group(1)), 0, 0]
            out.append(cur)
        if cur is not None and line.strip():
            cur[1] += 1
            cur[2] += len(line)
    return [tuple(g) for g in out]


def doc_sizes(repo, sha):
    files = [p for p in git(repo, "ls-tree", "-r", "--name-only", sha, "--",
                            "context").split("\n") if p.endswith(".md")]
    show = lambda p: git(repo, "show", f"{sha}:{p}")
    lines = {p: len(show(p).splitlines()) for p in sorted(files)}
    gs = gotchas(show("context/gotchas.md")) if "context/gotchas.md" in files else []
    longest = max(gs, key=lambda g: g[2], default=None)
    evidence = {p: sum(1 for l in show(p).split("\n") if re.match(r"^#{3,4} ", l))
                for p in sorted(files) if p.startswith("context/test-evidence/")}
    return {
        "context_lines": lines,
        "gotchas": {"count": len(gs), "longest": longest and
                     {"number": longest[0], "lines": longest[1], "chars": longest[2]}},
        "test_evidence": {"entries": sum(evidence.values()), "per_file": evidence},
    }


def report(repo, ref, since, min_revs, out):
    name, sha = resolve_ref(repo, ref)
    log = git2_log(repo, sha, since)
    commits = sum(1 for line in log.split("\n") if is_header(line))
    today = git(repo, "show", "-s", "--date=short", "--format=%cd", sha).strip()
    r = {"ref": name, "sha": sha, "since": since, "commits": commits, "out": out,
         "docs": doc_sizes(repo, sha), "tables": {}, "coupling": [], "soc": [], "age": []}
    rows = churn_rows(log)
    write_csv(os.path.join(out, "churn.csv"), CHURN_HEADER, rows)
    # Empty commits alone change no file: nothing to analyse, so no model is built.
    if rows:
        build_dir(out, log, rows, min_revs, today)
        r["tables"] = run_improvise(out)
        for a in ("coupling", "soc", "age"):
            r[a] = read_csv(os.path.join(out, f"{a}.csv"))
    return r


def table(rows):
    """A CSV improvise exported, as aligned text; the first column left-aligned."""
    n = max(len(r) for r in rows)
    widths = [max(len(r[i]) for r in rows if i < len(r)) for i in range(n)]
    return [("  " + "  ".join((c.ljust if i == 0 else c.rjust)(widths[i])
                              for i, c in enumerate(r))).rstrip() for r in rows]


TITLES = {
    "by-area": "Churn by area (improvise)",
    "by-kind": "Prose, code and other; ProseToCode = prose / code (improvise)",
    "area-by-week": "Lines added by area and ISO week (improvise)",
    "by-day": "Lines added per day, churn.csv beside code-maat's abs-churn (improvise)",
}


def text(r, top):
    out = [f"repo-stats: {r['ref']} ({r['sha'][:10]}), since {r['since']}, "
           f"{r['commits']} non-merge commits"]
    if not r["tables"]:
        out += ["", "No file changed in the window, so no model was built."]
    for t in TABLES:
        if r["tables"].get(t):
            out += ["", TITLES[t]] + table(r["tables"][t])
    out += ["", "Doc sizes (lines)"]
    out += [f"  {n:>6}  {path}" for path, n in r["docs"]["context_lines"].items()]
    rm, te = r["docs"]["gotchas"], r["docs"]["test_evidence"]
    lg = rm["longest"]
    out.append(f"  gotchas: {rm['count']}" + (
        f"; longest is {lg['number']} ({lg['lines']} lines, {lg['chars']} chars)"
        if lg else ""))
    out.append(f"  test-evidence entries: {te['entries']}")
    out += ["", f"Change coupling (code-maat; top {top})"]
    for c in r["coupling"][:top]:
        flags = [f for f, k in (("CROSSES-CRATE", "crosses-crate"),
                                ("AGENT-RULES", "agent-rules")) if c[k] == "yes"]
        out.append(f"  {c['degree']:>4}% {c['average-revs']:>4} revs  {c['entity']}"
                   f" <-> {c['coupled']}" + (f"  [{', '.join(flags)}]" if flags else ""))
    out += ["", f"Sum of coupling (code-maat; top {top})"]
    out += [f"  {s['soc']:>6}  {s['entity']}" for s in r["soc"][:top]]
    out += ["", f"Code age in months (code-maat; first {top})"]
    out += [f"  {a['age-months']:>4}  {a['entity']}" for a in r["age"][:top]]
    return "\n".join(out)


def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--ref", help="the ref to report on (default gh/main, else HEAD)")
    ap.add_argument("--since", default="7 days ago", help="git --since (default 7 days ago)")
    ap.add_argument("--min-revs", type=int, default=5,
                    help="code-maat --min-revs and --min-shared-revs (default 5)")
    ap.add_argument("--top", type=int, default=15, help="rows per ranked table")
    ap.add_argument("--out", default="repo-stats",
                    help="the directory for the CSVs, the model and its exports")
    ap.add_argument("--json", action="store_true", help="emit JSON instead of text")
    ap.add_argument("-C", dest="repo", default=".", help="the repository (default .)")
    a = ap.parse_args()
    os.makedirs(a.out, exist_ok=True)
    ours = {f"{n}.csv" for n in ["churn"] + ANALYSES}
    foreign = sorted(f for f in os.listdir(a.out) if f.lower().endswith(".csv") and f not in ours)
    if foreign:
        print(f"repo-stats: {a.out} holds CSVs this script did not write, and "
              f"`improvise import` would merge them: {', '.join(foreign)}", file=sys.stderr)
        return 2
    try:
        r = report(a.repo, a.ref, a.since, a.min_revs, os.path.abspath(a.out))
    except ImproviseFailed as e:
        print(f"repo-stats: {e}", file=sys.stderr)
        return 3
    print(json.dumps(r, indent=2) if a.json else text(r, a.top))
    return 0


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