spb/company-atlas

1"""One-off repair: remove snapshots/entities/events produced from mis-decoded bodies (2026-09-12 Brotli incident).23Before the fix, the fetcher advertised `Accept-Encoding: br` without the `brotli` codec installed, so Brotli-compressed responses4were stored and "parsed" as text: garbage titles (U+FFFD runs), 100 %-changed pages, meaningless events. This script:56  1. scans every snapshot's stored text (and title) and flags the corrupt ones (text quality < 0.99 or U+FFFD in the title),7  2. retracts events derived from them (or carrying U+FFFD) — status 'retracted' + reason, never deleted,8  3. deletes the corrupt snapshots, their change rows and the corrupt extracted entities (labels with U+FFFD),9  4. resets the affected sensors (hashes, last_snapshot_id, snapshot_count) and schedules them to run now, so the next observation10     becomes a clean baseline (entities flagged baseline, no bogus "change").1112Raw observation objects are kept (content-addressed archive). Usage: `.venv/bin/python scripts/purge_corrupt.py [--apply]`.13"""14from __future__ import annotations1516import asyncio17import sys18from datetime import UTC, datetime1920from companyatlas import archive21from companyatlas.db import dispose, execute, fetch_all, fetch_val, transaction22from companyatlas.fetch import text_quality2324APPLY = "--apply" in sys.argv25REASON = "corrupt extraction: response body was not decoded (Brotli without codec); repaired 2026-09-13"262728async def main() -> None:29    async with transaction() as conn:30        rows = await fetch_all(conn, "select id, sensor_id, title, text_key from snapshots order by sensor_id, fetched_at")31    corrupt: list[str] = []32    sensors: set[str] = set()33    for r in rows:34        bad = bool(r["title"] and "�" in r["title"])35        if not bad and r["text_key"]:36            try:37                bad = text_quality(archive.get_text(r["text_key"])) < 0.9938            except FileNotFoundError:39                bad = False40        if bad:41            corrupt.append(r["id"])42            sensors.add(r["sensor_id"])43    print(f"snapshots scanned={len(rows)} corrupt={len(corrupt)} sensors_affected={len(sensors)}")44    async with transaction() as conn:45        n_ev_text = await fetch_val(conn, "select count(*) from events where status = 'active' and (title ~ E'\\uFFFD' or coalesce(summary, '') ~ E'\\uFFFD')")46        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[])))",47                                    s=corrupt or ["-"])48        counts = {}49        for table, col in (("news_items", "title"), ("jobs", "title"), ("people", "name"), ("products", "name"), ("pricing_plans", "plan_name"), ("locations", "name")):50            counts[table] = await fetch_val(conn, f"select count(*) from {table} where {col} ~ E'\\uFFFD'")51        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 ["-"])52    print(f"events to retract: text={n_ev_text} snapshot-linked={n_ev_snap} · changes to delete={n_changes} · corrupt entities={counts}")53    if not APPLY:54        print("dry-run (pass --apply to repair)")55        return56    now = datetime.now(UTC)57    async with transaction() as conn:58        await execute(conn, """update events set status = 'retracted', retracted_reason = :r59                               where status = 'active' and (title ~ E'\\uFFFD' or coalesce(summary, '') ~ E'\\uFFFD'60                                     or snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))61                                     or change_id in (select id from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))))""",62                      r=REASON, s=corrupt or ["-"])63        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 ["-"])64        for table, col in (("news_items", "title"), ("jobs", "title"), ("people", "name"), ("products", "name"), ("pricing_plans", "plan_name"), ("locations", "name")):65            await execute(conn, f"delete from {table} where {col} ~ E'\\uFFFD'")66        await execute(conn, "update snapshots set previous_snapshot_id = null where previous_snapshot_id = any(cast(:s as text[]))", s=corrupt or ["-"])67        await execute(conn, "delete from snapshots where id = any(cast(:s as text[]))", s=corrupt or ["-"])68        for sid in sensors:69            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)70            n = await fetch_val(conn, "select count(*) from snapshots where sensor_id = :sid", sid=sid)71            if last:72                await execute(conn, """update sensors set last_snapshot_id = :snap, last_content_hash = :ch, last_normalized_hash = :nh, last_structural_hash = :sh,73                                       snapshot_count = :n, consecutive_unchanged = 0, next_run_at = :now, etag = null, last_modified = null,74                                       config = config - 'last_structured_hash', updated_at = :now where id = :sid""",75                              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)76            else:77                await execute(conn, """update sensors set last_snapshot_id = null, last_content_hash = null, last_normalized_hash = null, last_structural_hash = null,78                                       snapshot_count = 0, change_count = 0, meaningful_change_count = 0, event_count = 0, consecutive_unchanged = 0,79                                       next_run_at = :now, etag = null, last_modified = null, config = config - 'last_structured_hash', updated_at = :now80                                       where id = :sid""", now=now, sid=sid)81        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')")82    print(f"repaired: {len(corrupt)} snapshots removed, {len(sensors)} sensors reset and due now, events retracted, corrupt entities deleted")838485async def _run() -> None:86    try:87        await main()88    finally:89        await dispose()909192if __name__ == "__main__":93    asyncio.run(_run())94