zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
12 kB
338 lines
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = ["httpx", "pydantic-settings"]
5# ///
6"""Offline bulk-mirror classifier (Phase 1 of docs/spam-detection-plan.md).
7
8Scores every publisher (DID) on whether it looks like a machine-generated
9registry/feed mirror vs human long-form writing, and validates the score
10against a hand-labeled known set. This is the *emission brain* for the planned
11pub-search labeler: it decides which DIDs should carry a `bulk-mirror` account
12label. It does NOT ban anything and does NOT write — read-only.
13
14WHAT THIS IS: the *decision function* — "given a DID's records, is this a
15bulk-mirror?" — plus an offline harness to prove it. A single record can't be
16judged ("S07E14: The Cadillac" is innocent alone); only the pattern across a
17DID's records reveals the mirror, so the decision needs a per-DID aggregate.
18
19This script feeds that aggregate from turso for two reasons: (1) validate the
20scoring against the known set offline, and (2) the one-time BACKFILL of ~40k
21docs indexed before the labeler exists. At steady state there is no batch job
22and no turso — the live labeler is a firehose listener that maintains the same
23rolling per-DID aggregate in its own store and runs this same decision function.
24
25WHY THESE SIGNALS (the data forced them — see the plan):
26- volume & burst ALONE are useless: the corpus's #1 author by doc count
27 (coryd.dev, 941) is a real human, and real blogs backfill hundreds of posts
28 in a day. so volume only *corroborates*, never decides.
29- absence of curation proves nothing (the recommend graph is too sparse —
30 coryd.dev has 0 recommends). so curation is a VETO only: if an account HAS
31 curation, never emit.
32- the real discriminator is authorship / content SHAPE: templated titles
33 ("Train #N Delayed", "I felt X"), empty/templated content, thin length.
34
35Usage:
36 scripts/classify-bulk-mirror # rank + validate (min 100 docs)
37 scripts/classify-bulk-mirror --min-docs 50 # widen the candidate floor
38 scripts/classify-bulk-mirror --samples # print sample titles per row
39"""
40
41from __future__ import annotations
42
43import argparse
44import asyncio
45import math
46import os
47import re
48import sys
49import time
50from dataclasses import dataclass
51
52import httpx
53from pydantic_settings import BaseSettings, SettingsConfigDict
54
55# ── known set (hand-labeled, 2026-06-29) ────────────────────────────────────
56# the 3 historically-banned DIDs are purged from turso (no live rows to score),
57# so the live validation set is the 4 surfaced vs the 4 real humans.
58FLAG = {
59 "did:plc:jgg4dtdflzzemyvnybucnzdw": "chicagotransitalerts.app",
60 "did:plc:5swhfspkrynnbidlkrkch3lh": "prideraiser.org",
61 "did:plc:esvvys4mvoclui34shb23l5w": "eligundry.com",
62 "did:plc:gvudfu6dhl5cbpinhofa325s": "alekslessmann.de",
63}
64KEEP = {
65 "did:plc:sttgf52vkk46f6yuknvqxvgh": "coryd.dev",
66 "did:plc:77mn3ult3b72tpvtqqva6tat": "frankhecker.com",
67 "did:plc:rkjxbatkiros6f7pwtgsir54": "den.dev",
68 "did:plc:icpcpp5txyow3prnfgi533lj": "mikebifulco.com",
69}
70
71EMIT_THRESHOLD = 0.35 # score at/above which the labeler would emit bulk-mirror
72
73STOPWORDS = {
74 "the", "a", "an", "and", "or", "but", "of", "to", "in", "on", "for", "with",
75 "at", "by", "from", "is", "are", "was", "were", "be", "as", "it", "this",
76 "that", "i", "you", "we", "my", "your", "our", "me", "no", "not", "do",
77}
78
79CANARY_SLOW_SECS = 1.5
80
81
82def log(msg: str = "") -> None:
83 print(msg, flush=True)
84
85
86# ── turso plumbing (same shape as scripts/purge-banned-turso) ────────────────
87class Settings(BaseSettings):
88 model_config = SettingsConfigDict(
89 env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
90 )
91 turso_url: str
92 turso_token: str
93
94 @property
95 def turso_host(self) -> str:
96 url = self.turso_url
97 return url[len("libsql://"):] if url.startswith("libsql://") else url
98
99
100def _stmt(sql: str, args: list | None = None) -> dict:
101 s: dict = {"sql": sql}
102 if args:
103 s["args"] = [{"type": "text", "value": str(a)} for a in args]
104 return s
105
106
107async def _execute(client: httpx.AsyncClient, st: Settings, sql: str, args=None) -> dict:
108 r = await client.post(
109 f"https://{st.turso_host}/v2/pipeline",
110 headers={"Authorization": f"Bearer {st.turso_token}", "Content-Type": "application/json"},
111 json={"requests": [{"type": "execute", "stmt": _stmt(sql, args)}, {"type": "close"}]},
112 timeout=120,
113 )
114 r.raise_for_status()
115 res = r.json()["results"][0]["response"]
116 if res.get("type") == "error":
117 raise RuntimeError(res["error"])
118 return res["result"]
119
120
121async def rows(client, st, sql, args=None) -> list[list]:
122 res = await _execute(client, st, sql, args)
123 return [[c.get("value") for c in row] for row in res.get("rows", [])]
124
125
126async def canary(client, st) -> None:
127 t0 = time.monotonic()
128 await _execute(client, st, "SELECT 1")
129 dt = time.monotonic() - t0
130 if dt > CANARY_SLOW_SECS:
131 log(f"ABORT: turso canary {dt:.2f}s — prod shares this db; not piling on.")
132 sys.exit(2)
133
134
135# ── feature extraction ───────────────────────────────────────────────────────
136def _norm(title: str) -> str:
137 t = title.lower()
138 t = re.sub(r"\d+", "#", t)
139 t = re.sub(r"[^\w\s#]", " ", t)
140 return re.sub(r"\s+", " ", t).strip()
141
142
143def _toks(title: str) -> set[str]:
144 return {w for w in _norm(title).split() if len(w) > 1 and w not in STOPWORDS}
145
146
147@dataclass
148class Pub:
149 did: str
150 docs: int
151 pubs: int
152 avg_len: float
153 titles: list[str]
154 recos: int
155 subs: int
156
157 @property
158 def empty_title_frac(self) -> float:
159 return sum(1 for t in self.titles if not t.strip()) / max(1, len(self.titles))
160
161 @property
162 def digit_title_frac(self) -> float:
163 return sum(1 for t in self.titles if re.search(r"\d", t)) / max(1, len(self.titles))
164
165 @property
166 def distinct_norm_ratio(self) -> float:
167 return len({_norm(t) for t in self.titles}) / max(1, len(self.titles))
168
169 @property
170 def scaffold_coverage(self) -> float:
171 """How much of each title is shared scaffolding. Tokens appearing in
172 >=50% of titles form the scaffold; coverage = mean fraction of a
173 title's content tokens that are scaffold. High = templated."""
174 toksets = [_toks(t) for t in self.titles]
175 n = len(toksets)
176 if n == 0:
177 return 0.0
178 freq: dict[str, int] = {}
179 for ts in toksets:
180 for w in ts:
181 freq[w] = freq.get(w, 0) + 1
182 scaffold = {w for w, c in freq.items() if c >= 0.5 * n}
183 cov = []
184 for ts in toksets:
185 if not ts:
186 cov.append(1.0) # empty/punctuation-only title is maximally templated
187 else:
188 cov.append(len(ts & scaffold) / len(ts))
189 return sum(cov) / n
190
191 @property
192 def template(self) -> float:
193 return max(1.0 - self.distinct_norm_ratio, self.scaffold_coverage)
194
195 @property
196 def thinness(self) -> float:
197 return min(max(1.0 - self.avg_len / 800.0, 0.0), 1.0)
198
199 @property
200 def structural(self) -> float:
201 return max(self.digit_title_frac, self.empty_title_frac)
202
203 @property
204 def volume(self) -> float:
205 # 100 docs → 0, 10k docs → 1 (log-scaled corroboration only)
206 return min(max((math.log10(max(self.docs, 1)) - 2.0) / 2.0, 0.0), 1.0)
207
208 @property
209 def vetoed(self) -> bool:
210 return self.recos > 0 or self.subs > 0
211
212 @property
213 def score(self) -> float:
214 raw = (
215 0.45 * self.template
216 + 0.20 * self.structural
217 + 0.20 * self.thinness
218 + 0.15 * self.volume
219 )
220 return 0.0 if self.vetoed else raw
221
222
223async def build_pubs(client, st, min_docs: int) -> list[Pub]:
224 # Q1: per-DID aggregates over the whole corpus (one GROUP BY scan).
225 agg = await rows(client, st, f"""
226 SELECT did, COUNT(*) docs, COUNT(DISTINCT publication_uri) pubs,
227 AVG(LENGTH(content)) avg_len
228 FROM documents
229 WHERE is_bridgyfed = 0
230 GROUP BY did
231 HAVING docs >= {int(min_docs)}
232 ORDER BY docs DESC
233 """)
234 cand = {r[0]: (int(r[1]), int(r[2]), float(r[3] or 0)) for r in agg}
235 if not cand:
236 return []
237
238 await canary(client, st)
239
240 # Q2: titles for candidates only (bounded by candidate count).
241 dids = list(cand)
242 ph = ",".join("?" for _ in dids)
243 title_rows = await rows(
244 client, st,
245 f"SELECT did, title FROM documents WHERE is_bridgyfed = 0 AND did IN ({ph})",
246 dids,
247 )
248 titles: dict[str, list[str]] = {d: [] for d in dids}
249 for did, title in title_rows:
250 titles[did].append(title or "")
251
252 await canary(client, st)
253
254 # Q3: curation signal (the veto) per candidate.
255 recos = {
256 r[0]: int(r[1]) for r in await rows(client, st, f"""
257 SELECT d.did, COUNT(*) FROM recommends r
258 JOIN documents d ON r.document_uri = d.uri
259 WHERE d.did IN ({ph}) GROUP BY d.did
260 """, dids)
261 }
262 subs = {
263 r[0]: int(r[1]) for r in await rows(client, st, f"""
264 SELECT p.did, COUNT(*) FROM subscriptions s
265 JOIN publications p ON s.publication_uri = p.uri
266 WHERE p.did IN ({ph}) GROUP BY p.did
267 """, dids)
268 }
269
270 out = []
271 for did, (docs, pubs, avg_len) in cand.items():
272 out.append(Pub(did, docs, pubs, avg_len, titles[did], recos.get(did, 0), subs.get(did, 0)))
273 out.sort(key=lambda p: p.score, reverse=True)
274 return out
275
276
277def validate(pubs: list[Pub]) -> int:
278 by_did = {p.did: p for p in pubs}
279 log("\nvalidation against the known set (FLAG should be ≥ %.2f, KEEP below):" % EMIT_THRESHOLD)
280 ok = True
281 worst_flag, best_keep = 1.0, 0.0
282 for did, name in FLAG.items():
283 p = by_did.get(did)
284 if not p:
285 log(f" ?? FLAG {name:<26} no live rows (purged or below floor)")
286 continue
287 worst_flag = min(worst_flag, p.score)
288 hit = p.score >= EMIT_THRESHOLD
289 ok &= hit
290 log(f" {'✓' if hit else '✗'} FLAG {name:<26} score={p.score:.3f}")
291 for did, name in KEEP.items():
292 p = by_did.get(did)
293 if not p:
294 log(f" ?? KEEP {name:<26} no live rows")
295 continue
296 best_keep = max(best_keep, p.score)
297 hit = p.score < EMIT_THRESHOLD
298 ok &= hit
299 log(f" {'✓' if hit else '✗'} KEEP {name:<26} score={p.score:.3f}")
300 margin = worst_flag - best_keep
301 log(f"\nseparation margin (worst FLAG − best KEEP): {margin:+.3f}")
302 log("RESULT: " + ("PASS — clean separation" if ok else "FAIL — overlap, retune weights"))
303 return 0 if ok else 1
304
305
306async def main() -> int:
307 ap = argparse.ArgumentParser(description="offline bulk-mirror classifier")
308 ap.add_argument("--min-docs", type=int, default=100, help="candidate doc floor")
309 ap.add_argument("--top", type=int, default=25, help="rows to print")
310 ap.add_argument("--samples", action="store_true", help="print sample titles")
311 args = ap.parse_args()
312
313 st = Settings() # type: ignore
314 async with httpx.AsyncClient() as client:
315 pubs = await build_pubs(client, st, args.min_docs)
316
317 if not pubs:
318 log(f"no publishers with ≥{args.min_docs} docs")
319 return 0
320
321 known = {**FLAG, **KEEP}
322 log(f"{'score':>6} {'emit':>4} {'veto':>4} {'tmpl':>5} {'struc':>5} {'thin':>5} "
323 f"{'vol':>4} {'docs':>6} publisher")
324 for p in pubs[:args.top]:
325 tag = known.get(p.did, "")
326 mark = "·" + tag if tag else p.did[:18]
327 log(f"{p.score:6.3f} {'YES' if p.score >= EMIT_THRESHOLD and not p.vetoed else '-':>4} "
328 f"{'V' if p.vetoed else '-':>4} {p.template:5.2f} {p.structural:5.2f} "
329 f"{p.thinness:5.2f} {p.volume:4.2f} {p.docs:6d} {mark}")
330 if args.samples:
331 for t in p.titles[:4]:
332 log(f" · {t[:80]}")
333
334 return validate(pubs)
335
336
337if __name__ == "__main__":
338 raise SystemExit(asyncio.run(main()))