fix(demand): resolve source post with pg element evidence

examples/demand/evidence_pack_builder.py

@@ -118,22 +118,32 @@ def _build_evidence_pack(
         return _reject(item_reason)
 
     matched_post_ids = _merge_post_ids(itemset.get("matched_post_ids") for itemset in itemsets)
-    source_post_id = _choose_source_post_id(evidence_refs, demand_item, matched_post_ids)
-    if not source_post_id:
+    requested_source_post_id = _choose_source_post_id(evidence_refs, demand_item, matched_post_ids)
+    if not requested_source_post_id:
         return _reject("source_post_id cannot be resolved from DB-validated matched_post_ids")
-    if source_post_id not in set(matched_post_ids):
+    if requested_source_post_id not in set(matched_post_ids):
         return _reject(
-            f"source_post_id {source_post_id} is not in matched_post_ids for itemset_ids={itemset_ids}"
+            f"source_post_id {requested_source_post_id} is not in matched_post_ids for itemset_ids={itemset_ids}"
         )
 
     element_bindings = query_element_bindings_for_items(
         execution_id=int(execution_id),
         itemset_items=itemset_items,
-        post_ids=[source_post_id],
+        post_ids=[requested_source_post_id],
     )
     binding_reason = _validate_element_bindings(itemset_items, element_bindings)
     if binding_reason:
-        return _reject(binding_reason)
+        resolved_source_post_id, element_bindings, binding_reason = _resolve_source_post_with_bindings(
+            execution_id=int(execution_id),
+            itemset_items=itemset_items,
+            matched_post_ids=matched_post_ids,
+            preferred_post_id=requested_source_post_id,
+        )
+        if binding_reason:
+            return _reject(binding_reason)
+        source_post_id = resolved_source_post_id
+    else:
+        source_post_id = requested_source_post_id
 
     seed_terms = _resolve_seed_terms(evidence_refs, itemset_items, element_bindings)
     seed_reason = _validate_seed_terms(seed_terms, itemset_items, element_bindings)
@@ -419,6 +429,47 @@ def _choose_source_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]],

examples/demand/tests/test_pg_evidence_builder.py

@@ -57,6 +57,15 @@ def _binding():
     }
 
 
+def _empty_binding(post_ids=None):
+    binding = _binding()
+    binding["matched_element_count"] = 0
+    binding["matched_post_count"] = 0
+    binding["matched_post_ids"] = post_ids or []
+    binding["sample_elements"] = []
+    return binding
+
+
 class PgEvidenceBuilderTest(unittest.TestCase):
     def _build(self, evidence_refs):
         return build_evidence_pack(
@@ -126,6 +135,31 @@ class PgEvidenceBuilderTest(unittest.TestCase):
         self.assertFalse(result["success"])
         self.assertIn("not in matched_post_ids", result["reject_reason"])
 
+    def test_falls_back_to_db_validated_source_post_with_element_binding(self):
+        bindings = [
+            [_empty_binding()],
+            [{**_binding(), "matched_post_ids": ["p2"]}],
+            [{**_binding(), "matched_post_ids": ["p2"]}],
+        ]
+        with (
+            patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value=_execution()),
+            patch("examples.demand.evidence_pack_builder.query_itemset_evidence", return_value=[_itemset()]),
+            patch("examples.demand.evidence_pack_builder.query_itemset_items_with_categories", return_value=[_item()]),
+            patch("examples.demand.evidence_pack_builder.query_element_bindings_for_items", side_effect=bindings),
+            patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]),
+        ):
+            result = self._build(
+                {
+                    "source_kind": "pattern_itemset",
+                    "itemset_ids": [1607313],
+                    "source_post_id": "p1",
+                    "seed_terms": ["综合性腐败"],
+                }
+            )
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["evidence_pack"]["source_post_id"], "p2")
+
 
 if __name__ == "__main__":
     unittest.main()