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 sorted set of article keys. The topic is intentionally excluded from the hash: the same article may be classified under different topics across cycles (heuristic vs LLM-enriched), but it must always map to the same cluster_id so that ON CONFLICT DO UPDATE in upsert_clusters correctly merges them instead of creating duplicates.""" 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(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 # Keep the base cluster's original ID so the enrichment cache # (keyed by cluster_id) survives the merge. base.setdefault("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). # Re-derive the topic via the same heuristic (normalize_topic_from_title) # that new articles use, so that existing and new clusters with the same # headline land in the same by_topic bucket regardless of what LLM # enrichment previously stored on the cluster. if existing_clusters: for c in existing_clusters: if not _cluster_is_within_age_window(c, max_age_hours=max_age_hours): continue seed_title = c.get("headline") or "" topic = normalize_topic_from_title(seed_title) if seed_title else (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) # Assign stable IDs only to clusters that don't already have one. # Pre-seeded clusters from the DB carry their original cluster_id — # keeping it stable across cycles so the enrichment cache (keyed by # cluster_id) continues to work even after new articles are merged in. for c in clusters: if not c.get("cluster_id"): c["cluster_id"] = _stable_cluster_id(topic, c.get("articles", []) or []) by_topic[topic] = clusters # Cross-topic dedup: merge clusters with overlapping headlines and entities by_topic = _merge_duplicate_clusters(by_topic) # 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()} def _merge_duplicate_clusters( by_topic: Dict[str, List[Dict[str, Any]]], ) -> Dict[str, List[Dict[str, Any]]]: """Cross-topic dedup: merge clusters whose headlines and entities overlap. Catches the case where the same event arrives from different feeds with different article keys, lands in separate clusters with different stable IDs, but has nearly identical headlines and shared entities. Merge criteria: title_similarity >= 0.90 AND at least one shared entity. This is intentionally conservative to avoid merging distinct events. """ # Flatten all clusters with their topic all_clusters: list[tuple[str, dict]] = [] for topic, clusters in by_topic.items(): for c in clusters: all_clusters.append((topic, c)) n = len(all_clusters) if n <= 1: return by_topic # Union-Find parent = list(range(n)) 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 # Pre-extract normalized entity sets for each cluster cluster_ent_sets: list[set[str]] = [] cluster_heads: list[str] = [] for _, c in all_clusters: ents = {str(e).strip().lower() for e in (c.get("entities", []) or []) if str(e).strip()} cluster_ent_sets.append(ents) cluster_heads.append(str(c.get("headline", "") or "")) # Compare pairs — O(n^2) but n is small (clusters per cycle, not articles) TITLE_THRESHOLD = 0.90 for i in range(n): for j in range(i + 1, n): # Quick skip: if headlines are completely different, no need for entity check if _title_similarity(cluster_heads[i], cluster_heads[j]) < TITLE_THRESHOLD: continue # Check entity overlap (at least one shared entity) if not (cluster_ent_sets[i] & cluster_ent_sets[j]): continue union(i, j) # Group by component components: dict[int, list[int]] = {} for idx in range(n): root = find(idx) components.setdefault(root, []).append(idx) # Merge each component merged_by_topic: Dict[str, List[Dict[str, Any]]] = {} for root, members in components.items(): # Pick the base cluster (the one with the most sources, then most articles) best_idx = max(members, key=lambda i: ( len(all_clusters[i][1].get("sources", []) or []), len(all_clusters[i][1].get("articles", []) or []), )) base_topic, base = all_clusters[best_idx] if len(members) == 1: merged_by_topic.setdefault(base_topic, []).append(base) continue # Merge all clusters in this component into base all_articles: list[dict] = list(base.get("articles", []) or []) all_sources: list[str] = list(base.get("sources", []) or []) all_entities: list[str] = list(base.get("entities", []) or []) all_keywords: list[str] = list(base.get("keywords", []) or []) first_seen = base.get("first_seen", "") last_updated = base.get("last_updated", "") existing_article_keys = {_article_key(a) for a in all_articles} existing_ent_lower = {str(e).strip().lower() for e in all_entities} existing_kw_lower = {str(k).strip().lower() for k in all_keywords} for m_idx in members: if m_idx == best_idx: continue other = all_clusters[m_idx][1] # Merge articles (dedup by key) for a in other.get("articles", []) or []: ak = _article_key(a) if ak not in existing_article_keys: all_articles.append(a) existing_article_keys.add(ak) # Merge sources for s in other.get("sources", []) or []: if s not in all_sources: all_sources.append(s) # Merge entities (dedup case-insensitive) for e in other.get("entities", []) or []: el = str(e).strip().lower() if el not in existing_ent_lower: all_entities.append(e) existing_ent_lower.add(el) # Merge keywords (dedup case-insensitive) for k in other.get("keywords", []) or []: kl = str(k).strip().lower() if kl not in existing_kw_lower: all_keywords.append(k) existing_kw_lower.add(kl) # Timestamps 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["entities"] = all_entities base["keywords"] = all_keywords base["first_seen"] = first_seen base["last_updated"] = last_updated # Keep the base cluster's original ID so the enrichment cache # (keyed by cluster_id) survives the merge. base.setdefault("cluster_id", _stable_cluster_id(base.get("topic", "other"), all_articles)) merged_by_topic.setdefault(base_topic, []).append(base) return merged_by_topic