From ddd5fa94bdc60e212ecdc062c9b335224458f96e Mon Sep 17 00:00:00 2001 From: JunieXD <1240840076@qq.com> Date: Mon, 9 Nov 2026 23:16:27 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E5=8A=A0=E4=B8=8A=E6=8A=93=E5=8F=96?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E7=9A=84=E5=85=A5=E5=8F=A3=E5=92=8C=E9=95=9C?= =?UTF-8?q?=E5=83=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 抓取、读正文和初筛各步都有了，还差一个能常驻运行的进程把它们串起来，部署时也要一个镜像（#45）。 - `uv run crawler` 启动：启动时用 Pydantic 校验环境变量，缺了或不对就报出是哪几个、不带值，然后退出 - 每分钟领一次活，领到的来源同时各抓一轮；一轮里出意外异常只记日志，不影响其他来源，租约过期后网站会再发 - 收到 SIGTERM 不再领活，等进行中的几轮做完再退出 - 日志每行一个 JSON 对象，一轮的日志都带来源编号和开始时间；普通请求和浏览器下载也各记一条网址、状态码和耗时，httpx 自己的逐条日志关掉 - Dockerfile 按 uv 推荐的两段构建，带 Chromium 无头精简版，用普通用户运行 - AGENTS.md 写上抓取节点的环境变量和本地运行的命令，开发约定更新目录和日志的写法 --- AGENTS.md | 3 +- crawler/.dockerignore | 5 ++ crawler/.env.example | 4 + crawler/.gitignore | 1 + crawler/Dockerfile | 23 ++++++ crawler/pyproject.toml | 3 + crawler/src/crawler/crawl/browser.py | 9 +++ crawler/src/crawler/crawl/fetch.py | 14 +++- crawler/src/crawler/crawl/rounds.py | 4 +- crawler/src/crawler/log.py | 31 ++++++++ crawler/src/crawler/main.py | 112 +++++++++++++++++++++++++++ crawler/src/crawler/main_test.py | 81 +++++++++++++++++++ docs/development.md | 4 +- 13 files changed, 290 insertions(+), 4 deletions(-) create mode 100644 crawler/.dockerignore create mode 100644 crawler/.env.example create mode 100644 crawler/Dockerfile create mode 100644 crawler/src/crawler/log.py create mode 100644 crawler/src/crawler/main.py create mode 100644 crawler/src/crawler/main_test.py diff --git a/AGENTS.md b/AGENTS.md index 8bb29d1..06da5a6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -55,6 +55,7 @@ | `uv run pytest <路径>` | 只跑相关的测试 | | `uv run ruff format` | 格式化 | | `uv run datamodel-codegen` | 从网站的 `openapi/crawler.json` 重新生成接口的类型 `crawl/contract.py` | +| `uv run --env-file .env crawler` | 对着本地的 `pnpm dev` 抓：先 `cp .env.example .env`，填上 `pnpm crawler-token` 生成的令牌；浏览器的装法见[开发约定 §2](docs/development.md#2-抓取和判别) | ## 产品原则 @@ -89,7 +90,7 @@ - **上线前没有历史包袱**：表结构和接口直接改。 - **命名**：叫法见[开发约定 §6](docs/development.md#6-术语)。接口文档也是生成的文件。 - **哈希**：只有两处用哈希：会话、个人令牌和抓取节点令牌这些高熵的随机值，库里只存 SHA-256；鼠洞登录的 PKCE 按 OAuth 的规定算 SHA-256。是不是新通知比网址，正文变没变比文字。 -- **配置**：环境变量现在只有网站的 `DATABASE_URL`、`PUBLIC_URL` 和鼠洞登录用的 `MOUSEHOLE_URL`、`MOUSEHOLE_CLIENT_ID`、`MOUSEHOLE_CLIENT_SECRET`，本地的值见 `web/.env.example`。 +- **配置**：环境变量现在只有网站的 `DATABASE_URL`、`PUBLIC_URL` 和鼠洞登录用的 `MOUSEHOLE_URL`、`MOUSEHOLE_CLIENT_ID`、`MOUSEHOLE_CLIENT_SECRET`，本地的值见 `web/.env.example`；抓取节点的 `MOUSECLOCK_URL`、`MOUSECLOCK_TOKEN` 和初筛用的 `SYSTEM_ONE_API_KEY`，见 `crawler/.env.example`。 ## 可靠与安全 diff --git a/crawler/.dockerignore b/crawler/.dockerignore new file mode 100644 index 0000000..de384a6 --- /dev/null +++ b/crawler/.dockerignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +.env +screening-results/ +extraction-results/ diff --git a/crawler/.env.example b/crawler/.env.example new file mode 100644 index 0000000..b8dd5a5 --- /dev/null +++ b/crawler/.env.example @@ -0,0 +1,4 @@ +# Against the dev server (`pnpm dev` in web/); the token comes from `pnpm crawler-token `. +MOUSECLOCK_URL=http://localhost:5173 +MOUSECLOCK_TOKEN= +SYSTEM_ONE_API_KEY= diff --git a/crawler/.gitignore b/crawler/.gitignore index 9c66874..3267751 100644 --- a/crawler/.gitignore +++ b/crawler/.gitignore @@ -1,3 +1,4 @@ __pycache__/ screening-results/ extraction-results/ +.env diff --git a/crawler/Dockerfile b/crawler/Dockerfile new file mode 100644 index 0000000..910f4aa --- /dev/null +++ b/crawler/Dockerfile @@ -0,0 +1,23 @@ +# The crawler with Chromium's headless shell, built from this directory: `docker build -t crawler .` +# Run it with --init, which reaps the processes Chromium leaves behind, and a stop timeout long +# enough for the rounds in progress to finish. +FROM python:3.14-slim-trixie AS build +COPY --from=ghcr.io/astral-sh/uv:0.13.6 /uv /bin/ +ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_NO_DEV=1 UV_PYTHON_DOWNLOADS=0 +WORKDIR /app +RUN --mount=type=cache,target=/root/.cache/uv \ + --mount=type=bind,source=uv.lock,target=uv.lock \ + --mount=type=bind,source=pyproject.toml,target=pyproject.toml \ + uv sync --locked --no-install-project --no-editable +COPY . . +RUN --mount=type=cache,target=/root/.cache/uv uv sync --locked --no-editable + +FROM python:3.14-slim-trixie +ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright +COPY --from=build /app/.venv /app/.venv +RUN /app/.venv/bin/playwright install --with-deps --only-shell chromium \ + && rm -rf /var/lib/apt/lists/* \ + && useradd --create-home crawler +# The pages it opens are untrusted. +USER crawler +CMD ["/app/.venv/bin/crawler"] diff --git a/crawler/pyproject.toml b/crawler/pyproject.toml index e7c5072..4dc623d 100644 --- a/crawler/pyproject.toml +++ b/crawler/pyproject.toml @@ -15,6 +15,9 @@ dependencies = [ "xlrd>=2.0.2", ] +[project.scripts] +crawler = "crawler.main:main" + [dependency-groups] dev = [ "datamodel-code-generator>=0.86.0", diff --git a/crawler/src/crawler/crawl/browser.py b/crawler/src/crawler/crawl/browser.py index 833bb2f..8c32da7 100644 --- a/crawler/src/crawler/crawl/browser.py +++ b/crawler/src/crawler/crawl/browser.py @@ -136,6 +136,7 @@ def record_response(response: PlaywrightResponse) -> None: async def download(self, url: str, *, page_url: str) -> Response: for _ in range(MAX_REDIRECTS + 1): async with self._owner.fetcher.permit(url): + started = time.monotonic() try: response = await self._context.request.get( url, @@ -146,6 +147,14 @@ async def download(self, url: str, *, page_url: str) -> Response: fail_on_status_code=False, ) try: + logger.info( + "Browser downloaded", + extra={ + "url": url, + "status": response.status, + "elapsed": time.monotonic() - started, + }, + ) location = response.headers.get("location") if response.status in _REDIRECTS and location is not None: url = urljoin(response.url, location) diff --git a/crawler/src/crawler/crawl/fetch.py b/crawler/src/crawler/crawl/fetch.py index 742c017..e93a437 100644 --- a/crawler/src/crawler/crawl/fetch.py +++ b/crawler/src/crawler/crawl/fetch.py @@ -2,6 +2,7 @@ page visit or HTTP request at a time per site, with a pause between them.""" import asyncio +import logging import time from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager @@ -15,6 +16,7 @@ INTERVAL = 5.0 MAX_REDIRECTS = 5 TIMEOUT = 30.0 +logger = logging.getLogger(__name__) type Failure = Literal["timeout", "unreachable", "too_many_redirects"] @@ -90,9 +92,19 @@ async def get(self, url: str) -> Response: async def _send(self, url: str) -> httpx.Response: async with self.permit(url): + started = self._clock() try: - return await self._client.get(url) + response = await self._client.get(url) except httpx.TimeoutException as error: raise FetchError("timeout", url) from error except httpx.TransportError as error: raise FetchError("unreachable", url) from error + logger.info( + "Fetched", + extra={ + "url": url, + "status": response.status_code, + "elapsed": self._clock() - started, + }, + ) + return response diff --git a/crawler/src/crawler/crawl/rounds.py b/crawler/src/crawler/crawl/rounds.py index cf9dc2a..9f2b0f6 100644 --- a/crawler/src/crawler/crawl/rounds.py +++ b/crawler/src/crawler/crawl/rounds.py @@ -27,6 +27,7 @@ ) from crawler.crawl.fetch import FetchError from crawler.crawl.website import Website +from crawler.log import crawl_round from crawler.parse.indexing import Hyperlink logger = logging.getLogger(__name__) @@ -90,13 +91,14 @@ async def crawl_source( clock: Callable[[], float] = time.time, ) -> None: started_at = int(clock()) + crawl_round.set((source.id, started_at)) async with browser.session(source) as pages: try: items = await steps.list_items(pages, source) if not items: raise StepError("no_items") except (FetchError, StepError) as error: - logger.warning("Listing failed: %s", error, extra={"source_id": source.id}) + logger.warning("Listing failed: %s", error) failure, detail = _failure(error) await website.run( Run( diff --git a/crawler/src/crawler/log.py b/crawler/src/crawler/log.py new file mode 100644 index 0000000..dbf1dfa --- /dev/null +++ b/crawler/src/crawler/log.py @@ -0,0 +1,31 @@ +"""Log lines as one JSON object each on stdout, the way the website writes them, so `docker logs` +output can be filtered with jq (development §4).""" + +import json +import logging +from contextvars import ContextVar +from datetime import UTC, datetime + +# The source and start of the round a line comes from, which find its crawl run on the dashboard. +# Each round runs in its own task, so the value stays with that round's lines. +crawl_round: ContextVar[tuple[int, int] | None] = ContextVar("crawl_round", default=None) + +_RECORD_ATTRIBUTES = frozenset(vars(logging.makeLogRecord({}))) | {"message"} + + +class JsonFormatter(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + entry: dict[str, object] = { + "time": datetime.fromtimestamp(record.created, UTC).isoformat(timespec="milliseconds"), + "level": record.levelname.lower(), + "msg": record.getMessage(), + } + if (current := crawl_round.get()) is not None: + entry["source_id"], entry["started_at"] = current + # What the call passed as `extra`. + entry |= { + key: value for key, value in vars(record).items() if key not in _RECORD_ATTRIBUTES + } + if record.exc_info: + entry["stack"] = self.formatException(record.exc_info) + return json.dumps(entry, ensure_ascii=False, default=str) diff --git a/crawler/src/crawler/main.py b/crawler/src/crawler/main.py new file mode 100644 index 0000000..364b74d --- /dev/null +++ b/crawler/src/crawler/main.py @@ -0,0 +1,112 @@ +"""The crawler process: asks the website for work every minute and crawls each source it hands out, +all at once; on SIGTERM it stops asking and exits once the rounds in progress are done.""" + +import asyncio +import logging +import os +import signal +import sys +from collections.abc import Mapping +from contextlib import suppress +from typing import Annotated + +import httpx +from pydantic import BaseModel, ConfigDict, HttpUrl, StringConstraints, ValidationError + +from crawler.crawl.browser import Browser +from crawler.crawl.contract import ExtractedProgram, ExtractionContext, Usage, WorkSource +from crawler.crawl.fetch import Fetcher +from crawler.crawl.listing import list_items +from crawler.crawl.notice import read_notice +from crawler.crawl.rounds import NoticePage, StepError, Steps, crawl_source +from crawler.crawl.website import Website +from crawler.log import JsonFormatter +from crawler.screening.screen import Screen +from crawler.screening.system_one import SystemOne + +# Design §16: once a minute; the dashboard flags the crawler after ten minutes without asking. +POLL = 60.0 +logger = logging.getLogger(__name__) + +type Secret = Annotated[str, StringConstraints(min_length=1)] + + +class Config(BaseModel): + model_config = ConfigDict(alias_generator=str.upper) + + mouseclock_url: HttpUrl + mouseclock_token: Secret + system_one_api_key: Secret + + +def load_config(environ: Mapping[str, str]) -> Config: + try: + return Config.model_validate(environ) + except ValidationError as error: + # Names and reasons only: the values may be secrets. + problems = {str(e["loc"][0]): e["msg"] for e in error.errors(include_input=False)} + logger.error("Configuration invalid", extra={"variables": problems}) + raise SystemExit(1) from None + + +async def serve(website: Website, browser: Browser, steps: Steps, stopping: asyncio.Event) -> None: + # Leaving the group waits for the rounds still running. + async with asyncio.TaskGroup() as rounds: + while True: + try: + work = await website.work() + except httpx.HTTPError as error: + logger.warning("Asking for work failed: %s", error) + else: + for source in work.sources: + rounds.create_task(_crawl(source, browser, website, steps)) + with suppress(TimeoutError): + await asyncio.wait_for(stopping.wait(), POLL) + if stopping.is_set(): + break + + +async def _crawl(source: WorkSource, browser: Browser, website: Website, steps: Steps) -> None: + try: + await crawl_source(source, browser, website, steps) + except Exception: + # A bug or an outage in this round; the others go on, and the lease hands it out again. + logger.exception("Round failed") + + +async def _extract( + _page: NoticePage, _context: ExtractionContext +) -> tuple[list[ExtractedProgram], list[Usage]]: + raise StepError("error", "抽取还没接上") + + +async def _run(config: Config) -> None: + stopping = asyncio.Event() + + def stop() -> None: + logger.info("Stopping once the rounds in progress are done") + stopping.set() + + asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop) + system_one = SystemOne(config.system_one_api_key) + try: + async with ( + Fetcher() as fetcher, + Browser(fetcher) as browser, + Website(str(config.mouseclock_url), config.mouseclock_token) as website, + ): + steps = Steps( + list_items=list_items, screen=Screen(system_one), read=read_notice, extract=_extract + ) + await serve(website, browser, steps, stopping) + finally: + await system_one.close() + + +def main() -> None: + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(JsonFormatter()) + logging.basicConfig(level=logging.INFO, handlers=[handler]) + # Its line per request is noise next to the fetcher's own (development §4). + logging.getLogger("httpx").setLevel(logging.WARNING) + asyncio.run(_run(load_config(os.environ))) diff --git a/crawler/src/crawler/main_test.py b/crawler/src/crawler/main_test.py new file mode 100644 index 0000000..d6d6cf0 --- /dev/null +++ b/crawler/src/crawler/main_test.py @@ -0,0 +1,81 @@ +import asyncio +import io +import json +import logging +from typing import Any, Never + +import httpx +import pytest + +from crawler.crawl.browser import Browser, PageSession +from crawler.crawl.contract import ListedItem, WorkSource +from crawler.crawl.fetch import Fetcher +from crawler.crawl.rounds import StepError, Steps +from crawler.crawl.website import Website +from crawler.log import JsonFormatter +from crawler.main import load_config, serve + + +def test_bad_configuration_is_named_without_its_values(caplog: pytest.LogCaptureFixture) -> None: + environ = {"MOUSECLOCK_URL": "ftp://user:secret@site", "MOUSECLOCK_TOKEN": "token"} + + with pytest.raises(SystemExit): + load_config(environ) + + line = JsonFormatter().format(caplog.records[0]) + assert json.loads(line)["variables"].keys() == {"MOUSECLOCK_URL", "SYSTEM_ONE_API_KEY"} + assert "secret" not in line + + +def test_a_failing_round_leaves_the_others_and_stopping_waits_for_them() -> None: + sources = [ + WorkSource(id=n, url=f"http://a{n}/list.htm", browser=False, config=None, backfill=False) + for n in (1, 2) + ] + calls: list[tuple[str, Any]] = [] + + def handle(request: httpx.Request) -> httpx.Response: + path = request.url.path.removeprefix("/api/v1/crawler") + calls.append((f"{request.method} {path}", request.content and json.loads(request.content))) + if path == "/work": + work = {"sources": [s.model_dump(mode="json") for s in sources], "probes": []} + return httpx.Response(200, json=work) + return httpx.Response(204) + + async def list_items(_pages: PageSession, source: WorkSource) -> list[ListedItem]: + if source.id == 1: + raise RuntimeError("bug") + raise StepError("no_items") + + async def unused(*_: object) -> Never: + raise AssertionError + + async def main() -> None: + stopping = asyncio.Event() + stopping.set() + nowhere = httpx.MockTransport(lambda _: httpx.Response(500)) + async with ( + Fetcher(nowhere) as fetcher, + Website("http://site", "token", httpx.MockTransport(handle)) as website, + ): + await serve( + website, Browser(fetcher), Steps(list_items, unused, unused, unused), stopping + ) + + # Formatted as they are logged, inside the round. + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(JsonFormatter()) + logging.getLogger("crawler").addHandler(handler) + try: + asyncio.run(main()) + finally: + logging.getLogger("crawler").removeHandler(handler) + + assert [call for call, _ in calls] == ["GET /work", "POST /runs"] + assert calls[1][1]["source_id"] == 2 + lines = [json.loads(line) for line in stream.getvalue().splitlines()] + [failed] = [line for line in lines if line["msg"] == "Round failed"] + assert failed["source_id"] == 1 + assert isinstance(failed["started_at"], int) + assert "RuntimeError: bug" in failed["stack"] diff --git a/docs/development.md b/docs/development.md index bf9d70d..1487253 100644 --- a/docs/development.md +++ b/docs/development.md @@ -21,11 +21,13 @@ web/ 网站，写法照鼠洞 test/ Vitest 的全局准备：测试用的数据库；假的鼠洞，开发版也用它 crawler/ 抓取节点 src/crawler/ 一个领域一个包，测试和被测的文件放在一起，命名为 *_test.py + main.py 进程入口：读配置，每分钟领一次活，领到的来源同时各抓一轮 crawl/fetch.py 所有对来源网站的请求都经过这里 crawl/website.py 网站接口的客户端 crawl/rounds.py 一个来源的一轮：列表、初筛、抓正文、抽取、交结果；各步由 Steps 传进来 crawl/contract.py 网站接口的 Pydantic 模型，由 `uv run datamodel-codegen` 从 openapi/crawler.json 生成 check.sh 全部检查 + Dockerfile 抓取节点的镜像，带 Chromium 无头精简版 ``` 模块： @@ -118,7 +120,7 @@ CI 不装浏览器。自动测试用假的浏览器对象、固定响应和假 - 抓取记关键事实：来源和网址、抓取方式、状态码、跳转、耗时、失败类别，以及模型的名称、用量和耗时。每页只写几条，不逐个链接地记。 - 不写正文、附件内容、模型的输入和输出、用户的数据和任何密钥；外部服务返回的错误可能夹带这些内容，只记失败类别。公开网页的网址可以写。 -- 一轮抓取的日志都带同一个抓取记录的编号，方便从抓取看板查到对应的日志。 +- 每行一个 JSON 对象，写法和网站的一样。一轮抓取的日志都带来源编号 `source_id` 和开始时间 `started_at`，和抓取看板上的那一轮对得上。 ## 5. 测试 From e915e2d549fddb55c7f902bb2d8fac6d23044664 Mon Sep 17 00:00:00 2001 From: JunieXD <1240840076@qq.com> Date: Mon, 9 Nov 2026 23:52:20 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E5=88=9D=E7=AD=9B=E5=85=B1=E7=94=A8?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E4=B8=8A=E9=99=90=EF=BC=8C=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E6=97=B6=E8=AE=B0=E8=BF=9B=E6=8A=93=E5=8F=96=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 入口让几十个来源同时抓，原来每轮各自最多 2 个初筛请求，补抓往年时可能同时发出上百个，被限流后这一轮中断，也不交抓取记录，看板上看不到。 - 同时在调用的初筛请求整个进程共用一个上限 - 初筛不成时抛 StepError，这一轮照样往下走：新通知记进没抓成的通知，下一轮再初筛，进行中项目的通知照常抓 - 接口文档写上初筛不成的算没抓成，开发约定写上模型调用的并发上限整个进程共用 --- crawler/src/crawler/crawl/rounds.py | 20 +++++++--- crawler/src/crawler/crawl/rounds_test.py | 26 +++++++++++++ crawler/src/crawler/screening/screen.py | 16 ++++---- crawler/src/crawler/screening/screen_test.py | 39 ++++++++++++++++++-- docs/api.md | 2 +- docs/development.md | 2 +- 6 files changed, 88 insertions(+), 17 deletions(-) diff --git a/crawler/src/crawler/crawl/rounds.py b/crawler/src/crawler/crawl/rounds.py index 9f2b0f6..e8e19c9 100644 --- a/crawler/src/crawler/crawl/rounds.py +++ b/crawler/src/crawler/crawl/rounds.py @@ -118,14 +118,24 @@ async def crawl_source( listing = await website.listing(source.id, Listing(items=items)) new = [notice for notice in listing.pending if notice.step == "screen"] to_fetch = [notice for notice in listing.pending if notice.step == "fetch"] + failures: list[NoticeFailure] = [] if new: - results, usage = await steps.screen(source, new) - await website.screening(source.id, Screening(results=results, usage=usage)) - screened_in = {result.notice_id for result in results if result.screened_in} - to_fetch += [notice for notice in new if notice.id in screened_in] + try: + results, usage = await steps.screen(source, new) + except StepError as error: + # They stay unscreened and come back next round; the open programs' notices + # are still fetched. + logger.warning("Screening failed: %s", error) + failures += [ + NoticeFailure(url=notice.url, failure=error.failure, detail=error.detail) + for notice in new + ] + else: + await website.screening(source.id, Screening(results=results, usage=usage)) + screened_in = {result.notice_id for result in results if result.screened_in} + to_fetch += [notice for notice in new if notice.id in screened_in] useful = 0 - failures: list[NoticeFailure] = [] for notice in to_fetch: try: programs = await _fetch_notice(notice, source, pages, website, steps, clock) diff --git a/crawler/src/crawler/crawl/rounds_test.py b/crawler/src/crawler/crawl/rounds_test.py index dd7d203..d28f3d9 100644 --- a/crawler/src/crawler/crawl/rounds_test.py +++ b/crawler/src/crawler/crawl/rounds_test.py @@ -198,6 +198,32 @@ async def read(_pages: PageSession, _source: WorkSource, url: str) -> NoticePage ] +def test_notices_screening_failed_for_are_reported_and_the_round_goes_on() -> None: + site = FakeWebsite([pending(1, "screen"), pending(2, "fetch")], extracted=frozenset()) + + async def failing( + _source: WorkSource, _notices: list[PendingNotice] + ) -> tuple[list[ScreeningResult], list[Usage]]: + raise StepError("error", "http_429") + + async def read(_pages: PageSession, _source: WorkSource, _url: str) -> NoticePage | None: + return page("冬令营") + + site.run(Steps(list_items, failing, read, extract)) + + assert [call for call, _ in site.calls] == [ + "POST /sources/7/listing", + "POST /notices/2/snapshot", + "POST /snapshots/20/extraction", + "POST /runs", + ] + [run] = site.called("POST /runs") + assert "failure" not in run + assert run["notice_failures"] == [ + {"url": "http://a/1.htm", "failure": "error", "detail": "http_429"} + ] + + def test_a_list_that_cant_be_read_ends_the_round_with_the_reason() -> None: async def unreachable(_pages: PageSession, source: WorkSource) -> list[ListedItem]: raise FetchError("unreachable", source.url) diff --git a/crawler/src/crawler/screening/screen.py b/crawler/src/crawler/screening/screen.py index 00516e4..c17e33f 100644 --- a/crawler/src/crawler/screening/screen.py +++ b/crawler/src/crawler/screening/screen.py @@ -1,32 +1,34 @@ import asyncio from crawler.crawl.contract import PendingNotice, ScreeningResult, Usage, WorkSource +from crawler.crawl.rounds import StepError from crawler.screening.system_one import Call, SystemOne # Calibrated on this model's labeled titles (design §8.1); changing the model requires # rerunning crawler.screening.evaluate and selecting its threshold again. MODEL = "SemIf" THRESHOLD = 0.1 +# Calls in flight at once, across all the rounds that share this Screen. CONCURRENCY = 2 -class ScreeningError(Exception): - pass - - class Screen: def __init__(self, provider: SystemOne) -> None: self.provider = provider + self.slots = asyncio.Semaphore(CONCURRENCY) async def __call__( self, source: WorkSource, notices: list[PendingNotice] ) -> tuple[list[ScreeningResult], list[Usage]]: async def score(notice: PendingNotice) -> tuple[ScreeningResult, Usage]: - result = await self.provider.score(Call(model=MODEL, title=notice.title, section=None)) + async with self.slots: + result = await self.provider.score( + Call(model=MODEL, title=notice.title, section=None) + ) if result.error is not None: - raise ScreeningError(result.error) + raise StepError("error", f"初筛失败: {result.error}") if result.score is None or result.usage is None or result.returned_model is None: - raise ScreeningError("invalid_response") + raise StepError("error", "初筛失败: invalid_response") return ( ScreeningResult( notice_id=notice.id, diff --git a/crawler/src/crawler/screening/screen_test.py b/crawler/src/crawler/screening/screen_test.py index 9af3cb0..4c12a0b 100644 --- a/crawler/src/crawler/screening/screen_test.py +++ b/crawler/src/crawler/screening/screen_test.py @@ -5,7 +5,8 @@ import pytest from crawler.crawl.contract import PendingNotice, ScreeningResult, Usage, WorkSource -from crawler.screening.screen import THRESHOLD, Screen, ScreeningError +from crawler.crawl.rounds import StepError +from crawler.screening.screen import CONCURRENCY, THRESHOLD, Screen from crawler.screening.system_one import SystemOne @@ -71,7 +72,7 @@ async def out_of_order(request: httpx.Request) -> httpx.Response: assert {u.model for u in usage} == {"SemIf"} -def test_provider_failure_raises_screening_error() -> None: +def test_provider_failure_fails_the_screening() -> None: async def run() -> None: provider = SystemOne( "local-test", httpx.MockTransport(lambda _: httpx.Response(503, text="private body")) @@ -92,5 +93,37 @@ async def run() -> None: finally: await provider.close() - with pytest.raises(ScreeningError, match="http_503"): + with pytest.raises(StepError, match="http_503"): asyncio.run(run()) + + +def test_rounds_sharing_a_screen_share_its_limit() -> None: + in_flight = 0 + most = 0 + + async def handle(_request: httpx.Request) -> httpx.Response: + nonlocal in_flight, most + in_flight += 1 + most = max(most, in_flight) + # Lets every other call that can start get as far as here first. + for _ in range(20): + await asyncio.sleep(0) + in_flight -= 1 + return response(0.5) + + async def run() -> None: + provider = SystemOne("local-test", httpx.MockTransport(handle)) + screen = Screen(provider) + notices = [ + PendingNotice( + id=n, url=f"https://example.edu.cn/{n}", title="t", published_on=None, step="screen" + ) + for n in range(4) + ] + try: + await asyncio.gather(screen(source(), notices[:2]), screen(source(), notices[2:])) + finally: + await provider.close() + + asyncio.run(run()) + assert most == CONCURRENCY diff --git a/docs/api.md b/docs/api.md index 1ddf29f..ee5a6db 100644 --- a/docs/api.md +++ b/docs/api.md @@ -118,4 +118,4 @@ | `POST /crawler/runs` | 交一个来源这一轮的结果：开始和结束时间、成功还是失败和原因、列表里几条、新通知几条、其中有用的几条、哪些通知没抓成和原因。交了才算这一轮结束，失败了也等下一轮再抓 | | `POST /crawler/probes/{id}` | 交试抓的结果：抓到的条目和生成的配置，或者失败的原因，带模型用量 | -一个来源的一轮：领活 → 抓列表页、交列表 → 对没初筛的初筛、交分数 → 对要抓正文的抓正文和附件、交快照、上传附件 → 要抽取的抽取、交结果 → 交这一轮的结果，没抓成的通知和原因写在里面。重抓正文也在来源的这一轮里，守同样的频率，失败了一样记进抓取记录。 +一个来源的一轮：领活 → 抓列表页、交列表 → 对没初筛的初筛、交分数 → 对要抓正文的抓正文和附件、交快照、上传附件 → 要抽取的抽取、交结果 → 交这一轮的结果，没抓成的通知和原因写在里面；初筛不成的新通知也算没抓成，下一轮再初筛。重抓正文也在来源的这一轮里，守同样的频率，失败了一样记进抓取记录。 diff --git a/docs/development.md b/docs/development.md index 1487253..f8db5f0 100644 --- a/docs/development.md +++ b/docs/development.md @@ -75,7 +75,7 @@ crawler/ 抓取节点 - 浏览器抓取启动时关掉 Chromium 的自动化标记（`--disable-blink-features=AutomationControlled`），UA 用浏览器自己的，把 `HeadlessChrome` 写成 `Chrome`，末尾加上鼠钟的标识；瑞数这类脚本挑战两处都检查（[总体设计 §5.2](design.md#52-实测)）。不冒充别的浏览器和系统。 - 一轮里同一个网站共用一个无痕会话，抓完就关：脚本挑战设的 Cookie 在会话里有效，后面的页面不用再过挑战；不同网站、不同轮次之间不共享 Cookie。同时开的浏览器页面有上限，多的排队；拒绝弹窗和下载，附件由程序另外下载；不加载图片、字体和音视频，省对方的流量，也省内存。 - 解析和模型分工：`parse` 用程序取出能确定的东西，模型只做判断（[总体设计 §8.2](design.md#82-抽取)）。解析不猜：判断不了的交给模型或人，不硬配。 -- 判别模型和 LLM 的提示词放在调用它的代码旁边。输出都按固定的结构解析，解析不了的当失败处理。 +- 判别模型和 LLM 的提示词放在调用它的代码旁边。输出都按固定的结构解析，解析不了的当失败处理。同时在调用的个数整个进程共用一个上限，几轮同时抓也不超过，免得被限流。 - 每一轮抓取都交一份抓取记录，失败也交，写明原因；不在一轮里反复重试。各步抓不成时抛 `FetchError`（`Fetcher` 抛的）或 `StepError`（写明失败类别，比如 `http_status` 和「HTTP 403」），一轮的流程把它记进抓取记录；别的异常是 bug，照常抛出。 在 `crawler/` 安装项目依赖、Chromium 无头精简版及系统依赖： From 1f0f92b2c80d476fd0d811323e138b7149c9893b Mon Sep 17 00:00:00 2001 From: JunieXD <1240840076@qq.com> Date: Tue, 10 Nov 2026 00:23:43 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E6=8A=8A=E6=8A=BD=E5=8F=96=E6=8E=A5?= =?UTF-8?q?=E8=BF=9B=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #96 合并后，入口里占位的抽取换成正式的 Extract，抓取节点就能从头走到尾。 - 启动时一起校验 `LLM_API_KEY`，`Extract` 和其他客户端一样用 `async with` 打开 - 同时在调用的 LLM 请求整个进程共用一个上限，和初筛一样 - 抽取失败的原因写成「抽取失败: unknown_college」这样，带上具体的错误码，抓取看板和日志里能看出模型错在哪 --- AGENTS.md | 2 +- crawler/.env.example | 1 + crawler/src/crawler/extraction/extract.py | 16 +++- .../src/crawler/extraction/extract_test.py | 73 ++++++++++++++----- crawler/src/crawler/extraction/responses.py | 12 +-- .../src/crawler/extraction/responses_test.py | 2 +- crawler/src/crawler/main.py | 15 ++-- crawler/src/crawler/main_test.py | 6 +- 8 files changed, 86 insertions(+), 41 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 06da5a6..2799356 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -90,7 +90,7 @@ - **上线前没有历史包袱**：表结构和接口直接改。 - **命名**：叫法见[开发约定 §6](docs/development.md#6-术语)。接口文档也是生成的文件。 - **哈希**：只有两处用哈希：会话、个人令牌和抓取节点令牌这些高熵的随机值，库里只存 SHA-256；鼠洞登录的 PKCE 按 OAuth 的规定算 SHA-256。是不是新通知比网址，正文变没变比文字。 -- **配置**：环境变量现在只有网站的 `DATABASE_URL`、`PUBLIC_URL` 和鼠洞登录用的 `MOUSEHOLE_URL`、`MOUSEHOLE_CLIENT_ID`、`MOUSEHOLE_CLIENT_SECRET`，本地的值见 `web/.env.example`；抓取节点的 `MOUSECLOCK_URL`、`MOUSECLOCK_TOKEN` 和初筛用的 `SYSTEM_ONE_API_KEY`，见 `crawler/.env.example`。 +- **配置**：环境变量现在只有网站的 `DATABASE_URL`、`PUBLIC_URL` 和鼠洞登录用的 `MOUSEHOLE_URL`、`MOUSEHOLE_CLIENT_ID`、`MOUSEHOLE_CLIENT_SECRET`，本地的值见 `web/.env.example`；抓取节点的 `MOUSECLOCK_URL`、`MOUSECLOCK_TOKEN`，初筛用的 `SYSTEM_ONE_API_KEY` 和抽取用的 `LLM_API_KEY`，见 `crawler/.env.example`。 ## 可靠与安全 diff --git a/crawler/.env.example b/crawler/.env.example index b8dd5a5..98f3713 100644 --- a/crawler/.env.example +++ b/crawler/.env.example @@ -2,3 +2,4 @@ MOUSECLOCK_URL=http://localhost:5173 MOUSECLOCK_TOKEN= SYSTEM_ONE_API_KEY= +LLM_API_KEY= diff --git a/crawler/src/crawler/extraction/extract.py b/crawler/src/crawler/extraction/extract.py index f6dbfd0..392a17f 100644 --- a/crawler/src/crawler/extraction/extract.py +++ b/crawler/src/crawler/extraction/extract.py @@ -1,3 +1,4 @@ +import asyncio from types import TracebackType from typing import Annotated, Literal, Self @@ -19,6 +20,9 @@ from crawler.extraction.schema import response_schema from crawler.parse.indexing import Document, NoticeIndex, index_documents +# Calls in flight at once, across all the rounds that share this Extract. +CONCURRENCY = 2 + class UnresolvedSelection(BaseModel): model_config = ConfigDict(extra="forbid") @@ -132,6 +136,7 @@ def resolve_programs( class Extract: def __init__(self, key: str, transport: httpx.AsyncBaseTransport | None = None) -> None: self.llm = Responses(key, transport) + self.slots = asyncio.Semaphore(CONCURRENCY) async def __aenter__(self) -> Self: await self.llm.__aenter__() @@ -149,15 +154,18 @@ async def __call__( self, title: str, page: NoticePage, context: ExtractionContext ) -> tuple[list[ExtractedProgram], list[Usage]]: payload, index = prepare(title, page, context) - response = await self.llm.generate(payload) + async with self.slots: + response = await self.llm.generate(payload) try: selected = RoundSelection.model_validate_json(response.output_text(), strict=True) programs = resolve_programs(selected, index, context) - except (ValidationError, SelectionError) as error: - raise StepError("error", "Invalid extraction selection") from error + except ValidationError as error: + raise StepError("error", "抽取失败: invalid_selection") from error + except SelectionError as error: + raise StepError("error", f"抽取失败: {error}") from error usage = response.usage if usage is None: - raise StepError("error", "LLM did not return usage") + raise StepError("error", "抽取失败: missing_usage") return programs, [ Usage( model=response.model, diff --git a/crawler/src/crawler/extraction/extract_test.py b/crawler/src/crawler/extraction/extract_test.py index 4e70a9f..ceccfcb 100644 --- a/crawler/src/crawler/extraction/extract_test.py +++ b/crawler/src/crawler/extraction/extract_test.py @@ -7,7 +7,13 @@ from crawler.crawl.contract import AttachmentInfo, ExtractionContext from crawler.crawl.rounds import Attachment, NoticePage, StepError -from crawler.extraction.extract import Extract, RoundSelection, prepare, resolve_programs +from crawler.extraction.extract import ( + CONCURRENCY, + Extract, + RoundSelection, + prepare, + resolve_programs, +) from crawler.extraction.numbered import SelectedFields, SelectionError @@ -147,25 +153,31 @@ def test_raw_date_cannot_be_fabricated_or_replace_a_resolved_time() -> None: resolve_programs(RoundSelection.model_validate({"programs": [item]}), index, context()) -def test_invalid_selection_fails_the_entire_step() -> None: +def answer(text: str) -> httpx.Response: + return httpx.Response( + 200, + json={ + "model": "deepseek-flash", + "status": "completed", + "output": [{"type": "message", "content": [{"type": "output_text", "text": text}]}], + "usage": {"input_tokens": 100, "output_tokens": 10}, + }, + ) + + +# The code is what the dashboard shows for the notice, so it says what the model got wrong. +@pytest.mark.parametrize( + ("text", "code"), + [ + ("{}", "invalid_selection"), + (json.dumps({"programs": [selection() | {"college_id": 99}]}), "unknown_college"), + ], +) +def test_invalid_selection_fails_the_entire_step(text: str, code: str) -> None: async def run() -> None: - extract = Extract( - "local-test", - httpx.MockTransport( - lambda _: httpx.Response( - 200, - json={ - "model": "deepseek-flash", - "status": "completed", - "output": [ - {"type": "message", "content": [{"type": "output_text", "text": "{}"}]} - ], - }, - ) - ), - ) + extract = Extract("local-test", httpx.MockTransport(lambda _: answer(text))) async with extract: - with pytest.raises(StepError, match="Invalid extraction selection"): + with pytest.raises(StepError, match=f"抽取失败: {code}"): await extract( "通知", NoticePage("", "报名截止6月30日。", hyperlinks=(), attachments=[]), @@ -196,9 +208,32 @@ async def run() -> None: ), ) async with extract: - with pytest.raises(StepError, match="LLM did not return usage"): + with pytest.raises(StepError, match="missing_usage"): await extract( "通知", NoticePage("", "无关材料。", hyperlinks=(), attachments=[]), context() ) asyncio.run(run()) + + +def test_rounds_sharing_an_extract_share_its_limit() -> None: + in_flight = 0 + most = 0 + + async def handle(_request: httpx.Request) -> httpx.Response: + nonlocal in_flight, most + in_flight += 1 + most = max(most, in_flight) + # Lets every other call that can start get as far as here first. + for _ in range(20): + await asyncio.sleep(0) + in_flight -= 1 + return answer('{"programs":[]}') + + async def run() -> None: + async with Extract("local-test", httpx.MockTransport(handle)) as extract: + page = NoticePage("", "无关材料。", hyperlinks=(), attachments=[]) + await asyncio.gather(*(extract("通知", page, context()) for _ in range(4))) + + asyncio.run(run()) + assert most == CONCURRENCY diff --git a/crawler/src/crawler/extraction/responses.py b/crawler/src/crawler/extraction/responses.py index b2c996a..1e23ded 100644 --- a/crawler/src/crawler/extraction/responses.py +++ b/crawler/src/crawler/extraction/responses.py @@ -44,13 +44,13 @@ class Response(BaseModel): def output_text(self) -> str: if self.status != "completed": - raise StepError("error", "LLM response " + self.status) + raise StepError("error", f"抽取失败: {self.status}") parts = [part for item in self.output if item.type == "message" for part in item.content] if any(part.type != "output_text" or part.text is None for part in parts): - raise StepError("error", "LLM refused or returned non-text content") + raise StepError("error", "抽取失败: non_text_output") text = "".join(part.text for part in parts if part.text is not None) if not text: - raise StepError("error", "LLM returned empty content") + raise StepError("error", "抽取失败: empty_output") return text @@ -105,10 +105,10 @@ async def generate(self, payload: dict[str, JsonValue]) -> Response: try: response = await self.client.post(API_URL, json=payload) except httpx.HTTPError as error: - raise StepError("error", type(error).__name__) from error + raise StepError("error", f"抽取失败: {type(error).__name__}") from error if response.status_code != 200: - raise StepError("error", f"LLM HTTP {response.status_code}") + raise StepError("error", f"抽取失败: http_{response.status_code}") try: return Response.model_validate_json(response.content) except ValidationError as error: - raise StepError("error", "Invalid LLM response") from error + raise StepError("error", "抽取失败: invalid_response") from error diff --git a/crawler/src/crawler/extraction/responses_test.py b/crawler/src/crawler/extraction/responses_test.py index 78f63e4..cd4eab8 100644 --- a/crawler/src/crawler/extraction/responses_test.py +++ b/crawler/src/crawler/extraction/responses_test.py @@ -103,6 +103,6 @@ async def run() -> None: async with client: with pytest.raises(StepError) as caught: await client.generate({}) - assert caught.value.detail == f"LLM HTTP {status}" + assert caught.value.detail == f"抽取失败: http_{status}" asyncio.run(run()) diff --git a/crawler/src/crawler/main.py b/crawler/src/crawler/main.py index 364b74d..3f328fb 100644 --- a/crawler/src/crawler/main.py +++ b/crawler/src/crawler/main.py @@ -14,12 +14,13 @@ from pydantic import BaseModel, ConfigDict, HttpUrl, StringConstraints, ValidationError from crawler.crawl.browser import Browser -from crawler.crawl.contract import ExtractedProgram, ExtractionContext, Usage, WorkSource +from crawler.crawl.contract import WorkSource from crawler.crawl.fetch import Fetcher from crawler.crawl.listing import list_items from crawler.crawl.notice import read_notice -from crawler.crawl.rounds import NoticePage, StepError, Steps, crawl_source +from crawler.crawl.rounds import Steps, crawl_source from crawler.crawl.website import Website +from crawler.extraction.extract import Extract from crawler.log import JsonFormatter from crawler.screening.screen import Screen from crawler.screening.system_one import SystemOne @@ -37,6 +38,7 @@ class Config(BaseModel): mouseclock_url: HttpUrl mouseclock_token: Secret system_one_api_key: Secret + llm_api_key: Secret def load_config(environ: Mapping[str, str]) -> Config: @@ -74,12 +76,6 @@ async def _crawl(source: WorkSource, browser: Browser, website: Website, steps: logger.exception("Round failed") -async def _extract( - _page: NoticePage, _context: ExtractionContext -) -> tuple[list[ExtractedProgram], list[Usage]]: - raise StepError("error", "抽取还没接上") - - async def _run(config: Config) -> None: stopping = asyncio.Event() @@ -94,9 +90,10 @@ def stop() -> None: Fetcher() as fetcher, Browser(fetcher) as browser, Website(str(config.mouseclock_url), config.mouseclock_token) as website, + Extract(config.llm_api_key) as extract, ): steps = Steps( - list_items=list_items, screen=Screen(system_one), read=read_notice, extract=_extract + list_items=list_items, screen=Screen(system_one), read=read_notice, extract=extract ) await serve(website, browser, steps, stopping) finally: diff --git a/crawler/src/crawler/main_test.py b/crawler/src/crawler/main_test.py index d6d6cf0..744f679 100644 --- a/crawler/src/crawler/main_test.py +++ b/crawler/src/crawler/main_test.py @@ -23,7 +23,11 @@ def test_bad_configuration_is_named_without_its_values(caplog: pytest.LogCapture load_config(environ) line = JsonFormatter().format(caplog.records[0]) - assert json.loads(line)["variables"].keys() == {"MOUSECLOCK_URL", "SYSTEM_ONE_API_KEY"} + assert json.loads(line)["variables"].keys() == { + "MOUSECLOCK_URL", + "SYSTEM_ONE_API_KEY", + "LLM_API_KEY", + } assert "secret" not in line From 4a4a198a4313ee8ad87143e1abe43ca5b0477c45 Mon Sep 17 00:00:00 2001 From: JunieXD <1240840076@qq.com> Date: Tue, 10 Nov 2026 10:34:28 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E9=A2=86=E6=B4=BB=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E6=97=B6=E4=B8=8D=E4=B8=AD=E6=96=AD=E8=BF=9B=E8=A1=8C=E4=B8=AD?= =?UTF-8?q?=E7=9A=84=E8=BD=AE=E6=AC=A1=EF=BC=8C=E6=A8=A1=E5=9E=8B=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E8=AE=B0=E4=B8=8B=E7=94=A8=E9=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 网站返回的数据对不上契约时，领活抛出的校验错误没被捕获，会把正在抓的其他来源一起取消；领活失败的日志写了异常原文，可能带出令牌。开发约定 §4 要求记的模型名称、用量和耗时也还没记。 - 领活遇到网络错误或不合契约的响应，都只记错误类别和状态码，下一分钟再领 - 初筛每批、抽取每次调用各记一条日志，写明模型、用量和耗时；抽取在检查输出前就记，答错的那次也有记录 --- crawler/src/crawler/extraction/extract.py | 24 ++++++++-- crawler/src/crawler/main.py | 12 ++++- crawler/src/crawler/main_test.py | 55 +++++++++++++++++++++-- crawler/src/crawler/screening/screen.py | 14 ++++++ 4 files changed, 97 insertions(+), 8 deletions(-) diff --git a/crawler/src/crawler/extraction/extract.py b/crawler/src/crawler/extraction/extract.py index 392a17f..6147bbe 100644 --- a/crawler/src/crawler/extraction/extract.py +++ b/crawler/src/crawler/extraction/extract.py @@ -1,4 +1,6 @@ import asyncio +import logging +import time from types import TracebackType from typing import Annotated, Literal, Self @@ -22,6 +24,7 @@ # Calls in flight at once, across all the rounds that share this Extract. CONCURRENCY = 2 +logger = logging.getLogger(__name__) class UnresolvedSelection(BaseModel): @@ -155,7 +158,25 @@ async def __call__( ) -> tuple[list[ExtractedProgram], list[Usage]]: payload, index = prepare(title, page, context) async with self.slots: + started = time.monotonic() response = await self.llm.generate(payload) + elapsed = time.monotonic() - started + usage = response.usage + if usage is None: + raise StepError("error", "抽取失败: missing_usage") + # Before the answer is checked, so what a failed answer cost is logged too. + details = usage.input_tokens_details + logger.info( + "Extraction answered", + extra={ + "model": response.model, + "status": response.status, + "input_tokens": usage.input_tokens, + "output_tokens": usage.output_tokens, + "cached_tokens": details.cached_tokens if details else None, + "elapsed": elapsed, + }, + ) try: selected = RoundSelection.model_validate_json(response.output_text(), strict=True) programs = resolve_programs(selected, index, context) @@ -163,9 +184,6 @@ async def __call__( raise StepError("error", "抽取失败: invalid_selection") from error except SelectionError as error: raise StepError("error", f"抽取失败: {error}") from error - usage = response.usage - if usage is None: - raise StepError("error", "抽取失败: missing_usage") return programs, [ Usage( model=response.model, diff --git a/crawler/src/crawler/main.py b/crawler/src/crawler/main.py index 3f328fb..cad5393 100644 --- a/crawler/src/crawler/main.py +++ b/crawler/src/crawler/main.py @@ -57,8 +57,16 @@ async def serve(website: Website, browser: Browser, steps: Steps, stopping: asyn while True: try: work = await website.work() - except httpx.HTTPError as error: - logger.warning("Asking for work failed: %s", error) + except (httpx.HTTPError, ValidationError) as error: + # By kind only: the text may carry the token (development §4). A website that no + # longer matches this crawler is asked again next minute, like one that's down. + status = ( + error.response.status_code if isinstance(error, httpx.HTTPStatusError) else None + ) + logger.warning( + "Asking for work failed", + extra={"error": type(error).__name__, "status": status}, + ) else: for source in work.sources: rounds.create_task(_crawl(source, browser, website, steps)) diff --git a/crawler/src/crawler/main_test.py b/crawler/src/crawler/main_test.py index 744f679..13e59f6 100644 --- a/crawler/src/crawler/main_test.py +++ b/crawler/src/crawler/main_test.py @@ -16,6 +16,10 @@ from crawler.main import load_config, serve +async def unused(*_: object) -> Never: + raise AssertionError + + def test_bad_configuration_is_named_without_its_values(caplog: pytest.LogCaptureFixture) -> None: environ = {"MOUSECLOCK_URL": "ftp://user:secret@site", "MOUSECLOCK_TOKEN": "token"} @@ -51,9 +55,6 @@ async def list_items(_pages: PageSession, source: WorkSource) -> list[ListedItem raise RuntimeError("bug") raise StepError("no_items") - async def unused(*_: object) -> Never: - raise AssertionError - async def main() -> None: stopping = asyncio.Event() stopping.set() @@ -83,3 +84,51 @@ async def main() -> None: assert failed["source_id"] == 1 assert isinstance(failed["started_at"], int) assert "RuntimeError: bug" in failed["stack"] + + +def test_asking_for_work_that_fails_leaves_the_rounds_and_logs_no_details( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.setattr("crawler.main.POLL", 0) + source = WorkSource(id=1, url="http://a/list.htm", browser=False, config=None, backfill=False) + calls: list[str] = [] + stopping = asyncio.Event() + failed_twice = asyncio.Event() + + def handle(request: httpx.Request) -> httpx.Response: + path = request.url.path.removeprefix("/api/v1/crawler") + calls.append(f"{request.method} {path}") + if path != "/work": + return httpx.Response(204) + polls = calls.count("GET /work") + if polls == 1: + work = {"sources": [source.model_dump(mode="json")], "probes": []} + return httpx.Response(200, json=work) + if polls == 2: + # A website that no longer matches this crawler's contract. + return httpx.Response(200, json={"sources": "invalid", "probes": []}) + failed_twice.set() + raise httpx.LocalProtocolError(f"Illegal header value {request.headers['authorization']}") + + # Still in progress while the next two asks fail. + async def list_items(_pages: PageSession, _source: WorkSource) -> list[ListedItem]: + await failed_twice.wait() + stopping.set() + raise StepError("no_items") + + async def main() -> None: + nowhere = httpx.MockTransport(lambda _: httpx.Response(500)) + async with ( + Fetcher(nowhere) as fetcher, + Website("http://site", "token-secret", httpx.MockTransport(handle)) as website, + ): + await serve( + website, Browser(fetcher), Steps(list_items, unused, unused, unused), stopping + ) + + asyncio.run(main()) + + assert "POST /runs" in calls + failures = [r for r in caplog.records if r.getMessage() == "Asking for work failed"] + assert [vars(r)["error"] for r in failures[:2]] == ["ValidationError", "LocalProtocolError"] + assert all("token-secret" not in JsonFormatter().format(r) for r in caplog.records) diff --git a/crawler/src/crawler/screening/screen.py b/crawler/src/crawler/screening/screen.py index c17e33f..9c56c3a 100644 --- a/crawler/src/crawler/screening/screen.py +++ b/crawler/src/crawler/screening/screen.py @@ -1,4 +1,6 @@ import asyncio +import logging +import time from crawler.crawl.contract import PendingNotice, ScreeningResult, Usage, WorkSource from crawler.crawl.rounds import StepError @@ -10,6 +12,7 @@ THRESHOLD = 0.1 # Calls in flight at once, across all the rounds that share this Screen. CONCURRENCY = 2 +logger = logging.getLogger(__name__) class Screen: @@ -42,6 +45,7 @@ async def score(notice: PendingNotice) -> tuple[ScreeningResult, Usage]: ), ) + started = time.monotonic() results: list[ScreeningResult] = [] usage: list[Usage] = [] for offset in range(0, len(notices), CONCURRENCY): @@ -52,4 +56,14 @@ async def score(notice: PendingNotice) -> tuple[ScreeningResult, Usage]: for result, tokens in outcomes: results.append(result) usage.append(tokens) + logger.info( + "Screened", + extra={ + "model": MODEL, + "notices": len(results), + "input_tokens": sum(u.input_tokens for u in usage), + "output_tokens": sum(u.output_tokens for u in usage), + "elapsed": time.monotonic() - started, + }, + ) return results, usage