aps-agent/scripts/verify_ruiyang_demo.py

#!/usr/bin/env python3
"""Verify generated Ruiyang demo artifacts without importing the APS server."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_OUTPUT_DIR = REPO_ROOT / "outputs" / "ruiyang-demo"
DEFAULT_SOURCE_DIR = Path(
    os.environ.get("RUIYANG_DEMO_DIR")
    or (REPO_ROOT / "demo-data" / "ruiyang-source")
)
MIN_CANONICAL = {"boms": 3, "bomItems": 42, "routings": 3, "routingSteps": 15}
EXPECTED_MRP = {"orders": 1, "make": 106, "purchase": 124, "outsource": 0}
EXPECTED_PANEL = {
    "salesOrders": 1,
    "make": 106,
    "purchaseOrders": 124,
    "outsourceOrders": 0,
    "productionOrders": 1,
    "workOrders": 5,
}


def load_json(path: Path) -> Any:
    if not path.is_file():
        raise FileNotFoundError(path)
    return json.loads(path.read_text(encoding="utf-8"))


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def verify(output_dir: Path, expected_source_dir: Path, business_date: str) -> dict[str, Any]:
    output_dir = output_dir.resolve()
    summary = load_json(output_dir / "summary.json")
    manifest = load_json(output_dir / "manifest.json")
    compact = load_json(output_dir / "compact-ruiyang-demo.json")

    failures: list[str] = []
    if Path(summary.get("sourceDir") or "").resolve() != expected_source_dir.resolve():
        failures.append(f"sourceDir={summary.get('sourceDir')!r}")
    if (summary.get("dates") or {}).get("businessDate") != business_date:
        failures.append(f"businessDate={(summary.get('dates') or {}).get('businessDate')!r}")
    if summary.get("mrp") != EXPECTED_MRP:
        failures.append(f"mrp={summary.get('mrp')!r}")
    if summary.get("orderPanel") != EXPECTED_PANEL:
        failures.append(f"orderPanel={summary.get('orderPanel')!r}")

    projection = summary.get("projection") or {}
    canonical = ((projection.get("counts") or {}).get("canonical") or {})
    flex = ((projection.get("counts") or {}).get("flex") or {})
    if not projection.get("passed"):
        failures.append("projection gate not passed")
    for key, minimum in MIN_CANONICAL.items():
        if int(canonical.get(key) or 0) < minimum:
            failures.append(f"canonical.{key}={canonical.get(key)!r} < {minimum}")
    if canonical.get("bomItems") != flex.get("bom"):
        failures.append("canonical.bomItems != flex.bom")
    if canonical.get("routingSteps") != flex.get("routings"):
        failures.append("canonical.routingSteps != flex.routings")

    schedule = summary.get("schedule") or {}
    if schedule.get("poCount") != 1 or schedule.get("woCount") != 5:
        failures.append(f"schedule counts={schedule.get('poCount')}/{schedule.get('woCount')}")
    if schedule.get("conflictCount") != 0:
        failures.append(f"schedule conflictCount={schedule.get('conflictCount')}")
    if compact.get("schemaVersion") != "ruiyang-demo/1.0":
        failures.append("compact schemaVersion mismatch")

    artifact_count = 0
    for item in manifest.get("artifacts") or []:
        artifact_count += 1
        path = output_dir / item["name"]
        if not path.is_file():
            failures.append(f"missing artifact {item['name']}")
            continue
        if path.stat().st_size != item["bytes"]:
            failures.append(f"size mismatch {item['name']}")
        if sha256(path) != item["sha256"]:
            failures.append(f"sha256 mismatch {item['name']}")

    result = {
        "passed": not failures,
        "outputDir": str(output_dir),
        "sourceDir": str(expected_source_dir.resolve()),
        "businessDate": business_date,
        "artifactCount": artifact_count,
        "canonical": canonical,
        "mrp": summary.get("mrp"),
        "orderPanel": summary.get("orderPanel"),
        "schedule": {
            "poCount": schedule.get("poCount"),
            "woCount": schedule.get("woCount"),
            "conflictCount": schedule.get("conflictCount"),
        },
        "failures": failures,
    }
    if failures:
        raise RuntimeError("Ruiyang demo verification failed: " + "; ".join(failures))
    return result


def main() -> int:
    parser = argparse.ArgumentParser(description="Verify Ruiyang demo outputs")
    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
    parser.add_argument("--expected-source-dir", type=Path, default=DEFAULT_SOURCE_DIR)
    parser.add_argument("--business-date", default="2026-08-05")
    args = parser.parse_args()
    print(json.dumps(
        verify(args.output_dir, args.expected_source_dir, args.business_date),
        ensure_ascii=False,
        indent=2,
    ))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())