spb/company-atlas

1"""`catlas` crawl commands: onboard · discover · run-sensor · schedule · sensors · connectors · repair · stats-crawl."""2from __future__ import annotations34import json5from typing import Annotated, Any67import typer8from rich.table import Table910from companyatlas.cli import console, out, run_async111213def _emit(obj: Any) -> None:14    """Machine-readable line on stdout (never wrapped by the console)."""15    out.print(json.dumps(obj, default=str), soft_wrap=True, highlight=False, markup=False)161718def _table(title: str, columns: list[str], rows: list[list[Any]]) -> None:19    t = Table(title=title, show_lines=False, header_style="bold")20    for c in columns:21        t.add_column(c, overflow="fold")22    for r in rows:23        t.add_row(*[("" if v is None else str(v)) for v in r])24    console.print(t)252627def register(app: typer.Typer) -> None:28    @app.command()29    def onboard(limit: Annotated[int, typer.Option(help="max companies this run (0 = all pending)")] = 0, company: Annotated[str | None, typer.Option(help="slug / id / domain of one company")] = None,30                concurrency: int | None = None, fetch_now: bool = False, dry_run: bool = False) -> None:31        """Discover surfaces and create sensors for pending companies (or one company)."""32        from companyatlas.services.discovery import onboard_pending3334        stats = run_async(onboard_pending(limit, concurrency, company_slug=company, fetch_now=fetch_now, dry_run=dry_run))35        console.print(f"[bold]onboarding[/] claimed={stats.get('claimed', 0)} active={stats.get('active', 0)} failed={stats.get('failed', 0)} "36                      f"no_website={stats.get('no_website', 0)} sensors={stats.get('sensors', 0)}")37        _emit(stats)3839    @app.command()40    def discover(target: Annotated[str, typer.Argument(help="website URL or company slug")], dry_run: bool = True,41                 json_out: Annotated[bool, typer.Option("--json")] = False) -> None:42        """Run discovery for a website or an existing company and print the surface table (dry-run by default)."""43        from companyatlas.db import fetch_one, transaction44        from companyatlas.fetch import Fetcher45        from companyatlas.ids import new_id, slugify46        from companyatlas.services.discovery import discover_company47        from companyatlas.urls import registrable_domain4849        async def go() -> Any:50            async with transaction() as conn:51                company = await fetch_one(conn, "select * from companies where slug = :t or id = :t or canonical_domain = :t", t=target)52            if company is None:53                if "." not in target:54                    raise typer.BadParameter(f"{target!r} is neither a known company nor a website")55                website = target if target.startswith("http") else f"https://{target}"56                dom = registrable_domain(website)57                company = {"id": new_id("company"), "slug": slugify(dom.split(".")[0]), "display_name": dom, "canonical_domain": dom, "website": website, "tier": 3,58                           "importance": 0.3}59                if not dry_run:60                    raise typer.BadParameter("persisting discovery for an unknown website requires a seeded company — use --dry-run or seed it first")61            async with Fetcher() as fetcher:62                return await discover_company(company, fetcher=fetcher, dry_run=dry_run)6364        res = run_async(go())65        rows = [[s["surface"], s["url"], s["connector"], f"{s['confidence']:.2f}", s["method"], s["tier"], s["interval_s"] // 60, s["quality"]] for s in res.table()]66        _table(f"{res.canonical_domain} — {len(rows)} sensors · status={res.status} · {res.requests} requests · {res.duration_ms} ms",67               ["surface", "url", "connector", "conf", "method", "tier", "every (min)", "quality"], rows)68        if res.notes or res.error:69            console.print(f"[dim]notes: {'; '.join(res.notes)}[/]" + (f" [red]error: {res.error}[/]" if res.error else ""))70        if res.ats:71            console.print(f"[dim]ATS: {res.ats}[/]")72        if json_out:73            _emit({"status": res.status, "canonical_domain": res.canonical_domain, "final_url": res.final_url, "sensors": res.table(), "ats": res.ats,74                                  "notes": res.notes, "error": res.error, "requests": res.requests, "duration_ms": res.duration_ms})7576    @app.command("run-sensor")77    def run_sensor_cmd(ref: Annotated[str, typer.Argument(help="sensor id or URL")], file: Annotated[str | None, typer.Option("--file", help="fixture file instead of the network")] = None,78                       force: bool = False) -> None:79        """Run one sensor now (optionally against a local file) and print the outcome."""80        from companyatlas.fetch import Fetcher81        from companyatlas.services.pipeline import load_sensor, run_sensor, run_sensor_by_url_with_file8283        async def go() -> Any:84            if file:85                return await run_sensor_by_url_with_file(ref, file, force=force)86            row = await load_sensor(ref)87            if row is None:88                raise typer.BadParameter(f"sensor {ref!r} not found")89            async with Fetcher() as fetcher:90                return await run_sensor(row, fetcher=fetcher, worker="cli", force=force)9192        o = run_async(go())93        console.print(f"[bold]{o.status}[/] sensor={o.sensor_id} kind={o.kind} significance={o.significance} snapshot={o.snapshot_id} change={o.change_id} "94                      f"delta={o.delta_counts} interval={o.interval_s}s next={o.next_run_at} failure={o.failure_class} {o.error or ''}")95        _emit({"status": o.status, "sensor_id": o.sensor_id, "observation_id": o.observation_id, "snapshot_id": o.snapshot_id, "change_id": o.change_id,96                              "kind": o.kind, "significance": o.significance, "failure_class": o.failure_class, "error": o.error, "delta": o.delta_counts,97                              "interval_s": o.interval_s, "next_run_at": o.next_run_at, "duration_ms": o.duration_ms})9899    @app.command()100    def schedule(concurrency: int | None = None, no_onboarding: bool = False, once: bool = False, worker: str | None = None) -> None:101        """Run the crawl scheduler: claim due sensors, run them, onboarding worker, periodic tasks. Ctrl-C to stop."""102        from companyatlas.services.scheduler import run_scheduler103104        run_async(run_scheduler(concurrency=concurrency, onboarding=not no_onboarding, once=once, worker=worker))105106    @app.command()107    def sensors(company: str | None = None, status: str | None = None, surface: str | None = None, limit: int = 50,108                json_out: Annotated[bool, typer.Option("--json")] = False) -> None:109        """List sensors (filters: --company slug, --status, --surface)."""110        from companyatlas.db import fetch_all, transaction111112        async def go() -> list[dict[str, Any]]:113            async with transaction() as conn:114                return await fetch_all(conn, """115                    select s.id, c.slug, s.surface, s.connector_id, s.url, s.status, s.tier, s.current_interval_s, s.next_run_at, s.last_run_at, s.quality_score,116                           s.consecutive_failures, s.consecutive_unchanged, s.snapshot_count, s.change_count, s.meaningful_change_count, s.last_failure_class117                    from sensors s join companies c on c.id = s.company_id118                    where (:company is null or c.slug = :company or c.id = :company or c.canonical_domain = :company)119                      and (:status is null or s.status = :status) and (:surface is null or s.surface = :surface)120                    order by c.slug, s.surface limit :limit121                """, company=company, status=status, surface=surface, limit=limit)122123        rows = run_async(go())124        _table(f"{len(rows)} sensors", ["id", "company", "surface", "connector", "status", "tier", "every", "next", "quality", "fail", "unchg", "snaps", "chg", "mean", "url"],125               [[r["id"], r["slug"], r["surface"], r["connector_id"], r["status"], r["tier"], f"{r['current_interval_s'] // 60}m",126                 r["next_run_at"].strftime("%m-%d %H:%M") if r["next_run_at"] else "", r["quality_score"], r["consecutive_failures"], r["consecutive_unchanged"],127                 r["snapshot_count"], r["change_count"], r["meaningful_change_count"], r["url"]] for r in rows])128        if json_out:129            _emit(rows)130131    @app.command("connectors")132    def connectors_cmd(sync: bool = True) -> None:133        """List registered connector families and sync the `connectors` table."""134        from companyatlas.db import transaction135        from companyatlas.sdk.connector import all_connectors, sync_connectors_table136137        async def go() -> int:138            if not sync:139                return 0140            async with transaction() as conn:141                return await sync_connectors_table(conn)142143        n = run_async(go())144        _table(f"{len(all_connectors())} connectors (synced {n})", ["id", "name", "category", "fetch", "discovery", "incremental", "default interval", "url pattern"],145               [[c.meta.connector_id, c.meta.name, c.meta.category, c.meta.fetch_mode, c.meta.supports_discovery, c.meta.supports_incremental,146                 f"{c.meta.default_interval_s // 3600}h", c.meta.url_pattern or ""] for c in all_connectors()])147148    @app.command()149    def repair(limit: int = 50, dry_run: bool = False) -> None:150        """Auto-repair failing / stale / redirected sensors (retry → sitemap → navigation → identity check → migrate or review)."""151        from companyatlas.services.repair import repair_batch152153        stats = run_async(repair_batch(limit, dry_run=dry_run))154        _table(f"repair: examined={stats['examined']} recovered={stats.get('recovered', 0)} migrated={stats.get('migrated', 0)} review={stats.get('review', 0)}",155               ["sensor", "surface", "action", "url", "new url", "requests"],156               [[r.get("sensor_id"), r.get("surface"), r.get("action"), r.get("url"), r.get("new_url", ""), r.get("requests", 0)] for r in stats["results"]])157        _emit({k: v for k, v in stats.items() if k != "results"})158159    @app.command("stats-crawl")160    def stats_crawl() -> None:161        """Quick crawl counters: companies by onboarding status, sensors by status/tier, observations/changes today, failures, heartbeat."""162        from companyatlas.db import fetch_all, fetch_one, transaction163164        async def go() -> dict[str, Any]:165            async with transaction() as conn:166                return {167                    "companies": await fetch_all(conn, "select onboarding_status as k, count(*) as n from companies group by 1 order by 1"),168                    "sensors_status": await fetch_all(conn, "select status as k, count(*) as n from sensors group by 1 order by 1"),169                    "sensors_tier": await fetch_all(conn, "select tier as k, count(*) as n from sensors where status in ('active','pending','failing') group by 1 order by 1"),170                    "surfaces": await fetch_all(conn, "select surface as k, count(*) as n from sensors where status <> 'retired' group by 1 order by 2 desc limit 40"),171                    "today": await fetch_one(conn, """select (select count(*) from observations where fetched_at >= current_date) as observations,172                                                             (select count(*) from observations where fetched_at >= current_date and not_modified) as not_modified,173                                                             (select count(*) from observations where fetched_at >= current_date and failure_class is not null) as failed,174                                                             (select count(*) from snapshots where fetched_at >= current_date) as snapshots,175                                                             (select count(*) from changes where detected_at >= current_date) as changes,176                                                             (select count(*) from changes where detected_at >= current_date and kind in ('meaningful','major','critical')) as meaningful,177                                                             (select count(*) from sensors where status in ('active','failing','pending') and next_run_at <= now()) as due,178                                                             (select count(*) from jobs where status = 'open') as jobs_open,179                                                             (select count(*) from review_queue where status = 'open') as review_open"""),180                    "failures": await fetch_all(conn, "select failure_class as k, count(*) as n from failures where at >= now() - interval '24 hours' group by 1 order by 2 desc"),181                    "heartbeat": await fetch_one(conn, "select value, updated_at from settings_kv where key = 'scheduler:heartbeat'"),182                }183184        s = run_async(go())185        _table("companies", ["onboarding_status", "n"], [[r["k"], r["n"]] for r in s["companies"]])186        _table("sensors by status", ["status", "n"], [[r["k"], r["n"]] for r in s["sensors_status"]])187        _table("sensors by tier", ["tier", "n"], [[r["k"], r["n"]] for r in s["sensors_tier"]])188        _table("surfaces", ["surface", "n"], [[r["k"], r["n"]] for r in s["surfaces"]])189        _table("failures 24h", ["class", "n"], [[r["k"], r["n"]] for r in s["failures"]])190        today = s["today"] or {}191        console.print("[bold]today[/] " + " ".join(f"{k}={v}" for k, v in today.items()))192        hb = s["heartbeat"]193        if hb:194            v = hb["value"]195            console.print(f"[bold]heartbeat[/] worker={v.get('worker')} at={v.get('at')} inflight={v.get('inflight')} due={v.get('due')} tick_ms={v.get('tick_ms')}")196        _emit(s)197