From 2a74d3888d96fcd2e950ef059c8a46cef6a2baac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sat, 1 Aug 2026 13:54:16 +0800 Subject: [PATCH] feat: wechat group+member and talker scrapers Add scrape_wechat_groups (POST /wechatweb/chatroom/getChatroomList) and scrape_wechat_talkers (POST /wechatweb/wxFriends/talkerList), both reusing the page-number paginator. Groups parse the semicolon-joined memberIds into a member_ids_list so upsert_chat_group can mirror GroupMember rows (composite PK channel/room_id/member_id) idempotently alongside the ChatGroup row. Talkers are a sync-planning input (member_count>0 marks a group conversation); no dedicated table. Co-Authored-By: Claude --- yunke-server/app/scraper/wechat.py | 275 +++++++++++++++++- yunke-server/app/storage/repos.py | 62 +++- .../tests/cassettes/chatroom_fixture.py | 137 +++++++++ .../tests/cassettes/talkerList_fixture.py | 112 +++++++ yunke-server/tests/test_repos.py | 186 +++++++++++- .../tests/test_scrape_wechat_groups.py | 231 +++++++++++++++ .../tests/test_scrape_wechat_talkers.py | 164 +++++++++++ 7 files changed, 1163 insertions(+), 4 deletions(-) create mode 100644 yunke-server/tests/cassettes/chatroom_fixture.py create mode 100644 yunke-server/tests/cassettes/talkerList_fixture.py create mode 100644 yunke-server/tests/test_scrape_wechat_groups.py create mode 100644 yunke-server/tests/test_scrape_wechat_talkers.py diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index 831850a..d6f8424 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -15,6 +15,15 @@ Endpoints covered today: yields normalized :class:`app.db.models.WechatAccount` dicts (plus the embedded ``userId``/``userName`` staff pair). Page-number pagination (Task 2.1). +* :func:`scrape_wechat_groups` — wraps ``POST /wechatweb/chatroom/getChatroomList`` + and yields normalized :class:`app.db.models.ChatGroup` dicts, parsing the + semicolon-joined ``memberIds`` into a ``member_ids_list`` so the + persistence layer can mirror :class:`app.db.models.GroupMember` rows + (M2). Page-number pagination. +* :func:`scrape_wechat_talkers` — wraps ``POST /wechatweb/wxFriends/talkerList`` + and yields the per-staff-wechat conversation list (M2). Used by the sync + orchestrator to decide which conversations have recent traffic to pull. + Page-number pagination; the owning ``wechatId`` is sent in every request. All yield dicts (not ORM). Persistence is the caller's job (the :mod:`app.storage.repos` module bridges these dicts to the ORM tables). Each @@ -22,8 +31,10 @@ dict carries: * snake_case keys matching the target ORM columns (see :func:`_normalize` for friends, :func:`_normalize_message` for messages, - :func:`_normalize_department` for departments, and - :func:`_normalize_wxuser_account` for accounts), + :func:`_normalize_department` for departments, + :func:`_normalize_wxuser_account` for accounts, + :func:`_normalize_chat_group` for groups, and + :func:`_normalize_talker` for talkers), * ``channel == "wechat"`` (only on communication-channel entities; the organization entities ``Department`` / ``WechatAccount`` carry no channel), * ``raw_json`` — the original upstream record, verbatim, so nothing captured @@ -52,13 +63,17 @@ __all__ = [ "scrape_departments", "scrape_wechat_accounts", "scrape_wechat_friends", + "scrape_wechat_groups", "scrape_wechat_messages", + "scrape_wechat_talkers", ] FRIEND_LIST_PATH = "/wechatweb/wxFriends/friend_list/es" MSG_CHATS_PATH = "/wechatweb/msg/chats" DEPT_TREE_PATH = "/usercenter/dept/tree" WXUSER_LIST_PATH = "/wechatweb/wxUser/list" +CHATROOM_LIST_PATH = "/wechatweb/chatroom/getChatroomList" +TALKER_LIST_PATH = "/wechatweb/wxFriends/talkerList" # 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. @@ -700,3 +715,259 @@ async def scrape_wechat_accounts( build_body=_build_wxuser_body, ): yield _normalize_wxuser_account(record) + + +# =========================================================================== +# Chat-group scraper (M2) — POST /wechatweb/chatroom/getChatroomList +# =========================================================================== + + +def _parse_chatroom_total(resp: Any) -> int | None: + """Read the declared row count from the ``getChatroomList`` envelope. + + Shape: ``data.total`` (an int). 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("total") + return int(total) if isinstance(total, (int, float)) and not isinstance( + total, bool + ) else None + + +def _parse_chatroom_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the chatroom list from the ``getChatroomList`` envelope. + + Shape: ``data.data`` (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("data") + return rows if isinstance(rows, list) else [] + + +def _build_chatroom_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``getChatroomList`` 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 _parse_member_ids(member_ids: Any) -> list[str]: + """Split a semicolon-joined member id list, dropping empty segments. + + Upstream ``memberIds`` looks like ``"wxid_a;wxid_b;csf7169708"`` and may + carry a trailing ``";"`` (yielding an empty segment). Empties are dropped + so each entry is a real wxid; a missing/None/non-string value yields ``[]``. + """ + if not isinstance(member_ids, str): + return [] + return [seg for seg in member_ids.split(";") if seg] + + +def _normalize_chat_group(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream ``getChatroomList`` record to the ``ChatGroup`` shape. + + Adds ``channel="wechat"`` and preserves the original record under + ``raw_json``. The raw ``memberIds`` string is carried verbatim as + ``member_ids`` (matching the ``ChatGroup.member_ids`` Text column) AND + parsed into ``member_ids_list`` (a plain list of wxids with empties + dropped) so :func:`app.storage.repos.upsert_chat_group` can mirror one + :class:`app.db.models.GroupMember` row per entry without re-parsing. + + ``None``-safe: missing fields become ``None`` rather than raising, so a + partial payload does not break the whole scrape. + """ + member_ids = record.get("memberIds") + return { + "room_id": record.get("chatRoomId"), + "channel": "wechat", + "room_name": record.get("chatRoomName"), + "owner_account_id": record.get("wechatId"), + "owner_account_nickname": record.get("wechatNickName"), + "member_count": record.get("memberCount"), + "member_ids": member_ids, + "member_ids_list": _parse_member_ids(member_ids), + "head_image_url": record.get("headImageUrl"), + "owner_user_id": record.get("userId"), + "owner_user_phone": record.get("userPhone"), + "message_count": record.get("messageCount"), + "own_message_count": record.get("ownMessageCount"), + "loss_members": record.get("lossMembers"), + "add_members": record.get("addMembers"), + "last_chat_time": record.get("lastChatTime"), + "update_time": record.get("updateTime"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_wechat_groups( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WeChat chat-group records from ``getChatroomList``. + + Walks ``POST /wechatweb/chatroom/getChatroomList`` page by page until the + upstream reports no more rows (empty ``data.data`` array — the primary + stop signal), mapping each record to the ``ChatGroup`` column shape plus + a verbatim ``raw_json`` copy. + + The semicolon-joined ``memberIds`` field is carried both raw + (``member_ids``) and parsed (``member_ids_list``, empties dropped) so the + persistence layer can mirror :class:`app.db.models.GroupMember` rows + idempotently. + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client. + page_size: + Page size to request. Defaults to 50 (a sane middle ground between + request count and per-response payload size for this endpoint). + """ + async for record in paginate_by_page( + client, + CHATROOM_LIST_PATH, + page_size=page_size, + parse_total=_parse_chatroom_total, + parse_rows=_parse_chatroom_rows, + build_body=_build_chatroom_body, + ): + yield _normalize_chat_group(record) + + +# =========================================================================== +# Talker (conversation) list scraper (M2) — POST /wechatweb/wxFriends/talkerList +# =========================================================================== + + +def _parse_talker_total(resp: Any) -> int | None: + """Read the declared row count from the ``talkerList`` envelope. + + The talker endpoint signals "more pages" via ``data.hasMore`` (a boolean) + rather than a declared ``total``. There is no honest row count to read, + so this always returns ``None`` — the paginator then relies solely on the + empty-page stop signal (a final page whose ``data.page`` is empty). + """ + return None + + +def _parse_talker_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the talker list from the ``talkerList`` envelope. + + Shape: ``data.page`` (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("page") + return rows if isinstance(rows, list) else [] + + +def _build_talker_body( + page_num: int, page_size: int, *, wechat_id: str +) -> dict[str, Any]: + """Build the ``talkerList`` request body for a given page. + + The endpoint is per-staff-wechat: the owning ``wechatId`` MUST be sent in + every request so the upstream returns that account's conversation list. + """ + return { + "wechatId": wechat_id, + "pageNum": page_num, + "pageSize": page_size, + } + + +def _normalize_talker(record: dict[str, Any], wechat_id: str) -> dict[str, Any]: + """Map one upstream ``talkerList`` record to a talker dict. + + The talker list is primarily a sync-planning input: it tells the + orchestrator which conversations (direct chats AND group chats) have + recent traffic. ``member_count > 0`` marks a group conversation (the + ``friend_wechat_id`` is then a ``@chatroom`` id); ``member_count == 0`` + marks a 1:1 friend chat. The owning staff wechat is stamped on every + record as ``owner_account_id``. ``raw_json`` preserves the upstream + record verbatim. ``None``-safe on every field. + """ + return { + "friend_wechat_id": record.get("friendWechatId"), + "owner_account_id": wechat_id, + "nick_name": record.get("nickName"), + "remark_name": record.get("remarkName"), + "display_name": record.get("displayName"), + "message_count": record.get("messageCount"), + "last_chat_time": record.get("lastChatTime"), + "member_count": record.get("memberCount"), + "head_url": record.get("headUrl"), + "raw_json": record, + } + + +async def scrape_wechat_talkers( + client: YunkeClient, + wechat_id: str, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield the per-staff-wechat conversation list from ``talkerList``. + + Wraps ``POST /wechatweb/wxFriends/talkerList``. The endpoint is scoped to + one staff wechat account (``wechat_id`` → the ``wechatId`` request + field); each response carries a slice of that account's recent + conversations under ``data.page`` with a boolean ``data.hasMore``. + + The response envelope carries no declared ``total`` (only ``hasMore``), + so this scraper relies solely on the empty-page stop signal: it advances + ``pageNum`` until a page comes back with an empty ``data.page`` array. + A well-behaved upstream flips ``hasMore`` to false on the last real page + and then serves an empty page; the paginator stops on whichever signal + fires first. + + Talkers are NOT persisted to a dedicated table — the sync orchestrator + consumes this stream to decide which conversations have messages to pull + (``member_count > 0`` ⇒ group chat whose ``friend_wechat_id`` is a + ``@chatroom`` id). + + Parameters + ---------- + client: + Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are + handled by the client. + wechat_id: + Staff WeChat id (``wechatId`` request field) — the owning account + whose conversation list is being scraped. + page_size: + Page size to request. Defaults to 50. + """ + async for record in paginate_by_page( + client, + TALKER_LIST_PATH, + page_size=page_size, + parse_total=_parse_talker_total, + parse_rows=_parse_talker_rows, + build_body=_build_talker_body, + wechat_id=wechat_id, + ): + yield _normalize_talker(record, wechat_id) diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index 2f5437d..a7123a5 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -48,9 +48,18 @@ import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import DeclarativeBase -from app.db.models import Customer, Department, Message, Staff, WechatAccount +from app.db.models import ( + ChatGroup, + Customer, + Department, + GroupMember, + Message, + Staff, + WechatAccount, +) __all__ = [ + "upsert_chat_group", "upsert_customer", "upsert_customers", "upsert_department", @@ -383,3 +392,54 @@ async def upsert_staff_from_account(session: AsyncSession, rec: dict[str, Any]) if key in rec: staff_rec[key] = rec[key] await upsert_staff(session, staff_rec) + + +# --------------------------------------------------------------------------- +# ChatGroup + GroupMember upsert (M2) +# --------------------------------------------------------------------------- + + +async def upsert_chat_group(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert one :class:`ChatGroup` AND its :class:`GroupMember` rows. + + PK = ``room_id``. The group row is upserted first; then, for every id in + the record's ``member_ids_list`` (the scraper-parsed, empties-dropped + member wxid list), a :class:`GroupMember` row is upserted against the + composite PK ``(channel, room_id, member_id)``. Re-scraping a group + therefore refreshes the group row and re-asserts its members without + duplication (idempotent). + + The ``member_ids_list`` key is the parsed list shape produced by + :func:`app.scraper.wechat._normalize_chat_group`; it is NOT a + ``ChatGroup`` column, so it is dropped by :func:`_build_column_values` + when writing the group row (the raw semicolon-joined string is preserved + on ``ChatGroup.member_ids``). + + Per-member ``nickname`` / ``avatar`` / ``display_name`` are left ``None`` + here — Phase-0 does not yet resolve them (they need a separate per-member + API; see data dictionary §9.6). The repo writes a minimal member row keyed + by the composite PK so membership is captured now and enriched later. + """ + await _upsert_one(session, ChatGroup, rec) + + channel = rec.get("channel") or "wechat" + room_id = rec.get("room_id") + # No room_id -> nothing to key member rows on (defensive; the scraper + # always sets room_id). Skip the member loop rather than raising so a + # malformed record still lands its (PK-less) group upsert attempt above. + if not room_id: + return + + member_ids: list[Any] = list(rec.get("member_ids_list") or []) + for member_id in member_ids: + # Defensive: never insert an empty/falsy member id (would collide on + # an empty-PK row). The scraper already drops empties, but a hand- + # built record could slip one through. + if not member_id: + continue + member_rec: dict[str, Any] = { + "channel": channel, + "room_id": room_id, + "member_id": member_id, + } + await _upsert_one(session, GroupMember, member_rec) diff --git a/yunke-server/tests/cassettes/chatroom_fixture.py b/yunke-server/tests/cassettes/chatroom_fixture.py new file mode 100644 index 0000000..2e3c367 --- /dev/null +++ b/yunke-server/tests/cassettes/chatroom_fixture.py @@ -0,0 +1,137 @@ +"""Fixture payloads for ``POST /wechatweb/chatroom/getChatroomList`` (M2). + +These records mirror the real field names captured from the live CRM (see the +data dictionary §2: ``chatRoomId(@chatroom), chatRoomName, memberCount, +memberIds(分号分隔全部成员wxid), headImageUrl, wechatId/wechatNickName(归属), +userName/userPhone(群主员工), messageCount/ownMessageCount, lossMembers/ +addMembers, lastChatTime/updateTime``; total=1515). + +The envelope is the same flat shape as ``wxUser/list`` (no ``basicResult`` +wrapper): pagination fields sit directly under ``data``:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "total":.., "page":.., "pageCount":..}} + +The cassette ships three records across two pages so a test can drive the +page-number paginator across the empty-page stop signal: + +* page 1 -> 2 of the 3 fixture records, ``total=3`` +* page 2 -> the remaining 1 record, ``total=3`` +* (page 3 -> empty ``data`` array, served inline by the test) + +Records carry every field with a target ``ChatGroup`` column so the test can +assert a full mapping end-to-end. The ``memberIds`` field is the +semicolon-joined member wxid list; record 2 deliberately includes a trailing +empty segment (``"...;"``) so the parser's drop-empties path is exercised. +""" + +# The module name mirrors the upstream endpoint name (``chatroom``); the M2 +# spec mandates this exact filename. +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped chatroom records (3 total across 2 pages). +# --------------------------------------------------------------------------- + +CHATROOM_RECORDS: list[dict[str, Any]] = [ + { + "chatRoomId": "12345678901@chatroom", + "chatRoomName": "VIP客户交流群", + "memberCount": 3, + # 3 members, no empties -> member_ids_list length 3. + "memberIds": "wxid_member_a1;wxid_member_a2;csf7169708", + "headImageUrl": "https://wx.qlogo.cn/room_12345678901.png", + "wechatId": "wxid_staff_alice01", + "wechatNickName": "Alice WeChat", + "userId": "staff_user_001", + "userName": "Alice Wang", + "userPhone": "13800000001", + "messageCount": 4521, + "ownMessageCount": 1204, + "lossMembers": 2, + "addMembers": 5, + "lastChatTime": 1700000000123, + "updateTime": 1700000001000, + "version": "v1.20260731.room1", + }, + { + "chatRoomId": "98765432109@chatroom", + "chatRoomName": "华东销售大群", + "memberCount": 2, + # Trailing empty segment -> member_ids_list length 2 (empties dropped). + "memberIds": "wxid_member_b1;wxid_member_b2;", + "headImageUrl": "https://wx.qlogo.cn/room_98765432109.png", + "wechatId": "wxid_staff_bob02", + "wechatNickName": "Bob WeChat", + "userId": "staff_user_002", + "userName": "Bob Chen", + "userPhone": "13800000002", + "messageCount": 9821, + "ownMessageCount": 211, + "lossMembers": 0, + "addMembers": 1, + "lastChatTime": 1700000000456, + "updateTime": 1700000000800, + "version": "v1.20260731.room2", + }, + { + # A group whose owner has no resolved staff id (userId None) — the + # scraper must still yield it; the repo must still upsert its members. + "chatRoomId": "55555555555@chatroom", + "chatRoomName": "External Coordination", + "memberCount": 1, + "memberIds": "wxid_member_c1", + "headImageUrl": "https://wx.qlogo.cn/room_55555555555.png", + "wechatId": "wxid_staff_orphan04", + "wechatNickName": "Orphan WeChat", + "userId": None, + "userName": None, + "userPhone": None, + "messageCount": 12, + "ownMessageCount": 0, + "lossMembers": 0, + "addMembers": 0, + "lastChatTime": 1700000000789, + "updateTime": 1700000000900, + "version": "v1.20260731.room3", + }, +] + +# Default page size for the fixture (and the scraper). total=3 means the +# fixture fills exactly 2 pages of size 2 plus a terminating empty page. +TOTAL = len(CHATROOM_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_chatroom_list_response( + records: list[dict[str, Any]], + *, + total: int, + page_num: int = 1, + page_size: int = DEFAULT_PAGE_SIZE, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``getChatroomList`` response envelope. + + Shape (no ``basicResult`` wrapper; pagination fields directly under + ``data``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "total":.., "page":.., "pageCount":..}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-{page_num}", + "data": { + "data": list(records), + "total": total, + "page": page_num, + "pageCount": ( + (total + page_size - 1) // page_size if page_size else 1 + ), + }, + } diff --git a/yunke-server/tests/cassettes/talkerList_fixture.py b/yunke-server/tests/cassettes/talkerList_fixture.py new file mode 100644 index 0000000..62724a0 --- /dev/null +++ b/yunke-server/tests/cassettes/talkerList_fixture.py @@ -0,0 +1,112 @@ +"""Fixture payloads for ``POST /wechatweb/wxFriends/talkerList`` (M2). + +These records mirror the real field names captured from the live CRM (see the +data dictionary §2 会话列表 talker: ``friendWechatId, nickName, remarkName, +messageCount, lastChatTime, headUrl, memberCount, analyseResult``). The +endpoint is **per-staff-wechat**: the request body carries the ``wechatId`` of +the owning staff account, and the response lists that account's recent +conversations (direct chats AND group chats — ``memberCount > 0`` ⇒ group). + +The envelope carries the row list under ``data.page`` with a boolean +``data.hasMore`` flag (no ``total`` field is documented for this endpoint):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"page":[ {}, ... ], "hasMore":..}} + +Because the page-number paginator reuses the empty-page stop signal, the +cassette ships the records across two pages plus a terminating empty page: + +* page 1 -> 2 of the 3 fixture talks, ``hasMore=true`` +* page 2 -> the remaining 1 talk, ``hasMore=false`` +* (page 3 -> empty ``page`` array, served inline by the test) + +Records exercise both the direct-message path (``memberCount == 0``) and the +group-chat path (``memberCount > 0``) so the orchestrator can distinguish them. +""" + +# The module name uses camelCase deliberately to mirror the upstream endpoint +# name (``talkerList``); the M2 spec mandates this exact filename. +# ruff: noqa: N999 + +from __future__ import annotations + +from typing import Any + +# The owning staff wechat for this talker stream (the ``wechatId`` request +# field). All records below belong to this account's conversation list. +OWNER_WECHAT_ID = "wxid_staff_alice01" + +# --------------------------------------------------------------------------- +# Capture-shaped talker (conversation) records (3 total across 2 pages). +# --------------------------------------------------------------------------- + +TALKER_RECORDS: list[dict[str, Any]] = [ + { + # Direct message (1:1 friend). memberCount == 0. + "friendWechatId": "wxid_friend_0001", + "nickName": "Alice Wang", + "remarkName": "Alice-VIP", + "messageCount": 46, + "lastChatTime": 1700000000123, + "headUrl": "https://wx.qlogo.cn/avatar_alice.png", + "memberCount": 0, + "displayName": "", + "analyseResult": "", + }, + { + # Group chat. memberCount > 0 ⇒ the friendWechatId is a @chatroom id. + "friendWechatId": "12345678901@chatroom", + "nickName": "VIP客户交流群", + "remarkName": "", + "messageCount": 4521, + "lastChatTime": 1700000000456, + "headUrl": "https://wx.qlogo.cn/room_12345678901.png", + "memberCount": 3, + "displayName": "Alice (group)", + "analyseResult": "", + }, + { + # Another direct message — a less active contact. + "friendWechatId": "wxid_friend_0009", + "nickName": "Quiet Lead", + "remarkName": "", + "messageCount": 2, + "lastChatTime": 1700000000789, + "headUrl": "https://wx.qlogo.cn/avatar_quiet.png", + "memberCount": 0, + "displayName": "", + "analyseResult": "", + }, +] + +# Default page size for the fixture (and the scraper). 3 records fill exactly +# 2 pages of size 2 plus a terminating empty page. +TOTAL = len(TALKER_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_talker_list_response( + records: list[dict[str, Any]], + *, + page_num: int = 1, + page_size: int = DEFAULT_PAGE_SIZE, + has_more: bool = False, + trace_id: str | None = None, +) -> dict[str, Any]: + """Build a capture-shaped ``talkerList`` response envelope. + + Shape (row list under ``data.page``, boolean ``data.hasMore``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"page":[...], "hasMore":..}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-{page_num}", + "data": { + "page": list(records), + "hasMore": has_more, + }, + } diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index fbd7933..f711eaa 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -22,10 +22,19 @@ import sqlalchemy as sa from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.db.base import Base -from app.db.models import Customer, Department, Message, Staff, WechatAccount +from app.db.models import ( + ChatGroup, + Customer, + Department, + GroupMember, + Message, + Staff, + WechatAccount, +) from app.storage.repos import ( _coerce_bool, _coerce_int, + upsert_chat_group, upsert_customer, upsert_customers, upsert_department, @@ -701,3 +710,178 @@ async def test_upsert_wechat_account_without_user_id_skips_staff(): assert staff_rows == [] finally: await engine.dispose() + + +# --------------------------------------------------------------------------- +# ChatGroup + GroupMember upsert (M2) +# --------------------------------------------------------------------------- + + +async def test_upsert_chat_group_inserts_group_and_members(): + """Upsert writes the ChatGroup row AND one GroupMember per member_ids_list.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "room_id": "12345678901@chatroom", + "channel": "wechat", + "room_name": "VIP客户交流群", + "owner_account_id": "wxid_staff_alice01", + "member_count": 3, + "member_ids": "wxid_m1;wxid_m2;wxid_m3", + "member_ids_list": ["wxid_m1", "wxid_m2", "wxid_m3"], + "message_count": 4521, + } + await upsert_chat_group(session, rec) + await session.commit() + + group = ( + await session.execute( + sa.select(ChatGroup).where( + ChatGroup.room_id == "12345678901@chatroom" + ) + ) + ).scalar_one() + assert group.room_name == "VIP客户交流群" + assert group.owner_account_id == "wxid_staff_alice01" + assert group.member_count == 3 + assert group.message_count == 4521 + # The raw semicolon-joined string is stored on the group row. + assert group.member_ids == "wxid_m1;wxid_m2;wxid_m3" + + # Members: one row per member_ids_list entry, composite PK. + members = ( + await session.execute( + sa.select(GroupMember) + .where(GroupMember.room_id == "12345678901@chatroom") + .order_by(GroupMember.member_id) + ) + ).scalars().all() + assert len(members) == 3 + assert {m.member_id for m in members} == {"wxid_m1", "wxid_m2", "wxid_m3"} + assert {m.channel for m in members} == {"wechat"} + # Phase-0: per-member nickname/avatar are not yet populated. + assert all(m.nickname is None for m in members) + assert all(m.avatar is None for m in members) + finally: + await engine.dispose() + + +async def test_upsert_chat_group_reupsert_updates_group_keeps_members_unique(): + """Re-upserting a group refreshes its row; members stay unique (no dups).""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "room_id": "98765432109@chatroom", + "channel": "wechat", + "room_name": "Old Name", + "member_count": 2, + "member_ids": "wxid_m1;wxid_m2", + "member_ids_list": ["wxid_m1", "wxid_m2"], + "message_count": 10, + } + await upsert_chat_group(session, rec1) + await session.commit() + + rec2 = dict( + rec1, + room_name="New Name", + message_count=99, + ) + await upsert_chat_group(session, rec2) + await session.commit() + + # One group row, refreshed (name + count updated, not duplicated). + groups = ( + await session.execute( + sa.select(ChatGroup) + .where(ChatGroup.room_id == "98765432109@chatroom") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(groups) == 1 + assert groups[0].room_name == "New Name" + assert groups[0].message_count == 99 + + # Members still exactly 2 (idempotent re-upsert, no duplication). + members = ( + await session.execute( + sa.select(GroupMember).where( + GroupMember.room_id == "98765432109@chatroom" + ) + ) + ).scalars().all() + assert len(members) == 2 + assert {m.member_id for m in members} == {"wxid_m1", "wxid_m2"} + finally: + await engine.dispose() + + +async def test_upsert_chat_group_drops_empty_member_id(): + """An empty member id in member_ids_list is dropped (no empty-PK row). + + The scraper already drops empties when building ``member_ids_list``; this + test exercises the repo's own defensive guard so a hand-built record that + slips an empty id through cannot create a degenerate empty-PK member row. + """ + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "room_id": "55555555555@chatroom", + "channel": "wechat", + # Empty string sneaks through; repo must skip it. + "member_ids_list": ["", "wxid_only", ""], + } + await upsert_chat_group(session, rec) + await session.commit() + + members = ( + await session.execute( + sa.select(GroupMember).where( + GroupMember.room_id == "55555555555@chatroom" + ) + ) + ).scalars().all() + assert len(members) == 1 + assert members[0].member_id == "wxid_only" + finally: + await engine.dispose() + + +async def test_upsert_chat_group_with_no_members_writes_group_only(): + """A group with an empty member list still writes the group row.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "room_id": "empty@chatroom", + "channel": "wechat", + "room_name": "No Members Yet", + "member_ids_list": [], + } + await upsert_chat_group(session, rec) + await session.commit() + + # Group row written. + group = ( + await session.execute( + sa.select(ChatGroup).where(ChatGroup.room_id == "empty@chatroom") + ) + ).scalar_one() + assert group.room_name == "No Members Yet" + + # No member rows. + members = ( + await session.execute( + sa.select(GroupMember).where(GroupMember.room_id == "empty@chatroom") + ) + ).scalars().all() + assert members == [] + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_scrape_wechat_groups.py b/yunke-server/tests/test_scrape_wechat_groups.py new file mode 100644 index 0000000..2d928f2 --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_groups.py @@ -0,0 +1,231 @@ +"""Tests for the WeChat chat-group scraper (M2). + +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 for page-number endpoints): + +* page 1 -> 2 records, ``total=3`` +* page 2 -> 1 record, ``total=3`` +* page 3 -> empty ``data`` array (no more rows) + +The scraper must yield exactly the 3 records (normalized to ``ChatGroup`` +snake_case columns + ``raw_json``), parse ``memberIds`` into +``member_ids_list`` (dropping empty segments), make exactly 3 calls, and stop +without a 4th. +""" + +from __future__ import annotations + +import json as _json + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_groups +from app.session.yunke_client import YunkeClient +from tests.cassettes.chatroom_fixture import ( + CHATROOM_RECORDS, + DEFAULT_PAGE_SIZE, + TOTAL, + make_chatroom_list_response, +) + +BASE = "https://crm2.yunkecn.com" +CHATROOM_LIST_URL = f"{BASE}/wechatweb/chatroom/getChatroomList" + +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 + member_ids_list parsing +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_groups_yields_records_and_paginates(): + """Two full pages + a terminating empty page -> 3 records, 3 calls.""" + page1 = make_chatroom_list_response( + CHATROOM_RECORDS[:2], total=TOTAL, page_num=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_chatroom_list_response( + CHATROOM_RECORDS[2:], total=TOTAL, page_num=2, page_size=DEFAULT_PAGE_SIZE + ) + page3 = make_chatroom_list_response( + [], total=TOTAL, page_num=3, page_size=DEFAULT_PAGE_SIZE + ) + + route = respx.post(CHATROOM_LIST_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_wechat_groups(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["room_id"] for r in results} == { + rec["chatRoomId"] for rec in CHATROOM_RECORDS + } + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, CHATROOM_RECORDS, strict=True): + # Identity + channel. + assert result["room_id"] == record["chatRoomId"] + assert result["channel"] == "wechat" + + # Naming + ownership. + assert result["room_name"] == record["chatRoomName"] + assert result["owner_account_id"] == record["wechatId"] + assert result["owner_account_nickname"] == record["wechatNickName"] + assert result["owner_user_id"] == record["userId"] + assert result["owner_user_phone"] == record["userPhone"] + + # Counts. + assert result["member_count"] == record["memberCount"] + assert result["message_count"] == record["messageCount"] + assert result["own_message_count"] == record["ownMessageCount"] + assert result["loss_members"] == record["lossMembers"] + assert result["add_members"] == record["addMembers"] + + # Avatar + timestamps. + assert result["head_image_url"] == record["headImageUrl"] + assert result["last_chat_time"] == record["lastChatTime"] + assert result["update_time"] == record["updateTime"] + + # member_ids carried raw; member_ids_list parsed (empties dropped). + assert result["member_ids"] == record["memberIds"] + assert result["member_ids_list"] == [ + seg for seg in record["memberIds"].split(";") if seg + ] + + # Version carried verbatim; raw_json preserves the original record. + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +# --------------------------------------------------------------------------- +# member_ids_list parsing edge cases +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_groups_drops_empty_member_id_segments(): + """Trailing/empty ``memberIds`` segments are dropped from member_ids_list.""" + # Record 2 has a trailing ";" -> one empty segment that must be dropped. + page1 = make_chatroom_list_response( + [CHATROOM_RECORDS[1]], total=1, page_num=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_chatroom_list_response([], total=1, page_num=2) + respx.post(CHATROOM_LIST_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_groups(client, page_size=DEFAULT_PAGE_SIZE)] + finally: + await client.aclose() + + assert len(results) == 1 + # "wxid_member_b1;wxid_member_b2;" -> 2 ids (trailing empty dropped). + assert results[0]["member_ids_list"] == ["wxid_member_b1", "wxid_member_b2"] + + +@respx.mock +async def test_scrape_wechat_groups_handles_missing_member_ids(): + """A group with no ``memberIds`` field yields an empty member_ids_list.""" + record = {"chatRoomId": "x@chatroom", "chatRoomName": "Empty"} + page1 = make_chatroom_list_response([record], total=1, page_num=1) + page2 = make_chatroom_list_response([], total=1, page_num=2) + respx.post(CHATROOM_LIST_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_groups(client, page_size=DEFAULT_PAGE_SIZE)] + finally: + await client.aclose() + + assert len(results) == 1 + assert results[0]["member_ids"] is None + assert results[0]["member_ids_list"] == [] + + +# --------------------------------------------------------------------------- +# Pagination body: pageNum/pageSize advanced correctly +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_groups_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_chatroom_list_response( + [CHATROOM_RECORDS[0]], total=TOTAL, page_num=1, page_size=1 + ) + page2 = make_chatroom_list_response([], total=TOTAL, page_num=2, page_size=1) + route = respx.post(CHATROOM_LIST_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_groups(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 + + 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_wechat_groups_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_chatroom_list_response([], total=0, page_num=1) + route = respx.post(CHATROOM_LIST_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_wechat_groups(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1 diff --git a/yunke-server/tests/test_scrape_wechat_talkers.py b/yunke-server/tests/test_scrape_wechat_talkers.py new file mode 100644 index 0000000..a262b71 --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_talkers.py @@ -0,0 +1,164 @@ +"""Tests for the WeChat talker (conversation) list scraper (M2). + +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 conversations across 2 pages of size 2, plus a +terminating empty page (the page-number paginator's primary stop signal): + +* page 1 -> 2 records, ``hasMore=true`` +* page 2 -> 1 record, ``hasMore=false`` +* page 3 -> empty ``page`` array (no more rows) + +The scraper must yield exactly the 3 talkers (normalized), send the owning +``wechatId`` in every request body, advance ``pageNum``, and stop after the +empty page. Both the direct-message (``memberCount == 0``) and group-chat +(``memberCount > 0``) paths are exercised by the fixture. +""" + +from __future__ import annotations + +import json as _json + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_talkers +from app.session.yunke_client import YunkeClient +from tests.cassettes.talkerList_fixture import ( + DEFAULT_PAGE_SIZE, + OWNER_WECHAT_ID, + TALKER_RECORDS, + make_talker_list_response, +) + +BASE = "https://crm2.yunkecn.com" +TALKER_LIST_URL = f"{BASE}/wechatweb/wxFriends/talkerList" + +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 + wechatId in request body +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_talkers_yields_records_and_paginates(): + """Two full pages + a terminating empty page -> 3 talkers, 3 calls.""" + page1 = make_talker_list_response( + TALKER_RECORDS[:2], page_num=1, page_size=DEFAULT_PAGE_SIZE, has_more=True + ) + page2 = make_talker_list_response( + TALKER_RECORDS[2:], page_num=2, page_size=DEFAULT_PAGE_SIZE, has_more=False + ) + page3 = make_talker_list_response([], page_num=3, page_size=DEFAULT_PAGE_SIZE) + + route = respx.post(TALKER_LIST_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_wechat_talkers( + client, OWNER_WECHAT_ID, 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 talkers (not duplicated, not dropped). + assert len(results) == 3 + assert {r["friend_wechat_id"] for r in results} == { + rec["friendWechatId"] for rec in TALKER_RECORDS + } + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, TALKER_RECORDS, strict=True): + assert result["friend_wechat_id"] == record["friendWechatId"] + assert result["nick_name"] == record["nickName"] + assert result["remark_name"] == record["remarkName"] + assert result["message_count"] == record["messageCount"] + assert result["last_chat_time"] == record["lastChatTime"] + assert result["member_count"] == record["memberCount"] + assert result["head_url"] == record["headUrl"] + # The owning staff account is carried on every record. + assert result["owner_account_id"] == OWNER_WECHAT_ID + # raw_json preserves the original upstream record. + assert result["raw_json"] == record + + +@respx.mock +async def test_scrape_wechat_talkers_sends_wechat_id_and_pagination_body(): + """Every request carries the staff wechatId + advancing pageNum/pageSize.""" + page1 = make_talker_list_response( + TALKER_RECORDS[:1], page_num=1, page_size=1, has_more=True + ) + page2 = make_talker_list_response( + [], page_num=2, page_size=1, has_more=False + ) + route = respx.post(TALKER_LIST_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_talkers(client, OWNER_WECHAT_ID, 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 + + # Both requests must carry the owning wechatId + page slice. + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["wechatId"] == OWNER_WECHAT_ID + assert body1["pageNum"] == 1 + assert body1["pageSize"] == 1 + + body2 = _json.loads(route.calls[1].request.content.decode()) + assert body2["wechatId"] == OWNER_WECHAT_ID + assert body2["pageNum"] == 2 + assert body2["pageSize"] == 1 + + +# --------------------------------------------------------------------------- +# Empty first page -> no yields, single call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_talkers_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_talker_list_response([], page_num=1) + route = respx.post(TALKER_LIST_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_wechat_talkers(client, OWNER_WECHAT_ID)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1