diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index 6c6fe45..c49e291 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -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) diff --git a/yunke-server/app/scraper/whatsapp.py b/yunke-server/app/scraper/whatsapp.py index ad9dadb..2d8a6ca 100644 --- a/yunke-server/app/scraper/whatsapp.py +++ b/yunke-server/app/scraper/whatsapp.py @@ -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) diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index ef533a3..416ae6b 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -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, ) diff --git a/yunke-server/tests/test_scrape_wa_voice_calls.py b/yunke-server/tests/test_scrape_wa_voice_calls.py index 9dabb15..4f692f7 100644 --- a/yunke-server/tests/test_scrape_wa_voice_calls.py +++ b/yunke-server/tests/test_scrape_wa_voice_calls.py @@ -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): diff --git a/yunke-server/tests/test_scrape_wechat_voice_calls.py b/yunke-server/tests/test_scrape_wechat_voice_calls.py index 2f3b32b..690aa4b 100644 --- a/yunke-server/tests/test_scrape_wechat_voice_calls.py +++ b/yunke-server/tests/test_scrape_wechat_voice_calls.py @@ -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" # ---------------------------------------------------------------------------