diff --git a/yunke-server/app/scraper/whatsapp.py b/yunke-server/app/scraper/whatsapp.py index 8482277..ad9dadb 100644 --- a/yunke-server/app/scraper/whatsapp.py +++ b/yunke-server/app/scraper/whatsapp.py @@ -40,6 +40,7 @@ __all__ = [ "scrape_whatsapp_accounts", "scrape_whatsapp_friends", "scrape_whatsapp_messages", + "scrape_wa_voice_calls", ] WHATSAPP_ACCOUNTS_PATH = "/qqweb/WhatsAppUser/businessSalesList" @@ -47,9 +48,19 @@ WHATSAPP_FRIENDS_PATH = "/qqweb/WhatsAppUser/businessFriendList" # WhatsApp list endpoints return wall-clock strings like "2026-08-01 16:39" # (no seconds) for lastUpdate/createTime — parse to ms epoch (naive UTC). +# The voice-call endpoint reports ``startTime`` as ``YYYY-MM-DD HH:MM:SS`` +# (with seconds); the second format below matches it so this list is reused +# by the WA voice-call scraper without a private copy. _WA_DATETIME_FORMATS = ["%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S"] WHATSAPP_MESSAGES_PATH = "/qqweb/WhatsAppUser/chatRecords" +# The unified yunke voice-call endpoint is shared across the WeChat and +# WhatsApp channels — same path, same envelope, distinguished by the +# ``callSource`` flag in the request body (``"0"`` WeChat / ``"4"`` WA). +# ``scrape_wa_voice_calls`` below wraps the WA variant; the WeChat variant +# lives in :mod:`app.scraper.wechat` (``scrape_wechat_voice_calls``). +WA_VOICE_LIST_PATH = "/wechatweb/callRecord/queryWeChatVoiceList" + # =========================================================================== # Account scraper (M3) — POST /qqweb/WhatsAppUser/businessSalesList @@ -490,3 +501,175 @@ async def scrape_whatsapp_messages( if _is_wa_divider(record): continue yield _normalize_whatsapp_message(record) + + +# =========================================================================== +# Voice-call scraper (Plan 4a Task 1) — POST /wechatweb/callRecord/queryWeChatVoiceList +# with ``callSource=4`` (WhatsApp). +# +# Reuses the unified yunke voice interface also used by the WeChat voice +# scraper (``scrape_wechat_voice_calls``): same path, same flat envelope +# (rows under ``data.data``, declared total under ``data.total``, no +# ``basicResult`` wrapper). The only wire-level difference is the request +# body flag ``callSource="4"`` (WA) vs ``"0"`` (WeChat). Output records are +# tagged ``channel="wa_voice_call"`` so they share the ``PhoneCall`` table +# with the WeChat channel without colliding. +# =========================================================================== + + +def _parse_wa_voice_total(resp: Any) -> int | None: + """Read the declared row count from the ``queryWeChatVoiceList`` envelope. + + Shape: ``data.total`` (an int) — identical to the WeChat voice envelope + (same endpoint). Returns ``None`` if the path is absent so the paginator + falls back to the empty-page stop signal. (Live WA ``total`` is 143 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_wa_voice_rows(resp: Any) -> list[dict[str, Any]]: + """Extract the WA voice-call list from the ``queryWeChatVoiceList`` envelope. + + Shape: ``data.data`` (a list of records) — identical to the WeChat voice + envelope. 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_wa_voice_body(page_num: int, page_size: int) -> dict[str, Any]: + """Build the WA ``queryWeChatVoiceList`` request body for a given page. + + Adds ``callSource="4"`` (the WA flag; WeChat uses ``"0"``) to the + standard ``pageNum`` / ``pageSize`` pagination pair. No filter + parameters — the live capture carries none. + """ + return {"pageNum": page_num, "pageSize": page_size, "callSource": "4"} + + +def _normalize_wa_voice_call(record: dict[str, Any]) -> dict[str, Any]: + """Map one WA ``queryWeChatVoiceList`` record to the ``PhoneCall`` shape. + + Adds ``channel="wa_voice_call"`` and preserves the original record under + ``raw_json``. Field mapping is identical to the WeChat voice normalizer + (``_normalize_voice_call`` in :mod:`app.scraper.wechat`) — the endpoint + is shared, so the record schema is too. The two intentional differences: + + * ``channel`` is ``"wa_voice_call"`` (not ``"voice_call"``) so the WA + rows are distinguishable from WeChat rows in the shared ``PhoneCall`` + table. + * WA-specific fields with no matching ``PhoneCall`` column survive only + via ``raw_json``: ``weChatId`` (the owning WA account id, e.g. + ``@s.whatsapp.net``), ``callFileId``, ``nextId``, ``isRoom`` + and ``from`` (the WA source flag). + + Mapping (same as the WeChat voice normalizer, repeated here so this + scraper is self-documenting): + + * ``id`` -> ``call_id`` (PK) + * ``startTime`` -> ``start_call_time`` (``YYYY-MM-DD HH:MM:SS`` + -> ms epoch via :func:`parse_datetime_to_ms` + ``_WA_DATETIME_FORMATS``) + * ``duration`` -> ``call_duration`` (string seconds coerced to int) + * ``callStatus`` -> ``call_status`` (1=connected w/ recording, + 2=no answer) + * ``isSend`` -> ``incoming_call`` with INVERTED polarity + (``0=incoming -> 1`` / ``1=outgoing -> 0``; ``None`` preserved when + ``isSend`` is missing/non-numeric). + * ``userId`` -> ``user_id`` (owning staff) + * ``userName`` -> ``user_id_name`` + * ``talker`` -> ``call_number`` (the customer WA id) + * ``talkerNickName`` -> ``plan_customer_name`` + * ``ossFileName`` -> ``record_file`` (the recording URL; may be null) + * ``version`` -> ``version`` (carried verbatim) + + 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 + WA voice payload carries none of them. ``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": "wa_voice_call", + "start_call_time": parse_datetime_to_ms( + record.get("startTime"), _WA_DATETIME_FORMATS + ), + "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_wa_voice_calls( + client: YunkeClient, + *, + page_size: int = 50, +) -> AsyncIterator[dict[str, Any]]: + """Yield normalized WhatsApp voice-call records from ``queryWeChatVoiceList``. + + Wraps ``POST /wechatweb/callRecord/queryWeChatVoiceList`` with + ``callSource=4`` — the unified yunke voice interface's WA variant (the + WeChat variant, ``callSource=0``, lives in :mod:`app.scraper.wechat`). + Walks 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="wa_voice_call"`` plus a + verbatim ``raw_json`` copy. + + The envelope is identical to the WeChat voice endpoint: rows sit under + ``data.data``, declared total under ``data.total``, pagination fields + under ``data.pageNum`` / ``data.pageSize``. The WA-specific fields + (``weChatId`` / ``callFileId`` / ``nextId`` / ``isRoom`` / ``from``) + have no ``PhoneCall`` column and survive only via ``raw_json``. + + Persistence reuses :func:`app.storage.repos.upsert_voice_call` (Task 2 + wires the orchestration); the ``channel`` value distinguishes WA rows + from WeChat rows in the shared table. + + 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 WeChat voice + scraper; a sane middle ground between request count and per-response + payload size for this endpoint). + """ + async for record in paginate_by_page( + client, + WA_VOICE_LIST_PATH, + page_size=page_size, + parse_total=_parse_wa_voice_total, + parse_rows=_parse_wa_voice_rows, + build_body=_build_wa_voice_body, + ): + yield _normalize_wa_voice_call(record) diff --git a/yunke-server/tests/cassettes/wa_voice_fixture.py b/yunke-server/tests/cassettes/wa_voice_fixture.py new file mode 100644 index 0000000..69542e3 --- /dev/null +++ b/yunke-server/tests/cassettes/wa_voice_fixture.py @@ -0,0 +1,157 @@ +"""Fixture payloads for ``POST /wechatweb/callRecord/queryWeChatVoiceList`` +with ``callSource=4`` (WhatsApp voice calls). + +The WhatsApp voice-call endpoint is the SAME unified voice interface as the +WeChat one (``queryWeChatVoiceList``) — only the request flag differs +(``callSource="4"`` for WA vs. ``"0"`` for WeChat). The response envelope is +identical to :mod:`tests.cassettes.voiceCall_fixture` (no ``basicResult`` +wrapper; the row list and pagination fields sit directly under ``data``):: + + {"message":"调用成功","code":"10000","success":true,"traceId":"...", + "data":{"data":[...], "total":.., "pageNum":.., "pageSize":..}} + +WA voice records carry the same core schema as WeChat voice calls plus a +few WA-specific fields that have no matching ``PhoneCall`` column and +survive only via ``raw_json``: ``weChatId`` (here the owning WA account +id, e.g. ``@s.whatsapp.net``), ``callFileId``, ``nextId``, +``isRoom`` and ``from``. Live ``total`` is 143 per phase0-findings.md §4 +(WA voice capture). + +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 deliberately exercise the documented mapping edges: + +* ``wa_vc_1001`` — incoming connected call WITH a recording (``isSend=0`` -> + ``incoming_call=1``; ``callStatus=1``; non-null ``ossFileName``; + ``duration`` as a string). +* ``wa_vc_1002`` — outgoing no-answer call with NO recording (``isSend=1`` + -> ``incoming_call=0``; ``callStatus=2``; ``ossFileName=null``). +* ``wa_vc_1003`` — outgoing connected call with a recording. +""" + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Capture-shaped WA voice-call records (3 total across 2 pages). +# --------------------------------------------------------------------------- + +WA_VOICE_RECORDS: list[dict[str, Any]] = [ + { + # Incoming connected call with a recording. + "id": "wa_vc_1001", + "userId": "staff_user_001", + "userName": "Alice Wang", + "userWeChatNickName": "Alice WA", + "userWeChatAlias": "alice_wa", + "talker": "8613800000001@s.whatsapp.net", + "talkerNickName": "WA Customer A", + "talkerAlias": "wa_a", + "callType": 1, + "isSend": 0, # 0=incoming -> incoming_call=1 + "callStatus": 1, # 1=connected w/ recording + "ossFileName": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/whatsapp-call/" + "8m0atd/wa001/ossCallback/1700000001000_1700000151000.mp3" + ), + "duration": "151", # string seconds -> call_duration int 151 + "startTime": "2026-01-15 10:30:00", # parsed to ms-epoch UTC + "weChatId": "8613800000000@s.whatsapp.net", # owning WA account + "callFileId": "cf_wa_1001", + "nextId": "wa_vc_1002", + "isRoom": 0, + "from": 4, # WA source flag + "version": "v1.20260731.wa1001", + }, + { + # Outgoing no-answer call, no recording. + "id": "wa_vc_1002", + "userId": "staff_user_001", + "userName": "Alice Wang", + "userWeChatNickName": "Alice WA", + "userWeChatAlias": "alice_wa", + "talker": "8613800000002@s.whatsapp.net", + "talkerNickName": "WA Lead B", + "talkerAlias": "", + "callType": 1, + "isSend": 1, # 1=outgoing -> incoming_call=0 + "callStatus": 2, # 2=no answer + "ossFileName": None, # no recording -> record_file None + "duration": "0", + "startTime": "2026-01-15 11:00:00", + "weChatId": "8613800000000@s.whatsapp.net", + "callFileId": "cf_wa_1002", + "nextId": "wa_vc_1003", + "isRoom": 0, + "from": 4, + "version": "v1.20260731.wa1002", + }, + { + # Outgoing connected call with a recording. + "id": "wa_vc_1003", + "userId": "staff_user_002", + "userName": "Bob Chen", + "userWeChatNickName": "Bob WA", + "userWeChatAlias": "bob_wa", + "talker": "8613800000003@s.whatsapp.net", + "talkerNickName": "WA Customer C", + "talkerAlias": "wa_c", + "callType": 1, + "isSend": 1, # outgoing -> incoming_call=0 + "callStatus": 1, # connected w/ recording + "ossFileName": ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/whatsapp-call/" + "8m0atd/wa002/ossCallback/1700000003000_1700000045000.mp3" + ), + "duration": "45", + "startTime": "2026-01-15 12:00:00", + "weChatId": "8613800000004@s.whatsapp.net", + "callFileId": "cf_wa_1003", + "nextId": None, + "isRoom": 0, + "from": 4, + "version": "v1.20260731.wa1003", + }, +] + +# Default page size for the fixture (and the scraper). total=3 means the +# fixture fills exactly 2 pages of size 2 (2 + 1 records) plus a +# terminating empty page. +TOTAL = len(WA_VOICE_RECORDS) +DEFAULT_PAGE_SIZE = 2 + + +def make_wa_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`` (callSource=4) 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-wa-{page_num}", + "data": { + "data": list(records), + "total": total, + "pageNum": page_num, + "pageSize": page_size, + }, + } diff --git a/yunke-server/tests/test_scrape_wa_voice_calls.py b/yunke-server/tests/test_scrape_wa_voice_calls.py new file mode 100644 index 0000000..9dabb15 --- /dev/null +++ b/yunke-server/tests/test_scrape_wa_voice_calls.py @@ -0,0 +1,193 @@ +"""Tests for the WhatsApp voice-call scraper (Plan 4a Task 1). + +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 WA voice endpoint reuses the **unified** ``queryWeChatVoiceList`` +interface (the same one the WeChat voice scraper hits) with the +WA-specific ``callSource="4"`` flag — same envelope (``data.data`` / +``data.total``), same record schema, just a different ``channel`` label +(``"wa_voice_call"``). The fixture ships 3 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=3`` +* page 2 -> 1 record, ``total=3`` +* page 3 -> empty ``data`` array (no more rows) + +The scraper must: + +* make exactly 3 calls and stop without a 4th, +* send ``callSource=4`` + advancing ``pageNum``/``pageSize`` in the body, +* yield exactly the 3 records normalized to the ``PhoneCall`` column + shape with ``channel="wa_voice_call"``, +* preserve the WA-specific fields (``weChatId`` / ``callFileId`` / + ``nextId`` / ``isRoom`` / ``from``) via ``raw_json`` (no PhoneCall + columns exist for them). + +Field mapping asserted end-to-end (mirrors the WeChat voice test, +``test_scrape_wechat_voice_calls``): + +* ``id`` -> ``call_id`` (PK) +* ``talker`` -> ``call_number`` (the customer WA id) +* ``talkerNickName`` -> ``plan_customer_name`` +* ``ossFileName``-> ``record_file`` (may be null) +* ``startTime`` -> ``start_call_time`` (``YYYY-MM-DD HH:MM:SS`` -> ms epoch) +* ``duration`` -> ``call_duration`` (string seconds coerced to int) +* ``isSend`` -> ``incoming_call`` with INVERTED polarity (0=incoming + -> ``incoming_call=1``; 1=outgoing -> ``incoming_call=0``). +""" + +from __future__ import annotations + +import datetime as dt +import json as _json + +import httpx +import respx + +from app.scraper.whatsapp import scrape_wa_voice_calls +from app.session.yunke_client import YunkeClient +from tests.cassettes.wa_voice_fixture import ( + DEFAULT_PAGE_SIZE, + TOTAL, + WA_VOICE_RECORDS, + make_wa_voice_list_response, +) + +BASE = "https://crm2.yunkecn.com" +VOICE_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 + body + field mapping + raw_json +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wa_voice_calls_yields_records_and_paginates(): + """Two pages + a terminating empty page -> 3 records, 3 calls. + + Body must carry ``callSource=4`` (the WA flag) plus the standard + ``pageNum``/``pageSize`` pagination pair. Each yielded record is + normalized to the ``PhoneCall`` shape with ``channel="wa_voice_call"`` + and the WA-specific fields preserved via ``raw_json``. + """ + page1 = make_wa_voice_list_response( + WA_VOICE_RECORDS[:2], total=TOTAL, page_num=1, page_size=DEFAULT_PAGE_SIZE + ) + page2 = make_wa_voice_list_response( + WA_VOICE_RECORDS[2:], total=TOTAL, page_num=2, page_size=DEFAULT_PAGE_SIZE + ) + page3 = make_wa_voice_list_response( + [], total=TOTAL, page_num=3, page_size=DEFAULT_PAGE_SIZE + ) + + route = respx.post(VOICE_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_wa_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 3 fixture records (not duplicated, not dropped). + assert len(results) == len(WA_VOICE_RECORDS) + assert {r["call_id"] for r in results} == {rec["id"] for rec in WA_VOICE_RECORDS} + + # First request body: callSource=4 (WA) + standard pagination pair. + body1 = _json.loads(route.calls[0].request.content.decode()) + assert body1["callSource"] == "4" + assert body1["pageNum"] == 1 + assert body1["pageSize"] == DEFAULT_PAGE_SIZE + + # Field mapping for each record (compare against the source record). + for result, record in zip(results, WA_VOICE_RECORDS, strict=True): + # Identity + channel. + assert result["call_id"] == record["id"] + assert result["channel"] == "wa_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. + assert result["incoming_call"] == 1 - record["isSend"] + + # Version carried verbatim; raw_json preserves the original record + # (so WA-specific fields like weChatId / callFileId / nextId / isRoom + # / from survive end-to-end despite having no PhoneCall column). + assert result["version"] == record["version"] + assert result["raw_json"] == record + + +# --------------------------------------------------------------------------- +# Empty first page -> no yields, single call +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_scrape_wa_voice_calls_stops_on_empty_first_page(): + """If the very first page is empty, scraper stops without over-fetching.""" + empty = make_wa_voice_list_response([], total=0, page_num=1) + route = respx.post(VOICE_URL).mock( + return_value=httpx.Response(200, json=empty) + ) + + client = _primed_client() + try: + results = [r async for r in scrape_wa_voice_calls(client)] + finally: + await client.aclose() + + assert results == [] + assert route.call_count == 1