news-mcp/news_mcp/dedup/cluster.py

from __future__ import annotations

import asyncio
import hashlib
import re
from datetime import datetime, timezone, timedelta
from difflib import SequenceMatcher
from typing import Any, Dict, List
from urllib.parse import urlparse

from news_mcp.config import (
    NEWS_EMBEDDINGS_ENABLED,
    NEWS_EMBEDDING_SIMILARITY_THRESHOLD,
    NEWS_CLUSTER_MAX_AGE_HOURS,
)
from news_mcp.dedup.embedding_support import cosine_similarity, ollama_embed
from news_mcp.sources.news_feeds import normalize_topic_from_title


# ---------------------------------------------------------------------------
# Text helpers
# ---------------------------------------------------------------------------


def _normalize_title(title: str) -> str:
    t = title.lower().strip()
    t = re.sub(r"[^a-z0-9\s]", " ", t)
    t = re.sub(r"\s+", " ", t).strip()
    return t


def _title_similarity(a: str, b: str) -> float:
    return SequenceMatcher(None, _normalize_title(a), _normalize_title(b)).ratio()


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 _cluster_text(a: Dict[str, Any]) -> str:
    parts = [a.get("title", ""), a.get("summary", "") or ""]
    return "\n".join(p for p in parts if p).strip()


# ---------------------------------------------------------------------------
# Token / Jaccard signal
# ---------------------------------------------------------------------------

_STOPWORDS = frozenset(
    {
        "a", "an", "the", "of", "to", "in", "on", "at", "for", "by", "with",
        "and", "or", "but", "if", "is", "are", "was", "were", "be", "been",
        "being", "as", "from", "that", "this", "these", "those", "it", "its",
        "into", "over", "under", "than", "then", "so", "such", "no", "not",
        "do", "does", "did", "will", "would", "can", "could", "should", "may",
        "might", "has", "have", "had", "after", "before", "amid", "vs", "via",
        "us", "uk",
    }
)


def _tokens(text: str) -> set[str]:
    tokens = re.findall(r"[a-z0-9][a-z0-9\-]+", text.lower())
    return {t for t in tokens if len(t) >= 3 and t not in _STOPWORDS}


def _jaccard(a: set, b: set) -> float:
    if not a or not b:
        return 0.0
    inter = len(a & b)
    if inter == 0:
        return 0.0
    return inter / len(a | b)


# ---------------------------------------------------------------------------
# Composite similarity
# ---------------------------------------------------------------------------

DEFAULT_TITLE_THRESHOLD = 0.87
DEFAULT_JACCARD_THRESHOLD = 0.55


def _signals(article: Dict[str, Any], cluster: Dict[str, Any]) -> dict:
    """Per-pair similarity signals (title, jaccard, embedding cosine).

    Compares the article against ALL articles in the cluster and returns the
    best (max) signal across all comparisons.  The cosine signal uses the
    cluster-level embedding; title and jaccard are computed per-article and
    the maximum is returned so that a match against any cluster member counts.
    """
    a_title = str(article.get("title") or "")
    c_title = str(cluster.get("headline") or "")
    a_emb = article.get("_embedding")
    c_emb = cluster.get("embedding")
    cosine = cosine_similarity(a_emb, c_emb) if a_emb and c_emb else 0.0

    best_title = 0.0
    best_jaccard = 0.0
    a_text = _cluster_text(article)
    a_toks = _tokens(a_text) if a_text else set()

    # Compare against every article in the cluster, take the best scores.
    cluster_articles = cluster.get("articles") or ([{"title": c_title}] if c_title else [])
    for ca in cluster_articles:
        if not isinstance(ca, dict):
            continue
        # title signal
        ca_title = str(ca.get("title") or "")
        if a_title and ca_title:
            t = _title_similarity(a_title, ca_title)
            if t > best_title:
                best_title = t
        # jaccard signal
        ca_text = _cluster_text(ca)
        if a_text and ca_text:
            j = _jaccard(a_toks, _tokens(ca_text))
            if j > best_jaccard:
                best_jaccard = j
        # early exit: if both title and jaccard are already very high
        if best_title >= 0.95 and best_jaccard >= 0.80:
            break

    return {"title": best_title, "jaccard": best_jaccard, "cosine": cosine}


def _is_match(
    signals: dict,
    *,
    embeddings_enabled: bool,
    title_threshold: float = DEFAULT_TITLE_THRESHOLD,
    jaccard_threshold: float = DEFAULT_JACCARD_THRESHOLD,
) -> tuple[bool, str, float]:
    """Decide whether two items should merge based on the strongest signal.

    Cascade: cosine (if embeddings enabled) → title → jaccard → consensus.
    Returns (matched, signal_name, signal_value).
    """
    cosine_threshold = NEWS_EMBEDDING_SIMILARITY_THRESHOLD
    if embeddings_enabled and signals["cosine"] >= cosine_threshold:
        return True, "cosine", signals["cosine"]
    if signals["title"] >= title_threshold:
        return True, "title", signals["title"]
    if signals["jaccard"] >= jaccard_threshold:
        return True, "jaccard", signals["jaccard"]
    if (
        embeddings_enabled
        and signals["cosine"] >= 0.80
        and (signals["jaccard"] >= 0.30 or signals["title"] >= 0.55)
    ):
        val = (signals["cosine"] + max(signals["jaccard"], signals["title"])) / 2.0
        return True, "consensus", val
    return False, "none", 0.0


# ---------------------------------------------------------------------------
# Stable cluster ID
# ---------------------------------------------------------------------------

def _stable_cluster_id(topic: str, articles: List[Dict[str, Any]]) -> str:
    """Deterministic cluster ID derived from the topic and the sorted set of
    article keys.  Using the minimum key (lexicographic) as the seed ensures
    that no matter which article arrives first, the same set of articles always
    maps to the same cluster_id."""
    keys = sorted(_article_key(a) for a in articles if _article_key(a))
    if not keys:
        # Degenerate fallback — single article with empty url and title
        return hashlib.sha1(topic.encode("utf-8")).hexdigest()
    seed = keys[0]
    return hashlib.sha1(f"{topic}|{seed}".encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Temporal gating
# ---------------------------------------------------------------------------

def _parse_ts(ts_str: str) -> datetime | None:
    if not ts_str:
        return None
    try:
        s = str(ts_str).replace("Z", "+00:00")
        dt = datetime.fromisoformat(s)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc)
    except Exception:
        pass
    try:
        from email.utils import parsedate_to_datetime
        dt = parsedate_to_datetime(str(ts_str))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc)
    except Exception:
        return None


def _cluster_is_within_age_window(cluster: Dict[str, Any], *, max_age_hours: float) -> bool:
    """Return True if the cluster's last_updated is within the merge window."""
    if max_age_hours <= 0:
        return True  # 0 = no limit
    ts_str = cluster.get("last_updated") or cluster.get("timestamp") or ""
    dt = _parse_ts(ts_str)
    if dt is None:
        return True  # be lenient with unparseable timestamps
    cutoff = datetime.now(timezone.utc) - timedelta(hours=max_age_hours)
    return dt >= cutoff


# ---------------------------------------------------------------------------
# Embedding pre-computation (async internally)
# ---------------------------------------------------------------------------


async def _compute_embeddings_concurrently(
    articles: List[Dict[str, Any]],
) -> Dict[str, list[float] | None]:
    """Compute embeddings for unique article texts concurrently.

    Returns a cache dict: text -> embedding vector or None.
    """
    unique_texts: list[str] = []
    seen: set[str] = set()
    for a in articles:
        text = _cluster_text(a)
        if text and text not in seen:
            seen.add(text)
            unique_texts.append(text)

    emb_tasks = [ollama_embed(text) for text in unique_texts]
    emb_results = await asyncio.gather(*emb_tasks, return_exceptions=True)

    cache: Dict[str, list[float] | None] = {}
    for text, result in zip(unique_texts, emb_results):
        if isinstance(result, list):
            cache[text] = result
        else:
            cache[text] = None
    return cache


def _compute_embeddings_sync(
    articles: List[Dict[str, Any]],
) -> Dict[str, list[float] | None]:
    """Synchronous wrapper that runs the async embedding computation.

    Handles three cases:
    1. Already inside an async event loop (called from poller) -> schedule
       as a task and run it to completion on the running loop.
    2. No event loop at all (plain sync caller) -> use asyncio.run().
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # No running loop — safe to use asyncio.run()
        return asyncio.run(_compute_embeddings_concurrently(articles))

    # We're inside a running event loop (e.g. the poller). Create a new loop
    # in a thread to avoid blocking.
    import concurrent.futures
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(
            asyncio.run, _compute_embeddings_concurrently(articles)
        )
        return future.result()


# ---------------------------------------------------------------------------
# Orphan merge: detect clusters sharing articles and merge them
# ---------------------------------------------------------------------------

def _merge_orphan_clusters(
    clusters: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
    """Post-clustering pass: merge clusters that share article keys.

    This handles the case where two articles about the same event didn't match
    during the main loop (e.g. embeddings were temporarily unavailable) and
    ended up in separate clusters.  If two clusters share >= 1 article key, we
    merge them into one (keeping the earlier first_seen, recompute the stable
    ID from the union of articles).
    """
    if len(clusters) <= 1:
        return clusters

    # Build index: article_key -> list of cluster indices
    key_to_indices: dict[str, list[int]] = {}
    for idx, c in enumerate(clusters):
        for a in c.get("articles", []) or []:
            ak = _article_key(a)
            if ak:
                key_to_indices.setdefault(ak, []).append(idx)

    # Find connected components via Union-Find
    parent = list(range(len(clusters)))

    def find(x: int) -> int:
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    def union(a: int, b: int) -> None:
        ra, rb = find(a), find(b)
        if ra != rb:
            parent[ra] = rb

    for indices in key_to_indices.values():
        for i in range(1, len(indices)):
            union(indices[0], indices[i])

    # Group clusters by component
    components: dict[int, list[int]] = {}
    for idx in range(len(clusters)):
        root = find(idx)
        components.setdefault(root, []).append(idx)

    merged: List[Dict[str, Any]] = []
    for root, members in components.items():
        if len(members) == 1:
            merged.append(clusters[members[0]])
            continue

        # Merge all clusters in this component
        base = dict(clusters[members[0]])
        all_articles: list[dict] = list(base.get("articles", []) or [])
        all_sources: list[str] = list(base.get("sources", []) or [])
        first_seen = base.get("first_seen", "")
        last_updated = base.get("last_updated", "")

        for m_idx in members[1:]:
            other = clusters[m_idx]
            existing_keys = {_article_key(a) for a in all_articles}
            for a in other.get("articles", []) or []:
                ak = _article_key(a)
                if ak not in existing_keys:
                    all_articles.append(a)
                    existing_keys.add(ak)
            for s in other.get("sources", []) or []:
                if s not in all_sources:
                    all_sources.append(s)
            fs = other.get("first_seen", "")
            if fs and (not first_seen or fs < first_seen):
                first_seen = fs
            lu = other.get("last_updated", "")
            if lu and (not last_updated or lu > last_updated):
                last_updated = lu

        base["articles"] = all_articles
        base["sources"] = all_sources
        base["first_seen"] = first_seen
        base["last_updated"] = last_updated
        base["cluster_id"] = _stable_cluster_id(base.get("topic", "other"), all_articles)
        merged.append(base)

    return merged


# ---------------------------------------------------------------------------
# Public API (sync — backward compatible with tests)
# ---------------------------------------------------------------------------


def dedup_and_cluster_articles(
    articles: List[Dict[str, Any]],
    similarity_threshold: float | None = None,
    *,
    existing_clusters: List[Dict[str, Any]] | None = None,
    max_age_hours: float = 0,
) -> Dict[str, List[Dict[str, Any]]]:
    """Deduplicate raw articles into clusters keyed by topic.

    v1.3: stable cluster IDs, temporal gating, and orphan merge.

    Args:
        articles: new articles to cluster.
        similarity_threshold: override for the title-similarity threshold.
        existing_clusters: optional list of recent clusters from the DB to
            merge against (cross-cycle merge).  When provided, temporal
            gating via max_age_hours is applied to filter these.
        max_age_hours: only compare against existing_clusters updated within
            this many hours.  0 = no limit (compare against all provided).
    """

    title_threshold = similarity_threshold if similarity_threshold is not None else DEFAULT_TITLE_THRESHOLD

    # Pre-compute embeddings concurrently (sync boundary handles async internally)
    embedding_cache: Dict[str, list[float] | None] = {}
    if NEWS_EMBEDDINGS_ENABLED:
        embedding_cache = _compute_embeddings_sync(articles)

    by_topic: Dict[str, List[Dict[str, Any]]] = {}

    # Seed with existing clusters (filtered by age window)
    if existing_clusters:
        for c in existing_clusters:
            if not _cluster_is_within_age_window(c, max_age_hours=max_age_hours):
                continue
            topic = c.get("topic", "other") or "other"
            by_topic.setdefault(topic, []).append(dict(c))

    for a in articles:
        title = a.get("title") or ""
        if not title:
            continue
        topic = normalize_topic_from_title(title)
        article_text = _cluster_text(a)

        article_embedding = embedding_cache.get(article_text) if NEWS_EMBEDDINGS_ENABLED else None
        a_with_emb = dict(a)
        if article_embedding is not None:
            a_with_emb["_embedding"] = article_embedding

        by_topic.setdefault(topic, [])
        clusters = by_topic[topic]

        best_idx: int | None = None
        best_signal_name = "none"
        best_signal_value = 0.0
        for idx, c in enumerate(clusters):
            sigs = _signals(a_with_emb, c)
            matched, signal_name, signal_value = _is_match(
                sigs,
                embeddings_enabled=NEWS_EMBEDDINGS_ENABLED,
                title_threshold=title_threshold,
            )
            if matched and signal_value > best_signal_value:
                best_idx = idx
                best_signal_name = signal_name
                best_signal_value = signal_value

        if best_idx is not None:
            c = clusters[best_idx]
            existing_keys = {_article_key(x) for x in c.get("articles", []) or []}
            if _article_key(a) not in existing_keys:
                c["articles"].append(a)
            if a.get("source") and a["source"] not in c["sources"]:
                c["sources"].append(a["source"])
            c["last_updated"] = max(str(c.get("last_updated", "")), str(a.get("timestamp", "")))
            # Update cluster embedding to the new article's embedding so later
            # comparisons can match against the most recently added content.
            if NEWS_EMBEDDINGS_ENABLED and article_embedding is not None:
                c["embedding"] = article_embedding
                c["embedding_model"] = "ollama:nomic-embed-text"
            c.setdefault("_merge_signals", []).append(
                {"signal": best_signal_name, "value": round(best_signal_value, 3)}
            )
        else:
            cid = _stable_cluster_id(topic, [a])
            cluster_embedding = article_embedding if NEWS_EMBEDDINGS_ENABLED else None
            clusters.append(
                {
                    "cluster_id": cid,
                    "headline": title,
                    "summary": a.get("summary", ""),
                    "topic": topic,
                    "entities": [],
                    "sentiment": "neutral",
                    "importance": 0.0,
                    "sources": [a["source"]] if a.get("source") else [],
                    "timestamp": a.get("timestamp"),
                    "articles": [a],
                    "first_seen": a.get("timestamp"),
                    "last_updated": a.get("timestamp"),
                    "embedding": cluster_embedding,
                    "embedding_model": "ollama:nomic-embed-text" if cluster_embedding else None,
                }
            )

    # Post-clustering passes per topic
    for topic, clusters in by_topic.items():
        # Merge orphans (clusters that share articles)
        clusters = _merge_orphan_clusters(clusters)
        # Recompute stable IDs from the final article sets
        for c in clusters:
            c["cluster_id"] = _stable_cluster_id(topic, c.get("articles", []) or [])
        by_topic[topic] = clusters

    # Strip the internal merge audit trail before returning
    for clusters in by_topic.values():
        for c in clusters:
            c.pop("_merge_signals", None)

    return {topic: clusters for topic, clusters in by_topic.items()}