#!/usr/bin/env python3 """List DoX quality debt for a repo (STATE-WP-0077-T02). Scans local workplan frontmatter and intake YAML blocks; optionally lists hub intakes (vetted/routed) without DoC-Ok in notes. Usage: python scripts/quality_debt.py --here python scripts/quality_debt.py --repo-path /path/to/repo statehub quality-debt [--repo-path PATH] [--json] [--api-base URL] Exit 0 always (report only); use --strict for exit 1 when debt found. """ from __future__ import annotations import argparse import json import os import re import sys import urllib.error import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from quality_assessment import ( # noqa: E402 QualityDebtItem, debt_for_intake_meta, debt_for_workplan_meta, intake_has_doc_ok, ) try: import yaml except ImportError: # pragma: no cover yaml = None # type: ignore _SKIP_DIRS = frozenset({".git", "node_modules", ".venv", "history", "agents_backup", "dist"}) def _parse_frontmatter(text: str) -> dict: if not text.startswith("---"): return {} parts = text.split("---", 2) if len(parts) < 3: return {} raw = parts[1] if yaml is not None: try: data = yaml.safe_load(raw) or {} return data if isinstance(data, dict) else {} except Exception: pass # minimal fallback: key: value lines meta: dict = {} for line in raw.splitlines(): if ":" not in line: continue k, _, v = line.partition(":") meta[k.strip()] = v.strip().strip('"').strip("'") return meta def _scan_workplans(repo_dir: Path) -> list[QualityDebtItem]: items: list[QualityDebtItem] = [] wp_dir = repo_dir / "workplans" if not wp_dir.is_dir(): return items for path in sorted(wp_dir.rglob("*.md")): if path.name == "README.md": continue if "archived" in path.parts: # still scan finished debt in archived? skip archived for noise continue try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: continue meta = _parse_frontmatter(text) if not meta: continue rel = str(path.relative_to(repo_dir)) items.extend(debt_for_workplan_meta(meta, path=rel)) return items def _scan_intake_blocks(repo_dir: Path) -> list[QualityDebtItem]: items: list[QualityDebtItem] = [] if yaml is None: return items fence = re.compile(r"```ya?ml\n(.*?)```", re.DOTALL | re.IGNORECASE) for path in sorted(repo_dir.rglob("*.md")): if any(p in _SKIP_DIRS for p in path.parts): continue try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: continue rel = str(path.relative_to(repo_dir)) for m in fence.finditer(text): block = m.group(1) try: data = yaml.safe_load(block) except Exception: continue if not isinstance(data, dict): continue rid = str(data.get("id", "")) # intake ids: PREFIX-IN-NNNN or grandfathered AWQ- if not re.match(r"^[A-Z]+-IN-\d+", rid) and not re.match(r"^AWQ-\d+", rid): if data.get("kind") != "intake": continue items.extend(debt_for_intake_meta(data, path=rel)) return items def _hub_intake_debt(api_base: str) -> list[QualityDebtItem]: items: list[QualityDebtItem] = [] try: req = urllib.request.Request( f"{api_base.rstrip('/')}/intakes/", headers={"Accept": "application/json"}, ) with urllib.request.urlopen(req, timeout=15) as resp: rows = json.loads(resp.read()) except Exception as e: items.append( QualityDebtItem( kind="intake", record_id="(hub)", lifecycle_status="?", missing="DoC-Ok", path="api:/intakes/", detail=f"could not list intakes: {e}", ) ) return items if not isinstance(rows, list): return items for row in rows: if row.get("status") not in ("vetted", "routed"): continue if intake_has_doc_ok(row): continue items.append( QualityDebtItem( kind="intake", record_id=str(row.get("id", "")), lifecycle_status=str(row.get("status", "")), missing="DoC-Ok", path="api:/intakes/", detail=(row.get("title") or "")[:80], ) ) return items def collect( repo_dir: Path, *, api_base: str | None = None, include_hub_intakes: bool = True, ) -> list[QualityDebtItem]: items = _scan_workplans(repo_dir) + _scan_intake_blocks(repo_dir) if include_hub_intakes and api_base: items.extend(_hub_intake_debt(api_base)) return items def main() -> None: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--here", action="store_true", help="Use cwd as repo") ap.add_argument("--repo-path", default=None, help="Repo root (default: cwd)") ap.add_argument("--api-base", default=os.environ.get("API_BASE", "http://127.0.0.1:8000")) ap.add_argument("--no-hub", action="store_true", help="Skip hub intake list") ap.add_argument("--json", action="store_true", dest="as_json") ap.add_argument("--strict", action="store_true", help="Exit 1 if any debt") args = ap.parse_args() repo_dir = Path(args.repo_path or os.getcwd()).expanduser().resolve() items = collect( repo_dir, api_base=None if args.no_hub else args.api_base, include_hub_intakes=not args.no_hub, ) if args.as_json: print( json.dumps( [ { "kind": i.kind, "record_id": i.record_id, "lifecycle_status": i.lifecycle_status, "missing": i.missing, "path": i.path, "detail": i.detail, } for i in items ], indent=2, ) ) else: print(f"Quality debt — {repo_dir}") print(f" ({len(items)} item(s); lifecycle may advance without DoX-Ok)\n") if not items: print(" (none)") for i in items: print( f" [{i.kind}] {i.record_id} status={i.lifecycle_status} " f"missing {i.missing} ({i.path}) {i.detail}" ) print( "\nRecord assessments with quality_dor / quality_dod / quality_doc " "fields — see docs/work-record-quality-gates.md" ) if args.strict and items: sys.exit(1) sys.exit(0) if __name__ == "__main__": main()