sqlite_store.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726
  1. from __future__ import annotations
  2. import json
  3. import sqlite3
  4. from dataclasses import dataclass
  5. from datetime import datetime, timezone, timedelta
  6. from pathlib import Path
  7. from typing import Any
  8. from urllib.parse import urlparse
  9. from email.utils import parsedate_to_datetime
  10. from news_mcp.config import (
  11. NEWS_PRUNE_INTERVAL_HOURS,
  12. NEWS_PRUNING_ENABLED,
  13. NEWS_RETENTION_DAYS,
  14. )
  15. from news_mcp.entity_normalize import normalize_entities
  16. from news_mcp.trends_resolution import resolve_entity_via_trends
  17. @dataclass
  18. class ClusterRow:
  19. cluster_id: str
  20. topic: str
  21. payload: dict
  22. updated_at: datetime
  23. META_LAST_PRUNE_AT = "last_prune_at"
  24. def _article_key(article: dict[str, Any]) -> str:
  25. url = str(article.get("url") or "").strip()
  26. if not url:
  27. return str(article.get("title") or "")
  28. try:
  29. parsed = urlparse(url)
  30. parts = [p for p in parsed.path.split("/") if p]
  31. if parts:
  32. return parts[-1]
  33. except Exception:
  34. pass
  35. return url
  36. def _dedup_articles(articles: list[dict[str, Any]]) -> list[dict[str, Any]]:
  37. seen: set[str] = set()
  38. out: list[dict[str, Any]] = []
  39. for article in articles:
  40. key = _article_key(article)
  41. if key in seen:
  42. continue
  43. seen.add(key)
  44. out.append(article)
  45. return out
  46. def _has_valid_entity_resolutions(resolutions: Any, entities: list[str]) -> bool:
  47. if not isinstance(resolutions, list):
  48. return False
  49. if len(resolutions) != len(entities):
  50. return False
  51. for res in resolutions:
  52. if not isinstance(res, dict):
  53. return False
  54. if not res.get("normalized") or not res.get("canonical_label"):
  55. return False
  56. return True
  57. def sanitize_cluster_payload(cluster: dict[str, Any], *, include_resolutions: bool = True) -> dict[str, Any]:
  58. """Normalize cluster payload so every stored payload is internally consistent."""
  59. out = dict(cluster)
  60. raw_articles = out.get("articles", []) or []
  61. articles = [a for a in raw_articles if isinstance(a, dict)]
  62. out["articles"] = _dedup_articles(articles)
  63. raw_entities = out.get("entities", []) or []
  64. entities = normalize_entities(raw_entities)
  65. out["entities"] = entities
  66. if not include_resolutions:
  67. return out
  68. resolutions = out.get("entityResolutions", None)
  69. if entities:
  70. if not _has_valid_entity_resolutions(resolutions, entities):
  71. out["entityResolutions"] = [resolve_entity_via_trends(e) for e in entities]
  72. else:
  73. # Keep the empty case explicit and stable.
  74. out["entityResolutions"] = []
  75. return out
  76. class SQLiteClusterStore:
  77. def __init__(self, db_path: str | Path):
  78. self.db_path = str(db_path)
  79. self._init_db()
  80. def _conn(self) -> sqlite3.Connection:
  81. return sqlite3.connect(self.db_path)
  82. def _init_db(self) -> None:
  83. Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
  84. with self._conn() as conn:
  85. conn.execute("PRAGMA journal_mode=WAL")
  86. conn.execute("PRAGMA synchronous=NORMAL")
  87. conn.execute("PRAGMA busy_timeout=5000")
  88. conn.execute(
  89. """
  90. CREATE TABLE IF NOT EXISTS clusters (
  91. cluster_id TEXT PRIMARY KEY,
  92. topic TEXT NOT NULL,
  93. payload TEXT NOT NULL,
  94. updated_at TEXT NOT NULL,
  95. summary_payload TEXT,
  96. summary_updated_at TEXT
  97. )
  98. """
  99. )
  100. # If the table already exists without the summary columns,
  101. # add them (SQLite-friendly incremental migrations).
  102. for col_def in [
  103. "summary_payload TEXT",
  104. "summary_updated_at TEXT",
  105. ]:
  106. col = col_def.split()[0]
  107. try:
  108. conn.execute(f"ALTER TABLE clusters ADD COLUMN {col_def}")
  109. except sqlite3.OperationalError:
  110. pass
  111. conn.execute(
  112. "CREATE INDEX IF NOT EXISTS idx_clusters_topic ON clusters(topic)"
  113. )
  114. conn.execute(
  115. "CREATE INDEX IF NOT EXISTS idx_clusters_updated_at ON clusters(updated_at)"
  116. )
  117. try:
  118. cur = conn.execute("PRAGMA table_info(entity_metadata)")
  119. cols = [row[1] for row in cur.fetchall()]
  120. if cols and "entity_id" not in cols:
  121. conn.execute("DROP TABLE entity_metadata")
  122. except sqlite3.OperationalError:
  123. pass
  124. conn.execute(
  125. """
  126. CREATE TABLE IF NOT EXISTS entity_metadata (
  127. entity_id TEXT PRIMARY KEY,
  128. normalized_label TEXT NOT NULL,
  129. canonical_label TEXT,
  130. mid TEXT,
  131. sources_json TEXT,
  132. updated_at TEXT,
  133. last_requested_at TEXT
  134. )
  135. """
  136. )
  137. conn.execute(
  138. "CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_metadata_mid ON entity_metadata(mid) WHERE mid IS NOT NULL"
  139. )
  140. conn.execute(
  141. """
  142. CREATE TABLE IF NOT EXISTS feed_state (
  143. feed_key TEXT PRIMARY KEY,
  144. last_hash TEXT NOT NULL,
  145. updated_at TEXT NOT NULL
  146. )
  147. """
  148. )
  149. conn.execute(
  150. """
  151. CREATE TABLE IF NOT EXISTS meta (
  152. key TEXT PRIMARY KEY,
  153. value TEXT NOT NULL
  154. )
  155. """
  156. )
  157. def upsert_clusters(self, clusters: list[dict], topic: str) -> None:
  158. now = datetime.now(timezone.utc)
  159. with self._conn() as conn:
  160. for c in clusters:
  161. c = sanitize_cluster_payload(c)
  162. cluster_id = c["cluster_id"]
  163. payload = json.dumps(c, ensure_ascii=False)
  164. conn.execute(
  165. "INSERT INTO clusters(cluster_id, topic, payload, updated_at) VALUES(?,?,?,?) "
  166. "ON CONFLICT(cluster_id) DO UPDATE SET topic=excluded.topic, payload=excluded.payload, updated_at=excluded.updated_at",
  167. (cluster_id, topic, payload, now.isoformat()),
  168. )
  169. def upsert_cluster_summary(
  170. self,
  171. cluster_id: str,
  172. summary_payload: dict,
  173. ) -> None:
  174. now = datetime.now(timezone.utc).isoformat()
  175. with self._conn() as conn:
  176. conn.execute(
  177. "UPDATE clusters SET summary_payload=?, summary_updated_at=? WHERE cluster_id=?",
  178. (
  179. json.dumps(summary_payload, ensure_ascii=False),
  180. now,
  181. cluster_id,
  182. ),
  183. )
  184. def get_cluster_summary(self, cluster_id: str, ttl_hours: float) -> dict | None:
  185. cutoff = datetime.now(timezone.utc) - timedelta(hours=ttl_hours)
  186. cutoff_iso = cutoff.isoformat()
  187. with self._conn() as conn:
  188. cur = conn.execute(
  189. "SELECT summary_payload, summary_updated_at FROM clusters "
  190. "WHERE cluster_id=? AND summary_updated_at >= ?",
  191. (cluster_id, cutoff_iso),
  192. )
  193. row = cur.fetchone()
  194. if not row or not row[0]:
  195. return None
  196. return json.loads(row[0])
  197. def get_latest_clusters(self, topic: str, ttl_hours: float, limit: int) -> list[dict]:
  198. """Return newest clusters by *their own* timestamp.
  199. Filtering/sorting by the DB row's `updated_at` can drift away from the
  200. actual event time in `payload.timestamp`.
  201. """
  202. cutoff = datetime.now(timezone.utc) - timedelta(hours=float(ttl_hours))
  203. cutoff_ts = cutoff.timestamp()
  204. def _parse_payload_ts(ts: Any) -> float | None:
  205. if not ts:
  206. return None
  207. if isinstance(ts, (int, float)):
  208. return float(ts)
  209. text = str(ts).strip()
  210. try:
  211. dt = datetime.fromisoformat(text.replace('Z', '+00:00'))
  212. if dt.tzinfo is None:
  213. dt = dt.replace(tzinfo=timezone.utc)
  214. return dt.astimezone(timezone.utc).timestamp()
  215. except Exception:
  216. pass
  217. try:
  218. dt = parsedate_to_datetime(text)
  219. if dt.tzinfo is None:
  220. dt = dt.replace(tzinfo=timezone.utc)
  221. return dt.astimezone(timezone.utc).timestamp()
  222. except Exception:
  223. return None
  224. with self._conn() as conn:
  225. cur = conn.execute(
  226. "SELECT payload FROM clusters WHERE topic=? ORDER BY updated_at DESC",
  227. (topic,),
  228. )
  229. candidates = [json.loads(r[0]) for r in cur.fetchall()]
  230. filtered: list[dict] = []
  231. for c in candidates:
  232. ts = _parse_payload_ts(c.get("timestamp"))
  233. if ts is None:
  234. continue
  235. if ts >= cutoff_ts:
  236. filtered.append(c)
  237. filtered.sort(key=lambda c: _parse_payload_ts(c.get("timestamp")) or 0.0, reverse=True)
  238. return filtered[: int(limit)]
  239. def get_latest_clusters_all_topics(self, ttl_hours: float, limit: int) -> list[dict]:
  240. cutoff = datetime.now(timezone.utc) - timedelta(hours=float(ttl_hours))
  241. cutoff_ts = cutoff.timestamp()
  242. def _parse_payload_ts(ts: Any) -> float | None:
  243. if not ts:
  244. return None
  245. if isinstance(ts, (int, float)):
  246. return float(ts)
  247. text = str(ts).strip()
  248. try:
  249. dt = datetime.fromisoformat(text.replace('Z', '+00:00'))
  250. if dt.tzinfo is None:
  251. dt = dt.replace(tzinfo=timezone.utc)
  252. return dt.astimezone(timezone.utc).timestamp()
  253. except Exception:
  254. pass
  255. try:
  256. dt = parsedate_to_datetime(text)
  257. if dt.tzinfo is None:
  258. dt = dt.replace(tzinfo=timezone.utc)
  259. return dt.astimezone(timezone.utc).timestamp()
  260. except Exception:
  261. return None
  262. with self._conn() as conn:
  263. cur = conn.execute(
  264. "SELECT payload FROM clusters ORDER BY updated_at DESC",
  265. )
  266. candidates = [json.loads(r[0]) for r in cur.fetchall()]
  267. filtered: list[dict] = []
  268. for c in candidates:
  269. ts = _parse_payload_ts(c.get("timestamp"))
  270. if ts is None:
  271. continue
  272. if ts >= cutoff_ts:
  273. filtered.append(c)
  274. filtered.sort(key=lambda c: _parse_payload_ts(c.get("timestamp")) or 0.0, reverse=True)
  275. return filtered[: int(limit)]
  276. def get_cluster_by_id(self, cluster_id: str) -> dict | None:
  277. with self._conn() as conn:
  278. cur = conn.execute(
  279. "SELECT payload FROM clusters WHERE cluster_id=?",
  280. (cluster_id,),
  281. )
  282. row = cur.fetchone()
  283. return json.loads(row[0]) if row else None
  284. def get_feed_hash(self, feed_key: str) -> str | None:
  285. with self._conn() as conn:
  286. cur = conn.execute(
  287. "SELECT last_hash FROM feed_state WHERE feed_key=?",
  288. (feed_key,),
  289. )
  290. row = cur.fetchone()
  291. return row[0] if row else None
  292. def set_feed_hash(self, feed_key: str, last_hash: str) -> None:
  293. now = datetime.now(timezone.utc).isoformat()
  294. with self._conn() as conn:
  295. conn.execute(
  296. "INSERT INTO feed_state(feed_key, last_hash, updated_at) VALUES(?,?,?) "
  297. "ON CONFLICT(feed_key) DO UPDATE SET last_hash=excluded.last_hash, updated_at=excluded.updated_at",
  298. (feed_key, last_hash, now),
  299. )
  300. def get_feed_state(self, feed_key: str) -> dict | None:
  301. with self._conn() as conn:
  302. cur = conn.execute(
  303. "SELECT last_hash, updated_at FROM feed_state WHERE feed_key=?",
  304. (feed_key,),
  305. )
  306. row = cur.fetchone()
  307. if not row:
  308. return None
  309. return {"last_hash": row[0], "updated_at": row[1]}
  310. def get_all_feed_states(self) -> list[dict[str, Any]]:
  311. """All feed_state rows.
  312. The live writer keys feed state as ``newsfeeds:<sha1(comma_joined_urls)>``,
  313. so a hardcoded literal lookup never matches when more than one feed is
  314. configured. Use this for surfacing health information.
  315. """
  316. with self._conn() as conn:
  317. cur = conn.execute(
  318. "SELECT feed_key, last_hash, updated_at FROM feed_state ORDER BY updated_at DESC"
  319. )
  320. return [
  321. {"feed_key": row[0], "last_hash": row[1], "updated_at": row[2]}
  322. for row in cur.fetchall()
  323. ]
  324. def get_meta(self, key: str) -> str | None:
  325. with self._conn() as conn:
  326. cur = conn.execute("SELECT value FROM meta WHERE key=?", (key,))
  327. row = cur.fetchone()
  328. return row[0] if row else None
  329. def set_meta(self, key: str, value: str) -> None:
  330. with self._conn() as conn:
  331. conn.execute(
  332. "INSERT INTO meta(key, value) VALUES(?, ?) "
  333. "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
  334. (key, value),
  335. )
  336. def upsert_entity_metadata(
  337. self,
  338. normalized_label: str,
  339. canonical_label: str | None = None,
  340. mid: str | None = None,
  341. sources: list[str] | None = None,
  342. ) -> None:
  343. normalized_label = str(normalized_label or "").strip()
  344. if not normalized_label:
  345. return
  346. canonical_label = str(canonical_label).strip() if canonical_label else None
  347. mid = str(mid).strip() if mid else None
  348. entity_id = mid if mid else f"local:{normalized_label}"
  349. sources = sorted({s for s in (sources or []) if s})
  350. sources_json = json.dumps(sources, ensure_ascii=False)
  351. now = datetime.now(timezone.utc).isoformat()
  352. with self._conn() as conn:
  353. conn.execute(
  354. """
  355. INSERT INTO entity_metadata(entity_id, normalized_label, canonical_label, mid, sources_json, updated_at)
  356. VALUES(?,?,?,?,?,?)
  357. ON CONFLICT(entity_id) DO UPDATE SET
  358. canonical_label=excluded.canonical_label,
  359. mid=excluded.mid,
  360. sources_json=excluded.sources_json,
  361. updated_at=excluded.updated_at
  362. """,
  363. (entity_id, normalized_label, canonical_label, mid, sources_json, now),
  364. )
  365. def get_entity_metadata(self, normalized_label: str) -> dict[str, Any] | None:
  366. normalized_label = str(normalized_label or "").strip()
  367. if not normalized_label:
  368. return None
  369. with self._conn() as conn:
  370. cur = conn.execute(
  371. "SELECT entity_id, canonical_label, mid, sources_json, updated_at, last_requested_at "
  372. "FROM entity_metadata "
  373. "WHERE normalized_label=? "
  374. "ORDER BY CASE WHEN mid IS NOT NULL THEN 0 ELSE 1 END, "
  375. "COALESCE(last_requested_at, updated_at) DESC, updated_at DESC "
  376. "LIMIT 1",
  377. (normalized_label,),
  378. )
  379. row = cur.fetchone()
  380. if not row:
  381. return None
  382. sources = []
  383. if row[2]:
  384. try:
  385. sources = json.loads(row[2])
  386. except Exception:
  387. sources = []
  388. return {
  389. "entity_id": row[0],
  390. "normalized_label": normalized_label,
  391. "canonical_label": row[1],
  392. "mid": row[2],
  393. "sources": sources,
  394. "updated_at": row[4],
  395. "last_requested_at": row[5],
  396. }
  397. def record_entity_request(self, normalized_label: str, mid: str | None = None) -> None:
  398. normalized_label = str(normalized_label or "").strip()
  399. if not normalized_label:
  400. return
  401. mid = str(mid).strip() if mid else None
  402. entity_id = mid if mid else f"local:{normalized_label}"
  403. now = datetime.now(timezone.utc).isoformat()
  404. with self._conn() as conn:
  405. conn.execute(
  406. """
  407. INSERT INTO entity_metadata(entity_id, normalized_label, canonical_label, mid, sources_json, updated_at, last_requested_at)
  408. VALUES(?,?,?,?,?,?,?)
  409. ON CONFLICT(entity_id) DO UPDATE SET
  410. last_requested_at=excluded.last_requested_at
  411. """,
  412. (entity_id, normalized_label, None, mid, json.dumps([], ensure_ascii=False), now, now),
  413. )
  414. def prune_clusters(self, retention_days: float) -> int:
  415. retention_days = float(retention_days)
  416. if retention_days <= 0:
  417. return 0
  418. cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
  419. cutoff_iso = cutoff.isoformat()
  420. pruned_at = datetime.now(timezone.utc).isoformat()
  421. with self._conn() as conn:
  422. cur = conn.execute("DELETE FROM clusters WHERE updated_at < ?", (cutoff_iso,))
  423. deleted = int(cur.rowcount or 0)
  424. conn.execute(
  425. "INSERT INTO meta(key, value) VALUES(?, ?) "
  426. "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
  427. (META_LAST_PRUNE_AT, pruned_at),
  428. )
  429. return deleted
  430. def prune_if_due(self, pruning_enabled: bool, retention_days: float, interval_hours: float = 24.0) -> dict[str, Any]:
  431. retention_days = float(retention_days)
  432. interval_hours = float(interval_hours)
  433. if (not pruning_enabled) or retention_days <= 0:
  434. return {
  435. "enabled": bool(pruning_enabled),
  436. "deleted": 0,
  437. "due": False,
  438. "retention_days": retention_days,
  439. "interval_hours": interval_hours,
  440. "last_prune_at": self.get_meta(META_LAST_PRUNE_AT),
  441. }
  442. last_prune_at = self.get_meta(META_LAST_PRUNE_AT)
  443. now = datetime.now(timezone.utc)
  444. due = True
  445. if last_prune_at:
  446. try:
  447. last_dt = datetime.fromisoformat(last_prune_at)
  448. due = now - last_dt >= timedelta(hours=max(1.0, interval_hours))
  449. except Exception:
  450. due = True
  451. if not due:
  452. return {
  453. "enabled": True,
  454. "deleted": 0,
  455. "due": False,
  456. "retention_days": retention_days,
  457. "interval_hours": interval_hours,
  458. "last_prune_at": last_prune_at,
  459. }
  460. deleted = self.prune_clusters(retention_days)
  461. last_prune_at = self.get_meta(META_LAST_PRUNE_AT)
  462. return {
  463. "enabled": True,
  464. "deleted": deleted,
  465. "due": True,
  466. "retention_days": retention_days,
  467. "interval_hours": interval_hours,
  468. "last_prune_at": last_prune_at,
  469. }
  470. def get_prune_state(self, pruning_enabled: bool, retention_days: float, interval_hours: float = 24.0) -> dict[str, Any]:
  471. return {
  472. "enabled": bool(pruning_enabled),
  473. "retention_days": float(retention_days),
  474. "interval_hours": float(interval_hours),
  475. "last_prune_at": self.get_meta(META_LAST_PRUNE_AT),
  476. }
  477. # ------------------------------------------------------------------
  478. # Dashboard query helpers
  479. # ------------------------------------------------------------------
  480. def get_dashboard_stats(self) -> dict[str, Any]:
  481. """Aggregate status numbers for the health panel."""
  482. with self._conn() as conn:
  483. total_clusters = conn.execute("SELECT COUNT(*) FROM clusters").fetchone()[0]
  484. total_entities = conn.execute("SELECT COUNT(*) FROM entity_metadata").fetchone()[0]
  485. topic_counts = dict(conn.execute(
  486. "SELECT topic, COUNT(*) FROM clusters GROUP BY topic"
  487. ).fetchall())
  488. last_refresh = self.get_meta("last_refresh_at")
  489. feeds = {}
  490. for row in conn.execute("SELECT feed_key, last_hash, updated_at FROM feed_state"):
  491. feeds[row[0]] = {"last_hash": row[1], "updated_at": row[2]}
  492. last_prune = self.get_meta(META_LAST_PRUNE_AT)
  493. prune_state = self.get_prune_state(
  494. pruning_enabled=NEWS_PRUNING_ENABLED,
  495. retention_days=NEWS_RETENTION_DAYS,
  496. interval_hours=NEWS_PRUNE_INTERVAL_HOURS,
  497. )
  498. return {
  499. "total_clusters": total_clusters,
  500. "total_entities": total_entities,
  501. "clusters_by_topic": topic_counts,
  502. "last_refresh_at": last_refresh,
  503. "last_prune_at": last_prune,
  504. "prune_state": prune_state,
  505. "feeds": feeds,
  506. }
  507. def get_clusters_page(
  508. self,
  509. topic: str | None = None,
  510. hours: float = 24,
  511. limit: int = 20,
  512. offset: int = 0,
  513. ) -> list[dict[str, Any]]:
  514. """Paginated cluster listing for the dashboard."""
  515. cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
  516. query = "SELECT payload FROM clusters WHERE updated_at >= ?"
  517. params: list = [cutoff]
  518. if topic and topic != "all":
  519. query += " AND topic = ?"
  520. params.append(topic)
  521. query += " ORDER BY updated_at DESC LIMIT ? OFFSET ?"
  522. params.extend([limit, offset])
  523. with self._conn() as conn:
  524. cur = conn.execute(query, params)
  525. rows = cur.fetchall()
  526. clusters: list[dict[str, Any]] = []
  527. for (payload_text,) in rows:
  528. c = json.loads(payload_text)
  529. clusters.append({
  530. "cluster_id": c.get("cluster_id", ""),
  531. "headline": c.get("headline", ""),
  532. "topic": c.get("topic", ""),
  533. "sentiment": c.get("sentiment", "neutral"),
  534. "sentimentScore": c.get("sentimentScore"),
  535. "importance": c.get("importance", 0),
  536. "entities": c.get("entities", []),
  537. "sources": c.get("sources", []),
  538. "timestamp": c.get("timestamp", ""),
  539. "keywords": c.get("keywords", []),
  540. "article_count": len(c.get("articles", [])),
  541. })
  542. return clusters
  543. def get_sentiment_series(
  544. self,
  545. topic: str | None = None,
  546. hours: float = 24,
  547. bucket_hours: float = 1,
  548. ) -> list[dict[str, Any]]:
  549. """Sentiment score averaged per time bucket."""
  550. cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
  551. query = "SELECT payload FROM clusters WHERE updated_at >= ?"
  552. params: list = [cutoff]
  553. if topic and topic != "all":
  554. query += " AND topic = ?"
  555. params.append(topic)
  556. query += " ORDER BY updated_at ASC"
  557. with self._conn() as conn:
  558. cur = conn.execute(query, params)
  559. rows = cur.fetchall()
  560. def _parse_ts(ts: Any) -> datetime | None:
  561. if not ts:
  562. return None
  563. try:
  564. dt = datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
  565. if dt.tzinfo is None:
  566. dt = dt.replace(tzinfo=timezone.utc)
  567. return dt.astimezone(timezone.utc)
  568. except Exception:
  569. return None
  570. buckets: dict[datetime, list[float]] = {}
  571. for (payload_text,) in rows:
  572. c = json.loads(payload_text)
  573. dt = _parse_ts(c.get("timestamp"))
  574. score = c.get("sentimentScore")
  575. if dt is None or score is None:
  576. continue
  577. bucket_key = dt.replace(minute=0, second=0, microsecond=0)
  578. if bucket_hours > 1:
  579. bucket_key = bucket_key.replace(
  580. hour=(bucket_key.hour // int(bucket_hours)) * int(bucket_hours)
  581. )
  582. buckets.setdefault(bucket_key, []).append(float(score))
  583. series: list[dict[str, Any]] = []
  584. for bucket_key in sorted(buckets):
  585. scores = buckets[bucket_key]
  586. series.append({
  587. "time": bucket_key.isoformat(),
  588. "avg_sentiment": round(sum(scores) / len(scores), 3),
  589. "count": len(scores),
  590. "min": round(min(scores), 3),
  591. "max": round(max(scores), 3),
  592. })
  593. return series
  594. def get_entity_frequencies(
  595. self,
  596. hours: float = 24,
  597. limit: int = 30,
  598. ) -> list[dict[str, Any]]:
  599. """Top entities by mention count in recent clusters."""
  600. cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
  601. with self._conn() as conn:
  602. cur = conn.execute(
  603. "SELECT payload FROM clusters WHERE updated_at >= ? ORDER BY updated_at DESC LIMIT 500",
  604. (cutoff,),
  605. )
  606. rows = cur.fetchall()
  607. counter: dict[str, int] = {}
  608. for (payload_text,) in rows:
  609. c = json.loads(payload_text)
  610. for ent in c.get("entities", []):
  611. counter[ent] = counter.get(ent, 0) + 1
  612. sorted_entities = sorted(counter.items(), key=lambda x: -x[1])[:limit]
  613. result: list[dict[str, Any]] = []
  614. for label, count in sorted_entities:
  615. meta = self.get_entity_metadata(label)
  616. result.append({
  617. "label": label,
  618. "count": count,
  619. "canonical_label": meta["canonical_label"] if meta else label,
  620. "mid": meta["mid"] if meta else None,
  621. })
  622. return result
  623. def get_cluster_detail(self, cluster_id: str) -> dict[str, Any] | None:
  624. """Dashboard-optimized cluster detail fetch."""
  625. with self._conn() as conn:
  626. cur = conn.execute(
  627. "SELECT payload FROM clusters WHERE cluster_id = ?", (cluster_id,)
  628. )
  629. row = cur.fetchone()
  630. if not row:
  631. return None
  632. c = json.loads(row[0])
  633. summary = None
  634. if c.get("summary_payload"):
  635. try:
  636. summary = json.loads(c["summary_payload"])
  637. except Exception:
  638. pass
  639. return {
  640. "cluster_id": c.get("cluster_id"),
  641. "headline": c.get("headline", ""),
  642. "summary": c.get("summary", ""),
  643. "topic": c.get("topic", ""),
  644. "sentiment": c.get("sentiment", "neutral"),
  645. "sentimentScore": c.get("sentimentScore"),
  646. "importance": c.get("importance", 0),
  647. "entities": c.get("entities", []),
  648. "entityResolutions": c.get("entityResolutions", []),
  649. "keywords": c.get("keywords", []),
  650. "sources": c.get("sources", []),
  651. "timestamp": c.get("timestamp", ""),
  652. "first_seen": c.get("first_seen", ""),
  653. "last_updated": c.get("last_updated", ""),
  654. "article_count": len(c.get("articles", [])),
  655. "articles": c.get("articles", []),
  656. "summary_text": summary.get("mergedSummary", "") if summary else "",
  657. "key_facts": summary.get("keyFacts", []) if summary else [],
  658. }