"""Build DB-validated evidence packs for DemandAgent outputs.""" from __future__ import annotations import json import os import re from collections import defaultdict from collections.abc import Iterable, Mapping from typing import Any from examples.demand.db_manager import ( query_case_ids_by_post_ids, query_element_bindings_for_items, query_execution_for_evidence, query_itemset_evidence, query_itemset_items_with_categories, query_seed_points_for_itemsets, ) SOURCE_KIND_PATTERN_ITEMSET = "pattern_itemset" PATTERN_SOURCE_SYSTEM = "pg_pattern_v2" PATTERN_ITEMSET_SCOPE = "topic" def build_evidence_pack( execution_id: int, demand_item: Any, trace_id: str, demand_task_id: int | None, demand_content_id: int | None, *, demand_scope: Mapping[str, Any] | None = None, merge_leve2: str | None = None, platform: str | None = None, ) -> dict[str, Any]: """Return a DB-validated evidence pack or a reject result. This function is intentionally read-only. It never writes Demand rows or reject rows; callers can route the returned result into their own sink. """ try: return _build_evidence_pack( execution_id=execution_id, demand_item=demand_item, trace_id=trace_id, demand_task_id=demand_task_id, demand_content_id=demand_content_id, demand_scope=demand_scope, merge_leve2=merge_leve2, platform=platform, ) except Exception as exc: return _reject(f"db evidence validation failed: {exc}") def _build_evidence_pack( execution_id: int, demand_item: Any, trace_id: str, demand_task_id: int | None, demand_content_id: int | None, *, demand_scope: Mapping[str, Any] | None = None, merge_leve2: str | None = None, platform: str | None = None, ) -> dict[str, Any]: evidence_refs = _extract_evidence_refs(demand_item) resolved_scope = _resolve_demand_scope( execution_id=int(execution_id), evidence_refs=evidence_refs, demand_item=demand_item, demand_scope=demand_scope, merge_leve2=merge_leve2, platform=platform, ) scope_merge_leve2 = _clean_str(resolved_scope.get("merge_leve2")) scope_platform = _clean_str(resolved_scope.get("platform")) itemset_ids, invalid_itemset_ids = _normalize_int_list( _first_present( evidence_refs.get("itemset_ids"), evidence_refs.get("itemset_id"), _read_field(demand_item, "itemset_ids"), _read_field(demand_item, "itemset_id"), ) ) if invalid_itemset_ids: return _reject(f"invalid itemset_ids: {invalid_itemset_ids}") source_kind = _resolve_source_kind(evidence_refs, demand_item, itemset_ids) if source_kind and source_kind != SOURCE_KIND_PATTERN_ITEMSET: return _reject(f"unsupported source_kind={source_kind}; only pattern_itemset is supported") if not source_kind: return _reject("missing source_kind=pattern_itemset") if not itemset_ids: return _reject("missing itemset_ids for source_kind=pattern_itemset") if len(itemset_ids) != 1: return _reject("V2 exact evidence requires exactly one itemset_id per DemandItem") execution = query_execution_for_evidence(int(execution_id)) if not execution: return _reject(f"execution_id {execution_id} not found") execution_status = _clean_str(execution.get("status")).lower() if execution_status != "success": return _reject( f"execution_id {execution_id} status is {execution.get('status')}, not success" ) itemsets = query_itemset_evidence( execution_id=int(execution_id), itemset_ids=itemset_ids, merge_leve2=scope_merge_leve2 or None, platform=scope_platform or None, ) found_itemset_ids = {int(itemset["id"]) for itemset in itemsets} missing_itemset_ids = [itemset_id for itemset_id in itemset_ids if itemset_id not in found_itemset_ids] if missing_itemset_ids: return _reject( f"itemset_id {missing_itemset_ids[0]} not found under execution_id {execution_id}" ) invalid_itemset_reason = _validate_itemset_facts(itemsets) if invalid_itemset_reason: return _reject(invalid_itemset_reason) mining_config_ids = _unique_ints(itemset.get("mining_config_id") for itemset in itemsets) if len(mining_config_ids) != 1: return _reject( "itemset_ids must resolve to exactly one mining_config_id; " f"got {mining_config_ids}" ) itemset_items = query_itemset_items_with_categories( execution_id=int(execution_id), itemset_ids=itemset_ids, ) items_by_itemset: dict[int, list[dict[str, Any]]] = defaultdict(list) for item in itemset_items: items_by_itemset[int(item["itemset_id"])].append(item) item_reason = _validate_itemset_items( execution_id=int(execution_id), itemsets=itemsets, items_by_itemset=items_by_itemset, ) if item_reason: return _reject(item_reason) matched_post_ids = _merge_post_ids(itemset.get("matched_post_ids") for itemset in itemsets) requested_source_post_id = _choose_source_post_id(evidence_refs, demand_item, matched_post_ids) if not requested_source_post_id: return _reject("source_post_id cannot be resolved from DB-validated matched_post_ids") if requested_source_post_id not in set(matched_post_ids): return _reject( f"source_post_id {requested_source_post_id} is not in matched_post_ids for itemset_ids={itemset_ids}" ) element_bindings = query_element_bindings_for_items( execution_id=int(execution_id), itemset_items=itemset_items, post_ids=[requested_source_post_id], ) binding_reason = _validate_element_bindings(itemset_items, element_bindings) if binding_reason: resolved_source_post_id, element_bindings, binding_reason = _resolve_source_post_with_bindings( execution_id=int(execution_id), itemset_items=itemset_items, matched_post_ids=matched_post_ids, preferred_post_id=requested_source_post_id, ) if binding_reason: return _reject(binding_reason) source_post_id = resolved_source_post_id else: source_post_id = requested_source_post_id seed_terms = _build_seed_terms_from_itemset_items(itemset_items) seed_reason = _validate_seed_terms(seed_terms, itemset_items, []) if seed_reason: return _reject(seed_reason) case_rows = query_case_ids_by_post_ids(matched_post_ids) decode_case_ids = _unique_strings(row.get("case_id") for row in case_rows) query_seed_points = query_seed_points_for_itemsets( execution_id=int(execution_id), itemset_ids=itemset_ids, matched_post_ids=matched_post_ids, top_k=_env_int("DEMAND_QUERY_SEED_POINTS_TOP_K", 30), ) support_itemset = itemsets[0] scoped_post_count = support_itemset.get("scoped_post_count") filtered_absolute_support = support_itemset.get("filtered_absolute_support") evidence_pack = { "pattern_source_system": PATTERN_SOURCE_SYSTEM, "pattern_execution_id": int(execution_id), "mining_config_id": mining_config_ids[0], "source_kind": SOURCE_KIND_PATTERN_ITEMSET, "case_id_type": "post_id", "source_post_id": source_post_id, "category_bindings": _build_category_bindings(itemset_items), "element_bindings": _build_element_bindings(element_bindings), "itemset_ids": itemset_ids, "itemset_items": _build_itemset_items(itemset_items), "support": min(float(itemset["support"]) for itemset in itemsets), "absolute_support": min(int(itemset["absolute_support"]) for itemset in itemsets), "filtered_absolute_support": int(filtered_absolute_support) if filtered_absolute_support is not None else None, "scoped_post_count": int(scoped_post_count) if scoped_post_count is not None else None, "matched_post_ids": matched_post_ids, "video_ids": matched_post_ids, "case_ids": matched_post_ids, "decode_case_ids": decode_case_ids, "seed_terms": seed_terms, "query_seed_points": query_seed_points, "demand_scope": resolved_scope, "trace_id": trace_id, "demand_task_id": demand_task_id, "demand_content_id": demand_content_id, "source_certainty": "db_validated", "validation_status": "passed", } return {"success": True, "evidence_pack": evidence_pack} def _reject(reason: str) -> dict[str, Any]: return {"success": False, "reject_reason": reason} def _resolve_source_kind( evidence_refs: Mapping[str, Any], demand_item: Any, itemset_ids: list[int], ) -> str: explicit = _clean_str( _first_present( evidence_refs.get("source_kind"), _read_field(demand_item, "source_kind"), ) ) if explicit: return explicit source_tool = _clean_str(evidence_refs.get("source_tool")) if itemset_ids or source_tool == "get_itemset_detail": return SOURCE_KIND_PATTERN_ITEMSET return "" def _extract_evidence_refs(demand_item: Any) -> dict[str, Any]: value = _read_field(demand_item, "evidence_refs", {}) if isinstance(value, str): raw = value.strip() if not raw: return {} try: parsed = json.loads(raw) except json.JSONDecodeError: return {} return dict(parsed) if isinstance(parsed, Mapping) else {} return dict(value) if isinstance(value, Mapping) else {} def _resolve_demand_scope( *, execution_id: int, evidence_refs: Mapping[str, Any], demand_item: Any, demand_scope: Mapping[str, Any] | None, merge_leve2: str | None, platform: str | None, ) -> dict[str, Any]: raw_scope = {} for value in ( demand_scope, evidence_refs.get("demand_scope"), _read_field(demand_item, "demand_scope"), ): if isinstance(value, Mapping): raw_scope = dict(value) break scope_source = _clean_str(raw_scope.get("scope_source")) if not scope_source: scope_source = "odps_gap" if raw_scope.get("gap_dt") or raw_scope.get("lack_count") is not None else "manual_cli" resolved = { "scope_source": scope_source, "merge_leve2": _clean_str( _first_present( merge_leve2, raw_scope.get("merge_leve2"), raw_scope.get("merge_level2"), evidence_refs.get("merge_leve2"), evidence_refs.get("merge_level2"), ) ), "platform": _clean_str( _first_present( platform, raw_scope.get("platform"), raw_scope.get("platform_type"), evidence_refs.get("platform"), evidence_refs.get("platform_type"), ) ), "pattern_execution_id": int(execution_id), } for field in ("gap_dt", "requested_count", "lack_count"): value = raw_scope.get(field) if value is not None and value != "": resolved[field] = value return resolved def _env_int(name: str, default: int) -> int: try: return int(os.getenv(name, str(default))) except ValueError: return default def _read_field(source: Any, field: str, default: Any = None) -> Any: if source is None: return default if isinstance(source, Mapping): return source.get(field, default) if hasattr(source, field): return getattr(source, field) if hasattr(source, "model_dump"): dumped = source.model_dump() if isinstance(dumped, Mapping): return dumped.get(field, default) if hasattr(source, "dict"): dumped = source.dict() if isinstance(dumped, Mapping): return dumped.get(field, default) return default def _first_present(*values: Any) -> Any: for value in values: if value is None: continue if isinstance(value, str) and not value.strip(): continue if isinstance(value, (list, tuple, set, dict)) and len(value) == 0: continue return value return None def _clean_str(value: Any) -> str: return str(value).strip() if value is not None else "" def _normalize_list(value: Any) -> list[Any]: if value is None: return [] if isinstance(value, list): return value if isinstance(value, (tuple, set)): return list(value) if isinstance(value, str): raw = value.strip() if not raw: return [] try: parsed = json.loads(raw) except json.JSONDecodeError: return [part.strip() for part in raw.split(",") if part.strip()] if isinstance(parsed, list): return parsed if parsed is None: return [] return [parsed] return [value] def _normalize_int_list(value: Any) -> tuple[list[int], list[Any]]: result: list[int] = [] invalid: list[Any] = [] seen: set[int] = set() for raw in _normalize_list(value): try: parsed = int(raw) except (TypeError, ValueError): invalid.append(raw) continue if parsed not in seen: seen.add(parsed) result.append(parsed) return result, invalid def _unique_ints(values: Iterable[Any]) -> list[int]: result: list[int] = [] seen: set[int] = set() for value in values: if value is None: continue parsed = int(value) if parsed not in seen: seen.add(parsed) result.append(parsed) return result def _unique_strings(values: Iterable[Any]) -> list[str]: result: list[str] = [] seen: set[str] = set() for value in values: text = _clean_str(value) if text and text not in seen: seen.add(text) result.append(text) return result def _validate_itemset_facts(itemsets: list[dict[str, Any]]) -> str | None: for itemset in itemsets: itemset_id = itemset.get("id") execution_id = itemset.get("execution_id") if itemset.get("mining_config_id") is None: return f"itemset_id {itemset_id} missing mining_config_id" if int(itemset.get("mining_config_execution_id") or -1) != int(execution_id): return ( f"itemset_id {itemset_id} mining_config_id {itemset.get('mining_config_id')} " f"does not belong to execution_id {execution_id}" ) if _clean_str(itemset.get("scope")) != PATTERN_ITEMSET_SCOPE: return ( f"itemset_id {itemset_id} scope={itemset.get('scope')} is not " f"{PATTERN_ITEMSET_SCOPE}" ) if _clean_str(itemset.get("mining_config_scope")) != PATTERN_ITEMSET_SCOPE: return ( f"itemset_id {itemset_id} mining_config_id {itemset.get('mining_config_id')} " f"scope={itemset.get('mining_config_scope')} is not {PATTERN_ITEMSET_SCOPE}" ) if itemset.get("support") is None: return f"itemset_id {itemset_id} missing support" if itemset.get("absolute_support") is None: return f"itemset_id {itemset_id} missing absolute_support" matched_post_ids = itemset.get("matched_post_ids") or [] if not matched_post_ids: return f"itemset_id {itemset_id} missing matched_post_ids" scoped_post_count = itemset.get("scoped_post_count") if scoped_post_count is not None: scoped_count = int(scoped_post_count) if len(matched_post_ids) != scoped_count: return ( f"itemset_id {itemset_id} matched_post_ids count {len(matched_post_ids)} " f"does not equal scoped_post_count {scoped_count}" ) if scoped_count > int(itemset["absolute_support"]): return ( f"itemset_id {itemset_id} scoped_post_count {scoped_count} " f"is greater than absolute_support {itemset['absolute_support']}" ) elif len(matched_post_ids) != int(itemset["absolute_support"]): return ( f"itemset_id {itemset_id} matched_post_ids count {len(matched_post_ids)} " f"does not equal absolute_support {itemset['absolute_support']}" ) return None def _validate_itemset_items( execution_id: int, itemsets: list[dict[str, Any]], items_by_itemset: dict[int, list[dict[str, Any]]], ) -> str | None: for itemset in itemsets: itemset_id = int(itemset["id"]) items = items_by_itemset.get(itemset_id, []) if not items: return f"itemset_id {itemset_id} has no itemset_items" expected_count = itemset.get("item_count") if expected_count is not None and int(expected_count) != len(items): return ( f"itemset_id {itemset_id} item_count mismatch: " f"expected {expected_count}, got {len(items)}" ) for item in items: category_id = item.get("category_id") if category_id is None: return f"itemset_id {itemset_id} has item without category_id" if not item.get("category_found"): return ( f"category_id {category_id} for itemset_id {itemset_id} " f"not found under execution_id {execution_id}" ) if int(item.get("category_execution_id")) != int(execution_id): return ( f"category_id {category_id} belongs to execution_id " f"{item.get('category_execution_id')}, not {execution_id}" ) return None def _merge_post_ids(post_id_groups: Iterable[Any]) -> list[str]: merged: list[str] = [] seen: set[str] = set() for group in post_id_groups: for raw_post_id in _normalize_list(group): post_id = _clean_str(raw_post_id) if post_id and post_id not in seen: seen.add(post_id) merged.append(post_id) return merged def _choose_source_post_id( evidence_refs: Mapping[str, Any], demand_item: Any, matched_post_ids: list[str], ) -> str: candidate = _clean_str( _first_present( evidence_refs.get("source_post_id"), evidence_refs.get("post_id"), _read_field(demand_item, "source_post_id"), _read_field(demand_item, "post_id"), ) ) if candidate: return candidate candidate_posts = _normalize_list( _first_present( evidence_refs.get("video_ids"), evidence_refs.get("matched_post_ids"), _read_field(demand_item, "video_ids"), ) ) matched_set = set(matched_post_ids) for raw_post_id in candidate_posts: post_id = _clean_str(raw_post_id) if post_id in matched_set: return post_id return matched_post_ids[0] if matched_post_ids else "" def _resolve_source_post_with_bindings( execution_id: int, itemset_items: list[dict[str, Any]], matched_post_ids: list[str], preferred_post_id: str, ) -> tuple[str, list[dict[str, Any]], str | None]: broad_bindings = query_element_bindings_for_items( execution_id=int(execution_id), itemset_items=itemset_items, post_ids=matched_post_ids, limit_per_item=max(len(matched_post_ids), 200), ) by_item_id = { int(binding["itemset_item_id"]): set(_normalize_list(binding.get("matched_post_ids"))) for binding in broad_bindings if binding.get("itemset_item_id") is not None } candidate_posts: set[str] | None = set(matched_post_ids) for item in itemset_items: item_id = int(item["itemset_item_id"]) item_posts = {str(post_id) for post_id in by_item_id.get(item_id, set())} if not item_posts: return "", broad_bindings, f"itemset_item_id {item_id} has no exact PG topic element binding" candidate_posts = candidate_posts & item_posts if candidate_posts is not None else item_posts ordered_candidates = [post_id for post_id in matched_post_ids if post_id in (candidate_posts or set())] if not ordered_candidates: return "", broad_bindings, "no source_post_id can close all PG topic element bindings" source_post_id = preferred_post_id if preferred_post_id in ordered_candidates else ordered_candidates[0] exact_bindings = query_element_bindings_for_items( execution_id=int(execution_id), itemset_items=itemset_items, post_ids=[source_post_id], ) binding_reason = _validate_element_bindings(itemset_items, exact_bindings) if binding_reason: return "", exact_bindings, binding_reason return source_post_id, exact_bindings, None def _validate_element_bindings( itemset_items: list[dict[str, Any]], element_bindings: list[dict[str, Any]], ) -> str | None: by_item_id = { int(binding["itemset_item_id"]): binding for binding in element_bindings if binding.get("itemset_item_id") is not None } for item in itemset_items: item_id = int(item["itemset_item_id"]) binding = by_item_id.get(item_id) if not binding: return f"itemset_item_id {item_id} has no element binding" if int(binding.get("matched_element_count") or 0) <= 0: return f"itemset_item_id {item_id} has no exact PG topic element binding" if int(binding.get("matched_post_count") or 0) <= 0: return f"itemset_item_id {item_id} has no matched post binding" return None def _build_seed_terms_from_itemset_items(itemset_items: list[dict[str, Any]]) -> list[str]: candidates: list[str] = [] for item in itemset_items: for value in ( item.get("element_name"), item.get("category_name"), _last_path_part(item.get("category_path")), _last_path_part(item.get("category_full_path")), ): text = _clean_str(value) if text and text not in candidates: candidates.append(text) filtered = [ term for term in candidates if _normalize_term(term) not in {"其他", "其它", "未知", "内容", "视频"} ] source = filtered or candidates max_terms = max(_env_int("DEMAND_SEED_TERMS_MAX", 10), 1) return source[:max_terms] def _validate_seed_terms( seed_terms: list[str], itemset_items: list[dict[str, Any]], element_bindings: list[dict[str, Any]], ) -> str | None: if not seed_terms: return "seed_terms cannot be resolved from itemset evidence" covered_terms = _build_covered_terms(itemset_items, element_bindings) uncovered = [term for term in seed_terms if _normalize_term(term) not in covered_terms] if uncovered: return f"seed_terms not covered by DB evidence: {uncovered}" return None def _build_covered_terms( itemset_items: list[dict[str, Any]], element_bindings: list[dict[str, Any]], ) -> set[str]: terms: set[str] = set() for item in itemset_items: for field in ("element_name", "category_name", "category_path", "category_full_path"): _add_term_variants(terms, item.get(field)) for binding in element_bindings: _add_term_variants(terms, binding.get("element_name")) for sample in binding.get("sample_elements") or []: _add_term_variants(terms, sample.get("name")) _add_term_variants(terms, sample.get("category_path")) return terms def _add_term_variants(terms: set[str], value: Any) -> None: text = _clean_str(value) if not text: return terms.add(_normalize_term(text)) for part in _split_pathish(text): terms.add(_normalize_term(part)) def _normalize_term(value: Any) -> str: return re.sub(r"\s+", "", _clean_str(value)) def _split_pathish(value: Any) -> list[str]: text = _clean_str(value) if not text: return [] return [part.strip() for part in re.split(r"[>/|,，]+", text) if part.strip()] def _last_path_part(value: Any) -> str: parts = _split_pathish(value) return parts[-1] if parts else "" def _build_category_bindings(itemset_items: list[dict[str, Any]]) -> list[dict[str, Any]]: bindings: list[dict[str, Any]] = [] seen: set[tuple[int, int | None]] = set() for item in itemset_items: key = (int(item["category_id"]), item.get("itemset_id")) if key in seen: continue seen.add(key) bindings.append( { "itemset_id": item.get("itemset_id"), "itemset_item_id": item.get("itemset_item_id"), "category_id": item.get("category_id"), "category_name": item.get("category_name"), "category_path": item.get("category_path") or item.get("category_full_path"), "category_full_path": item.get("category_full_path"), "category_source_type": item.get("category_source_type"), "category_level": item.get("category_level"), "point_type": item.get("point_type"), "dimension": item.get("dimension"), "element_name": item.get("element_name"), } ) return bindings def _build_element_bindings(element_bindings: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ { "itemset_id": binding.get("itemset_id"), "itemset_item_id": binding.get("itemset_item_id"), "category_id": binding.get("category_id"), "point_type": binding.get("point_type"), "dimension": binding.get("dimension"), "element_name": binding.get("element_name"), "matched_element_count": binding.get("matched_element_count"), "matched_post_count": binding.get("matched_post_count"), "matched_post_ids": binding.get("matched_post_ids") or [], "sample_elements": binding.get("sample_elements") or [], } for binding in element_bindings ] def _build_itemset_items(itemset_items: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ { "itemset_id": item.get("itemset_id"), "category_id": item.get("category_id"), "category_path": item.get("category_path") or item.get("category_full_path"), "point_type": item.get("point_type"), "dimension": item.get("dimension"), "element_name": item.get("element_name"), } for item in itemset_items ]