데이터 게이트 진척도 검증 체인 추가

tools/validate_data_gated_progress_v1.py

@@ -0,0 +1,89 @@
+"""validate_data_gated_progress_v1.py — DATA_GATED_PROGRESS_VALIDATE_V1
+
+Temp/data_gated_progress_v1.json의 기본 계약을 검증한다.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_INPUT = ROOT / "Temp" / "data_gated_progress_v1.json"
+DEFAULT_OUT = ROOT / "Temp" / "validate_data_gated_progress_v1.json"
+FORMULA_ID = "DATA_GATED_PROGRESS_VALIDATE_V1"
+
+
+def _load(path: Path) -> Any:
+    if not path.exists():
+        return {}
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return {}
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--input", default=str(DEFAULT_INPUT))
+    ap.add_argument("--out", default=str(DEFAULT_OUT))
+    args = ap.parse_args()
+
+    input_path = Path(args.input)
+    input_path = input_path if input_path.is_absolute() else ROOT / input_path
+    out_path = Path(args.out)
+    out_path = out_path if out_path.is_absolute() else ROOT / out_path
+
+    payload = _load(input_path)
+    errors: list[str] = []
+
+    if not isinstance(payload, dict) or not payload:
+        errors.append("payload must be object")
+    else:
+        if payload.get("formula_id") != "DATA_GATED_PROGRESS_V1":
+            errors.append("formula_id mismatch")
+        if payload.get("gate") not in {"PASS", "IN_PROGRESS"}:
+            errors.append(f"gate={payload.get('gate')}")
+        for key in ["as_of", "done_count", "total_count", "items"]:
+            if key not in payload:
+                errors.append(f"missing field: {key}")
+
+        done_count = payload.get("done_count")
+        total_count = payload.get("total_count")
+        items = payload.get("items")
+
+        if isinstance(done_count, int) and isinstance(total_count, int):
+            if done_count < 0:
+                errors.append("done_count < 0")
+            if total_count <= 0:
+                errors.append("total_count <= 0")
+            if done_count > total_count:
+                errors.append("done_count > total_count")
+
+        if isinstance(items, list):
+            if not items:
+                errors.append("items empty")
+            for idx, item in enumerate(items):
+                if not isinstance(item, dict):
+                    errors.append(f"item[{idx}] not object")
+                    continue
+                for key in ["id", "label", "status", "source"]:
+                    if key not in item:
+                        errors.append(f"item[{idx}] missing {key}")
+                if item.get("status") not in {"DONE", "IN_PROGRESS", "DATA_GATED", "PASS", "FAIL", "USER_ACTION_REQUIRED"}:
+                    errors.append(f"item[{idx}].status={item.get('status')}")
+
+    result = {
+        "formula_id": FORMULA_ID,
+        "gate": "PASS" if not errors else "FAIL",
+        "checked_file": str(input_path.as_posix()),
+        "errors": errors,
+    }
+    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 0 if not errors else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())