From 903286f37965e6585985c0a9d3eef459f9bef1d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sat, 1 Aug 2026 11:39:55 +0800 Subject: [PATCH] feat: wechat message scraper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- yunke-server/app/scraper/base.py | 81 ++++++- yunke-server/app/scraper/parsers.py | 51 ++++ yunke-server/app/scraper/wechat.py | 220 +++++++++++++++++- .../tests/cassettes/msg_chats_fixture.py | 147 ++++++++++++ yunke-server/tests/test_parsers.py | 44 ++++ .../tests/test_scrape_wechat_messages.py | 217 +++++++++++++++++ 6 files changed, 744 insertions(+), 16 deletions(-) create mode 100644 yunke-server/app/scraper/parsers.py create mode 100644 yunke-server/tests/cassettes/msg_chats_fixture.py create mode 100644 yunke-server/tests/test_parsers.py create mode 100644 yunke-server/tests/test_scrape_wechat_messages.py diff --git a/yunke-server/app/scraper/base.py b/yunke-server/app/scraper/base.py index cf151b1..77bfcf0 100644 --- a/yunke-server/app/scraper/base.py +++ b/yunke-server/app/scraper/base.py @@ -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. diff --git a/yunke-server/app/scraper/parsers.py b/yunke-server/app/scraper/parsers.py new file mode 100644 index 0000000..e7f4747 --- /dev/null +++ b/yunke-server/app/scraper/parsers.py @@ -0,0 +1,51 @@ +"""Field / type-code normalization helpers for upstream record parsers. + +The yunke CRM uses opaque integer type codes (``type`` / ``messageType``) on +WeChat message records. The storage layer prefers stable enum strings so that +media dispatch (image / voice / video / file) and presentation (text vs. +date divider) can be keyed off a single normalized value rather than the +raw code, whose exact mapping for non-text codes is a Phase-0 unknown. + +This module is intentionally tiny: one total function that maps the +documented codes and falls through to ``"unknown"`` for everything else. +It MUST NOT raise — every record pulled during a scrape passes through here, +and one unexpected code must not abort the whole pull. The raw code is +preserved on the record's ``raw_json`` so the persistence layer can recover +it when needed. +""" + +from __future__ import annotations + +from typing import Any + +__all__ = ["normalize_wechat_type"] + +# Documented WeChat message type codes (see Task 1.6 spec). +# * ``1`` — plain text (and sticker text). The only non-media type we map +# today; everything else falls through to ``"unknown"``. +# * ``-2`` — date divider. NOT a real message; the scraper uses this label as +# its drop signal (filtered out, never yielded, never persisted). +# Image / voice / video / file / transfer / redpacket / system codes are +# observed in the wild but their exact integer values are a Phase-0 unknown. +# When confirmed, extend this table — for now they pass through as +# ``"unknown"`` so the message is still ingested rather than dropped. +_WECHAT_TYPE_NAMES: dict[int, str] = { + 1: "text", + -2: "divider", +} + + +def normalize_wechat_type(raw: Any) -> str: + """Map an upstream WeChat ``type`` / ``messageType`` code to a stable string. + + Returns ``"text"`` for code ``1``, ``"divider"`` for code ``-2``, and + ``"unknown"`` for any other / missing / unparsable value. + + Never raises: an unexpected type must not break the scrape. The raw code + is preserved on ``raw_json`` so the persistence layer can recover it. + """ + try: + code = int(raw) + except (TypeError, ValueError): + return "unknown" + return _WECHAT_TYPE_NAMES.get(code, "unknown") diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index 56ab654..bcf6e65 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -1,16 +1,24 @@ -"""WeChat friend-list scraper. +"""WeChat scrapers. -Wraps ``POST /wechatweb/wxFriends/friend_list/es`` and yields normalized dicts -ready to be upserted into :class:`app.db.models.Customer` (persistence is a -LATER task — Task 1.7). Each dict carries: +Two endpoints today: -* snake_case keys matching the ``Customer`` columns (see :func:`_normalize`), +* :func:`scrape_wechat_friends` — wraps ``POST /wechatweb/wxFriends/friend_list/es`` + and yields normalized dicts ready to be upserted into + :class:`app.db.models.Customer`. +* :func:`scrape_wechat_messages` — wraps ``POST /wechatweb/msg/chats`` and + yields normalized chat-message dicts (see Task 1.6). + +Both yield dicts (not ORM). Persistence is a LATER task — Task 1.7. Each dict +carries: + +* snake_case keys matching the target ORM columns (see :func:`_normalize` + for friends and :func:`_normalize_message` for messages), * ``channel == "wechat"``, * ``raw_json`` — the original upstream record, verbatim, so nothing captured is lost even when a column has no target field. -Field names are mapped from the real capture (see Task 1.5 spec). The mapping -is a straight ``camelCase -> snake_case`` translation with a couple of +Friend field names are mapped from the real capture (see Task 1.5 spec). The +mapping is a straight ``camelCase -> snake_case`` translation with a couple of disambiguations where the source has two fields for one column: * ``phone`` <- ``phone`` @@ -24,12 +32,18 @@ from __future__ import annotations from collections.abc import AsyncIterator from typing import Any -from app.scraper.base import paginate_by_page +from app.scraper.base import paginate_by_cursor, paginate_by_page +from app.scraper.parsers import normalize_wechat_type from app.session.yunke_client import YunkeClient -__all__ = ["scrape_wechat_friends"] +__all__ = ["scrape_wechat_friends", "scrape_wechat_messages"] FRIEND_LIST_PATH = "/wechatweb/wxFriends/friend_list/es" +MSG_CHATS_PATH = "/wechatweb/msg/chats" + +# Upstream sends form-encoded bodies for ``msg/chats`` (NOT JSON), so we have +# to override the client's default JSON content-type on every request. +_FORM_CONTENT_TYPE = "application/x-www-form-urlencoded" def _parse_total(resp: Any) -> int | None: @@ -169,3 +183,191 @@ async def scrape_wechat_friends( user_id=user_id, ): yield _normalize(record) + + +# =========================================================================== +# Message scraper (Task 1.6) — POST /wechatweb/msg/chats +# =========================================================================== + + +async def _fetch_msg_page( + client: YunkeClient, + *, + wechat_id: str, + friend_id: str, + types: str, + start: int, + end_cursor: int, +) -> tuple[list[dict[str, Any]], int, bool]: + """Fetch one backward page of ``msg/chats``. + + Returns the ``(rows, next_cursor, has_more)`` triple expected by + :func:`paginate_by_cursor`. ``next_cursor`` is ``data.end - 1`` so the + next call's window is ``(end-1, ...]`` — i.e. strictly older than the + oldest message in this batch. + """ + resp = await client.post( + MSG_CHATS_PATH, + data={ + "wechatId": wechat_id, + "friendId": friend_id, + "types": types, + "start": str(start), + "end": str(end_cursor), + }, + headers={"content-type": _FORM_CONTENT_TYPE}, + ) + + try: + body = resp.json() + except (AttributeError, ValueError): + # Malformed envelope — treat as empty + no-more so the paginator stops. + return [], end_cursor, False + + data = body.get("data") if isinstance(body, dict) else None + if not isinstance(data, dict): + return [], end_cursor, False + + rows = data.get("messages") + if not isinstance(rows, list): + rows = [] + + end = data.get("end") + next_cursor = ( + int(end) - 1 if isinstance(end, (int, float)) and not isinstance(end, bool) + else end_cursor + ) + has_more = bool(data.get("hasLast")) + return rows, next_cursor, has_more + + +def _normalize_message(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream ``msg/chats`` record to the ``Message`` column shape. + + Adds ``channel="wechat"`` and preserves the original record under + ``raw_json``. The ``type`` field is normalized via + :func:`normalize_wechat_type` to a stable enum string; the raw code is + still recoverable from ``raw_json`` for the persistence layer. + + ``None``-safe: missing fields become ``None`` rather than raising, so a + partial payload does not break the whole scrape. Value coercion + (e.g. ``isSend``/``mine``/``isDel`` ``int`` -> ``bool``) is deferred to + the persistence layer — here we carry the raw values through, except for + ``is_send`` which is pulled from ``mine`` (falling back to ``isSend``) + since that disambiguation is straightforward. + """ + # `mine` and `isSend` carry the same semantic; the live capture uses both. + # Prefer `mine`, fall back to `isSend`. Missing-key .get() returns None, + # but `mine` may legitimately be the int 0 (falsy but present) — guard + # explicitly with `is None`. + mine = record.get("mine") + is_send = mine if mine is not None else record.get("isSend") + + raw_type = record.get("messageType") + if raw_type is None: + raw_type = record.get("type") + + # `timestamp` is the canonical field; fall back to `createTime`. + timestamp = record.get("timestamp") + if timestamp is None: + timestamp = record.get("createTime") + + return { + "msg_id": record.get("msgSvrId"), + "channel": "wechat", + "owner_account_id": record.get("wechatId"), + "talker_id": record.get("talker"), + "ori_talker": record.get("oriTalker"), + "room_id": record.get("roomid"), + "is_send": is_send, + "type": normalize_wechat_type(raw_type), + "text": record.get("text"), + "file_url": record.get("file"), + "file_th_url": record.get("fileTh"), + "file_hd_url": record.get("fileHd"), + "md5": record.get("md5"), + "md5_th": record.get("md5Th"), + "md5_hd": record.get("md5Hd"), + "duration": record.get("duration"), + "transcription_text": record.get("transcriptionText"), + "img_path": record.get("imgPath"), + "original_image": record.get("originalImage"), + "pic_type": record.get("picType"), + "pay_msg_id": record.get("paymsgid"), + "refer_msg": record.get("referMsg"), + "sender_name": record.get("name"), + "sender_head_url": record.get("headUrl"), + "timestamp": timestamp, + "is_del": record.get("isDel"), + "version": record.get("version"), + "raw_json": record, + } + + +def _is_divider(record: dict[str, Any]) -> bool: + """True iff this record is a date-divider (``type``/``messageType`` == -2). + + These are not real messages — the upstream inserts them as visual date + bars in the chat UI — and must never be yielded or persisted. Per the + Task 1.6 spec we drop a record when EITHER ``type`` or ``messageType`` is + ``-2``; all other unknown codes pass through. + """ + return normalize_wechat_type(record.get("messageType")) == "divider" or ( + normalize_wechat_type(record.get("type")) == "divider" + ) + + +async def scrape_wechat_messages( + client: YunkeClient, + wechat_id: str, + friend_id: str, + start: int, + end: int, + *, + types: str = "", +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WeChat chat messages, paging backward by timestamp. + + Wraps ``POST /wechatweb/msg/chats`` (form-encoded body, NOT JSON). The + endpoint pages BACKWARD: each response carries ``data.end`` (oldest + timestamp in this batch) and ``data.hasLast`` (older messages exist?). + The scraper advances the cursor to ``end - 1`` and repeats until + ``hasLast`` is false, then stops. + + Date dividers (``type == -2``) are dropped — they are UI chrome, not real + messages. All other type codes pass through (mapped to ``"unknown"`` when + not in our small known-table) so no real message is silently lost. + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client. + wechat_id: + Staff WeChat id (``wechatId`` form field) — maps to + ``owner_account_id`` on each yielded record. + friend_id: + Friend ``wxid`` OR chatroom id (``friendId`` form field) — selects + the conversation to pull. + start, end: + Time window in ms epoch. ``end`` is also the initial backward cursor; + the first request queries the ``(start, end]`` window and the cursor + walks toward ``start`` from there. + types: + Optional upstream message-type filter (passed through verbatim to the + form body). Empty by default — pull all types. + """ + async for record in paginate_by_cursor( + lambda end_cursor: _fetch_msg_page( + client, + wechat_id=wechat_id, + friend_id=friend_id, + types=types, + start=start, + end_cursor=end_cursor, + ), + initial_cursor=end, + ): + if _is_divider(record): + continue + yield _normalize_message(record) diff --git a/yunke-server/tests/cassettes/msg_chats_fixture.py b/yunke-server/tests/cassettes/msg_chats_fixture.py new file mode 100644 index 0000000..71dc8ab --- /dev/null +++ b/yunke-server/tests/cassettes/msg_chats_fixture.py @@ -0,0 +1,147 @@ +"""Fixture payloads for ``POST /wechatweb/msg/chats`` (Task 1.6). + +These records mirror the real field names captured from the live CRM (see the +Task 1.6 spec). The cassette is shaped to drive the backward-cursor paginator +across two pages: + +* page 1 -> ``messages=[m1, m_datebar(-2), m2]``, ``end=E1``, ``hasLast=true`` +* page 2 -> ``messages=[m3]``, ``end=E2``, ``hasLast=false`` + +The ``-2`` record is a date divider that the scraper must DROP. ``E1`` is the +oldest timestamp in page 1's batch, so the backward-cursor advance must set +the next request's ``end`` to ``E1 - 1``. Page 2 reports ``hasLast=false`` so +the scraper stops without a third fetch. +""" + +from __future__ import annotations + +from typing import Any + +# Oldest-timestamp-in-batch returned by each page; the scraper uses these to +# advance the backward cursor (next_request_end = page_end - 1). +E1 = 1784517338999 +E2 = 1784517000000 + +# Page 1 records, EXACT live-capture field names. m1 is text/direct/incoming, +# m_datebar is a -2 divider (must be filtered), m2 is image/chatroom/outgoing. +MESSAGE_RECORDS_PAGE1: list[dict[str, Any]] = [ + { # m1 — text message, incoming, direct (empty roomid). + "msgSvrId": "1000001", + "type": 1, + "messageType": 1, + "text": "hi alice", + "wechatId": "wxid_staff_alice", + "talker": "wxid_friend_0001", + "oriTalker": "wxid_friend_0001", + "roomid": "", + "mine": 0, + "isSend": 0, + "timestamp": E1 + 600000, + "createTime": E1 + 600000, + "name": "Alice Wang", + "headUrl": "https://cdn.example.com/alice.png", + "isDel": 0, + "version": "v1.0", + }, + { # date divider — type == -2; scraper MUST drop this, never yield it. + "msgSvrId": "1000000", + "type": -2, + "messageType": -2, + "text": "2026-01-01", + "wechatId": "wxid_staff_alice", + "talker": "wxid_friend_0001", + "oriTalker": "wxid_friend_0001", + "roomid": "", + "mine": 0, + "isSend": 0, + "timestamp": E1 + 300000, + "createTime": E1 + 300000, + "name": "", + "headUrl": "", + "isDel": 0, + "version": "v1.0", + }, + { # m2 — image message, outgoing, in a chatroom. + "msgSvrId": "1000002", + "type": 3, + "messageType": 3, + "text": "[image]", + "file": "https://cdn.example.com/img2.jpg", + "fileTh": "https://cdn.example.com/img2_th.jpg", + "fileHd": "https://cdn.example.com/img2_hd.jpg", + "md5": "md5img2", + "md5Th": "md5img2th", + "md5Hd": "md5img2hd", + "imgPath": "/tmp/img2.jpg", + "originalImage": "/tmp/img2_orig.jpg", + "picType": 1, + "wechatId": "wxid_staff_alice", + "talker": "wxid_friend_0001", + "oriTalker": "room12345@chatroom", + "roomid": "room12345@chatroom", + "mine": 1, + "isSend": 1, + "timestamp": E1 + 100, + "createTime": E1 + 100, + "name": "Bob Chen", + "headUrl": "https://cdn.example.com/bob.png", + "isDel": 0, + "version": "v1.0", + }, +] + +# Page 2 record — voice message, oldest in the history window. +MESSAGE_RECORDS_PAGE2: list[dict[str, Any]] = [ + { + "msgSvrId": "999999", + "type": 34, + "messageType": 34, + "text": "[voice]", + "file": "https://cdn.example.com/voice3.amr", + "md5": "md5voice3", + "duration": 5, + "transcriptionText": "hello from yesterday", + "wechatId": "wxid_staff_alice", + "talker": "wxid_friend_0001", + "oriTalker": "wxid_friend_0001", + "roomid": "", + "mine": 1, + "isSend": 1, + "timestamp": E2, + "createTime": E2, + "name": "Alice Wang", + "headUrl": "https://cdn.example.com/alice.png", + "isDel": 0, + "version": "v1.0", + "paymsgid": "pay-abc-123", + "referMsg": {"text": "earlier msg", "msgSvrId": "888888"}, + }, +] + + +def make_msg_chats_response( + messages: list[dict[str, Any]], + *, + end: int, + has_last: bool, + trace_id: str = "trace-msg-1", +) -> dict[str, Any]: + """Build a capture-shaped ``msg/chats`` response envelope. + + Shape (EXACT, per Task 1.6 spec):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"messages":[...], "end":, + "hasLast":}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id, + "data": { + "messages": list(messages), + "end": end, + "hasLast": has_last, + }, + } diff --git a/yunke-server/tests/test_parsers.py b/yunke-server/tests/test_parsers.py new file mode 100644 index 0000000..29a5aa5 --- /dev/null +++ b/yunke-server/tests/test_parsers.py @@ -0,0 +1,44 @@ +"""Tests for the message-type normalization parser (Task 1.6). + +The parser must be total — it never raises on weird input — because every +incoming WeChat record runs through it during a scrape, and one malformed +type code must not abort the whole pull. Unknown codes pass through as +``"unknown"`` so the message is still ingested (and recoverable via +``raw_json``); only the ``-2`` date-divider code maps to ``"divider"``, +which the scraper uses as its drop signal. +""" + +from __future__ import annotations + +import pytest + +from app.scraper.parsers import normalize_wechat_type + + +@pytest.mark.parametrize( + "raw,expected", + [ + # Documented codes. + (1, "text"), + (-2, "divider"), + # Phase-0 unknown codes — pass through as "unknown", never dropped. + (3, "unknown"), # image + (34, "unknown"), # voice + (49, "unknown"), # file / link / transfer + (10000, "unknown"), + # Stringified ints (the upstream API may send either form). + ("1", "text"), + ("-2", "divider"), + ("34", "unknown"), + ], +) +def test_normalize_wechat_type_maps_known_and_unknown(raw, expected): + assert normalize_wechat_type(raw) == expected + + +def test_normalize_wechat_type_never_raises(): + """Whatever we throw at it, it returns a string — never raises.""" + for raw in (None, "", "junk", "1.5", [], {}, object(), float("nan")): + out = normalize_wechat_type(raw) + assert isinstance(out, str) + assert out == "unknown" diff --git a/yunke-server/tests/test_scrape_wechat_messages.py b/yunke-server/tests/test_scrape_wechat_messages.py new file mode 100644 index 0000000..48cbaf8 --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_messages.py @@ -0,0 +1,217 @@ +"""Tests for the WeChat message scraper (Task 1.6). + +Drives the real backward-cursor pagination + normalization logic through +``respx`` by wiring a real ``YunkeClient`` (primed with a fake session so the +login flow is skipped) against a mocked transport. Two pages are served: + +* page 1 -> ``messages=[m1, m_datebar(-2), m2]``, ``end=E1``, ``hasLast=true`` +* page 2 -> ``messages=[m3]``, ``end=E2``, ``hasLast=false`` + +The scraper must yield exactly ``[m1, m2, m3]`` (the ``-2`` divider dropped), +map fields per the live-capture names, AND the 2nd request body's ``end`` must +equal ``E1 - 1`` (proving the backward cursor advances by one ms). +""" + +from __future__ import annotations + +import urllib.parse + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_messages +from app.session.yunke_client import YunkeClient +from tests.cassettes.msg_chats_fixture import ( + E1, + E2, + MESSAGE_RECORDS_PAGE1, + MESSAGE_RECORDS_PAGE2, + make_msg_chats_response, +) + +BASE = "https://crm2.yunkecn.com" +MSG_CHATS_URL = f"{BASE}/wechatweb/msg/chats" + +USERID = "a" * 32 +COMPANY = "8m0atd" + +WECHAT_ID = "wxid_staff_alice" +FRIEND_ID = "wxid_friend_0001" +START = 1 +END = 1784518000000 # the initial upper-bound cursor + + +def _primed_client() -> YunkeClient: + """A YunkeClient already in a logged-in state (login flow skipped).""" + return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY) + + +def _form(sent: httpx.Request) -> dict[str, str]: + """Parse the form-encoded body of an outgoing request into a flat dict. + + ``keep_blank_values=True`` so an empty form field (e.g. ``types=``) is + preserved — the real API contract treats the empty ``types`` field as + "no filter", so its presence in the body is part of what we assert. + """ + parsed = urllib.parse.parse_qs(sent.content.decode(), keep_blank_values=True) + return {k: v[0] for k, v in parsed.items()} + + +# --------------------------------------------------------------------------- +# Test A: end-to-end backward-cursor pagination + divider drop + field mapping +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_messages_drops_divider_and_paginates_backward(): + page1 = make_msg_chats_response( + MESSAGE_RECORDS_PAGE1, end=E1, has_last=True, trace_id="trace-1" + ) + page2 = make_msg_chats_response( + MESSAGE_RECORDS_PAGE2, end=E2, has_last=False, trace_id="trace-2" + ) + + route = respx.post(MSG_CHATS_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + ] + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_wechat_messages( + client, WECHAT_ID, FRIEND_ID, START, END + ) + ] + finally: + await client.aclose() + + # --- Exactly 2 calls: hasLast=false on page 2 means no 3rd fetch. --- + assert route.call_count == 2 + + # --- Exactly [m1, m2, m3] yielded (the -2 divider dropped). --- + assert [r["msg_id"] for r in results] == ["1000001", "1000002", "999999"] + assert len(results) == 3 + + r1, r2, r3 = results + + # m1 — text / direct / incoming. + assert r1["msg_id"] == "1000001" + assert r1["channel"] == "wechat" + assert r1["owner_account_id"] == WECHAT_ID + assert r1["talker_id"] == "wxid_friend_0001" + assert r1["ori_talker"] == "wxid_friend_0001" + assert r1["room_id"] == "" # raw value carried through + assert r1["is_send"] == 0 # pulled from `mine` + assert r1["type"] == "text" + assert r1["text"] == "hi alice" + assert r1["timestamp"] == E1 + 600000 + assert r1["sender_name"] == "Alice Wang" + assert r1["sender_head_url"] == "https://cdn.example.com/alice.png" + assert r1["raw_json"] == MESSAGE_RECORDS_PAGE1[0] + + # m2 — image / chatroom / outgoing. + assert r2["msg_id"] == "1000002" + assert r2["room_id"] == "room12345@chatroom" + assert r2["ori_talker"] == "room12345@chatroom" + assert r2["is_send"] == 1 + assert r2["type"] == "unknown" # unmapped code passes through + assert r2["file_url"] == "https://cdn.example.com/img2.jpg" + assert r2["file_th_url"] == "https://cdn.example.com/img2_th.jpg" + assert r2["file_hd_url"] == "https://cdn.example.com/img2_hd.jpg" + assert r2["md5"] == "md5img2" + assert r2["md5_th"] == "md5img2th" + assert r2["md5_hd"] == "md5img2hd" + assert r2["img_path"] == "/tmp/img2.jpg" + assert r2["original_image"] == "/tmp/img2_orig.jpg" + assert r2["pic_type"] == 1 + assert r2["raw_json"] == MESSAGE_RECORDS_PAGE1[2] + + # m3 — voice / transcription. + assert r3["msg_id"] == "999999" + assert r3["type"] == "unknown" + assert r3["duration"] == 5 + assert r3["transcription_text"] == "hello from yesterday" + assert r3["pay_msg_id"] == "pay-abc-123" + assert r3["refer_msg"] == {"text": "earlier msg", "msgSvrId": "888888"} + assert r3["raw_json"] == MESSAGE_RECORDS_PAGE2[0] + + # --- Backward cursor advance: 2nd request body's end == E1 - 1. --- + first_req = route.calls[0].request + second_req = route.calls[1].request + + assert _form(first_req)["end"] == str(END) + assert _form(second_req)["end"] == str(E1 - 1) + + # --- Form-encoding is actually sent (NOT JSON). Content-type overridden. --- + assert ( + first_req.headers["content-type"] + == "application/x-www-form-urlencoded" + ) + form_first = _form(first_req) + assert form_first["wechatId"] == WECHAT_ID + assert form_first["friendId"] == FRIEND_ID + assert form_first["types"] == "" + assert form_first["start"] == str(START) + assert form_first["end"] == str(END) + + +# --------------------------------------------------------------------------- +# Test B: hasLast=false on first page -> stop after one call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_messages_stops_when_has_last_false_on_first_page(): + """A single page with hasLast=false: scraper yields the records and stops.""" + page = make_msg_chats_response( + [MESSAGE_RECORDS_PAGE2[0]], end=E2, has_last=False + ) + route = respx.post(MSG_CHATS_URL).mock( + return_value=httpx.Response(200, json=page) + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_wechat_messages( + client, WECHAT_ID, FRIEND_ID, START, END + ) + ] + finally: + await client.aclose() + + assert [r["msg_id"] for r in results] == ["999999"] + assert route.call_count == 1 + + +# --------------------------------------------------------------------------- +# Test C: empty messages array on the first page -> stop without a 3rd fetch +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_messages_empty_first_page_yields_nothing(): + """An empty messages array is a defensive stop signal — no rows, one call.""" + page = make_msg_chats_response([], end=E1, has_last=True) + route = respx.post(MSG_CHATS_URL).mock( + return_value=httpx.Response(200, json=page) + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_wechat_messages( + client, WECHAT_ID, FRIEND_ID, START, END + ) + ] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1