f820ced5c9
Record the flavor backfill cohort, residual-intake join, and agent convention that default views omit residuals.
348 lines
12 KiB
Python
348 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""One-shot CUST-WP-0072-T02 helper: classify open workplans and patch files.
|
|
|
|
Does not invent depends_on edges. Does not promote residuals.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import urllib.request
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
API = "http://127.0.0.1:8000"
|
|
OPEN = {"proposed", "ready", "active", "blocked", "backlog"}
|
|
FLAVORS = {"planning", "implementation", "refactoring", "extension", "residual"}
|
|
WP_ID = re.compile(
|
|
r"\b([A-Z][A-Z0-9]*(?:-[A-Z][A-Z0-9]*)*-WP-(?:ADHOC-)?[0-9]{4}(?:[a-z])?)\b"
|
|
)
|
|
ORIGIN_LINE = re.compile(
|
|
r"Origin:\s*residual\s+from\s+`?([A-Z][A-Z0-9]*(?:-[A-Z][A-Z0-9]*)*-WP-(?:ADHOC-)?[0-9]{4})",
|
|
re.I,
|
|
)
|
|
PLAN_KW = re.compile(
|
|
r"\b(spec|design|canon|policy|review|declare|founding|charter|"
|
|
r"convention|assessment|plan-derived|establish intent)\b",
|
|
re.I,
|
|
)
|
|
IMPL_KW = re.compile(
|
|
r"\b(implement|deploy|admit|migrate|build|provision|activate|"
|
|
r"cut over|restore|bind)\b",
|
|
re.I,
|
|
)
|
|
REF_KW = re.compile(r"\b(refactor|rename|terminology|cleanup|strangler)\b", re.I)
|
|
EXT_KW = re.compile(r"\b(adapt(?:ation)?|extension|follow-on)\b", re.I)
|
|
TITLE_RESIDUAL = re.compile(
|
|
r"(authorized live residuals|residual from|residual of|residual handoff|"
|
|
r"^residual\b)",
|
|
re.I,
|
|
)
|
|
|
|
|
|
def get(path: str):
|
|
with urllib.request.urlopen(API + path, timeout=30) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def parse_fm(text: str) -> tuple[dict, str, str]:
|
|
if not text.startswith("---"):
|
|
return {}, text, ""
|
|
rest = text[3:]
|
|
end = rest.find("\n---")
|
|
if end < 0:
|
|
return {}, text, ""
|
|
raw = rest[:end].lstrip("\n")
|
|
body = rest[end + 4 :]
|
|
loaded = yaml.safe_load(raw) or {}
|
|
if not isinstance(loaded, dict):
|
|
loaded = {}
|
|
return loaded, body, raw
|
|
|
|
|
|
def classify(meta: dict, title: str, body: str, status: str) -> tuple[str, str]:
|
|
existing = str(meta.get("flavor") or "").strip().lower()
|
|
if existing in FLAVORS:
|
|
return existing, "already-set"
|
|
origin = str(meta.get("origin") or "").strip().lower()
|
|
if origin in {"residual", "handoff"}:
|
|
return "residual", "origin"
|
|
if TITLE_RESIDUAL.search(title or "") or ORIGIN_LINE.search(body[:1500] or ""):
|
|
return "residual", "prose"
|
|
if REF_KW.search(title or ""):
|
|
return "refactoring", "keyword"
|
|
if EXT_KW.search(title or "") and not IMPL_KW.search(title or ""):
|
|
return "extension", "keyword"
|
|
if status == "proposed":
|
|
if IMPL_KW.search(title or ""):
|
|
return "implementation", "proposed-implement"
|
|
return "planning", "proposed"
|
|
if PLAN_KW.search(title or "") and not IMPL_KW.search(title or ""):
|
|
return "planning", "keyword"
|
|
return "implementation", "default"
|
|
|
|
|
|
def existing_depends(meta: dict, self_id: str) -> list[str]:
|
|
ids: list[str] = []
|
|
for key in ("depends_on", "depends_on_workplans"):
|
|
val = meta.get(key)
|
|
if isinstance(val, str):
|
|
ids.extend(WP_ID.findall(val))
|
|
elif isinstance(val, list):
|
|
for item in val:
|
|
ids.extend(WP_ID.findall(str(item)))
|
|
ids.extend(WP_ID.findall(str(meta.get("blocked_on") or "")))
|
|
out: list[str] = []
|
|
seen: set[str] = set()
|
|
for item in ids:
|
|
if item == self_id or item in seen:
|
|
continue
|
|
seen.add(item)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
def index_workplan_files(repo_path: Path) -> dict[str, Path]:
|
|
found: dict[str, Path] = {}
|
|
for directory in (repo_path / "workplans", repo_path / "workplans" / "archived"):
|
|
if not directory.is_dir():
|
|
continue
|
|
for path in directory.glob("*.md"):
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except OSError:
|
|
continue
|
|
meta, _, _ = parse_fm(text)
|
|
wid = str(meta.get("id") or "").strip()
|
|
if wid:
|
|
found.setdefault(wid, path)
|
|
return found
|
|
|
|
|
|
def insert_after_status(raw: str, flavor: str) -> str:
|
|
if re.search(r"^flavor:\s*", raw, re.M):
|
|
return raw
|
|
lines = raw.splitlines()
|
|
out = []
|
|
inserted = False
|
|
for line in lines:
|
|
out.append(line)
|
|
if not inserted and re.match(r"^status:\s*", line):
|
|
out.append(f"flavor: {flavor}")
|
|
inserted = True
|
|
if not inserted:
|
|
out.append(f"flavor: {flavor}")
|
|
return "\n".join(out)
|
|
|
|
|
|
def insert_depends_on(raw: str, deps: list[str]) -> str:
|
|
if re.search(r"^depends_on:\s*", raw, re.M):
|
|
return raw
|
|
block = "depends_on:\n" + "\n".join(f" - {item}" for item in deps)
|
|
lines = raw.splitlines()
|
|
out = []
|
|
inserted = False
|
|
for line in lines:
|
|
out.append(line)
|
|
if not inserted and re.match(r"^flavor:\s*", line):
|
|
out.extend(block.splitlines())
|
|
inserted = True
|
|
if not inserted:
|
|
out.extend(block.splitlines())
|
|
return "\n".join(out)
|
|
|
|
|
|
def insert_origin(raw: str, origin: str, origin_ref: str | None) -> str:
|
|
lines = raw.splitlines()
|
|
have_origin = any(re.match(r"^origin:\s*", line) for line in lines)
|
|
have_ref = any(re.match(r"^origin_ref:\s*", line) for line in lines)
|
|
extra = []
|
|
if not have_origin:
|
|
extra.append(f"origin: {origin}")
|
|
if origin_ref and not have_ref:
|
|
extra.append(f"origin_ref: {origin_ref}")
|
|
if not extra:
|
|
return raw
|
|
out = []
|
|
inserted = False
|
|
for line in lines:
|
|
out.append(line)
|
|
if not inserted and re.match(r"^flavor:\s*", line):
|
|
out.extend(extra)
|
|
inserted = True
|
|
if not inserted:
|
|
out.extend(extra)
|
|
return "\n".join(out)
|
|
|
|
|
|
def patch_residual_tasks(body: str) -> tuple[str, int]:
|
|
count = 0
|
|
|
|
def repl(match: re.Match[str]) -> str:
|
|
nonlocal count
|
|
block = match.group(0)
|
|
inner = match.group(1)
|
|
if re.search(r"^flavor:\s*", inner, re.M):
|
|
return block
|
|
inner_lines = inner.splitlines()
|
|
new_inner = []
|
|
inserted = False
|
|
for line in inner_lines:
|
|
new_inner.append(line)
|
|
if not inserted and re.match(r"^status:\s*", line):
|
|
new_inner.append("flavor: residual")
|
|
inserted = True
|
|
count += 1
|
|
if not inserted:
|
|
new_inner.append("flavor: residual")
|
|
count += 1
|
|
return "```task\n" + "\n".join(new_inner) + "\n```"
|
|
|
|
patched = re.sub(r"```task\n(.*?)```", repl, body, flags=re.S)
|
|
return patched, count
|
|
|
|
|
|
def rebuild(raw: str, body: str) -> str:
|
|
if not raw.endswith("\n"):
|
|
raw += "\n"
|
|
if body.startswith("\n"):
|
|
return f"---\n{raw}---{body}"
|
|
return f"---\n{raw}---\n{body}"
|
|
|
|
|
|
def main() -> None:
|
|
repos = {row["id"]: row for row in get("/repos/")}
|
|
open_wps = [
|
|
row
|
|
for row in get("/workplans/")
|
|
if (row.get("status") or "").lower() in OPEN
|
|
and "@retired" not in (row.get("slug") or "")
|
|
]
|
|
coverage = []
|
|
changed_by_repo: dict[str, list[str]] = defaultdict(list)
|
|
stats = Counter()
|
|
|
|
for wp in open_wps:
|
|
repo = repos.get(wp.get("repo_id") or "") or {}
|
|
repo_slug = repo.get("slug") or ""
|
|
local = repo.get("local_path")
|
|
record = {
|
|
"repo": repo_slug,
|
|
"hub_slug": wp.get("slug"),
|
|
"hub_status": wp.get("status"),
|
|
"title": wp.get("title"),
|
|
"file": "",
|
|
"id": "",
|
|
"flavor": "",
|
|
"flavor_reason": "",
|
|
"depends_on": "",
|
|
"origin": "",
|
|
"origin_ref": "",
|
|
"changed": "no",
|
|
"note": "",
|
|
}
|
|
if not local or not Path(local).is_dir():
|
|
record["note"] = "no-local-path"
|
|
stats["skip_no_path"] += 1
|
|
coverage.append(record)
|
|
continue
|
|
files = index_workplan_files(Path(local))
|
|
# Prefer backing path when it exists
|
|
file_path = None
|
|
rel = wp.get("backing_relative_path")
|
|
fname = wp.get("backing_filename")
|
|
if rel and (Path(local) / rel).is_file():
|
|
file_path = Path(local) / rel
|
|
elif fname and (Path(local) / "workplans" / fname).is_file():
|
|
file_path = Path(local) / "workplans" / fname
|
|
text = ""
|
|
meta: dict = {}
|
|
if file_path:
|
|
text = file_path.read_text(encoding="utf-8")
|
|
meta, _, _ = parse_fm(text)
|
|
wid = str(meta.get("id") or "").strip()
|
|
if not wid:
|
|
# match by scanning index using hub slug upper
|
|
guess = (wp.get("slug") or "").upper().replace("_", "-")
|
|
file_path = files.get(guess)
|
|
if file_path:
|
|
text = file_path.read_text(encoding="utf-8")
|
|
meta, _, _ = parse_fm(text)
|
|
wid = str(meta.get("id") or "").strip()
|
|
if not file_path or not wid:
|
|
record["note"] = "no-file"
|
|
stats["skip_no_file"] += 1
|
|
coverage.append(record)
|
|
continue
|
|
|
|
body_holder = parse_fm(text)
|
|
meta, body, raw = body_holder
|
|
title = str(meta.get("title") or wp.get("title") or "")
|
|
status = str(meta.get("status") or wp.get("status") or "").lower()
|
|
flavor, reason = classify(meta, title, body, status)
|
|
deps = existing_depends(meta, wid)
|
|
origin = str(meta.get("origin") or "").strip()
|
|
origin_ref = str(meta.get("origin_ref") or "").strip()
|
|
origin_match = ORIGIN_LINE.search(body[:2000])
|
|
new_raw = raw
|
|
new_body = body
|
|
changed = False
|
|
task_patches = 0
|
|
|
|
if str(meta.get("flavor") or "").strip().lower() not in FLAVORS:
|
|
new_raw = insert_after_status(new_raw, flavor)
|
|
changed = True
|
|
if deps and not meta.get("depends_on"):
|
|
new_raw = insert_depends_on(new_raw, deps)
|
|
changed = True
|
|
if flavor == "residual":
|
|
if not origin and origin_match:
|
|
origin = "residual"
|
|
if not origin_ref and origin_match:
|
|
origin_ref = origin_match.group(1)
|
|
if origin == "residual" or origin_ref:
|
|
before = new_raw
|
|
new_raw = insert_origin(new_raw, origin or "residual", origin_ref or None)
|
|
if new_raw != before:
|
|
changed = True
|
|
new_body, task_patches = patch_residual_tasks(new_body)
|
|
if task_patches:
|
|
changed = True
|
|
|
|
record.update(
|
|
{
|
|
"file": str(file_path),
|
|
"id": wid,
|
|
"flavor": flavor,
|
|
"flavor_reason": reason,
|
|
"depends_on": ",".join(deps) if deps else "none",
|
|
"origin": origin,
|
|
"origin_ref": origin_ref,
|
|
}
|
|
)
|
|
if changed:
|
|
file_path.write_text(rebuild(new_raw, new_body), encoding="utf-8")
|
|
record["changed"] = "yes"
|
|
if task_patches:
|
|
record["note"] = f"task_flavor={task_patches}"
|
|
changed_by_repo[repo_slug].append(str(file_path))
|
|
stats["changed"] += 1
|
|
else:
|
|
stats["unchanged"] += 1
|
|
coverage.append(record)
|
|
|
|
out_dir = Path(__file__).resolve().parent
|
|
(out_dir / "coverage.json").write_text(
|
|
json.dumps({"stats": dict(stats), "rows": coverage}, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
(out_dir / "changed-by-repo.json").write_text(
|
|
json.dumps(changed_by_repo, indent=2) + "\n", encoding="utf-8"
|
|
)
|
|
print(json.dumps({"stats": dict(stats), "repos_changed": len(changed_by_repo)}, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|