From d2ceba201e0dd4ff70ffc20ababb733eaeccfaee Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 28 Apr 2026 21:05:45 -0400 Subject: [PATCH] crawler: live stderr heartbeat during BFS + ingest (rate-limited) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A no-cap crawl could take many seconds with no feedback. Wires the existing aborist.progress.Progress into both phases so stderr shows heartbeats every 2s (Progress's default interval). Discovery phase: prints prefix='crawl', counts discovered URLs, shows queue depth as the secondary number ('inserted' slot in the Progress format — works fine, semantically "still to do"). Ingest phase: prefix='ingest', total_estimate=len(urls) so the user sees percent + ETA. Threads through ingest_source's existing progress= parameter. Plus three banner lines to stderr at phase boundaries (start crawl, end discovery, start ingest) so even sub-2s crawls show signs of life. All flushed via Progress's flush=True path; stderr is line-buffered by default so this works without PYTHONUNBUFFERED. Tests stay green (Progress goes to stderr, pytest captures only the stdout summary). 19 bridge tests + 273 default suite, all passing. --- aborist/cli.py | 30 +++++++++++++++++++++++++++--- aborist/sources/crawler/bridge.py | 28 +++++++++++++++++++++++++--- 2 files changed, 52 insertions(+), 6 deletions(-) diff --git a/aborist/cli.py b/aborist/cli.py index 28c7e77..04f9c31 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -2120,17 +2120,41 @@ def _cmd_crawl(args: argparse.Namespace) -> int: print(f"error: {e}", file=sys.stderr) return 2 - urls = crawl_seed( - args.seed_url, max_depth=args.depth, max_pages=args.max_pages + from aborist.progress import Progress + + cap = "no cap" if args.max_pages == 0 else f"max {args.max_pages}" + print( + f" crawl: seed={args.seed_url} depth={args.depth} {cap}", + file=sys.stderr, + flush=True, ) + crawl_progress = Progress(prefix="crawl ") + urls = crawl_seed( + args.seed_url, + max_depth=args.depth, + max_pages=args.max_pages, + progress=crawl_progress, + ) + print( + f" crawl: discovery done — {len(urls)} URLs", + file=sys.stderr, + flush=True, + ) + if not args.ingest: for u in urls: print(u) return 0 + print( + f" ingest: starting on {len(urls)} URLs", + file=sys.stderr, + flush=True, + ) + ingest_progress = Progress(prefix="ingest ", total_estimate=len(urls)) conn = connect(args.db) try: - result = ingest_crawled(conn, urls) + result = ingest_crawled(conn, urls, progress=ingest_progress) finally: conn.close() print(json.dumps( diff --git a/aborist/sources/crawler/bridge.py b/aborist/sources/crawler/bridge.py index 7101554..922e75f 100644 --- a/aborist/sources/crawler/bridge.py +++ b/aborist/sources/crawler/bridge.py @@ -30,6 +30,7 @@ import httpx from aborist.document import Document from aborist.ingest import ingest_source +from aborist.progress import Progress from aborist.sources.crawler.async_web_fetcher import AsyncWebFetcher from aborist.sources.html_page import USER_AGENT, parse_html from aborist.store import transaction @@ -53,12 +54,18 @@ async def _crawl_seed_async( max_depth: int, max_pages: int, fetcher: AsyncWebFetcher | None = None, + progress: Progress | None = None, ) -> list[str]: """BFS from `seed_url`, staying on the same hostname. Discovery only — fetches each page enough to extract links, no aborist-side ingestion happens here. Returns a deduplicated list of URLs in BFS order, capped at `max_pages`. + + Pass ``progress`` (an `aborist.progress.Progress`) for stderr + heartbeats while the BFS runs. Lines are rate-limited by the + Progress instance (default 2s) so a fast crawl prints once or + twice; a slow one prints every interval. """ fetcher = fetcher or AsyncWebFetcher(user_agent=USER_AGENT) seen: set[str] = set() @@ -85,6 +92,8 @@ async def _crawl_seed_async( continue discovered.append(url) + if progress is not None: + progress.tick(len(discovered), inserted=len(queue)) if depth >= max_depth: continue @@ -102,6 +111,8 @@ async def _crawl_seed_async( continue queue.append((normalized, depth + 1)) + if progress is not None: + progress.done(len(discovered), inserted=len(queue)) return discovered @@ -110,13 +121,20 @@ def crawl_seed( *, max_depth: int = 2, max_pages: int = 0, + progress: Progress | None = None, ) -> list[str]: """Sync wrapper around the async crawler. Returns same-domain URLs. - ``max_pages=0`` means "no cap" — depth is the only bound. + ``max_pages=0`` means "no cap" — depth is the only bound. Pass + ``progress`` for stderr heartbeats while BFS runs. """ return asyncio.run( - _crawl_seed_async(seed_url, max_depth=max_depth, max_pages=max_pages) + _crawl_seed_async( + seed_url, + max_depth=max_depth, + max_pages=max_pages, + progress=progress, + ) ) @@ -168,6 +186,7 @@ def ingest_crawled( urls: Iterable[str], *, timeout: float = 30.0, + progress: Progress | None = None, ) -> dict: """Fetch each URL, ingest into `conn`, record ETag + Last-Modified. @@ -177,9 +196,12 @@ def ingest_crawled( the document AND its http_meta or neither (FK guarantees this on cascade delete; the upsert happens inside the same transaction boundary as the audit event). + + Pass ``progress`` for stderr heartbeats while ingest runs (the + underlying ingest_source supports it natively). """ src = _CrawledHtmlSource(urls, timeout=timeout) - stats = ingest_source(conn, src) + stats = ingest_source(conn, src, progress=progress) written: list[dict] = [] if not src.http_meta: return {