Spaces:
Running
Running
File size: 5,655 Bytes
a96145c a6515c4 a96145c a6515c4 a96145c 1a87329 8914f3e 1a87329 8914f3e a96145c a6515c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | import logging
from collections import Counter
from typing import Optional
from .models import NewsCluster, NewsItem
logger = logging.getLogger(__name__)
class NewsAggregator:
def __init__(self, config):
self.config = config
self._encoder = None
self._setup_encoder()
def _setup_encoder(self):
if not self.config.use_local_models:
logger.info("Local models disabled — using keyword similarity")
return
try:
from sentence_transformers import SentenceTransformer
model_name = f"sentence-transformers/{self.config.embedding_model}"
logger.info("Loading embedding model: %s ...", model_name)
self._encoder = SentenceTransformer(model_name)
except ImportError:
logger.warning("sentence-transformers not available — using keyword fallback")
except Exception as exc:
logger.warning("Embedding model failed: %s", exc)
def compute_similarity(self, a: str, b: str) -> float:
if self._encoder:
emb_a = self._encoder.encode(a, normalize_embeddings=True)
emb_b = self._encoder.encode(b, normalize_embeddings=True)
return float(emb_a @ emb_b)
return self._keyword_overlap(a, b)
@staticmethod
def _keyword_overlap(a: str, b: str) -> float:
words_a = set(a.lower().split())
words_b = set(b.lower().split())
if not words_a or not words_b:
return 0.0
common = words_a & words_b
return len(common) / max(len(words_a), len(words_b))
def cluster_news(self, items: list[NewsItem]) -> list[NewsCluster]:
clusters: list[list[NewsItem]] = []
for item in items:
if not item.analysis or not item.article:
continue
text = f"{item.article.title or item.post.title} {item.analysis.summary}"
placed = False
for cluster in clusters:
rep = cluster[0]
if not rep.analysis:
continue
rep_text = f"{rep.article.title or rep.post.title} {rep.analysis.summary}"
if self.compute_similarity(text, rep_text) >= self.config.similarity_threshold:
cluster.append(item)
placed = True
break
if not placed:
clusters.append([item])
return self._rank_clusters(clusters)
def _rank_clusters(self, raw: list[list[NewsItem]]) -> list[NewsCluster]:
scored = []
for group in raw:
if not group:
continue
topic = self._main_topic(group)
# Highest scoring post represents the cluster
best = max(group, key=lambda x: x.post.score)
avg_trust = sum(
it.analysis.trustworthiness_score for it in group if it.analysis
) / max(len(group), 1)
avg_pop = sum(it.post.score for it in group) / max(len(group), 1)
# Pick the first non-empty image
image_url = ""
for it in group:
src = it.article.image_url if it.article else ""
if src:
image_url = src
break
if it.post.image_url:
image_url = it.post.image_url
break
cluster_score = self._cluster_score(group, avg_trust)
cluster = NewsCluster(
topic=topic,
articles=group,
total_coverage=len(group),
avg_trustworthiness=avg_trust,
avg_popularity=avg_pop,
top_post_url=best.post.url,
final_score=cluster_score,
image_url=image_url,
)
scored.append(cluster)
return sorted(scored, key=lambda c: c.final_score, reverse=True)
def _cluster_score(self, group: list[NewsItem], avg_trust: float) -> float:
scores = []
for item in group:
s = avg_trust * 0.50
# Content quality: longer articles score higher
if item.article and item.article.text:
title_len = len(item.article.title or "")
text_len = len(item.article.text)
if text_len > title_len * 3:
s += 0.15
elif text_len > title_len * 1.5:
s += 0.08
# Successfully extracted vs title-only
if item.article and item.article.extraction_success:
s += 0.10
else:
s -= 0.10
# More topics = richer article
if item.analysis and item.analysis.topics:
s += min(len(item.analysis.topics) * 0.06, 0.18)
# Having a category means we actually understood it
if item.analysis and item.analysis.category != "General":
s += 0.05
# Penalty for sourcing from another outlet's reporting
if item.analysis:
s -= item.analysis.sourcing_penalty * 0.40
scores.append(max(0.05, min(1.0, s)))
return sum(scores) / max(len(scores), 1)
@staticmethod
def _main_topic(cluster: list[NewsItem]) -> str:
counter: Counter[str] = Counter()
for item in cluster:
if item.analysis:
for t in item.analysis.topics:
counter[t] += 1
if counter:
return counter.most_common(1)[0][0]
return (cluster[0].article.title or cluster[0].post.title or "")[:60]
|