news-mcp/news_mcp/storage/sqlite_store.py

from __future__ import annotations

import json
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from email.utils import parsedate_to_datetime

from news_mcp.config import (
    NEWS_PRUNE_INTERVAL_HOURS,
    NEWS_PRUNING_ENABLED,
    NEWS_RETENTION_DAYS,
)
from news_mcp.entity_normalize import normalize_entities
from news_mcp.trends_resolution import resolve_entity_via_trends


def _normalize_ts(ts: Any) -> str:
    """Parse any timestamp string and return ISO 8601 UTC.
    
    Handles ISO 8601, RFC 2822/HTTP-date, and unix epoch seconds.
    Returns empty string if unparseable.
    """
    if ts is None:
        return ""
    if isinstance(ts, (int, float)):
        try:
            dt = datetime.fromtimestamp(float(ts), tz=timezone.utc)
            return dt.strftime("%Y-%m-%dT%H:%M:%S+00:00")
        except Exception:
            return ""
    text = str(ts).strip()
    if not text:
        return ""
    # Try ISO 8601
    try:
        dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00")
    except Exception:
        pass
    # Try RFC 2822 / HTTP-date
    try:
        dt = parsedate_to_datetime(text)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00")
    except Exception:
        pass
    # Return original if we can't parse — better than losing data
    return text


def _read_ts(ts: Any) -> float | None:
    """Parse a stored, already-normalized ISO 8601 UTC timestamp to a unix float.

    All payload.timestamp / payload.first_seen / payload.last_updated values
    are guaranteed YYYY-MM-DDTHH:MM:SS+00:00 at write time (enforced by
    sanitize_cluster_payload → _normalize_ts).  Only datetime.fromisoformat is
    needed here.  Do NOT add RFC 2822 / parsedate_to_datetime fallbacks — if
    this function can't parse a stored timestamp it means the normalization
    pipeline has a bug that should be fixed there, not papered over here.
    """
    if not ts:
        return None
    try:
        dt = datetime.fromisoformat(str(ts).strip())
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc).timestamp()
    except Exception:
        return None


@dataclass
class ClusterRow:
    cluster_id: str
    topic: str
    payload: dict
    updated_at: datetime


META_LAST_PRUNE_AT = "last_prune_at"


def _article_key(article: dict[str, Any]) -> str:
    url = str(article.get("url") or "").strip()
    if not url:
        return str(article.get("title") or "")
    try:
        parsed = urlparse(url)
        parts = [p for p in parsed.path.split("/") if p]
        if parts:
            return parts[-1]
    except Exception:
        pass
    return url


def _dedup_articles(articles: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen: set[str] = set()
    out: list[dict[str, Any]] = []
    for article in articles:
        key = _article_key(article)
        if key in seen:
            continue
        seen.add(key)
        out.append(article)
    return out


def _has_valid_entity_resolutions(resolutions: Any, entities: list[str]) -> bool:
    if not isinstance(resolutions, list):
        return False
    if len(resolutions) != len(entities):
        return False
    for res in resolutions:
        if not isinstance(res, dict):
            return False
        if not res.get("normalized") or not res.get("canonical_label"):
            return False
    return True


def sanitize_cluster_payload(cluster: dict[str, Any], *, include_resolutions: bool = True) -> dict[str, Any]:
    """Normalize cluster payload so every stored payload is internally consistent."""
    out = dict(cluster)

    raw_articles = out.get("articles", []) or []
    articles = [a for a in raw_articles if isinstance(a, dict)]
    # Normalize article timestamps
    for a in articles:
        if "timestamp" in a:
            a["timestamp"] = _normalize_ts(a["timestamp"])
    out["articles"] = _dedup_articles(articles)

    raw_entities = out.get("entities", []) or []
    entities = normalize_entities(raw_entities)
    out["entities"] = entities

    # Normalize cluster-level timestamps
    for field in ("timestamp", "last_updated", "first_seen"):
        if field in out and out[field]:
            out[field] = _normalize_ts(out[field])

    if not include_resolutions:
        return out

    resolutions = out.get("entityResolutions", None)
    if entities:
        if not _has_valid_entity_resolutions(resolutions, entities):
            out["entityResolutions"] = [resolve_entity_via_trends(e) for e in entities]
    else:
        # Keep the empty case explicit and stable.
        out["entityResolutions"] = []

    return out


class SQLiteClusterStore:
    def __init__(self, db_path: str | Path):
        self.db_path = str(db_path)
        self._init_db()

    def _conn(self) -> sqlite3.Connection:
        return sqlite3.connect(self.db_path)

    def _init_db(self) -> None:
        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
        with self._conn() as conn:
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute("PRAGMA synchronous=NORMAL")
            conn.execute("PRAGMA busy_timeout=5000")
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS clusters (
                  cluster_id TEXT PRIMARY KEY,
                  topic TEXT NOT NULL,
                  payload TEXT NOT NULL,
                  updated_at TEXT NOT NULL,
                  summary_payload TEXT,
                  summary_updated_at TEXT
                )
                """
            )

            # If the table already exists without the summary columns,
            # add them (SQLite-friendly incremental migrations).
            for col_def in [
                "summary_payload TEXT",
                "summary_updated_at TEXT",
            ]:
                col = col_def.split()[0]
                try:
                    conn.execute(f"ALTER TABLE clusters ADD COLUMN {col_def}")
                except sqlite3.OperationalError:
                    pass
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_clusters_topic ON clusters(topic)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_clusters_updated_at ON clusters(updated_at)"
            )

            try:
                cur = conn.execute("PRAGMA table_info(entity_metadata)")
                cols = [row[1] for row in cur.fetchall()]
                if cols and "entity_id" not in cols:
                    conn.execute("DROP TABLE entity_metadata")
            except sqlite3.OperationalError:
                pass
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS entity_metadata (
                  entity_id TEXT PRIMARY KEY,
                  normalized_label TEXT NOT NULL,
                  canonical_label TEXT,
                  mid TEXT,
                  sources_json TEXT,
                  updated_at TEXT,
                  last_requested_at TEXT
                )
                """
            )
            conn.execute(
                "CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_metadata_mid ON entity_metadata(mid) WHERE mid IS NOT NULL"
            )

            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS feed_state (
                  feed_key TEXT PRIMARY KEY,
                  last_hash TEXT NOT NULL,
                  last_item_count INTEGER,
                  enabled INTEGER DEFAULT 1,
                  updated_at TEXT NOT NULL
                )
                """
            )
            try:
                conn.execute("ALTER TABLE feed_state ADD COLUMN enabled INTEGER DEFAULT 1")
            except sqlite3.OperationalError:
                pass

            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS meta (
                  key TEXT PRIMARY KEY,
                  value TEXT NOT NULL
                )
                """
            )

    def upsert_clusters(self, clusters: list[dict], topic: str) -> None:
        now = datetime.now(timezone.utc)
        with self._conn() as conn:
            for c in clusters:
                c = sanitize_cluster_payload(c)
                cluster_id = c["cluster_id"]
                payload = json.dumps(c, ensure_ascii=False)
                conn.execute(
                    "INSERT INTO clusters(cluster_id, topic, payload, updated_at) VALUES(?,?,?,?) "
                    "ON CONFLICT(cluster_id) DO UPDATE SET topic=excluded.topic, payload=excluded.payload, updated_at=excluded.updated_at",
                    (cluster_id, topic, payload, now.isoformat()),
                )

    def upsert_cluster_summary(
        self,
        cluster_id: str,
        summary_payload: dict,
    ) -> None:
        now = datetime.now(timezone.utc).isoformat()
        with self._conn() as conn:
            conn.execute(
                "UPDATE clusters SET summary_payload=?, summary_updated_at=? WHERE cluster_id=?",
                (
                    json.dumps(summary_payload, ensure_ascii=False),
                    now,
                    cluster_id,
                ),
            )

    def get_cluster_summary(self, cluster_id: str, ttl_hours: float) -> dict | None:
        cutoff = datetime.now(timezone.utc) - timedelta(hours=ttl_hours)
        cutoff_iso = cutoff.isoformat()
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT summary_payload, summary_updated_at FROM clusters "
                "WHERE cluster_id=? AND summary_updated_at >= ?",
                (cluster_id, cutoff_iso),
            )
            row = cur.fetchone()
            if not row or not row[0]:
                return None
            return json.loads(row[0])

    def get_latest_clusters(self, topic: str, ttl_hours: float, limit: int) -> list[dict]:
        """Return newest clusters by their own event timestamp (payload.timestamp).

        payload.timestamp is guaranteed ISO 8601 UTC — use _read_ts, not raw
        JSON parsing with RFC 2822 fallbacks.
        """
        cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=float(ttl_hours))).timestamp()

        with self._conn() as conn:
            cur = conn.execute(
                "SELECT payload FROM clusters WHERE topic=? ORDER BY updated_at DESC",
                (topic,),
            )
            candidates = [json.loads(r[0]) for r in cur.fetchall()]

        filtered = [c for c in candidates if (_read_ts(c.get("timestamp")) or 0.0) >= cutoff_ts]
        filtered.sort(key=lambda c: _read_ts(c.get("timestamp")) or 0.0, reverse=True)
        return filtered[: int(limit)]

    def get_latest_clusters_all_topics(self, ttl_hours: float, limit: int) -> list[dict]:
        """Return newest clusters across all topics by event timestamp.

        payload.timestamp is guaranteed ISO 8601 UTC — use _read_ts, not raw
        JSON parsing with RFC 2822 fallbacks.
        """
        cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=float(ttl_hours))).timestamp()

        with self._conn() as conn:
            cur = conn.execute("SELECT payload FROM clusters ORDER BY updated_at DESC")
            candidates = [json.loads(r[0]) for r in cur.fetchall()]

        filtered = [c for c in candidates if (_read_ts(c.get("timestamp")) or 0.0) >= cutoff_ts]
        filtered.sort(key=lambda c: _read_ts(c.get("timestamp")) or 0.0, reverse=True)
        return filtered[: int(limit)]

    def get_cluster_by_id(self, cluster_id: str) -> dict | None:
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT payload FROM clusters WHERE cluster_id=?",
                (cluster_id,),
            )
            row = cur.fetchone()
            return json.loads(row[0]) if row else None

    def get_feed_hash(self, feed_key: str) -> str | None:
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT last_hash FROM feed_state WHERE feed_key=?",
                (feed_key,),
            )
            row = cur.fetchone()
            return row[0] if row else None

    def set_feed_hash(self, feed_key: str, last_hash: str) -> None:
        self.set_feed_state(feed_key, last_hash, None)

    def set_feed_state(self, feed_key: str, last_hash: str, last_item_count: int | None = None) -> None:
        now = datetime.now(timezone.utc).isoformat()
        with self._conn() as conn:
            conn.execute(
                "INSERT INTO feed_state(feed_key, last_hash, last_item_count, updated_at) VALUES(?,?,?,?) "
                "ON CONFLICT(feed_key) DO UPDATE SET last_hash=excluded.last_hash, last_item_count=excluded.last_item_count, updated_at=excluded.updated_at",
                (feed_key, last_hash, last_item_count, now),
            )

    def get_feed_state(self, feed_key: str) -> dict | None:
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT last_hash, updated_at FROM feed_state WHERE feed_key=?",
                (feed_key,),
            )
            row = cur.fetchone()
            if not row:
                return None
            return {"last_hash": row[0], "updated_at": row[1]}

    def get_all_feed_states(self) -> list[dict[str, Any]]:
        """All feed_state rows."""
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT feed_key, last_hash, last_item_count, enabled, updated_at FROM feed_state ORDER BY updated_at DESC"
            )
            return [
                {
                    "feed_key": row[0],
                    "last_hash": row[1],
                    "last_item_count": row[2],
                    "enabled": bool(row[3]),
                    "updated_at": row[4],
                }
                for row in cur.fetchall()
            ]

    def feed_ensure_seeded(self, feed_urls: list[str]) -> None:
        """Insert any feed URLs not yet present in feed_state (enabled by default)."""
        if not feed_urls:
            return
        with self._conn() as conn:
            for url in feed_urls:
                conn.execute(
                    "INSERT OR IGNORE INTO feed_state(feed_key, last_hash, last_item_count, enabled, updated_at) VALUES(?, '', 0, 1, '')",
                    (url,),
                )

    def get_feed_state_list(self) -> list[dict[str, Any]]:
        """Return all feeds with enabled/disabled status for the dashboard."""
        return self.get_all_feed_states()

    def set_feed_enabled(self, feed_url: str, enabled: bool) -> bool:
        """Toggle a feed's enabled state. Returns True if the feed existed and was updated."""
        with self._conn() as conn:
            cur = conn.execute(
                "UPDATE feed_state SET enabled = ? WHERE feed_key = ?",
                (1 if enabled else 0, feed_url),
            )
            return cur.rowcount > 0

    def get_enabled_feed_urls(self, feed_urls: list[str]) -> list[str]:
        """From a list of configured feed URLs, return only those that are enabled in feed_state.

        URLs not yet present in feed_state are seeded as enabled.
        """
        self.feed_ensure_seeded(feed_urls)
        with self._conn() as conn:
            placeholders = ",".join("?" for _ in feed_urls) if feed_urls else ""
            if placeholders:
                cur = conn.execute(
                    f"SELECT feed_key FROM feed_state WHERE feed_key IN ({placeholders}) AND enabled = 1",
                    feed_urls,
                )
            else:
                cur = conn.execute("SELECT feed_key FROM feed_state WHERE enabled = 1")
            return [row[0] for row in cur.fetchall()]

    def get_meta(self, key: str) -> str | None:
        with self._conn() as conn:
            cur = conn.execute("SELECT value FROM meta WHERE key=?", (key,))
            row = cur.fetchone()
            return row[0] if row else None

    def set_meta(self, key: str, value: str) -> None:
        with self._conn() as conn:
            conn.execute(
                "INSERT INTO meta(key, value) VALUES(?, ?) "
                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
                (key, value),
            )

    def upsert_entity_metadata(
        self,
        normalized_label: str,
        canonical_label: str | None = None,
        mid: str | None = None,
        sources: list[str] | None = None,
    ) -> None:
        normalized_label = str(normalized_label or "").strip()
        if not normalized_label:
            return
        canonical_label = str(canonical_label).strip() if canonical_label else None
        mid = str(mid).strip() if mid else None
        entity_id = mid if mid else f"local:{normalized_label}"
        sources = sorted({s for s in (sources or []) if s})
        sources_json = json.dumps(sources, ensure_ascii=False)
        now = datetime.now(timezone.utc).isoformat()
        with self._conn() as conn:
            conn.execute(
                """
                INSERT INTO entity_metadata(entity_id, normalized_label, canonical_label, mid, sources_json, updated_at)
                VALUES(?,?,?,?,?,?)
                ON CONFLICT(entity_id) DO UPDATE SET
                  canonical_label=excluded.canonical_label,
                  mid=excluded.mid,
                  sources_json=excluded.sources_json,
                  updated_at=excluded.updated_at
                """,
                (entity_id, normalized_label, canonical_label, mid, sources_json, now),
            )

    def get_entity_metadata(self, normalized_label: str) -> dict[str, Any] | None:
        normalized_label = str(normalized_label or "").strip()
        if not normalized_label:
            return None
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT entity_id, canonical_label, mid, sources_json, updated_at, last_requested_at "
                "FROM entity_metadata "
                "WHERE normalized_label=? "
                "ORDER BY CASE WHEN mid IS NOT NULL THEN 0 ELSE 1 END, "
                "COALESCE(last_requested_at, updated_at) DESC, updated_at DESC "
                "LIMIT 1",
                (normalized_label,),
            )
            row = cur.fetchone()
            if not row:
                return None
            sources = []
            if row[2]:
                try:
                    sources = json.loads(row[2])
                except Exception:
                    sources = []
            return {
                "entity_id": row[0],
                "normalized_label": normalized_label,
                "canonical_label": row[1],
                "mid": row[2],
                "sources": sources,
                "updated_at": row[4],
                "last_requested_at": row[5],
            }

    def record_entity_request(self, normalized_label: str, mid: str | None = None) -> None:
        normalized_label = str(normalized_label or "").strip()
        if not normalized_label:
            return
        mid = str(mid).strip() if mid else None
        entity_id = mid if mid else f"local:{normalized_label}"
        now = datetime.now(timezone.utc).isoformat()
        with self._conn() as conn:
            conn.execute(
                """
                INSERT INTO entity_metadata(entity_id, normalized_label, canonical_label, mid, sources_json, updated_at, last_requested_at)
                VALUES(?,?,?,?,?,?,?)
                ON CONFLICT(entity_id) DO UPDATE SET
                  last_requested_at=excluded.last_requested_at
                """,
                (entity_id, normalized_label, None, mid, json.dumps([], ensure_ascii=False), now, now),
            )

    def get_failed_enrichment_clusters(self, max_retries: int = 3) -> list[dict]:
        """Return clusters whose last enrichment failed and haven't exceeded max_retries.

        These are candidates for re-enrichment on the next polling cycle.
        """
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT payload FROM clusters "
                "WHERE json_extract(payload, '$.enrichment_failed_at') IS NOT NULL "
                "AND (json_extract(payload, '$.enrichment_retry_count') IS NULL "
                "     OR json_extract(payload, '$.enrichment_retry_count') < ?) "
                "ORDER BY updated_at DESC LIMIT 500",
                (max_retries,),
            )
            rows = cur.fetchall()
        return [json.loads(r[0]) for r in rows]

    def prune_clusters(self, retention_days: float) -> int:
        retention_days = float(retention_days)
        if retention_days <= 0:
            return 0
        cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
        cutoff_iso = cutoff.isoformat()
        pruned_at = datetime.now(timezone.utc).isoformat()
        with self._conn() as conn:
            cur = conn.execute("DELETE FROM clusters WHERE updated_at < ?", (cutoff_iso,))
            deleted = int(cur.rowcount or 0)
            conn.execute(
                "INSERT INTO meta(key, value) VALUES(?, ?) "
                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
                (META_LAST_PRUNE_AT, pruned_at),
            )
            return deleted

    def prune_if_due(self, pruning_enabled: bool, retention_days: float, interval_hours: float = 24.0) -> dict[str, Any]:
        retention_days = float(retention_days)
        interval_hours = float(interval_hours)
        if (not pruning_enabled) or retention_days <= 0:
            return {
                "enabled": bool(pruning_enabled),
                "deleted": 0,
                "due": False,
                "retention_days": retention_days,
                "interval_hours": interval_hours,
                "last_prune_at": self.get_meta(META_LAST_PRUNE_AT),
            }

        last_prune_at = self.get_meta(META_LAST_PRUNE_AT)
        now = datetime.now(timezone.utc)
        due = True
        if last_prune_at:
            try:
                last_dt = datetime.fromisoformat(last_prune_at)
                due = now - last_dt >= timedelta(hours=max(1.0, interval_hours))
            except Exception:
                due = True

        if not due:
            return {
                "enabled": True,
                "deleted": 0,
                "due": False,
                "retention_days": retention_days,
                "interval_hours": interval_hours,
                "last_prune_at": last_prune_at,
            }

        deleted = self.prune_clusters(retention_days)
        last_prune_at = self.get_meta(META_LAST_PRUNE_AT)
        return {
            "enabled": True,
            "deleted": deleted,
            "due": True,
            "retention_days": retention_days,
            "interval_hours": interval_hours,
            "last_prune_at": last_prune_at,
        }

    def get_prune_state(self, pruning_enabled: bool, retention_days: float, interval_hours: float = 24.0) -> dict[str, Any]:
        return {
            "enabled": bool(pruning_enabled),
            "retention_days": float(retention_days),
            "interval_hours": float(interval_hours),
            "last_prune_at": self.get_meta(META_LAST_PRUNE_AT),
        }

    # ------------------------------------------------------------------
    # Dashboard query helpers
    # ------------------------------------------------------------------

    def get_dashboard_stats(self) -> dict[str, Any]:
        """Aggregate status numbers for the health panel."""
        with self._conn() as conn:
            total_clusters = conn.execute("SELECT COUNT(*) FROM clusters").fetchone()[0]
            total_entities = conn.execute("SELECT COUNT(*) FROM entity_metadata").fetchone()[0]
            topic_counts = dict(conn.execute(
                "SELECT topic, COUNT(*) FROM clusters GROUP BY topic"
            ).fetchall())
            last_refresh = self.get_meta("last_refresh_at")
            feeds = {}
            for row in conn.execute("SELECT feed_key, last_hash, last_item_count, updated_at FROM feed_state"):
                feeds[row[0]] = {"last_hash": row[1], "last_item_count": row[2], "updated_at": row[3]}
            last_prune = self.get_meta(META_LAST_PRUNE_AT)
            prune_state = self.get_prune_state(
                pruning_enabled=NEWS_PRUNING_ENABLED,
                retention_days=NEWS_RETENTION_DAYS,
                interval_hours=NEWS_PRUNE_INTERVAL_HOURS,
            )
            return {
                "total_clusters": total_clusters,
                "total_entities": total_entities,
                "clusters_by_topic": topic_counts,
                "last_refresh_at": last_refresh,
                "last_prune_at": last_prune,
                "prune_state": prune_state,
                "feeds": feeds,
            }

    def get_clusters_page(
        self,
        topic: str | None = None,
        hours: float = 24,
        limit: int = 20,
        offset: int = 0,
    ) -> list[dict[str, Any]]:
        """Paginated cluster listing filtered by payload.timestamp (event time).

        payload.timestamp is guaranteed ISO 8601 UTC — filtered and sorted
        using _read_ts, not updated_at (row modification time).
        """
        cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp()

        query = "SELECT payload FROM clusters"
        params: list = []
        if topic and topic != "all":
            query += " WHERE topic = ?"
            params.append(topic)

        with self._conn() as conn:
            rows = conn.execute(query, params).fetchall()

        filtered = [json.loads(r[0]) for r in rows]
        filtered = [c for c in filtered if (_read_ts(c.get("timestamp")) or 0.0) >= cutoff_ts]
        filtered.sort(key=lambda c: _read_ts(c.get("timestamp")) or 0.0, reverse=True)
        page = filtered[offset:offset + limit]

        return [
            {
                "cluster_id": c.get("cluster_id", ""),
                "headline": c.get("headline", ""),
                "topic": c.get("topic", ""),
                "sentiment": c.get("sentiment", "neutral"),
                "sentimentScore": c.get("sentimentScore"),
                "importance": c.get("importance", 0),
                "entities": c.get("entities", []),
                "sources": c.get("sources", []),
                "timestamp": c.get("timestamp", ""),
                "keywords": c.get("keywords", []),
                "article_count": len(c.get("articles", [])),
            }
            for c in page
        ]

    def get_sentiment_series(
            self,
            topic: str | None = None,
            hours: float = 24,
            bucket_hours: float = 1,
        ) -> list[dict[str, Any]]:
            """Sentiment score averaged per time bucket.

            Filters by payload.timestamp (event time, ISO 8601 UTC guaranteed).
            """
            cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp()
            query = "SELECT payload FROM clusters"
            params: list = []
            if topic and topic != "all":
                query += " WHERE topic = ?"
                params.append(topic)

            with self._conn() as conn:
                rows = conn.execute(query, params).fetchall()

            buckets: dict[datetime, list[float]] = {}
            for (payload_text,) in rows:
                c = json.loads(payload_text)
                ts = _read_ts(c.get("timestamp"))
                score = c.get("sentimentScore")
                if ts is None or score is None:
                    continue
                if ts < cutoff_ts:
                    continue
                dt = datetime.fromtimestamp(ts, tz=timezone.utc)
                bucket_key = dt.replace(minute=0, second=0, microsecond=0)
                if bucket_hours > 1:
                    bucket_key = bucket_key.replace(
                        hour=(bucket_key.hour // int(bucket_hours)) * int(bucket_hours)
                    )
                buckets.setdefault(bucket_key, []).append(float(score))

            return [
                {
                    "time": bucket_key.isoformat(),
                    "avg_sentiment": round(sum(scores) / len(scores), 3),
                    "count": len(scores),
                    "min": round(min(scores), 3),
                    "max": round(max(scores), 3),
                }
                for bucket_key, scores in sorted(buckets.items())
            ]

    def get_entity_frequencies(
        self,
        hours: float = 24,
        limit: int = 30,
    ) -> list[dict[str, Any]]:
        """Top entities by mention count filtered by payload.timestamp (ISO 8601 UTC guaranteed)."""
        cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp()

        with self._conn() as conn:
            rows = conn.execute("SELECT payload FROM clusters").fetchall()

        counter: dict[str, int] = {}
        for (payload_text,) in rows:
            c = json.loads(payload_text)
            if (_read_ts(c.get("timestamp")) or 0.0) < cutoff_ts:
                continue
            for ent in c.get("entities", []):
                counter[ent] = counter.get(ent, 0) + 1

        result: list[dict[str, Any]] = []
        for label, count in sorted(counter.items(), key=lambda x: -x[1])[:limit]:
            meta = self.get_entity_metadata(label)
            result.append({
                "label": label,
                "count": count,
                "canonical_label": meta["canonical_label"] if meta else label,
                "mid": meta["mid"] if meta else None,
            })
        return result

    def get_cluster_detail(self, cluster_id: str) -> dict[str, Any] | None:
        """Dashboard-optimized cluster detail fetch."""
        with self._conn() as conn:
            cur = conn.execute(
                "SELECT payload FROM clusters WHERE cluster_id = ?", (cluster_id,)
            )
            row = cur.fetchone()
            if not row:
                return None
            c = json.loads(row[0])
            summary = None
            if c.get("summary_payload"):
                try:
                    summary = json.loads(c["summary_payload"])
                except Exception:
                    pass
            return {
                "cluster_id": c.get("cluster_id"),
                "headline": c.get("headline", ""),
                "summary": c.get("summary", ""),
                "topic": c.get("topic", ""),
                "sentiment": c.get("sentiment", "neutral"),
                "sentimentScore": c.get("sentimentScore"),
                "importance": c.get("importance", 0),
                "entities": c.get("entities", []),
                "entityResolutions": c.get("entityResolutions", []),
                "keywords": c.get("keywords", []),
                "sources": c.get("sources", []),
                "timestamp": c.get("timestamp", ""),
                "first_seen": c.get("first_seen", ""),
                "last_updated": c.get("last_updated", ""),
                "article_count": len(c.get("articles", [])),
                "articles": c.get("articles", []),
                "summary_text": summary.get("mergedSummary", "") if summary else "",
                "key_facts": summary.get("keyFacts", []) if summary else [],
            }