feat(sync): 接 WA 自动同步(accounts+friends wiring to orchestrator/scheduler)

Co-Authored-By: Claude <noreply@anthropic.com>

yunke-server/app/sync/orchestrator.py

@@ -60,7 +60,11 @@ from app.scraper.wechat import (
     scrape_wechat_messages,
     scrape_wechat_voice_calls,
 )
-from app.scraper.whatsapp import scrape_wa_voice_calls
+from app.scraper.whatsapp import (
+    scrape_wa_voice_calls,
+    scrape_whatsapp_accounts,
+    scrape_whatsapp_friends,
+)
 from app.session.yunke_client import YunkeClient
 from app.storage.object_store import ObjectStore
 from app.storage.repos import (
@@ -91,6 +95,8 @@ __all__ = [
     "sync_wechat_messages",
     "sync_wechat_voice_calls",
     "sync_wa_voice_calls",
+    "sync_whatsapp_accounts",
+    "sync_whatsapp_friends",
 ]
 
 # ---------------------------------------------------------------------------
@@ -106,6 +112,12 @@ ENTITY_WECHAT_FRIEND = "wechat_friend"
 ENTITY_WECHAT_GROUP = "wechat_group"
 ENTITY_WECHAT_VOICE_CALL = "wechat_voice_call"
 ENTITY_WA_VOICE_CALL = "wa_voice_call"
+# M3 (WhatsApp accounts + friends). The WA account scraper wraps
+# ``businessSalesList`` (WhatsappAccount table); the WA friend scraper wraps
+# ``businessFriendList`` (Customer table, channel="whatsapp"). Both share the
+# org's department tree with WeChat (sync_departments seeds it).
+ENTITY_WHATSAPP_ACCOUNT = "whatsapp_account"
+ENTITY_WHATSAPP_FRIEND = "whatsapp_friend"
 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.
@@ -118,6 +130,7 @@ ENTITY_SMS = "sms"
 ENTITY_CALL_RECORDINGS = "call_recordings"
 
 CHANNEL_WECHAT = "wechat"
+CHANNEL_WHATSAPP = "whatsapp"
 
 
 # ---------------------------------------------------------------------------
@@ -598,6 +611,77 @@ async def sync_wa_voice_calls(
     )
 
 
+async def sync_whatsapp_accounts(
+    session: AsyncSession,
+    client: YunkeClient,
+) -> SyncResult:
+    """Full re-scrape of all WhatsApp staff accounts (+ embedded staff rows).
+
+    Mirrors :func:`sync_wechat_accounts` but the WA ``businessSalesList``
+    endpoint needs NO ``department_id`` (its body is just the page slice), so
+    there is no root-department resolution / skip path here. Idempotent upsert
+    via :func:`upsert_whatsapp_account` (same ``uid`` twice -> one refreshed
+    row; the companion :func:`upsert_staff_from_account` covers the embedded
+    staff row automatically).
+    """
+    return await _sync_full_rescrape(
+        session,
+        entity_type=ENTITY_WHATSAPP_ACCOUNT,
+        channel=CHANNEL_WHATSAPP,
+        scope_id=None,
+        scrape=lambda: scrape_whatsapp_accounts(client),
+        upsert=upsert_whatsapp_account,
+    )
+
+
+async def sync_whatsapp_friends(
+    session: AsyncSession,
+    client: YunkeClient,
+    *,
+    department_id: str | None = None,
+) -> SyncResult:
+    """Full re-scrape of all WhatsApp friends (customers) and idempotent upsert.
+
+    Mirrors :func:`sync_wechat_accounts`'s department-resolution path: the WA
+    ``businessFriendList`` form body REQUIRES ``departmentId`` (the company-root
+    dept id) or it returns empty (live payload carries ``departmentId`` +
+    ``type``). 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:`Customer` table with
+    ``channel='whatsapp'`` (stamped by :func:`_normalize_whatsapp_friend`),
+    so :func:`upsert_customer` is reused directly — exactly the same pattern as
+    :func:`sync_wechat_friends`.
+    """
+    resolved = department_id
+    if resolved is None:
+        resolved = await _find_root_department_id(session)
+    if not resolved:
+        logger.warning(
+            "sync_whatsapp_friends: no root department_id resolved"
+            " (sync departments first?); skipping businessFriendList scrape"
+        )
+        return SyncResult(
+            entity=ENTITY_WHATSAPP_FRIEND,
+            status="skipped",
+            fetched=0,
+            upserted=0,
+            error="no root department_id available",
+        )
+    return await _sync_full_rescrape(
+        session,
+        entity_type=ENTITY_WHATSAPP_FRIEND,
+        channel=CHANNEL_WHATSAPP,
+        scope_id=None,
+        scrape=lambda: scrape_whatsapp_friends(client, department_id=resolved),
+        upsert=upsert_customer,
+    )
+
+
 async def sync_call_recordings(
     session: AsyncSession, client: YunkeClient, store: ObjectStore
 ) -> SyncResult:

yunke-server/app/sync/scheduler.py

@@ -65,6 +65,8 @@ from app.sync.orchestrator import (
     sync_wechat_groups,
     sync_wechat_messages,
     sync_wechat_voice_calls,
+    sync_whatsapp_accounts,
+    sync_whatsapp_friends,
 )
 
 logger = logging.getLogger(__name__)
@@ -160,6 +162,18 @@ async def _run_wa_voice_calls(
     return [await sync_wa_voice_calls(session, client)]
 
 
+async def _run_wa_accounts(
+    session: AsyncSession, client: YunkeClient, store: ObjectStore
+) -> list[SyncResult]:
+    return [await sync_whatsapp_accounts(session, client)]
+
+
+async def _run_wa_friends(
+    session: AsyncSession, client: YunkeClient, store: ObjectStore
+) -> list[SyncResult]:
+    return [await sync_whatsapp_friends(session, client)]
+
+
 async def _run_phone_calls(
     session: AsyncSession, client: YunkeClient, store: ObjectStore
 ) -> list[SyncResult]:
@@ -216,6 +230,12 @@ async def _run_all(
     This is the entity the scheduler's daily job runs.
     """
     results = await _run_all_wechat(session, client, store)
+    # WhatsApp entities (accounts + friends). WA voice calls are already
+    # included via ``_run_all_wechat``; accounts + friends need their own
+    # kick here. WA friends depend on the department tree (root dept id),
+    # 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))
     results.append(await sync_phone_calls(session, client))
     results.append(await sync_sms(session, client))
     return results
@@ -233,6 +253,8 @@ ENTITY_FUNCS: dict[
     "groups": _run_groups,
     "voice_calls": _run_voice_calls,
     "wa_voice_calls": _run_wa_voice_calls,
+    "wa_accounts": _run_wa_accounts,
+    "wa_friends": _run_wa_friends,
     "phone_calls": _run_phone_calls,
     "sms": _run_sms,
     "call_recordings": _run_call_recordings,