"""One-off repair: remove snapshots/entities/events produced from mis-decoded bodies (2026-09-12 Brotli incident). Before the fix, the fetcher advertised `Accept-Encoding: br` without the `brotli` codec installed, so Brotli-compressed responses were stored and "parsed" as text: garbage titles (U+FFFD runs), 100 %-changed pages, meaningless events. This script: 1. scans every snapshot's stored text (and title) and flags the corrupt ones (text quality < 0.99 or U+FFFD in the title), 2. retracts events derived from them (or carrying U+FFFD) — status 'retracted' + reason, never deleted, 3. deletes the corrupt snapshots, their change rows and the corrupt extracted entities (labels with U+FFFD), 4. resets the affected sensors (hashes, last_snapshot_id, snapshot_count) and schedules them to run now, so the next observation becomes a clean baseline (entities flagged baseline, no bogus "change"). Raw observation objects are kept (content-addressed archive). Usage: `.venv/bin/python scripts/purge_corrupt.py [--apply]`. """ from __future__ import annotations import asyncio import sys from datetime import UTC, datetime from companyatlas import archive from companyatlas.db import dispose, execute, fetch_all, fetch_val, transaction from companyatlas.fetch import text_quality APPLY = "--apply" in sys.argv REASON = "corrupt extraction: response body was not decoded (Brotli without codec); repaired 2026-09-13" async def main() -> None: async with transaction() as conn: rows = await fetch_all(conn, "select id, sensor_id, title, text_key from snapshots order by sensor_id, fetched_at") corrupt: list[str] = [] sensors: set[str] = set() for r in rows: bad = bool(r["title"] and "�" in r["title"]) if not bad and r["text_key"]: try: bad = text_quality(archive.get_text(r["text_key"])) < 0.99 except FileNotFoundError: bad = False if bad: corrupt.append(r["id"]) sensors.add(r["sensor_id"]) print(f"snapshots scanned={len(rows)} corrupt={len(corrupt)} sensors_affected={len(sensors)}") async with transaction() as conn: n_ev_text = await fetch_val(conn, "select count(*) from events where status = 'active' and (title ~ E'\\uFFFD' or coalesce(summary, '') ~ E'\\uFFFD')") n_ev_snap = await fetch_val(conn, "select count(*) from events where status = 'active' and (snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[])))", s=corrupt or ["-"]) counts = {} for table, col in (("news_items", "title"), ("jobs", "title"), ("people", "name"), ("products", "name"), ("pricing_plans", "plan_name"), ("locations", "name")): counts[table] = await fetch_val(conn, f"select count(*) from {table} where {col} ~ E'\\uFFFD'") n_changes = await fetch_val(conn, "select count(*) from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))", s=corrupt or ["-"]) print(f"events to retract: text={n_ev_text} snapshot-linked={n_ev_snap} · changes to delete={n_changes} · corrupt entities={counts}") if not APPLY: print("dry-run (pass --apply to repair)") return now = datetime.now(UTC) async with transaction() as conn: await execute(conn, """update events set status = 'retracted', retracted_reason = :r where status = 'active' and (title ~ E'\\uFFFD' or coalesce(summary, '') ~ E'\\uFFFD' or snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[])) or change_id in (select id from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))))""", r=REASON, s=corrupt or ["-"]) await execute(conn, "delete from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))", s=corrupt or ["-"]) for table, col in (("news_items", "title"), ("jobs", "title"), ("people", "name"), ("products", "name"), ("pricing_plans", "plan_name"), ("locations", "name")): await execute(conn, f"delete from {table} where {col} ~ E'\\uFFFD'") await execute(conn, "update snapshots set previous_snapshot_id = null where previous_snapshot_id = any(cast(:s as text[]))", s=corrupt or ["-"]) await execute(conn, "delete from snapshots where id = any(cast(:s as text[]))", s=corrupt or ["-"]) for sid in sensors: last = await fetch_all(conn, "select id, content_hash, normalized_hash, structural_hash from snapshots where sensor_id = :sid order by fetched_at desc limit 1", sid=sid) n = await fetch_val(conn, "select count(*) from snapshots where sensor_id = :sid", sid=sid) if last: await execute(conn, """update sensors set last_snapshot_id = :snap, last_content_hash = :ch, last_normalized_hash = :nh, last_structural_hash = :sh, snapshot_count = :n, consecutive_unchanged = 0, next_run_at = :now, etag = null, last_modified = null, config = config - 'last_structured_hash', updated_at = :now where id = :sid""", snap=last[0]["id"], ch=last[0]["content_hash"], nh=last[0]["normalized_hash"], sh=last[0]["structural_hash"], n=n, now=now, sid=sid) else: await execute(conn, """update sensors set last_snapshot_id = null, last_content_hash = null, last_normalized_hash = null, last_structural_hash = null, snapshot_count = 0, change_count = 0, meaningful_change_count = 0, event_count = 0, consecutive_unchanged = 0, next_run_at = :now, etag = null, last_modified = null, config = config - 'last_structured_hash', updated_at = :now where id = :sid""", now=now, sid=sid) await execute(conn, "update companies c set last_event_at = (select max(detected_at) from events e where e.company_id = c.id and e.status = 'active')") print(f"repaired: {len(corrupt)} snapshots removed, {len(sensors)} sensors reset and due now, events retracted, corrupt entities deleted") async def _run() -> None: try: await main() finally: await dispose() if __name__ == "__main__": asyncio.run(_run())