spb/company-atlas

1"""Audit stored typed entities against the precision rules (`connectors/_precision`) — read-only by default.23For each table it samples N rows (seeded, reproducible), applies the same validators the connector and the pipeline now apply, and4prints a precision estimate (share of rows the rules accept, with a 95 % interval) plus the rows that would now be rejected, grouped5by reason. Run it before and after a deploy to measure the effect on production.67    DATABASE_URL=… .venv/bin/python scripts/audit_extractions.py                     # sample 200 rows per table8    .venv/bin/python scripts/audit_extractions.py --sample 500 --table people,locations --show 409    .venv/bin/python scripts/audit_extractions.py --purge                            # delete rows failing the rules (see below)1011Scope: by default only rows attached to a `generic-html-v1` sensor are audited (the rows the rules were written for); `--all-sources`12widens the read-only audit to enrichment-sourced rows (e.g. Wikidata people) as a recall check on the validators.13`--purge` scans the *whole* table (not a sample) and deletes HTML-sourced rows that fail the validators, for `people`, `products`,14`pricing_plans`, `locations` and `jobs` (`sensors.connector_id = 'generic-html-v1'`); it never touches ATS jobs, enrichment rows or `news_items`.15People whose only defect is a stoplisted title ("Contact", "Read more") are kept with `title = null`. Rows the rules would *rename*16(swapped person cards, country-as-name locations) are reported, not rewritten: the next crawl re-extracts them correctly and the17reconciliation marks the old row `no_longer_listed`. Events derived from purged rows are history and stay untouched.18"""19from __future__ import annotations2021import argparse22import asyncio23import math24from collections import Counter, defaultdict25from collections.abc import Callable, Sequence26from dataclasses import dataclass, field27from typing import Any2829from companyatlas.connectors import _precision as P30from companyatlas.db import dispose, execute, fetch_all, transaction31from companyatlas.sdk.models import ExtractedLocation, ExtractedPlan3233DEFAULT_SAMPLE = 20034DEFAULT_SHOW = 2535DEFAULT_SEED = 0.4236Z_95 = 1.9637LABEL_WIDTH = 7038HTML_CONNECTOR = "generic-html-v1"39PURGE_BATCH = 500404142@dataclass(slots=True)43class Outcome:44    """What the rules say about one stored row."""45    status: str                      # accept | reject | normalise46    reason: str | None = None47    fix: str | None = None           # normalise: what would change (for the report) — "title → null" is applied under --purge484950@dataclass(slots=True)51class TableSpec:52    name: str53    label_col: str54    select_sql: str                  # must select id + the columns `judge` reads; `{where}` and `{limit}` are filled in55    judge: Callable[[dict[str, Any]], Outcome]56    purgeable: bool57    label: Callable[[dict[str, Any]], str]58    summary_cols: Sequence[str] = field(default_factory=tuple)596061# ------------------------------------------------------------------------------------------------------------ judges (one per table)626364def _judge_person(row: dict[str, Any]) -> Outcome:65    fixed = P.normalize_person(row["name"] or "", row.get("title"))66    if fixed is None:67        return Outcome("reject", P.person_verdict(row["name"] or "", row.get("title")).reason)68    name, title = fixed69    if name != (row["name"] or "").strip():70        return Outcome("normalise", "name/title swapped", fix=f"name → {name!r}, title → {title!r}")71    if (title or None) != (row.get("title") or None):72        return Outcome("normalise", "title is a link label", fix="title → null")73    return Outcome("accept")747576def _judge_product(row: dict[str, Any]) -> Outcome:77    v = P.product_verdict(row["name"] or "")78    return Outcome("accept") if v.ok else Outcome("reject", v.reason)798081def _judge_plan(row: dict[str, Any]) -> Outcome:82    plan = ExtractedPlan(plan_name=row["plan_name"] or "", price=float(row["price"]) if row.get("price") is not None else None,83                         price_text=row.get("price_text"), contact_sales=bool(row.get("contact_sales")))84    v = P.plan_verdict(plan)85    return Outcome("accept") if v.ok else Outcome("reject", v.reason)868788def _judge_location(row: dict[str, Any]) -> Outcome:89    loc = ExtractedLocation(name=row["name"] or "", kind=row.get("kind") or "office", city=row.get("city"), region=row.get("region"),90                            country=row.get("country"), address_text=None)91    v = P.location_verdict(loc)92    if not v.ok:93        return Outcome("reject", v.reason)94    fixed = P.normalize_location(loc)95    if fixed is not None and fixed.name != loc.name:96        return Outcome("normalise", "country used as the name", fix=f"name → {fixed.name!r}, city → {fixed.city!r}")97    if fixed is not None and fixed.city != loc.city:98        return Outcome("normalise", "city cell is not a city", fix=f"city → {fixed.city!r}")99    return Outcome("accept")100101102def _judge_job(row: dict[str, Any]) -> Outcome:103    v = P.job_verdict(row["title"] or "", url=row.get("url"), location=row.get("location_text"), department=row.get("department"))104    if not v.ok:105        return Outcome("reject", v.reason)106    clean, _ = P.clean_job_title(row["title"] or "")107    if clean and clean != (row["title"] or "").strip():108        return Outcome("normalise", "display title carries a marker / id", fix=f"title → {clean!r}")109    return Outcome("accept")110111112def _judge_news(row: dict[str, Any]) -> Outcome:113    v = P.news_verdict(row["title"] or "", url=row.get("url"), published_at=row.get("published_at"))114    return Outcome("accept") if v.ok else Outcome("reject", v.reason)115116117def _lbl(*parts: Any) -> str:118    return " | ".join(str(p) for p in parts if p not in (None, ""))119120121# Every select joins `sensors` so rows can be scoped to the HTML connector (`{where}`): enrichment-sourced rows (Wikidata people…)122# have no HTML sensor and are only ever *audited* with --all-sources, never purged.123TABLES: dict[str, TableSpec] = {124    "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}",125                        _judge_person, True, lambda r: _lbl(r["name"], r.get("title"))),126    "products": TableSpec("products", "name", "select t.id, t.name from products t left join sensors s on s.id = t.sensor_id {where} {limit}",127                          _judge_product, True, lambda r: _lbl(r["name"])),128    "pricing_plans": TableSpec("pricing_plans", "plan_name",129                               "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 "130                               "{where} {limit}", _judge_plan, True, lambda r: _lbl(r["plan_name"], r.get("price_text"))),131    "locations": TableSpec("locations", "name",132                           "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}",133                           _judge_location, True, lambda r: _lbl(r["name"], r.get("city"), r.get("country"), r.get("kind"))),134    "jobs": TableSpec("jobs", "title",135                      "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}",136                      _judge_job, True, lambda r: _lbl(r["title"], r.get("location_text"))),137    "news_items": TableSpec("news_items", "title",138                            "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}",139                            _judge_news, False, lambda r: _lbl(r["title"], r.get("url"))),140}141PURGEABLE = [t for t, spec in TABLES.items() if spec.purgeable]142HTML_ONLY_WHERE = f"where s.connector_id = '{HTML_CONNECTOR}'"143144145# ------------------------------------------------------------------------------------------------------------ reporting146147148def _interval(k: int, n: int) -> tuple[float, float]:149    """Wilson 95 % interval for a proportion k/n."""150    if n == 0:151        return 0.0, 0.0152    p = k / n153    denom = 1 + Z_95 ** 2 / n154    centre = (p + Z_95 ** 2 / (2 * n)) / denom155    half = Z_95 * math.sqrt(p * (1 - p) / n + Z_95 ** 2 / (4 * n * n)) / denom156    return max(0.0, centre - half), min(1.0, centre + half)157158159def _sql(spec: TableSpec, *, sample: int | None, html_only: bool) -> str:160    where = HTML_ONLY_WHERE if (html_only or spec.name == "jobs") else ""     # jobs: ATS boards are always out of scope161    limit = f"order by random() limit {int(sample)}" if sample else ""162    return spec.select_sql.format(where=where, limit=limit)163164165async def _audit_table(conn: Any, spec: TableSpec, *, sample: int | None, show: int, html_only: bool) -> tuple[list[str], list[str]]:166    rows = await fetch_all(conn, _sql(spec, sample=sample, html_only=html_only))167    outcomes = [(r, spec.judge(r)) for r in rows]168    accepted = sum(1 for _, o in outcomes if o.status == "accept")169    normalised = [(r, o) for r, o in outcomes if o.status == "normalise"]170    rejected = [(r, o) for r, o in outcomes if o.status == "reject"]171    n = len(rows)172    lo, hi = _interval(accepted + len(normalised), n)173    print(f"\n== {spec.name}: sampled {n} · accepted {accepted} · normalised {len(normalised)} · rejected {len(rejected)}")174    if n:175        print(f"   precision estimate (accepted + normalisable) = {(accepted + len(normalised)) / n:.1%}  [{lo:.1%}, {hi:.1%}]"176              f"   rejected share = {len(rejected) / n:.1%}")177    reasons: Counter[str] = Counter(o.reason or "?" for _, o in rejected)178    by_reason: dict[str, list[str]] = defaultdict(list)179    for r, o in rejected:180        by_reason[o.reason or "?"].append(spec.label(r))181    for reason, count in reasons.most_common():182        print(f"   - {reason}: {count}")183        for lbl in by_reason[reason][:show]:184            print(f"       ✗ {lbl[:LABEL_WIDTH]}")185    if normalised:186        print(f"   ~ would be normalised ({len(normalised)}):")187        for r, o in normalised[:show]:188            print(f"       ~ {spec.label(r)[:LABEL_WIDTH]}  →  {o.fix}")189    return [r["id"] for r, _ in rejected], [r["id"] for r, o in normalised if o.fix == "title → null"]190191192async def _purge(conn: Any, spec: TableSpec, reject_ids: list[str], null_title_ids: list[str]) -> None:193    for i in range(0, len(reject_ids), PURGE_BATCH):194        await execute(conn, f"delete from {spec.name} where id = any(cast(:ids as text[]))", ids=reject_ids[i:i + PURGE_BATCH])195    if spec.name == "people":196        for i in range(0, len(null_title_ids), PURGE_BATCH):197            await execute(conn, "update people set title = null where id = any(cast(:ids as text[]))", ids=null_title_ids[i:i + PURGE_BATCH])198    print(f"   purged {len(reject_ids)} row(s) from {spec.name}" + (f", cleared {len(null_title_ids)} title(s)" if spec.name == "people" else ""))199200201async def main(args: argparse.Namespace) -> None:202    names = [t.strip() for t in args.table.split(",")] if args.table else list(TABLES)203    unknown = [t for t in names if t not in TABLES]204    if unknown:205        raise SystemExit(f"unknown table(s): {unknown}; choose from {list(TABLES)}")206    if args.purge:207        names = [t for t in names if TABLES[t].purgeable]208        print(f"PURGE mode: full scan of {names}; rows failing the precision rules will be deleted (news and ATS jobs are never touched)")209    sample = None if args.purge else args.sample210    html_only = args.purge or not args.all_sources                                 # a purge is always scoped to HTML-sourced rows211    print(f"precision rules {P.PRECISION_VERSION} · sample={'all' if sample is None else sample} · seed={args.seed} · "212          f"scope={'HTML-sourced rows (generic-html-v1)' if html_only else 'all sources'}")213    async with transaction() as conn:214        await execute(conn, "select setseed(:s)", s=args.seed)215        for name in names:216            spec = TABLES[name]217            reject_ids, null_title_ids = await _audit_table(conn, spec, sample=sample, show=args.show, html_only=html_only)218            if args.purge:219                await _purge(conn, spec, reject_ids, null_title_ids)220    print("\ndone" if args.purge else "\nread-only audit done (pass --purge to delete the rejected rows of the purgeable tables)")221222223def _parse() -> argparse.Namespace:224    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])225    ap.add_argument("--sample", type=int, default=DEFAULT_SAMPLE, help=f"rows sampled per table (default {DEFAULT_SAMPLE})")226    ap.add_argument("--table", type=str, default="", help=f"comma-separated subset of {list(TABLES)}")227    ap.add_argument("--show", type=int, default=DEFAULT_SHOW, help="examples printed per reason")228    ap.add_argument("--seed", type=float, default=DEFAULT_SEED, help="Postgres setseed() value in [-1, 1] for reproducible samples")229    ap.add_argument("--purge", action="store_true", help=f"delete rows failing the rules in {PURGEABLE} (full scan, not a sample, HTML-sourced only)")230    ap.add_argument("--all-sources", action="store_true", help="read-only audit over every source (enrichment rows included); ignored with --purge")231    return ap.parse_args()232233234async def _run() -> None:235    try:236        await main(_parse())237    finally:238        await dispose()239240241if __name__ == "__main__":242    asyncio.run(_run())243