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

"""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_sources,
    query_source_elements,
)
from examples.demand.pg_pattern_repository import DEFAULT_DEMAND_PLATFORM, normalize_platform


SOURCE_KIND_PATTERN_ITEMSET = "pattern_itemset"
SOURCE_KIND_HIGH_WEIGHT_ELEMENT = "high_weight_element"
SOURCE_KIND_HIGH_WEIGHT_CATEGORY = "high_weight_category"
SOURCE_KIND_ELEMENT_CO_OCCURRENCE = "element_co_occurrence"
SOURCE_KIND_CATEGORY_CO_OCCURRENCE = "category_co_occurrence"
SOURCE_KIND_MULTI_SOURCE = "multi_source"
SUPPORTED_SOURCE_KINDS = {
    SOURCE_KIND_PATTERN_ITEMSET,
    SOURCE_KIND_HIGH_WEIGHT_ELEMENT,
    SOURCE_KIND_HIGH_WEIGHT_CATEGORY,
    SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
    SOURCE_KIND_CATEGORY_CO_OCCURRENCE,
}
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"))

    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"
        )

    source_candidates = _extract_evidence_sources(evidence_refs, demand_item)
    if not source_candidates:
        return _reject("missing evidence_refs.sources or resolvable evidence source")

    resolved_sources: list[dict[str, Any]] = []
    reject_reasons: list[str] = []
    for source in source_candidates:
        source_kind = _resolve_source_kind(source, demand_item, [])
        if source_kind == SOURCE_KIND_PATTERN_ITEMSET:
            resolved, reason = _resolve_pattern_itemset_source(
                execution_id=int(execution_id),
                source=source,
                demand_item=demand_item,
                evidence_refs=evidence_refs,
                scope_merge_leve2=scope_merge_leve2 or None,
                scope_platform=scope_platform or None,
            )
        else:
            resolved, reason = _resolve_elementish_source(
                execution_id=int(execution_id),
                source=source,
                demand_item=demand_item,
                source_kind=source_kind,
                scope_merge_leve2=scope_merge_leve2 or None,
                scope_platform=scope_platform or None,
            )
        if resolved:
            resolved_sources.append(resolved)
        elif reason:
            reject_reasons.append(reason)

    if not resolved_sources:
        return _reject("no evidence source passed DB validation: " + "; ".join(reject_reasons[:5]))

    matched_post_ids = _merge_post_ids(source.get("matched_post_ids") for source in resolved_sources)
    if not matched_post_ids:
        return _reject("validated evidence sources produced no matched_post_ids")

    validated_source_post_ids = _unique_strings(source.get("source_post_id") for source in resolved_sources)
    source_post_id = next((post_id for post_id in validated_source_post_ids if post_id in set(matched_post_ids)), "")
    if not source_post_id:
        source_post_id = _choose_source_post_id(evidence_refs, demand_item, matched_post_ids)
    if not source_post_id:
        source_post_id = matched_post_ids[0]
    if source_post_id not in set(matched_post_ids):
        return _reject(f"source_post_id {source_post_id} is not in validated matched_post_ids")

    itemset_ids = _unique_ints(
        itemset_id
        for source in resolved_sources
        for itemset_id in source.get("itemset_ids", [])
    )
    itemset_items = _dedupe_itemset_items(
        item
        for source in resolved_sources
        for item in source.get("raw_itemset_items", [])
    )
    category_bindings = _dedupe_bindings(
        binding
        for source in resolved_sources
        for binding in source.get("category_bindings", [])
    )
    element_bindings = _dedupe_bindings(
        binding
        for source in resolved_sources
        for binding in source.get("element_bindings", [])
    )
    if not category_bindings:
        return _reject("validated evidence sources produced no category_bindings")
    if not element_bindings:
        return _reject("validated evidence sources produced no element_bindings")

    seed_terms = _build_seed_terms_from_sources(demand_item, resolved_sources, itemset_items, element_bindings)
    if not seed_terms:
        return _reject("seed_terms cannot be resolved from DB-validated evidence sources")

    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_top_k = min(max(_env_int("DEMAND_QUERY_SEED_POINTS_TOP_K", 30), 0), 30)
    query_seed_points = query_seed_points_for_sources(
        execution_id=int(execution_id),
        matched_post_ids=matched_post_ids,
        category_ids=_unique_ints(binding.get("category_id") for binding in category_bindings),
        top_k=query_seed_points_top_k,
    )
    mining_config_ids = _unique_ints(
        config_id
        for source in resolved_sources
        for config_id in source.get("mining_config_ids", [])
    )
    absolute_support = _resolve_absolute_support(execution, resolved_sources, matched_post_ids)
    source_kinds = _unique_strings(source.get("source_kind") for source in resolved_sources)
    source_kind = source_kinds[0] if len(source_kinds) == 1 else SOURCE_KIND_MULTI_SOURCE

    evidence_pack = {
        "pattern_source_system": PATTERN_SOURCE_SYSTEM,
        "pattern_execution_id": int(execution_id),
        "mining_config_id": mining_config_ids[0] if mining_config_ids else None,
        "source_kind": source_kind,
        "case_id_type": "post_id",
        "source_post_id": source_post_id,
        "evidence_sources": _build_evidence_sources_payload(resolved_sources),
        "category_bindings": category_bindings,
        "element_bindings": element_bindings,
        "itemset_ids": itemset_ids,
        "itemset_items": _build_itemset_items(itemset_items),
        "support": _resolve_support(execution, resolved_sources, matched_post_ids),
        "absolute_support": absolute_support,
        "filtered_absolute_support": len(matched_post_ids),
        "scoped_post_count": len(matched_post_ids),
        "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 _extract_evidence_sources(evidence_refs: Mapping[str, Any], demand_item: Any) -> list[dict[str, Any]]:
    raw_sources = evidence_refs.get("sources")
    sources: list[dict[str, Any]] = []
    if isinstance(raw_sources, list):
        for raw_source in raw_sources:
            if isinstance(raw_source, Mapping):
                sources.append(dict(raw_source))
    elif isinstance(raw_sources, Mapping):
        sources.append(dict(raw_sources))

    if not sources and evidence_refs:
        sources.append(dict(evidence_refs))

    if not sources:
        names = _normalize_string_list(_read_field(demand_item, "element_names"))
        if names:
            sources.append(
                {
                    "source_kind": SOURCE_KIND_HIGH_WEIGHT_ELEMENT,
                    "source_tool": "create_demand_items.element_names_fallback",
                    "element_names": names,
                }
            )
    return sources


def _resolve_pattern_itemset_source(
        *,
        execution_id: int,
        source: Mapping[str, Any],
        demand_item: Any,
        evidence_refs: Mapping[str, Any],
        scope_merge_leve2: str | None,
        scope_platform: str | None,
) -> tuple[dict[str, Any] | None, str | None]:
    itemset_ids, invalid_itemset_ids = _normalize_int_list(
        _first_present(
            source.get("itemset_ids"),
            source.get("itemset_id"),
            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 None, f"invalid itemset_ids: {invalid_itemset_ids}"
    if not itemset_ids:
        itemset_ids = _extract_itemset_ids_from_text(
            source.get("reason"),
            source.get("desc"),
            source.get("description"),
            _read_field(demand_item, "reason"),
            _read_field(demand_item, "desc"),
        )
    if not itemset_ids:
        return None, "pattern_itemset source missing itemset_ids"

    itemsets = query_itemset_evidence(
        execution_id=execution_id,
        itemset_ids=itemset_ids,
        merge_leve2=scope_merge_leve2,
        platform=scope_platform,
    )
    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 None, 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 None, invalid_itemset_reason

    itemset_items = query_itemset_items_with_categories(
        execution_id=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=execution_id,
        itemsets=itemsets,
        items_by_itemset=items_by_itemset,
    )
    if item_reason:
        return None, 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(source, demand_item, matched_post_ids)
    if not requested_source_post_id:
        return None, "source_post_id cannot be resolved from DB-validated itemset posts"
    if requested_source_post_id not in set(matched_post_ids):
        return None, (
            f"source_post_id {requested_source_post_id} is not in matched_post_ids "
            f"for itemset_ids={itemset_ids}"
        )

    element_bindings = query_element_bindings_for_items(
        execution_id=execution_id,
        itemset_items=itemset_items,
        post_ids=[requested_source_post_id],
    )
    binding_reason = _validate_element_bindings(itemset_items, element_bindings)
    if binding_reason:
        source_post_id, element_bindings, binding_reason = _resolve_source_post_with_bindings(
            execution_id=execution_id,
            itemset_items=itemset_items,
            matched_post_ids=matched_post_ids,
            preferred_post_id=requested_source_post_id,
        )
        if binding_reason:
            return None, binding_reason
    else:
        source_post_id = requested_source_post_id

    return {
        "source_kind": SOURCE_KIND_PATTERN_ITEMSET,
        "source_tool": _clean_str(source.get("source_tool")) or "get_itemset_detail",
        "itemset_ids": itemset_ids,
        "mining_config_ids": _unique_ints(itemset.get("mining_config_id") for itemset in itemsets),
        "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,
        "source_post_id": source_post_id,
        "raw_itemset_items": itemset_items,
        "category_bindings": _build_category_bindings(itemset_items),
        "element_bindings": _build_element_bindings(element_bindings),
        "seed_terms": _build_seed_terms_from_itemset_items(itemset_items),
    }, None


def _resolve_elementish_source(
        *,
        execution_id: int,
        source: Mapping[str, Any],
        demand_item: Any,
        source_kind: str,
        scope_merge_leve2: str | None,
        scope_platform: str | None,
) -> tuple[dict[str, Any] | None, str | None]:
    if source_kind not in SUPPORTED_SOURCE_KINDS or source_kind == SOURCE_KIND_PATTERN_ITEMSET:
        return None, f"unsupported source_kind={source_kind or '<empty>'}"

    element_names = _source_element_names(source, demand_item)
    category_ids, invalid_category_ids = _normalize_int_list(
        _first_present(source.get("category_ids"), source.get("category_id"))
    )
    if invalid_category_ids:
        return None, f"invalid category_ids: {invalid_category_ids}"
    category_names = _normalize_string_list(
        _first_present(source.get("category_names"), source.get("category_name"), source.get("categories"))
    )
    category_paths = _normalize_string_list(
        _first_present(source.get("category_paths"), source.get("category_path"))
    )
    element_types = _normalize_string_list(
        _first_present(source.get("element_types"), source.get("element_type"), source.get("dimension"))
    )

    groups: list[dict[str, Any]] = []
    if source_kind in {SOURCE_KIND_HIGH_WEIGHT_ELEMENT, SOURCE_KIND_ELEMENT_CO_OCCURRENCE} and element_names:
        groups.extend({"element_names": [name]} for name in element_names)
    elif source_kind in {SOURCE_KIND_HIGH_WEIGHT_CATEGORY, SOURCE_KIND_CATEGORY_CO_OCCURRENCE}:
        if category_ids:
            groups.extend({"category_ids": [category_id]} for category_id in category_ids)
        elif category_paths:
            groups.extend({"category_paths": [path]} for path in category_paths)
        elif category_names:
            groups.extend({"category_names": [name]} for name in category_names)
        elif element_names:
            groups.extend({"category_names": [name]} for name in element_names)

    if not groups:
        groups.append(
            {
                "element_names": element_names,
                "category_ids": category_ids,
                "category_names": category_names,
                "category_paths": category_paths,
            }
        )

    if not any(_source_group_has_criteria(group) for group in groups):
        return None, f"{source_kind} source has no resolvable names or categories"

    rows_by_group: list[list[dict[str, Any]]] = []
    for group in groups:
        rows = query_source_elements(
            execution_id=execution_id,
            element_names=group.get("element_names"),
            category_ids=group.get("category_ids"),
            category_names=group.get("category_names"),
            category_paths=group.get("category_paths"),
            element_types=element_types,
            merge_leve2=scope_merge_leve2,
            platform=scope_platform,
            limit=_env_int("DEMAND_EVIDENCE_SOURCE_ELEMENT_LIMIT", 20000),
        )
        if not rows:
            return None, f"{source_kind} source cannot resolve DB rows for {group}"
        rows_by_group.append(rows)

    matched_sets = [
        {str(row["post_id"]) for row in rows if row.get("post_id") is not None}
        for rows in rows_by_group
    ]
    matched_post_set = set.intersection(*matched_sets) if matched_sets else set()
    if not matched_post_set:
        return None, f"{source_kind} source has no common scoped support posts"

    matched_post_ids = sorted(matched_post_set)
    all_rows = [
        dict(row)
        for rows in rows_by_group
        for row in rows
        if str(row.get("post_id") or "") in matched_post_set
    ]
    category_bindings = _build_source_category_bindings(all_rows, source_kind=source_kind)
    element_bindings = _build_source_element_bindings(all_rows, source_kind=source_kind)
    if not category_bindings or not element_bindings:
        return None, f"{source_kind} source produced no category/element bindings"

    return {
        "source_kind": source_kind,
        "source_tool": _clean_str(source.get("source_tool")),
        "source_terms": _source_terms_from_bindings(category_bindings, element_bindings),
        "itemset_ids": [],
        "mining_config_ids": [],
        "support": None,
        "absolute_support": len(matched_post_ids),
        "matched_post_ids": matched_post_ids,
        "source_post_id": _choose_source_post_id(source, demand_item, matched_post_ids),
        "raw_itemset_items": [],
        "category_bindings": category_bindings,
        "element_bindings": element_bindings,
        "seed_terms": _source_terms_from_bindings(category_bindings, element_bindings),
    }, None


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:
        aliases = {
            "element": SOURCE_KIND_HIGH_WEIGHT_ELEMENT,
            "元素": SOURCE_KIND_HIGH_WEIGHT_ELEMENT,
            "high_weight": SOURCE_KIND_HIGH_WEIGHT_ELEMENT,
            "category": SOURCE_KIND_HIGH_WEIGHT_CATEGORY,
            "分类": SOURCE_KIND_HIGH_WEIGHT_CATEGORY,
            "co_occurrence": SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
            "co-occurrence": SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
            "co occurrence": SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
            "element_co-occurrence": SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
            "element co occurrence": SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
            "category_co-occurrence": SOURCE_KIND_CATEGORY_CO_OCCURRENCE,
            "category co occurrence": SOURCE_KIND_CATEGORY_CO_OCCURRENCE,
            "关系": SOURCE_KIND_ELEMENT_CO_OCCURRENCE,
            "pattern": SOURCE_KIND_PATTERN_ITEMSET,
        }
        return aliases.get(explicit, explicit)

    source_tool = _clean_str(evidence_refs.get("source_tool"))
    if itemset_ids or source_tool == "get_itemset_detail":
        return SOURCE_KIND_PATTERN_ITEMSET
    if source_tool == "get_element_co_occurrences":
        return SOURCE_KIND_ELEMENT_CO_OCCURRENCE
    if source_tool == "get_category_co_occurrences":
        return SOURCE_KIND_CATEGORY_CO_OCCURRENCE
    if evidence_refs.get("category_id") or evidence_refs.get("category_ids"):
        return SOURCE_KIND_HIGH_WEIGHT_CATEGORY
    if evidence_refs.get("category_path") or evidence_refs.get("category_paths"):
        return SOURCE_KIND_HIGH_WEIGHT_CATEGORY
    if evidence_refs.get("category_name") or evidence_refs.get("category_names"):
        return SOURCE_KIND_HIGH_WEIGHT_CATEGORY
    if evidence_refs.get("element_names") or evidence_refs.get("element_name") or evidence_refs.get("name"):
        return SOURCE_KIND_HIGH_WEIGHT_ELEMENT
    return ""


def _extract_itemset_ids_from_text(*values: Any) -> list[int]:
    ids: list[int] = []
    for value in values:
        if value is None:
            continue
        text = str(value)
        for match in re.finditer(r"itemset\\s*[_#：:：-]?\\s*(\\d{3,})", text, flags=re.IGNORECASE):
            ids.append(int(match.group(1)))
    return _unique_ints(ids)


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": normalize_platform(
            _clean_str(
                _first_present(
                    platform,
                    raw_scope.get("platform"),
                    raw_scope.get("platform_type"),
                    evidence_refs.get("platform"),
                    evidence_refs.get("platform_type"),
                    DEFAULT_DEMAND_PLATFORM,
                )
            )
        ),
        "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 _normalize_string_list(value: Any) -> list[str]:
    result: list[str] = []
    seen: set[str] = set()
    for raw in _normalize_list(value):
        text = _clean_str(raw)
        if text and text not in seen:
            seen.add(text)
            result.append(text)
    return result


def _source_group_has_criteria(group: Mapping[str, Any]) -> bool:
    return any(_normalize_list(value) for value in group.values())


def _source_element_names(source: Mapping[str, Any], demand_item: Any) -> list[str]:
    return _normalize_string_list(
        _first_present(
            source.get("element_names"),
            source.get("element_name"),
            source.get("names"),
            source.get("name"),
            source.get("seed_terms"),
            _read_field(demand_item, "element_names"),
        )
    )


def _unique_ints(values: Iterable[Any]) -> list[int]:
    result: list[int] = []
    seen: set[int] = set()
    for value in values:
        if value is None:
            continue
        try:
            parsed = int(value)
        except (TypeError, ValueError):
            continue
        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
    return _maybe_limit_seed_terms(source)


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_source_category_bindings(rows: list[dict[str, Any]], *, source_kind: str) -> list[dict[str, Any]]:
    bindings: list[dict[str, Any]] = []
    seen: set[tuple[Any, ...]] = set()
    for row in rows:
        category_id = row.get("category_id")
        category_path = row.get("category_path") or row.get("category_full_path")
        if category_id is None and not category_path:
            continue
        key = (category_id, category_path, row.get("element_type"), source_kind)
        if key in seen:
            continue
        seen.add(key)
        bindings.append(
            {
                "source_kind": source_kind,
                "category_id": int(category_id) if category_id is not None else None,
                "category_name": row.get("category_name") or _last_path_part(category_path),
                "category_path": category_path,
                "category_full_path": row.get("category_full_path") or category_path,
                "category_source_type": row.get("category_source_type"),
                "category_level": row.get("category_level"),
                "category_nature": row.get("category_nature"),
                "point_type": row.get("point_type"),
                "dimension": row.get("element_type"),
                "element_name": row.get("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_source_element_bindings(rows: list[dict[str, Any]], *, source_kind: str) -> list[dict[str, Any]]:
    grouped: dict[tuple[Any, ...], dict[str, Any]] = {}
    for row in rows:
        key = (
            source_kind,
            row.get("category_id"),
            row.get("point_type"),
            row.get("element_type"),
            row.get("name"),
        )
        group = grouped.setdefault(
            key,
            {
                "source_kind": source_kind,
                "category_id": int(row["category_id"]) if row.get("category_id") is not None else None,
                "point_type": row.get("point_type"),
                "dimension": row.get("element_type"),
                "element_name": row.get("name"),
                "matched_element_count": 0,
                "matched_post_ids": [],
                "sample_elements": [],
            },
        )
        group["matched_element_count"] += 1
        post_id = _clean_str(row.get("post_id"))
        if post_id and post_id not in group["matched_post_ids"]:
            group["matched_post_ids"].append(post_id)
        if len(group["sample_elements"]) < 20:
            group["sample_elements"].append(_source_element_sample(row))

    result: list[dict[str, Any]] = []
    for group in grouped.values():
        group["matched_post_count"] = len(group["matched_post_ids"])
        result.append(group)
    result.sort(
        key=lambda item: (
            -int(item.get("matched_post_count") or 0),
            str(item.get("element_name") or ""),
            str(item.get("category_id") or ""),
        )
    )
    return result


def _source_element_sample(row: Mapping[str, Any]) -> dict[str, Any]:
    return {
        "id": row.get("id"),
        "name": row.get("name"),
        "post_id": _clean_str(row.get("post_id")),
        "point_text": row.get("point_text"),
        "point_type": row.get("point_type"),
        "category_id": row.get("category_id"),
        "element_type": row.get("element_type"),
        "source_table": row.get("source_table"),
        "category_path": row.get("category_path") or row.get("category_full_path"),
        "topic_point_id": row.get("topic_point_id"),
        "source_element_id": row.get("source_element_id"),
    }


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
    ]


def _dedupe_itemset_items(items: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    result: list[dict[str, Any]] = []
    seen: set[tuple[Any, ...]] = set()
    for item in items:
        key = (
            item.get("itemset_item_id"),
            item.get("itemset_id"),
            item.get("category_id"),
            item.get("point_type"),
            item.get("dimension"),
            item.get("element_name"),
        )
        if key in seen:
            continue
        seen.add(key)
        result.append(dict(item))
    return result


def _dedupe_bindings(bindings: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    result: list[dict[str, Any]] = []
    seen: set[tuple[Any, ...]] = set()
    for binding in bindings:
        key = (
            binding.get("source_kind"),
            binding.get("itemset_id"),
            binding.get("itemset_item_id"),
            binding.get("category_id"),
            binding.get("category_path"),
            binding.get("point_type"),
            binding.get("dimension"),
            binding.get("element_name"),
        )
        if key in seen:
            continue
        seen.add(key)
        result.append(dict(binding))
    return result


def _source_terms_from_bindings(
        category_bindings: list[dict[str, Any]],
        element_bindings: list[dict[str, Any]],
) -> list[str]:
    terms: list[str] = []
    for binding in element_bindings:
        for value in (binding.get("element_name"),):
            text = _clean_str(value)
            if text and text not in terms:
                terms.append(text)
    for binding in category_bindings:
        for value in (binding.get("category_name"), _last_path_part(binding.get("category_path"))):
            text = _clean_str(value)
            if text and text not in terms:
                terms.append(text)
    return terms


def _build_seed_terms_from_sources(
        demand_item: Any,
        resolved_sources: list[dict[str, Any]],
        itemset_items: list[dict[str, Any]],
        element_bindings: list[dict[str, Any]],
) -> list[str]:
    del demand_item
    candidates: list[str] = []
    for source in resolved_sources:
        for value in source.get("seed_terms", []) or []:
            text = _clean_str(value)
            if text and text not in candidates:
                candidates.append(text)
        for value in source.get("source_terms", []) or []:
            text = _clean_str(value)
            if text and text not in candidates:
                candidates.append(text)

    if not candidates and itemset_items:
        candidates.extend(_build_seed_terms_from_itemset_items(itemset_items))
    if not candidates:
        candidates.extend(_source_terms_from_bindings([], element_bindings))

    filtered = [
        term for term in candidates
        if _normalize_term(term) not in {"其他", "其它", "未知", "内容", "视频"}
    ]
    source = filtered or candidates
    return _maybe_limit_seed_terms(source)


def _maybe_limit_seed_terms(seed_terms: list[str]) -> list[str]:
    raw_limit = os.getenv("DEMAND_SEED_TERMS_MAX")
    if raw_limit is None or not str(raw_limit).strip():
        return seed_terms
    try:
        max_terms = int(str(raw_limit).strip())
    except ValueError:
        return seed_terms
    if max_terms <= 0:
        return seed_terms
    return seed_terms[:max_terms]


def _resolve_support(
        execution: Mapping[str, Any],
        resolved_sources: list[dict[str, Any]],
        matched_post_ids: list[str],
) -> float:
    itemset_supports = [
        float(source["support"])
        for source in resolved_sources
        if source.get("support") is not None
    ]
    if itemset_supports:
        return min(itemset_supports)
    post_count = int(execution.get("post_count") or 0)
    return float(len(matched_post_ids)) / float(post_count) if post_count > 0 else 0.0


def _resolve_absolute_support(
        execution: Mapping[str, Any],
        resolved_sources: list[dict[str, Any]],
        matched_post_ids: list[str],
) -> int:
    del execution
    itemset_supports = [
        int(source["absolute_support"])
        for source in resolved_sources
        if source.get("absolute_support") is not None and source.get("itemset_ids")
    ]
    return max([len(matched_post_ids), *itemset_supports])


def _build_evidence_sources_payload(resolved_sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
    payload: list[dict[str, Any]] = []
    for source in resolved_sources:
        payload.append(
            {
                "source_kind": source.get("source_kind"),
                "source_tool": source.get("source_tool"),
                "itemset_ids": source.get("itemset_ids") or [],
                "mining_config_ids": source.get("mining_config_ids") or [],
                "source_terms": source.get("source_terms") or source.get("seed_terms") or [],
                "source_post_id": source.get("source_post_id"),
                "matched_post_count": len(source.get("matched_post_ids") or []),
                "matched_post_ids": source.get("matched_post_ids") or [],
                "support": source.get("support"),
                "absolute_support": source.get("absolute_support"),
            }
        )
    return payload