management/analysis/scripts/common.py

"""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.
VENDORED = re.compile(r"(^|/)(node_modules|dist|build|\.yarn|vendor)/")

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/<name>, 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