"""Shared helpers for the neckbeard field-test inventory scripts. Determinism rules obeyed by every consumer of this module: * No wall-clock time enters any output file. Where an output needs a "days since" figure, it is measured against REFERENCE_DATE, which is the committer date of the management repo's HEAD -- a property of the tree under analysis, not of the moment the script runs. * Every collection written out is sorted by an explicit key. * Paths in outputs are repo-relative, never absolute. """ import os import re import subprocess from datetime import date from pathlib import Path # Repo layout. WORKSPACE holds the management clone plus a components/ # directory; override with NB_WORKSPACE when the checkout lives elsewhere. MGMT_REPO = Path(__file__).resolve().parents[2] WORKSPACE = Path(os.environ.get("NB_WORKSPACE", MGMT_REPO.parent)) COMPONENTS_DIR = WORKSPACE / "components" DATA_DIR = MGMT_REPO / "analysis" / "data" GITLAB_HOST = "https://git.lab" GROUP = "axion1337.chat" # Frozen in analysis/SCOPE.md, confirmed by the human at the Phase-0 STOP. COMPONENTS = [ ("threadnet-call", "ThreadNet Call"), ("thread-net-git", "ThreadNet Git"), ("threadnet-operating", "ThreadNet Operating"), ("axion1337.chat-gitops", "ThreadNet Server Suite"), ("ThreadNet-Web", "ThreadNet Web"), ] # Named out of scope as analysis targets by the human. References into # them are still recorded, tagged points-outside-scope. OUT_OF_SCOPE_PROJECTS = ["game-operating", "gameserver"] OUT_OF_SCOPE_SUBGROUPS = ["vendor", "Archiv"] # Tracked-but-vendored paths, excluded from trees and scans. Deliberately # narrow: an earlier broader pattern also excluded build/ and dist/, which # silently dropped 7 tracked files -- ThreadNet-Web keeps its desktop icons # under apps/desktop/build/, tracked source assets, not build output. That # made this analysis flag documented icon paths as missing. VENDORED = re.compile(r"(^|/)(node_modules|\.yarn)/") DOC_SUFFIXES = {".md", ".rst"} # The management repo is measured at its default branch, never at the # analysis branch: otherwise this analysis observes its own commits and # its own analysis/ files and reports them as project reality. BASELINE_REF = {"management": "main"} ANALYSIS_DIR = "analysis/" def repos(): """(name, path) for the management repo and every in-scope component.""" out = [("management", MGMT_REPO)] out += [(slug, COMPONENTS_DIR / slug) for slug, _ in COMPONENTS] return out def ref_of(name): return BASELINE_REF.get(name, "HEAD") def assert_untouched(): """Fail loudly if this session modified anything outside analysis/. The mandate forbids changing any existing file. That is checked here mechanically rather than trusted, on every run. """ ref = ref_of("management") diff = git(MGMT_REPO, "diff", "--name-only", ref, "--", ".", f":(exclude){ANALYSIS_DIR}") untracked = git(MGMT_REPO, "ls-files", "--others", "--exclude-standard", "--", ".", f":(exclude){ANALYSIS_DIR}") offending = [p for p in (diff + untracked).split("\n") if p.strip()] if offending: raise SystemExit( "ABORT: this session changed files outside analysis/, which the " "field-test mandate forbids:\n " + "\n ".join(offending)) def git(repo, *args, env_tz=None): """Run git in repo, return stdout as text. Raises on non-zero exit. env_tz pins the timezone git renders dates in. Date rules in this project are stated in UTC; rendering them in the machine's local zone silently misclassifies every commit. """ env = None if env_tz: env = dict(os.environ, TZ=env_tz) res = subprocess.run( ["git", "-C", str(repo), *args], capture_output=True, text=True, errors="replace", env=env, ) if res.returncode != 0: raise RuntimeError(f"git {' '.join(args)} in {repo}: {res.stderr.strip()}") return res.stdout def tracked_files(name, repo, skip_vendored=True): """Sorted repo-relative paths tracked at the repo's baseline ref. Reads the ref, not the index, so the analysis branch's own files stay out of the inventory. """ files = [f for f in git(repo, "ls-tree", "-r", "-z", "--name-only", ref_of(name)).split("\0") if f] files = [f for f in files if not f.startswith(ANALYSIS_DIR)] if skip_vendored: files = [f for f in files if not VENDORED.search(f)] return sorted(files) def last_commit(name, repo, path): """(iso_date, author_name, short_sha) of the newest commit touching path.""" out = git(repo, "log", "-1", "--format=%ad\t%an\t%h", "--date=short", ref_of(name), "--", path) if not out.strip(): return ("", "", "") return tuple(out.strip().split("\t")) def _reference_date(): out = git(MGMT_REPO, "log", "-1", "--format=%cd", "--date=short", ref_of("management")) y, m, d = (int(x) for x in out.strip().split("-")) return date(y, m, d) REFERENCE_DATE = _reference_date() def days_since(iso_date): """Whole days from iso_date to REFERENCE_DATE; '' when the date is empty.""" if not iso_date: return "" y, m, d = (int(x) for x in iso_date.split("-")) return (REFERENCE_DATE - date(y, m, d)).days def cell(text): """Make a value safe for a TSV cell: no tabs, no newlines, trimmed.""" return str(text).replace("\t", " ").replace("\r", " ").replace("\n", " ").strip() def write_tsv(name, header, rows): """Write sorted rows to analysis/data/, tab separated.""" DATA_DIR.mkdir(parents=True, exist_ok=True) path = DATA_DIR / name with path.open("w", encoding="utf-8") as fh: fh.write("\t".join(header) + "\n") for row in rows: fh.write("\t".join(cell(v) for v in row) + "\n") print(f" {name}: {len(rows)} rows") return path