123 lines
3.5 KiB
Python
123 lines
3.5 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Mechanical workstream→workplan prose sweep for agent guidance (CUST-WP-0055 T04).
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
python tools/sweep_workstream_prose.py --repo ops-warden
|
||
|
|
python tools/sweep_workstream_prose.py --repo ops-warden --repo hub-core --dry-run
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import re
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
HOME = Path("/home/worsch")
|
||
|
|
TARGET_NAMES = {"AGENTS.md", "SCOPE.md", "INTENT.md", "README.md", "CLAUDE.md"}
|
||
|
|
TARGET_RULES = ".claude/rules"
|
||
|
|
|
||
|
|
PRESERVE = (
|
||
|
|
"state_hub_workstream_id",
|
||
|
|
"workstream_id",
|
||
|
|
"workstream_slug",
|
||
|
|
"workstream_title",
|
||
|
|
"workstream_status",
|
||
|
|
"open_workstreams",
|
||
|
|
"create_workstream",
|
||
|
|
"update_workstream",
|
||
|
|
"list_workstreams",
|
||
|
|
"update_workstream_status",
|
||
|
|
"/workstreams",
|
||
|
|
"workstreams/",
|
||
|
|
"workstream-dod",
|
||
|
|
"workstream-health",
|
||
|
|
"workstream-lifecycle",
|
||
|
|
"workstream-kpi",
|
||
|
|
"related_workstream",
|
||
|
|
"by_workstream",
|
||
|
|
)
|
||
|
|
|
||
|
|
REPLACEMENTS = (
|
||
|
|
(re.compile(r"\bWorkstreams\b"), "Workplans"),
|
||
|
|
(re.compile(r"\bworkstreams\b"), "workplans"),
|
||
|
|
(re.compile(r"\bWorkstream\b"), "Workplan"),
|
||
|
|
(re.compile(r"\bworkstream\b"), "workplan"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _protect(text: str) -> tuple[str, dict[str, str]]:
|
||
|
|
tokens: dict[str, str] = {}
|
||
|
|
|
||
|
|
def repl(match: re.Match[str]) -> str:
|
||
|
|
key = f"__KEEP_{len(tokens)}__"
|
||
|
|
tokens[key] = match.group(0)
|
||
|
|
return key
|
||
|
|
|
||
|
|
for term in sorted(PRESERVE, key=len, reverse=True):
|
||
|
|
text = re.sub(re.escape(term), repl, text)
|
||
|
|
return text, tokens
|
||
|
|
|
||
|
|
|
||
|
|
def _restore(text: str, tokens: dict[str, str]) -> str:
|
||
|
|
for key, value in tokens.items():
|
||
|
|
text = text.replace(key, value)
|
||
|
|
return text
|
||
|
|
|
||
|
|
|
||
|
|
def sweep_text(text: str) -> str:
|
||
|
|
protected, tokens = _protect(text)
|
||
|
|
updated = protected
|
||
|
|
for pattern, replacement in REPLACEMENTS:
|
||
|
|
updated = pattern.sub(replacement, updated)
|
||
|
|
return _restore(updated, tokens)
|
||
|
|
|
||
|
|
|
||
|
|
def target_files(repo_root: Path) -> list[Path]:
|
||
|
|
paths: list[Path] = []
|
||
|
|
for name in TARGET_NAMES:
|
||
|
|
path = repo_root / name
|
||
|
|
if path.is_file():
|
||
|
|
paths.append(path)
|
||
|
|
rules_dir = repo_root / TARGET_RULES
|
||
|
|
if rules_dir.is_dir():
|
||
|
|
paths.extend(sorted(rules_dir.glob("*.md")))
|
||
|
|
return paths
|
||
|
|
|
||
|
|
|
||
|
|
def sweep_repo(repo_root: Path, *, dry_run: bool) -> list[str]:
|
||
|
|
changed: list[str] = []
|
||
|
|
for path in target_files(repo_root):
|
||
|
|
original = path.read_text(encoding="utf-8")
|
||
|
|
updated = sweep_text(original)
|
||
|
|
if updated != original:
|
||
|
|
changed.append(str(path.relative_to(repo_root)))
|
||
|
|
if not dry_run:
|
||
|
|
path.write_text(updated, encoding="utf-8")
|
||
|
|
return changed
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("--repo", action="append", required=True, help="Repo slug under ~/")
|
||
|
|
parser.add_argument("--dry-run", action="store_true")
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
exit_code = 0
|
||
|
|
for slug in args.repo:
|
||
|
|
root = HOME / slug
|
||
|
|
if not root.is_dir():
|
||
|
|
print(f"SKIP missing checkout: {slug}")
|
||
|
|
exit_code = 1
|
||
|
|
continue
|
||
|
|
changed = sweep_repo(root, dry_run=args.dry_run)
|
||
|
|
if changed:
|
||
|
|
mode = "would update" if args.dry_run else "updated"
|
||
|
|
print(f"{slug}: {mode} {len(changed)} file(s)")
|
||
|
|
for rel in changed:
|
||
|
|
print(f" {rel}")
|
||
|
|
else:
|
||
|
|
print(f"{slug}: no changes")
|
||
|
|
return exit_code
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|