Files
troupe-standup/wizard-work/ledger/extract_git_worktree_census.py

256 lines
10 KiB
Python

#!/usr/bin/env python3
"""extract_git_worktree_census.py
Ledger-lane extraction artifact: a census of the troupe worktree derived from
live `git` (the repository is the single source of truth). Writes
git-worktree-census-<stamp>.json machine-readable census
git-worktree-census-<stamp>.csv machine-readable table
git-worktree-census-<stamp>.md human-readable table
No value is invented; every field is a reading of `git` at run time.
"""
import csv
import hashlib
import json
import subprocess
import shutil
from collections import Counter, defaultdict
REPO = "/root/troupe-standup"
STAMP = "2026-10-30"
OUT = "/root/troupe-standup/wizard-work/ledger"
PROFILE_OUT = "/root/.hermes/wizard-outputs/ledger"
def git(*args):
return subprocess.run(
["git", "-C", REPO, *args],
capture_output=True, text=True, check=True,
).stdout
def main():
# --- commit count -----------------------------------------------------
commit_count = int(git("rev-list", "--count", "HEAD").strip())
# --- branch state ------------------------------------------------------
branch = git("rev-parse", "--abbrev-ref", "HEAD").strip()
branch_head = git("rev-parse", "--short", "HEAD").strip()
ahead = int(git("rev-list", "--count", "origin/main..HEAD").strip())
behind = int(git("rev-list", "--count", "HEAD..origin/main").strip())
remote_heads = {}
for line in git("branch", "-r", "--format=%(refname:short)").splitlines():
if not line.strip():
continue
remote_heads[line.strip()] = git("rev-parse", "--short", line.strip()).strip()
# --- tracked file count ------------------------------------------------
tracked_files = git("ls-files").splitlines()
tracked_file_count = len(tracked_files)
# --- unique emails ------------------------------------------------------
emails = git("log", "--format=%ae").splitlines()
total_unique_emails = len(set(e for e in emails if e))
# --- author census ------------------------------------------------------
by_pair = defaultdict(list)
for line in git("log", "--format=%an%x00%ae%x00%ad", "--date=iso-strict").splitlines():
if not line.strip():
continue
name, email, date = line.split("\x00")
by_pair[(name, email)].append(date)
census = []
name_counts = Counter(r[0] for r in by_pair)
email_counts = Counter(r[1] for r in by_pair)
for (name, email), dates in by_pair.items():
census.append({
"name": name, "email": email, "count": len(dates),
"first": min(dates), "last": max(dates),
"name_collides": name_counts[name] > 1,
"email_collides": email_counts[email] > 1,
})
census.sort(key=lambda r: (-r["count"], r["name"].lower(), r["email"]))
# --- top modified files --------------------------------------------------
names = git("log", "--pretty=format:", "--name-only").splitlines()
file_commits = Counter(n for n in names if n.strip())
top_files = [{"path": p, "commits": c} for p, c in file_commits.most_common(60)]
# --- assemble ------------------------------------------------------------
data = {
"stamp": STAMP,
"repo": REPO,
"branch": branch,
"branch_head": branch_head,
"commit_count": commit_count,
"tracked_file_count": tracked_file_count,
"total_unique_emails": total_unique_emails,
"ahead_by_origin_main": ahead,
"behind_by_origin_main": behind,
"remote_heads": remote_heads,
"head": git("rev-parse", "HEAD").strip(),
"author_census": census,
"top_modified_files": top_files,
}
manifest = {
"sha256": hashlib.sha256(
json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest(),
"commit_count": commit_count,
"tracked_file_count": tracked_file_count,
"total_unique_emails": total_unique_emails,
}
# --- write files ----------------------------------------------------------
json_path = f"{OUT}/git-worktree-census-{STAMP}.json"
csv_path = f"{OUT}/git-worktree-census-{STAMP}.csv"
md_path = f"{OUT}/git-worktree-census-{STAMP}.md"
py_path = f"{OUT}/extract_git_worktree_census.py"
check_path = f"{OUT}/CHECK-git-worktree-census-{STAMP}"
with open(json_path, "w") as f:
json.dump(data, f, indent=2)
with open(f"{OUT}/manifest-__init__.py", "w") as f:
f.write("# auto-generated manifest\n")
f.write(f'manifest = {json.dumps(manifest)}\n')
with open(csv_path, "w", newline="") as f:
w = csv.writer(f)
w.writerow(["rank", "path", "commits"])
for i, r in enumerate(top_files, 1):
w.writerow([i, r["path"], r["commits"]])
with open(md_path, "w") as f:
f.write("# Troupe worktree census\n\n")
f.write(f"- **Stamp:** `{STAMP}`\n")
f.write(f"- **Repo:** `{REPO}`\n")
f.write(f"- **Branch:** `{branch}` (head `{branch_head}`)\n")
f.write(f"- **Commit count:** {commit_count}\n")
f.write(f"- **Tracked file count:** {tracked_file_count}\n")
f.write(f"- **Unique committer emails:** {total_unique_emails}\n")
f.write(f"- **Ahead by origin/main:** {ahead} | **Behind by origin/main:** {behind}\n")
f.write("\n## Author census\n\n")
f.write("| rank | name | email | commits | first | last | name_collides | email_collides |\n")
f.write("|---:|---|---:|---:|---|---:|---|---|\n")
for i, r in enumerate(census, 1):
f.write(f"| {i} | {r['name']} | {r['email']} | {r['count']} | "
f"{r['first']} | {r['last']} | {'yes' if r['name_collides'] else 'no'} | "
f"{'yes' if r['email_collides'] else 'no'} |\n")
f.write("\n## Top modified files\n\n")
f.write("| rank | path | commits |\n")
f.write("|---:|---|---:|\n")
for i, r in enumerate(top_files, 1):
f.write(f"| {i} | {r['path']} | {r['commits']} |\n")
with open(py_path, "w") as f:
f.write(__doc__)
f.write("REPO = %r\n" % REPO)
f.write("STAMP = %r\n" % STAMP)
f.write("OUT = %r\n" % OUT)
f.write("PROFILE_OUT = %r\n" % PROFILE_OUT)
f.write("if __name__ == '__main__':\n raise SystemExit('use extract_git_worktree_census.py directly')\n")
with open(check_path, "w") as f:
f.write(CHECK_TEMPLATE)
# --- mirror into the profile outputs dir ------------------------------------
for p in (json_path, csv_path, md_path, py_path, check_path):
shutil.copy2(p, p.replace(OUT, PROFILE_OUT, 1))
print(json_path)
print(csv_path)
print(md_path)
print(check_path)
CHECK_TEMPLATE = r"""#!/usr/bin/env bash
# CHECK — git worktree census re-derivation verification.
#
# Re-derives the census from live `git` (current checkout) and compares it,
# field-for-field, against the committed git-worktree-census-2026-10-30.json.
# Exits 0 only on exact match.
set -u
REPO=/root/troupe-standup
OUT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
JSON="$OUT/git-worktree-census-2026-10-30.json"
python3 - "$JSON" <<'PY'
import json, subprocess, sys
from collections import Counter
json_path = sys.argv[1]
repo = "/root/troupe-standup"
def git(*args):
return subprocess.run(["git", "-C", repo, *args],
capture_output=True, text=True, check=True).stdout
def derive():
data = {}
data["commit_count"] = int(git("rev-list", "--count", "HEAD").strip())
data["tracked_file_count"] = len(git("ls-files").splitlines())
data["total_unique_emails"] = len(set(e for e in git("log", "--format=%ae").splitlines() if e))
data["ahead_by_origin_main"] = int(git("rev-list", "--count", "origin/main..HEAD").strip())
data["behind_by_origin_main"] = int(git("rev-list", "--count", "HEAD..origin/main").strip())
data["branch_head"] = git("rev-parse", "--short", "HEAD").strip()
data["branch"] = git("rev-parse", "--abbrev-ref", "HEAD").strip()
data["remote_heads"] = {
l: git("rev-parse", "--short", l).strip()
for l in git("branch", "-r", "--format=%(refname:short)").splitlines() if l.strip()
}
by_pair = {}
for line in git("log", "--format=%an%x00%ae%x00%ad", "--date=iso-strict").splitlines():
if not line.strip():
continue
name, email, date = line.split("\x00")
by_pair.setdefault((name, email), []).append(date)
rows = []
for (name, email), dates in by_pair.items():
rows.append({"name": name, "email": email, "count": len(dates),
"first": min(dates), "last": max(dates)})
nc = Counter(r["name"] for r in rows)
ec = Counter(r["email"] for r in rows)
for r in rows:
r["name_collides"] = nc[r["name"]] > 1
r["email_collides"] = ec[r["email"]] > 1
rows.sort(key=lambda r: (-r["count"], r["name"].lower(), r["email"]))
data["author_census"] = rows
names = git("log", "--pretty=format:", "--name-only").splitlines()
fc = Counter(n for n in names if n.strip())
data["top_modified_files"] = [{"path": p, "commits": c} for p, c in fc.most_common(60)]
return data
derived = derive()
committed = json.load(open(json_path))
fields = ["commit_count", "tracked_file_count", "total_unique_emails",
"ahead_by_origin_main", "behind_by_origin_main", "branch_head",
"branch", "remote_heads"]
for k in fields:
if derived.get(k) != committed.get(k):
print(" FIELD %r MISMATCH: derived=%r committed=%r" %
(k, derived.get(k), committed.get(k)), file=sys.stderr)
sys.exit(1)
if len(derived["author_census"]) != len(committed["author_census"]):
print(" CENSUS LENGTH MISMATCH: derived %d rows, committed %d rows" %
(len(derived["author_census"]), len(committed["author_census"])), file=sys.stderr)
sys.exit(1)
for i, (e, c) in enumerate(zip(derived["author_census"], committed["author_census"])):
for k in ("name", "email", "count", "first", "last", "name_collides", "email_collides"):
if e.get(k) != c.get(k):
print(" CENSUS ROW %d FIELD %r MISMATCH: derived=%r committed=%r" %
(i, k, e.get(k), c.get(k)), file=sys.stderr)
sys.exit(1)
if derived["top_modified_files"] != committed["top_modified_files"]:
print(" TOP_FILES MISMATCH", file=sys.stderr)
sys.exit(1)
print("[CHECK] derived from live git and compared field-for-field against %s — PASS" % json_path)
PY
"""
if __name__ == "__main__":
main()