diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 303a735..693530b 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/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: diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index b6f3205..2b2f5dd 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/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,