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

"""Build DB-validated evidence packs for DemandAgent outputs."""

from __future__ import annotations

import json
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,
)


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,
) -> 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,
        )
    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,
) -> dict[str, Any]:
    evidence_refs = _extract_evidence_refs(demand_item)
    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")

    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)
    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 = _resolve_seed_terms(evidence_refs, itemset_items, element_bindings)
    seed_reason = _validate_seed_terms(seed_terms, itemset_items, element_bindings)
    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)

    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),
        "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,
        "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 _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"
        if 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 _resolve_seed_terms(
        evidence_refs: Mapping[str, Any],
        itemset_items: list[dict[str, Any]],
        element_bindings: list[dict[str, Any]],
) -> list[str]:
    provided = _unique_strings(_normalize_list(evidence_refs.get("seed_terms")))
    if provided:
        covered_terms = _build_covered_terms(itemset_items, element_bindings)
        verified = [term for term in provided if _normalize_term(term) in covered_terms]
        if verified:
            return verified

    derived: 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 derived:
                derived.append(text)
    return derived


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
    ]