From a0db4535f1ccde419a48c8ddc785ebc0c4409be8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Mon, 10 Aug 2026 17:50:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(sync):=20WA=20messages=20=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=90=8C=E6=AD=A5(per-conversation=20=E5=A2=9E?= =?UTF-8?q?=E9=87=8F=E6=B0=B4=E4=BD=8D,=E9=95=9C=E5=83=8F=20sync=5Fwechat?= =?UTF-8?q?=5Fmessages)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude --- yunke-server/app/sync/orchestrator.py | 103 ++++++++++++++++++++++++++ yunke-server/app/sync/scheduler.py | 41 ++++++++-- 2 files changed, 137 insertions(+), 7 deletions(-) diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 5cbecd3..4219693 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -64,6 +64,7 @@ from app.scraper.whatsapp import ( scrape_wa_voice_calls, scrape_whatsapp_accounts, scrape_whatsapp_friends, + scrape_whatsapp_messages, ) from app.session.yunke_client import YunkeClient from app.storage.object_store import ObjectStore @@ -98,6 +99,7 @@ __all__ = [ "sync_wa_voice_calls", "sync_whatsapp_accounts", "sync_whatsapp_friends", + "sync_whatsapp_messages", ] # --------------------------------------------------------------------------- @@ -683,6 +685,107 @@ async def sync_whatsapp_friends( ) +async def sync_whatsapp_messages( + session: AsyncSession, + client: YunkeClient, + uid: str, + friend_uid: str, +) -> SyncResult: + """Incremental sync of one WhatsApp conversation's messages. + + Mirrors :func:`sync_wechat_messages` for the WA channel: reads + ``last_timestamp`` for ``(message, whatsapp, ":")`` + (default ``0`` on first sync), queries ``[last_timestamp, now]`` via + :func:`scrape_whatsapp_messages`, upserts every yielded message into the + shared ``message`` table (``channel='whatsapp'`` — stamped by the scraper's + normalizer), and advances the watermark to ``max(timestamp)``. A re-run + with no new messages re-queries the empty window and fetches nothing — + the watermark advanced past the previously-seen rows. + + Per Phase 1 this entrypoint does NOT do best-effort media download (unlike + :func:`sync_wechat_messages`); WA media localization is deferred to + Phase 2. The message record itself is still upserted (so the conversation + history is mirrored) — only the media fetch is skipped. + + Returns a :class:`SyncResult`. On a scraper failure, records a + ``status='failed'`` SyncRun with the error and returns a failed result + (does NOT re-raise), so a single broken conversation does NOT abort a + multi-conversation batch. + """ + scope_id = f"{uid}:{friend_uid}" + last_ts = await get_last_timestamp( + session, + ENTITY_MESSAGE, + channel=CHANNEL_WHATSAPP, + scope_id=scope_id, + default=0, + ) + end = _now_ms() + + run = await _start_run( + session, + ENTITY_MESSAGE, + channel=CHANNEL_WHATSAPP, + scope_id=scope_id, + ) + + # --- Scrape (may raise) --- + try: + msgs: list[dict[str, Any]] = [ + msg + async for msg in scrape_whatsapp_messages( + client, uid, friend_uid, last_ts, end + ) + ] + except Exception as exc: + logger.exception( + "scrape_whatsapp_messages failed for %s", scope_id + ) + error = repr(exc) + await _finish_run( + session, run, status="failed", fetched=0, upserted=0, error=error + ) + return SyncResult( + entity=ENTITY_MESSAGE, + status="failed", + fetched=0, + upserted=0, + channel=CHANNEL_WHATSAPP, + scope_id=scope_id, + error=error, + ) + + # --- Upsert (idempotent; same msg_id twice -> one refreshed row) --- + upserted = await upsert_messages(session, msgs) if msgs else 0 + + # NOTE: no media download for WA (Phase 2; records-only for now). + + # --- Advance watermark to max(timestamp) seen in this batch --- + timestamps = [ + m.get("timestamp") for m in msgs if m.get("timestamp") is not None + ] + if timestamps: + await set_watermark( + session, + ENTITY_MESSAGE, + channel=CHANNEL_WHATSAPP, + scope_id=scope_id, + last_timestamp=max(timestamps), + ) + + await _finish_run( + session, run, status="success", fetched=len(msgs), upserted=upserted + ) + return SyncResult( + entity=ENTITY_MESSAGE, + status="success", + fetched=len(msgs), + upserted=upserted, + channel=CHANNEL_WHATSAPP, + scope_id=scope_id, + ) + + 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 2b2f5dd..f6c1202 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -67,6 +67,7 @@ from app.sync.orchestrator import ( sync_wechat_voice_calls, sync_whatsapp_accounts, sync_whatsapp_friends, + sync_whatsapp_messages, ) logger = logging.getLogger(__name__) @@ -87,22 +88,23 @@ DEFAULT_MAX_CONVERSATIONS_PER_RUN = 200 async def _discover_conversations( session: AsyncSession, *, + channel: str = "wechat", limit: int = DEFAULT_MAX_CONVERSATIONS_PER_RUN, ) -> list[tuple[str, str]]: - """Return distinct ``(owner_account_id, friend_id)`` WeChat pairs to sync. + """Return distinct ``(owner_account_id, friend_id)`` pairs to sync. - Source: the local ``customer`` mirror (where - ``channel='wechat'`` and ``owner_account_id`` is non-empty). Reading the - already-mirrored table — instead of the upstream ``talkerList`` endpoint - — keeps the discovery call off the risk-control budget and makes the - incremental run purely a function of the customers we already know + Source: the local ``customer`` mirror, filtered by ``channel`` + (default ``'wechat'``; pass ``'whatsapp'`` for WA conversations). Reading + the already-mirrored table — instead of the upstream ``talkerList`` + endpoint — keeps the discovery call off the risk-control budget and makes + the incremental run purely a function of the customers we already know about. Order is unspecified; ``limit`` caps work per run. """ from app.db.models import Customer stmt = ( sa.select(Customer.owner_account_id, Customer.friend_id) - .where(Customer.channel == "wechat") + .where(Customer.channel == channel) .where(Customer.owner_account_id.is_not(None)) .where(Customer.owner_account_id != "") .distinct() @@ -208,6 +210,26 @@ async def _run_messages( return results +async def _run_wa_messages( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> list[SyncResult]: + """Incremental message sync across all locally-known WA conversations. + + Mirrors :func:`_run_messages`: discovers ``(uid, friend_uid)`` pairs from + the local ``customer`` mirror (``channel='whatsapp'``) and calls + :func:`sync_whatsapp_messages` per conversation. ``store`` is accepted + for signature parity with the other runners but is NOT used — WA media + download is deferred to Phase 2. + """ + conversations = await _discover_conversations(session, channel="whatsapp") + results: list[SyncResult] = [] + for uid, friend_uid in conversations: + results.append( + await sync_whatsapp_messages(session, client, uid, friend_uid) + ) + return results + + async def _run_all_wechat( session: AsyncSession, client: YunkeClient, store: ObjectStore ) -> list[SyncResult]: @@ -236,6 +258,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 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). + results.extend(await _run_wa_messages(session, client, store)) results.append(await sync_phone_calls(session, client)) results.append(await sync_sms(session, client)) return results @@ -259,6 +285,7 @@ ENTITY_FUNCS: dict[ "sms": _run_sms, "call_recordings": _run_call_recordings, "messages": _run_messages, + "wa_messages": _run_wa_messages, "all_wechat": _run_all_wechat, "all": _run_all, }