90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""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())
|