diff --git a/prototype/cluster.py b/prototype/cluster.py new file mode 100644 index 0000000..0bc7517 --- /dev/null +++ b/prototype/cluster.py @@ -0,0 +1,464 @@ +#!/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'
  • {html.escape(m.source)}
    • ' + for m in s.members) + summ = f"

    {html.escape(s.summary)}

    " if s.summary else "" + rows.append(f""" +
    +
    {rank} +

    {html.escape(s.title)}

    +
    {s.coverage_count} source{'s' if s.coverage_count>1 else ''} ยท {_fmt(s.first_published)}
    + {summ} + +
    """) + body = "\n".join(rows) or "

    No stories.

    " + html_doc = f""" +News Aggregator โ€” stories + +

    News Aggregator โ€” story clusters

    +
    {len(stories)} stories
    +{body} +""" + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "index.html").write_text(html_doc, encoding="utf-8") + (out_dir / "stories.json").write_text(json.dumps([ + {"rank": r, "title": s.title, "coverage": s.coverage_count, + "first_published": s.first_published, "summary": s.summary, + "members": [{"url": m.url, "source": m.source} for m in s.members]} + for r, s in enumerate(stories, 1)], indent=2), encoding="utf-8") + + +def serve(port: int = 8765) -> None: + out_dir = Path(__file__).parent / "out" + os.chdir(out_dir) + + class H(http.server.SimpleHTTPRequestHandler): + def log_message(self, *a): # keep the console clean + pass + + print(f"serving {out_dir} on http://localhost:{port}") + http.server.ThreadingHTTPServer(("0.0.0.0", port), H).serve_forever() + + +# --------------------------------------------------------------------------- +# main +# --------------------------------------------------------------------------- + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="news aggregator story layer") + p.add_argument("--opml", default=str(Path(__file__).parent / "feeds.opml"), + help="OPML of feeds to fetch directly (skip Docker/FreshRSS)") + p.add_argument("--snapshot", help="JSON snapshot of FreshRSS items instead of live fetch") + p.add_argument("--llm", action="store_true", help="LLM-summarize top-k via LLM_BASE_URL") + p.add_argument("--topk", type=int, default=10, help="cap on # stories summarized") + p.add_argument("--serve", nargs="?", const="8765", type=str, metavar="PORT", + help="serve out/ on the given port (default 8765). Accepts " + "'--serve 8765' or '--serve :8765'") + p.add_argument("--no-fetch", action="store_true", help="skip live fetch (offline demo)") + args = p.parse_args(argv) + + out_dir = Path(__file__).parent / "out" + + if args.snapshot: + items = load_snapshot(args.snapshot) + src = f"snapshot {args.snapshot}" + elif args.no_fetch: + items = [] + src = "offline" + else: + print(f"fetching OPML: {args.opml}") + items = fetch_opml(args.opml) + src = args.opml + + # exact dedup by url+title handled inside cluster() + print(f" {len(items)} items parsed from {src}") + stories = Clusterer().cluster(items) + now = time.time() + scored = compute_scores(stories, now) + stories_sorted = [s for _, s in scored] + print(f" {len(stories)} story clusters (was {len(items)} items)") + + if args.llm and stories_sorted: + print(f" summarizing top-{min(args.topk, len(stories_sorted))} with LLM...") + llm_summarize_topk(scored, args.topk) + + render_html(stories_sorted, out_dir) + print(f" wrote {out_dir / 'index.html'}") + + if args.serve is not None: + port = int(re.sub(r"[^0-9]", "", args.serve) or "8765") + serve(port) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())