feat: wechat message scraper
Adds Task 1.6: a backward-cursor paginator in base.py, a message-type normalizer in parsers.py, and the WeChat chat-message scraper in wechat.py. - app/scraper/parsers.py: normalize_wechat_type() maps 1 -> "text", -2 -> "divider", everything else -> "unknown". Total (never raises); unknown codes pass through so no real message is silently dropped. - app/scraper/base.py: paginate_by_cursor() generic async generator. Takes a fetch(cursor) -> (rows, next_cursor, has_more) coroutine; advances the cursor and stops on has_more=False OR an empty page (defensive). Same RateLimiter integration as paginate_by_page. - app/scraper/wechat.py: scrape_wechat_messages() wraps POST /wechatweb/msg/chats. Body is form-encoded (NOT JSON) — content-type is overridden per request to application/x-www-form-urlencoded via the YunkeClient.post(headers=...) override hook. Pages backward by setting end = data.end - 1 each batch, repeats until hasLast=false. Drops any record whose type/messageType is -2 (date divider UI chrome). Yields normalized dicts (snake_case Message columns + channel="wechat" + raw_json); the raw type code is preserved on raw_json for the persistence layer (Task 1.7). Tests: 13 new (10 parser parametrized + 3 scraper). respx-mocked msg/chats with two backward pages — asserts [m1, m2, m3] yielded (the -2 divider dropped), full field mapping vs the live-capture names, the content-type is form-encoded (not JSON), and the 2nd request body's end == E1 - 1 (proving the backward cursor advance). Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1,42 +1,54 @@
|
||||
"""Reusable building blocks for list scrapers.
|
||||
|
||||
This module establishes the pagination pattern every list scraper in the
|
||||
This module establishes the pagination patterns every list scraper in the
|
||||
project reuses (see design doc §6.2 拉取层). It is intentionally small:
|
||||
|
||||
* :func:`paginate_by_page` — a generic async generator that walks page-numbered
|
||||
endpoints until a page comes back empty (the primary stop signal used by the
|
||||
yunke CRM), with a defensive over-fetch ceiling keyed off the declared total.
|
||||
* :func:`paginate_by_cursor` — a generic async generator that walks a
|
||||
cursor-paginated endpoint until the fetch callable reports no more pages.
|
||||
Used by the WeChat message scraper, whose ``msg/chats`` endpoint pages
|
||||
BACKWARD by timestamp cursor (``end = prev_end - 1`` until ``hasLast``
|
||||
flips false).
|
||||
* :class:`RateLimiter` — a tiny asyncio limiter that enforces a minimum
|
||||
interval between successive calls. Scrapers that need to be polite to the
|
||||
upstream construct one and ``await`` :meth:`acquire` before each call.
|
||||
|
||||
The paginator talks to :class:`app.session.yunke_client.YunkeClient` (which
|
||||
already injects auth headers + handles 401 re-login). It is generic over the
|
||||
response shape via three callables:
|
||||
``paginate_by_page`` talks to :class:`app.session.yunke_client.YunkeClient`
|
||||
(which already injects auth headers + handles 401 re-login). It is generic
|
||||
over the response shape via three callables:
|
||||
|
||||
* ``parse_total(resp) -> int | None`` — read the declared row count from the
|
||||
envelope (used only for the over-fetch safety; ``None`` disables it).
|
||||
* ``parse_rows(resp) -> list[dict]`` — extract the list of records.
|
||||
* ``build_body(page_num, page_size, **extra) -> dict`` — build the request
|
||||
body for a given page.
|
||||
|
||||
``paginate_by_cursor`` is decoupled from the client entirely: it takes a
|
||||
single ``fetch(cursor)`` coroutine that performs whatever request the caller
|
||||
needs and returns ``(rows, next_cursor, has_more)``. The cursor-paginator's
|
||||
job is just to advance the cursor and stop on ``has_more == False``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from collections.abc import AsyncIterator, Callable
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||||
from typing import Any
|
||||
|
||||
from app.session.yunke_client import YunkeClient
|
||||
|
||||
__all__ = ["RateLimiter", "paginate_by_page"]
|
||||
__all__ = ["RateLimiter", "paginate_by_cursor", "paginate_by_page"]
|
||||
|
||||
# Informal type aliases for the three response-handling callables. The yunke
|
||||
# Informal type aliases for the response-handling callables. The yunke
|
||||
# response shapes are dict-shaped JSON and do not benefit from finer typing.
|
||||
ParseTotal = Callable[[Any], "int | None"]
|
||||
ParseRows = Callable[[Any], "list[dict[str, Any]]"]
|
||||
BuildBody = Callable[..., "dict[str, Any]"]
|
||||
# fetch(cursor) -> (rows, next_cursor, has_more) — see ``paginate_by_cursor``.
|
||||
FetchCursor = Callable[[Any], Awaitable["tuple[list[dict[str, Any]], Any, bool]"]]
|
||||
|
||||
|
||||
async def paginate_by_page(
|
||||
@@ -105,6 +117,61 @@ async def paginate_by_page(
|
||||
page_num += 1
|
||||
|
||||
|
||||
async def paginate_by_cursor(
|
||||
fetch: FetchCursor,
|
||||
*,
|
||||
initial_cursor: Any,
|
||||
rate_limiter: RateLimiter | None = None,
|
||||
) -> AsyncIterator[dict[str, Any]]:
|
||||
"""Walk a cursor-paginated endpoint, yielding rows until no more pages.
|
||||
|
||||
Used by the WeChat message scraper, whose ``msg/chats`` endpoint pages
|
||||
BACKWARD by timestamp cursor: each response carries ``data.end`` (the
|
||||
oldest timestamp in the batch) and ``data.hasLast`` (true iff there are
|
||||
older messages still). The caller advances the cursor to ``end - 1`` and
|
||||
repeats until ``hasLast`` is false.
|
||||
|
||||
The contract is intentionally client-agnostic. Each call to
|
||||
``fetch(cursor)`` must return a 3-tuple ``(rows, next_cursor, has_more)``:
|
||||
|
||||
* ``rows`` — list of records from this page (may be empty).
|
||||
* ``next_cursor`` — cursor for the next page (e.g. ``prev_end - 1``).
|
||||
* ``has_more`` — True if there are more pages to fetch.
|
||||
|
||||
The loop yields every row of every page, then advances ``cursor`` to
|
||||
``next_cursor`` and repeats. It stops after:
|
||||
|
||||
* a page whose ``has_more`` is False (the documented primary stop signal
|
||||
for cursor endpoints — e.g. ``hasLast=false``), OR
|
||||
* an empty ``rows`` list (a defensive secondary stop, so a misbehaving
|
||||
upstream that never flips ``has_more`` to False cannot loop forever).
|
||||
|
||||
``rate_limiter`` is acquired before each fetch when supplied, so cursor
|
||||
scrapers get the same inter-call spacing as page scrapers.
|
||||
"""
|
||||
cursor = initial_cursor
|
||||
while True:
|
||||
if rate_limiter is not None:
|
||||
await rate_limiter.acquire()
|
||||
|
||||
rows, next_cursor, has_more = await fetch(cursor)
|
||||
|
||||
# Defensive stop: an empty page is treated as end-of-data even if the
|
||||
# upstream claims has_more. This mirrors paginate_by_page's primary
|
||||
# stop signal and bounds the worst case (a stuck cursor that never
|
||||
# advances would otherwise loop forever).
|
||||
if not rows:
|
||||
return
|
||||
|
||||
for row in rows:
|
||||
yield row
|
||||
|
||||
if not has_more:
|
||||
return
|
||||
|
||||
cursor = next_cursor
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""Minimal asyncio limiter enforcing a minimum interval between calls.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user