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, }