diff --git a/news/cli.py b/news/cli.py index 714eba0..9788a02 100644 --- a/news/cli.py +++ b/news/cli.py @@ -51,7 +51,10 @@ def cmd_clusters(args) -> int: ).fetchall() for r in rows: buckets = cluster_mod.bucket_count(con, r["id"]) - print(f"\n[{r['size']:>3} items, {buckets} independent bucket(s)] {r['lead']}") + origins = cluster_mod.origin_count( + con, r["id"], Config().clustering["syndication_threshold"]) + synd = f", {buckets - origins} syndicated" if origins < buckets else "" + print(f"\n[{r['size']:>3} items, {origins} independent origin(s){synd}] {r['lead']}") if args.verbose: for m in con.execute( """SELECT i.title, s.name FROM cluster_members cm diff --git a/news/cluster.py b/news/cluster.py index 4f4098e..adca632 100644 --- a/news/cluster.py +++ b/news/cluster.py @@ -22,6 +22,7 @@ feeds importance, and it merges unrelated stories into one notification. from __future__ import annotations +import re import sqlite3 from dataclasses import dataclass from datetime import datetime, timedelta @@ -33,6 +34,11 @@ from .util import stable_id THRESHOLD = 0.82 WINDOW_HOURS = 48 +# Above this cosine, two items from different newsrooms are the same syndicated +# copy rather than independent reporting. High on purpose: independently written +# headlines about one event sit far below it (the eight Renoir-heist reports all +# stayed separate), while wire reprints are effectively identical text. +SYNDICATION_THRESHOLD = 0.97 @dataclass(frozen=True) @@ -142,12 +148,85 @@ def assign( return out -def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int: - """Distinct editorial-independence buckets in a cluster. +_PUNCT = re.compile(r"[^a-z0-9 ]") - This, not raw member count, is the corroboration signal. Ten outlets - reprinting one wire story is one bucket; the old system counted it as ten - and then fired a priority-5 push on it. + +def _normalise_title(title: str) -> str: + return " ".join(_PUNCT.sub("", title.lower()).split()) + + +def origin_count( + con: sqlite3.Connection, + cluster_id: str, + syndication_threshold: float = SYNDICATION_THRESHOLD, +) -> int: + """Number of independent newsrooms that originated reporting in a cluster. + + This, not member count and not bucket count, is the corroboration signal. + Two items share an origin when EITHER holds: + + 1. Same `bucket` -- the same newsroom, however many feeds or distinct + stories it filed. Catches NYT running five different AfD pieces. + 2. Near-identical text -- syndicated copy, whoever published it. Catches a + local station reprinting a wire dispatch, which no static bucket can + express: WCPO is genuinely independent on Cincinnati stories and a pure + reprinter on national ones. + + Those two rules are complementary and both are needed: (1) is same-owner / + different-text, (2) is different-owner / same-text. Merges are transitive. + + Note this is where a lexical comparison IS the right tool, unlike + clustering. Paraphrases of one story share almost no wording, which is why + the SimHash stage was measured and removed -- but syndicated copy is + literally the same bytes. + """ + rows = con.execute( + """SELECT i.id, i.title, i.content_hash, s.bucket + FROM cluster_members cm + JOIN items i ON i.id = cm.item_id + JOIN sources s ON s.id = i.source_id + WHERE cm.cluster_id = ?""", + (cluster_id,), + ).fetchall() + if not rows: + return 0 + + vectors: dict[str, np.ndarray] = {} + for row in rows: + got = con.execute( + "SELECT vec FROM embeddings WHERE content_hash=?", (row["content_hash"],) + ).fetchone() + if got is not None: + vectors[row["content_hash"]] = store.from_blob(got["vec"]) + + parent = list(range(len(rows))) + + def find(a: int) -> int: + while parent[a] != a: + parent[a] = parent[parent[a]] + a = parent[a] + return a + + titles = [_normalise_title(r["title"]) for r in rows] + for i in range(len(rows)): + for j in range(i + 1, len(rows)): + same = rows[i]["bucket"] == rows[j]["bucket"] or titles[i] == titles[j] + if not same: + vi = vectors.get(rows[i]["content_hash"]) + vj = vectors.get(rows[j]["content_hash"]) + if vi is not None and vj is not None: + same = float(vi @ vj) >= syndication_threshold + if same: + parent[find(i)] = find(j) + + return len({find(i) for i in range(len(rows))}) + + +def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int: + """Distinct configured buckets, ignoring per-story syndication. + + Kept for diagnostics: comparing it with `origin_count` shows how much + syndication a cluster contains. Routing should use `origin_count`. """ row = con.execute( """SELECT COUNT(DISTINCT s.bucket) AS n diff --git a/news/config.py b/news/config.py index ed15980..ca8c307 100644 --- a/news/config.py +++ b/news/config.py @@ -82,7 +82,11 @@ class Config: @property def clustering(self) -> dict[str, Any]: c = self.section("clustering") - return {"threshold": c.get("threshold", 0.82), "window_hours": c.get("window_hours", 48)} + return { + "threshold": c.get("threshold", 0.82), + "window_hours": c.get("window_hours", 48), + "syndication_threshold": c.get("syndication_threshold", 0.97), + } @property def scoring(self) -> dict[str, Any]: diff --git a/news/pipeline.py b/news/pipeline.py index af0d3be..b046d75 100644 --- a/news/pipeline.py +++ b/news/pipeline.py @@ -129,7 +129,8 @@ def run_cycle(con: sqlite3.Connection, cfg: Config, embedder: Embedder) -> Cycle clustering = cfg.clustering assignments = cluster.assign( con, list(pending), vectors, - threshold=clustering["threshold"], window_hours=clustering["window_hours"]) + threshold=clustering["threshold"], + window_hours=clustering["window_hours"]) report.clustered = len(assignments) report.new_clusters = sum(1 for a in assignments if a.created) return report diff --git a/profile/config.toml b/profile/config.toml index 993a6a0..0d0c284 100644 --- a/profile/config.toml +++ b/profile/config.toml @@ -40,6 +40,12 @@ dim = 384 threshold = 0.82 window_hours = 48 +# Above this cosine, two items from DIFFERENT newsrooms are treated as the same +# syndicated copy rather than independent corroboration. Deliberately high: +# eight outlets each writing their own Renoir-heist headline stayed separate, +# while a local station reprinting a wire dispatch collapsed correctly. +syndication_threshold = 0.97 + [scoring] # Compare blocks by z-score WITHIN each block, never by raw cosine: cosine # magnitude is not comparable across blocks, and a plain max let one block win diff --git a/tests/test_cluster.py b/tests/test_cluster.py index 5b59831..30315aa 100644 --- a/tests/test_cluster.py +++ b/tests/test_cluster.py @@ -96,3 +96,60 @@ def test_syndicated_partners_in_one_bucket_count_once(con): assert out[0].cluster_id == out[1].cluster_id assert cluster.bucket_count(con, out[0].cluster_id) == 1, ( "syndicated partners must not read as independent confirmation") + + +def _cluster_of(con, items_spec, vec): + """Put every item in one cluster and return its id.""" + rows = [add(con, iid, sid, when, chash) for iid, sid, when, chash in items_spec] + out = cluster.assign(con, rows, {c: vec for *_, c in items_spec}) + return out[0].cluster_id + + +def test_origin_count_collapses_syndicated_copy_across_newsrooms(con): + """Different owners, identical text: a wire reprint is not corroboration. + + This is the case no static bucket can express -- a local station is + independent on local stories and a pure reprinter on national ones. + """ + v = unit(1, 0, 0) + con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)", + ("hw", 3, store.to_blob(v))) + rows = [add(con, "w1", "s1", "2026-01-01T00:00:00+00:00", "hw"), + add(con, "w2", "s2", "2026-01-01T00:20:00+00:00", "hw")] + # identical headline text across two different buckets + con.execute("UPDATE items SET title='Wire dispatch: minister resigns' WHERE id IN ('w1','w2')") + rows = con.execute("SELECT * FROM items WHERE id IN ('w1','w2') ORDER BY published_at").fetchall() + cid = cluster.assign(con, list(rows), {"hw": v})[0].cluster_id + assert cluster.bucket_count(con, cid) == 2 + assert cluster.origin_count(con, cid) == 1 + + +def test_origin_count_keeps_independently_written_reports_separate(con): + """Same event, different newsrooms, different wording: real corroboration.""" + # cosine ~0.90: above the 0.82 clustering threshold so they land in one + # cluster, below the 0.97 syndication threshold so they stay distinct + # origins. The first draft of this test used ~0.999 vectors and correctly + # merged them -- the test data was wrong, not the rule. + a, b = unit(1, 0, 0), unit(0.90, 0.436, 0) + for h, v in (("ha", a), ("hb", b)): + con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)", + (h, 3, store.to_blob(v))) + add(con, "r1", "s1", "2026-01-01T00:00:00+00:00", "ha") + add(con, "r2", "s2", "2026-01-01T00:30:00+00:00", "hb") + con.execute("UPDATE items SET title='Thieves steal two paintings' WHERE id='r1'") + con.execute("UPDATE items SET title='Renoir works taken in museum heist' WHERE id='r2'") + rows = con.execute("SELECT * FROM items WHERE id IN ('r1','r2') ORDER BY published_at").fetchall() + cid = cluster.assign(con, list(rows), {"ha": a, "hb": b})[0].cluster_id + assert cluster.origin_count(con, cid) == 2 + + +def test_origin_count_never_exceeds_bucket_count(con): + """Syndication detection may only ever reduce corroboration.""" + v = unit(1, 0, 0) + con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)", + ("hx", 3, store.to_blob(v))) + specs = [("x1", "s1", "2026-01-01T00:00:00+00:00", "hx"), + ("x2", "s2", "2026-01-01T00:10:00+00:00", "hx"), + ("x3", "s1", "2026-01-01T00:20:00+00:00", "hx")] + cid = _cluster_of(con, specs, v) + assert cluster.origin_count(con, cid) <= cluster.bucket_count(con, cid) diff --git a/tests/test_config.py b/tests/test_config.py index 2e511a2..7c24b16 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -64,7 +64,8 @@ def test_machinery_values_are_read_from_config(cfgdir): assert cfg.http["timeout_secs"] == 7 assert cfg.http["concurrency"] == 3 assert cfg.embedding == {"model": "intfloat/multilingual-e5-small", "dim": 512} - assert cfg.clustering == {"threshold": 0.5, "window_hours": 6} + assert cfg.clustering["threshold"] == 0.5 + assert cfg.clustering["window_hours"] == 6 assert cfg.scoring["veto_absolute"] == 0.99