diff --git a/yunke-server/app/scraper/parsers.py b/yunke-server/app/scraper/parsers.py index c206870..b9d7928 100644 --- a/yunke-server/app/scraper/parsers.py +++ b/yunke-server/app/scraper/parsers.py @@ -1,24 +1,29 @@ """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. +The yunke CRM uses opaque integer type codes (``type`` / ``messageType`` on +WeChat, ``type`` / ``msgType`` on WhatsApp) on 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. -This module is intentionally tiny: one total function that maps the -Phase-0 confirmed codes and falls through to ``"unknown"`` for everything -else. It MUST NOT raise — every record pulled during a scrape passes through +This module is intentionally tiny: one total function per channel that maps +the Phase-0 confirmed codes and falls through to ``"unknown"`` for everything +else. Each 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. +preserved on the record's ``raw_json`` so the persistence layer can recover it +when needed. + +:func:`coerce_int_or_none` is the shared scraper-layer int coercion (mirrors +:func:`app.storage.repos._coerce_int`) used to emit a ready-to-store +``type_raw`` int and to test raw type codes (e.g. the ``-2`` divider drop) +without the scraper taking a DB dependency. """ from __future__ import annotations from typing import Any -__all__ = ["normalize_wechat_type"] +__all__ = ["coerce_int_or_none", "normalize_wechat_type", "normalize_whatsapp_type"] # Phase-0 confirmed WeChat message type codes (see # docs/superpowers/specs/phase0-findings.md §2). Captured by live inspection @@ -63,3 +68,68 @@ def normalize_wechat_type(raw: Any) -> str: except (TypeError, ValueError): return "unknown" return _WECHAT_TYPE_NAMES.get(code, "unknown") + + +# Phase-0 confirmed WhatsApp message type codes (see data dictionary §3 + +# phase0-findings.md §5). Captured by live inspection of the yunke CRM. +# * ``0`` — plain text. +# * ``1`` — image / media. ``mediaName`` is a deferred-download FILENAME +# (not a URL); the media layer resolves it via a separate endpoint. +# * ``-2`` — date divider. NOT a real message; the scraper uses this label as +# its drop signal (filtered out, never yielded, never persisted). +# Other codes pass through as ``"unknown"`` so the message is still ingested +# (raw code kept on ``raw_json``) rather than dropped. Extend this table as +# each is confirmed. +_WHATSAPP_TYPE_NAMES: dict[int, str] = { + 0: "text", + 1: "image", + -2: "divider", +} + + +def normalize_whatsapp_type(raw: Any) -> str: + """Map an upstream WhatsApp ``type`` / ``msgType`` code to a stable string. + + Confirmed mappings: ``0→text``, ``1→image``, ``-2→divider``. Any other / + missing / unparsable value returns ``"unknown"``. + + 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 _WHATSAPP_TYPE_NAMES.get(code, "unknown") + + +def coerce_int_or_none(value: Any) -> int | None: + """Coerce a possibly-string numeric value to ``int``; ``None`` preserved. + + Mirrors :func:`app.storage.repos._coerce_int` so the scraper can emit a + ready-to-store ``type_raw`` int and check raw type codes (e.g. the ``-2`` + divider drop) without depending on the storage layer (the scraper has no + DB dependency by design). A non-numeric value (``"text"`` / ``None`` / + missing) returns ``None``. + + Kept in this module (rather than per-scraper) so both the WeChat and + WhatsApp scrapers share one well-tested coercion instead of each carrying + a private copy. + """ + if value is None: + return None + if isinstance(value, bool): + return int(value) + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + if isinstance(value, str): + s = value.strip() + if not s: + return None + try: + return int(s) + except ValueError: + return None + return None diff --git a/yunke-server/app/scraper/whatsapp.py b/yunke-server/app/scraper/whatsapp.py new file mode 100644 index 0000000..d2d3a0b --- /dev/null +++ b/yunke-server/app/scraper/whatsapp.py @@ -0,0 +1,460 @@ +"""WhatsApp scrapers (M3). + +Endpoints covered: + +* :func:`scrape_whatsapp_accounts` — wraps ``POST /qqweb/WhatsAppUser/businessSalesList`` + and yields normalized :class:`app.db.models.WhatsappAccount` dicts (plus the + embedded ``userId`` / ``userName`` staff pair). Page-number pagination. +* :func:`scrape_whatsapp_friends` — wraps ``POST /qqweb/WhatsAppUser/businessFriendList`` + and yields normalized dicts ready to be upserted into + :class:`app.db.models.Customer` with ``channel="whatsapp"``. Page-number + pagination. +* :func:`scrape_whatsapp_messages` — wraps ``POST /qqweb/WhatsAppUser/chatRecords`` + and yields normalized chat-message dicts with ``channel="whatsapp"``. + Backward-cursor pagination by timestamp (the same strategy as WeChat + ``msg/chats``), but with a **JSON** request body (NOT form-encoded). + +All three reuse :func:`app.scraper.base.paginate_by_page` / +:func:`paginate_by_cursor` and emit snake_case dicts + a verbatim ``raw_json`` +copy, mirroring :mod:`app.scraper.wechat`. + +Envelope difference from WeChat list endpoints: rows sit under +``data.overviews`` (NOT ``data.data``) and the declared row count is +``data.totalCount`` (NOT ``data.total``); the page count is ``data.pageCount``. +The same envelope is shared by ``businessSalesList`` and ``businessFriendList``. + +Auth is the same crm2 session cookie the WeChat scrapers use (Phase-0 §1 +confirmed cookie-only auth suffices for every data endpoint). +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any + +from app.scraper.base import paginate_by_cursor, paginate_by_page +from app.scraper.parsers import coerce_int_or_none, normalize_whatsapp_type +from app.session.yunke_client import YunkeClient + +__all__ = [ + "scrape_whatsapp_accounts", + "scrape_whatsapp_friends", + "scrape_whatsapp_messages", +] + +WHATSAPP_ACCOUNTS_PATH = "/qqweb/WhatsAppUser/businessSalesList" +WHATSAPP_FRIENDS_PATH = "/qqweb/WhatsAppUser/businessFriendList" +WHATSAPP_MESSAGES_PATH = "/qqweb/WhatsAppUser/chatRecords" + + +# =========================================================================== +# Account scraper (M3) — POST /qqweb/WhatsAppUser/businessSalesList +# =========================================================================== + + +def _parse_wa_account_total(resp: Any) -> int | None: + """Read the declared row count from the ``businessSalesList`` envelope. + + Shape: ``data.totalCount`` (an int). The page count is ``data.pageCount`` + but the paginator only consumes the row-count ceiling; the empty-page + signal remains the primary stop. Returns ``None`` if the path is absent so + the paginator falls back to the empty-page stop signal. + """ + try: + body = resp.json() + except (AttributeError, ValueError): + return None + data = body.get("data") if isinstance(body, dict) else None + if not isinstance(data, dict): + return None + total = data.get("totalCount") + if isinstance(total, bool): + return None + return int(total) if isinstance(total, (int, float)) else None + + +def _parse_wa_account_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the account list from the ``businessSalesList`` envelope. + + Shape: ``data.overviews`` (a list of records). Missing / non-list values + collapse to an empty list so the paginator sees "empty page". + """ + try: + body = resp.json() + except (AttributeError, ValueError): + return [] + data = body.get("data") if isinstance(body, dict) else None + if not isinstance(data, dict): + return [] + rows = data.get("overviews") + return rows if isinstance(rows, list) else [] + + +def _build_wa_account_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``businessSalesList`` request body for a given page. + + The live capture carries no filter parameters — just the page slice. + """ + return {"pageNum": page_num, "pageSize": page_size} + + +def _normalize_whatsapp_account(record: dict[str, Any]) -> dict[str, Any]: + """Map one ``businessSalesList`` record to the ``WhatsappAccount`` shape. + + Carries the embedded staff pair (``user_id`` <- ``userId``, + ``user_name`` <- ``userName``) so the persistence layer can upsert the + owning :class:`app.db.models.Staff` row alongside the account. Adds + ``raw_json`` (the verbatim upstream record). ``None``-safe: missing fields + become ``None`` rather than raising, so a partial payload does not break + the scrape. + """ + return { + "uid": record.get("uid"), + "user_id": record.get("userId"), + "user_name": record.get("userName"), + "nick_name": record.get("nickName"), + "phone": record.get("phone"), + "sale_phone": record.get("salePhone"), + "sales_bind_phone": record.get("salesBindPhone"), + "contacts_num": record.get("contactsNum"), + "group_num": record.get("groupNum"), + "talker_count": record.get("talkerCount"), + "friend_add_num": record.get("friendAddNum"), + "is_on_job": record.get("isOnJob"), + "last_update": record.get("lastUpdate"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_whatsapp_accounts( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WhatsApp staff-account records from ``businessSalesList``. + + Walks the endpoint page by page until the upstream reports no more rows + (empty ``data.overviews`` array — the primary stop signal), mapping each + record to the ``WhatsappAccount`` column shape plus a verbatim ``raw_json`` + copy. + + The envelope differs from WeChat ``wxUser/list``: rows sit under + ``data.overviews`` (not ``data.data``) and the declared total is + ``data.totalCount`` (not ``data.total``). The page count is + ``data.pageCount``. + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client. + page_size: + Page size to request. Defaults to 50 (matches the observed live page + size for this endpoint). + """ + async for record in paginate_by_page( + client, + WHATSAPP_ACCOUNTS_PATH, + page_size=page_size, + parse_total=_parse_wa_account_total, + parse_rows=_parse_wa_account_rows, + build_body=_build_wa_account_body, + ): + yield _normalize_whatsapp_account(record) + + +# =========================================================================== +# Friend scraper (M3) — POST /qqweb/WhatsAppUser/businessFriendList +# =========================================================================== + + +def _parse_wa_friend_total(resp: Any) -> int | None: + """Read the declared row count from the ``businessFriendList`` envelope. + + Shape: ``data.totalCount`` (live total is 11292 per phase0-findings.md §5). + """ + try: + body = resp.json() + except (AttributeError, ValueError): + return None + data = body.get("data") if isinstance(body, dict) else None + if not isinstance(data, dict): + return None + total = data.get("totalCount") + if isinstance(total, bool): + return None + return int(total) if isinstance(total, (int, float)) else None + + +def _parse_wa_friend_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the friend list from the ``businessFriendList`` envelope. + + Shape: ``data.overviews`` (same envelope as ``businessSalesList``). + """ + try: + body = resp.json() + except (AttributeError, ValueError): + return [] + data = body.get("data") if isinstance(body, dict) else None + if not isinstance(data, dict): + return [] + rows = data.get("overviews") + return rows if isinstance(rows, list) else [] + + +def _build_wa_friend_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``businessFriendList`` request body for a given page.""" + return {"pageNum": page_num, "pageSize": page_size} + + +def _normalize_whatsapp_friend(record: dict[str, Any]) -> dict[str, Any]: + """Map one ``businessFriendList`` record to the ``Customer`` column shape. + + Adds ``channel="whatsapp"``. Field mapping per phase0-findings.md §5: + + * ``friend_id`` <- ``friendUid`` (PK, ``@s.whatsapp.net``) + * ``owner_account_id`` <- ``uid`` (the owning staff WA account) + * ``user_id`` <- ``userId`` (owning staff) + * ``sale_name`` <- ``salesNickName`` (owning staff sale name) + * ``friend_nickname`` <- ``nickName`` + * ``phone`` <- ``phone`` + * ``message_count`` <- ``messageCount`` + * ``last_chat_time`` <- ``lastUpdate`` (ms epoch of last activity) + + Preserves the verbatim record under ``raw_json`` (so the owner-staff + fields ``userName`` / ``salePhone`` / ``salesBindPhone`` / ``createTime`` + and the per-contact counts ``groupNum`` / ``contactsNum`` / ``talkerCount`` + / ``friendAddNum`` / ``sensitiveWordNum`` survive end-to-end even though + they have no direct ``Customer`` column). ``None``-safe on every field. + """ + return { + "friend_id": record.get("friendUid"), + "channel": "whatsapp", + "owner_account_id": record.get("uid"), + "friend_nickname": record.get("nickName"), + "phone": record.get("phone"), + "sale_name": record.get("salesNickName"), + "user_id": record.get("userId"), + "message_count": record.get("messageCount"), + "last_chat_time": record.get("lastUpdate"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_whatsapp_friends( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WhatsApp contact records from ``businessFriendList``. + + Walks the endpoint page by page until the upstream reports no more rows + (empty ``data.overviews`` array — the primary stop signal), mapping each + record to the ``Customer`` column shape with ``channel="whatsapp"`` plus a + verbatim ``raw_json`` copy. + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client. + page_size: + Page size to request. Defaults to 50. + """ + async for record in paginate_by_page( + client, + WHATSAPP_FRIENDS_PATH, + page_size=page_size, + parse_total=_parse_wa_friend_total, + parse_rows=_parse_wa_friend_rows, + build_body=_build_wa_friend_body, + ): + yield _normalize_whatsapp_friend(record) + + +# =========================================================================== +# Message scraper (M3) — POST /qqweb/WhatsAppUser/chatRecords +# =========================================================================== + + +async def _fetch_wa_msg_page( + client: YunkeClient, + *, + uid: str, + friend_uid: str, + start: int, + end_cursor: int, +) -> tuple[list[dict[str, Any]], int, bool]: + """Fetch one backward page of ``chatRecords``. + + Returns the ``(rows, next_cursor, has_more)`` triple expected by + :func:`paginate_by_cursor`. Unlike WeChat ``msg/chats`` this endpoint takes + a **JSON** body (content-type ``application/json``), not form-encoded. + + ``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; ``has_more`` is ``data.hasLast``. + """ + resp = await client.post( + WHATSAPP_MESSAGES_PATH, + json={ + "uid": uid, + "friendUid": friend_uid, + "start": start, + "end": end_cursor, + }, + ) + + 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_whatsapp_message(record: dict[str, Any]) -> dict[str, Any]: + """Map one ``chatRecords`` record to the ``Message`` column shape. + + Adds ``channel="whatsapp"`` and preserves the original record under + ``raw_json``. Per design §4.3 the message type is carried as TWO fields: + + * ``type`` — the normalized enum label (``"text"`` / ``"image"`` / + ``"unknown"`` / ...) via :func:`normalize_whatsapp_type`. + * ``type_raw`` — the original upstream type code (``type`` falling back to + ``msgType``), coerced to ``int | None`` so the raw code survives + end-to-end without round-tripping through ``raw_json``. + + ``mediaName`` (a deferred-download filename, NOT a URL — see phase0 §6) is + carried on ``file_url`` so the media layer can route on ``channel`` / + ``type``; this mirrors the WeChat ``type=3`` image-legacy pattern where an + opaque download key (not a URL) is stored on ``file_url``. ``raw_json`` + preserves it verbatim either way. ``quotedId`` / ``quotedMessage`` collapse + into ``refer_msg`` as a small dict when either is present. No ``room_id`` + for WA 1:1 chats. + + ``None``-safe on every field. + """ + # ``type`` is the documented field; fall back to ``msgType`` (some captures + # populate one, some the other — same disambiguation strategy as WeChat). + raw_type = record.get("type") + if raw_type is None: + raw_type = record.get("msgType") + + # ``createTime`` is the canonical timestamp; fall back to ``dateTime``. + timestamp = record.get("createTime") + if timestamp is None: + timestamp = record.get("dateTime") + + # Quoted-message reference: preserve both id + content when either present + # (a bare None/None pair would just be noise on the JSON column). + quoted_id = record.get("quotedId") + quoted_message = record.get("quotedMessage") + refer_msg = ( + {"quotedId": quoted_id, "quotedMessage": quoted_message} + if quoted_id is not None or quoted_message is not None + else None + ) + + return { + "msg_id": record.get("keyId"), + "channel": "whatsapp", + "owner_account_id": record.get("uid"), + "talker_id": record.get("friendUid"), + "is_send": record.get("isSend"), + "type": normalize_whatsapp_type(raw_type), + "type_raw": coerce_int_or_none(raw_type), + "text": record.get("content"), + "file_url": record.get("mediaName"), + "refer_msg": refer_msg, + "sender_name": record.get("nickName"), + "timestamp": timestamp, + "raw_json": record, + } + + +def _is_wa_divider(record: dict[str, Any]) -> bool: + """True iff this record is a date-divider (raw type code ``-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 M3 + spec we drop a record when EITHER ``type`` or ``msgType`` is ``-2``; all + other unknown codes pass through. + + The check is on the RAW code (not the normalized label) so the drop + decision is robust against any future change to + :func:`normalize_whatsapp_type`'s label vocabulary. + """ + return coerce_int_or_none(record.get("type")) == -2 or ( + coerce_int_or_none(record.get("msgType")) == -2 + ) + + +async def scrape_whatsapp_messages( + client: YunkeClient, + uid: str, + friend_uid: str, + start: int, + end: int, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WhatsApp chat messages, paging backward by timestamp. + + Wraps ``POST /qqweb/WhatsAppUser/chatRecords`` with a **JSON** body + (``{uid, friendUid, start, end}`` — NOT form-encoded like WeChat + ``msg/chats``). 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. + uid: + Owning staff WhatsApp account id (``uid`` JSON field, of the form + ``@s.whatsapp.net``) — maps to ``owner_account_id`` on each + yielded record. + friend_uid: + Customer WhatsApp id (``friendUid`` JSON field) — selects the 1:1 + 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. + """ + async for record in paginate_by_cursor( + lambda end_cursor: _fetch_wa_msg_page( + client, + uid=uid, + friend_uid=friend_uid, + start=start, + end_cursor=end_cursor, + ), + initial_cursor=end, + ): + if _is_wa_divider(record): + continue + yield _normalize_whatsapp_message(record) diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index beff727..ffbde0f 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -63,6 +63,7 @@ from app.db.models import ( PhoneCall, Staff, WechatAccount, + WhatsappAccount, ) __all__ = [ @@ -76,6 +77,7 @@ __all__ = [ "upsert_staff_from_account", "upsert_voice_call", "upsert_wechat_account", + "upsert_whatsapp_account", ] @@ -411,6 +413,25 @@ async def upsert_staff_from_account(session: AsyncSession, rec: dict[str, Any]) await upsert_staff(session, staff_rec) +async def upsert_whatsapp_account(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert a :class:`WhatsappAccount` AND its embedded Staff row. + + PK = ``uid`` (of the form ``@s.whatsapp.net``). Mirrors + :func:`upsert_wechat_account`: the companion :func:`upsert_staff_from_account` + covers the embedded staff row (``user_id`` + ``user_name``); this helper + calls it automatically so an account scrape that lands accounts also lands + the owning staff — callers do not need to upsert staff separately. + + The WA account payload (unlike the WeChat one) DOES carry ``sale_phone``, + so the embedded staff row picks it up via the shared staff-from-account + helper. The staff upsert is best-effort: an account record with no + ``user_id`` still writes the account row; the staff upsert silently + no-ops. + """ + await _upsert_one(session, WhatsappAccount, rec) + await upsert_staff_from_account(session, rec) + + # --------------------------------------------------------------------------- # ChatGroup + GroupMember upsert (M2) # --------------------------------------------------------------------------- diff --git a/yunke-server/tests/cassettes/wa_accounts_fixture.py b/yunke-server/tests/cassettes/wa_accounts_fixture.py new file mode 100644 index 0000000..61e28e3 --- /dev/null +++ b/yunke-server/tests/cassettes/wa_accounts_fixture.py @@ -0,0 +1,148 @@ +"""Fixture payloads for ``POST /qqweb/WhatsAppUser/businessSalesList`` (M3). + +These records mirror the real field names captured from the live CRM (see the +data dictionary §3 and phase0-findings.md §5): + +``uid`` (PK, ``@s.whatsapp.net``), ``userId`` / ``userName`` (owning +staff real name), ``nickName``, ``phone`` / ``salePhone`` / ``salesBindPhone``, +``contactsNum`` / ``groupNum`` / ``talkerCount`` / ``friendAddNum``, +``isOnJob``, ``lastUpdate``. + +The envelope differs from the WeChat list endpoints: rows sit under +``data.overviews`` (NOT ``data.data``) and the declared row count is +``data.totalCount`` (NOT ``data.total``); the page count is ``data.pageCount``:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"page":.., "pageCount":.., "totalCount":.., "overviews":[...]}} + +The cassette ships two pages so a test can drive the page-number paginator +across the empty-page stop signal: + +* page 1 -> 2 of the 4 fixture records, ``totalCount=4`` +* page 2 -> the other 2 records, ``totalCount=4`` +* (page 3 -> empty ``overviews`` array, served inline by the test) + +Records carry every field with a target ``WhatsappAccount`` column so the test +can assert a full mapping end-to-end, plus the embedded ``userId`` + ``userName`` +staff pair (the owning employee) so the embedded-staff repo upsert can be +exercised against capture-shaped input (including a no-``userId`` orphan to +exercise the best-effort skip path). +""" + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped WhatsApp staff-account records (4 total across 2 pages). +# --------------------------------------------------------------------------- + +WA_ACCOUNT_RECORDS: list[dict[str, Any]] = [ + { + "uid": "8613800000001@s.whatsapp.net", + "userId": "wa_staff_001", + "userName": "Alice Wang", + "nickName": "Alice WA", + "phone": "8613800000001", + "salePhone": "8613800000001", + "salesBindPhone": "8613900000001", + "contactsNum": 421, + "groupNum": 12, + "talkerCount": 38, + "friendAddNum": 7, + "isOnJob": 1, + "lastUpdate": 1700000000123, + "version": "v1.20260731.alice.wa", + }, + { + "uid": "8613800000002@s.whatsapp.net", + "userId": "wa_staff_002", + "userName": "Bob Chen", + "nickName": "Bob WA", + "phone": "8613800000002", + "salePhone": "8613800000002", + "salesBindPhone": "8613900000002", + "contactsNum": 58, + "groupNum": 2, + "talkerCount": 9, + "friendAddNum": 1, + "isOnJob": 1, + "lastUpdate": 1700000000456, + "version": "v1.20260731.bob.wa", + }, + { + "uid": "8613800000003@s.whatsapp.net", + "userId": "wa_staff_003", + "userName": "Carol Liu", + "nickName": "Carol WA", + "phone": "8613800000003", + "salePhone": "8613800000003", + "salesBindPhone": "8613900000003", + "contactsNum": 1899, + "groupNum": 64, + "talkerCount": 210, + "friendAddNum": 33, + "isOnJob": 0, # departed staff — exercises the falsy isOnJob path + "lastUpdate": 1700000000789, + "version": "v1.20260731.carol.wa", + }, + { + # Account with NO embedded staff (userId missing) — the scraper must + # still yield it; the embedded-staff repo upsert must skip it cleanly. + "uid": "8613800000004@s.whatsapp.net", + "userId": None, + "userName": None, + "nickName": "Orphan WA", + "phone": "8613800000004", + "salePhone": "8613800000004", + "salesBindPhone": "8613900000004", + "contactsNum": 3, + "groupNum": 0, + "talkerCount": 1, + "friendAddNum": 0, + "isOnJob": 0, + "lastUpdate": 1700000001000, + "version": "v1.20260731.orphan.wa", + }, +] + +# Default page size for the fixture (and the scraper). totalCount=4 means the +# fixture fills exactly 2 pages of size 2 plus a terminating empty page. +TOTAL = len(WA_ACCOUNT_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_wa_accounts_response( + records: list[dict[str, Any]], + *, + total_count: int, + page: int = 1, + page_size: int = DEFAULT_PAGE_SIZE, + page_count: int | None = None, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``businessSalesList`` response envelope. + + Shape:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"page":.., "pageCount":.., "totalCount":.., "overviews":[...]}} + + ``page_count`` defaults to ``ceil(total_count / page_size)`` when omitted. + """ + if page_count is None: + import math + + page_count = math.ceil(total_count / page_size) if page_size else 1 + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-wa-acct-{page}", + "data": { + "page": page, + "pageCount": page_count, + "totalCount": total_count, + "overviews": list(records), + }, + } diff --git a/yunke-server/tests/cassettes/wa_friends_fixture.py b/yunke-server/tests/cassettes/wa_friends_fixture.py new file mode 100644 index 0000000..119e5d6 --- /dev/null +++ b/yunke-server/tests/cassettes/wa_friends_fixture.py @@ -0,0 +1,129 @@ +"""Fixture payloads for ``POST /qqweb/WhatsAppUser/businessFriendList`` (M3). + +These records mirror the real field names captured from the live CRM (see +phase0-findings.md §5; live total = **11292**): + +``uid`` (owning staff ``@s.whatsapp.net``), ``userId`` / ``userName`` / +``salePhone`` / ``salesNickName`` / ``salesBindPhone`` (owning staff), +``friendUid`` (customer -> ``friend_id`` PK), ``nickName`` (customer nickname), +``phone``, ``createTime`` / ``lastUpdate``, ``groupNum`` / ``contactsNum`` / +``talkerCount`` / ``friendAddNum`` / ``messageCount`` / ``sensitiveWordNum``. + +The envelope is the SAME shape as ``businessSalesList``: rows under +``data.overviews``, declared count under ``data.totalCount``, page count under +``data.pageCount``:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"page":.., "pageCount":.., "totalCount":.., "overviews":[...]}} + +The fixture's ``TOTAL`` is pinned to the record count (not the live 11292) so +the paginator's over-fetch ceiling does not fire early in the multi-page test. +""" + +from __future__ import annotations + +import math +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped WhatsApp contact records (3 total across 2 pages). +# --------------------------------------------------------------------------- + +WA_FRIEND_RECORDS: list[dict[str, Any]] = [ + { + "uid": "8613800000001@s.whatsapp.net", # owning staff account + "userId": "wa_staff_001", + "userName": "Alice Wang", + "salePhone": "8613800000001", + "salesNickName": "Alice Sale", + "salesBindPhone": "8613900000001", + "friendUid": "8613800000111@s.whatsapp.net", # customer -> friend_id + "nickName": "Customer One", + "phone": "8613800000111", + "createTime": 1699000000000, + "lastUpdate": 1700000000123, + "groupNum": 2, + "contactsNum": 421, + "talkerCount": 38, + "friendAddNum": 1, + "messageCount": 57, + "sensitiveWordNum": 0, + "version": "v1.20260731.cust1", + }, + { + "uid": "8613800000001@s.whatsapp.net", + "userId": "wa_staff_001", + "userName": "Alice Wang", + "salePhone": "8613800000001", + "salesNickName": "Alice Sale", + "salesBindPhone": "8613900000001", + "friendUid": "8613800000222@s.whatsapp.net", + "nickName": "Customer Two", + "phone": "8613800000222", + "createTime": 1699100000000, + "lastUpdate": 1700000000456, + "groupNum": 1, + "contactsNum": 421, + "talkerCount": 38, + "friendAddNum": 1, + "messageCount": 9, + "sensitiveWordNum": 0, + "version": "v1.20260731.cust2", + }, + { + # Contact owned by a different staff account. + "uid": "8613800000002@s.whatsapp.net", + "userId": "wa_staff_002", + "userName": "Bob Chen", + "salePhone": "8613800000002", + "salesNickName": "Bob Sale", + "salesBindPhone": "8613900000002", + "friendUid": "8613800000333@s.whatsapp.net", + "nickName": "Customer Three", + "phone": "8613800000333", + "createTime": 1699200000000, + "lastUpdate": 1700000000789, + "groupNum": 0, + "contactsNum": 58, + "talkerCount": 9, + "friendAddNum": 0, + "messageCount": 4, + "sensitiveWordNum": 1, + "version": "v1.20260731.cust3", + }, +] + +TOTAL = len(WA_FRIEND_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_wa_friends_response( + records: list[dict[str, Any]], + *, + total_count: int, + page: int = 1, + page_size: int = DEFAULT_PAGE_SIZE, + page_count: int | None = None, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``businessFriendList`` response envelope. + + Shape (same as ``businessSalesList``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"page":.., "pageCount":.., "totalCount":.., "overviews":[...]}} + """ + if page_count is None: + page_count = math.ceil(total_count / page_size) if page_size else 1 + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-wa-friend-{page}", + "data": { + "page": page, + "pageCount": page_count, + "totalCount": total_count, + "overviews": list(records), + }, + } diff --git a/yunke-server/tests/cassettes/wa_messages_fixture.py b/yunke-server/tests/cassettes/wa_messages_fixture.py new file mode 100644 index 0000000..85b30b5 --- /dev/null +++ b/yunke-server/tests/cassettes/wa_messages_fixture.py @@ -0,0 +1,142 @@ +"""Fixture payloads for ``POST /qqweb/WhatsAppUser/chatRecords`` (M3). + +These records mirror the real field names captured from the live CRM (see the +data dictionary §3 + phase0-findings.md §6): + +``keyId`` (PK), ``type`` / ``msgType`` (``0``=text, ``1``=image/media, +``-2``=date divider), ``content`` (text), ``mediaName`` (media FILENAME — NOT a +URL; deferred download), ``nickName`` (sender), ``uid`` (owning staff account), +``friendUid`` (the other party), ``isSend``, ``createTime`` / ``dateTime``, +``timezone``, ``companyCode``, ``quotedId`` / ``quotedMessage`` (quoted reply). + +The request is a **JSON body** ``{uid, friendUid, start, end}`` (ms epochs) — +NOT form-encoded like WeChat ``msg/chats``. The response pages BACKWARD: each +page 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:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"messages":[...], "start":.., "end":.., "hasNext":.., "hasLast":..}} + +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``. +""" + +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/incoming, +# m_datebar is a -2 divider (must be filtered), m2 is image/outgoing. +WA_MESSAGE_RECORDS_PAGE1: list[dict[str, Any]] = [ + { # m1 — text message, incoming. + "keyId": "wa_1000001", + "type": 0, + "msgType": 0, + "content": "hello from whatsapp", + "nickName": "Customer One", + "uid": "8613800000001@s.whatsapp.net", + "friendUid": "8613800000111@s.whatsapp.net", + "isSend": 0, + "createTime": E1 + 600000, + "dateTime": E1 + 600000, + "timezone": "Asia/Shanghai", + "companyCode": "8m0atd", + "version": "v1.0", + }, + { # date divider — type == -2; scraper MUST drop this, never yield it. + "keyId": "wa_1000000", + "type": -2, + "msgType": -2, + "content": "2026-01-01", + "nickName": "", + "uid": "8613800000001@s.whatsapp.net", + "friendUid": "8613800000111@s.whatsapp.net", + "isSend": 0, + "createTime": E1 + 300000, + "dateTime": E1 + 300000, + "timezone": "Asia/Shanghai", + "companyCode": "8m0atd", + "version": "v1.0", + }, + { # m2 — image message, outgoing, with a media filename (deferred download). + "keyId": "wa_1000002", + "type": 1, + "msgType": 1, + "content": "[image]", + "mediaName": "8613800000001@s.whatsapp.net_wa_1000002.jpg", + "nickName": "Alice Wang", + "uid": "8613800000001@s.whatsapp.net", + "friendUid": "8613800000111@s.whatsapp.net", + "isSend": 1, + "createTime": E1 + 100, + "dateTime": E1 + 100, + "timezone": "Asia/Shanghai", + "companyCode": "8m0atd", + "version": "v1.0", + }, +] + +# Page 2 record — text message with a quoted reply; oldest in the window. +WA_MESSAGE_RECORDS_PAGE2: list[dict[str, Any]] = [ + { + "keyId": "wa_999999", + "type": 0, + "msgType": 0, + "content": "replying to an earlier message", + "nickName": "Customer One", + "uid": "8613800000001@s.whatsapp.net", + "friendUid": "8613800000111@s.whatsapp.net", + "isSend": 0, + "createTime": E2, + "dateTime": E2, + "timezone": "Asia/Shanghai", + "companyCode": "8m0atd", + "quotedId": "wa_888888", + "quotedMessage": "the earlier message text", + "version": "v1.0", + }, +] + + +def make_wa_messages_response( + messages: list[dict[str, Any]], + *, + end: int, + has_last: bool, + start: int = 1, + has_next: bool = False, + trace_id: str = "trace-wa-msg-1", +) -> dict[str, Any]: + """Build a capture-shaped ``chatRecords`` response envelope. + + Shape (EXACT, per M3 spec):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"messages":[...], "start":.., "end":, + "hasNext":, "hasLast":}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id, + "data": { + "messages": list(messages), + "start": start, + "end": end, + "hasNext": has_next, + "hasLast": has_last, + }, + } diff --git a/yunke-server/tests/test_parsers.py b/yunke-server/tests/test_parsers.py index ea72468..10062a3 100644 --- a/yunke-server/tests/test_parsers.py +++ b/yunke-server/tests/test_parsers.py @@ -12,7 +12,11 @@ from __future__ import annotations import pytest -from app.scraper.parsers import normalize_wechat_type +from app.scraper.parsers import ( + coerce_int_or_none, + normalize_wechat_type, + normalize_whatsapp_type, +) @pytest.mark.parametrize( @@ -52,3 +56,62 @@ def test_normalize_wechat_type_never_raises(): out = normalize_wechat_type(raw) assert isinstance(out, str) assert out == "unknown" + + +# =========================================================================== +# WhatsApp message-type normalization (M3) +# =========================================================================== + + +@pytest.mark.parametrize( + "raw,expected", + [ + # Phase-0 confirmed WhatsApp codes (see data dictionary §3): + (0, "text"), + (1, "image"), # image / media + (-2, "divider"), # date separator — NOT a real message (scraper drops it) + # Unconfirmed / other media codes — pass through as "unknown", never dropped. + (2, "unknown"), + (3, "unknown"), + (13, "unknown"), + (100, "unknown"), + # Stringified ints (the upstream API may send either form). + ("0", "text"), + ("1", "image"), + ("-2", "divider"), + ("2", "unknown"), + ], +) +def test_normalize_whatsapp_type_maps_known_and_unknown(raw, expected): + assert normalize_whatsapp_type(raw) == expected + + +def test_normalize_whatsapp_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_whatsapp_type(raw) + assert isinstance(out, str) + assert out == "unknown" + + +# =========================================================================== +# coerce_int_or_none — shared scraper helper (M3) +# =========================================================================== + + +def test_coerce_int_or_none_handles_documented_cases(): + """``type_raw``-style fields: str -> int; None preserved; 0 preserved.""" + assert coerce_int_or_none(None) is None + assert coerce_int_or_none(0) == 0 + assert coerce_int_or_none(2) == 2 + assert coerce_int_or_none(-2) == -2 + assert coerce_int_or_none("0") == 0 + assert coerce_int_or_none("2") == 2 + assert coerce_int_or_none("-2") == -2 + + +def test_coerce_int_or_none_returns_none_for_non_numeric_string(): + """A non-numeric value (e.g. a label) -> None.""" + assert coerce_int_or_none("text") is None + assert coerce_int_or_none("") is None + assert coerce_int_or_none("unknown") is None diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index b5a8312..c709656 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -31,6 +31,7 @@ from app.db.models import ( PhoneCall, Staff, WechatAccount, + WhatsappAccount, ) from app.storage.repos import ( _coerce_bool, @@ -44,6 +45,7 @@ from app.storage.repos import ( upsert_staff_from_account, upsert_voice_call, upsert_wechat_account, + upsert_whatsapp_account, ) # --------------------------------------------------------------------------- @@ -1004,3 +1006,216 @@ async def test_upsert_voice_call_null_record_file_and_coerces_ints(): assert fetched.incoming_call == 1 finally: await engine.dispose() + + +# --------------------------------------------------------------------------- +# WhatsappAccount + embedded Staff upsert (M3) +# --------------------------------------------------------------------------- + + +async def test_upsert_whatsapp_account_writes_account_and_embedded_staff(): + """WhatsappAccount upsert creates the account row AND the embedded Staff row.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "uid": "8613800000001@s.whatsapp.net", + "user_id": "wa_u1", + "user_name": "Alice Wang", + "nick_name": "Alice WA", + "phone": "8613800000001", + "sale_phone": "8613800000001", + "sales_bind_phone": "8613900000001", + "contacts_num": 421, + "group_num": 12, + "talker_count": 38, + "friend_add_num": 7, + "is_on_job": 1, + "last_update": 1700000000123, + } + await upsert_whatsapp_account(session, rec) + await session.commit() + + acct = ( + await session.execute( + sa.select(WhatsappAccount).where( + WhatsappAccount.uid == "8613800000001@s.whatsapp.net" + ) + ) + ).scalar_one() + assert acct.user_id == "wa_u1" + assert acct.user_name == "Alice Wang" + assert acct.sale_phone == "8613800000001" + assert acct.sales_bind_phone == "8613900000001" + assert acct.contacts_num == 421 + assert acct.is_on_job is True + assert acct.last_update == 1700000000123 + + # Embedded staff row created (PK = user_id), name forwarded. + staff = ( + await session.execute( + sa.select(Staff).where(Staff.user_id == "wa_u1") + ) + ).scalar_one() + assert staff.name == "Alice Wang" + assert staff.is_on_job is True + assert staff.sale_phone == "8613800000001" + finally: + await engine.dispose() + + +async def test_upsert_whatsapp_account_is_idempotent_and_refreshes_staff(): + """Re-upserting the account refreshes BOTH account + embedded staff rows.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "uid": "8613800000002@s.whatsapp.net", + "user_id": "wa_u2", + "user_name": "Old Name", + "nick_name": "Old Nick", + "is_on_job": 1, + } + rec2 = { + "uid": "8613800000002@s.whatsapp.net", + "user_id": "wa_u2", + "user_name": "New Name", + "nick_name": "New Nick", + "is_on_job": 0, # staff departed since last scrape + } + await upsert_whatsapp_account(session, rec1) + await session.commit() + await upsert_whatsapp_account(session, rec2) + await session.commit() + + # One account row, refreshed. + acct_rows = ( + await session.execute( + sa.select(WhatsappAccount) + .where(WhatsappAccount.uid == "8613800000002@s.whatsapp.net") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(acct_rows) == 1 + assert acct_rows[0].nick_name == "New Nick" + + # One staff row, refreshed (not duplicated). + staff_rows = ( + await session.execute( + sa.select(Staff) + .where(Staff.user_id == "wa_u2") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(staff_rows) == 1 + assert staff_rows[0].name == "New Name" + assert staff_rows[0].is_on_job is False + finally: + await engine.dispose() + + +async def test_upsert_whatsapp_account_without_user_id_skips_staff(): + """Account with no user_id is written; embedded-staff upsert is skipped.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "uid": "8613800000009@s.whatsapp.net", + "nick_name": "Orphan WA", + # no user_id / user_name + } + await upsert_whatsapp_account(session, rec) + await session.commit() + + # Account row written. + acct = ( + await session.execute( + sa.select(WhatsappAccount).where( + WhatsappAccount.uid == "8613800000009@s.whatsapp.net" + ) + ) + ).scalar_one() + assert acct.nick_name == "Orphan WA" + + # No staff rows at all. + staff_rows = ( + await session.execute(sa.select(Staff)) + ).scalars().all() + assert staff_rows == [] + finally: + await engine.dispose() + + +async def test_upsert_customer_with_whatsapp_channel(): + """WA contacts flow through the channel-generic upsert_customer. + + The ``Customer`` table is shared across WeChat / WhatsApp / SMS peers; the + per-row ``channel`` value disambiguates them. This pins that a WA-shaped + record (``channel="whatsapp"``) upserts cleanly against the same PK contract. + """ + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "friend_id": "8613800000111@s.whatsapp.net", + "channel": "whatsapp", + "owner_account_id": "8613800000001@s.whatsapp.net", + "friend_nickname": "Customer One", + "phone": "8613800000111", + "sale_name": "Alice Sale", + "user_id": "wa_staff_001", + "message_count": 57, + } + await upsert_customer(session, rec) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Customer).where( + Customer.friend_id == "8613800000111@s.whatsapp.net" + ) + ) + ).scalar_one() + assert fetched.channel == "whatsapp" + assert fetched.owner_account_id == "8613800000001@s.whatsapp.net" + assert fetched.friend_nickname == "Customer One" + assert fetched.sale_name == "Alice Sale" + assert fetched.message_count == 57 + finally: + await engine.dispose() + + +async def test_upsert_messages_with_whatsapp_channel(): + """WA messages flow through the channel-generic upsert_messages.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "msg_id": "wa_msg_1", + "channel": "whatsapp", + "owner_account_id": "8613800000001@s.whatsapp.net", + "talker_id": "8613800000111@s.whatsapp.net", + "is_send": 0, + "type": "text", + "type_raw": 0, + "text": "hi", + } + await upsert_messages(session, [rec]) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "wa_msg_1") + ) + ).scalar_one() + assert fetched.channel == "whatsapp" + assert fetched.type == "text" + assert fetched.type_raw == 0 + assert fetched.is_send is False + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_scrape_whatsapp_accounts.py b/yunke-server/tests/test_scrape_whatsapp_accounts.py new file mode 100644 index 0000000..5ff1e77 --- /dev/null +++ b/yunke-server/tests/test_scrape_whatsapp_accounts.py @@ -0,0 +1,217 @@ +"""Tests for the WhatsApp staff-account scraper (M3). + +Drives the real page-number 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. + +The fixture ships 4 records across 2 pages of size 2, plus a terminating +empty page (the documented primary stop signal for page-number endpoints): + +* page 1 -> 2 records, ``totalCount=4`` (rows under ``data.overviews``) +* page 2 -> 2 records, ``totalCount=4`` +* page 3 -> empty ``overviews`` array (no more rows) + +The scraper must yield exactly the 4 records (normalized to ``WhatsappAccount`` +snake_case columns + ``raw_json`` + the embedded staff pair), make exactly 3 +calls, and stop without a 4th. The empty-page stop is the primary signal; +``totalCount=4`` is only a defensive over-fetch ceiling that must NOT trigger +early (we yield exactly 4 == total, so the strict-``>`` ceiling never fires). +""" + +from __future__ import annotations + +import json as _json + +import httpx +import respx + +from app.scraper.whatsapp import scrape_whatsapp_accounts +from app.session.yunke_client import YunkeClient +from tests.cassettes.wa_accounts_fixture import ( + DEFAULT_PAGE_SIZE, + TOTAL, + WA_ACCOUNT_RECORDS, + make_wa_accounts_response, +) + +BASE = "https://crm2.yunkecn.com" +WA_ACCOUNTS_URL = f"{BASE}/qqweb/WhatsAppUser/businessSalesList" + +USERID = "a" * 32 +COMPANY = "8m0atd" + + +def _primed_client() -> YunkeClient: + """A YunkeClient already in a logged-in state (login flow skipped).""" + return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY) + + +# --------------------------------------------------------------------------- +# End-to-end: pagination + field mapping + embedded staff + raw_json +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_accounts_yields_records_and_paginates(): + """Two full pages + a terminating empty page -> 4 records, 3 calls.""" + page1 = make_wa_accounts_response( + WA_ACCOUNT_RECORDS[:2], total_count=TOTAL, page=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_wa_accounts_response( + WA_ACCOUNT_RECORDS[2:], total_count=TOTAL, page=2, page_size=DEFAULT_PAGE_SIZE + ) + page3 = make_wa_accounts_response( + [], total_count=TOTAL, page=3, page_size=DEFAULT_PAGE_SIZE + ) + + route = respx.post(WA_ACCOUNTS_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + httpx.Response(200, json=page3), + ] + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_whatsapp_accounts(client, page_size=DEFAULT_PAGE_SIZE) + ] + finally: + await client.aclose() + + # Pagination: exactly 3 calls (page1, page2, empty-page stop). + assert route.call_count == 3 + + # Exactly the 4 fixture records (not duplicated, not dropped). + assert len(results) == 4 + assert {r["uid"] for r in results} == {rec["uid"] for rec in WA_ACCOUNT_RECORDS} + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, WA_ACCOUNT_RECORDS, strict=True): + # Identity PK. + assert result["uid"] == record["uid"] + + # Embedded staff pair (the owning employee). + assert result["user_id"] == record["userId"] + assert result["user_name"] == record["userName"] + + # Naming + phones. + assert result["nick_name"] == record["nickName"] + assert result["phone"] == record["phone"] + assert result["sale_phone"] == record["salePhone"] + assert result["sales_bind_phone"] == record["salesBindPhone"] + + # Counts. + assert result["contacts_num"] == record["contactsNum"] + assert result["group_num"] == record["groupNum"] + assert result["talker_count"] == record["talkerCount"] + assert result["friend_add_num"] == record["friendAddNum"] + + # Status / timestamps. + assert result["is_on_job"] == record["isOnJob"] + assert result["last_update"] == record["lastUpdate"] + + # Version carried verbatim; raw_json preserves the original record. + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +# --------------------------------------------------------------------------- +# Pagination body: pageNum/pageSize advanced correctly +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_accounts_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_wa_accounts_response( + WA_ACCOUNT_RECORDS[:1], total_count=TOTAL, page=1, page_size=1 + ) + page2 = make_wa_accounts_response([], total_count=TOTAL, page=2, page_size=1) + route = respx.post(WA_ACCOUNTS_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_whatsapp_accounts(client, page_size=1)] + finally: + await client.aclose() + + # One record on page 1, empty page 2 stops the loop. + assert len(results) == 1 + assert route.call_count == 2 + + # First request: pageNum=1. + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["pageNum"] == 1 + assert body1["pageSize"] == 1 + + # Second request: pageNum advanced to 2. + body2 = _json.loads(route.calls[1].request.content.decode()) + assert body2["pageNum"] == 2 + assert body2["pageSize"] == 1 + + +# --------------------------------------------------------------------------- +# Empty first page -> no yields, single call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_accounts_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_wa_accounts_response([], total_count=0, page=1) + route = respx.post(WA_ACCOUNTS_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_whatsapp_accounts(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1 + + +# --------------------------------------------------------------------------- +# Envelope: rows are read from data.overviews (NOT data.data) +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_accounts_reads_overviews_envelope(): + """Rows under ``data.overviews`` are picked up (not the WeChat ``data.data``).""" + # A response that ALSO carries an empty ``data.data`` list must not confuse + # the scraper — it must read ``overviews`` and yield the record. + page = make_wa_accounts_response( + WA_ACCOUNT_RECORDS[:1], total_count=1, page=1, page_size=1 + ) + page["data"]["data"] = [] # WeChat-style key present but empty + respx.post(WA_ACCOUNTS_URL).mock(return_value=httpx.Response(200, json=page)) + # Serve an empty page next so the paginator stops. + respx.post(WA_ACCOUNTS_URL).mock( + side_effect=[ + httpx.Response(200, json=page), + httpx.Response( + 200, + json=make_wa_accounts_response([], total_count=1, page=2, page_size=1), + ), + ] + ) + + client = _primed_client() + try: + results = [r async for r in scrape_whatsapp_accounts(client, page_size=1)] + finally: + await client.aclose() + + assert len(results) == 1 + assert results[0]["uid"] == WA_ACCOUNT_RECORDS[0]["uid"] diff --git a/yunke-server/tests/test_scrape_whatsapp_friends.py b/yunke-server/tests/test_scrape_whatsapp_friends.py new file mode 100644 index 0000000..ac9d810 --- /dev/null +++ b/yunke-server/tests/test_scrape_whatsapp_friends.py @@ -0,0 +1,169 @@ +"""Tests for the WhatsApp contact-list scraper (M3). + +Drives the real page-number 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. + +The fixture ships 3 records across 2 pages of size 2, plus a terminating +empty page (the documented primary stop signal): + +* page 1 -> 2 records, ``totalCount=3`` (rows under ``data.overviews``) +* page 2 -> 1 record, ``totalCount=3`` +* page 3 -> empty ``overviews`` array (no more rows) + +The scraper must yield exactly the 3 records normalized to the ``Customer`` +column shape with ``channel="whatsapp"``, ``friend_id`` <- ``friendUid`` and +``owner_account_id`` <- ``uid``. +""" + +from __future__ import annotations + +import json as _json + +import httpx +import respx + +from app.scraper.whatsapp import scrape_whatsapp_friends +from app.session.yunke_client import YunkeClient +from tests.cassettes.wa_friends_fixture import ( + DEFAULT_PAGE_SIZE, + TOTAL, + WA_FRIEND_RECORDS, + make_wa_friends_response, +) + +BASE = "https://crm2.yunkecn.com" +WA_FRIENDS_URL = f"{BASE}/qqweb/WhatsAppUser/businessFriendList" + +USERID = "a" * 32 +COMPANY = "8m0atd" + + +def _primed_client() -> YunkeClient: + """A YunkeClient already in a logged-in state (login flow skipped).""" + return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY) + + +# --------------------------------------------------------------------------- +# End-to-end: pagination + field mapping + channel/ownership + raw_json +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_friends_yields_normalized_records_and_paginates(): + """Two full pages + a terminating empty page -> 3 records, 3 calls.""" + page1 = make_wa_friends_response( + WA_FRIEND_RECORDS[:2], total_count=TOTAL, page=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_wa_friends_response( + WA_FRIEND_RECORDS[2:], total_count=TOTAL, page=2, page_size=DEFAULT_PAGE_SIZE + ) + page3 = make_wa_friends_response( + [], total_count=TOTAL, page=3, page_size=DEFAULT_PAGE_SIZE + ) + + route = respx.post(WA_FRIENDS_URL).mock( + side_effect=[ + httpx.Response(200, json=page1), + httpx.Response(200, json=page2), + httpx.Response(200, json=page3), + ] + ) + + client = _primed_client() + try: + results = [ + r async for r in scrape_whatsapp_friends(client, page_size=DEFAULT_PAGE_SIZE) + ] + finally: + await client.aclose() + + # Pagination: exactly 3 calls (page1, page2, empty-page stop). + assert route.call_count == 3 + + # Exactly the 3 fixture records (not duplicated, not dropped). + assert len(results) == 3 + assert {r["friend_id"] for r in results} == { + rec["friendUid"] for rec in WA_FRIEND_RECORDS + } + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, WA_FRIEND_RECORDS, strict=True): + # Identity / ownership / channel. + assert result["friend_id"] == record["friendUid"] + assert result["channel"] == "whatsapp" + assert result["owner_account_id"] == record["uid"] + assert result["user_id"] == record["userId"] + + # Naming / phone. + assert result["friend_nickname"] == record["nickName"] + assert result["phone"] == record["phone"] + assert result["sale_name"] == record["salesNickName"] + + # Counts / timestamps. + assert result["message_count"] == record["messageCount"] + assert result["last_chat_time"] == record["lastUpdate"] + + # Version carried verbatim; raw_json preserves the original record. + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +# --------------------------------------------------------------------------- +# Pagination body: pageNum/pageSize advanced correctly +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_friends_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_wa_friends_response( + WA_FRIEND_RECORDS[:1], total_count=TOTAL, page=1, page_size=1 + ) + page2 = make_wa_friends_response([], total_count=TOTAL, page=2, page_size=1) + route = respx.post(WA_FRIENDS_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_whatsapp_friends(client, page_size=1)] + finally: + await client.aclose() + + assert len(results) == 1 + assert route.call_count == 2 + + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["pageNum"] == 1 + assert body1["pageSize"] == 1 + + body2 = _json.loads(route.calls[1].request.content.decode()) + assert body2["pageNum"] == 2 + assert body2["pageSize"] == 1 + + +# --------------------------------------------------------------------------- +# Empty first page -> no yields, single call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_friends_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_wa_friends_response([], total_count=0, page=1) + route = respx.post(WA_FRIENDS_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_whatsapp_friends(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1 diff --git a/yunke-server/tests/test_scrape_whatsapp_messages.py b/yunke-server/tests/test_scrape_whatsapp_messages.py new file mode 100644 index 0000000..a7462c8 --- /dev/null +++ b/yunke-server/tests/test_scrape_whatsapp_messages.py @@ -0,0 +1,208 @@ +"""Tests for the WhatsApp message scraper (M3). + +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, set ``channel="whatsapp"``, send a +**JSON** body (``uid``/``friendUid``/``start``/``end``), AND the 2nd request +body's ``end`` must equal ``E1 - 1`` (proving the backward cursor advances by +one ms). +""" + +from __future__ import annotations + +import json as _json +from typing import Any + +import httpx +import respx + +from app.scraper.whatsapp import scrape_whatsapp_messages +from app.session.yunke_client import YunkeClient +from tests.cassettes.wa_messages_fixture import ( + E1, + E2, + WA_MESSAGE_RECORDS_PAGE1, + WA_MESSAGE_RECORDS_PAGE2, + make_wa_messages_response, +) + +BASE = "https://crm2.yunkecn.com" +WA_MESSAGES_URL = f"{BASE}/qqweb/WhatsAppUser/chatRecords" + +USERID = "a" * 32 +COMPANY = "8m0atd" + +UID = "8613800000001@s.whatsapp.net" +FRIEND_UID = "8613800000111@s.whatsapp.net" +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 _json_body(sent: httpx.Request) -> dict[str, Any]: + """Parse the JSON body of an outgoing request into a dict.""" + return _json.loads(sent.content.decode()) + + +# --------------------------------------------------------------------------- +# Test A: end-to-end backward-cursor pagination + divider drop + field mapping +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_messages_drops_divider_and_paginates_backward(): + page1 = make_wa_messages_response( + WA_MESSAGE_RECORDS_PAGE1, end=E1, has_last=True, trace_id="trace-wa-1" + ) + page2 = make_wa_messages_response( + WA_MESSAGE_RECORDS_PAGE2, end=E2, has_last=False, trace_id="trace-wa-2" + ) + + route = respx.post(WA_MESSAGES_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_whatsapp_messages( + client, UID, FRIEND_UID, 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] == ["wa_1000001", "wa_1000002", "wa_999999"] + assert len(results) == 3 + + r1, r2, r3 = results + + # m1 — text / incoming. + assert r1["msg_id"] == "wa_1000001" + assert r1["channel"] == "whatsapp" + assert r1["owner_account_id"] == UID + assert r1["talker_id"] == FRIEND_UID + assert r1["is_send"] == 0 + assert r1["type"] == "text" # code 0 -> text (phase0 §3) + assert r1["type_raw"] == 0 # original raw code preserved + assert r1["text"] == "hello from whatsapp" + assert r1["timestamp"] == E1 + 600000 + assert r1["sender_name"] == "Customer One" + assert r1["raw_json"] == WA_MESSAGE_RECORDS_PAGE1[0] + + # m2 — image / outgoing / media filename (deferred download, not a URL). + assert r2["msg_id"] == "wa_1000002" + assert r2["is_send"] == 1 + assert r2["type"] == "image" # code 1 -> image/media (phase0 §3) + assert r2["type_raw"] == 1 + assert r2["text"] == "[image]" + assert r2["file_url"] == "8613800000001@s.whatsapp.net_wa_1000002.jpg" + assert r2["raw_json"] == WA_MESSAGE_RECORDS_PAGE1[2] + + # m3 — text / incoming / quoted reply. + assert r3["msg_id"] == "wa_999999" + assert r3["type"] == "text" + assert r3["type_raw"] == 0 + assert r3["text"] == "replying to an earlier message" + assert r3["refer_msg"] == { + "quotedId": "wa_888888", + "quotedMessage": "the earlier message text", + } + assert r3["raw_json"] == WA_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 + + body1 = _json_body(first_req) + body2 = _json_body(second_req) + + assert body1["end"] == END + assert body2["end"] == E1 - 1 + + # --- JSON body is actually sent (NOT form-encoded). --- + assert first_req.headers["content-type"] == "application/json" + assert body1["uid"] == UID + assert body1["friendUid"] == FRIEND_UID + assert body1["start"] == START + assert body1["end"] == END + # ``start`` is constant across pages (only ``end`` walks backward). + assert body2["start"] == START + assert body2["uid"] == UID + assert body2["friendUid"] == FRIEND_UID + + +# --------------------------------------------------------------------------- +# Test B: hasLast=false on first page -> stop after one call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_whatsapp_messages_stops_when_has_last_false_on_first_page(): + """A single page with hasLast=false: scraper yields the records and stops.""" + page = make_wa_messages_response( + [WA_MESSAGE_RECORDS_PAGE2[0]], end=E2, has_last=False + ) + route = respx.post(WA_MESSAGES_URL).mock( + return_value=httpx.Response(200, json=page) + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_whatsapp_messages( + client, UID, FRIEND_UID, START, END + ) + ] + finally: + await client.aclose() + + assert [r["msg_id"] for r in results] == ["wa_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_whatsapp_messages_empty_first_page_yields_nothing(): + """An empty messages array is a defensive stop signal — no rows, one call.""" + page = make_wa_messages_response([], end=E1, has_last=True) + route = respx.post(WA_MESSAGES_URL).mock( + return_value=httpx.Response(200, json=page) + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_whatsapp_messages( + client, UID, FRIEND_UID, START, END + ) + ] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1