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 json
import re
import sys
import urllib.request
from collections import defaultdict
from pathlib import Path

DEFAULT_API_BASE = "http://127.0.0.1:8000"
HOME = Path("/home/worsch")
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 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) -> dict:
    patterns: dict[str, int] = defaultdict(int)
    buckets: dict[str, int] = defaultdict(int)
    total = 0
    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
        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
        rel = str(path.relative_to(root))
        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)
    return {
        "occurrences": total,
        "files": files,
        "patterns": dict(patterns),
        "buckets": dict(buckets),
        "top_files": top_files[:10],
    }


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")
    args = parser.parse_args()

    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}}
    pattern_totals: dict[str, int] = defaultdict(int)

    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
        data = scan_repo(root)
        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 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())