diff --git a/news/cli.py b/news/cli.py index 9788a02..9b9b315 100644 --- a/news/cli.py +++ b/news/cli.py @@ -50,11 +50,15 @@ def cmd_clusters(args) -> int: (args.min_size, args.limit), ).fetchall() for r in rows: - buckets = cluster_mod.bucket_count(con, r["id"]) - 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']}") + sc = cfg.scoring + co = cluster_mod.corroboration( + con, r["id"], cfg.clustering["syndication_threshold"], + sc["corroboration_pickup_weight"], sc["corroboration_depth_weight"]) + extra = "".join( + (f", {co.pickups} syndicated pickup(s)" if co.pickups else ""), + ) + (f", {co.depth} extra piece(s) same newsroom" if co.depth else "") + print(f"\n[{r['size']:>3} items | corroboration {co.score:.2f} " + f"= {co.origins} origin(s){extra}] {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 adca632..469f757 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 math import re import sqlite3 from dataclasses import dataclass @@ -39,6 +40,9 @@ WINDOW_HOURS = 48 # 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 +# Tiebreaker weights. Small on purpose: origins must stay the dominant term. +PICKUP_WEIGHT = 0.3 +DEPTH_WEIGHT = 0.2 @dataclass(frozen=True) @@ -155,33 +159,52 @@ def _normalise_title(title: str) -> str: return " ".join(_PUNCT.sub("", title.lower()).split()) -def origin_count( +@dataclass(frozen=True) +class Corroboration: + """How much independent judgement a cluster represents. + + Not one number, because "several outlets carried this" covers four + situations that mean different things. Splitting on same/different newsroom + and same/different text: + + same newsroom, same text -> pure duplicate (BBC World + BBC top). + Zero information. The editorial signal + there is the front-page placement, which + `sources.prominence` captures separately. + same newsroom, diff text -> DEPTH. A newsroom filing four pieces on + one story is a real judgement of + importance. + diff newsroom, same text -> PICKUPS. Syndicated copy. Each carrier + chose to run it: weak signal, not zero. + diff newsroom, diff text -> ORIGINS. Independent judgement AND + independent newsgathering. Strongest. + + Measured on a real 91-cluster corpus: 163 origins, 90 depth, 11 pickups. + Depth dominates because these sources are mostly national outlets writing + their own copy; pickups would matter far more with local papers on the wire. + """ + + origins: int + pickups: int + depth: int + score: float + + +def corroboration( con: sqlite3.Connection, cluster_id: str, syndication_threshold: float = SYNDICATION_THRESHOLD, -) -> int: - """Number of independent newsrooms that originated reporting in a cluster. + pickup_weight: float = PICKUP_WEIGHT, + depth_weight: float = DEPTH_WEIGHT, +) -> Corroboration: + """Score = origins + w_p*log1p(pickups) + w_d*log1p(depth). - 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. + Origins dominate by design; the other two break ties among the many + two- and three-origin clusters. The log damping stops a high-volume feed + inflating `depth` by publishing many near-identical pieces. """ rows = con.execute( - """SELECT i.id, i.title, i.content_hash, s.bucket + """SELECT 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 @@ -189,7 +212,7 @@ def origin_count( (cluster_id,), ).fetchall() if not rows: - return 0 + return Corroboration(0, 0, 0, 0.0) vectors: dict[str, np.ndarray] = {} for row in rows: @@ -199,27 +222,61 @@ def origin_count( 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 - + n = len(rows) 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 same_text(i: int, j: int) -> bool: + if titles[i] == titles[j]: + return True + vi, vj = vectors.get(rows[i]["content_hash"]), vectors.get(rows[j]["content_hash"]) + return vi is not None and vj is not None and float(vi @ vj) >= syndication_threshold + + def components(pred) -> list[int]: + parent = list(range(n)) + + def find(a: int) -> int: + while parent[a] != a: + parent[a] = parent[parent[a]] + a = parent[a] + return a + + for i in range(n): + for j in range(i + 1, n): + if pred(i, j): + parent[find(i)] = find(j) + return [find(i) for i in range(n)] + + # Text groups: items that are the same piece of writing, wherever published. + text_group = components(same_text) + # Origins: same newsroom OR same writing. + origin = components(lambda i, j: rows[i]["bucket"] == rows[j]["bucket"] or same_text(i, j)) + + origins = len(set(origin)) + + # Pickups: for each piece of writing carried by more than one newsroom, + # every carrier beyond the first. + buckets_per_text: dict[int, set[str]] = {} + for idx, g in enumerate(text_group): + buckets_per_text.setdefault(g, set()).add(rows[idx]["bucket"]) + pickups = sum(max(0, len(b) - 1) for b in buckets_per_text.values()) + + # Depth: for each newsroom, every distinct piece it filed beyond the first. + texts_per_bucket: dict[str, set[int]] = {} + for idx, g in enumerate(text_group): + texts_per_bucket.setdefault(rows[idx]["bucket"], set()).add(g) + depth = sum(max(0, len(t) - 1) for t in texts_per_bucket.values()) + + score = origins + pickup_weight * math.log1p(pickups) + depth_weight * math.log1p(depth) + return Corroboration(origins, pickups, depth, score) + + +def origin_count( + con: sqlite3.Connection, + cluster_id: str, + syndication_threshold: float = SYNDICATION_THRESHOLD, +) -> int: + """Independent originating newsrooms. See `corroboration` for the full split.""" + return corroboration(con, cluster_id, syndication_threshold).origins def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int: diff --git a/news/config.py b/news/config.py index ca8c307..f9e2bbe 100644 --- a/news/config.py +++ b/news/config.py @@ -95,6 +95,8 @@ class Config: "normalize": s.get("normalize", "zscore-per-block"), "veto_absolute": s.get("veto_absolute", 0.60), "veto_margin_over_interest": s.get("veto_margin_over_interest", 0.3), + "corroboration_pickup_weight": s.get("corroboration_pickup_weight", 0.3), + "corroboration_depth_weight": s.get("corroboration_depth_weight", 0.2), } # -- reader ------------------------------------------------------------ diff --git a/profile/config.toml b/profile/config.toml index 0d0c284..ddd4228 100644 --- a/profile/config.toml +++ b/profile/config.toml @@ -57,3 +57,13 @@ normalize = "zscore-per-block" # similarity OR when the veto signal beats the best interest signal by a margin. veto_absolute = 0.60 veto_margin_over_interest = 0.3 + +# Corroboration = origins + pickup_weight*log1p(pickups) + depth_weight*log1p(depth). +# "Several outlets carried this" covers four different situations; see +# news/cluster.py:Corroboration. Origins (independent newsrooms doing their own +# reporting) dominate; the other two break ties among the many two- and +# three-origin clusters. Measured on a real corpus: 163 origins, 90 depth, +# 11 pickups -- depth matters far more here because these sources mostly write +# their own copy. Raise pickup_weight if you add local papers running wire. +corroboration_pickup_weight = 0.3 +corroboration_depth_weight = 0.2 diff --git a/tests/test_cluster.py b/tests/test_cluster.py index 30315aa..637e0cf 100644 --- a/tests/test_cluster.py +++ b/tests/test_cluster.py @@ -153,3 +153,114 @@ def test_origin_count_never_exceeds_bucket_count(con): ("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) + + +def equicos(n, c=0.90, dim=None): + """n unit vectors with pairwise cosine exactly `c`. + + Needed because these tests must sit in a narrow band: above the 0.82 + clustering threshold so the items form ONE cluster, below the 0.97 + syndication threshold so they read as distinct pieces of writing. Earlier + drafts of this test used ad-hoc vectors that fell outside it in both + directions and tested nothing. + """ + dim = dim or n + 1 + out = [] + for i in range(n): + v = np.zeros(dim, dtype=np.float32) + v[0] = np.sqrt(c) + v[1 + i] = np.sqrt(1 - c) + out.append(v) + return out + + +def _mk(con, spec, vec_for): + """spec: list of (item_id, source_id, title, content_hash). One cluster.""" + when = "2026-01-01T00:{:02d}:00+00:00" + rows = [] + for n, (iid, sid, title, chash) in enumerate(spec): + con.execute( + """INSERT INTO items (id,source_id,title,summary,url,canonical_url, + published_at,first_seen_at,content_hash) + VALUES (?,?,?,'',?,?,?,?,?)""", + (iid, sid, title, f"http://x/{iid}", f"http://x/{iid}", + when.format(n), when.format(n), chash)) + v = vec_for(chash) + con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)", + (chash, len(v), store.to_blob(v))) + rows = con.execute("SELECT * FROM items ORDER BY published_at").fetchall() + v = {r["content_hash"]: vec_for(r["content_hash"]) for r in rows} + return cluster.assign(con, list(rows), v)[0].cluster_id + + +def _sources(con, *pairs): + for sid, bucket in pairs: + store.upsert_source(con, id=sid, name=sid, url=f"http://{sid}", kind="rss", + bucket=bucket, weight=1.0, prominence=0, enabled=1, + poll_interval_secs=900) + + +def test_corroboration_ranks_the_four_cases_in_the_right_order(con): + """The 2x2 that motivates the split, each case scored in isolation. + + Ordering must be: four independent newsrooms > four outlets running one + wire story > one newsroom filing four pieces > one story on two feeds of + the same newsroom. + """ + # "wire" is one shared vector (cosine 1.0 -> syndicated); t0..t3 and u0..u3 + # are pairwise 0.90 -> one cluster, four distinct pieces of writing. + distinct = equicos(4, 0.90, dim=6) + table = {"wire": np.eye(6, dtype=np.float32)[0]} + for i in range(4): + table[f"t{i}"] = distinct[i] + table[f"u{i}"] = distinct[i] + + def vec(h): + return table[h] + + # (a) one newsroom, one story, two feeds -> pure duplicate + c1 = store.connect(":memory:") + _sources(c1, ("f1", "bbc"), ("f2", "bbc")) + cid = _mk(c1, [("a1", "f1", "Same headline", "wire"), + ("a2", "f2", "Same headline", "wire")], vec) + dup = cluster.corroboration(c1, cid) + + # (b) one newsroom, four distinct pieces -> depth + c2 = store.connect(":memory:") + _sources(c2, ("g1", "nyt")) + cid = _mk(c2, [(f"b{i}", "g1", f"Distinct angle {i}", f"t{i}") for i in range(4)], vec) + deep = cluster.corroboration(c2, cid) + + # (c) four newsrooms, one syndicated wire story -> pickups + c3 = store.connect(":memory:") + _sources(c3, *[(f"h{i}", f"local{i}") for i in range(4)]) + cid = _mk(c3, [(f"c{i}", f"h{i}", "Wire dispatch text", "wire") for i in range(4)], vec) + wire = cluster.corroboration(c3, cid) + + # (d) four newsrooms each writing their own -> origins + c4 = store.connect(":memory:") + _sources(c4, *[(f"k{i}", f"nat{i}") for i in range(4)]) + cid = _mk(c4, [(f"d{i}", f"k{i}", f"Own reporting {i}", f"u{i}") for i in range(4)], vec) + orig = cluster.corroboration(c4, cid) + + assert dup.origins == 1 and dup.pickups == 0 and dup.depth == 0 + assert deep.origins == 1 and deep.depth == 3 + assert wire.origins == 1 and wire.pickups == 3 + assert orig.origins == 4 + + assert orig.score > wire.score > deep.score > dup.score, ( + f"expected origins > pickups > depth > duplicate, got " + f"{orig.score:.2f} / {wire.score:.2f} / {deep.score:.2f} / {dup.score:.2f}") + + +def test_origins_dominate_the_tiebreakers(con): + """No amount of depth or pickups may outweigh one extra real newsroom.""" + c = store.connect(":memory:") + _sources(c, ("m1", "one")) + vs = equicos(20, 0.90) + table = {f"v{i}": vs[i] for i in range(20)} + cid = _mk(c, [(f"e{i}", "m1", f"Piece {i}", f"v{i}") for i in range(20)], + lambda h: table[h]) + twenty_deep = cluster.corroboration(c, cid) + assert twenty_deep.score < 2.0, ( + "20 pieces from one newsroom must still rank below two independent ones")