|
|
@@ -15,11 +15,24 @@ from examples.demand.db_manager import (
|
|
|
query_execution_for_evidence,
|
|
|
query_itemset_evidence,
|
|
|
query_itemset_items_with_categories,
|
|
|
- query_seed_points_for_itemsets,
|
|
|
+ query_seed_points_for_sources,
|
|
|
+ query_source_elements,
|
|
|
)
|
|
|
|
|
|
|
|
|
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"
|
|
|
|
|
|
@@ -77,8 +90,184 @@ def _build_evidence_pack(
|
|
|
)
|
|
|
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 = 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=_env_int("DEMAND_QUERY_SEED_POINTS_TOP_K", 30),
|
|
|
+ )
|
|
|
+ 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"),
|
|
|
@@ -86,145 +275,188 @@ def _build_evidence_pack(
|
|
|
)
|
|
|
)
|
|
|
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")
|
|
|
+ return None, f"invalid itemset_ids: {invalid_itemset_ids}"
|
|
|
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"
|
|
|
- )
|
|
|
+ return None, "pattern_itemset source missing itemset_ids"
|
|
|
|
|
|
itemsets = query_itemset_evidence(
|
|
|
- execution_id=int(execution_id),
|
|
|
+ execution_id=execution_id,
|
|
|
itemset_ids=itemset_ids,
|
|
|
- merge_leve2=scope_merge_leve2 or None,
|
|
|
- platform=scope_platform or None,
|
|
|
+ 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 _reject(
|
|
|
- f"itemset_id {missing_itemset_ids[0]} not found under execution_id {execution_id}"
|
|
|
- )
|
|
|
+ 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 _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}"
|
|
|
- )
|
|
|
+ return None, invalid_itemset_reason
|
|
|
|
|
|
itemset_items = query_itemset_items_with_categories(
|
|
|
- execution_id=int(execution_id),
|
|
|
+ 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=int(execution_id),
|
|
|
+ execution_id=execution_id,
|
|
|
itemsets=itemsets,
|
|
|
items_by_itemset=items_by_itemset,
|
|
|
)
|
|
|
if item_reason:
|
|
|
- return _reject(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(evidence_refs, demand_item, matched_post_ids)
|
|
|
+ requested_source_post_id = _choose_source_post_id(source, 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")
|
|
|
+ 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 _reject(
|
|
|
- f"source_post_id {requested_source_post_id} is not in matched_post_ids for itemset_ids={itemset_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=int(execution_id),
|
|
|
+ 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:
|
|
|
- resolved_source_post_id, element_bindings, binding_reason = _resolve_source_post_with_bindings(
|
|
|
- execution_id=int(execution_id),
|
|
|
+ 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 _reject(binding_reason)
|
|
|
- source_post_id = resolved_source_post_id
|
|
|
+ return None, binding_reason
|
|
|
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],
|
|
|
+ return {
|
|
|
"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),
|
|
|
+ "source_tool": _clean_str(source.get("source_tool")) or "get_itemset_detail",
|
|
|
"itemset_ids": itemset_ids,
|
|
|
- "itemset_items": _build_itemset_items(itemset_items),
|
|
|
+ "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),
|
|
|
- "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}
|
|
|
+ "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 _reject(reason: str) -> dict[str, Any]:
|
|
|
- return {"success": False, "reject_reason": reason}
|
|
|
+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)
|
|
|
+
|
|
|
+ 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(
|
|
|
@@ -239,11 +471,39 @@ def _resolve_source_kind(
|
|
|
)
|
|
|
)
|
|
|
if explicit:
|
|
|
- return 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 ""
|
|
|
|
|
|
|
|
|
@@ -394,13 +654,44 @@ def _normalize_int_list(value: Any) -> tuple[list[int], list[Any]]:
|
|
|
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
|
|
|
- parsed = int(value)
|
|
|
+ try:
|
|
|
+ parsed = int(value)
|
|
|
+ except (TypeError, ValueError):
|
|
|
+ continue
|
|
|
if parsed not in seen:
|
|
|
seen.add(parsed)
|
|
|
result.append(parsed)
|
|
|
@@ -625,8 +916,7 @@ def _build_seed_terms_from_itemset_items(itemset_items: list[dict[str, Any]]) ->
|
|
|
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]
|
|
|
+ return _maybe_limit_seed_terms(source)
|
|
|
|
|
|
|
|
|
def _validate_seed_terms(
|
|
|
@@ -712,6 +1002,36 @@ def _build_category_bindings(itemset_items: list[dict[str, Any]]) -> list[dict[s
|
|
|
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 [
|
|
|
{
|
|
|
@@ -730,6 +1050,66 @@ def _build_element_bindings(element_bindings: list[dict[str, Any]]) -> list[dict
|
|
|
]
|
|
|
|
|
|
|
|
|
+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 [
|
|
|
{
|
|
|
@@ -742,3 +1122,155 @@ def _build_itemset_items(itemset_items: list[dict[str, Any]]) -> list[dict[str,
|
|
|
}
|
|
|
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
|
SamLee