the-custodian/tools/scan_workstream_terminology.py

#!/usr/bin/env python3
"""Scan registered repos for legacy ``workstream`` terminology.

Usage:
    python tools/scan_workstream_terminology.py [--api-base URL] [--json]
    python tools/scan_workstream_terminology.py --repo the-custodian

Reads repo slugs from State Hub ``GET /repos/`` and scans checkouts under
``/home/worsch/<slug>`` (or host_paths when available). Excludes VCS/vendor
directories. Emits per-repo occurrence counts and pattern buckets for fleet
terminology refactor planning (CUST-WP-0055).
"""
from __future__ import annotations

import argparse
import fnmatch
import json
import re
import sys
import urllib.request
from collections import defaultdict
from pathlib import Path

import yaml

DEFAULT_API_BASE = "http://127.0.0.1:8000"
HOME = Path("/home/worsch")
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_ALLOWLIST_PATH = REPO_ROOT / "tools" / "scan_workstream_allowlist.yaml"
PROSE_PATTERN = "workstream "
EXCLUDE_DIRS = {
    ".git",
    "node_modules",
    "__pycache__",
    ".venv",
    "venv",
    "dist",
    "build",
    ".next",
    "target",
}
TEXT_EXTENSIONS = {
    ".md",
    ".py",
    ".ts",
    ".tsx",
    ".js",
    ".jsx",
    ".yaml",
    ".yml",
    ".json",
    ".sql",
    ".sh",
    ".toml",
    ".rs",
    ".go",
    ".java",
    ".html",
    ".css",
    ".scss",
    ".vue",
    ".rb",
    ".php",
    ".cs",
    ".swift",
    ".kt",
    ".hcl",
    ".tf",
    ".mdx",
    ".jinja",
    ".j2",
    ".template",
    ".cfg",
    ".ini",
    ".txt",
    ".graphql",
    ".proto",
}
PATTERN_LABELS = [
    ("state_hub_workstream_id", "frontmatter:state_hub_workstream_id"),
    ("workstream_id", "api:workstream_id"),
    ("create_workstream", "mcp:create_workstream"),
    ("update_workstream", "mcp:update_workstream"),
    ("list_workstreams", "mcp:list_workstreams"),
    ("open_workstreams", "api:open_workstreams"),
    ("workstreams/", "path:workstreams/"),
    ("workstream ", "prose:workstream "),
    ("workstreams", "term:workstreams"),
]
NAMED_FILES = {"Makefile", "Dockerfile", "AGENTS.md", "CLAUDE.md", "SCOPE.md", "INTENT.md", "README"}


def load_allowlist(path: Path = DEFAULT_ALLOWLIST_PATH) -> dict:
    if not path.is_file():
        return {"version": 0, "global": {}, "per_repo": {}, "patterns_always_allowed": []}
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    data.setdefault("global", {})
    data.setdefault("per_repo", {})
    data.setdefault("patterns_always_allowed", [])
    return data


def _matches_any(rel_path: str, prefixes: list[str], globs: list[str]) -> bool:
    for prefix in prefixes:
        if rel_path == prefix or rel_path.startswith(prefix):
            return True
    for pattern in globs:
        if fnmatch.fnmatch(rel_path, pattern):
            return True
    return False


def path_is_excluded(repo_slug: str, rel_path: str, allowlist: dict) -> bool:
    repo_rules = allowlist.get("per_repo", {}).get(repo_slug, {})
    if repo_rules.get("exclude_repo"):
        return True
    global_rules = allowlist.get("global", {})
    if _matches_any(
        rel_path,
        global_rules.get("path_prefixes", []),
        global_rules.get("path_globs", []),
    ):
        return True
    return _matches_any(
        rel_path,
        repo_rules.get("path_prefixes", []),
        repo_rules.get("path_globs", []),
    )


def prose_gate_violations(root: Path, repo_slug: str, allowlist: dict) -> list[tuple[str, int]]:
    violations: list[tuple[str, int]] = []
    for path in root.rglob("*"):
        if not path.is_file() or any(part in EXCLUDE_DIRS for part in path.parts):
            continue
        if path.suffix.lower() not in TEXT_EXTENSIONS and path.name not in NAMED_FILES:
            continue
        rel = str(path.relative_to(root))
        if path_is_excluded(repo_slug, rel, allowlist):
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        hits = len(re.findall(re.escape(PROSE_PATTERN), text, flags=re.I))
        if hits:
            violations.append((rel, hits))
    violations.sort(key=lambda item: (-item[1], item[0]))
    return violations


def bucket(path: str) -> str:
    p = path.lower()
    if p.startswith("workplans/"):
        return "workplans"
    if p in {"agents.md", "claude.md", "scope.md", "intent.md", "readme.md"}:
        return "agent-guidance"
    if "/tests/" in p or p.startswith("tests/"):
        return "tests"
    if "/migrations/" in p or p.startswith("migrations/"):
        return "migrations"
    if p.endswith(".py"):
        return "python"
    if p.endswith((".ts", ".tsx", ".js", ".jsx")):
        return "frontend"
    if p.endswith((".yaml", ".yml")):
        return "yaml"
    if p.endswith(".md"):
        return "markdown-other"
    return "other"


def fetch_repos(api_base: str) -> list[dict]:
    with urllib.request.urlopen(f"{api_base.rstrip('/')}/repos/") as resp:
        return json.load(resp)


def scan_repo(
    root: Path,
    *,
    repo_slug: str | None = None,
    allowlist: dict | None = None,
    apply_allowlist: bool = False,
) -> dict:
    patterns: dict[str, int] = defaultdict(int)
    buckets: dict[str, int] = defaultdict(int)
    total = 0
    files = 0
    excluded_files = 0
    top_files: list[tuple[int, str]] = []
    for path in root.rglob("*"):
        if not path.is_file() or any(part in EXCLUDE_DIRS for part in path.parts):
            continue
        if path.suffix.lower() not in TEXT_EXTENSIONS and path.name not in NAMED_FILES:
            continue
        rel = str(path.relative_to(root))
        if apply_allowlist and allowlist is not None and repo_slug and path_is_excluded(repo_slug, rel, allowlist):
            excluded_files += 1
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        matches = re.findall(r"workstream", text, flags=re.I)
        if not matches:
            continue
        count = len(matches)
        total += count
        files += 1
        top_files.append((count, rel))
        buckets[bucket(rel)] += count
        low = text.lower()
        for needle, label in PATTERN_LABELS:
            hits = low.count(needle.lower())
            if hits:
                patterns[label] += hits
    top_files.sort(reverse=True)
    result = {
        "occurrences": total,
        "files": files,
        "patterns": dict(patterns),
        "buckets": dict(buckets),
        "top_files": top_files[:10],
    }
    if apply_allowlist:
        result["excluded_files"] = excluded_files
    return result


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api-base", default=DEFAULT_API_BASE)
    parser.add_argument("--repo", help="Scan a single repo slug")
    parser.add_argument("--json", action="store_true", help="Emit JSON report")
    parser.add_argument(
        "--apply-allowlist",
        action="store_true",
        help="Exclude allowlisted paths from counts (CUST-WP-0055 T08)",
    )
    parser.add_argument(
        "--allowlist",
        default=str(DEFAULT_ALLOWLIST_PATH),
        help="Path to scan allowlist YAML",
    )
    parser.add_argument(
        "--check-prose-gate",
        action="store_true",
        help="Exit 1 when prose 'workstream ' hits remain outside the allowlist",
    )
    args = parser.parse_args()

    allowlist = load_allowlist(Path(args.allowlist))
    repos = fetch_repos(args.api_base)
    if args.repo:
        repos = [r for r in repos if r["slug"] == args.repo]
        if not repos:
            print(f"Repo '{args.repo}' not registered", file=sys.stderr)
            return 1

    report: dict = {
        "repos": {},
        "totals": {"occurrences": 0, "files": 0, "repos_with_hits": 0},
        "allowlist_applied": args.apply_allowlist or args.check_prose_gate,
    }
    pattern_totals: dict[str, int] = defaultdict(int)
    prose_violations: dict[str, list[dict[str, int | str]]] = {}

    for repo in sorted(repos, key=lambda r: r["slug"]):
        slug = repo["slug"]
        root = HOME / slug
        if not root.is_dir():
            report["repos"][slug] = {"missing_checkout": True}
            continue
        if allowlist.get("per_repo", {}).get(slug, {}).get("exclude_repo"):
            report["repos"][slug] = {"excluded_repo": True}
            continue
        data = scan_repo(
            root,
            repo_slug=slug,
            allowlist=allowlist,
            apply_allowlist=args.apply_allowlist,
        )
        if args.check_prose_gate:
            violations = prose_gate_violations(root, slug, allowlist)
            if violations:
                prose_violations[slug] = [
                    {"path": path, "hits": hits} for path, hits in violations
                ]
        if data["occurrences"]:
            report["repos"][slug] = data
            report["totals"]["occurrences"] += data["occurrences"]
            report["totals"]["files"] += data["files"]
            report["totals"]["repos_with_hits"] += 1
            for label, count in data["patterns"].items():
                pattern_totals[label] += count

    report["pattern_totals"] = dict(sorted(pattern_totals.items(), key=lambda x: -x[1]))
    if prose_violations:
        report["prose_gate_violations"] = prose_violations

    if args.check_prose_gate and prose_violations:
        if args.json:
            print(json.dumps(report, indent=2))
        else:
            print("Prose gate failed — 'workstream ' hits outside allowlist:", file=sys.stderr)
            for slug, rows in sorted(prose_violations.items()):
                for row in rows[:10]:
                    print(f"  {slug}: {row['hits']:3d}  {row['path']}", file=sys.stderr)
        return 1

    if args.json:
        print(json.dumps(report, indent=2))
        return 0

    totals = report["totals"]
    print(f"Repos scanned: {len(repos)}")
    print(f"Repos with hits: {totals['repos_with_hits']}")
    print(f"Total occurrences: {totals['occurrences']}")
    print(f"Total files: {totals['files']}")
    print("\nTop repos:")
    ranked = sorted(
        ((slug, data) for slug, data in report["repos"].items() if data.get("occurrences")),
        key=lambda item: item[1]["occurrences"],
        reverse=True,
    )
    for slug, data in ranked[:25]:
        print(f"  {data['occurrences']:5d} in {data['files']:4d} files  {slug}")
    print("\nPattern totals:")
    for label, count in report["pattern_totals"].items():
        print(f"  {count:5d}  {label}")
    missing = [slug for slug, data in report["repos"].items() if data.get("missing_checkout")]
    if missing:
        print("\nMissing checkouts:")
        for slug in missing:
            print(f"  {slug}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())