"""Audit stored typed entities against the precision rules (`connectors/_precision`) — read-only by default. For each table it samples N rows (seeded, reproducible), applies the same validators the connector and the pipeline now apply, and prints a precision estimate (share of rows the rules accept, with a 95 % interval) plus the rows that would now be rejected, grouped by reason. Run it before and after a deploy to measure the effect on production. DATABASE_URL=… .venv/bin/python scripts/audit_extractions.py # sample 200 rows per table .venv/bin/python scripts/audit_extractions.py --sample 500 --table people,locations --show 40 .venv/bin/python scripts/audit_extractions.py --purge # delete rows failing the rules (see below) Scope: by default only rows attached to a `generic-html-v1` sensor are audited (the rows the rules were written for); `--all-sources` widens the read-only audit to enrichment-sourced rows (e.g. Wikidata people) as a recall check on the validators. `--purge` scans the *whole* table (not a sample) and deletes HTML-sourced rows that fail the validators, for `people`, `products`, `pricing_plans`, `locations` and `jobs` (`sensors.connector_id = 'generic-html-v1'`); it never touches ATS jobs, enrichment rows or `news_items`. People whose only defect is a stoplisted title ("Contact", "Read more") are kept with `title = null`. Rows the rules would *rename* (swapped person cards, country-as-name locations) are reported, not rewritten: the next crawl re-extracts them correctly and the reconciliation marks the old row `no_longer_listed`. Events derived from purged rows are history and stay untouched. """ from __future__ import annotations import argparse import asyncio import math from collections import Counter, defaultdict from collections.abc import Callable, Sequence from dataclasses import dataclass, field from typing import Any from companyatlas.connectors import _precision as P from companyatlas.db import dispose, execute, fetch_all, transaction from companyatlas.sdk.models import ExtractedLocation, ExtractedPlan DEFAULT_SAMPLE = 200 DEFAULT_SHOW = 25 DEFAULT_SEED = 0.42 Z_95 = 1.96 LABEL_WIDTH = 70 HTML_CONNECTOR = "generic-html-v1" PURGE_BATCH = 500 @dataclass(slots=True) class Outcome: """What the rules say about one stored row.""" status: str # accept | reject | normalise reason: str | None = None fix: str | None = None # normalise: what would change (for the report) — "title → null" is applied under --purge @dataclass(slots=True) class TableSpec: name: str label_col: str select_sql: str # must select id + the columns `judge` reads; `{where}` and `{limit}` are filled in judge: Callable[[dict[str, Any]], Outcome] purgeable: bool label: Callable[[dict[str, Any]], str] summary_cols: Sequence[str] = field(default_factory=tuple) # ------------------------------------------------------------------------------------------------------------ judges (one per table) def _judge_person(row: dict[str, Any]) -> Outcome: fixed = P.normalize_person(row["name"] or "", row.get("title")) if fixed is None: return Outcome("reject", P.person_verdict(row["name"] or "", row.get("title")).reason) name, title = fixed if name != (row["name"] or "").strip(): return Outcome("normalise", "name/title swapped", fix=f"name → {name!r}, title → {title!r}") if (title or None) != (row.get("title") or None): return Outcome("normalise", "title is a link label", fix="title → null") return Outcome("accept") def _judge_product(row: dict[str, Any]) -> Outcome: v = P.product_verdict(row["name"] or "") return Outcome("accept") if v.ok else Outcome("reject", v.reason) def _judge_plan(row: dict[str, Any]) -> Outcome: plan = ExtractedPlan(plan_name=row["plan_name"] or "", price=float(row["price"]) if row.get("price") is not None else None, price_text=row.get("price_text"), contact_sales=bool(row.get("contact_sales"))) v = P.plan_verdict(plan) return Outcome("accept") if v.ok else Outcome("reject", v.reason) def _judge_location(row: dict[str, Any]) -> Outcome: loc = ExtractedLocation(name=row["name"] or "", kind=row.get("kind") or "office", city=row.get("city"), region=row.get("region"), country=row.get("country"), address_text=None) v = P.location_verdict(loc) if not v.ok: return Outcome("reject", v.reason) fixed = P.normalize_location(loc) if fixed is not None and fixed.name != loc.name: return Outcome("normalise", "country used as the name", fix=f"name → {fixed.name!r}, city → {fixed.city!r}") if fixed is not None and fixed.city != loc.city: return Outcome("normalise", "city cell is not a city", fix=f"city → {fixed.city!r}") return Outcome("accept") def _judge_job(row: dict[str, Any]) -> Outcome: v = P.job_verdict(row["title"] or "", url=row.get("url"), location=row.get("location_text"), department=row.get("department")) if not v.ok: return Outcome("reject", v.reason) clean, _ = P.clean_job_title(row["title"] or "") if clean and clean != (row["title"] or "").strip(): return Outcome("normalise", "display title carries a marker / id", fix=f"title → {clean!r}") return Outcome("accept") def _judge_news(row: dict[str, Any]) -> Outcome: v = P.news_verdict(row["title"] or "", url=row.get("url"), published_at=row.get("published_at")) return Outcome("accept") if v.ok else Outcome("reject", v.reason) def _lbl(*parts: Any) -> str: return " | ".join(str(p) for p in parts if p not in (None, "")) # Every select joins `sensors` so rows can be scoped to the HTML connector (`{where}`): enrichment-sourced rows (Wikidata people…) # have no HTML sensor and are only ever *audited* with --all-sources, never purged. TABLES: dict[str, TableSpec] = { "people": TableSpec("people", "name", "select t.id, t.name, t.title from people t left join sensors s on s.id = t.sensor_id {where} {limit}", _judge_person, True, lambda r: _lbl(r["name"], r.get("title"))), "products": TableSpec("products", "name", "select t.id, t.name from products t left join sensors s on s.id = t.sensor_id {where} {limit}", _judge_product, True, lambda r: _lbl(r["name"])), "pricing_plans": TableSpec("pricing_plans", "plan_name", "select t.id, t.plan_name, t.price, t.price_text, t.contact_sales from pricing_plans t left join sensors s on s.id = t.sensor_id " "{where} {limit}", _judge_plan, True, lambda r: _lbl(r["plan_name"], r.get("price_text"))), "locations": TableSpec("locations", "name", "select t.id, t.name, t.kind, t.city, t.region, t.country from locations t left join sensors s on s.id = t.sensor_id {where} {limit}", _judge_location, True, lambda r: _lbl(r["name"], r.get("city"), r.get("country"), r.get("kind"))), "jobs": TableSpec("jobs", "title", "select t.id, t.title, t.url, t.location_text, t.department from jobs t left join sensors s on s.id = t.sensor_id {where} {limit}", _judge_job, True, lambda r: _lbl(r["title"], r.get("location_text"))), "news_items": TableSpec("news_items", "title", "select t.id, t.title, t.url, t.published_at from news_items t left join sensors s on s.id = t.sensor_id {where} {limit}", _judge_news, False, lambda r: _lbl(r["title"], r.get("url"))), } PURGEABLE = [t for t, spec in TABLES.items() if spec.purgeable] HTML_ONLY_WHERE = f"where s.connector_id = '{HTML_CONNECTOR}'" # ------------------------------------------------------------------------------------------------------------ reporting def _interval(k: int, n: int) -> tuple[float, float]: """Wilson 95 % interval for a proportion k/n.""" if n == 0: return 0.0, 0.0 p = k / n denom = 1 + Z_95 ** 2 / n centre = (p + Z_95 ** 2 / (2 * n)) / denom half = Z_95 * math.sqrt(p * (1 - p) / n + Z_95 ** 2 / (4 * n * n)) / denom return max(0.0, centre - half), min(1.0, centre + half) def _sql(spec: TableSpec, *, sample: int | None, html_only: bool) -> str: where = HTML_ONLY_WHERE if (html_only or spec.name == "jobs") else "" # jobs: ATS boards are always out of scope limit = f"order by random() limit {int(sample)}" if sample else "" return spec.select_sql.format(where=where, limit=limit) async def _audit_table(conn: Any, spec: TableSpec, *, sample: int | None, show: int, html_only: bool) -> tuple[list[str], list[str]]: rows = await fetch_all(conn, _sql(spec, sample=sample, html_only=html_only)) outcomes = [(r, spec.judge(r)) for r in rows] accepted = sum(1 for _, o in outcomes if o.status == "accept") normalised = [(r, o) for r, o in outcomes if o.status == "normalise"] rejected = [(r, o) for r, o in outcomes if o.status == "reject"] n = len(rows) lo, hi = _interval(accepted + len(normalised), n) print(f"\n== {spec.name}: sampled {n} · accepted {accepted} · normalised {len(normalised)} · rejected {len(rejected)}") if n: print(f" precision estimate (accepted + normalisable) = {(accepted + len(normalised)) / n:.1%} [{lo:.1%}, {hi:.1%}]" f" rejected share = {len(rejected) / n:.1%}") reasons: Counter[str] = Counter(o.reason or "?" for _, o in rejected) by_reason: dict[str, list[str]] = defaultdict(list) for r, o in rejected: by_reason[o.reason or "?"].append(spec.label(r)) for reason, count in reasons.most_common(): print(f" - {reason}: {count}") for lbl in by_reason[reason][:show]: print(f" ✗ {lbl[:LABEL_WIDTH]}") if normalised: print(f" ~ would be normalised ({len(normalised)}):") for r, o in normalised[:show]: print(f" ~ {spec.label(r)[:LABEL_WIDTH]} → {o.fix}") return [r["id"] for r, _ in rejected], [r["id"] for r, o in normalised if o.fix == "title → null"] async def _purge(conn: Any, spec: TableSpec, reject_ids: list[str], null_title_ids: list[str]) -> None: for i in range(0, len(reject_ids), PURGE_BATCH): await execute(conn, f"delete from {spec.name} where id = any(cast(:ids as text[]))", ids=reject_ids[i:i + PURGE_BATCH]) if spec.name == "people": for i in range(0, len(null_title_ids), PURGE_BATCH): await execute(conn, "update people set title = null where id = any(cast(:ids as text[]))", ids=null_title_ids[i:i + PURGE_BATCH]) print(f" purged {len(reject_ids)} row(s) from {spec.name}" + (f", cleared {len(null_title_ids)} title(s)" if spec.name == "people" else "")) async def main(args: argparse.Namespace) -> None: names = [t.strip() for t in args.table.split(",")] if args.table else list(TABLES) unknown = [t for t in names if t not in TABLES] if unknown: raise SystemExit(f"unknown table(s): {unknown}; choose from {list(TABLES)}") if args.purge: names = [t for t in names if TABLES[t].purgeable] print(f"PURGE mode: full scan of {names}; rows failing the precision rules will be deleted (news and ATS jobs are never touched)") sample = None if args.purge else args.sample html_only = args.purge or not args.all_sources # a purge is always scoped to HTML-sourced rows print(f"precision rules {P.PRECISION_VERSION} · sample={'all' if sample is None else sample} · seed={args.seed} · " f"scope={'HTML-sourced rows (generic-html-v1)' if html_only else 'all sources'}") async with transaction() as conn: await execute(conn, "select setseed(:s)", s=args.seed) for name in names: spec = TABLES[name] reject_ids, null_title_ids = await _audit_table(conn, spec, sample=sample, show=args.show, html_only=html_only) if args.purge: await _purge(conn, spec, reject_ids, null_title_ids) print("\ndone" if args.purge else "\nread-only audit done (pass --purge to delete the rejected rows of the purgeable tables)") def _parse() -> argparse.Namespace: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--sample", type=int, default=DEFAULT_SAMPLE, help=f"rows sampled per table (default {DEFAULT_SAMPLE})") ap.add_argument("--table", type=str, default="", help=f"comma-separated subset of {list(TABLES)}") ap.add_argument("--show", type=int, default=DEFAULT_SHOW, help="examples printed per reason") ap.add_argument("--seed", type=float, default=DEFAULT_SEED, help="Postgres setseed() value in [-1, 1] for reproducible samples") ap.add_argument("--purge", action="store_true", help=f"delete rows failing the rules in {PURGEABLE} (full scan, not a sample, HTML-sourced only)") ap.add_argument("--all-sources", action="store_true", help="read-only audit over every source (enrichment rows included); ignored with --purge") return ap.parse_args() async def _run() -> None: try: await main(_parse()) finally: await dispose() if __name__ == "__main__": asyncio.run(_run())