"""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", "source_kind": "pattern_itemset", "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", "mining_config_id", "itemset_ids", "itemset_items", "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 not isinstance(evidence_pack.get("itemset_ids"), list) or len(evidence_pack["itemset_ids"]) != 1: errors.append("itemset_ids must contain exactly one itemset_id") 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 term in evidence_pack.get("seed_terms") or []: if str(term).replace(" ", "") not in seed_candidates: errors.append(f"seed_term not derived from itemset_items: {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()