"""Read-only PostgreSQL Pattern V2 repository for DemandAgent. This module is the DemandAgent boundary to PG `open_aigc.public` Pattern V2 facts. It intentionally exposes the old Demand-side query shapes while reading only PG main-truth tables: - pattern_mining_execution - pattern_mining_category - pattern_mining_element - pattern_mining_config - pattern_itemset - pattern_itemset_item - pattern_itemset_post """ from __future__ import annotations import json import os import re from contextlib import contextmanager from datetime import date, datetime from typing import Any, Iterable, Mapping, Sequence try: import psycopg2 from psycopg2.extras import RealDictCursor except Exception: # pragma: no cover - exercised in environments without deps psycopg2 = None RealDictCursor = None 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( r"\b(insert|update|delete|create|alter|drop|truncate|copy|grant|revoke|merge|call)\b", re.IGNORECASE, ) def _load_dotenv_once() -> None: try: from dotenv import load_dotenv except Exception: return load_dotenv() def _resolve_dsn() -> str: _load_dotenv_once() dsn = os.getenv("DEMAND_PATTERN_PG_DSN") or os.getenv("PGVECTOR_DSN") if not dsn: raise RuntimeError("PG Pattern V2 DSN missing: set DEMAND_PATTERN_PG_DSN or PGVECTOR_DSN") return dsn def _assert_readonly_sql(sql: str) -> None: if not _READONLY_SQL_RE.search(sql): raise RuntimeError("PG Pattern repository only allows SELECT/SHOW/EXPLAIN queries") if _BLOCKED_SQL_RE.search(sql): 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: raise RuntimeError("psycopg2 is required for PG Pattern V2 reads") conn = psycopg2.connect(_resolve_dsn()) try: conn.set_session(readonly=True, autocommit=True) yield conn finally: conn.close() def _fetch_all(sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> list[dict[str, Any]]: _assert_readonly_sql(sql) with _connect_readonly() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute(sql, params) return [_row_to_dict(row) for row in cursor.fetchall()] def _fetch_one(sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> dict[str, Any] | None: rows = _fetch_all(sql, params) return rows[0] if rows else None def _cursor_fetch_all(cursor: Any, sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> list[dict[str, Any]]: _assert_readonly_sql(sql) cursor.execute(sql, params) return [_row_to_dict(row) for row in cursor.fetchall()] def _cursor_fetch_one(cursor: Any, sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> dict[str, Any] | None: rows = _cursor_fetch_all(cursor, sql, params) return rows[0] if rows else None def _serialize_db_value(value: Any) -> Any: if isinstance(value, (datetime, date)): return value.isoformat() return value def _row_to_dict(row: Any) -> dict[str, Any]: if row is None: return {} mapping = getattr(row, "_mapping", row) return {key: _serialize_db_value(value) for key, value in dict(mapping).items()} def _jsonish_to_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, bytes): value = value.decode("utf-8") 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 _to_post_id_list(value: Any) -> list[str]: post_ids: list[str] = [] seen: set[str] = set() for item in _jsonish_to_list(value): post_id = str(item).strip() if item is not None else "" if post_id and post_id not in seen: seen.add(post_id) post_ids.append(post_id) return post_ids def _to_str_filter_list(value: Any) -> list[str]: result: list[str] = [] seen: set[str] = set() for item in _jsonish_to_list(value): text = str(item).strip() if item is not None else "" if text and text not in seen: seen.add(text) result.append(text) return result PLATFORM_ALIASES = { "piaoquan": ["piaoquan", "票圈"], "票圈": ["票圈", "piaoquan"], } DEFAULT_DEMAND_PLATFORM = "piaoquan" PLATFORM_CANONICAL = { "票圈": "piaoquan", } def normalize_platform(value: Any) -> str: """Return the canonical machine value for a platform label.""" text = str(value).strip() if value is not None else "" return PLATFORM_CANONICAL.get(text, text) def _expand_platform_filters(values: list[str]) -> list[str]: """Expand known platform aliases used by Hive and PG post rows.""" result: list[str] = [] seen: set[str] = set() for value in values: canonical = normalize_platform(value) for candidate in PLATFORM_ALIASES.get(canonical, [canonical]): if candidate and candidate not in seen: seen.add(candidate) result.append(candidate) return result def _to_int_list(values: Any) -> list[int]: result: list[int] = [] seen: set[int] = set() for value in _jsonish_to_list(values): if value is None or value == "": continue try: int_value = int(value) except (TypeError, ValueError): continue if int_value not in seen: seen.add(int_value) result.append(int_value) return result def _scope_filter_payload(merge_leve2: Any = None, platform: Any = None) -> dict[str, list[str]]: return { "merge_leve2": _to_str_filter_list(merge_leve2), "platform": _expand_platform_filters(_to_str_filter_list(platform)), } def _scope_is_active(scope_filter: Mapping[str, list[str]]) -> bool: return bool(scope_filter.get("merge_leve2") or scope_filter.get("platform")) def _clean_names(names: Iterable[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for raw in names: name = str(raw).strip() if raw is not None else "" if name and name not in seen: seen.add(name) result.append(name) return result def query_execution_for_evidence(execution_id: int) -> dict[str, Any] | None: """Read PG Pattern V2 execution facts for evidence validation.""" row = _fetch_one( """ SELECT id, snapshot_date, is_current, is_retired, status, post_filter, post_count, post_ids_digest, classify_execution_id, category_count, element_count, topic_itemset_count AS itemset_count, topic_itemset_count, topic_element_itemset_count, cross_itemset_count, script_sequence_count, paragraph_itemset_count, start_time, end_time, error_message FROM pattern_mining_execution WHERE id = %s LIMIT 1 """, (int(execution_id),), ) if not row: return None row["pattern_source_system"] = PG_PATTERN_SOURCE_SYSTEM return row def query_latest_success_execution_id() -> int | None: row = _fetch_one( """ SELECT id FROM pattern_mining_execution WHERE status = 'success' AND COALESCE(topic_itemset_count, 0) > 0 ORDER BY snapshot_date DESC NULLS LAST, id DESC LIMIT 1 """ ) return int(row["id"]) if row else None def query_itemset_evidence( execution_id: int, itemset_ids: Iterable[Any], *, merge_leve2: Any = None, platform: Any = None, ) -> list[dict[str, Any]]: """Read PG topic-scope itemset facts and aggregate matched posts.""" clean_ids = _to_int_list(itemset_ids or []) if not clean_ids: return [] scope_filter = _scope_filter_payload(merge_leve2=merge_leve2, platform=platform) scoped = _scope_is_active(scope_filter) scope_clauses: list[str] = [] scope_params: list[Any] = [] if scope_filter["merge_leve2"]: scope_clauses.append("p_scope.merge_leve2 = ANY(%s)") scope_params.append(scope_filter["merge_leve2"]) if scope_filter["platform"]: scope_clauses.append("p_scope.platform = ANY(%s)") scope_params.append(scope_filter["platform"]) scope_where_sql = f" AND {' AND '.join(scope_clauses)}" if scope_clauses else "" rows = _fetch_all( f""" WITH selected AS ( SELECT i.id, i.execution_id, i.mining_config_id, cfg.execution_id AS mining_config_execution_id, cfg.scope AS mining_config_scope, cfg.name AS mining_config_name, cfg.algorithm AS mining_config_algorithm, cfg.params::text AS mining_config_params, cfg.params ->> 'dimension_mode' AS mining_config_dimension_mode, COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS mining_config_target_depth, i.scope, i.combination_type, i.item_count, i.support, i.absolute_support, i.dimensions::text AS dimensions, i.is_cross_point, i.is_cross_layer FROM pattern_itemset i JOIN pattern_mining_config cfg ON cfg.id = i.mining_config_id AND cfg.execution_id = i.execution_id AND cfg.scope = %s WHERE i.execution_id = %s AND i.scope = %s AND i.id = ANY(%s) ) SELECT s.*, COALESCE( ( SELECT array_agg(p_all.post_id ORDER BY p_all.post_id) FROM pattern_itemset_post p_all WHERE p_all.itemset_id = s.id AND p_all.execution_id = s.execution_id ), ARRAY[]::varchar[] ) AS global_matched_post_ids, COALESCE( ( SELECT array_agg(DISTINCT p_scope.post_id ORDER BY p_scope.post_id) FROM pattern_itemset_post p_scope JOIN post post_scope ON post_scope.post_id = p_scope.post_id WHERE p_scope.itemset_id = s.id AND p_scope.execution_id = s.execution_id {scope_where_sql.replace('p_scope.merge_leve2', 'post_scope.merge_leve2').replace('p_scope.platform', 'post_scope.platform')} ), ARRAY[]::varchar[] ) AS filtered_matched_post_ids FROM selected s ORDER BY s.id """, tuple([TOPIC_SCOPE, int(execution_id), TOPIC_SCOPE, clean_ids, *scope_params]), ) result: list[dict[str, Any]] = [] for row in rows: itemset = dict(row) for key in ( "id", "execution_id", "mining_config_id", "mining_config_execution_id", "item_count", "absolute_support", ): if itemset.get(key) is not None: itemset[key] = int(itemset[key]) itemset["support"] = float(itemset["support"]) if itemset.get("support") is not None else None itemset["is_cross_point"] = bool(itemset.get("is_cross_point")) itemset["is_cross_layer"] = bool(itemset.get("is_cross_layer")) itemset["dimensions"] = _jsonish_to_list(itemset.get("dimensions")) if isinstance(itemset.get("mining_config_params"), str): try: itemset["mining_config_params"] = json.loads(itemset["mining_config_params"]) except json.JSONDecodeError: pass global_post_ids = _to_post_id_list(itemset.get("global_matched_post_ids")) filtered_post_ids = _to_post_id_list(itemset.get("filtered_matched_post_ids")) itemset["global_matched_post_ids"] = global_post_ids itemset["filtered_matched_post_ids"] = filtered_post_ids if scoped else [] itemset["filtered_absolute_support"] = len(filtered_post_ids) if scoped else None itemset["scoped_post_count"] = len(filtered_post_ids) if scoped else None itemset["scope_filter"] = scope_filter if scoped else {} itemset["matched_post_ids"] = filtered_post_ids if scoped else global_post_ids result.append(itemset) by_id = {itemset["id"]: itemset for itemset in result} return [by_id[itemset_id] for itemset_id in clean_ids if itemset_id in by_id] def query_itemset_items_with_categories( execution_id: int, itemset_ids: Iterable[Any], ) -> list[dict[str, Any]]: """Read PG itemset items and their exact category tree nodes.""" clean_ids = _to_int_list(itemset_ids or []) if not clean_ids: return [] rows = _fetch_all( """ SELECT ii.id AS itemset_item_id, ii.itemset_id, ii.layer, ii.point_type, ii.dimension, ii.category_id, ii.category_path, ii.element_name, ii.element_id, ii.post_count, c.id AS bound_category_id, c.execution_id AS category_execution_id, c.source_stable_id AS category_source_stable_id, c.source_type AS category_source_type, c.name AS category_name, c.path AS category_full_path, c.level AS category_level, c.parent_id AS category_parent_id, c.parent_source_stable_id AS category_parent_source_stable_id, c.classified_as AS category_nature FROM pattern_itemset_item ii JOIN pattern_itemset i ON i.id = ii.itemset_id AND i.execution_id = %s AND i.scope = %s LEFT JOIN pattern_mining_category c ON c.id = ii.category_id AND c.execution_id = i.execution_id WHERE ii.itemset_id = ANY(%s) ORDER BY ii.itemset_id, ii.id """, (int(execution_id), TOPIC_SCOPE, clean_ids), ) result: list[dict[str, Any]] = [] for row in rows: item = dict(row) for key in ( "itemset_item_id", "itemset_id", "category_id", "element_id", "post_count", "bound_category_id", "category_execution_id", "category_source_stable_id", "category_level", "category_parent_id", "category_parent_source_stable_id", ): if item.get(key) is not None: item[key] = int(item[key]) item["category_found"] = item.get("bound_category_id") is not None result.append(item) return result def query_element_bindings_for_items( execution_id: int, itemset_items: Sequence[Mapping[str, Any]], post_ids: Iterable[Any] | None = None, limit_per_item: int = 200, ) -> list[dict[str, Any]]: """Bind itemset items to real PG Pattern V2 source-post elements.""" clean_post_ids = _to_post_id_list(post_ids) if not itemset_items: return [] bindings: list[dict[str, Any]] = [] with _connect_readonly() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cursor: for raw_item in itemset_items: item = dict(raw_item) category_id = item.get("bound_category_id") or item.get("category_id") if category_id is None: continue clauses = [ "execution_id = %s", "category_id = %s", "source_table = %s", ] params: list[Any] = [int(execution_id), int(category_id), TOPIC_ELEMENT_SOURCE_TABLE] if item.get("point_type"): clauses.append("point_type = %s") params.append(item["point_type"]) if _is_element_type_dimension(item.get("dimension")): clauses.append("element_type = %s") params.append(item["dimension"]) if item.get("element_name"): clauses.append("name = %s") params.append(item["element_name"]) if clean_post_ids: clauses.append("post_id = ANY(%s)") params.append(clean_post_ids) where_sql = " AND ".join(clauses) count_row = _cursor_fetch_one( cursor, f""" SELECT COUNT(*) AS matched_element_count, COUNT(DISTINCT post_id) AS matched_post_count FROM pattern_mining_element WHERE {where_sql} """, tuple(params), ) or {} post_rows = _cursor_fetch_all( cursor, f""" SELECT DISTINCT post_id FROM pattern_mining_element WHERE {where_sql} ORDER BY post_id LIMIT %s """, tuple(params + [int(limit_per_item)]), ) sample_rows = _cursor_fetch_all( cursor, f""" SELECT id, post_id, source_table, source_element_id, point_type, point_text, element_type, name, category_id, category_path, topic_point_id FROM pattern_mining_element WHERE {where_sql} ORDER BY post_id, id LIMIT %s """, tuple(params + [min(int(limit_per_item), 20)]), ) bindings.append( { "itemset_id": item.get("itemset_id"), "itemset_item_id": item.get("itemset_item_id"), "category_id": int(category_id), "point_type": item.get("point_type"), "dimension": item.get("dimension"), "element_name": item.get("element_name"), "matched_element_count": int(count_row.get("matched_element_count") or 0), "matched_post_count": int(count_row.get("matched_post_count") or 0), "matched_post_ids": _to_post_id_list([row["post_id"] for row in post_rows]), "sample_elements": sample_rows, } ) return bindings def query_video_ids_by_names(execution_id: int, names: Iterable[str]) -> list[str]: """Resolve names to PG source post IDs by category name/path leaf or element name.""" clean_names = _clean_names(names) if not clean_names: return [] rows = _fetch_all( """ WITH target_categories AS ( SELECT id FROM pattern_mining_category WHERE execution_id = %s AND ( name = ANY(%s) OR regexp_replace(path, '^.*/', '') = ANY(%s) ) ), matched_elements AS ( SELECT post_id FROM pattern_mining_element WHERE execution_id = %s AND source_table = %s AND ( name = ANY(%s) OR category_id IN (SELECT id FROM target_categories) ) ) SELECT DISTINCT post_id FROM matched_elements WHERE post_id IS NOT NULL ORDER BY post_id """, (int(execution_id), clean_names, clean_names, int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, clean_names), ) return _to_post_id_list([row["post_id"] for row in rows]) def query_category_level(execution_id: int, name: str) -> int | None: clean_name = str(name).strip() if name is not None else "" if not clean_name: return None row = _fetch_one( """ SELECT level FROM pattern_mining_category WHERE execution_id = %s AND (name = %s OR regexp_replace(path, '^.*/', '') = %s) ORDER BY id DESC LIMIT 1 """, (int(execution_id), clean_name, clean_name), ) return int(row["level"]) if row and row.get("level") is not None else None def query_case_ids_by_post_ids(post_ids: Iterable[Any]) -> list[dict[str, Any]]: """PG Pattern V2 has post_id semantics for MVP exact evidence. DemandAgent no longer reads the old MySQL decode-case table in this PG-only path. `case_ids` is filled from validated post IDs by the evidence builder; `decode_case_ids` remains an optional empty compatibility field. """ return [] def query_categories( execution_id: int, *, source_type: str | None = None, keyword: str | None = None, category_ids: Iterable[Any] | None = None, limit: int | None = None, ) -> list[dict[str, Any]]: clauses = ["execution_id = %s"] params: list[Any] = [int(execution_id)] if source_type: clauses.append("source_type = %s") params.append(source_type) if keyword: clauses.append("(name ILIKE %s OR path ILIKE %s)") like = f"%{keyword}%" params.extend([like, like]) clean_ids = _to_int_list(category_ids or []) if clean_ids: clauses.append("id = ANY(%s)") params.append(clean_ids) limit_sql = "LIMIT %s" if limit else "" if limit: params.append(int(limit)) return _fetch_all( f""" SELECT id, execution_id, source_stable_id, source_type, name, description, path, level, parent_id, parent_source_stable_id, element_count, classified_as AS category_nature FROM pattern_mining_category WHERE {" AND ".join(clauses)} ORDER BY source_type, path, id {limit_sql} """, tuple(params), ) def query_elements( execution_id: int, *, keyword: str | None = None, element_type: str | None = None, post_ids: Iterable[Any] | None = None, category_ids: Iterable[Any] | None = None, merge_leve2: Any = None, platform: Any = None, limit: int | None = None, ) -> list[dict[str, Any]]: clauses = [ "e.execution_id = %s", "e.source_table = %s", ] params: list[Any] = [int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE] if keyword: clauses.append("e.name ILIKE %s") params.append(f"%{keyword}%") if element_type: clauses.append("e.element_type = %s") params.append(element_type) clean_posts = _to_post_id_list(post_ids) if post_ids is not None and not clean_posts: return [] if clean_posts: clauses.append("e.post_id = ANY(%s)") params.append(clean_posts) clean_categories = _to_int_list(category_ids or []) if clean_categories: clauses.append("e.category_id = ANY(%s)") params.append(clean_categories) scope_filter = _scope_filter_payload(merge_leve2=merge_leve2, platform=platform) scope_active = _scope_is_active(scope_filter) join_post_sql = "" if scope_active: join_post_sql = "JOIN post post_scope ON post_scope.post_id = e.post_id" if scope_filter["merge_leve2"]: clauses.append("post_scope.merge_leve2 = ANY(%s)") params.append(scope_filter["merge_leve2"]) if scope_filter["platform"]: clauses.append("post_scope.platform = ANY(%s)") params.append(scope_filter["platform"]) limit_sql = "LIMIT %s" if limit else "" if limit: params.append(int(limit)) return _fetch_all( f""" SELECT e.id, e.execution_id, e.post_id, e.source_table, e.source_element_id, e.element_type, e.element_sub_type, e.name, e.description, e.category_id, e.category_path, e.point_type, e.point_text, e.topic_point_id FROM pattern_mining_element e {join_post_sql} WHERE {" AND ".join(clauses)} ORDER BY e.post_id, e.id {limit_sql} """, tuple(params), ) def query_source_elements( execution_id: int, *, element_names: Iterable[Any] | None = None, category_ids: Iterable[Any] | None = None, category_names: Iterable[Any] | None = None, category_paths: Iterable[Any] | None = None, element_types: Iterable[Any] | None = None, post_ids: Iterable[Any] | None = None, merge_leve2: Any = None, platform: Any = None, limit: int = 20000, ) -> list[dict[str, Any]]: """Read source elements for DB-validating non-itemset evidence sources.""" clauses = [ "e.execution_id = %s", "e.source_table = %s", ] params: list[Any] = [int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE] clean_names = _clean_names(str(name) for name in _jsonish_to_list(element_names)) if clean_names: clauses.append("e.name = ANY(%s)") params.append(clean_names) clean_category_ids = _to_int_list(category_ids or []) if clean_category_ids: clauses.append("e.category_id = ANY(%s)") params.append(clean_category_ids) clean_category_names = _clean_names(str(name) for name in _jsonish_to_list(category_names)) if clean_category_names: clauses.append("(c.name = ANY(%s) OR regexp_replace(e.category_path, '^.*/', '') = ANY(%s))") params.extend([clean_category_names, clean_category_names]) clean_category_paths = _clean_names(str(path) for path in _jsonish_to_list(category_paths)) if clean_category_paths: clauses.append("(c.path = ANY(%s) OR e.category_path = ANY(%s))") params.extend([clean_category_paths, clean_category_paths]) clean_element_types = _clean_names(str(kind) for kind in _jsonish_to_list(element_types)) if clean_element_types: clauses.append("e.element_type = ANY(%s)") params.append(clean_element_types) clean_post_ids = _to_post_id_list(post_ids) if clean_post_ids: clauses.append("e.post_id = ANY(%s)") params.append(clean_post_ids) scope_filter = _scope_filter_payload(merge_leve2=merge_leve2, platform=platform) scope_active = _scope_is_active(scope_filter) join_post_sql = "" if scope_active: join_post_sql = "JOIN post post_scope ON post_scope.post_id = e.post_id" if scope_filter["merge_leve2"]: clauses.append("post_scope.merge_leve2 = ANY(%s)") params.append(scope_filter["merge_leve2"]) if scope_filter["platform"]: clauses.append("post_scope.platform = ANY(%s)") params.append(scope_filter["platform"]) params.append(int(limit)) return _fetch_all( f""" SELECT e.id, e.execution_id, e.post_id, e.source_table, e.source_element_id, e.element_type, e.element_sub_type, e.name, e.description, e.category_id, e.category_path, e.point_type, e.point_text, e.topic_point_id, c.name AS category_name, c.path AS category_full_path, c.source_type AS category_source_type, c.level AS category_level, c.parent_id AS category_parent_id, c.classified_as AS category_nature FROM pattern_mining_element e LEFT JOIN pattern_mining_category c ON c.id = e.category_id AND c.execution_id = e.execution_id {join_post_sql} WHERE {" AND ".join(clauses)} ORDER BY e.post_id, e.id LIMIT %s """, tuple(params), ) def query_topic_itemsets( execution_id: int, *, category_ids: Iterable[Any] | None = None, dimension_mode: str | None = None, min_support: int | None = None, min_item_count: int | None = None, max_item_count: int | None = None, sort_by: str = "absolute_support", limit: int = 100, merge_leve2: Any = None, platform: Any = None, ) -> list[dict[str, Any]]: scope_filter = _scope_filter_payload(merge_leve2=merge_leve2, platform=platform) scoped = _scope_is_active(scope_filter) clauses = [ "i.execution_id = %s", "i.scope = %s", ] where_params: list[Any] = [int(execution_id), TOPIC_SCOPE] clean_category_ids = _to_int_list(category_ids or []) for category_id in clean_category_ids: clauses.append( """ EXISTS ( SELECT 1 FROM pattern_itemset_item ii_filter WHERE ii_filter.itemset_id = i.id AND ii_filter.category_id = %s ) """ ) where_params.append(category_id) if dimension_mode: clauses.append("cfg.params ->> 'dimension_mode' = %s") where_params.append(dimension_mode) if min_support is not None: if scoped: clauses.append("COALESCE(scoped_posts.filtered_absolute_support, 0) >= %s") else: clauses.append("i.absolute_support >= %s") where_params.append(int(min_support)) if min_item_count is not None: clauses.append("i.item_count >= %s") where_params.append(int(min_item_count)) if max_item_count is not None: clauses.append("i.item_count <= %s") where_params.append(int(max_item_count)) if scoped: order_sql = { "item_count": "i.item_count DESC NULLS LAST, COALESCE(scoped_posts.filtered_absolute_support, 0) DESC", "support": "COALESCE(scoped_posts.filtered_absolute_support, 0) DESC, i.support DESC NULLS LAST", "absolute_support": "COALESCE(scoped_posts.filtered_absolute_support, 0) DESC, i.absolute_support DESC NULLS LAST", }.get( str(sort_by or "absolute_support"), "COALESCE(scoped_posts.filtered_absolute_support, 0) DESC, i.absolute_support DESC NULLS LAST", ) else: order_sql = { "support": "i.support DESC NULLS LAST, i.absolute_support DESC NULLS LAST", "item_count": "i.item_count DESC NULLS LAST, i.absolute_support DESC NULLS LAST", "absolute_support": "i.absolute_support DESC NULLS LAST, i.support DESC NULLS LAST", }.get(str(sort_by or "absolute_support"), "i.absolute_support DESC NULLS LAST, i.support DESC NULLS LAST") scope_clauses: list[str] = [] scope_params: list[Any] = [] if scope_filter["merge_leve2"]: scope_clauses.append("post_scope.merge_leve2 = ANY(%s)") scope_params.append(scope_filter["merge_leve2"]) if scope_filter["platform"]: scope_clauses.append("post_scope.platform = ANY(%s)") scope_params.append(scope_filter["platform"]) scope_where_sql = f" AND {' AND '.join(scope_clauses)}" if scope_clauses else "" if scoped: return _fetch_all( f""" WITH scoped_posts AS ( SELECT p_scope.itemset_id, COALESCE( array_agg(DISTINCT p_scope.post_id ORDER BY p_scope.post_id), ARRAY[]::varchar[] ) AS filtered_matched_post_ids, COUNT(DISTINCT p_scope.post_id) AS filtered_absolute_support FROM pattern_itemset_post p_scope JOIN post post_scope ON post_scope.post_id = p_scope.post_id WHERE p_scope.execution_id = %s {scope_where_sql} GROUP BY p_scope.itemset_id ) SELECT i.id, i.execution_id, i.mining_config_id, cfg.scope AS mining_config_scope, cfg.name AS mining_config_name, cfg.params AS mining_config_params, cfg.params ->> 'dimension_mode' AS dimension_mode, COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS target_depth, i.scope, i.combination_type, i.item_count, i.support, i.absolute_support, scoped_posts.filtered_matched_post_ids, scoped_posts.filtered_absolute_support, scoped_posts.filtered_absolute_support AS scoped_post_count, i.dimensions, i.is_cross_point, i.is_cross_layer, %s::jsonb AS scope_filter FROM pattern_itemset i JOIN pattern_mining_config cfg ON cfg.id = i.mining_config_id AND cfg.execution_id = i.execution_id AND cfg.scope = %s JOIN scoped_posts ON scoped_posts.itemset_id = i.id WHERE {" AND ".join(clauses)} ORDER BY {order_sql}, i.id LIMIT %s """, tuple([ int(execution_id), *scope_params, json.dumps(scope_filter, ensure_ascii=False), TOPIC_SCOPE, *where_params, int(limit), ]), ) return _fetch_all( f""" SELECT i.id, i.execution_id, i.mining_config_id, cfg.scope AS mining_config_scope, cfg.name AS mining_config_name, cfg.params AS mining_config_params, cfg.params ->> 'dimension_mode' AS dimension_mode, COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS target_depth, i.scope, i.combination_type, i.item_count, i.support, i.absolute_support, NULL::varchar[] AS filtered_matched_post_ids, NULL::bigint AS filtered_absolute_support, NULL::bigint AS scoped_post_count, i.dimensions, i.is_cross_point, i.is_cross_layer, %s::jsonb AS scope_filter FROM pattern_itemset i JOIN pattern_mining_config cfg ON cfg.id = i.mining_config_id AND cfg.execution_id = i.execution_id AND cfg.scope = %s WHERE {" AND ".join(clauses)} ORDER BY {order_sql}, i.id LIMIT %s """, tuple([ json.dumps({}, ensure_ascii=False), TOPIC_SCOPE, *where_params, int(limit), ]), ) def query_seed_points_for_itemsets( execution_id: int, itemset_ids: Iterable[Any], matched_post_ids: Iterable[Any], *, top_k: int = 30, ) -> list[dict[str, Any]]: """Rank searchable source points for scoped itemset support posts. The result is intentionally derived from the same PG element schema used by `element_bindings.sample_elements`, but it is a separate search-seed pool: it spans all validated support posts and ranks point_text by distinct-post coverage. """ clean_itemset_ids = _to_int_list(itemset_ids or []) clean_post_ids = _to_post_id_list(list(matched_post_ids or [])) if not clean_itemset_ids or not clean_post_ids: return [] rows = _fetch_all( """ WITH item_cats AS ( SELECT ii.id AS itemset_item_id, ii.itemset_id, ii.point_type, ii.dimension, ii.category_id, ii.category_path, ii.element_name FROM pattern_itemset_item ii JOIN pattern_itemset i ON i.id = ii.itemset_id AND i.execution_id = %s AND i.scope = %s WHERE ii.itemset_id = ANY(%s) AND ii.category_id IS NOT NULL ) SELECT e.id, e.post_id, e.source_table, e.source_element_id, e.point_type, e.point_text, e.element_type, e.name, e.category_id, e.category_path, e.topic_point_id, ic.itemset_id AS matched_itemset_id, ic.itemset_item_id AS matched_itemset_item_id, ic.category_id AS matched_category_id FROM pattern_mining_element e JOIN item_cats ic ON ic.category_id = e.category_id AND (ic.point_type IS NULL OR ic.point_type = e.point_type) AND (ic.dimension NOT IN ('实质', '形式', '意图') OR ic.dimension = e.element_type) AND (ic.element_name IS NULL OR ic.element_name = e.name) WHERE e.execution_id = %s AND e.source_table = %s AND e.post_id = ANY(%s) AND e.point_type IN ('灵感点', '目的点') AND e.point_text IS NOT NULL AND e.point_text <> '' ORDER BY e.point_text, e.point_type, e.post_id, e.id """, ( int(execution_id), TOPIC_SCOPE, clean_itemset_ids, int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, clean_post_ids, ), ) return _rank_query_seed_point_rows(rows, top_k=top_k) def query_seed_points_for_sources( execution_id: int, matched_post_ids: Iterable[Any], *, category_ids: Iterable[Any] | None = None, top_k: int = 30, ) -> list[dict[str, Any]]: """Rank searchable source points for any DB-validated evidence source.""" clean_post_ids = _to_post_id_list(list(matched_post_ids or [])) if not clean_post_ids: return [] clauses = [ "e.execution_id = %s", "e.source_table = %s", "e.post_id = ANY(%s)", "e.point_type IN ('灵感点', '目的点')", "e.point_text IS NOT NULL", "e.point_text <> ''", ] params: list[Any] = [int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, clean_post_ids] clean_category_ids = _to_int_list(category_ids or []) if clean_category_ids: clauses.append("e.category_id = ANY(%s)") params.append(clean_category_ids) rows = _fetch_all( f""" SELECT e.id, e.post_id, e.source_table, e.source_element_id, e.point_type, e.point_text, e.element_type, e.name, e.category_id, e.category_path, e.topic_point_id, e.category_id AS matched_category_id FROM pattern_mining_element e WHERE {" AND ".join(clauses)} ORDER BY e.point_text, e.point_type, e.post_id, e.id """, tuple(params), ) return _rank_query_seed_point_rows(rows, top_k=top_k) def _rank_query_seed_point_rows(rows: list[dict[str, Any]], *, top_k: int = 30) -> list[dict[str, Any]]: grouped: dict[tuple[str, str], dict[str, Any]] = {} for row in rows: point_text = str(row.get("point_text") or "").strip() point_type = str(row.get("point_type") or "").strip() if not point_text or point_type not in {"灵感点", "目的点"}: continue key = (point_text, point_type) group = grouped.setdefault( key, { "representative": dict(row), "post_ids": set(), "matched_category_ids": set(), "matched_itemset_item_ids": set(), }, ) if row.get("post_id") is not None: group["post_ids"].add(str(row["post_id"])) if row.get("matched_category_id") is not None: group["matched_category_ids"].add(int(row["matched_category_id"])) if row.get("matched_itemset_item_id") is not None: group["matched_itemset_item_ids"].add(int(row["matched_itemset_item_id"])) ranked = sorted( grouped.items(), key=lambda item: (-len(item[1]["post_ids"]), item[0][0], item[0][1]), ) result: list[dict[str, Any]] = [] for rank, ((point_text, point_type), group) in enumerate(ranked[: max(int(top_k or 30), 0)], start=1): representative = dict(group["representative"]) representative.pop("matched_itemset_id", None) representative.pop("matched_itemset_item_id", None) representative.pop("matched_category_id", None) representative["point_text"] = point_text representative["point_type"] = point_type representative["coverage_post_count"] = len(group["post_ids"]) representative["rank"] = rank representative["matched_category_ids"] = sorted(group["matched_category_ids"]) representative["matched_itemset_item_ids"] = sorted(group["matched_itemset_item_ids"]) result.append(representative) return result def query_weight_score_rows( execution_id: int, *, level: str, dimension: str, limit: int = 5000, ) -> list[dict[str, Any]]: """Aggregate PG topic elements into old weight-score JSON shapes.""" if level == "元素": return _fetch_all( """ SELECT name, element_type, category_id, category_path, COUNT(*) AS occurrence_count, COUNT(DISTINCT post_id) AS post_count FROM pattern_mining_element WHERE execution_id = %s AND source_table = %s AND element_type = %s AND name IS NOT NULL AND name <> '' GROUP BY name, element_type, category_id, category_path ORDER BY COUNT(DISTINCT post_id) DESC, COUNT(*) DESC, name LIMIT %s """, (int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, dimension, int(limit)), ) if level == "分类": return _fetch_all( """ SELECT COALESCE(c.name, regexp_replace(e.category_path, '^.*/', '')) AS category, e.category_id, e.category_path, e.element_type, COUNT(*) AS occurrence_count, COUNT(DISTINCT e.post_id) AS post_count, COALESCE(MAX(c.level), array_length(string_to_array(trim(both '/' from e.category_path), '/'), 1)) AS level FROM pattern_mining_element e LEFT JOIN pattern_mining_category c ON c.id = e.category_id AND c.execution_id = e.execution_id WHERE e.execution_id = %s AND e.source_table = %s AND e.element_type = %s AND e.category_id IS NOT NULL AND e.category_path IS NOT NULL GROUP BY e.category_id, e.category_path, e.element_type, c.name ORDER BY COUNT(DISTINCT e.post_id) DESC, COUNT(*) DESC, e.category_path LIMIT %s """, (int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, dimension, int(limit)), ) raise ValueError(f"unsupported weight score level: {level}")