feat(scraper): WA 语音通话抓取 scrape_wa_voice_calls（callSource=4）

复用云客统一语音接口 queryWeChatVoiceList + callSource=4；channel=wa_voice_call 区分；WA 特有字段落 raw_json；复用 PhoneCall 表 + upsert_voice_call。 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-07 09:26:16 +08:00
parent a66c3be00a
commit 169a64ee29
3 changed files with 533 additions and 0 deletions
@@ -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. ``<phone>@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,
+        },
+    }
@@ -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