xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scraper): WA 群 scraper(businessFriendList+type=1,待风控重置后实测)

Browse Source 
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-11 16:04:03 +08:00
parent 17ee5a1a6c
commit 0e1cb87156
4 changed files with 509 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/scraper/whatsapp.py
+123 -1
View File
@@ -9,6 +9,12 @@ Endpoints covered:
and yields normalized dicts ready to be upserted into
:class:`app.db.models.Customer` with ``channel="whatsapp"``. Page-number
pagination.
* :func:`scrape_whatsapp_groups` — wraps the SAME ``businessFriendList``
endpoint with ``type="1"`` to select WA chat-groups (``"0"`` selects
friends) and yields normalized :class:`app.db.models.ChatGroup` dicts with
``channel="whatsapp"``. The ``type`` group switch is UNVERIFIED (see the
section docstring) — the structure is correct, only the param name may need
a post-risk-control adjustment.
* :func:`scrape_whatsapp_messages` — wraps ``POST /qqweb/WhatsAppUser/chatRecords``
and yields normalized chat-message dicts with ``channel="whatsapp"``.
Backward-cursor pagination by timestamp (the same strategy as WeChat
@@ -34,13 +40,15 @@ from typing import Any
from app.scraper.base import paginate_by_cursor, paginate_by_page
from app.scraper.parsers import coerce_int_or_none, normalize_whatsapp_type, parse_datetime_to_ms
from app.scraper.wechat import _normalize_chat_group
from app.session.yunke_client import YunkeClient
__all__ = [
"scrape_wa_voice_calls",
"scrape_whatsapp_accounts",
"scrape_whatsapp_friends",
"scrape_whatsapp_groups",
"scrape_whatsapp_messages",
"scrape_wa_voice_calls",
]
WHATSAPP_ACCOUNTS_PATH = "/qqweb/WhatsAppUser/businessSalesList"
@@ -317,6 +325,120 @@ async def scrape_whatsapp_friends(
yield _normalize_whatsapp_friend(record)
# ===========================================================================
# Chat-group scraper — POST /qqweb/WhatsAppUser/businessFriendList
# with ``type="1"`` (groups). ``type="0"`` selects friends (see above).
#
# UNVERIFIED (云客 risk-control blocking live API calls as of 2026-08-11):
# the group-type body parameter is inferred, not captured. The WA friends
# body sends ``type="0"`` and the WA CRM SPA exposes two parallel routes —
# ``friendlist/0`` (friends) and ``friendlist/1`` (groups) — so by symmetry
# ``type="1"`` is the most likely group selector. If the post-risk-control
# smoke returns 0 rows, try alternative param names BEFORE concluding the
# endpoint is wrong: ``friendType: "1"``, ``friendTypeYunke: "2"`` (the
# WeChat switch value), or a sibling path entirely. The field mapping and
# pagination structure here are correct regardless; only the switch param
# name may need adjustment.
# ===========================================================================
def _build_wa_group_body(
page_num: int, page_size: int, *, department_id: str = ""
) -> dict[str, Any]:
"""Build the ``businessFriendList`` request body for chat-groups (type=1).
Identical to :func:`_build_wa_friend_body` except for ``type``: ``"1"``
selects groups (``"0"`` selects friends). UNVERIFIED — see the section
docstring above for the post-risk-control verification plan. ``departmentId``
is omitted when blank so a test/fixture call without one still forms a
valid body (the live endpoint requires it for non-empty results, same as
the friends variant).
"""
body: dict[str, Any] = {
"searchInfo": "",
"friendSearchInfo": "",
"page": page_num,
"pageSize": page_size,
# UNVERIFIED group-type switch — see section docstring.
"type": "1",
}
if department_id:
body["departmentId"] = department_id
return body
def _normalize_whatsapp_group(record: dict[str, Any]) -> dict[str, Any]:
"""Map one WA ``businessFriendList`` group record to the ``ChatGroup`` shape.
Reuses :func:`app.scraper.wechat._normalize_chat_group` for the full field
mapping — the WA endpoint returns the SAME field names the WeChat
``friend_list/es`` endpoint does for the group-type switch (``friendWxId``
is the ``@chatroom`` id, ``friendWxNickname`` is the group name,
``friendNum`` is the member count, etc.) — then overrides ``channel`` to
``"whatsapp"`` so WA groups are distinguishable from WeChat groups in the
shared :class:`app.db.models.ChatGroup` table.
Member ids are NOT surfaced by this endpoint (only the ``friendNum``
count); ``member_ids`` / ``member_ids_list`` are emitted as ``None`` /
``[]`` by the shared WeChat normalizer and left untouched here.
"""
out = _normalize_chat_group(record)
out["channel"] = "whatsapp"
return out
async def scrape_whatsapp_groups(
client: YunkeClient,
*,
department_id: str = "",
page_size: int = 50,
) -> AsyncIterator[dict[str, Any]]:
"""Yield normalized WhatsApp chat-group records from ``businessFriendList``.
Mirrors :func:`scrape_whatsapp_friends` but sends ``type="1"`` in the body
to select groups (the WA CRM SPA's ``friendlist/1`` route). Shares the
friends endpoint's envelope (rows under ``data.overviews``, declared total
under ``data.totalCount``), so the same ``_parse_wa_friend_total`` /
``_parse_wa_friend_rows`` helpers are reused.
Each record is mapped to the ``ChatGroup`` column shape with
``channel="whatsapp"`` (via :func:`_normalize_whatsapp_group`, which reuses
the WeChat group normalizer for the field mapping) plus a verbatim
``raw_json`` copy. Member ids are NOT available — ``member_ids`` /
``member_ids_list`` are ``None`` / ``[]``.
.. note::
UNVERIFIED (2026-08-11, risk-control): the ``type="1"`` group switch is
inferred from the WA SPA's ``friendlist/0`` / ``friendlist/1`` routes,
not a live capture. If the post-risk-control smoke returns 0 rows, try
``friendType`` / ``friendTypeYunke`` before concluding the endpoint is
wrong. See the section docstring above.
Parameters
----------
client:
Authenticated :class:`YunkeClient`. Auth headers / 401 re-login are
handled by the client.
department_id:
Company-root dept id (the ``departmentId`` body field). Required by the
live endpoint for non-empty results (same as friends); resolved by the
orchestrator when omitted.
page_size:
Page size to request. Defaults to 50.
"""
async for record in paginate_by_page(
client,
WHATSAPP_FRIENDS_PATH,
page_size=page_size,
parse_total=_parse_wa_friend_total,
parse_rows=_parse_wa_friend_rows,
build_body=_build_wa_group_body,
department_id=department_id,
):
yield _normalize_whatsapp_group(record)
# ===========================================================================
# Message scraper (M3) — POST /qqweb/WhatsAppUser/chatRecords
# ===========================================================================
yunke-server/app/sync/orchestrator.py
+66 -1
View File
@@ -69,6 +69,7 @@ from app.scraper.whatsapp import (
scrape_wa_voice_calls,
scrape_whatsapp_accounts,
scrape_whatsapp_friends,
scrape_whatsapp_groups,
scrape_whatsapp_messages,
)
from app.session.yunke_client import YunkeClient
@@ -100,14 +101,15 @@ __all__ = [
"sync_sms",
"sync_voice_clip_audio",
"sync_voice_clips",
"sync_wa_voice_calls",
"sync_wechat_accounts",
"sync_wechat_friends",
"sync_wechat_groups",
"sync_wechat_messages",
"sync_wechat_voice_calls",
"sync_wa_voice_calls",
"sync_whatsapp_accounts",
"sync_whatsapp_friends",
"sync_whatsapp_groups",
"sync_whatsapp_messages",
]
@@ -130,6 +132,12 @@ ENTITY_WA_VOICE_CALL = "wa_voice_call"
# org's department tree with WeChat (sync_departments seeds it).
ENTITY_WHATSAPP_ACCOUNT = "whatsapp_account"
ENTITY_WHATSAPP_FRIEND = "whatsapp_friend"
# WhatsApp chat-groups share the ``ChatGroup`` table with wechat groups (the
# ``channel`` column distinguishes them). The scraper wraps
# ``businessFriendList`` with the group-type switch (UNVERIFIED param — see
# :func:`app.scraper.whatsapp.scrape_whatsapp_groups`); records land with
# ``channel='whatsapp'``.
ENTITY_WHATSAPP_GROUP = "whatsapp_group"
ENTITY_MESSAGE = "message"
# M4 (/pc/ CRM module) — phone calls + SMS share no channel column with the
# wechat-family entities; they track their own SyncRun / SyncState rows.
@@ -703,6 +711,63 @@ async def sync_whatsapp_friends(
)
async def sync_whatsapp_groups(
session: AsyncSession,
client: YunkeClient,
*,
department_id: str | None = None,
) -> SyncResult:
"""Full re-scrape of all WhatsApp chat-groups and idempotent upsert.
Mirrors :func:`sync_wechat_groups` (the WA analogue) but with the WA
department-resolution path borrowed from :func:`sync_whatsapp_friends`:
the WA ``businessFriendList`` body REQUIRES ``departmentId`` (the
company-root dept id) or it returns empty. 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.
Records land in the shared :class:`ChatGroup` table with
``channel='whatsapp'`` (stamped by :func:`_normalize_whatsapp_group`), so
:func:`upsert_chat_group` is reused directly — exactly the same pattern as
:func:`sync_wechat_groups`. The ``channel`` column distinguishes WA groups
from WeChat groups.
.. note::
UNVERIFIED (2026-08-11, risk-control): the WA group ``type="1"`` body
switch is inferred, not captured. See
:func:`app.scraper.whatsapp.scrape_whatsapp_groups`. If a post-risk-
control smoke returns 0 rows, the param name is the likely culprit.
"""
resolved = department_id
if resolved is None:
resolved = await _find_root_department_id(session)
if not resolved:
logger.warning(
"sync_whatsapp_groups: no root department_id resolved"
" (sync departments first?); skipping businessFriendList scrape"
)
return SyncResult(
entity=ENTITY_WHATSAPP_GROUP,
status="skipped",
fetched=0,
upserted=0,
error="no root department_id available",
)
return await _sync_full_rescrape(
session,
entity_type=ENTITY_WHATSAPP_GROUP,
channel=CHANNEL_WHATSAPP,
scope_id=None,
scrape=lambda: scrape_whatsapp_groups(client, department_id=resolved),
upsert=upsert_chat_group,
)
async def sync_whatsapp_messages(
session: AsyncSession,
client: YunkeClient,
yunke-server/app/sync/scheduler.py
+12
View File
@@ -69,6 +69,7 @@ from app.sync.orchestrator import (
sync_wechat_voice_calls,
sync_whatsapp_accounts,
sync_whatsapp_friends,
sync_whatsapp_groups,
sync_whatsapp_messages,
)
@@ -178,6 +179,12 @@ async def _run_wa_friends(
return [await sync_whatsapp_friends(session, client)]
async def _run_wa_groups(
session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> list[SyncResult]:
return [await sync_whatsapp_groups(session, client)]
async def _run_phone_calls(
session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> list[SyncResult]:
@@ -274,6 +281,10 @@ async def _run_all(
# which ``_run_all_wechat`` seeds via sync_departments — so order matters.
results.append(await sync_whatsapp_accounts(session, client))
results.append(await sync_whatsapp_friends(session, client))
# WA chat-groups: share the ChatGroup table with wechat groups (channel
# distinguishes). Same department_id dependency as wa_friends. UNVERIFIED
# group-type switch — see scrape_whatsapp_groups.
results.append(await sync_whatsapp_groups(session, client))
# WA messages: per-conversation incremental sync, discovered from the
# already-mirrored ``customer`` table (channel='whatsapp'). Must run
# AFTER wa_friends (which seeds the customer rows the discovery reads).
@@ -307,6 +318,7 @@ ENTITY_FUNCS: dict[
"wa_voice_calls": _run_wa_voice_calls,
"wa_accounts": _run_wa_accounts,
"wa_friends": _run_wa_friends,
"wa_groups": _run_wa_groups,
"phone_calls": _run_phone_calls,
"sms": _run_sms,
"voice_clips": _run_voice_clips,
yunke-server/tests/test_scrape_whatsapp_groups.py
+308
View File
@@ -0,0 +1,308 @@
"""Tests for the WhatsApp chat-group scraper.
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.
WA groups reuse the SAME upstream endpoint as WA friends
(``POST /qqweb/WhatsAppUser/businessFriendList``); the ``type="1"`` body
parameter selects groups (``"0"`` selects friends). The response envelope is
identical to the WA friends scraper's (``data.overviews`` +
``data.totalCount``), so this test reuses
:func:`tests.cassettes.wa_friends_fixture.make_wa_friends_response` to build
capture-shaped envelopes.
The records mirror the field names returned by the WeChat
``friend_list/es`` group endpoint (``friendWxId`` / ``friendWxNickname`` /
``friendNum`` / ...) — the WA endpoint is assumed to return the same shape for
groups because both channels share the yunke CRM's friend-list contract.
``_normalize_whatsapp_group`` reuses the WeChat
:func:`app.scraper.wechat._normalize_chat_group` field mapping and overrides
``channel`` to ``"whatsapp"``.
NOTE: the group-type switch ``type="1"`` is UNVERIFIED (云客 risk-control
blocking live API calls as of 2026-08-11) — see the scraper docstring. These
tests pin the STRUCTURE (body shape, normalization, channel override,
pagination); only the param NAME may need a post-risk-control adjustment.
The fixture ships 3 records across 2 pages of size 2, plus a terminating
empty page (the documented primary stop signal for page-number endpoints):
* page 1 -> 2 records, ``totalCount=3``
* page 2 -> 1 record, ``totalCount=3``
* page 3 -> empty ``overviews`` array (no more rows)
The scraper must yield exactly the 3 records (normalized to ``ChatGroup``
snake_case columns + ``raw_json``, ``channel="whatsapp"``), make exactly 3
calls, and stop without a 4th. Member details are NOT surfaced by this
endpoint — ``member_ids`` is ``None`` and ``member_ids_list`` is ``[]`` on
every yielded record.
"""
from __future__ import annotations
import json as _json
from typing import Any
import httpx
import respx
from app.scraper.whatsapp import scrape_whatsapp_groups
from app.session.yunke_client import YunkeClient
from tests.cassettes.wa_friends_fixture import make_wa_friends_response
BASE = "https://crm2.yunkecn.com"
WA_FRIENDS_URL = f"{BASE}/qqweb/WhatsAppUser/businessFriendList"
USERID = "a" * 32
COMPANY = "8m0atd"
# Realistic, capture-shaped WA group records. Field names mirror the WeChat
# ``friend_list/es`` group payload (verified 2026-08-10 Playwright capture) —
# the WA ``businessFriendList`` endpoint is assumed to return the SAME shape
# for the group-type switch because both channels share the yunke CRM's
# friend-list contract. Every field with a target ``ChatGroup`` column is
# populated so the test asserts a full mapping end-to-end. (Member ids are
# NOT in this payload — the endpoint surfaces only the ``friendNum`` count —
# so ``member_ids`` / ``member_ids_list`` are exercised as None / [] on every
# record.)
WA_GROUP_RECORDS: list[dict[str, Any]] = [
{
"friendWxId": "120363000000000000@g.us", # WA group id (not @chatroom)
"friendWxNickname": "VIP WA Customer Group",
"friendNum": 5,
"friendHeadImageUrl": "https://wx.qlogo.cn/wa_group_001.png",
"wxId": "8613800000001@s.whatsapp.net", # owning WA staff account
"ownerName": "Alice Wang",
"userId": "wa_staff_001",
"phone": "8613800000001",
"wechatPhone": "8613900000001",
"messageCount": 4521,
"sendNum": 1204,
"lastChatTime": 1700000000123,
"version": "v1.20260811.waroom1",
},
{
"friendWxId": "120363111111111111@g.us",
"friendWxNickname": "East Region WA Sales",
"friendNum": 8,
"friendHeadImageUrl": "https://wx.qlogo.cn/wa_group_002.png",
"wxId": "8613800000002@s.whatsapp.net",
"ownerName": "Bob Chen",
"userId": "wa_staff_002",
"phone": "8613800000002",
"wechatPhone": "8613900000002",
"messageCount": 9821,
"sendNum": 211,
"lastChatTime": 1700000000456,
"version": "v1.20260811.waroom2",
},
{
# A group whose owner has no resolved staff id (userId None) — the
# scraper must still yield it; None-safe fields stay None.
"friendWxId": "120363222222222222@g.us",
"friendWxNickname": "External WA Coord",
"friendNum": 3,
"friendHeadImageUrl": "https://wx.qlogo.cn/wa_group_003.png",
"wxId": "8613800000004@s.whatsapp.net",
"ownerName": "Orphan WA",
"userId": None,
"phone": None,
"wechatPhone": None,
"messageCount": 12,
"sendNum": 0,
"lastChatTime": 1700000000789,
"version": "v1.20260811.waroom3",
},
]
TOTAL = len(WA_GROUP_RECORDS)
DEFAULT_PAGE_SIZE = 2
def _primed_client() -> YunkeClient:
"""A YunkeClient already in a logged-in state (login flow skipped)."""
return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY)
# ---------------------------------------------------------------------------
# End-to-end: pagination + field mapping + channel override
# ---------------------------------------------------------------------------
@respx.mock
async def test_scrape_whatsapp_groups_yields_records_and_paginates():
"""Two full pages + a terminating empty page -> 3 records, 3 calls."""
page1 = make_wa_friends_response(
WA_GROUP_RECORDS[:2], total_count=TOTAL, page=1, page_size=DEFAULT_PAGE_SIZE
)
page2 = make_wa_friends_response(
WA_GROUP_RECORDS[2:], total_count=TOTAL, page=2, page_size=DEFAULT_PAGE_SIZE
)
page3 = make_wa_friends_response(
[], total_count=TOTAL, page=3, page_size=DEFAULT_PAGE_SIZE
)
route = respx.post(WA_FRIENDS_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_whatsapp_groups(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) == 3
assert {r["room_id"] for r in results} == {
rec["friendWxId"] for rec in WA_GROUP_RECORDS
}
# Field mapping for each record (compare against the source record).
for result, record in zip(results, WA_GROUP_RECORDS, strict=True):
# Identity + channel override (whatsapp, NOT wechat).
assert result["room_id"] == record["friendWxId"]
assert result["channel"] == "whatsapp"
# Naming + ownership.
assert result["room_name"] == record["friendWxNickname"]
assert result["owner_account_id"] == record["wxId"]
assert result["owner_account_nickname"] == record["ownerName"]
assert result["owner_user_id"] == record["userId"]
# phone preferred, wechatPhone fallback.
assert result["owner_user_phone"] == (
record["phone"] or record["wechatPhone"]
)
# Counts.
assert result["member_count"] == record["friendNum"]
assert result["message_count"] == record["messageCount"]
assert result["own_message_count"] == record["sendNum"]
# Avatar + timestamps.
assert result["head_image_url"] == record["friendHeadImageUrl"]
assert result["last_chat_time"] == record["lastChatTime"]
# Member details NOT surfaced by this endpoint -> None / [].
assert result["member_ids"] is None
assert result["member_ids_list"] == []
# Fields the endpoint does not carry.
assert result["loss_members"] is None
assert result["add_members"] is None
assert result["update_time"] is None
# Version carried verbatim; raw_json preserves the original record.
assert result["version"] == record["version"]
assert result["raw_json"] == record
# ---------------------------------------------------------------------------
# member_ids behaviour: always None / [] (not surfaced by this endpoint)
# ---------------------------------------------------------------------------
@respx.mock
async def test_scrape_whatsapp_groups_emits_empty_member_ids():
"""The endpoint does not surface member ids; they are always
``None`` / ``[]`` on yielded records (member mirroring deferred)."""
record = dict(WA_GROUP_RECORDS[0])
page1 = make_wa_friends_response([record], total_count=1, page=1)
page2 = make_wa_friends_response([], total_count=1, page=2)
respx.post(WA_FRIENDS_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_whatsapp_groups(client, page_size=DEFAULT_PAGE_SIZE)
]
finally:
await client.aclose()
assert len(results) == 1
assert results[0]["member_ids"] is None
assert results[0]["member_ids_list"] == []
# ---------------------------------------------------------------------------
# Pagination body: page/pageSize advanced + type="1" (UNVERIFIED group switch)
# ---------------------------------------------------------------------------
@respx.mock
async def test_scrape_whatsapp_groups_sends_pagination_body():
"""Outgoing body carries page/pageSize, departmentId, and ``type="1"``.
The ``type="1"`` group switch is UNVERIFIED (see scraper docstring); this
test pins the current choice so a future post-risk-control adjustment is
surfaced as a deliberate test update, not a silent regression.
"""
page1 = make_wa_friends_response(
[WA_GROUP_RECORDS[0]], total_count=TOTAL, page=1, page_size=1
)
page2 = make_wa_friends_response([], total_count=TOTAL, page=2, page_size=1)
route = respx.post(WA_FRIENDS_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_whatsapp_groups(
client, department_id="ROOT_DEPT", 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())
# businessFriendList uses `page` (not pageNum) + departmentId + type.
assert body1["page"] == 1
assert body1["pageSize"] == 1
assert body1["departmentId"] == "ROOT_DEPT"
# The crucial (UNVERIFIED) switch: type="1" selects GROUPS ("0" = friends).
assert body1["type"] == "1"
body2 = _json.loads(route.calls[1].request.content.decode())
assert body2["page"] == 2
assert body2["pageSize"] == 1
assert body2["type"] == "1"
# ---------------------------------------------------------------------------
# Empty first page -> no yields, single call
# ---------------------------------------------------------------------------
@respx.mock
async def test_scrape_whatsapp_groups_stops_on_empty_first_page():
"""If the very first page is empty, scraper stops without over-fetching."""
empty = make_wa_friends_response([], total_count=0, page=1)
route = respx.post(WA_FRIENDS_URL).mock(
return_value=httpx.Response(200, json=empty)
)
client = _primed_client()
try:
results = [r async for r in scrape_whatsapp_groups(client)]
finally:
await client.aclose()
assert results == []
assert route.call_count == 1