"""`catlas` crawl commands: onboard · discover · run-sensor · schedule · sensors · connectors · repair · stats-crawl.""" from __future__ import annotations import json from typing import Annotated, Any import typer from rich.table import Table from companyatlas.cli import console, out, run_async def _emit(obj: Any) -> None: """Machine-readable line on stdout (never wrapped by the console).""" out.print(json.dumps(obj, default=str), soft_wrap=True, highlight=False, markup=False) def _table(title: str, columns: list[str], rows: list[list[Any]]) -> None: t = Table(title=title, show_lines=False, header_style="bold") for c in columns: t.add_column(c, overflow="fold") for r in rows: t.add_row(*[("" if v is None else str(v)) for v in r]) console.print(t) def register(app: typer.Typer) -> None: @app.command() 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, concurrency: int | None = None, fetch_now: bool = False, dry_run: bool = False) -> None: """Discover surfaces and create sensors for pending companies (or one company).""" from companyatlas.services.discovery import onboard_pending stats = run_async(onboard_pending(limit, concurrency, company_slug=company, fetch_now=fetch_now, dry_run=dry_run)) console.print(f"[bold]onboarding[/] claimed={stats.get('claimed', 0)} active={stats.get('active', 0)} failed={stats.get('failed', 0)} " f"no_website={stats.get('no_website', 0)} sensors={stats.get('sensors', 0)}") _emit(stats) @app.command() def discover(target: Annotated[str, typer.Argument(help="website URL or company slug")], dry_run: bool = True, json_out: Annotated[bool, typer.Option("--json")] = False) -> None: """Run discovery for a website or an existing company and print the surface table (dry-run by default).""" from companyatlas.db import fetch_one, transaction from companyatlas.fetch import Fetcher from companyatlas.ids import new_id, slugify from companyatlas.services.discovery import discover_company from companyatlas.urls import registrable_domain async def go() -> Any: async with transaction() as conn: company = await fetch_one(conn, "select * from companies where slug = :t or id = :t or canonical_domain = :t", t=target) if company is None: if "." not in target: raise typer.BadParameter(f"{target!r} is neither a known company nor a website") website = target if target.startswith("http") else f"https://{target}" dom = registrable_domain(website) company = {"id": new_id("company"), "slug": slugify(dom.split(".")[0]), "display_name": dom, "canonical_domain": dom, "website": website, "tier": 3, "importance": 0.3} if not dry_run: raise typer.BadParameter("persisting discovery for an unknown website requires a seeded company — use --dry-run or seed it first") async with Fetcher() as fetcher: return await discover_company(company, fetcher=fetcher, dry_run=dry_run) res = run_async(go()) 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()] _table(f"{res.canonical_domain} — {len(rows)} sensors · status={res.status} · {res.requests} requests · {res.duration_ms} ms", ["surface", "url", "connector", "conf", "method", "tier", "every (min)", "quality"], rows) if res.notes or res.error: console.print(f"[dim]notes: {'; '.join(res.notes)}[/]" + (f" [red]error: {res.error}[/]" if res.error else "")) if res.ats: console.print(f"[dim]ATS: {res.ats}[/]") if json_out: _emit({"status": res.status, "canonical_domain": res.canonical_domain, "final_url": res.final_url, "sensors": res.table(), "ats": res.ats, "notes": res.notes, "error": res.error, "requests": res.requests, "duration_ms": res.duration_ms}) @app.command("run-sensor") 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, force: bool = False) -> None: """Run one sensor now (optionally against a local file) and print the outcome.""" from companyatlas.fetch import Fetcher from companyatlas.services.pipeline import load_sensor, run_sensor, run_sensor_by_url_with_file async def go() -> Any: if file: return await run_sensor_by_url_with_file(ref, file, force=force) row = await load_sensor(ref) if row is None: raise typer.BadParameter(f"sensor {ref!r} not found") async with Fetcher() as fetcher: return await run_sensor(row, fetcher=fetcher, worker="cli", force=force) o = run_async(go()) console.print(f"[bold]{o.status}[/] sensor={o.sensor_id} kind={o.kind} significance={o.significance} snapshot={o.snapshot_id} change={o.change_id} " f"delta={o.delta_counts} interval={o.interval_s}s next={o.next_run_at} failure={o.failure_class} {o.error or ''}") _emit({"status": o.status, "sensor_id": o.sensor_id, "observation_id": o.observation_id, "snapshot_id": o.snapshot_id, "change_id": o.change_id, "kind": o.kind, "significance": o.significance, "failure_class": o.failure_class, "error": o.error, "delta": o.delta_counts, "interval_s": o.interval_s, "next_run_at": o.next_run_at, "duration_ms": o.duration_ms}) @app.command() def schedule(concurrency: int | None = None, no_onboarding: bool = False, once: bool = False, worker: str | None = None) -> None: """Run the crawl scheduler: claim due sensors, run them, onboarding worker, periodic tasks. Ctrl-C to stop.""" from companyatlas.services.scheduler import run_scheduler run_async(run_scheduler(concurrency=concurrency, onboarding=not no_onboarding, once=once, worker=worker)) @app.command() def sensors(company: str | None = None, status: str | None = None, surface: str | None = None, limit: int = 50, json_out: Annotated[bool, typer.Option("--json")] = False) -> None: """List sensors (filters: --company slug, --status, --surface).""" from companyatlas.db import fetch_all, transaction async def go() -> list[dict[str, Any]]: async with transaction() as conn: return await fetch_all(conn, """ 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, s.consecutive_failures, s.consecutive_unchanged, s.snapshot_count, s.change_count, s.meaningful_change_count, s.last_failure_class from sensors s join companies c on c.id = s.company_id where (:company is null or c.slug = :company or c.id = :company or c.canonical_domain = :company) and (:status is null or s.status = :status) and (:surface is null or s.surface = :surface) order by c.slug, s.surface limit :limit """, company=company, status=status, surface=surface, limit=limit) rows = run_async(go()) _table(f"{len(rows)} sensors", ["id", "company", "surface", "connector", "status", "tier", "every", "next", "quality", "fail", "unchg", "snaps", "chg", "mean", "url"], [[r["id"], r["slug"], r["surface"], r["connector_id"], r["status"], r["tier"], f"{r['current_interval_s'] // 60}m", r["next_run_at"].strftime("%m-%d %H:%M") if r["next_run_at"] else "", r["quality_score"], r["consecutive_failures"], r["consecutive_unchanged"], r["snapshot_count"], r["change_count"], r["meaningful_change_count"], r["url"]] for r in rows]) if json_out: _emit(rows) @app.command("connectors") def connectors_cmd(sync: bool = True) -> None: """List registered connector families and sync the `connectors` table.""" from companyatlas.db import transaction from companyatlas.sdk.connector import all_connectors, sync_connectors_table async def go() -> int: if not sync: return 0 async with transaction() as conn: return await sync_connectors_table(conn) n = run_async(go()) _table(f"{len(all_connectors())} connectors (synced {n})", ["id", "name", "category", "fetch", "discovery", "incremental", "default interval", "url pattern"], [[c.meta.connector_id, c.meta.name, c.meta.category, c.meta.fetch_mode, c.meta.supports_discovery, c.meta.supports_incremental, f"{c.meta.default_interval_s // 3600}h", c.meta.url_pattern or ""] for c in all_connectors()]) @app.command() def repair(limit: int = 50, dry_run: bool = False) -> None: """Auto-repair failing / stale / redirected sensors (retry → sitemap → navigation → identity check → migrate or review).""" from companyatlas.services.repair import repair_batch stats = run_async(repair_batch(limit, dry_run=dry_run)) _table(f"repair: examined={stats['examined']} recovered={stats.get('recovered', 0)} migrated={stats.get('migrated', 0)} review={stats.get('review', 0)}", ["sensor", "surface", "action", "url", "new url", "requests"], [[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"]]) _emit({k: v for k, v in stats.items() if k != "results"}) @app.command("stats-crawl") def stats_crawl() -> None: """Quick crawl counters: companies by onboarding status, sensors by status/tier, observations/changes today, failures, heartbeat.""" from companyatlas.db import fetch_all, fetch_one, transaction async def go() -> dict[str, Any]: async with transaction() as conn: return { "companies": await fetch_all(conn, "select onboarding_status as k, count(*) as n from companies group by 1 order by 1"), "sensors_status": await fetch_all(conn, "select status as k, count(*) as n from sensors group by 1 order by 1"), "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"), "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"), "today": await fetch_one(conn, """select (select count(*) from observations where fetched_at >= current_date) as observations, (select count(*) from observations where fetched_at >= current_date and not_modified) as not_modified, (select count(*) from observations where fetched_at >= current_date and failure_class is not null) as failed, (select count(*) from snapshots where fetched_at >= current_date) as snapshots, (select count(*) from changes where detected_at >= current_date) as changes, (select count(*) from changes where detected_at >= current_date and kind in ('meaningful','major','critical')) as meaningful, (select count(*) from sensors where status in ('active','failing','pending') and next_run_at <= now()) as due, (select count(*) from jobs where status = 'open') as jobs_open, (select count(*) from review_queue where status = 'open') as review_open"""), "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"), "heartbeat": await fetch_one(conn, "select value, updated_at from settings_kv where key = 'scheduler:heartbeat'"), } s = run_async(go()) _table("companies", ["onboarding_status", "n"], [[r["k"], r["n"]] for r in s["companies"]]) _table("sensors by status", ["status", "n"], [[r["k"], r["n"]] for r in s["sensors_status"]]) _table("sensors by tier", ["tier", "n"], [[r["k"], r["n"]] for r in s["sensors_tier"]]) _table("surfaces", ["surface", "n"], [[r["k"], r["n"]] for r in s["surfaces"]]) _table("failures 24h", ["class", "n"], [[r["k"], r["n"]] for r in s["failures"]]) today = s["today"] or {} console.print("[bold]today[/] " + " ".join(f"{k}={v}" for k, v in today.items())) hb = s["heartbeat"] if hb: v = hb["value"] 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')}") _emit(s)