feat(sync): WA messages 自动同步(per-conversation 增量水位,镜像 sync_wechat_messages)

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

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, "<uid>:<friend_uid>")``
+    (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:

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