evidence_pack_builder.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744
  1. """Build DB-validated evidence packs for DemandAgent outputs."""
  2. from __future__ import annotations
  3. import json
  4. import os
  5. import re
  6. from collections import defaultdict
  7. from collections.abc import Iterable, Mapping
  8. from typing import Any
  9. from examples.demand.db_manager import (
  10. query_case_ids_by_post_ids,
  11. query_element_bindings_for_items,
  12. query_execution_for_evidence,
  13. query_itemset_evidence,
  14. query_itemset_items_with_categories,
  15. query_seed_points_for_itemsets,
  16. )
  17. SOURCE_KIND_PATTERN_ITEMSET = "pattern_itemset"
  18. PATTERN_SOURCE_SYSTEM = "pg_pattern_v2"
  19. PATTERN_ITEMSET_SCOPE = "topic"
  20. def build_evidence_pack(
  21. execution_id: int,
  22. demand_item: Any,
  23. trace_id: str,
  24. demand_task_id: int | None,
  25. demand_content_id: int | None,
  26. *,
  27. demand_scope: Mapping[str, Any] | None = None,
  28. merge_leve2: str | None = None,
  29. platform: str | None = None,
  30. ) -> dict[str, Any]:
  31. """Return a DB-validated evidence pack or a reject result.
  32. This function is intentionally read-only. It never writes Demand rows or
  33. reject rows; callers can route the returned result into their own sink.
  34. """
  35. try:
  36. return _build_evidence_pack(
  37. execution_id=execution_id,
  38. demand_item=demand_item,
  39. trace_id=trace_id,
  40. demand_task_id=demand_task_id,
  41. demand_content_id=demand_content_id,
  42. demand_scope=demand_scope,
  43. merge_leve2=merge_leve2,
  44. platform=platform,
  45. )
  46. except Exception as exc:
  47. return _reject(f"db evidence validation failed: {exc}")
  48. def _build_evidence_pack(
  49. execution_id: int,
  50. demand_item: Any,
  51. trace_id: str,
  52. demand_task_id: int | None,
  53. demand_content_id: int | None,
  54. *,
  55. demand_scope: Mapping[str, Any] | None = None,
  56. merge_leve2: str | None = None,
  57. platform: str | None = None,
  58. ) -> dict[str, Any]:
  59. evidence_refs = _extract_evidence_refs(demand_item)
  60. resolved_scope = _resolve_demand_scope(
  61. execution_id=int(execution_id),
  62. evidence_refs=evidence_refs,
  63. demand_item=demand_item,
  64. demand_scope=demand_scope,
  65. merge_leve2=merge_leve2,
  66. platform=platform,
  67. )
  68. scope_merge_leve2 = _clean_str(resolved_scope.get("merge_leve2"))
  69. scope_platform = _clean_str(resolved_scope.get("platform"))
  70. itemset_ids, invalid_itemset_ids = _normalize_int_list(
  71. _first_present(
  72. evidence_refs.get("itemset_ids"),
  73. evidence_refs.get("itemset_id"),
  74. _read_field(demand_item, "itemset_ids"),
  75. _read_field(demand_item, "itemset_id"),
  76. )
  77. )
  78. if invalid_itemset_ids:
  79. return _reject(f"invalid itemset_ids: {invalid_itemset_ids}")
  80. source_kind = _resolve_source_kind(evidence_refs, demand_item, itemset_ids)
  81. if source_kind and source_kind != SOURCE_KIND_PATTERN_ITEMSET:
  82. return _reject(f"unsupported source_kind={source_kind}; only pattern_itemset is supported")
  83. if not source_kind:
  84. return _reject("missing source_kind=pattern_itemset")
  85. if not itemset_ids:
  86. return _reject("missing itemset_ids for source_kind=pattern_itemset")
  87. if len(itemset_ids) != 1:
  88. return _reject("V2 exact evidence requires exactly one itemset_id per DemandItem")
  89. execution = query_execution_for_evidence(int(execution_id))
  90. if not execution:
  91. return _reject(f"execution_id {execution_id} not found")
  92. execution_status = _clean_str(execution.get("status")).lower()
  93. if execution_status != "success":
  94. return _reject(
  95. f"execution_id {execution_id} status is {execution.get('status')}, not success"
  96. )
  97. itemsets = query_itemset_evidence(
  98. execution_id=int(execution_id),
  99. itemset_ids=itemset_ids,
  100. merge_leve2=scope_merge_leve2 or None,
  101. platform=scope_platform or None,
  102. )
  103. found_itemset_ids = {int(itemset["id"]) for itemset in itemsets}
  104. missing_itemset_ids = [itemset_id for itemset_id in itemset_ids if itemset_id not in found_itemset_ids]
  105. if missing_itemset_ids:
  106. return _reject(
  107. f"itemset_id {missing_itemset_ids[0]} not found under execution_id {execution_id}"
  108. )
  109. invalid_itemset_reason = _validate_itemset_facts(itemsets)
  110. if invalid_itemset_reason:
  111. return _reject(invalid_itemset_reason)
  112. mining_config_ids = _unique_ints(itemset.get("mining_config_id") for itemset in itemsets)
  113. if len(mining_config_ids) != 1:
  114. return _reject(
  115. "itemset_ids must resolve to exactly one mining_config_id; "
  116. f"got {mining_config_ids}"
  117. )
  118. itemset_items = query_itemset_items_with_categories(
  119. execution_id=int(execution_id),
  120. itemset_ids=itemset_ids,
  121. )
  122. items_by_itemset: dict[int, list[dict[str, Any]]] = defaultdict(list)
  123. for item in itemset_items:
  124. items_by_itemset[int(item["itemset_id"])].append(item)
  125. item_reason = _validate_itemset_items(
  126. execution_id=int(execution_id),
  127. itemsets=itemsets,
  128. items_by_itemset=items_by_itemset,
  129. )
  130. if item_reason:
  131. return _reject(item_reason)
  132. matched_post_ids = _merge_post_ids(itemset.get("matched_post_ids") for itemset in itemsets)
  133. requested_source_post_id = _choose_source_post_id(evidence_refs, demand_item, matched_post_ids)
  134. if not requested_source_post_id:
  135. return _reject("source_post_id cannot be resolved from DB-validated matched_post_ids")
  136. if requested_source_post_id not in set(matched_post_ids):
  137. return _reject(
  138. f"source_post_id {requested_source_post_id} is not in matched_post_ids for itemset_ids={itemset_ids}"
  139. )
  140. element_bindings = query_element_bindings_for_items(
  141. execution_id=int(execution_id),
  142. itemset_items=itemset_items,
  143. post_ids=[requested_source_post_id],
  144. )
  145. binding_reason = _validate_element_bindings(itemset_items, element_bindings)
  146. if binding_reason:
  147. resolved_source_post_id, element_bindings, binding_reason = _resolve_source_post_with_bindings(
  148. execution_id=int(execution_id),
  149. itemset_items=itemset_items,
  150. matched_post_ids=matched_post_ids,
  151. preferred_post_id=requested_source_post_id,
  152. )
  153. if binding_reason:
  154. return _reject(binding_reason)
  155. source_post_id = resolved_source_post_id
  156. else:
  157. source_post_id = requested_source_post_id
  158. seed_terms = _build_seed_terms_from_itemset_items(itemset_items)
  159. seed_reason = _validate_seed_terms(seed_terms, itemset_items, [])
  160. if seed_reason:
  161. return _reject(seed_reason)
  162. case_rows = query_case_ids_by_post_ids(matched_post_ids)
  163. decode_case_ids = _unique_strings(row.get("case_id") for row in case_rows)
  164. query_seed_points = query_seed_points_for_itemsets(
  165. execution_id=int(execution_id),
  166. itemset_ids=itemset_ids,
  167. matched_post_ids=matched_post_ids,
  168. top_k=_env_int("DEMAND_QUERY_SEED_POINTS_TOP_K", 30),
  169. )
  170. support_itemset = itemsets[0]
  171. scoped_post_count = support_itemset.get("scoped_post_count")
  172. filtered_absolute_support = support_itemset.get("filtered_absolute_support")
  173. evidence_pack = {
  174. "pattern_source_system": PATTERN_SOURCE_SYSTEM,
  175. "pattern_execution_id": int(execution_id),
  176. "mining_config_id": mining_config_ids[0],
  177. "source_kind": SOURCE_KIND_PATTERN_ITEMSET,
  178. "case_id_type": "post_id",
  179. "source_post_id": source_post_id,
  180. "category_bindings": _build_category_bindings(itemset_items),
  181. "element_bindings": _build_element_bindings(element_bindings),
  182. "itemset_ids": itemset_ids,
  183. "itemset_items": _build_itemset_items(itemset_items),
  184. "support": min(float(itemset["support"]) for itemset in itemsets),
  185. "absolute_support": min(int(itemset["absolute_support"]) for itemset in itemsets),
  186. "filtered_absolute_support": int(filtered_absolute_support) if filtered_absolute_support is not None else None,
  187. "scoped_post_count": int(scoped_post_count) if scoped_post_count is not None else None,
  188. "matched_post_ids": matched_post_ids,
  189. "video_ids": matched_post_ids,
  190. "case_ids": matched_post_ids,
  191. "decode_case_ids": decode_case_ids,
  192. "seed_terms": seed_terms,
  193. "query_seed_points": query_seed_points,
  194. "demand_scope": resolved_scope,
  195. "trace_id": trace_id,
  196. "demand_task_id": demand_task_id,
  197. "demand_content_id": demand_content_id,
  198. "source_certainty": "db_validated",
  199. "validation_status": "passed",
  200. }
  201. return {"success": True, "evidence_pack": evidence_pack}
  202. def _reject(reason: str) -> dict[str, Any]:
  203. return {"success": False, "reject_reason": reason}
  204. def _resolve_source_kind(
  205. evidence_refs: Mapping[str, Any],
  206. demand_item: Any,
  207. itemset_ids: list[int],
  208. ) -> str:
  209. explicit = _clean_str(
  210. _first_present(
  211. evidence_refs.get("source_kind"),
  212. _read_field(demand_item, "source_kind"),
  213. )
  214. )
  215. if explicit:
  216. return explicit
  217. source_tool = _clean_str(evidence_refs.get("source_tool"))
  218. if itemset_ids or source_tool == "get_itemset_detail":
  219. return SOURCE_KIND_PATTERN_ITEMSET
  220. return ""
  221. def _extract_evidence_refs(demand_item: Any) -> dict[str, Any]:
  222. value = _read_field(demand_item, "evidence_refs", {})
  223. if isinstance(value, str):
  224. raw = value.strip()
  225. if not raw:
  226. return {}
  227. try:
  228. parsed = json.loads(raw)
  229. except json.JSONDecodeError:
  230. return {}
  231. return dict(parsed) if isinstance(parsed, Mapping) else {}
  232. return dict(value) if isinstance(value, Mapping) else {}
  233. def _resolve_demand_scope(
  234. *,
  235. execution_id: int,
  236. evidence_refs: Mapping[str, Any],
  237. demand_item: Any,
  238. demand_scope: Mapping[str, Any] | None,
  239. merge_leve2: str | None,
  240. platform: str | None,
  241. ) -> dict[str, Any]:
  242. raw_scope = {}
  243. for value in (
  244. demand_scope,
  245. evidence_refs.get("demand_scope"),
  246. _read_field(demand_item, "demand_scope"),
  247. ):
  248. if isinstance(value, Mapping):
  249. raw_scope = dict(value)
  250. break
  251. scope_source = _clean_str(raw_scope.get("scope_source"))
  252. if not scope_source:
  253. scope_source = "odps_gap" if raw_scope.get("gap_dt") or raw_scope.get("lack_count") is not None else "manual_cli"
  254. resolved = {
  255. "scope_source": scope_source,
  256. "merge_leve2": _clean_str(
  257. _first_present(
  258. merge_leve2,
  259. raw_scope.get("merge_leve2"),
  260. raw_scope.get("merge_level2"),
  261. evidence_refs.get("merge_leve2"),
  262. evidence_refs.get("merge_level2"),
  263. )
  264. ),
  265. "platform": _clean_str(
  266. _first_present(
  267. platform,
  268. raw_scope.get("platform"),
  269. raw_scope.get("platform_type"),
  270. evidence_refs.get("platform"),
  271. evidence_refs.get("platform_type"),
  272. )
  273. ),
  274. "pattern_execution_id": int(execution_id),
  275. }
  276. for field in ("gap_dt", "requested_count", "lack_count"):
  277. value = raw_scope.get(field)
  278. if value is not None and value != "":
  279. resolved[field] = value
  280. return resolved
  281. def _env_int(name: str, default: int) -> int:
  282. try:
  283. return int(os.getenv(name, str(default)))
  284. except ValueError:
  285. return default
  286. def _read_field(source: Any, field: str, default: Any = None) -> Any:
  287. if source is None:
  288. return default
  289. if isinstance(source, Mapping):
  290. return source.get(field, default)
  291. if hasattr(source, field):
  292. return getattr(source, field)
  293. if hasattr(source, "model_dump"):
  294. dumped = source.model_dump()
  295. if isinstance(dumped, Mapping):
  296. return dumped.get(field, default)
  297. if hasattr(source, "dict"):
  298. dumped = source.dict()
  299. if isinstance(dumped, Mapping):
  300. return dumped.get(field, default)
  301. return default
  302. def _first_present(*values: Any) -> Any:
  303. for value in values:
  304. if value is None:
  305. continue
  306. if isinstance(value, str) and not value.strip():
  307. continue
  308. if isinstance(value, (list, tuple, set, dict)) and len(value) == 0:
  309. continue
  310. return value
  311. return None
  312. def _clean_str(value: Any) -> str:
  313. return str(value).strip() if value is not None else ""
  314. def _normalize_list(value: Any) -> list[Any]:
  315. if value is None:
  316. return []
  317. if isinstance(value, list):
  318. return value
  319. if isinstance(value, (tuple, set)):
  320. return list(value)
  321. if isinstance(value, str):
  322. raw = value.strip()
  323. if not raw:
  324. return []
  325. try:
  326. parsed = json.loads(raw)
  327. except json.JSONDecodeError:
  328. return [part.strip() for part in raw.split(",") if part.strip()]
  329. if isinstance(parsed, list):
  330. return parsed
  331. if parsed is None:
  332. return []
  333. return [parsed]
  334. return [value]
  335. def _normalize_int_list(value: Any) -> tuple[list[int], list[Any]]:
  336. result: list[int] = []
  337. invalid: list[Any] = []
  338. seen: set[int] = set()
  339. for raw in _normalize_list(value):
  340. try:
  341. parsed = int(raw)
  342. except (TypeError, ValueError):
  343. invalid.append(raw)
  344. continue
  345. if parsed not in seen:
  346. seen.add(parsed)
  347. result.append(parsed)
  348. return result, invalid
  349. def _unique_ints(values: Iterable[Any]) -> list[int]:
  350. result: list[int] = []
  351. seen: set[int] = set()
  352. for value in values:
  353. if value is None:
  354. continue
  355. parsed = int(value)
  356. if parsed not in seen:
  357. seen.add(parsed)
  358. result.append(parsed)
  359. return result
  360. def _unique_strings(values: Iterable[Any]) -> list[str]:
  361. result: list[str] = []
  362. seen: set[str] = set()
  363. for value in values:
  364. text = _clean_str(value)
  365. if text and text not in seen:
  366. seen.add(text)
  367. result.append(text)
  368. return result
  369. def _validate_itemset_facts(itemsets: list[dict[str, Any]]) -> str | None:
  370. for itemset in itemsets:
  371. itemset_id = itemset.get("id")
  372. execution_id = itemset.get("execution_id")
  373. if itemset.get("mining_config_id") is None:
  374. return f"itemset_id {itemset_id} missing mining_config_id"
  375. if int(itemset.get("mining_config_execution_id") or -1) != int(execution_id):
  376. return (
  377. f"itemset_id {itemset_id} mining_config_id {itemset.get('mining_config_id')} "
  378. f"does not belong to execution_id {execution_id}"
  379. )
  380. if _clean_str(itemset.get("scope")) != PATTERN_ITEMSET_SCOPE:
  381. return (
  382. f"itemset_id {itemset_id} scope={itemset.get('scope')} is not "
  383. f"{PATTERN_ITEMSET_SCOPE}"
  384. )
  385. if _clean_str(itemset.get("mining_config_scope")) != PATTERN_ITEMSET_SCOPE:
  386. return (
  387. f"itemset_id {itemset_id} mining_config_id {itemset.get('mining_config_id')} "
  388. f"scope={itemset.get('mining_config_scope')} is not {PATTERN_ITEMSET_SCOPE}"
  389. )
  390. if itemset.get("support") is None:
  391. return f"itemset_id {itemset_id} missing support"
  392. if itemset.get("absolute_support") is None:
  393. return f"itemset_id {itemset_id} missing absolute_support"
  394. matched_post_ids = itemset.get("matched_post_ids") or []
  395. if not matched_post_ids:
  396. return f"itemset_id {itemset_id} missing matched_post_ids"
  397. scoped_post_count = itemset.get("scoped_post_count")
  398. if scoped_post_count is not None:
  399. scoped_count = int(scoped_post_count)
  400. if len(matched_post_ids) != scoped_count:
  401. return (
  402. f"itemset_id {itemset_id} matched_post_ids count {len(matched_post_ids)} "
  403. f"does not equal scoped_post_count {scoped_count}"
  404. )
  405. if scoped_count > int(itemset["absolute_support"]):
  406. return (
  407. f"itemset_id {itemset_id} scoped_post_count {scoped_count} "
  408. f"is greater than absolute_support {itemset['absolute_support']}"
  409. )
  410. elif len(matched_post_ids) != int(itemset["absolute_support"]):
  411. return (
  412. f"itemset_id {itemset_id} matched_post_ids count {len(matched_post_ids)} "
  413. f"does not equal absolute_support {itemset['absolute_support']}"
  414. )
  415. return None
  416. def _validate_itemset_items(
  417. execution_id: int,
  418. itemsets: list[dict[str, Any]],
  419. items_by_itemset: dict[int, list[dict[str, Any]]],
  420. ) -> str | None:
  421. for itemset in itemsets:
  422. itemset_id = int(itemset["id"])
  423. items = items_by_itemset.get(itemset_id, [])
  424. if not items:
  425. return f"itemset_id {itemset_id} has no itemset_items"
  426. expected_count = itemset.get("item_count")
  427. if expected_count is not None and int(expected_count) != len(items):
  428. return (
  429. f"itemset_id {itemset_id} item_count mismatch: "
  430. f"expected {expected_count}, got {len(items)}"
  431. )
  432. for item in items:
  433. category_id = item.get("category_id")
  434. if category_id is None:
  435. return f"itemset_id {itemset_id} has item without category_id"
  436. if not item.get("category_found"):
  437. return (
  438. f"category_id {category_id} for itemset_id {itemset_id} "
  439. f"not found under execution_id {execution_id}"
  440. )
  441. if int(item.get("category_execution_id")) != int(execution_id):
  442. return (
  443. f"category_id {category_id} belongs to execution_id "
  444. f"{item.get('category_execution_id')}, not {execution_id}"
  445. )
  446. return None
  447. def _merge_post_ids(post_id_groups: Iterable[Any]) -> list[str]:
  448. merged: list[str] = []
  449. seen: set[str] = set()
  450. for group in post_id_groups:
  451. for raw_post_id in _normalize_list(group):
  452. post_id = _clean_str(raw_post_id)
  453. if post_id and post_id not in seen:
  454. seen.add(post_id)
  455. merged.append(post_id)
  456. return merged
  457. def _choose_source_post_id(
  458. evidence_refs: Mapping[str, Any],
  459. demand_item: Any,
  460. matched_post_ids: list[str],
  461. ) -> str:
  462. candidate = _clean_str(
  463. _first_present(
  464. evidence_refs.get("source_post_id"),
  465. evidence_refs.get("post_id"),
  466. _read_field(demand_item, "source_post_id"),
  467. _read_field(demand_item, "post_id"),
  468. )
  469. )
  470. if candidate:
  471. return candidate
  472. candidate_posts = _normalize_list(
  473. _first_present(
  474. evidence_refs.get("video_ids"),
  475. evidence_refs.get("matched_post_ids"),
  476. _read_field(demand_item, "video_ids"),
  477. )
  478. )
  479. matched_set = set(matched_post_ids)
  480. for raw_post_id in candidate_posts:
  481. post_id = _clean_str(raw_post_id)
  482. if post_id in matched_set:
  483. return post_id
  484. return matched_post_ids[0] if matched_post_ids else ""
  485. def _resolve_source_post_with_bindings(
  486. execution_id: int,
  487. itemset_items: list[dict[str, Any]],
  488. matched_post_ids: list[str],
  489. preferred_post_id: str,
  490. ) -> tuple[str, list[dict[str, Any]], str | None]:
  491. broad_bindings = query_element_bindings_for_items(
  492. execution_id=int(execution_id),
  493. itemset_items=itemset_items,
  494. post_ids=matched_post_ids,
  495. limit_per_item=max(len(matched_post_ids), 200),
  496. )
  497. by_item_id = {
  498. int(binding["itemset_item_id"]): set(_normalize_list(binding.get("matched_post_ids")))
  499. for binding in broad_bindings
  500. if binding.get("itemset_item_id") is not None
  501. }
  502. candidate_posts: set[str] | None = set(matched_post_ids)
  503. for item in itemset_items:
  504. item_id = int(item["itemset_item_id"])
  505. item_posts = {str(post_id) for post_id in by_item_id.get(item_id, set())}
  506. if not item_posts:
  507. return "", broad_bindings, f"itemset_item_id {item_id} has no exact PG topic element binding"
  508. candidate_posts = candidate_posts & item_posts if candidate_posts is not None else item_posts
  509. ordered_candidates = [post_id for post_id in matched_post_ids if post_id in (candidate_posts or set())]
  510. if not ordered_candidates:
  511. return "", broad_bindings, "no source_post_id can close all PG topic element bindings"
  512. source_post_id = preferred_post_id if preferred_post_id in ordered_candidates else ordered_candidates[0]
  513. exact_bindings = query_element_bindings_for_items(
  514. execution_id=int(execution_id),
  515. itemset_items=itemset_items,
  516. post_ids=[source_post_id],
  517. )
  518. binding_reason = _validate_element_bindings(itemset_items, exact_bindings)
  519. if binding_reason:
  520. return "", exact_bindings, binding_reason
  521. return source_post_id, exact_bindings, None
  522. def _validate_element_bindings(
  523. itemset_items: list[dict[str, Any]],
  524. element_bindings: list[dict[str, Any]],
  525. ) -> str | None:
  526. by_item_id = {
  527. int(binding["itemset_item_id"]): binding
  528. for binding in element_bindings
  529. if binding.get("itemset_item_id") is not None
  530. }
  531. for item in itemset_items:
  532. item_id = int(item["itemset_item_id"])
  533. binding = by_item_id.get(item_id)
  534. if not binding:
  535. return f"itemset_item_id {item_id} has no element binding"
  536. if int(binding.get("matched_element_count") or 0) <= 0:
  537. return f"itemset_item_id {item_id} has no exact PG topic element binding"
  538. if int(binding.get("matched_post_count") or 0) <= 0:
  539. return f"itemset_item_id {item_id} has no matched post binding"
  540. return None
  541. def _build_seed_terms_from_itemset_items(itemset_items: list[dict[str, Any]]) -> list[str]:
  542. candidates: list[str] = []
  543. for item in itemset_items:
  544. for value in (
  545. item.get("element_name"),
  546. item.get("category_name"),
  547. _last_path_part(item.get("category_path")),
  548. _last_path_part(item.get("category_full_path")),
  549. ):
  550. text = _clean_str(value)
  551. if text and text not in candidates:
  552. candidates.append(text)
  553. filtered = [
  554. term for term in candidates
  555. if _normalize_term(term) not in {"其他", "其它", "未知", "内容", "视频"}
  556. ]
  557. source = filtered or candidates
  558. max_terms = max(_env_int("DEMAND_SEED_TERMS_MAX", 10), 1)
  559. return source[:max_terms]
  560. def _validate_seed_terms(
  561. seed_terms: list[str],
  562. itemset_items: list[dict[str, Any]],
  563. element_bindings: list[dict[str, Any]],
  564. ) -> str | None:
  565. if not seed_terms:
  566. return "seed_terms cannot be resolved from itemset evidence"
  567. covered_terms = _build_covered_terms(itemset_items, element_bindings)
  568. uncovered = [term for term in seed_terms if _normalize_term(term) not in covered_terms]
  569. if uncovered:
  570. return f"seed_terms not covered by DB evidence: {uncovered}"
  571. return None
  572. def _build_covered_terms(
  573. itemset_items: list[dict[str, Any]],
  574. element_bindings: list[dict[str, Any]],
  575. ) -> set[str]:
  576. terms: set[str] = set()
  577. for item in itemset_items:
  578. for field in ("element_name", "category_name", "category_path", "category_full_path"):
  579. _add_term_variants(terms, item.get(field))
  580. for binding in element_bindings:
  581. _add_term_variants(terms, binding.get("element_name"))
  582. for sample in binding.get("sample_elements") or []:
  583. _add_term_variants(terms, sample.get("name"))
  584. _add_term_variants(terms, sample.get("category_path"))
  585. return terms
  586. def _add_term_variants(terms: set[str], value: Any) -> None:
  587. text = _clean_str(value)
  588. if not text:
  589. return
  590. terms.add(_normalize_term(text))
  591. for part in _split_pathish(text):
  592. terms.add(_normalize_term(part))
  593. def _normalize_term(value: Any) -> str:
  594. return re.sub(r"\s+", "", _clean_str(value))
  595. def _split_pathish(value: Any) -> list[str]:
  596. text = _clean_str(value)
  597. if not text:
  598. return []
  599. return [part.strip() for part in re.split(r"[>/|,,]+", text) if part.strip()]
  600. def _last_path_part(value: Any) -> str:
  601. parts = _split_pathish(value)
  602. return parts[-1] if parts else ""
  603. def _build_category_bindings(itemset_items: list[dict[str, Any]]) -> list[dict[str, Any]]:
  604. bindings: list[dict[str, Any]] = []
  605. seen: set[tuple[int, int | None]] = set()
  606. for item in itemset_items:
  607. key = (int(item["category_id"]), item.get("itemset_id"))
  608. if key in seen:
  609. continue
  610. seen.add(key)
  611. bindings.append(
  612. {
  613. "itemset_id": item.get("itemset_id"),
  614. "itemset_item_id": item.get("itemset_item_id"),
  615. "category_id": item.get("category_id"),
  616. "category_name": item.get("category_name"),
  617. "category_path": item.get("category_path") or item.get("category_full_path"),
  618. "category_full_path": item.get("category_full_path"),
  619. "category_source_type": item.get("category_source_type"),
  620. "category_level": item.get("category_level"),
  621. "point_type": item.get("point_type"),
  622. "dimension": item.get("dimension"),
  623. "element_name": item.get("element_name"),
  624. }
  625. )
  626. return bindings
  627. def _build_element_bindings(element_bindings: list[dict[str, Any]]) -> list[dict[str, Any]]:
  628. return [
  629. {
  630. "itemset_id": binding.get("itemset_id"),
  631. "itemset_item_id": binding.get("itemset_item_id"),
  632. "category_id": binding.get("category_id"),
  633. "point_type": binding.get("point_type"),
  634. "dimension": binding.get("dimension"),
  635. "element_name": binding.get("element_name"),
  636. "matched_element_count": binding.get("matched_element_count"),
  637. "matched_post_count": binding.get("matched_post_count"),
  638. "matched_post_ids": binding.get("matched_post_ids") or [],
  639. "sample_elements": binding.get("sample_elements") or [],
  640. }
  641. for binding in element_bindings
  642. ]
  643. def _build_itemset_items(itemset_items: list[dict[str, Any]]) -> list[dict[str, Any]]:
  644. return [
  645. {
  646. "itemset_id": item.get("itemset_id"),
  647. "category_id": item.get("category_id"),
  648. "category_path": item.get("category_path") or item.get("category_full_path"),
  649. "point_type": item.get("point_type"),
  650. "dimension": item.get("dimension"),
  651. "element_name": item.get("element_name"),
  652. }
  653. for item in itemset_items
  654. ]