fix(scraper): voice body 加 departmentId/isRoom/时间窗 + sync 解析 root department_id

实测 queryWeChatVoiceList 必须带 callSource+departmentId+isRoom+startTime+endTime，否则 200 但 data 空（0 数据）。WA callSource=4 / 微信 callSource=0，宽窗抓全量。 sync_*_voice_calls 照 sync_wechat_accounts 模式解析 _find_root_department_id。 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-07 09:49:31 +08:00
parent 18620870c6
commit 79b89bc919
5 changed files with 210 additions and 21 deletions
@@ -1140,12 +1140,41 @@ def _parse_voice_rows(resp: Any) -> list[dict[str, Any]]:
    return rows if isinstance(rows, list) else []


-def _build_voice_body(page_num: int, page_size: int) -> dict[str, Any]:
+def _build_voice_body(
+    page_num: int,
+    page_size: int,
+    *,
+    department_id: str = "",
+    call_source: str = "0",
+    start: str = "2000-01-01 00:00:00",
+    end: str = "2099-12-31 23:59:59",
+) -> dict[str, Any]:
    """Build the ``queryWeChatVoiceList`` request body for a given page.

-    The live capture carries no filter parameters — just the page slice.
+    Live diagnosis (2026-08-07 browser capture) confirmed the endpoint
+    REQUIRES ``callSource`` + ``departmentId`` + ``isRoom`` + ``startTime`` +
+    ``endTime`` — omit any one and the HTTP layer returns 200 with an EMPTY
+    ``data.data`` array (0 rows). The WeChat variant uses
+    ``call_source="0"`` (the WA variant in :mod:`app.scraper.whatsapp`
+    passes ``"4"``); a wide time window (2000-01-01 .. 2099-12-31) pulls
+    the full history in one scrape. ``userId`` / ``sort`` / ``ids`` may
+    stay empty. ``departmentId`` is omitted when blank so a test/fixture
+    call without one still forms a valid body.
    """
-    return {"pageNum": page_num, "pageSize": page_size}
+    body: dict[str, Any] = {
+        "pageNum": page_num,
+        "pageSize": page_size,
+        "callSource": call_source,
+        "isRoom": "0",
+        "startTime": start,
+        "endTime": end,
+        "sort": "",
+        "ids": [],
+        "userId": "",
+    }
+    if department_id:
+        body["departmentId"] = department_id
+    return body


 def _parse_voice_start_time(value: Any) -> int | None:
@@ -1222,6 +1251,7 @@ def _normalize_voice_call(record: dict[str, Any]) -> dict[str, Any]:
 async def scrape_wechat_voice_calls(
    client: YunkeClient,
    *,
+    department_id: str = "",
    page_size: int = 50,
 ) -> AsyncIterator[dict[str, Any]]:
    """Yield normalized WeChat voice-call records from ``queryWeChatVoiceList``.
@@ -1231,6 +1261,14 @@ async def scrape_wechat_voice_calls(
    primary stop signal), mapping each record to the ``PhoneCall`` column
    shape with ``channel="voice_call"`` plus a verbatim ``raw_json`` copy.

+    The endpoint REQUIRES ``callSource`` + ``departmentId`` + ``isRoom`` +
+    a ``startTime``/``endTime`` window — without them it returns 200 + an
+    empty ``data.data`` (0 rows). The orchestrator resolves the
+    company-root ``department_id`` from the mirrored ``Department`` table
+    (see :func:`app.sync.orchestrator.sync_wechat_voice_calls`); the
+    default wide time window ``2000-01-01 .. 2099-12-31`` pulls the full
+    history in one scrape and is not exposed as a parameter.
+
    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`` /
@@ -1241,6 +1279,10 @@ async def scrape_wechat_voice_calls(
    client:
        Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are
        handled by the client.
+    department_id:
+        Company-root department id (the ``departmentId`` form field).
+        Required by the live endpoint for non-empty results; resolved by
+        the orchestrator when omitted (empty skips the field in the body).
    page_size:
        Page size to request. Defaults to 50 (a sane middle ground between
        request count and per-response payload size for this endpoint).
@@ -1252,5 +1294,6 @@ async def scrape_wechat_voice_calls(
        parse_total=_parse_voice_total,
        parse_rows=_parse_voice_rows,
        build_body=_build_voice_body,
+        department_id=department_id,
    ):
        yield _normalize_voice_call(record)
@@ -556,14 +556,39 @@ def _parse_wa_voice_rows(resp: Any) -> list[dict[str, Any]]:
    return rows if isinstance(rows, list) else []


-def _build_wa_voice_body(page_num: int, page_size: int) -> dict[str, Any]:
+def _build_wa_voice_body(
+    page_num: int,
+    page_size: int,
+    *,
+    department_id: str = "",
+    start: str = "2000-01-01 00:00:00",
+    end: str = "2099-12-31 23:59:59",
+) -> 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.
+    Live diagnosis (2026-08-07 browser capture) confirmed the endpoint
+    REQUIRES ``callSource`` + ``departmentId`` + ``isRoom`` + ``startTime`` +
+    ``endTime`` — omit any one and the HTTP layer returns 200 with an EMPTY
+    ``data.data`` array (0 rows). The WA variant uses ``callSource="4"``;
+    a wide time window (2000-01-01 .. 2099-12-31) pulls the full history in
+    one scrape (live total=608). ``userId`` / ``sort`` / ``ids`` may stay
+    empty. ``departmentId`` is omitted when blank so a test/fixture call
+    without one still forms a valid body.
    """
-    return {"pageNum": page_num, "pageSize": page_size, "callSource": "4"}
+    body: dict[str, Any] = {
+        "pageNum": page_num,
+        "pageSize": page_size,
+        "callSource": "4",
+        "isRoom": "0",
+        "startTime": start,
+        "endTime": end,
+        "sort": "",
+        "ids": [],
+        "userId": "",
+    }
+    if department_id:
+        body["departmentId"] = department_id
+    return body


 def _normalize_wa_voice_call(record: dict[str, Any]) -> dict[str, Any]:
@@ -632,6 +657,7 @@ def _normalize_wa_voice_call(record: dict[str, Any]) -> dict[str, Any]:
 async def scrape_wa_voice_calls(
    client: YunkeClient,
    *,
+    department_id: str = "",
    page_size: int = 50,
 ) -> AsyncIterator[dict[str, Any]]:
    """Yield normalized WhatsApp voice-call records from ``queryWeChatVoiceList``.
@@ -644,6 +670,14 @@ async def scrape_wa_voice_calls(
    the ``PhoneCall`` column shape with ``channel="wa_voice_call"`` plus a
    verbatim ``raw_json`` copy.

+    The endpoint REQUIRES ``callSource`` + ``departmentId`` + ``isRoom`` +
+    a ``startTime``/``endTime`` window — without them it returns 200 + an
+    empty ``data.data`` (0 rows). The orchestrator resolves the
+    company-root ``department_id`` from the mirrored ``Department`` table
+    (see :func:`app.sync.orchestrator.sync_wa_voice_calls`); the default
+    wide time window ``2000-01-01 .. 2099-12-31`` pulls the full history
+    (live total=608) and is not exposed as a parameter.
+
    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
@@ -659,6 +693,10 @@ async def scrape_wa_voice_calls(
    client:
        Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are
        handled by the client.
+    department_id:
+        Company-root department id (the ``departmentId`` form field).
+        Required by the live endpoint for non-empty results; resolved by
+        the orchestrator when omitted (empty skips the field in the body).
    page_size:
        Page size to request. Defaults to 50 (matches the WeChat voice
        scraper; a sane middle ground between request count and per-response
@@ -671,5 +709,6 @@ async def scrape_wa_voice_calls(
        parse_total=_parse_wa_voice_total,
        parse_rows=_parse_wa_voice_rows,
        build_body=_build_wa_voice_body,
+        department_id=department_id,
    ):
        yield _normalize_wa_voice_call(record)
@@ -488,21 +488,54 @@ async def sync_wechat_groups(


 async def sync_wechat_voice_calls(
-    session: AsyncSession, client: YunkeClient
+    session: AsyncSession,
+    client: YunkeClient,
+    *,
+    department_id: str | None = None,
 ) -> SyncResult:
-    """Full re-scrape of all WeChat voice-call records and idempotent upsert."""
+    """Full re-scrape of all WeChat voice-call records and idempotent upsert.
+
+    ``department_id`` is the company-root dept id REQUIRED by the
+    ``queryWeChatVoiceList`` form body — without it (and the accompanying
+    ``callSource`` / ``isRoom`` / ``startTime`` / ``endTime`` filters) the
+    endpoint returns 200 with an EMPTY ``data.data`` array (0 rows; live
+    diagnosis 2026-08-07). When omitted, the orchestrator resolves it from
+    the mirrored ``Department`` table via :func:`_find_root_department_id`
+    — which is why :func:`sync_departments` must run first. If no root can
+    be resolved, the scrape is SKIPPED with a logged warning (returns a
+    ``status='skipped'`` :class:`SyncResult` and records no ``SyncRun``) so
+    a missing org tree does not abort a multi-entity batch.
+    """
+    resolved = department_id
+    if resolved is None:
+        resolved = await _find_root_department_id(session)
+    if not resolved:
+        logger.warning(
+            "sync_wechat_voice_calls: no root department_id resolved"
+            " (sync departments first?); skipping queryWeChatVoiceList scrape"
+        )
+        return SyncResult(
+            entity=ENTITY_WECHAT_VOICE_CALL,
+            status="skipped",
+            fetched=0,
+            upserted=0,
+            error="no root department_id available",
+        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WECHAT_VOICE_CALL,
        channel=None,
        scope_id=None,
-        scrape=lambda: scrape_wechat_voice_calls(client),
+        scrape=lambda: scrape_wechat_voice_calls(client, department_id=resolved),
        upsert=upsert_voice_call,
    )


 async def sync_wa_voice_calls(
-    session: AsyncSession, client: YunkeClient
+    session: AsyncSession,
+    client: YunkeClient,
+    *,
+    department_id: str | None = None,
 ) -> SyncResult:
    """Full re-scrape of all WhatsApp voice-call records and idempotent upsert.

@@ -513,13 +546,36 @@ async def sync_wa_voice_calls(
    twice -> one refreshed row). The ``channel="wa_voice_call"`` value the
    scraper stamps distinguishes WA rows from WeChat rows in the shared
    ``PhoneCall`` table; the SyncRun row records ``entity_type='wa_voice_call'``.
+
+    ``department_id`` resolution mirrors :func:`sync_wechat_accounts`: the
+    ``queryWeChatVoiceList`` form body REQUIRES the company-root
+    ``departmentId`` (live diagnosis 2026-08-07 — without it the endpoint
+    returns 200 + empty ``data.data``, 0 rows). When omitted, the
+    orchestrator resolves it from the mirrored ``Department`` table; if no
+    root can be resolved, the scrape is SKIPPED with a logged warning and a
+    ``status='skipped'`` :class:`SyncResult`.
    """
+    resolved = department_id
+    if resolved is None:
+        resolved = await _find_root_department_id(session)
+    if not resolved:
+        logger.warning(
+            "sync_wa_voice_calls: no root department_id resolved"
+            " (sync departments first?); skipping queryWeChatVoiceList scrape"
+        )
+        return SyncResult(
+            entity=ENTITY_WA_VOICE_CALL,
+            status="skipped",
+            fetched=0,
+            upserted=0,
+            error="no root department_id available",
+        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WA_VOICE_CALL,
        channel=None,
        scope_id=None,
-        scrape=lambda: scrape_wa_voice_calls(client),
+        scrape=lambda: scrape_wa_voice_calls(client, department_id=resolved),
        upsert=upsert_voice_call,
    )

@@ -91,10 +91,13 @@ def _ms(date_str: str) -> int:
 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``.
+    Body must carry the full filter set the live endpoint REQUIRES
+    (``callSource=4`` + ``departmentId`` + ``isRoom`` + ``startTime`` +
+    ``endTime`` — omit any and the endpoint returns 200 with empty
+    ``data.data``, 0 rows) 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
@@ -119,7 +122,7 @@ async def test_scrape_wa_voice_calls_yields_records_and_paginates():
        results = [
            r
            async for r in scrape_wa_voice_calls(
-                client, page_size=DEFAULT_PAGE_SIZE
+                client, department_id="DEPT_TEST", page_size=DEFAULT_PAGE_SIZE
            )
        ]
    finally:
@@ -132,11 +135,29 @@ async def test_scrape_wa_voice_calls_yields_records_and_paginates():
    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.
+    # First request body: full required filter set the live endpoint demands.
+    # callSource=4 (WA) + departmentId + isRoom="0" + wide time window +
+    # empty userId/sort/ids (the live endpoint accepts these blank).
    body1 = _json.loads(route.calls[0].request.content.decode())
    assert body1["callSource"] == "4"
+    assert body1["departmentId"] == "DEPT_TEST"
+    assert body1["isRoom"] == "0"
+    assert body1["startTime"] == "2000-01-01 00:00:00"
+    assert body1["endTime"] == "2099-12-31 23:59:59"
    assert body1["pageNum"] == 1
    assert body1["pageSize"] == DEFAULT_PAGE_SIZE
+    # The blank-but-required keys are present (the live endpoint expects
+    # them in the body even when empty).
+    assert body1["userId"] == ""
+    assert body1["sort"] == ""
+    assert body1["ids"] == []
+
+    # Second page advances pageNum and keeps the same filter set verbatim.
+    body2 = _json.loads(route.calls[1].request.content.decode())
+    assert body2["pageNum"] == 2
+    assert body2["pageSize"] == DEFAULT_PAGE_SIZE
+    assert body2["callSource"] == "4"
+    assert body2["departmentId"] == "DEPT_TEST"

    # Field mapping for each record (compare against the source record).
    for result, record in zip(results, WA_VOICE_RECORDS, strict=True):
@@ -205,7 +205,15 @@ async def test_scrape_wechat_voice_calls_null_oss_file_name_yields_none_record_f

@respx.mock
 async def test_scrape_wechat_voice_calls_sends_pagination_body():
-    """Outgoing request body carries pageNum/pageSize; advances per page."""
+    """Body carries the full required filter set + advancing pageNum/pageSize.
+
+    Live diagnosis (2026-08-07) showed ``queryWeChatVoiceList`` REQUIRES
+    ``callSource=0`` + ``departmentId`` + ``isRoom`` + ``startTime`` +
+    ``endTime`` — omit any and the endpoint returns 200 with an empty
+    ``data.data`` (0 rows). The WeChat variant uses ``callSource="0"``;
+    the wide time window ``2000-01-01 .. 2099-12-31`` pulls the full
+    history in one scrape.
+    """
    page1 = make_voice_list_response(
        VOICE_RECORDS[:1], total=TOTAL, page_num=1, page_size=1
    )
@@ -221,7 +229,12 @@ async def test_scrape_wechat_voice_calls_sends_pagination_body():

    client = _primed_client()
    try:
-        results = [r async for r in scrape_wechat_voice_calls(client, page_size=1)]
+        results = [
+            r
+            async for r in scrape_wechat_voice_calls(
+                client, department_id="DEPT_TEST", page_size=1
+            )
+        ]
    finally:
        await client.aclose()

@@ -229,13 +242,30 @@ async def test_scrape_wechat_voice_calls_sends_pagination_body():
    assert len(results) == 1
    assert route.call_count == 2

+    # Page 1 body: full required filter set + standard pagination pair.
    body1 = _json.loads(route.calls[0].request.content.decode())
+    assert body1["callSource"] == "0"
+    assert body1["departmentId"] == "DEPT_TEST"
+    assert body1["isRoom"] == "0"
+    assert body1["startTime"] == "2000-01-01 00:00:00"
+    assert body1["endTime"] == "2099-12-31 23:59:59"
    assert body1["pageNum"] == 1
    assert body1["pageSize"] == 1
+    # Blank-but-required keys are present (the live endpoint expects them
+    # in the body even when empty).
+    assert body1["userId"] == ""
+    assert body1["sort"] == ""
+    assert body1["ids"] == []

+    # Page 2 advances pageNum and keeps the same filter set verbatim.
    body2 = _json.loads(route.calls[1].request.content.decode())
    assert body2["pageNum"] == 2
    assert body2["pageSize"] == 1
+    assert body2["callSource"] == "0"
+    assert body2["departmentId"] == "DEPT_TEST"
+    assert body2["isRoom"] == "0"
+    assert body2["startTime"] == "2000-01-01 00:00:00"
+    assert body2["endTime"] == "2099-12-31 23:59:59"


 # ---------------------------------------------------------------------------