fix(demand): map topic item dimensions to pg elements

examples/demand/pg_pattern_repository.py

@@ -33,6 +33,7 @@ except Exception:  # pragma: no cover - exercised in environments without deps
 PG_PATTERN_SOURCE_SYSTEM = "pg_pattern_v2"
 TOPIC_SCOPE = "topic"
 TOPIC_ELEMENT_SOURCE_TABLE = "post_decode_topic_point_element"
+ELEMENT_TYPE_DIMENSIONS = {"实质", "形式", "意图"}
 
 _READONLY_SQL_RE = re.compile(r"^\s*(select|with|show|explain)\b", re.IGNORECASE | re.DOTALL)
 _BLOCKED_SQL_RE = re.compile(
@@ -64,6 +65,11 @@ def _assert_readonly_sql(sql: str) -> None:
         raise RuntimeError("PG Pattern repository blocked a non-read-only SQL keyword")
 
 
+def _is_element_type_dimension(dimension: Any) -> bool:
+    """Return whether an item dimension maps to pattern_mining_element.element_type."""
+    return dimension in ELEMENT_TYPE_DIMENSIONS
+
+
 @contextmanager
 def _connect_readonly():
     if psycopg2 is None:
@@ -421,7 +427,7 @@ def query_element_bindings_for_items(
                 if item.get("point_type"):
                     clauses.append("point_type = %s")
                     params.append(item["point_type"])
-                if item.get("dimension"):
+                if _is_element_type_dimension(item.get("dimension")):
                     clauses.append("element_type = %s")
                     params.append(item["dimension"])
                 if item.get("element_name"):

examples/demand/tests/test_pg_pattern_repository.py

@@ -0,0 +1,17 @@
+import unittest
+
+from examples.demand.pg_pattern_repository import _is_element_type_dimension
+
+
+class PgPatternRepositoryTest(unittest.TestCase):
+    def test_topic_itemset_need_dimension_does_not_filter_element_type(self):
+        self.assertFalse(_is_element_type_dimension("需求"))
+
+    def test_real_element_dimensions_filter_element_type(self):
+        self.assertTrue(_is_element_type_dimension("实质"))
+        self.assertTrue(_is_element_type_dimension("形式"))
+        self.assertTrue(_is_element_type_dimension("意图"))
+
+
+if __name__ == "__main__":
+    unittest.main()