repo_stats

  1#!/usr/bin/env python3
  2"""repo_stats — the README's CI-health, LOC and Python-coverage badges, as shields.io endpoint JSON.
  3
  4    scripts/repo_stats.py write --out-dir stats \
  5        --repo marola-dev/marola --run-id 123 --exclude-job repo-stats
  6    scripts/repo_stats.py write --out-dir stats     # LOC + Python coverage (no --run-id: no CI badge)
  7    scripts/repo_stats.py write --out-dir stats --no-python-coverage   # skip coverage.py
  8    scripts/repo_stats.py python-coverage           # just print the measured % (no files written)
  9    scripts/repo_stats.py --self-test               # shaping + counting rules (just quality-other)
 10
 11Four files, the same shields.io "endpoint" shape ci.yml already writes for the Scala coverage
 12(`{"schemaVersion": 1, "label": ..., "message": ..., "color": ...}`), published to
 13marola-site's `site-data` branch and copied into the map by its site.yml, where the README reads
 14them live:
 15
 16    ci.json               "ci steps"        19/20 green
 17    scala-loc.json        "scala"           6,865 LOC
 18    python-loc.json       "python"          2,877 LOC
 19    python-coverage.json  "py-cov" 73%
 20
 21CI health is *step*-level, not job-level: ci.yml has three jobs but ~20 named steps, and
 22"build-test passed" hides which of them actually ran. Only steps that ran count — a step whose
 23`conclusion` is `skipped` (most of them are gated on `needs.changes.outputs.*`) or still `null`
 24(this job's own later steps) is left out of both numerator and denominator, so a docs-only push
 25reports 8/8 rather than a misleading 8/20. `--exclude-job` drops the reporting job itself, whose
 26steps are by definition still running while it asks.
 27
 28LOC is `cloc` (from `nix develop .#lint`, locally and in ci.yml), counted over the four
 29Scala modules and the Python trees, code lines only — blanks and comments excluded by cloc, and
 30`target/`, `__pycache__/`, virtualenvs and `node_modules/` excluded by path.
 31
 32Python coverage is measured, never estimated — but read the label narrowly. marola has no pytest
 33suite; every `scripts/**/*.py` is tested by its own `--self-test` flag, the list `just
 34quality-other` runs. So this badge is *statement coverage of `scripts/` while those self-tests
 35run*, and nothing more: a branch a self-test never bothers to call is uncovered by construction,
 36which is why the figure sits in the 70s rather than the 90s. `dspy/` and `finetune/` are out of
 37scope — no `--self-test` entry point, and importing them needs torch/DSPy — so they are neither
 38numerator nor denominator, while a `scripts/*.py` that grows without a self-test does count
 39(at 0%), which is the point. Mechanically: one `coverage run --parallel-mode` per self-test into
 40a temp data file, then `coverage combine` + `coverage json`.
 41
 42Standard library only; `cloc` and `coverage` are the external tools, and only `write` needs them.
 43"""
 44
 45import argparse
 46import json
 47import shutil
 48import subprocess
 49import sys
 50import tempfile
 51from pathlib import Path
 52
 53SCHEMA = 1
 54
 55SCALA_PATHS = ("core", "local", "cli")
 56PYTHON_PATHS = ("dspy", "finetune", "scripts")
 57EXCLUDE_DIRS = ("target", "__pycache__", ".venv", "venv", "node_modules")
 58
 59SCALA_COLOR = "DC322F"  # = the README's hand-written Scala badge
 60PYTHON_COLOR = "3776AB"  # = python.org's brand blue, as used by shields' own python logo
 61
 62# Every `python3 <script> --self-test` line of justfile's `quality-other`, in its order. Keeping
 63# the two lists equal is what makes the badge honest: the number below is exactly what that gate
 64# already runs, not a second, friendlier suite. (The `.sh` self-tests in the same recipe are not
 65# Python and cannot contribute statements.)
 66SELF_TEST_SCRIPTS = (
 67    "scripts/smoke_record.py",
 68    "scripts/benchmark_gate.py",
 69    "scripts/repo_stats.py",
 70    "scripts/arxiv_digest.py",
 71    "scripts/awesome_agentic_digest.py",
 72    "scripts/ocr-post.py",
 73    "scripts/mip_graph.py",
 74    "scripts/strip_external_scripts.py",
 75    "scripts/analyze_training.py",
 76)
 77# Measured tree. `dspy/`/`finetune/` are excluded on purpose — see the module docstring.
 78COVERAGE_SOURCE = "scripts"
 79
 80# A step that reached one of these actually executed; anything else (`skipped`, `neutral`, or a
 81# null conclusion for a step still queued/running) is not evidence either way and is not counted.
 82RAN = ("success", "failure", "cancelled", "timed_out")
 83GREEN = ("success",)
 84
 85
 86# ---------------------------------------------------------------------------------------------
 87# Pure shaping — everything below self-tests without a network call or a `cloc` on PATH.
 88
 89
 90def badge(label: str, message: str, color: str) -> dict:
 91    """One shields.io endpoint document (https://shields.io/badges/endpoint-badge)."""
 92    return {"schemaVersion": SCHEMA, "label": label, "message": message, "color": color}
 93
 94
 95def count_steps(jobs: list[dict], exclude_job: str | None = None) -> tuple[int, int]:
 96    """(green, ran) over every step of every job of one run — skipped/unfinished steps ignored."""
 97    green = ran = 0
 98    for job in jobs:
 99        if exclude_job and job.get("name") == exclude_job:
100            continue
101        for step in job.get("steps") or []:
102            conclusion = step.get("conclusion")
103            if conclusion not in RAN:
104                continue
105            ran += 1
106            if conclusion in GREEN:
107                green += 1
108    return green, ran
109
110
111def ci_badge(green: int, ran: int) -> dict:
112    """Green only when every step that ran passed — a single red step is not a rounding error."""
113    if ran == 0:
114        return badge("ci steps", "no data", "lightgrey")
115    color = "brightgreen" if green == ran else "yellow" if green >= 0.9 * ran else "red"
116    return badge("ci steps", f"{green}/{ran} green", color)
117
118
119def loc_badge(label: str, code: int, color: str) -> dict:
120    return badge(label, f"{code:,} LOC", color)
121
122
123def parse_cloc(payload: str, language: str) -> int:
124    """Code lines for one language out of `cloc --json`; 0 when it found none of that language."""
125    return int(json.loads(payload).get(language, {}).get("code", 0))
126
127
128def coverage_badge(percent: float | None) -> dict:
129    """Statement coverage of `scripts/` under its own self-tests. Same thresholds as ci.yml's
130    Scala badge (red < 50 ≤ yellow < 80 ≤ green) so the two read on one scale. The label is
131    `py-cov`, paired with ci.yml's `sc-cov` — short enough that the two badges sit side by side
132    without wrapping, and still distinguishable at a glance, which is the only thing the label
133    has to do."""
134    if percent is None:
135        return badge("py-cov", "no data", "lightgrey")
136    color = "red" if percent < 50 else "yellow" if percent < 80 else "green"
137    return badge("py-cov", f"{percent:.0f}%", color)
138
139
140def parse_coverage_json(payload: str) -> float:
141    """The overall statement percentage out of `coverage json` (`totals.percent_covered`)."""
142    return float(json.loads(payload)["totals"]["percent_covered"])
143
144
145def coverage_run_argv(exe: list[str], script: str, data_file: Path) -> list[str]:
146    """One instrumented self-test run. `--parallel-mode` keeps the ten runs from overwriting each
147    other's data file; `--source` fixes the measured tree so a `scripts/*.py` that no self-test
148    imports still lands in the denominator at 0% instead of vanishing from the report."""
149    return [
150        *exe,
151        "run",
152        "--parallel-mode",
153        f"--data-file={data_file}",
154        f"--source={COVERAGE_SOURCE}",
155        script,
156        "--self-test",
157    ]
158
159
160# ---------------------------------------------------------------------------------------------
161# The two effectful sources: this run's jobs (GitHub API, via `gh`) and `cloc`.
162
163
164def fetch_jobs(repo: str, run_id: str) -> list[dict]:
165    """Every job of one workflow run, steps included. Needs `actions: read` on the token."""
166    jobs: list[dict] = []
167    page = 1
168    while True:
169        url = f"repos/{repo}/actions/runs/{run_id}/jobs?per_page=100&page={page}"
170        out = subprocess.run(
171            ["gh", "api", "-H", "Accept: application/vnd.github+json", url],
172            capture_output=True,
173            text=True,
174            check=True,
175        ).stdout
176        batch = json.loads(out).get("jobs") or []
177        jobs.extend(batch)
178        if len(batch) < 100:
179            return jobs
180        page += 1
181
182
183def cloc_code(paths: tuple[str, ...], language: str, root: Path) -> int:
184    """Code lines of `language` under `paths`; a path that does not exist is simply skipped."""
185    if not shutil.which("cloc"):
186        raise SystemExit(
187            "repo_stats: `cloc` is not on PATH (nix develop has it, and ci.yml takes it from nix develop .#lint)"
188        )
189    present = [p for p in paths if (root / p).exists()]
190    if not present:
191        return 0
192    out = subprocess.run(
193        [
194            "cloc",
195            "--json",
196            "--quiet",
197            f"--exclude-dir={','.join(EXCLUDE_DIRS)}",
198            f"--include-lang={language}",
199            *present,
200        ],
201        capture_output=True,
202        text=True,
203        check=True,
204        cwd=root,
205    ).stdout
206    # cloc prints nothing at all when no file of that language survived the filters.
207    return parse_cloc(out, language) if out.strip() else 0
208
209
210def coverage_exe(which=shutil.which, has_module=None) -> list[str]:
211    """How to invoke coverage.py here, as an argv prefix.
212
213    Two shapes: nix's `python3Packages.coverage` puts a wrapped `coverage` on PATH but *not* on
214    this interpreter's import path, while a pip or distro install does the opposite. Prefer the
215    executable, fall back to `-m`, fail loudly if neither.
216    """
217    if has_module is None:
218
219        def has_module() -> bool:
220            import importlib.util
221
222            return importlib.util.find_spec("coverage") is not None
223
224    if which("coverage"):
225        return ["coverage"]
226    if has_module():
227        return [sys.executable, "-m", "coverage"]
228    raise SystemExit(
229        "repo_stats: coverage.py is not installed (nix develop has it, and ci.yml takes it "
230        "from nix develop .#lint) — or pass --no-python-coverage"
231    )
232
233
234def python_coverage(root: Path) -> float:
235    """Run every self-test under coverage.py and return the combined statement percentage.
236
237    The data files live in a temp directory, so a run leaves no `.coverage*` behind in the repo.
238    A self-test that *fails* aborts the measurement rather than quietly reporting a smaller
239    number — `just quality-other` is the gate for that, and a green badge over a red self-test
240    would be worse than no badge.
241    """
242    exe = coverage_exe()
243    with tempfile.TemporaryDirectory() as tmp:
244        data_file = Path(tmp) / ".coverage"
245        for script in SELF_TEST_SCRIPTS:
246            subprocess.run(
247                coverage_run_argv(exe, script, data_file),
248                cwd=root,
249                check=True,
250                capture_output=True,
251                text=True,
252            )
253        subprocess.run(
254            [*exe, "combine", f"--data-file={data_file}", tmp],
255            cwd=root,
256            check=True,
257            capture_output=True,
258            text=True,
259        )
260        report = Path(tmp) / "coverage.json"
261        subprocess.run(
262            [*exe, "json", f"--data-file={data_file}", "-o", str(report)],
263            cwd=root,
264            check=True,
265            capture_output=True,
266            text=True,
267        )
268        return parse_coverage_json(report.read_text())
269
270
271def write(out_dir: Path, badges: dict[str, dict]) -> list[Path]:
272    out_dir.mkdir(parents=True, exist_ok=True)
273    written = []
274    for name, doc in badges.items():
275        path = out_dir / name
276        path.write_text(json.dumps(doc) + "\n")
277        written.append(path)
278    return written
279
280
281def collect(args) -> dict[str, dict]:
282    root = Path(args.root)
283    badges = {
284        "scala-loc.json": loc_badge("scala", cloc_code(SCALA_PATHS, "Scala", root), SCALA_COLOR),
285        "python-loc.json": loc_badge(
286            "python", cloc_code(PYTHON_PATHS, "Python", root), PYTHON_COLOR
287        ),
288    }
289    if args.run_id:
290        green, ran = count_steps(fetch_jobs(args.repo, args.run_id), args.exclude_job)
291        badges["ci.json"] = ci_badge(green, ran)
292    if not args.no_python_coverage:
293        badges["python-coverage.json"] = coverage_badge(python_coverage(root))
294    return badges
295
296
297# ---------------------------------------------------------------------------------------------
298
299
300def self_test() -> int:
301    jobs = [
302        {
303            "name": "build-test",
304            "steps": [
305                {"name": "checkout", "conclusion": "success"},
306                {"name": "compile+test", "conclusion": "success"},
307                {"name": "coverage", "conclusion": "skipped"},  # main-only, on a PR
308            ],
309        },
310        {
311            "name": "quality-other",
312            "steps": [
313                {"name": "ruff", "conclusion": "success"},
314                {"name": "actionlint", "conclusion": "failure"},
315                {"name": "hadolint", "conclusion": "skipped"},
316                {"name": "docker compose config", "conclusion": None},  # never reached
317            ],
318        },
319        {
320            "name": "repo-stats",  # the reporting job itself: excluded, steps still running
321            "steps": [
322                {"name": "checkout", "conclusion": "success"},
323                {"name": "badges", "conclusion": None},
324            ],
325        },
326    ]
327    assert count_steps(jobs, "repo-stats") == (3, 4), count_steps(jobs, "repo-stats")
328    # Without the exclusion the reporting job's own finished steps leak in.
329    assert count_steps(jobs) == (4, 5), count_steps(jobs)
330    # A job with no steps at all (queued, or `steps` absent) contributes nothing, never crashes.
331    assert count_steps([{"name": "changes"}, {"name": "x", "steps": None}]) == (0, 0)
332
333    # Skipped steps stay out of the denominator: a docs-only push is 2/2, not 2/9.
334    docs_only = [
335        {
336            "name": "quality-other",
337            "steps": [{"conclusion": "success"}] * 2 + [{"conclusion": "skipped"}] * 7,
338        }
339    ]
340    assert count_steps(docs_only) == (2, 2)
341    assert ci_badge(*count_steps(docs_only))["message"] == "2/2 green"
342
343    assert ci_badge(20, 20) == {
344        "schemaVersion": 1,
345        "label": "ci steps",
346        "message": "20/20 green",
347        "color": "brightgreen",
348    }
349    assert ci_badge(19, 20)["color"] == "yellow", "one red step out of twenty: not green, not red"
350    assert ci_badge(17, 20)["color"] == "red"
351    assert ci_badge(0, 0) == {
352        "schemaVersion": 1,
353        "label": "ci steps",
354        "message": "no data",
355        "color": "lightgrey",
356    }
357
358    assert loc_badge("scala", 6865, SCALA_COLOR)["message"] == "6,865 LOC"
359    assert loc_badge("python", 0, PYTHON_COLOR)["message"] == "0 LOC"
360
361    cloc_json = json.dumps(
362        {
363            "header": {"cloc_version": "2.10"},
364            "Scala": {"nFiles": 84, "blank": 1050, "comment": 1673, "code": 6865},
365            "SUM": {"blank": 1050, "comment": 1673, "code": 6865, "nFiles": 84},
366        }
367    )
368    assert parse_cloc(cloc_json, "Scala") == 6865
369    assert parse_cloc(cloc_json, "Python") == 0, "a language cloc did not find is 0, not an error"
370
371    # --- Python coverage: shaping, parsing, the argv builder and how coverage.py is located.
372    assert coverage_badge(73.0) == {
373        "schemaVersion": 1,
374        "label": "py-cov",
375        "message": "73%",
376        "color": "yellow",
377    }
378    assert coverage_badge(80.0)["color"] == "green", "the ci.yml Scala badge's own boundary"
379    assert coverage_badge(49.9)["color"] == "red"
380    assert coverage_badge(100.0)["message"] == "100%", "no decimals on the badge"
381    assert coverage_badge(None) == {
382        "schemaVersion": 1,
383        "label": "py-cov",
384        "message": "no data",
385        "color": "lightgrey",
386    }
387    # Both coverage badges must be distinguishable at a glance — this is the whole reason the
388    # label is not just "coverage" like ci.yml's Scala one used to be. `py-cov` here pairs with
389    # `sc-cov` in ci.yml; if one is renamed the other has to follow.
390    assert coverage_badge(73.0)["label"] != ci_badge(1, 1)["label"]
391
392    cov_json = json.dumps(
393        {
394            "meta": {"version": "7.15.4"},
395            "files": {"scripts/repo_stats.py": {"summary": {"percent_covered": 75.0}}},
396            "totals": {
397                "covered_lines": 1328,
398                "num_statements": 1820,
399                "percent_covered": 72.96703296703296,
400            },
401        }
402    )
403    assert round(parse_coverage_json(cov_json), 2) == 72.97
404    assert coverage_badge(parse_coverage_json(cov_json))["message"] == "73%"
405
406    argv = coverage_run_argv(["coverage"], "scripts/mip_graph.py", Path("/tmp/x/.coverage"))
407    assert argv[:2] == ["coverage", "run"]
408    assert "--parallel-mode" in argv, "ten runs into one data file need parallel mode"
409    assert "--data-file=/tmp/x/.coverage" in argv, "data files stay out of the repo"
410    assert f"--source={COVERAGE_SOURCE}" in argv
411    assert argv[-2:] == ["scripts/mip_graph.py", "--self-test"]
412    assert coverage_run_argv([sys.executable, "-m", "coverage"], "s.py", Path("d"))[1] == "-m"
413
414    assert coverage_exe(which=lambda _: "/usr/bin/coverage") == ["coverage"], "prefer the exe"
415    assert coverage_exe(which=lambda _: None, has_module=lambda: True) == [
416        sys.executable,
417        "-m",
418        "coverage",
419    ], "nix's coverage is on PATH; Ubuntu's python3-coverage is only importable"
420    try:
421        coverage_exe(which=lambda _: None, has_module=lambda: False)
422        raise AssertionError("a missing coverage.py must fail loudly, not report 0%")
423    except SystemExit as exc:
424        assert "--no-python-coverage" in str(exc), str(exc)
425
426    # The badge is only honest while this list is exactly justfile's; drift is the failure mode.
427    root = Path(__file__).resolve().parent.parent
428    for script in SELF_TEST_SCRIPTS:
429        assert (root / script).exists(), f"{script} is in SELF_TEST_SCRIPTS but not on disk"
430    justfile = (root / "justfile").read_text()
431    for script in SELF_TEST_SCRIPTS:
432        assert f"python3 {script} --self-test" in justfile, f"{script} left quality-other"
433    in_recipe = {
434        line.split()[1]
435        for line in justfile.splitlines()
436        if line.strip().startswith("python3 scripts/") and line.strip().endswith("--self-test")
437    }
438    assert in_recipe == set(SELF_TEST_SCRIPTS), sorted(in_recipe ^ set(SELF_TEST_SCRIPTS))
439
440    with tempfile.TemporaryDirectory() as tmp:
441        out = Path(tmp) / "stats"
442        paths = write(out, {"ci.json": ci_badge(20, 20), "scala-loc.json": loc_badge("s", 1, "x")})
443        assert [p.name for p in paths] == ["ci.json", "scala-loc.json"]
444        assert json.loads((out / "ci.json").read_text())["message"] == "20/20 green"
445        # Rewriting replaces rather than appends — every run publishes a whole document.
446        write(out, {"ci.json": ci_badge(1, 2)})
447        assert json.loads((out / "ci.json").read_text())["message"] == "1/2 green"
448
449    write_args = build_parser().parse_args(["write", "--out-dir", "x"])
450    assert write_args.repo == "marola-dev/marola", write_args.repo
451
452    print(
453        "repo_stats self-test: ok (step counting, badge shaping, cloc/coverage parsing, "
454        "the coverage argv + exe resolution, SELF_TEST_SCRIPTS vs. justfile, write, "
455        "the --repo default)"
456    )
457    return 0
458
459
460def build_parser() -> argparse.ArgumentParser:
461    ap = argparse.ArgumentParser(
462        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
463    )
464    ap.add_argument("--self-test", action="store_true")
465    sub = ap.add_subparsers(dest="cmd")
466    w = sub.add_parser("write", help="write the badge JSONs into --out-dir")
467    w.add_argument("--out-dir", required=True, type=Path)
468    w.add_argument("--root", default=".", help="repo root the LOC paths are relative to")
469    w.add_argument(
470        "--repo", default="marola-dev/marola", help="owner/name, for the CI-health badge"
471    )
472    w.add_argument("--run-id", help="workflow run to report on; omitted = LOC badges only")
473    w.add_argument("--exclude-job", help="job name to leave out (the reporting job itself)")
474    w.add_argument(
475        "--no-python-coverage",
476        action="store_true",
477        help="skip the coverage.py run (no coverage.py installed, or LOC/CI badges only)",
478    )
479    c = sub.add_parser(
480        "python-coverage", help="print the measured statement %% of scripts/ and exit"
481    )
482    c.add_argument("--root", default=".", help="repo root the self-test paths are relative to")
483    return ap
484
485
486def main(argv: list[str]) -> int:
487    ap = build_parser()
488    args = ap.parse_args(argv)
489    if args.self_test:
490        return self_test()
491    if args.cmd == "python-coverage":
492        percent = python_coverage(Path(args.root))
493        print(f"{percent:.1f}% ({coverage_badge(percent)['message']} on the badge)")
494        return 0
495    if args.cmd != "write":
496        ap.print_help()
497        return 2
498    badges = collect(args)
499    for path in write(args.out_dir, badges):
500        print(f"{path}: {path.read_text().strip()}")
501    return 0
502
503
504if __name__ == "__main__":
505    sys.exit(main(sys.argv[1:]))