diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index d6f8424..46a05f6 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -24,6 +24,11 @@ Endpoints covered today: 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. +* :func:`scrape_wechat_voice_calls` — wraps ``POST /wechatweb/callRecord/queryWeChatVoiceList`` + and yields normalized WeChat voice-call records (M2). Each record is mapped + to the ``PhoneCall`` column shape with ``channel="voice_call"``; the + recording URL (``ossFileName``) lands on ``record_file``. Page-number + pagination. 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 @@ -33,8 +38,9 @@ dict carries: for friends, :func:`_normalize_message` for messages, :func:`_normalize_department` for departments, :func:`_normalize_wxuser_account` for accounts, - :func:`_normalize_chat_group` for groups, and - :func:`_normalize_talker` for talkers), + :func:`_normalize_chat_group` for groups, + :func:`_normalize_talker` for talkers, and + :func:`_normalize_voice_call` for voice calls), * ``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,6 +58,7 @@ disambiguations where the source has two fields for one column: from __future__ import annotations +import datetime as dt from collections.abc import AsyncIterator from typing import Any @@ -66,6 +73,7 @@ __all__ = [ "scrape_wechat_groups", "scrape_wechat_messages", "scrape_wechat_talkers", + "scrape_wechat_voice_calls", ] FRIEND_LIST_PATH = "/wechatweb/wxFriends/friend_list/es" @@ -74,6 +82,7 @@ DEPT_TREE_PATH = "/usercenter/dept/tree" WXUSER_LIST_PATH = "/wechatweb/wxUser/list" CHATROOM_LIST_PATH = "/wechatweb/chatroom/getChatroomList" TALKER_LIST_PATH = "/wechatweb/wxFriends/talkerList" +VOICE_LIST_PATH = "/wechatweb/callRecord/queryWeChatVoiceList" # 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. @@ -971,3 +980,190 @@ async def scrape_wechat_talkers( wechat_id=wechat_id, ): yield _normalize_talker(record, wechat_id) + + +# =========================================================================== +# Voice-call scraper (M2, Task 2.4) — POST /wechatweb/callRecord/queryWeChatVoiceList +# =========================================================================== + + +def _parse_voice_total(resp: Any) -> int | None: + """Read the declared row count from the ``queryWeChatVoiceList`` envelope. + + Shape: ``data.total`` (an int). Returns ``None`` if the path is absent so + the paginator falls back to the empty-page stop signal. (Live ``total`` + is 333 per phase0-findings.md §4.) + """ + 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_voice_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the voice-call list from the ``queryWeChatVoiceList`` 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_voice_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the ``queryWeChatVoiceList`` 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_voice_start_time(value: Any) -> int | None: + """Coerce a voice-call ``startTime`` to a ms-epoch int (the PhoneCall column type). + + The voice-call endpoint reports ``startTime`` as a ``YYYY-MM-DD HH:MM:SS`` + datetime string (confirmed by the Task 2.4 spec). To be robust against the + upstream also emitting a numeric ms-epoch (as other call endpoints do), the + following shapes are accepted: + + * ``int`` / ``float`` — treated as ms-epoch already, returned as ``int``. + * numeric ``str`` (e.g. ``"1700000000123"``) -> ``int``. + * ``"YYYY-MM-DD HH:MM:SS"`` datetime string -> parsed and converted to + ms-epoch. The CRM is a Chinese deployment and the wall-clock times are + presumably CST (UTC+8); this scraper parses them as **naive UTC** for + determinism — a documented simplification. If production later needs + CST-accurate buckets, swap the ``tzinfo`` attached here. + * ``None`` / empty / unparseable -> ``None`` (no spurious value stored). + """ + if value is None: + return None + if isinstance(value, bool): + return int(value) + if isinstance(value, (int, float)): + return int(value) + if isinstance(value, str): + s = value.strip() + if not s: + return None + # Numeric string -> ms-epoch int (no datetime round-trip). + try: + return int(s) + except ValueError: + pass + # Documented "YYYY-MM-DD HH:MM:SS" datetime string -> ms-epoch. + # ``.replace(tzinfo=...)`` is chained on the strptime so ruff's + # DTZ007 sees the naive datetime being made aware (and so the + # resulting epoch is well-defined). + try: + parsed = dt.datetime.strptime(s, "%Y-%m-%d %H:%M:%S").replace( + tzinfo=dt.UTC + ) + except ValueError: + return None + return int(parsed.timestamp() * 1000) + return None + + +def _normalize_voice_call(record: dict[str, Any]) -> dict[str, Any]: + """Map one upstream ``queryWeChatVoiceList`` record to the ``PhoneCall`` shape. + + Adds ``channel="voice_call"`` and preserves the original record under + ``raw_json``. Field mapping per phase0-findings.md §4 + the PhoneCall + column set: + + * ``id`` -> ``call_id`` (PK) + * ``duration`` -> ``call_duration`` (string seconds coerced to int) + * ``startTime`` -> ``start_call_time`` (datetime string -> ms epoch; + see :func:`_parse_voice_start_time`) + * ``callStatus`` -> ``call_status`` (1=connected w/ recording, + 2=no answer) + * ``isSend`` -> ``incoming_call`` with INVERTED polarity. + Voice-call ``isSend`` is ``0=incoming / 1=outgoing``; ``PhoneCall. + incoming_call`` (matching the ``phone_call`` channel) is + ``1=incoming / 0=outgoing``. To keep the column's meaning identical + across both call channels we store ``incoming_call = 1 - isSend`` + (``None`` preserved when ``isSend`` is missing/non-numeric). + * ``userId`` -> ``user_id`` (owning staff) + * ``userName`` -> ``user_id_name`` + * ``talker`` -> ``call_number`` (the customer wxid; for a room + call this is the ``@chatroom`` id — there is no separate member-sender + field on the voice-call record) + * ``talkerNickName`` -> ``plan_customer_name`` + * ``ossFileName`` -> ``record_file`` (the recording URL; may be null) + + Non-applicable PhoneCall columns (``ring_time`` / ``sim_phone`` / + ``sim_number`` / ``imei`` / ``is_upload`` / ``download_count`` / ``ai_asr`` + / ``qc_status`` / ``department_id``) are left ``None`` — the voice-call + payload carries none of them. The owning ``weChatId`` has no matching + PhoneCall column (there is no ``owner_account_id`` on this model), so it + is preserved only in ``raw_json``. ``None``-safe on every mapped field. + """ + is_send = _coerce_int_or_none(record.get("isSend")) + # Flip isSend (0=incoming/1=outgoing) to incoming_call (1=incoming/0=outgoing). + incoming_call = None if is_send is None else 1 - is_send + return { + "call_id": record.get("id"), + "channel": "voice_call", + "start_call_time": _parse_voice_start_time(record.get("startTime")), + "call_duration": _coerce_int_or_none(record.get("duration")), + "call_status": _coerce_int_or_none(record.get("callStatus")), + "incoming_call": incoming_call, + "user_id": record.get("userId"), + "user_id_name": record.get("userName"), + "call_number": record.get("talker"), + "plan_customer_name": record.get("talkerNickName"), + "record_file": record.get("ossFileName"), + "version": record.get("version"), + "raw_json": record, + } + + +async def scrape_wechat_voice_calls( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WeChat voice-call records from ``queryWeChatVoiceList``. + + Walks ``POST /wechatweb/callRecord/queryWeChatVoiceList`` page by page + until the upstream reports no more rows (empty ``data.data`` array — the + primary stop signal), mapping each record to the ``PhoneCall`` column + shape with ``channel="voice_call"`` plus a verbatim ``raw_json`` copy. + + The envelope is the same flat shape as ``wxUser/list`` (no + ``basicResult`` wrapper): the row list and pagination fields sit directly + under ``data`` (``data.data`` / ``data.total`` / ``data.pageNum`` / + ``data.pageSize``). + + 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, + VOICE_LIST_PATH, + page_size=page_size, + parse_total=_parse_voice_total, + parse_rows=_parse_voice_rows, + build_body=_build_voice_body, + ): + yield _normalize_voice_call(record) diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index a7123a5..ef053ba 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -19,13 +19,18 @@ Upstream CRM values are not always typed the way the columns want them. The following coercions are applied when building column values from the input dict (see :func:`_coerce_bool`, :func:`_coerce_int`): -* Booleans (``is_block`` / ``is_send`` / ``is_del`` / ``is_on_job``): - ``1`` / ``"1"`` / ``True`` -> ``True``; ``0`` / ``"0"`` / ``False`` -> - ``False``; ``None`` preserved. +* Booleans (``is_block`` / ``is_send`` / ``is_del`` / ``is_on_job`` / + ``is_upload``): ``1`` / ``"1"`` / ``True`` -> ``True``; ``0`` / ``"0"`` / + ``False`` -> ``False``; ``None`` preserved. * ``gender``: ``"0"`` / ``"1"`` / ``"2"`` (str) -> ``int`` when not ``None``. * ``type_raw`` (Message): the raw upstream type code, coerced to ``int | None``. ``Message.type`` is a normalized label string and is stored as-is (no int coercion). +* PhoneCall int fields (``call_duration`` / ``start_call_time`` / + ``call_status`` / ``incoming_call`` / ``qc_status``): coerced to + ``int | None``. The voice-call scraper already parses these to ints + (incl. ``startTime`` datetime -> ms epoch); the repo coercion is a + defensive net for hand-built records that arrive as numeric strings. * Timestamps (``last_chat_time`` / ``timestamp`` / ``last_update`` / ...): passed through as-is (ms-epoch ``int`` or ``None``). * ``raw_json``: the original record dict, preserved verbatim. When the input @@ -54,6 +59,7 @@ from app.db.models import ( Department, GroupMember, Message, + PhoneCall, Staff, WechatAccount, ) @@ -66,6 +72,7 @@ __all__ = [ "upsert_messages", "upsert_staff", "upsert_staff_from_account", + "upsert_voice_call", "upsert_wechat_account", ] @@ -90,6 +97,8 @@ _INT_FIELDS: frozenset[str] = frozenset( "direction", "incoming_call", "call_status", + "call_duration", + "start_call_time", "qc_status", } ) @@ -443,3 +452,25 @@ async def upsert_chat_group(session: AsyncSession, rec: dict[str, Any]) -> None: "member_id": member_id, } await _upsert_one(session, GroupMember, member_rec) + + +# --------------------------------------------------------------------------- +# PhoneCall (voice_call) upsert (M2, Task 2.4) +# --------------------------------------------------------------------------- + + +async def upsert_voice_call(session: AsyncSession, rec: dict[str, Any]) -> None: + """Idempotently upsert one :class:`PhoneCall` (PK = ``call_id``). + + Same ``call_id`` twice -> one row, with non-PK fields refreshed from the + later record. Applies the documented int coercions (``call_duration`` / + ``start_call_time`` / ``call_status`` / ``incoming_call`` -> int when they + arrive as numeric strings; the voice-call scraper already emits ints, but + a hand-built record may not). ``is_upload`` is a bool field and goes + through :func:`_coerce_bool` like the other bool columns. + + The voice-call channel (``channel="voice_call"``) and the phone-call + channel (``channel="phone_call"``) share the ``PhoneCall`` table; the + per-row ``channel`` value disambiguates them. + """ + await _upsert_one(session, PhoneCall, rec) diff --git a/yunke-server/tests/cassettes/voiceCall_fixture.py b/yunke-server/tests/cassettes/voiceCall_fixture.py new file mode 100644 index 0000000..7e8a9d4 --- /dev/null +++ b/yunke-server/tests/cassettes/voiceCall_fixture.py @@ -0,0 +1,202 @@ +"""Fixture payloads for ``POST /wechatweb/callRecord/queryWeChatVoiceList`` (M2). + +These records mirror the real field names confirmed by the Phase-0 capture +(see phase0-findings.md §4 语音通话): ``id, userId/userName/userPhone/ +userWeChatNickName/userWeChatHeadImg`` (staff), ``talker/talkerNickName/ +talkerAlias/talkerHeadImg`` (customer), ``callType, isSend(0呼入/1呼出), +callStatus(1接通有录音/2无应答), ossFileName(录音URL, may be null), +ossFileExist, duration(秒,字符串)/durationFile(秒,int), startTime/endTime, +weChatId, remark, isRoom, callFileId``. Live ``total`` is 333. + +The envelope is the same flat shape as ``wxUser/list`` (no ``basicResult`` +wrapper): the row list + pagination fields sit directly under ``data``:: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "total":.., "pageNum":.., "pageSize":..}} + +The cassette ships four records across 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, ``total=4`` +* page 2 -> the other 2 records, ``total=4`` +* (page 3 -> empty ``data`` array, served inline by the test) + +Records deliberately exercise the documented mapping edges: + +* ``vc_1001`` — incoming connected call WITH a recording (``isSend=0`` -> + ``incoming_call=1``; ``callStatus=1``; non-null ``ossFileName``; + ``duration`` as a string). +* ``vc_1002`` — outgoing no-answer call with NO recording (``isSend=1`` -> + ``incoming_call=0``; ``callStatus=2``; ``ossFileName=null``). +* ``vc_1003`` — outgoing connected call with a recording. +* ``vc_1004`` — incoming connected call inside a chatroom (``isRoom=1`` and + ``talker`` is a ``@chatroom`` id), exercising the room-call path. +""" + +# The module name mirrors the upstream endpoint name (``callRecord/ +# queryWeChatVoiceList`` → ``voiceCall``); the M2 spec uses this naming family. +# ruff: noqa: N999 + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped voice-call records (4 total across 2 pages). +# --------------------------------------------------------------------------- + +VOICE_RECORDS: list[dict[str, Any]] = [ + { + # Incoming connected call with a recording. + "id": "vc_1001", + "userId": "staff_user_001", + "userName": "Alice Wang", + "userPhone": "13800000001", + "userWeChatNickName": "Alice WeChat", + "userWeChatHeadImg": "https://wx.qlogo.cn/alice.png", + "talker": "wxid_friend_0001", + "talkerNickName": "VIP Customer A", + "talkerAlias": "vip_a", + "talkerHeadImg": "https://wx.qlogo.cn/vip_a.png", + "callType": 1, + "isSend": 0, # 0=incoming -> incoming_call=1 + "callStatus": 1, # 1=connected w/ recording + "ossFileName": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/wechat-call/" + "8m0atd/imei001/ossCallback/1700000000000_1700000151000.mp3" + ), + "ossFileExist": 1, + "duration": "151", # string seconds -> call_duration int 151 + "durationFile": 151, + "startTime": "2026-01-15 10:30:00", # parsed to ms-epoch UTC + "endTime": "2026-01-15 10:32:31", + "weChatId": "wxid_staff_alice01", + "remark": "通话时长02:31", + "isRoom": 0, + "callFileId": "cf_1001", + "version": "v1.20260731.vc1001", + }, + { + # Outgoing no-answer call, no recording. + "id": "vc_1002", + "userId": "staff_user_001", + "userName": "Alice Wang", + "userPhone": "13800000001", + "userWeChatNickName": "Alice WeChat", + "userWeChatHeadImg": "https://wx.qlogo.cn/alice.png", + "talker": "wxid_friend_0002", + "talkerNickName": "Lead B", + "talkerAlias": "", + "talkerHeadImg": "https://wx.qlogo.cn/lead_b.png", + "callType": 1, + "isSend": 1, # 1=outgoing -> incoming_call=0 + "callStatus": 2, # 2=no answer + "ossFileName": None, # no recording -> record_file None + "ossFileExist": 0, + "duration": "0", + "durationFile": 0, + "startTime": "2026-01-15 11:00:00", + "endTime": "2026-01-15 11:00:00", + "weChatId": "wxid_staff_alice01", + "remark": "对方无应答", + "isRoom": 0, + "callFileId": "cf_1002", + "version": "v1.20260731.vc1002", + }, + { + # Outgoing connected call with a recording. + "id": "vc_1003", + "userId": "staff_user_002", + "userName": "Bob Chen", + "userPhone": "13800000002", + "userWeChatNickName": "Bob WeChat", + "userWeChatHeadImg": "https://wx.qlogo.cn/bob.png", + "talker": "wxid_friend_0003", + "talkerNickName": "Customer C", + "talkerAlias": "c_alias", + "talkerHeadImg": "https://wx.qlogo.cn/customer_c.png", + "callType": 1, + "isSend": 1, # outgoing -> incoming_call=0 + "callStatus": 1, # connected w/ recording + "ossFileName": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/wechat-call/" + "8m0atd/imei002/ossCallback/1700000003000_1700000045000.mp3" + ), + "ossFileExist": 1, + "duration": "45", + "durationFile": 45, + "startTime": "2026-01-15 12:00:00", + "endTime": "2026-01-15 12:00:45", + "weChatId": "wxid_staff_bob02", + "remark": "通话时长00:45", + "isRoom": 0, + "callFileId": "cf_1003", + "version": "v1.20260731.vc1003", + }, + { + # Incoming connected call inside a chatroom (a group voice call). + "id": "vc_1004", + "userId": "staff_user_001", + "userName": "Alice Wang", + "userPhone": "13800000001", + "userWeChatNickName": "Alice WeChat", + "userWeChatHeadImg": "https://wx.qlogo.cn/alice.png", + # For a room call the "talker" is the chatroom id. + "talker": "12345678901@chatroom", + "talkerNickName": "VIP客户交流群", + "talkerAlias": "", + "talkerHeadImg": "https://wx.qlogo.cn/room_12345678901.png", + "callType": 2, + "isSend": 0, # incoming -> incoming_call=1 + "callStatus": 1, # connected w/ recording + "ossFileName": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/wechat-call/" + "8m0atd/imei001/ossCallback/1700000005000_1700000212000.mp3" + ), + "ossFileExist": 1, + "duration": "207", + "durationFile": 207, + "startTime": "2026-01-16 09:00:00", + "endTime": "2026-01-16 09:03:27", + "weChatId": "wxid_staff_alice01", + "remark": "群通话03:27", + "isRoom": 1, # this is a group voice call + "callFileId": "cf_1004", + "version": "v1.20260731.vc1004", + }, +] + +# Default page size for the fixture (and the scraper). total=4 means the +# fixture fills exactly 2 pages of size 2 plus a terminating empty page. +TOTAL = len(VOICE_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_voice_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 ``queryWeChatVoiceList`` response envelope. + + Shape (no ``basicResult`` wrapper; pagination fields directly under + ``data``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "total":.., "pageNum":.., "pageSize":..}} + """ + return { + "message": "调用成功", + "code": "10000", + "success": True, + "traceId": trace_id or f"trace-{page_num}", + "data": { + "data": list(records), + "total": total, + "pageNum": page_num, + "pageSize": page_size, + }, + } diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index f711eaa..a4ff322 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -28,6 +28,7 @@ from app.db.models import ( Department, GroupMember, Message, + PhoneCall, Staff, WechatAccount, ) @@ -41,6 +42,7 @@ from app.storage.repos import ( upsert_messages, upsert_staff, upsert_staff_from_account, + upsert_voice_call, upsert_wechat_account, ) @@ -885,3 +887,84 @@ async def test_upsert_chat_group_with_no_members_writes_group_only(): assert members == [] finally: await engine.dispose() + + +# --------------------------------------------------------------------------- +# PhoneCall (voice_call) upsert (Task 2.4) +# --------------------------------------------------------------------------- + + +async def test_upsert_voice_call_is_idempotent_and_updates_record_file(): + """Same call_id twice with different record_file/version -> 1 row, refreshed.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec1 = { + "call_id": "vc_dup", + "channel": "voice_call", + "record_file": "https://example.com/call_v1.mp3", + "call_duration": 10, + "version": "v1", + } + rec2 = { + "call_id": "vc_dup", + "channel": "voice_call", + "record_file": "https://example.com/call_v2.mp3", + "call_duration": 12, + "version": "v2", + } + + await upsert_voice_call(session, rec1) + await session.commit() + await upsert_voice_call(session, rec2) + await session.commit() + + rows = ( + await session.execute( + sa.select(PhoneCall) + .where(PhoneCall.call_id == "vc_dup") + .execution_options(populate_existing=True) + ) + ).scalars().all() + assert len(rows) == 1 + assert rows[0].call_id == "vc_dup" + assert rows[0].channel == "voice_call" + assert rows[0].record_file == "https://example.com/call_v2.mp3" + assert rows[0].call_duration == 12 + assert rows[0].version == "v2" + finally: + await engine.dispose() + + +async def test_upsert_voice_call_null_record_file_and_coerces_ints(): + """null record_file -> None; numeric-str call_duration/start_call_time/ + call_status/incoming_call coerced to int.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "call_id": "vc_null", + "channel": "voice_call", + "record_file": None, + "call_duration": "151", # str -> int + "start_call_time": "1700000000123", # numeric str -> int (ms epoch) + "call_status": "1", # str -> int + "incoming_call": "1", # str -> int + } + await upsert_voice_call(session, rec) + await session.commit() + + fetched = ( + await session.execute( + sa.select(PhoneCall).where(PhoneCall.call_id == "vc_null") + ) + ).scalar_one() + assert fetched.record_file is None + assert fetched.call_duration == 151 + assert fetched.start_call_time == 1700000000123 + assert fetched.call_status == 1 + assert fetched.incoming_call == 1 + finally: + await engine.dispose() diff --git a/yunke-server/tests/test_scrape_wechat_messages_group.py b/yunke-server/tests/test_scrape_wechat_messages_group.py new file mode 100644 index 0000000..9798b60 --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_messages_group.py @@ -0,0 +1,145 @@ +"""Tests for the WeChat message scraper on GROUP conversations (Task 2.3). + +The existing :func:`scrape_wechat_messages` already maps ``roomid`` -> +``room_id`` and ``talker`` -> ``talker_id``, so group chats — where the +conversation id ``friendId`` is a ``xxx@chatroom`` id and every message +carries ``roomid`` set to that chatroom id plus a per-member ``talker`` wxid +(the sender inside the group) — should flow through unchanged. + +This test locks that behavior in: a single group page yields messages with +``room_id`` populated to the chatroom id and ``talker_id`` equal to the +SENDING MEMBER's wxid (NOT the chatroom id), so group traffic is +attributable to the individual sender while still being scoped to the room. + +No scraper code change is expected; if the scraper already handles it, this +test just guards the behavior against regressions. +""" + +from __future__ import annotations + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_messages +from app.session.yunke_client import YunkeClient +from tests.cassettes.msg_chats_fixture import make_msg_chats_response + +BASE = "https://crm2.yunkecn.com" +MSG_CHATS_URL = f"{BASE}/wechatweb/msg/chats" + +USERID = "a" * 32 +COMPANY = "8m0atd" + +WECHAT_ID = "wxid_staff_alice" +# The conversation id (friendId) for a group chat is the @chatroom id. +ROOM_ID = "12345678901@chatroom" +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) + + +# Two group messages from two DIFFERENT members of the same chatroom. The +# `talker` is the individual member's wxid; `roomid` is the chatroom id (same +# for every message in the conversation); the upstream sets `oriTalker` to the +# chatroom id for group messages (it carries the conversation key, not the +# sender). +GROUP_MESSAGES = [ + { + "msgSvrId": "grp-2000001", + "type": 1, + "messageType": 1, + "text": "hello group from alice", + "wechatId": WECHAT_ID, + "talker": "wxid_member_a1", # the SENDER (a group member) + "oriTalker": ROOM_ID, # upstream convention: oriTalker = chatroom id + "roomid": ROOM_ID, + "mine": 1, + "isSend": 1, + "timestamp": END - 600000, + "createTime": END - 600000, + "name": "Alice (group)", + "headUrl": "https://cdn.example.com/alice.png", + "isDel": 0, + "version": "v1.0", + }, + { + "msgSvrId": "grp-2000002", + "type": 1, + "messageType": 1, + "text": "hi from bob", + "wechatId": WECHAT_ID, + "talker": "wxid_member_b2", # a DIFFERENT member + "oriTalker": ROOM_ID, + "roomid": ROOM_ID, + "mine": 0, + "isSend": 0, + "timestamp": END - 300000, + "createTime": END - 300000, + "name": "Bob (group)", + "headUrl": "https://cdn.example.com/bob.png", + "isDel": 0, + "version": "v1.0", + }, +] + + +@respx.mock +async def test_scrape_wechat_messages_group_yields_room_id_and_member_talker(): + """Group messages keep ``room_id`` populated and ``talker_id`` = member wxid. + + The scraper must NOT collapse ``talker_id`` to the chatroom id (that would + lose the per-message sender) and must NOT clear ``room_id`` (that would + make the message look like a 1:1 direct message). + """ + # `end` = oldest ts in batch; hasLast=false -> single page, no 2nd fetch. + page = make_msg_chats_response( + GROUP_MESSAGES, end=END - 600000, has_last=False + ) + route = respx.post(MSG_CHATS_URL).mock( + return_value=httpx.Response(200, json=page) + ) + + client = _primed_client() + try: + results = [ + r + async for r in scrape_wechat_messages( + client, WECHAT_ID, ROOM_ID, START, END + ) + ] + finally: + await client.aclose() + + # Single page, hasLast=false -> exactly one call. + assert route.call_count == 1 + assert len(results) == 2 + + r1, r2 = results + + # Both messages carry the chatroom id on room_id... + assert r1["room_id"] == ROOM_ID + assert r2["room_id"] == ROOM_ID + + # ...and the SENDING MEMBER's wxid on talker_id (NOT the chatroom id). + assert r1["talker_id"] == "wxid_member_a1" + assert r2["talker_id"] == "wxid_member_b2" + + # ori_talker carries the chatroom id verbatim (upstream convention). + assert r1["ori_talker"] == ROOM_ID + assert r2["ori_talker"] == ROOM_ID + + # Standard fields still flow through. + assert r1["channel"] == "wechat" + assert r1["owner_account_id"] == WECHAT_ID + assert r1["msg_id"] == "grp-2000001" + assert r1["text"] == "hello group from alice" + assert r1["raw_json"] == GROUP_MESSAGES[0] + + # The friendId we passed in is the chatroom id; it is NOT stamped on the + # record separately (room_id already carries it, so there is no duplicate + # "conversation id" field on Message). + assert r2["msg_id"] == "grp-2000002" diff --git a/yunke-server/tests/test_scrape_wechat_voice_calls.py b/yunke-server/tests/test_scrape_wechat_voice_calls.py new file mode 100644 index 0000000..2f3b32b --- /dev/null +++ b/yunke-server/tests/test_scrape_wechat_voice_calls.py @@ -0,0 +1,261 @@ +"""Tests for the WeChat voice-call scraper (Task 2.4). + +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 page-number paginator's primary stop signal): + +* page 1 -> 2 records, ``total=4`` +* page 2 -> 2 records, ``total=4`` +* page 3 -> empty ``data`` array (no more rows) + +The scraper must yield exactly the 4 records normalized to the ``PhoneCall`` +column shape (``channel="voice_call"``), make exactly 3 calls, and stop +without a 4th. Field mapping asserted end-to-end incl.: + +* ``id`` -> ``call_id`` (PK) +* ``duration`` -> ``call_duration`` (string seconds coerced to int) +* ``startTime`` -> ``start_call_time`` (``YYYY-MM-DD HH:MM:SS`` -> ms epoch) +* ``ossFileName``-> ``record_file`` (may be null) +* ``isSend`` -> ``incoming_call`` with INVERTED polarity + (voice-call ``isSend`` is 0=incoming/1=outgoing; ``PhoneCall.incoming_call`` + is 1=incoming/0=outgoing — so ``incoming_call = 1 - isSend``). +* ``talker`` -> ``call_number`` (the customer / room id) +* ``talkerNickName`` -> ``plan_customer_name`` +""" + +from __future__ import annotations + +import datetime as dt +import json as _json + +import httpx +import respx + +from app.scraper.wechat import scrape_wechat_voice_calls +from app.session.yunke_client import YunkeClient +from tests.cassettes.voiceCall_fixture import ( + DEFAULT_PAGE_SIZE, + TOTAL, + VOICE_RECORDS, + make_voice_list_response, +) + +BASE = "https://crm2.yunkecn.com" +VOICE_LIST_URL = f"{BASE}/wechatweb/callRecord/queryWeChatVoiceList" + +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) + + +def _ms(date_str: str) -> int: + """Expected ms-epoch for a ``YYYY-MM-DD HH:MM:SS`` string parsed as UTC. + + Mirrors the scraper's documented parsing rule so the assertion is an + independent recomputation (not a self-reference). + """ + return int( + dt.datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S") + .replace(tzinfo=dt.UTC) + .timestamp() + * 1000 + ) + + +# --------------------------------------------------------------------------- +# End-to-end: pagination + field mapping + raw_json +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_voice_calls_yields_records_and_paginates(): + """Two full pages + a terminating empty page -> 4 records, 3 calls.""" + page1 = make_voice_list_response( + VOICE_RECORDS[:2], total=TOTAL, page_num=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_voice_list_response( + VOICE_RECORDS[2:], total=TOTAL, page_num=2, page_size=DEFAULT_PAGE_SIZE + ) + page3 = make_voice_list_response( + [], total=TOTAL, page_num=3, page_size=DEFAULT_PAGE_SIZE + ) + + route = respx.post(VOICE_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_voice_calls( + 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["call_id"] for r in results} == {rec["id"] for rec in VOICE_RECORDS} + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, VOICE_RECORDS, strict=True): + # Identity + channel. + assert result["call_id"] == record["id"] + assert result["channel"] == "voice_call" + + # Staff side. + assert result["user_id"] == record["userId"] + assert result["user_id_name"] == record["userName"] + + # Customer side: talker -> call_number, talkerNickName -> plan_customer_name. + assert result["call_number"] == record["talker"] + assert result["plan_customer_name"] == record["talkerNickName"] + + # Recording URL (may be None). + assert result["record_file"] == record["ossFileName"] + + # Status + duration + start time coercions. + assert result["call_status"] == record["callStatus"] + assert result["call_duration"] == int(record["duration"]) + assert result["start_call_time"] == _ms(record["startTime"]) + + # incoming_call polarity: isSend 0=incoming -> 1; 1=outgoing -> 0. + if record["isSend"] == 0: + assert result["incoming_call"] == 1 + else: + assert result["incoming_call"] == 0 + + # Version carried verbatim; raw_json preserves the original record. + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +# --------------------------------------------------------------------------- +# Null ossFileName -> record_file None +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_voice_calls_null_oss_file_name_yields_none_record_file(): + """A record with ``ossFileName=null`` -> ``record_file`` None.""" + rec = { + "id": "vc_null", + "userId": "staff_user_001", + "userName": "Alice Wang", + "talker": "wxid_friend_x", + "talkerNickName": "Customer X", + "isSend": 1, + "callStatus": 2, + "ossFileName": None, + "duration": "0", + "startTime": "2026-01-15 10:00:00", + "weChatId": "wxid_staff_alice01", + } + page1 = make_voice_list_response( + [rec], total=1, page_num=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_voice_list_response( + [], total=1, page_num=2, page_size=DEFAULT_PAGE_SIZE + ) + respx.post(VOICE_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_voice_calls( + client, page_size=DEFAULT_PAGE_SIZE + ) + ] + finally: + await client.aclose() + + assert len(results) == 1 + assert results[0]["record_file"] is None + assert results[0]["call_status"] == 2 + # Outgoing (isSend=1) -> incoming_call=0. + assert results[0]["incoming_call"] == 0 + + +# --------------------------------------------------------------------------- +# Pagination body: pageNum/pageSize advanced correctly +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wechat_voice_calls_sends_pagination_body(): + """Outgoing request body carries pageNum/pageSize; advances per page.""" + page1 = make_voice_list_response( + VOICE_RECORDS[:1], total=TOTAL, page_num=1, page_size=1 + ) + page2 = make_voice_list_response( + [], total=TOTAL, page_num=2, page_size=1 + ) + route = respx.post(VOICE_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_voice_calls(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_voice_calls_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_voice_list_response([], total=0, page_num=1) + route = respx.post(VOICE_LIST_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_wechat_voice_calls(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1