Agent/examples/process_pipeline/script/validate_schema.py

import json
from pathlib import Path

def check_keys(data, expected_keys, path_context=""):
    missing = [k for k in expected_keys if k not in data]
    if missing:
        return f"{path_context} missing keys: {missing}"
    return None

def validate_case(data):
    if not isinstance(data, dict): return "Root is not a dict"
    err = check_keys(data, ["requirement", "cases"])
    if err: return err
    if not isinstance(data["cases"], list): return "'cases' is not a list"

    if len(data["cases"]) == 0:
        return "'cases' array is empty"

    for i, c in enumerate(data["cases"]):
        err = check_keys(c, ["id", "title", "platform", "source_url", "metrics", "user_feedback", "images", "input_details", "output_details", "workflow_process"], f"cases[{i}]")
        if err: return err
        if not isinstance(c.get("images", []), list): return f"cases[{i}].images must be a list"

        # 检查关键字段是否为空
        if not (c.get("title") or "").strip():
            return f"cases[{i}].title is empty"
        wp = c.get("workflow_process")
        if not wp or (isinstance(wp, str) and not wp.strip()) or (isinstance(wp, list) and len(wp) == 0):
            return f"cases[{i}].workflow_process is empty"
    return None

def validate_blueprint(data):
    if not isinstance(data, dict): return "Root is not a dict"
    err = check_keys(data, ["requirement", "distilled_cases", "blueprints"])
    if err: return err

    if not isinstance(data["blueprints"], list): return "'blueprints' is not a list"
    if len(data["blueprints"]) == 0:
        return "'blueprints' array is empty"

    for i, bp in enumerate(data["blueprints"]):
        err = check_keys(bp, ["name", "phases", "reasoning"], f"blueprints[{i}]")
        if err: return err
        if not isinstance(bp.get("phases", []), list): return f"blueprints[{i}].phases must be a list"

        # 检查关键字段是否为空
        if not (bp.get("name") or "").strip():
            return f"blueprints[{i}].name is empty"
        if len(bp.get("phases", [])) == 0:
            return f"blueprints[{i}].phases array is empty"

    if not isinstance(data["distilled_cases"], list): return "'distilled_cases' is not a list"
    if len(data["distilled_cases"]) == 0:
        return "'distilled_cases' array is empty"

    for i, dc in enumerate(data["distilled_cases"]):
        err = check_keys(dc, ["id", "title", "source_url", "user_feedback", "workflow_process"], f"distilled_cases[{i}]")
        if err: return err

        # 检查关键字段是否为空
        if not (dc.get("title") or "").strip():
            return f"distilled_cases[{i}].title is empty"
    return None

def validate_capabilities(data):
    if not isinstance(data, dict): return "Root is not a dict"
    err = check_keys(data, ["extracted_capabilities", "requirement"])
    if err: return err

    if not isinstance(data["extracted_capabilities"], list): return "'extracted_capabilities' is not a list"
    if len(data["extracted_capabilities"]) == 0:
        return "'extracted_capabilities' array is empty"

    for i, cap in enumerate(data["extracted_capabilities"]):
        err = check_keys(cap, ["id", "name", "description", "criterion", "effects", "implements", "is_new", "case_references"], f"extracted_capabilities[{i}]")
        if err: return err
        if not isinstance(cap.get("effects", []), list): return f"extracted_capabilities[{i}].effects must be a list"
        if not isinstance(cap.get("case_references", []), list): return f"extracted_capabilities[{i}].case_references must be a list"

        # 检查关键字段是否为空
        if not (cap.get("name") or "").strip():
            return f"extracted_capabilities[{i}].name is empty"
        if not (cap.get("description") or "").strip():
            return f"extracted_capabilities[{i}].description is empty"
    return None

def validate_strategy(data):
    if not isinstance(data, dict): return "Root is not a dict"
    err = check_keys(data, ["requirement", "strategies", "uncovered_requirements"])
    if err: return err

    if not isinstance(data["strategies"], list): return "'strategies' is not a list"
    if len(data["strategies"]) == 0:
        return "'strategies' array is empty"

    for i, strat in enumerate(data["strategies"]):
        err = check_keys(strat, ["is_selected", "name", "source", "workflow_outline", "highlight_coverage", "baseline_coverage", "reasoning", "why_not", "could_switch_if", "coverage_score", "coverage_explanation"], f"strategies[{i}]")
        if err: return err

        # 检查关键字段是否为空
        if not (strat.get("name") or "").strip():
            return f"strategies[{i}].name is empty"
        if not (strat.get("reasoning") or "").strip():
            return f"strategies[{i}].reasoning is empty"

        if isinstance(strat.get("workflow_outline"), list):
            if len(strat["workflow_outline"]) == 0:
                return f"strategies[{i}].workflow_outline array is empty"

            for j, wo in enumerate(strat["workflow_outline"]):
                err = check_keys(wo, ["phase", "description", "capabilities"], f"strategies[{i}].workflow_outline[{j}]")
                if err: return err
                if not isinstance(wo.get("capabilities", []), list): return f"strategies[{i}].workflow_outline[{j}].capabilities must be a list"

                # 检查关键字段是否为空
                if not (wo.get("phase") or "").strip():
                    return f"strategies[{i}].workflow_outline[{j}].phase is empty"
                if not (wo.get("description") or "").strip():
                    return f"strategies[{i}].workflow_outline[{j}].description is empty"
    return None

def check_missing_files(base_dir):
    """检查每个需求目录是否缺少必需的文件"""
    missing_files = []

    # 获取所有需求目录（格式为 001, 002, ...）
    req_dirs = sorted([d for d in base_dir.iterdir() if d.is_dir() and d.name.isdigit()])

    for req_dir in req_dirs:
        req_id = req_dir.name

        # 检查必需的文件
        required_files = {
            "raw_cases": req_dir / "raw_cases",
            "blueprint.json": req_dir / "blueprint.json",
            "capabilities_extracted.json": req_dir / "capabilities_extracted.json",
            "strategy.json": req_dir / "strategy.json"
        }

        for file_name, file_path in required_files.items():
            if file_name == "raw_cases":
                # raw_cases 是目录，检查是否存在且至少有一个 case 文件
                if not file_path.exists():
                    missing_files.append((req_id, f"raw_cases directory missing"))
                elif not list(file_path.glob("case_*.json")):
                    missing_files.append((req_id, f"raw_cases directory exists but contains no case files"))
            else:
                # 其他是文件
                if not file_path.exists():
                    missing_files.append((req_id, f"{file_name} missing"))

    return missing_files

def main():
    base_dir = Path(__file__).parent.parent / "output"
    if not base_dir.exists():
        print(f"Error: {base_dir} does not exist.")
        return

    # 检查文件缺失
    print(f"[Start] Checking for missing files...")
    missing_files = check_missing_files(base_dir)

    if missing_files:
        print(f"[WARNING] Found {len(missing_files)} missing files:")
        for req_id, issue in missing_files:
            print(f"   - REQ_{req_id}: {issue}")
        print("-" * 50)
    else:
        print("[OK] All required files are present.")
        print("-" * 50)

    # 检查 schema
    json_files = list(base_dir.rglob("*.json"))
    total_files = len(json_files)

    format_errors = []

    print(f"[Start] Validating schema for {total_files} JSON files...")

    for file_path in json_files:
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception as e:
            format_errors.append((file_path, f"JSON Parsing Error: {e}"))
            continue
            
        filename = file_path.name
        rel_path = file_path.relative_to(base_dir.parent)
        
        err = None
        if filename.startswith("case_"):
            err = validate_case(data)
        elif filename == "blueprint.json":
            err = validate_blueprint(data)
        elif filename == "capabilities_extracted.json":
            err = validate_capabilities(data)
        elif filename == "strategy.json":
            err = validate_strategy(data)
        else:
            # Unknown json file
            pass
            
        if err:
            format_errors.append((rel_path, f"Schema mismatch: {err}"))

    report_path = Path(__file__).parent / "schema_errors_report.txt"

    print("-" * 50)
    with open(report_path, "w", encoding="utf-8") as out_f:
        if not format_errors:
            msg = f"[OK] All {total_files} JSON files match their expected schemas perfectly!"
            print(msg)
            out_f.write(msg + "\n")
        else:
            msg = f"[ERROR] Found {len(format_errors)} files with incorrect schemas/formats:"
            print(msg)
            out_f.write(msg + "\n")
            for path, error in format_errors:
                print(f"   - {path}: {error}")
                out_f.write(f"   - {path}: {error}\n")
    print("-" * 50)
    print(f"Schema error details saved to {report_path}")

if __name__ == "__main__":
    main()