pg_pattern_repository.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810
  1. """Read-only PostgreSQL Pattern V2 repository for DemandAgent.
  2. This module is the DemandAgent boundary to PG `open_aigc.public` Pattern V2
  3. facts. It intentionally exposes the old Demand-side query shapes while reading
  4. only PG main-truth tables:
  5. - pattern_mining_execution
  6. - pattern_mining_category
  7. - pattern_mining_element
  8. - pattern_mining_config
  9. - pattern_itemset
  10. - pattern_itemset_item
  11. - pattern_itemset_post
  12. """
  13. from __future__ import annotations
  14. import json
  15. import os
  16. import re
  17. from contextlib import contextmanager
  18. from datetime import date, datetime
  19. from typing import Any, Iterable, Mapping, Sequence
  20. try:
  21. import psycopg2
  22. from psycopg2.extras import RealDictCursor
  23. except Exception: # pragma: no cover - exercised in environments without deps
  24. psycopg2 = None
  25. RealDictCursor = None
  26. PG_PATTERN_SOURCE_SYSTEM = "pg_pattern_v2"
  27. TOPIC_SCOPE = "topic"
  28. TOPIC_ELEMENT_SOURCE_TABLE = "post_decode_topic_point_element"
  29. ELEMENT_TYPE_DIMENSIONS = {"实质", "形式", "意图"}
  30. _READONLY_SQL_RE = re.compile(r"^\s*(select|with|show|explain)\b", re.IGNORECASE | re.DOTALL)
  31. _BLOCKED_SQL_RE = re.compile(
  32. r"\b(insert|update|delete|create|alter|drop|truncate|copy|grant|revoke|merge|call)\b",
  33. re.IGNORECASE,
  34. )
  35. def _load_dotenv_once() -> None:
  36. try:
  37. from dotenv import load_dotenv
  38. except Exception:
  39. return
  40. load_dotenv()
  41. def _resolve_dsn() -> str:
  42. _load_dotenv_once()
  43. dsn = os.getenv("DEMAND_PATTERN_PG_DSN") or os.getenv("PGVECTOR_DSN")
  44. if not dsn:
  45. raise RuntimeError("PG Pattern V2 DSN missing: set DEMAND_PATTERN_PG_DSN or PGVECTOR_DSN")
  46. return dsn
  47. def _assert_readonly_sql(sql: str) -> None:
  48. if not _READONLY_SQL_RE.search(sql):
  49. raise RuntimeError("PG Pattern repository only allows SELECT/SHOW/EXPLAIN queries")
  50. if _BLOCKED_SQL_RE.search(sql):
  51. raise RuntimeError("PG Pattern repository blocked a non-read-only SQL keyword")
  52. def _is_element_type_dimension(dimension: Any) -> bool:
  53. """Return whether an item dimension maps to pattern_mining_element.element_type."""
  54. return dimension in ELEMENT_TYPE_DIMENSIONS
  55. @contextmanager
  56. def _connect_readonly():
  57. if psycopg2 is None:
  58. raise RuntimeError("psycopg2 is required for PG Pattern V2 reads")
  59. conn = psycopg2.connect(_resolve_dsn())
  60. try:
  61. conn.set_session(readonly=True, autocommit=True)
  62. yield conn
  63. finally:
  64. conn.close()
  65. def _fetch_all(sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> list[dict[str, Any]]:
  66. _assert_readonly_sql(sql)
  67. with _connect_readonly() as conn:
  68. with conn.cursor(cursor_factory=RealDictCursor) as cursor:
  69. cursor.execute(sql, params)
  70. return [_row_to_dict(row) for row in cursor.fetchall()]
  71. def _fetch_one(sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> dict[str, Any] | None:
  72. rows = _fetch_all(sql, params)
  73. return rows[0] if rows else None
  74. def _cursor_fetch_all(cursor: Any, sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> list[dict[str, Any]]:
  75. _assert_readonly_sql(sql)
  76. cursor.execute(sql, params)
  77. return [_row_to_dict(row) for row in cursor.fetchall()]
  78. def _cursor_fetch_one(cursor: Any, sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> dict[str, Any] | None:
  79. rows = _cursor_fetch_all(cursor, sql, params)
  80. return rows[0] if rows else None
  81. def _serialize_db_value(value: Any) -> Any:
  82. if isinstance(value, (datetime, date)):
  83. return value.isoformat()
  84. return value
  85. def _row_to_dict(row: Any) -> dict[str, Any]:
  86. if row is None:
  87. return {}
  88. mapping = getattr(row, "_mapping", row)
  89. return {key: _serialize_db_value(value) for key, value in dict(mapping).items()}
  90. def _jsonish_to_list(value: Any) -> list[Any]:
  91. if value is None:
  92. return []
  93. if isinstance(value, list):
  94. return value
  95. if isinstance(value, (tuple, set)):
  96. return list(value)
  97. if isinstance(value, bytes):
  98. value = value.decode("utf-8")
  99. if isinstance(value, str):
  100. raw = value.strip()
  101. if not raw:
  102. return []
  103. try:
  104. parsed = json.loads(raw)
  105. except json.JSONDecodeError:
  106. return [part.strip() for part in raw.split(",") if part.strip()]
  107. if isinstance(parsed, list):
  108. return parsed
  109. if parsed is None:
  110. return []
  111. return [parsed]
  112. return [value]
  113. def _to_post_id_list(value: Any) -> list[str]:
  114. post_ids: list[str] = []
  115. seen: set[str] = set()
  116. for item in _jsonish_to_list(value):
  117. post_id = str(item).strip() if item is not None else ""
  118. if post_id and post_id not in seen:
  119. seen.add(post_id)
  120. post_ids.append(post_id)
  121. return post_ids
  122. def _to_int_list(values: Any) -> list[int]:
  123. result: list[int] = []
  124. seen: set[int] = set()
  125. for value in _jsonish_to_list(values):
  126. if value is None or value == "":
  127. continue
  128. try:
  129. int_value = int(value)
  130. except (TypeError, ValueError):
  131. continue
  132. if int_value not in seen:
  133. seen.add(int_value)
  134. result.append(int_value)
  135. return result
  136. def _clean_names(names: Iterable[str]) -> list[str]:
  137. result: list[str] = []
  138. seen: set[str] = set()
  139. for raw in names:
  140. name = str(raw).strip() if raw is not None else ""
  141. if name and name not in seen:
  142. seen.add(name)
  143. result.append(name)
  144. return result
  145. def query_execution_for_evidence(execution_id: int) -> dict[str, Any] | None:
  146. """Read PG Pattern V2 execution facts for evidence validation."""
  147. row = _fetch_one(
  148. """
  149. SELECT
  150. id,
  151. snapshot_date,
  152. is_current,
  153. is_retired,
  154. status,
  155. post_filter,
  156. post_count,
  157. post_ids_digest,
  158. classify_execution_id,
  159. category_count,
  160. element_count,
  161. topic_itemset_count AS itemset_count,
  162. topic_itemset_count,
  163. topic_element_itemset_count,
  164. cross_itemset_count,
  165. script_sequence_count,
  166. paragraph_itemset_count,
  167. start_time,
  168. end_time,
  169. error_message
  170. FROM pattern_mining_execution
  171. WHERE id = %s
  172. LIMIT 1
  173. """,
  174. (int(execution_id),),
  175. )
  176. if not row:
  177. return None
  178. row["pattern_source_system"] = PG_PATTERN_SOURCE_SYSTEM
  179. return row
  180. def query_latest_success_execution_id() -> int | None:
  181. row = _fetch_one(
  182. """
  183. SELECT id
  184. FROM pattern_mining_execution
  185. WHERE status = 'success'
  186. AND COALESCE(topic_itemset_count, 0) > 0
  187. ORDER BY snapshot_date DESC NULLS LAST, id DESC
  188. LIMIT 1
  189. """
  190. )
  191. return int(row["id"]) if row else None
  192. def query_itemset_evidence(execution_id: int, itemset_ids: Iterable[Any]) -> list[dict[str, Any]]:
  193. """Read PG topic-scope itemset facts and aggregate matched posts."""
  194. clean_ids = _to_int_list(itemset_ids or [])
  195. if not clean_ids:
  196. return []
  197. rows = _fetch_all(
  198. """
  199. SELECT
  200. i.id,
  201. i.execution_id,
  202. i.mining_config_id,
  203. cfg.execution_id AS mining_config_execution_id,
  204. cfg.scope AS mining_config_scope,
  205. cfg.name AS mining_config_name,
  206. cfg.algorithm AS mining_config_algorithm,
  207. cfg.params::text AS mining_config_params,
  208. cfg.params ->> 'dimension_mode' AS mining_config_dimension_mode,
  209. COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS mining_config_target_depth,
  210. i.scope,
  211. i.combination_type,
  212. i.item_count,
  213. i.support,
  214. i.absolute_support,
  215. i.dimensions::text AS dimensions,
  216. i.is_cross_point,
  217. i.is_cross_layer,
  218. COALESCE(
  219. array_agg(p.post_id ORDER BY p.post_id)
  220. FILTER (WHERE p.post_id IS NOT NULL),
  221. ARRAY[]::varchar[]
  222. ) AS matched_post_ids
  223. FROM pattern_itemset i
  224. JOIN pattern_mining_config cfg
  225. ON cfg.id = i.mining_config_id
  226. AND cfg.execution_id = i.execution_id
  227. AND cfg.scope = %s
  228. LEFT JOIN pattern_itemset_post p
  229. ON p.itemset_id = i.id
  230. AND p.execution_id = i.execution_id
  231. WHERE i.execution_id = %s
  232. AND i.scope = %s
  233. AND i.id = ANY(%s)
  234. GROUP BY
  235. i.id,
  236. i.execution_id,
  237. i.mining_config_id,
  238. cfg.execution_id,
  239. cfg.scope,
  240. cfg.name,
  241. cfg.algorithm,
  242. cfg.params::text,
  243. cfg.params ->> 'dimension_mode',
  244. COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth'),
  245. i.scope,
  246. i.combination_type,
  247. i.item_count,
  248. i.support,
  249. i.absolute_support,
  250. i.dimensions::text,
  251. i.is_cross_point,
  252. i.is_cross_layer
  253. ORDER BY i.id
  254. """,
  255. (TOPIC_SCOPE, int(execution_id), TOPIC_SCOPE, clean_ids),
  256. )
  257. result: list[dict[str, Any]] = []
  258. for row in rows:
  259. itemset = dict(row)
  260. for key in (
  261. "id",
  262. "execution_id",
  263. "mining_config_id",
  264. "mining_config_execution_id",
  265. "item_count",
  266. "absolute_support",
  267. ):
  268. if itemset.get(key) is not None:
  269. itemset[key] = int(itemset[key])
  270. itemset["support"] = float(itemset["support"]) if itemset.get("support") is not None else None
  271. itemset["is_cross_point"] = bool(itemset.get("is_cross_point"))
  272. itemset["is_cross_layer"] = bool(itemset.get("is_cross_layer"))
  273. itemset["dimensions"] = _jsonish_to_list(itemset.get("dimensions"))
  274. if isinstance(itemset.get("mining_config_params"), str):
  275. try:
  276. itemset["mining_config_params"] = json.loads(itemset["mining_config_params"])
  277. except json.JSONDecodeError:
  278. pass
  279. itemset["matched_post_ids"] = _to_post_id_list(itemset.get("matched_post_ids"))
  280. result.append(itemset)
  281. by_id = {itemset["id"]: itemset for itemset in result}
  282. return [by_id[itemset_id] for itemset_id in clean_ids if itemset_id in by_id]
  283. def query_itemset_items_with_categories(
  284. execution_id: int,
  285. itemset_ids: Iterable[Any],
  286. ) -> list[dict[str, Any]]:
  287. """Read PG itemset items and their exact category tree nodes."""
  288. clean_ids = _to_int_list(itemset_ids or [])
  289. if not clean_ids:
  290. return []
  291. rows = _fetch_all(
  292. """
  293. SELECT
  294. ii.id AS itemset_item_id,
  295. ii.itemset_id,
  296. ii.layer,
  297. ii.point_type,
  298. ii.dimension,
  299. ii.category_id,
  300. ii.category_path,
  301. ii.element_name,
  302. ii.element_id,
  303. ii.post_count,
  304. c.id AS bound_category_id,
  305. c.execution_id AS category_execution_id,
  306. c.source_stable_id AS category_source_stable_id,
  307. c.source_type AS category_source_type,
  308. c.name AS category_name,
  309. c.path AS category_full_path,
  310. c.level AS category_level,
  311. c.parent_id AS category_parent_id,
  312. c.parent_source_stable_id AS category_parent_source_stable_id,
  313. c.classified_as AS category_nature
  314. FROM pattern_itemset_item ii
  315. JOIN pattern_itemset i
  316. ON i.id = ii.itemset_id
  317. AND i.execution_id = %s
  318. AND i.scope = %s
  319. LEFT JOIN pattern_mining_category c
  320. ON c.id = ii.category_id
  321. AND c.execution_id = i.execution_id
  322. WHERE ii.itemset_id = ANY(%s)
  323. ORDER BY ii.itemset_id, ii.id
  324. """,
  325. (int(execution_id), TOPIC_SCOPE, clean_ids),
  326. )
  327. result: list[dict[str, Any]] = []
  328. for row in rows:
  329. item = dict(row)
  330. for key in (
  331. "itemset_item_id",
  332. "itemset_id",
  333. "category_id",
  334. "element_id",
  335. "post_count",
  336. "bound_category_id",
  337. "category_execution_id",
  338. "category_source_stable_id",
  339. "category_level",
  340. "category_parent_id",
  341. "category_parent_source_stable_id",
  342. ):
  343. if item.get(key) is not None:
  344. item[key] = int(item[key])
  345. item["category_found"] = item.get("bound_category_id") is not None
  346. result.append(item)
  347. return result
  348. def query_element_bindings_for_items(
  349. execution_id: int,
  350. itemset_items: Sequence[Mapping[str, Any]],
  351. post_ids: Iterable[Any] | None = None,
  352. limit_per_item: int = 200,
  353. ) -> list[dict[str, Any]]:
  354. """Bind itemset items to real PG Pattern V2 source-post elements."""
  355. clean_post_ids = _to_post_id_list(post_ids)
  356. if not itemset_items:
  357. return []
  358. bindings: list[dict[str, Any]] = []
  359. with _connect_readonly() as conn:
  360. with conn.cursor(cursor_factory=RealDictCursor) as cursor:
  361. for raw_item in itemset_items:
  362. item = dict(raw_item)
  363. category_id = item.get("bound_category_id") or item.get("category_id")
  364. if category_id is None:
  365. continue
  366. clauses = [
  367. "execution_id = %s",
  368. "category_id = %s",
  369. "source_table = %s",
  370. ]
  371. params: list[Any] = [int(execution_id), int(category_id), TOPIC_ELEMENT_SOURCE_TABLE]
  372. if item.get("point_type"):
  373. clauses.append("point_type = %s")
  374. params.append(item["point_type"])
  375. if _is_element_type_dimension(item.get("dimension")):
  376. clauses.append("element_type = %s")
  377. params.append(item["dimension"])
  378. if item.get("element_name"):
  379. clauses.append("name = %s")
  380. params.append(item["element_name"])
  381. if clean_post_ids:
  382. clauses.append("post_id = ANY(%s)")
  383. params.append(clean_post_ids)
  384. where_sql = " AND ".join(clauses)
  385. count_row = _cursor_fetch_one(
  386. cursor,
  387. f"""
  388. SELECT
  389. COUNT(*) AS matched_element_count,
  390. COUNT(DISTINCT post_id) AS matched_post_count
  391. FROM pattern_mining_element
  392. WHERE {where_sql}
  393. """,
  394. tuple(params),
  395. ) or {}
  396. post_rows = _cursor_fetch_all(
  397. cursor,
  398. f"""
  399. SELECT DISTINCT post_id
  400. FROM pattern_mining_element
  401. WHERE {where_sql}
  402. ORDER BY post_id
  403. LIMIT %s
  404. """,
  405. tuple(params + [int(limit_per_item)]),
  406. )
  407. sample_rows = _cursor_fetch_all(
  408. cursor,
  409. f"""
  410. SELECT
  411. id,
  412. post_id,
  413. source_table,
  414. source_element_id,
  415. point_type,
  416. point_text,
  417. element_type,
  418. name,
  419. category_id,
  420. category_path,
  421. topic_point_id
  422. FROM pattern_mining_element
  423. WHERE {where_sql}
  424. ORDER BY post_id, id
  425. LIMIT %s
  426. """,
  427. tuple(params + [min(int(limit_per_item), 20)]),
  428. )
  429. bindings.append(
  430. {
  431. "itemset_id": item.get("itemset_id"),
  432. "itemset_item_id": item.get("itemset_item_id"),
  433. "category_id": int(category_id),
  434. "point_type": item.get("point_type"),
  435. "dimension": item.get("dimension"),
  436. "element_name": item.get("element_name"),
  437. "matched_element_count": int(count_row.get("matched_element_count") or 0),
  438. "matched_post_count": int(count_row.get("matched_post_count") or 0),
  439. "matched_post_ids": _to_post_id_list([row["post_id"] for row in post_rows]),
  440. "sample_elements": sample_rows,
  441. }
  442. )
  443. return bindings
  444. def query_video_ids_by_names(execution_id: int, names: Iterable[str]) -> list[str]:
  445. """Resolve names to PG source post IDs by category name/path leaf or element name."""
  446. clean_names = _clean_names(names)
  447. if not clean_names:
  448. return []
  449. rows = _fetch_all(
  450. """
  451. WITH target_categories AS (
  452. SELECT id
  453. FROM pattern_mining_category
  454. WHERE execution_id = %s
  455. AND (
  456. name = ANY(%s)
  457. OR regexp_replace(path, '^.*/', '') = ANY(%s)
  458. )
  459. ),
  460. matched_elements AS (
  461. SELECT post_id
  462. FROM pattern_mining_element
  463. WHERE execution_id = %s
  464. AND source_table = %s
  465. AND (
  466. name = ANY(%s)
  467. OR category_id IN (SELECT id FROM target_categories)
  468. )
  469. )
  470. SELECT DISTINCT post_id
  471. FROM matched_elements
  472. WHERE post_id IS NOT NULL
  473. ORDER BY post_id
  474. """,
  475. (int(execution_id), clean_names, clean_names, int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, clean_names),
  476. )
  477. return _to_post_id_list([row["post_id"] for row in rows])
  478. def query_category_level(execution_id: int, name: str) -> int | None:
  479. clean_name = str(name).strip() if name is not None else ""
  480. if not clean_name:
  481. return None
  482. row = _fetch_one(
  483. """
  484. SELECT level
  485. FROM pattern_mining_category
  486. WHERE execution_id = %s
  487. AND (name = %s OR regexp_replace(path, '^.*/', '') = %s)
  488. ORDER BY id DESC
  489. LIMIT 1
  490. """,
  491. (int(execution_id), clean_name, clean_name),
  492. )
  493. return int(row["level"]) if row and row.get("level") is not None else None
  494. def query_case_ids_by_post_ids(post_ids: Iterable[Any]) -> list[dict[str, Any]]:
  495. """PG Pattern V2 has post_id semantics for MVP exact evidence.
  496. DemandAgent no longer reads the old MySQL decode-case table in this PG-only
  497. path. `case_ids` is filled from validated post IDs by the evidence builder;
  498. `decode_case_ids` remains an optional empty compatibility field.
  499. """
  500. return []
  501. def query_categories(
  502. execution_id: int,
  503. *,
  504. source_type: str | None = None,
  505. keyword: str | None = None,
  506. category_ids: Iterable[Any] | None = None,
  507. limit: int | None = None,
  508. ) -> list[dict[str, Any]]:
  509. clauses = ["execution_id = %s"]
  510. params: list[Any] = [int(execution_id)]
  511. if source_type:
  512. clauses.append("source_type = %s")
  513. params.append(source_type)
  514. if keyword:
  515. clauses.append("(name ILIKE %s OR path ILIKE %s)")
  516. like = f"%{keyword}%"
  517. params.extend([like, like])
  518. clean_ids = _to_int_list(category_ids or [])
  519. if clean_ids:
  520. clauses.append("id = ANY(%s)")
  521. params.append(clean_ids)
  522. limit_sql = "LIMIT %s" if limit else ""
  523. if limit:
  524. params.append(int(limit))
  525. return _fetch_all(
  526. f"""
  527. SELECT
  528. id,
  529. execution_id,
  530. source_stable_id,
  531. source_type,
  532. name,
  533. description,
  534. path,
  535. level,
  536. parent_id,
  537. parent_source_stable_id,
  538. element_count,
  539. classified_as AS category_nature
  540. FROM pattern_mining_category
  541. WHERE {" AND ".join(clauses)}
  542. ORDER BY source_type, path, id
  543. {limit_sql}
  544. """,
  545. tuple(params),
  546. )
  547. def query_elements(
  548. execution_id: int,
  549. *,
  550. keyword: str | None = None,
  551. element_type: str | None = None,
  552. post_ids: Iterable[Any] | None = None,
  553. category_ids: Iterable[Any] | None = None,
  554. limit: int | None = None,
  555. ) -> list[dict[str, Any]]:
  556. clauses = [
  557. "execution_id = %s",
  558. "source_table = %s",
  559. ]
  560. params: list[Any] = [int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE]
  561. if keyword:
  562. clauses.append("name ILIKE %s")
  563. params.append(f"%{keyword}%")
  564. if element_type:
  565. clauses.append("element_type = %s")
  566. params.append(element_type)
  567. clean_posts = _to_post_id_list(post_ids)
  568. if clean_posts:
  569. clauses.append("post_id = ANY(%s)")
  570. params.append(clean_posts)
  571. clean_categories = _to_int_list(category_ids or [])
  572. if clean_categories:
  573. clauses.append("category_id = ANY(%s)")
  574. params.append(clean_categories)
  575. limit_sql = "LIMIT %s" if limit else ""
  576. if limit:
  577. params.append(int(limit))
  578. return _fetch_all(
  579. f"""
  580. SELECT
  581. id,
  582. execution_id,
  583. post_id,
  584. source_table,
  585. source_element_id,
  586. element_type,
  587. element_sub_type,
  588. name,
  589. description,
  590. category_id,
  591. category_path,
  592. point_type,
  593. point_text,
  594. topic_point_id
  595. FROM pattern_mining_element
  596. WHERE {" AND ".join(clauses)}
  597. ORDER BY post_id, id
  598. {limit_sql}
  599. """,
  600. tuple(params),
  601. )
  602. def query_topic_itemsets(
  603. execution_id: int,
  604. *,
  605. category_ids: Iterable[Any] | None = None,
  606. dimension_mode: str | None = None,
  607. min_support: int | None = None,
  608. min_item_count: int | None = None,
  609. max_item_count: int | None = None,
  610. sort_by: str = "absolute_support",
  611. limit: int = 100,
  612. ) -> list[dict[str, Any]]:
  613. clauses = [
  614. "i.execution_id = %s",
  615. "i.scope = %s",
  616. ]
  617. params: list[Any] = [int(execution_id), TOPIC_SCOPE]
  618. clean_category_ids = _to_int_list(category_ids or [])
  619. for category_id in clean_category_ids:
  620. clauses.append(
  621. """
  622. EXISTS (
  623. SELECT 1
  624. FROM pattern_itemset_item ii_filter
  625. WHERE ii_filter.itemset_id = i.id
  626. AND ii_filter.category_id = %s
  627. )
  628. """
  629. )
  630. params.append(category_id)
  631. if dimension_mode:
  632. clauses.append("cfg.params ->> 'dimension_mode' = %s")
  633. params.append(dimension_mode)
  634. if min_support is not None:
  635. clauses.append("i.absolute_support >= %s")
  636. params.append(int(min_support))
  637. if min_item_count is not None:
  638. clauses.append("i.item_count >= %s")
  639. params.append(int(min_item_count))
  640. if max_item_count is not None:
  641. clauses.append("i.item_count <= %s")
  642. params.append(int(max_item_count))
  643. order_sql = {
  644. "support": "i.support DESC NULLS LAST, i.absolute_support DESC NULLS LAST",
  645. "item_count": "i.item_count DESC NULLS LAST, i.absolute_support DESC NULLS LAST",
  646. "absolute_support": "i.absolute_support DESC NULLS LAST, i.support DESC NULLS LAST",
  647. }.get(str(sort_by or "absolute_support"), "i.absolute_support DESC NULLS LAST, i.support DESC NULLS LAST")
  648. params.append(int(limit))
  649. return _fetch_all(
  650. f"""
  651. SELECT
  652. i.id,
  653. i.execution_id,
  654. i.mining_config_id,
  655. cfg.scope AS mining_config_scope,
  656. cfg.name AS mining_config_name,
  657. cfg.params AS mining_config_params,
  658. cfg.params ->> 'dimension_mode' AS dimension_mode,
  659. COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS target_depth,
  660. i.scope,
  661. i.combination_type,
  662. i.item_count,
  663. i.support,
  664. i.absolute_support,
  665. i.dimensions,
  666. i.is_cross_point,
  667. i.is_cross_layer
  668. FROM pattern_itemset i
  669. JOIN pattern_mining_config cfg
  670. ON cfg.id = i.mining_config_id
  671. AND cfg.execution_id = i.execution_id
  672. AND cfg.scope = %s
  673. WHERE {" AND ".join(clauses)}
  674. ORDER BY {order_sql}, i.id
  675. LIMIT %s
  676. """,
  677. tuple([TOPIC_SCOPE, *params]),
  678. )
  679. def query_weight_score_rows(
  680. execution_id: int,
  681. *,
  682. level: str,
  683. dimension: str,
  684. limit: int = 5000,
  685. ) -> list[dict[str, Any]]:
  686. """Aggregate PG topic elements into old weight-score JSON shapes."""
  687. if level == "元素":
  688. return _fetch_all(
  689. """
  690. SELECT
  691. name,
  692. element_type,
  693. category_id,
  694. category_path,
  695. COUNT(*) AS occurrence_count,
  696. COUNT(DISTINCT post_id) AS post_count
  697. FROM pattern_mining_element
  698. WHERE execution_id = %s
  699. AND source_table = %s
  700. AND element_type = %s
  701. AND name IS NOT NULL
  702. AND name <> ''
  703. GROUP BY name, element_type, category_id, category_path
  704. ORDER BY COUNT(DISTINCT post_id) DESC, COUNT(*) DESC, name
  705. LIMIT %s
  706. """,
  707. (int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, dimension, int(limit)),
  708. )
  709. if level == "分类":
  710. return _fetch_all(
  711. """
  712. SELECT
  713. COALESCE(c.name, regexp_replace(e.category_path, '^.*/', '')) AS category,
  714. e.category_id,
  715. e.category_path,
  716. e.element_type,
  717. COUNT(*) AS occurrence_count,
  718. COUNT(DISTINCT e.post_id) AS post_count,
  719. COALESCE(MAX(c.level), array_length(string_to_array(trim(both '/' from e.category_path), '/'), 1)) AS level
  720. FROM pattern_mining_element e
  721. LEFT JOIN pattern_mining_category c
  722. ON c.id = e.category_id
  723. AND c.execution_id = e.execution_id
  724. WHERE e.execution_id = %s
  725. AND e.source_table = %s
  726. AND e.element_type = %s
  727. AND e.category_id IS NOT NULL
  728. AND e.category_path IS NOT NULL
  729. GROUP BY e.category_id, e.category_path, e.element_type, c.name
  730. ORDER BY COUNT(DISTINCT e.post_id) DESC, COUNT(*) DESC, e.category_path
  731. LIMIT %s
  732. """,
  733. (int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, dimension, int(limit)),
  734. )
  735. raise ValueError(f"unsupported weight score level: {level}")