the-custodian/tools/add_archive_terminology_note.py
codex e0f93304bf
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s
CUST-WP-0055: finish T02/T06/T07 and close workplan
Add archive terminology note tool, extend scan allowlist for generated
trees, mark remaining tasks done, and add grandfather notes to archived
workplans in this repo.
2026-07-08 20:26:24 +02:00

106 lines
No EOL
3.4 KiB
Python
Executable file

#!/usr/bin/env python3
"""Add a terminology grandfather note to archived workplan files (CUST-WP-0055 T07).
Usage:
python tools/add_archive_terminology_note.py --repo the-custodian
python tools/add_archive_terminology_note.py --all-repos
python tools/add_archive_terminology_note.py --repo the-custodian --dry-run
"""
from __future__ import annotations
import argparse
import re
import sys
import urllib.request
from pathlib import Path
HOME = Path("/home/worsch")
DEFAULT_API_BASE = "http://127.0.0.1:8000"
ARCHIVE_DIR = Path("workplans/archived")
NOTE_MARKER = "Terminology note:"
NOTE_BODY = (
"> **Terminology note:** Historical text in this archived workplan may use "
'the legacy term "workstream". The fleet term is **workplan** '
"(`canon/standards/workplan-terminology-fleet_v0.1.md`).\n"
)
FRONTMATTER_RE = re.compile(r"^---\n.*?\n---\n", re.DOTALL)
def list_repos(api_base: str) -> list[str]:
try:
with urllib.request.urlopen(f"{api_base.rstrip('/')}/repos/", timeout=10) as resp:
import json
rows = json.loads(resp.read().decode())
return sorted({r["slug"] for r in rows if r.get("slug")})
except Exception:
return []
def archived_files(repo_root: Path) -> list[Path]:
archive = repo_root / ARCHIVE_DIR
if not archive.is_dir():
return []
return sorted(p for p in archive.rglob("*.md") if p.is_file())
def add_note(text: str) -> tuple[str, bool]:
if NOTE_MARKER in text:
return text, False
match = FRONTMATTER_RE.match(text)
if not match:
return text, False
insert_at = match.end()
return text[:insert_at] + "\n" + NOTE_BODY + text[insert_at:], True
def process_repo(repo_root: Path, *, dry_run: bool) -> tuple[int, int]:
changed = 0
scanned = 0
for path in archived_files(repo_root):
scanned += 1
original = path.read_text(encoding="utf-8")
updated, did_change = add_note(original)
if not did_change:
continue
changed += 1
if dry_run:
print(f"would update: {path}")
else:
path.write_text(updated, encoding="utf-8")
print(f"updated: {path}")
return scanned, changed
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repo", action="append", default=[], help="repo slug under ~/")
parser.add_argument("--all-repos", action="store_true", help="every registered repo")
parser.add_argument("--api-base", default=DEFAULT_API_BASE)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args(argv)
repos = args.repo or (list_repos(args.api_base) if args.all_repos else [])
if not repos:
print("No repos selected.", file=sys.stderr)
return 1
total_scanned = 0
total_changed = 0
for slug in repos:
root = HOME / slug
if not root.is_dir():
print(f"skip missing checkout: {slug}")
continue
scanned, changed = process_repo(root, dry_run=args.dry_run)
total_scanned += scanned
total_changed += changed
if scanned:
print(f"{slug}: {changed}/{scanned} archived files updated")
print(f"done: {total_changed} files updated ({total_scanned} scanned)")
return 0
if __name__ == "__main__":
raise SystemExit(main())