news-mcp/news_mcp/dashboard/dashboard_store.py

from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from typing import Any

from news_mcp.config import (
    NEWS_PRUNE_INTERVAL_HOURS,
    NEWS_PRUNING_ENABLED,
    NEWS_REFRESH_INTERVAL_SECONDS,
    NEWS_RETENTION_DAYS,
    DEFAULT_TOPICS,
)
from news_mcp.storage.sqlite_store import SQLiteClusterStore, _read_ts


class DashboardStore:
    """Read-only query layer for the dashboard."""

    def __init__(self, store=None):
        if store is not None:
            self._store = store
        else:
            from news_mcp.config import DB_PATH
            self._store = SQLiteClusterStore(DB_PATH)

    # ── Health & Stats ──────────────────────────────────────────────

    def get_dashboard_stats(self) -> dict[str, Any]:
        with self._store._conn() as conn:
            total_clusters = conn.execute("SELECT COUNT(*) FROM clusters").fetchone()[0]
            total_entities = conn.execute("SELECT COUNT(*) FROM entity_metadata").fetchone()[0]
            cluster_entities = conn.execute(
                "SELECT COUNT(DISTINCT e.value) "
                "FROM clusters, json_each(clusters.payload, '$.entities') AS e"
            ).fetchone()[0]
            topic_counts = dict(conn.execute(
                "SELECT topic, COUNT(*) FROM clusters GROUP BY topic"
            ).fetchall())

        last_refresh = self._store.get_meta("last_refresh_at")
        last_prune = self._store.get_meta("last_prune_at")
        
        # Freshness: did a refresh happen recently? (within 2x the configured interval)
        fresh = False
        if last_refresh:
            try:
                dt = datetime.fromisoformat(last_refresh.replace("Z", "+00:00"))
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=timezone.utc)
                age_hours = (datetime.now(timezone.utc) - dt).total_seconds() / 3600
                fresh = age_hours < max(1.0, NEWS_REFRESH_INTERVAL_SECONDS / 3600) * 2
            except Exception:
                pass

        feeds = {}
        with self._store._conn() as conn:
            for row in conn.execute("SELECT feed_key, last_hash, last_item_count, enabled, updated_at FROM feed_state ORDER BY updated_at DESC"):
                feeds[row[0]] = {"last_hash": row[1], "last_item_count": row[2], "enabled": bool(row[3]), "updated_at": row[4]}

        return {
            "total_clusters": total_clusters,
            "total_entities": total_entities,
            "cluster_entities": cluster_entities,
            "clusters_by_topic": topic_counts,
            "last_refresh_at": last_refresh,
            "last_prune_at": last_prune,
            "data_fresh": fresh,
            "feeds": feeds,
            "feed_count": len(feeds),
            "pruning": {
                "enabled": NEWS_PRUNING_ENABLED,
                "retention_days": NEWS_RETENTION_DAYS,
                "interval_hours": NEWS_PRUNE_INTERVAL_HOURS,
                "last_prune_at": last_prune,
            },
        }

    # ── Clusters ────────────────────────────────────────────────────

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

        payload.timestamp is guaranteed ISO 8601 UTC — uses _read_ts from
        sqlite_store. Do NOT filter by updated_at (row mod time).
        Returns {"clusters": [...], "total": int}.
        """
        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._store._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 {
            "clusters": [
                {
                    "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
            ],
            "total": len(filtered),
        }

    def get_cluster_detail(self, cluster_id: str) -> dict[str, Any] | None:
        with self._store._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 [],
            }

    # ── Sentiment Series ────────────────────────────────────────────

    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._store._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 or 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())
        ]

    # ── Entity Frequencies ──────────────────────────────────────────

    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._store._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._store.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

    # ── Keyword Frequencies ─────────────────────────────────────────

    def get_keyword_frequencies(
        self,
        hours: float = 24,
        limit: int = 30,
    ) -> list[dict[str, Any]]:
        """Top keywords by occurrence count filtered by payload.timestamp (ISO 8601 UTC guaranteed).

        Excludes keywords that are already entities in the same cluster,
        and excludes DEFAULT_TOPICS labels (crypto, macro, regulation, ai, other).
        """
        cutoff_ts = (datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp()
        _topic_labels = {t.lower() for t in DEFAULT_TOPICS}

        with self._store._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
            ents_in_cluster = {str(e).strip().lower() for e in (c.get("entities", []) or []) if str(e).strip()}
            for kw in c.get("keywords", []):
                kw_str = str(kw).strip()
                if not kw_str:
                    continue
                if kw_str.lower() in _topic_labels:
                    continue
                if kw_str.lower() in ents_in_cluster:
                    continue
                counter[kw_str] = counter.get(kw_str, 0) + 1

        return [
            {"label": label, "count": count}
            for label, count in sorted(counter.items(), key=lambda x: -x[1])[:limit]
        ]