diff --git a/devplacepy/services/jobs/deepsearch/crawl.py b/devplacepy/services/jobs/deepsearch/crawl.py index ca963031..74d917f3 100644 --- a/devplacepy/services/jobs/deepsearch/crawl.py +++ b/devplacepy/services/jobs/deepsearch/crawl.py @@ -302,99 +302,102 @@ async def crawl( total = min(len(level_candidates), max_pages) cancelled = False + pw = None browser = None - async with async_playwright() as pw: - try: - browser = await pw.chromium.launch( - headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"] - ) - except Exception as exc: - logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc) + try: + pw = await async_playwright().start() + browser = await pw.chromium.launch( + headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"] + ) + except Exception as exc: + logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc) - try: - for level in range(max(1, depth)): - if cancelled or fetched >= max_pages or not level_candidates: + try: + for level in range(max(1, depth)): + if cancelled or fetched >= max_pages or not level_candidates: + break + next_candidates: list[dict] = [] + for start in range(0, len(level_candidates), CRAWL_CONCURRENCY): + if fetched >= max_pages: break - next_candidates: list[dict] = [] - for start in range(0, len(level_candidates), CRAWL_CONCURRENCY): + if await should_stop(): + emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"}) + cancelled = True + break + batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched] + for candidate in batch: + emit( + { + "type": "progress", + "done": fetched, + "total": total, + "url": candidate["url"], + "depth": level, + "message": f"Reading {candidate['url']}", + } + ) + if is_cached(candidate["url"]): + emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"}) + fetch_start = time.perf_counter() + results = await asyncio.gather( + *(_resolve_candidate(candidate, level, browser) for candidate in batch), + return_exceptions=True, + ) + elapsed_ms = int((time.perf_counter() - fetch_start) * 1000) + for candidate, page in zip(batch, results): + url = candidate["url"] + if isinstance(page, BaseException): + logger.info("deepsearch fetch crashed for %s: %s", url, page) + page = None + if page is None: + emit( + { + "type": "page_skipped", + "url": url, + "reason": "no readable content", + "elapsed_ms": elapsed_ms, + } + ) + continue if fetched >= max_pages: break - if await should_stop(): - emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"}) - cancelled = True - break - batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched] - for candidate in batch: + digest = content_hash(page.text) + if digest in outcome.seen_hashes: emit( { - "type": "progress", - "done": fetched, - "total": total, - "url": candidate["url"], - "depth": level, - "message": f"Reading {candidate['url']}", - } - ) - if is_cached(candidate["url"]): - emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"}) - fetch_start = time.perf_counter() - results = await asyncio.gather( - *(_resolve_candidate(candidate, level, browser) for candidate in batch), - return_exceptions=True, - ) - elapsed_ms = int((time.perf_counter() - fetch_start) * 1000) - for candidate, page in zip(batch, results): - url = candidate["url"] - if isinstance(page, BaseException): - logger.info("deepsearch fetch crashed for %s: %s", url, page) - page = None - if page is None: - emit( - { - "type": "page_skipped", - "url": url, - "reason": "no readable content", - "elapsed_ms": elapsed_ms, - } - ) - continue - if fetched >= max_pages: - break - digest = content_hash(page.text) - if digest in outcome.seen_hashes: - emit( - { - "type": "page_duplicate", - "url": url, - "reason": "duplicate content", - "elapsed_ms": elapsed_ms, - } - ) - continue - outcome.seen_hashes.add(digest) - outcome.pages.append(page) - fetched += 1 - emit( - { - "type": "page_loaded", - "url": page.url, - "title": page.title, - "source": page.source, - "depth": level, - "render": page.source == "playwright", + "type": "page_duplicate", + "url": url, + "reason": "duplicate content", "elapsed_ms": elapsed_ms, - "done": fetched, - "total": total, } ) - if level + 1 < depth: - for link in relevant_links(page.links, query, LINKS_PER_PAGE): - if link not in seen_urls: - seen_urls.add(link) - next_candidates.append({"url": link}) - level_candidates = next_candidates - total = min(total + len(next_candidates), max_pages) - finally: - if browser is not None: - await browser.close() + continue + outcome.seen_hashes.add(digest) + outcome.pages.append(page) + fetched += 1 + emit( + { + "type": "page_loaded", + "url": page.url, + "title": page.title, + "source": page.source, + "depth": level, + "render": page.source == "playwright", + "elapsed_ms": elapsed_ms, + "done": fetched, + "total": total, + } + ) + if level + 1 < depth: + for link in relevant_links(page.links, query, LINKS_PER_PAGE): + if link not in seen_urls: + seen_urls.add(link) + next_candidates.append({"url": link}) + level_candidates = next_candidates + total = min(total + len(next_candidates), max_pages) + finally: + if browser is not None: + await browser.close() + if pw is not None: + await pw.stop() return outcome diff --git a/tests/unit/services/jobs/deepsearch/crawl.py b/tests/unit/services/jobs/deepsearch/crawl.py index cbe38353..949b9d11 100644 --- a/tests/unit/services/jobs/deepsearch/crawl.py +++ b/tests/unit/services/jobs/deepsearch/crawl.py @@ -110,3 +110,28 @@ def test_crawl_falls_back_to_snippet_when_fetch_empty(monkeypatch): ) assert len(outcome.pages) == 1 assert outcome.pages[0].source == "search" + + +def test_crawl_degrades_to_httpx_when_the_playwright_driver_cannot_start(monkeypatch): + import playwright.async_api as playwright_api + + class FailingDriver: + async def start(self): + raise RuntimeError("driver missing") + + monkeypatch.setattr(playwright_api, "async_playwright", lambda: FailingDriver()) + + async def fake_fetch(url, depth, browser=None): + return CrawledPage( + url=url, title="T", text="body " * 100, source="httpx", status=200, depth=depth + ) + + monkeypatch.setattr(crawl_module, "fetch_page", fake_fetch) + candidates = [ + {"url": "https://example.com/a", "title": "A", "description": "", "content": ""} + ] + outcome = run_async( + crawl(candidates, 6, lambda f: None, lambda u: False, _no_stop, query="q", depth=1) + ) + assert len(outcome.pages) == 1 + assert outcome.pages[0].source == "httpx"