#!/usr/bin/env python3
"""Story layer for the news aggregator.
fetch -> dedup -> cluster -> rank -> [LLM top-k] -> out/index.html
Reads from one of:
* a FreshRSS export (JSON snapshot on disk), or
* the starter OPML directly over HTTP (no Docker needed for dev).
Usage:
python cluster.py # ingest -> out/index.html
python cluster.py --opml feeds.opml # fetch stub feeds directly (no FreshRSS)
python cluster.py --llm # LLM-summarize top-k via Hermes at :8080
python cluster.py --serve :8765 # serve out/index.html on port 8765
python cluster.py --topk 10 # cap LLM work to top-k stories
Env (--llm only):
LLM_BASE_URL default http://127.0.0.1:8080/v1 (host llama.cpp / Hermes)
LLM_MODEL default hermes
"""
from __future__ import annotations
import argparse
import hashlib
import html
import http.server
import json
import os
import random
import re
import sys
import time
import xml.etree.ElementTree as ET
from collections import Counter
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
try:
import httpx
except ImportError: # pragma: no cover
httpx = None
# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
@dataclass
class Item:
guid: str
url: str
title: str
summary: str
source: str
published: float # epoch seconds
@property
def normalized_title(self) -> str:
return re.sub(r"[^a-z0-9 ]+", " ", self.title.lower()).strip()
@dataclass
class Story:
id: int
title: str # canonical (longest) title among members
coverage_count: int
first_published: float
last_published: float
members: list = field(default_factory=list)
summary: str = ""
# ---------------------------------------------------------------------------
# Ingest
# ---------------------------------------------------------------------------
def parse_opml(path: str) -> list[str]:
root = ET.parse(path).getroot()
urls = []
for el in root.iter("outline"):
if el.attrib.get("type") == "rss" and el.attrib.get("xmlUrl"):
urls.append(el.attrib["xmlUrl"])
return urls
ATOM = "{http://www.w3.org/2005/Atom}"
def _atom(e, tag):
"""Fetch a namespaced Atom child by local name."""
return e.find(f".//{ATOM}{tag}")
def parse_rss_text(text: str, source: str) -> list[Item]:
items: list[Item] = []
root = ET.fromstring(text)
is_atom = root.tag == f"{ATOM}feed"
def pick(entry, rss_tag, atom_tag, get_text=True):
if is_atom:
el = _atom(entry, atom_tag)
if el is not None:
if el.tag == f"{ATOM}link":
return el.attrib.get("href")
return el.text
else:
el = entry.find(rss_tag)
if el is not None:
return el.text
return None
for entry in root.iter():
local = entry.tag.rsplit("}", 1)[-1]
if not is_atom and local != "item":
continue
if is_atom and local != "entry":
continue
title = pick(entry, "title", "title")
title = title or "(untitled)"
url = pick(entry, "link", "link")
guid = pick(entry, "guid", "id") or url or ""
summary = pick(entry, "description", "summary") or ""
if is_atom and str(summary).startswith("<"):
summary = "".join(_atom(entry, "summary").itertext()) if _atom(entry, "summary") is not None else ""
if not str(title).strip():
continue
published = _pub_to_ts(pick(entry, "pubDate", "updated") or "")
items.append(Item(guid=str(guid), url=str(url or guid), title=html.unescape(str(title)).strip(),
summary=html.unescape(str(summary)).strip(), source=source,
published=published))
return items
def _pub_to_ts(s: str) -> float:
s = s.strip()
for fmt in ("%a, %d %b %Y %H:%M:%S %z", "%a, %d %b %Y %H:%M:%S %Z",
"%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%dT%H:%M:%SZ"):
try:
return datetime.strptime(s, fmt).timestamp()
except ValueError:
continue
try:
return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
except ValueError:
return time.time()
def fetch_opml(opml: str) -> list[Item]:
"""Fetch every feed in the OPML directly. Tolerant of individual failures."""
if httpx is None:
raise SystemExit("httpx not installed โ run: pip install -r requirements.txt")
all_items: list[Item] = []
with httpx.Client(timeout=25.0, follow_redirects=True, headers={"User-Agent": UA}) as c:
for url in parse_opml(opml):
try:
r = c.get(url)
r.raise_for_status()
got = parse_rss_text(r.text, url)
print(f" {len(got):4d} {url}")
all_items.extend(got)
except Exception as e: # noqa: BLE001
print(f" FAIL {url}: {e}")
return all_items
def load_snapshot(path: str) -> list[Item]:
"""Load a FreshRSS export JSON: [{"url","title","summary","source","published"}, ...]"""
raw = json.loads(Path(path).read_text(encoding="utf-8"))
items = []
for i, r in enumerate(raw):
items.append(Item(guid=str(r.get("guid", i)), url=r.get("url", ""),
title=r.get("title", ""), summary=r.get("summary", ""),
source=r.get("source", ""), published=float(r.get("published", time.time()))))
return items
# ---------------------------------------------------------------------------
# Dedup + clustering (MinHash + LSH -> connected components)
# ---------------------------------------------------------------------------
def _word_shingles(text: str, k: int = 5):
words = text.split()
if len(words) < k:
return set(words) if words else set()
return set(" ".join(words[i:i + k]) for i in range(len(words) - k + 1))
def _minhash_signature(shingles: set, perms):
sig = []
for a, b in perms:
m = None
for s in shingles:
h = (a * hash(s) + b) % 2**31
m = h if m is None else min(m, h)
sig.append(m if m is not None else 0)
return sig
class Clusterer:
def __init__(self, num_perm: int = 128, rows: int = 8, k: int = 4,
sim_threshold: float = 0.40):
# LSH bands: num_perm rows per band -> number of bands = num_perm // rows
# k = word-shingle width over title + body (bodies carry the signal in
# syndicated rewrites; titles routinely diverge).
self.num_perm = num_perm
self.rows = rows
self.k = k
self.sim_threshold = sim_threshold
random.seed(7)
self.perms = [(random.randint(1, 2**31 - 1), random.randint(1, 2**31 - 1))
for _ in range(num_perm)]
def signature(self, item: Item) -> list:
text = f"{item.normalized_title} {re.sub(r'[^a-z0-9 ]+', ' ', item.summary.lower())}"
return _minhash_signature(_word_shingles(text, self.k), self.perms) or [0] * self.num_perm
@staticmethod
def jaccard(s1: list, s2: list) -> float:
eq = sum(1 for a, b in zip(s1, s2) if a == b)
return eq / len(s1)
def _bands(self, sig: list):
bands = []
for i in range(0, self.num_perm, self.rows):
bands.append(tuple(sig[i:i + self.rows]))
return bands
def cluster(self, items: list[Item]) -> list[Story]:
# --- dedup fast path: exact normalized title OR same canonical url ---
seen_title: dict[str, int] = {}
seen_url: dict[str, int] = {}
groups: list[list[Item]] = []
for it in items:
key = it.normalized_title
matched = seen_title.get(key) or seen_url.get(it.url.strip().rstrip("/"))
if matched is not None:
groups[matched].append(it)
continue
groups.append([it])
seen_title[key] = len(groups) - 1
seen_url.setdefault(it.url.strip().rstrip("/"), len(groups) - 1)
# --- MinHash signatures: representative = longest member of each group ---
sigs = [self.signature(max(g, key=lambda it: len(it.title))) for g in groups]
union = [-1] * len(groups)
def find(x):
while union[x] != -1:
x = union[x]
return x
def merge(a, b):
ra, rb = find(a), find(b)
if ra != rb:
union[ra] = rb
n = len(groups)
if n <= 2000:
# Prototype scale: exact pairwise over MinHash signatures. Guarantees
# the merge for near-dups that LSH banding would miss (band all-equal
# probability collapses as jaccard drops toward the threshold).
for i in range(n):
for j in range(i + 1, n):
if self.jaccard(sigs[i], sigs[j]) >= self.sim_threshold:
merge(i, j)
else:
# Large corpus: LSH bands prune candidates, then confirm by jaccard.
buckets: dict[tuple, list[int]] = {}
for idx, sig in enumerate(sigs):
for band in self._bands(sig):
buckets.setdefault(band, []).append(idx)
for band, idxs in buckets.items():
if len(idxs) < 2:
continue
for i in range(len(idxs)):
for j in range(i + 1, len(idxs)):
a, b = idxs[i], idxs[j]
if self.jaccard(sigs[a], sigs[b]) >= self.sim_threshold:
merge(a, b)
clusters: dict[int, list] = {}
for gi in range(len(groups)):
clusters.setdefault(find(gi), []).extend(groups[gi])
return [self._make_story(i, members) for i, members in
enumerate(clusters.values())]
@staticmethod
def _make_story(i, members: list[Item]) -> Story:
members.sort(key=lambda it: it.published)
longest = max(members, key=lambda it: len(it.title))
return Story(
id=i,
title=longest.title,
coverage_count=len(members),
first_published=members[0].published,
last_published=members[-1].published,
members=members,
summary="",
)
# ---------------------------------------------------------------------------
# Rank
# ---------------------------------------------------------------------------
def compute_scores(stories: list[Story], now: float, alpha: float = 1.8) -> list:
"""score = authority * (coverage_count + 1) / (age_h + 2)^1.8 (HN-style decay)"""
scored = []
for s in stories:
age_h = max(0.0, (now - s.first_published) / 3600.0)
rank = 1.0 * (s.coverage_count + 1) / (age_h + 2.0) ** alpha
scored.append((rank, s))
scored.sort(key=lambda x: -x[0])
return scored
# ---------------------------------------------------------------------------
# LLM summary of top-k only (capped; host llama.cpp/Hermes, never DeepSeek)
# ---------------------------------------------------------------------------
def llm_summarize_topk(scored, topk: int) -> None:
base = os.environ.get("LLM_BASE_URL", "http://127.0.0.1:8080/v1").rstrip("/")
model = os.environ.get("LLM_MODEL", "hermes")
if httpx is None:
print(" (httpx missing; skipping LLM)")
return
for rank, story in scored[:topk]:
lead = story.members[0].url or story.members[0].summary
text = f"{story.title}\n\n{story.members[0].summary[:300]}"
prompt = ("Summarize this news story cluster in 2 punchy sentences. "
f"({story.coverage_count} sources). Story:\n{text}")
try:
with httpx.Client(timeout=120.0) as c:
r = c.post(f"{base}/chat/completions", json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 120,
})
r.raise_for_status()
story.summary = r.json()["choices"][0]["message"]["content"].strip()
print(f" [llm] {story.title[:50]}...")
except Exception as e: # noqa: BLE001
print(f" [llm FAIL] {story.title[:40]}: {e}")
# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------
def _fmt(ts: float) -> str:
return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%b %d %H:%M")
def render_html(stories: list, out_dir: Path) -> None:
rows = []
for rank, s in enumerate(stories, 1):
mems = "".join(
f'