demand-agent-new/examples/demand/validate_v2_demand_content.py

"""Validate DemandAgent V2 demand_content JSON contracts."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any


ALLOWED_POINT_TYPES = {"灵感点", "目的点"}


def _load_rows(path: Path) -> list[dict[str, Any]]:
    loaded = json.loads(path.read_text(encoding="utf-8"))
    if isinstance(loaded, list):
        return [row for row in loaded if isinstance(row, dict)]
    if isinstance(loaded, dict) and isinstance(loaded.get("rows"), list):
        return [row for row in loaded["rows"] if isinstance(row, dict)]
    raise ValueError(f"{path} is not a demand_content row array")


def _as_ext_data(row: dict[str, Any]) -> dict[str, Any]:
    value = row.get("ext_data")
    if isinstance(value, dict):
        return value
    if isinstance(value, str) and value.strip():
        parsed = json.loads(value)
        if isinstance(parsed, dict):
            return parsed
    return {}


def _derive_seed_candidates(itemset_items: list[dict[str, Any]]) -> set[str]:
    result: set[str] = set()
    for item in itemset_items:
        for field in ("element_name", "category_name", "category_path", "category_full_path"):
            value = str(item.get(field) or "").strip()
            if not value:
                continue
            result.add(value.replace(" ", ""))
            for part in value.replace(">", "/").split("/"):
                part = part.strip()
                if part:
                    result.add(part.replace(" ", ""))
    return result


def validate_row(row: dict[str, Any], *, require_scope: bool = True) -> list[str]:
    errors: list[str] = []
    for field in ("name", "merge_leve2", "dt", "ext_data"):
        if not row.get(field):
            errors.append(f"missing top-level {field}")

    ext_data = _as_ext_data(row)
    evidence_pack = ext_data.get("evidence_pack")
    if not isinstance(evidence_pack, dict):
        return [*errors, "missing ext_data.evidence_pack"]

    expected = {
        "pattern_source_system": "pg_pattern_v2",
        "case_id_type": "post_id",
        "source_certainty": "db_validated",
        "validation_status": "passed",
    }
    for field, expected_value in expected.items():
        if evidence_pack.get(field) != expected_value:
            errors.append(f"invalid evidence_pack.{field}: {evidence_pack.get(field)!r}")

    required_non_empty = [
        "source_post_id",
        "pattern_execution_id",
        "source_kind",
        "evidence_sources",
        "category_bindings",
        "element_bindings",
        "support",
        "absolute_support",
        "matched_post_ids",
        "video_ids",
        "case_ids",
        "seed_terms",
        "trace_id",
    ]
    for field in required_non_empty:
        value = evidence_pack.get(field)
        if value is None or value == "" or value == []:
            errors.append(f"missing evidence_pack.{field}")
    if evidence_pack.get("source_kind") not in {
        "high_weight_element",
        "high_weight_category",
        "element_co_occurrence",
        "category_co_occurrence",
        "pattern_itemset",
        "multi_source",
    }:
        errors.append(f"invalid source_kind: {evidence_pack.get('source_kind')!r}")
    if not isinstance(evidence_pack.get("evidence_sources"), list) or not evidence_pack["evidence_sources"]:
        errors.append("missing evidence_sources")
    if not isinstance(evidence_pack.get("itemset_ids", []), list):
        errors.append("itemset_ids must be an array")
    if not isinstance(evidence_pack.get("itemset_items", []), list):
        errors.append("itemset_items must be an array")

    matched_post_ids = [str(post_id) for post_id in evidence_pack.get("matched_post_ids") or []]
    if str(evidence_pack.get("source_post_id") or "") not in set(matched_post_ids):
        errors.append("source_post_id not in matched_post_ids")
    if [str(post_id) for post_id in evidence_pack.get("video_ids") or []] != matched_post_ids:
        errors.append("video_ids must equal matched_post_ids")
    if [str(post_id) for post_id in evidence_pack.get("case_ids") or []] != matched_post_ids:
        errors.append("case_ids must equal matched_post_ids")

    query_seed_points = evidence_pack.get("query_seed_points")
    if not isinstance(query_seed_points, list):
        errors.append("query_seed_points must exist as array")
    else:
        previous_coverage: int | None = None
        for index, point in enumerate(query_seed_points, start=1):
            if not isinstance(point, dict):
                errors.append(f"query_seed_points[{index}] must be object")
                continue
            if point.get("point_type") not in ALLOWED_POINT_TYPES:
                errors.append(f"query_seed_points[{index}].point_type invalid")
            coverage = int(point.get("coverage_post_count") or 0)
            if coverage <= 0:
                errors.append(f"query_seed_points[{index}].coverage_post_count invalid")
            if int(point.get("rank") or 0) != index:
                errors.append(f"query_seed_points[{index}].rank invalid")
            if previous_coverage is not None and coverage > previous_coverage:
                errors.append("query_seed_points not sorted by coverage_post_count desc")
            previous_coverage = coverage

    itemset_items = evidence_pack.get("itemset_items") or []
    seed_candidates = _derive_seed_candidates(itemset_items)
    for source in evidence_pack.get("evidence_sources") or []:
        if isinstance(source, dict):
            for value in source.get("source_terms") or []:
                seed_candidates.add(str(value).replace(" ", ""))
    for term in evidence_pack.get("seed_terms") or []:
        if str(term).replace(" ", "") not in seed_candidates:
            errors.append(f"seed_term not derived from DB evidence sources: {term}")

    demand_scope = evidence_pack.get("demand_scope")
    if require_scope and not isinstance(demand_scope, dict):
        errors.append("missing demand_scope")
    elif isinstance(demand_scope, dict):
        if demand_scope.get("merge_leve2") and demand_scope.get("merge_leve2") != row.get("merge_leve2"):
            errors.append("demand_scope.merge_leve2 mismatch")
        if demand_scope.get("scope_source") == "odps_gap":
            for field in ("gap_dt", "requested_count", "lack_count"):
                if demand_scope.get(field) in (None, ""):
                    errors.append(f"missing demand_scope.{field}")

    scoped_count = evidence_pack.get("scoped_post_count")
    if require_scope and scoped_count is None:
        errors.append("missing scoped_post_count")
    if scoped_count is not None:
        scoped_count = int(scoped_count)
        if scoped_count <= 0:
            errors.append("scoped_post_count must be positive")
        if len(matched_post_ids) != scoped_count:
            errors.append("matched_post_ids length must equal scoped_post_count")
        if scoped_count > int(evidence_pack.get("absolute_support") or 0):
            errors.append("scoped_post_count must not exceed absolute_support")
        if evidence_pack.get("filtered_absolute_support") is not None:
            if int(evidence_pack["filtered_absolute_support"]) != scoped_count:
                errors.append("filtered_absolute_support must equal scoped_post_count")
    return errors


def validate_file(path: Path, *, require_scope: bool = True) -> list[str]:
    errors: list[str] = []
    rows = _load_rows(path)
    for index, row in enumerate(rows, start=1):
        for error in validate_row(row, require_scope=require_scope):
            errors.append(f"{path}:{index}: {error}")
    return errors


def main() -> None:
    parser = argparse.ArgumentParser(description="Validate DemandAgent V2 demand_content JSON.")
    parser.add_argument("paths", nargs="+", type=Path)
    parser.add_argument("--allow-no-scope", action="store_true")
    args = parser.parse_args()

    errors: list[str] = []
    for path in args.paths:
        errors.extend(validate_file(path, require_scope=not args.allow_no_scope))
    if errors:
        print("\n".join(errors), file=sys.stderr)
        raise SystemExit(1)
    print(json.dumps({"success": True, "files": [str(path) for path in args.paths]}, ensure_ascii=False))


if __name__ == "__main__":
    main()