加上抓取节点的入口和镜像

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` 和抽取用的 `LLM_API_KEY`，见 `crawler/.env.example`。
 
 ## 可靠与安全
 

crawler/.dockerignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+.env
+screening-results/
+extraction-results/

crawler/.env.example

@@ -0,0 +1,5 @@
+# Against the dev server (`pnpm dev` in web/); the token comes from `pnpm crawler-token <name>`.
+MOUSECLOCK_URL=http://localhost:5173
+MOUSECLOCK_TOKEN=
+SYSTEM_ONE_API_KEY=
+LLM_API_KEY=

crawler/.gitignore

@@ -1,3 +1,4 @@
 __pycache__/
 screening-results/
 extraction-results/
+.env

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"]

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",

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)

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

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(
@@ -116,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)

crawler/src/crawler/crawl/rounds_test.py

@@ -287,6 +287,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)

crawler/src/crawler/extraction/extract.py

@@ -1,3 +1,6 @@
+import asyncio
+import logging
+import time
 from types import TracebackType
 from typing import Annotated, Literal, Self
 
@@ -19,6 +22,10 @@
 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
+logger = logging.getLogger(__name__)
+
 
 class UnresolvedSelection(BaseModel):
     model_config = ConfigDict(extra="forbid")
@@ -132,6 +139,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 +157,33 @@ 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:
+            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)
-        except (ValidationError, SelectionError) as error:
-            raise StepError("error", "Invalid extraction selection") from error
-        usage = response.usage
-        if usage is None:
-            raise StepError("error", "LLM did not return usage")
+        except ValidationError as error:
+            raise StepError("error", "抽取失败: invalid_selection") from error
+        except SelectionError as error:
+            raise StepError("error", f"抽取失败: {error}") from error
         return programs, [
             Usage(
                 model=response.model,

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