QuantEngineByItz/tools/validate_db_first_pipeline_v1.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def main() -> int:
    errors: list[str] = []

    spec_path = ROOT / "spec" / "02_data_contract.yaml"
    server_path = ROOT / "src" / "quant_engine" / "snapshot_admin_server_v1.py"
    collector_path = ROOT / "src" / "quant_engine" / "kis_data_collection_v1.py"

    # Check if files exist before reading
    spec_exists = spec_path.exists()
    server_exists = server_path.exists()
    collector_exists = collector_path.exists()

    spec_text = spec_path.read_text(encoding="utf-8") if spec_exists else ""
    server_text = server_path.read_text(encoding="utf-8") if server_exists else ""
    collector_text = collector_path.read_text(encoding="utf-8") if collector_exists else ""

    if not spec_exists:
        print(f"Warning: {spec_path} not found, skipping validation")
    if not server_exists:
        print(f"Warning: {server_path} not found, skipping server validation")
    if not collector_exists:
        print(f"Warning: {collector_path} not found, skipping collector validation")

    # Only check markers in files that exist
    marker_checks = {
        "spec/db-first": (spec_text, "DB 기반 수집 결과를 바탕으로 생성된 파생 보고서 증빙", spec_exists),
        "spec/db-first-xlsx": (spec_text, "xlsx는 HTS 잔고·거래내역 판독 또는 DB 반영 이전의 보조 감사 소스", spec_exists),
        "server/json-role": (server_text, "derived_report_evidence", server_exists),
        "server/json-evidence": (server_text, "Derived JSON Evidence Preview", server_exists),
        "server/collection-trend": (server_text, "collectionTrendChart", server_exists),
        "collector/db-canonical": (collector_text, "SQLite as the canonical persistence layer", collector_exists),
    }

    for name, (haystack, marker, should_check) in marker_checks.items():
        if should_check and marker not in haystack:
            errors.append(f"missing marker: {name}")

    # If no files exist, pass with warning
    if not (spec_exists or server_exists or collector_exists):
        print(json.dumps({"gate": "PASS", "errors": [], "note": "Legacy Python files not found, skipping DB pipeline validation"}, ensure_ascii=False, indent=2))
        return 0

    if errors:
        print(json.dumps({"gate": "FAIL", "errors": errors}, ensure_ascii=False, indent=2))
        return 1
    print(json.dumps({"gate": "PASS", "errors": []}, ensure_ascii=False, indent=2))
    return 0


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