diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 9876e31..ef533a3 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -57,6 +57,7 @@ from app.scraper.wechat import ( scrape_wechat_messages, scrape_wechat_voice_calls, ) +from app.scraper.whatsapp import scrape_wa_voice_calls from app.session.yunke_client import YunkeClient from app.storage.object_store import ObjectStore from app.storage.repos import ( @@ -84,6 +85,7 @@ __all__ = [ "sync_wechat_groups", "sync_wechat_messages", "sync_wechat_voice_calls", + "sync_wa_voice_calls", ] # --------------------------------------------------------------------------- @@ -98,6 +100,7 @@ ENTITY_WECHAT_ACCOUNT = "wechat_account" ENTITY_WECHAT_FRIEND = "wechat_friend" ENTITY_WECHAT_GROUP = "wechat_group" ENTITY_WECHAT_VOICE_CALL = "wechat_voice_call" +ENTITY_WA_VOICE_CALL = "wa_voice_call" 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. @@ -498,6 +501,29 @@ async def sync_wechat_voice_calls( ) +async def sync_wa_voice_calls( + session: AsyncSession, client: YunkeClient +) -> SyncResult: + """Full re-scrape of all WhatsApp voice-call records and idempotent upsert. + + Mirrors :func:`sync_wechat_voice_calls`: the WA variant of the unified + ``queryWeChatVoiceList`` endpoint (``callSource=4``) has no ``since`` + param, so "incremental" = re-scrape everything and rely on + :func:`upsert_voice_call`'s idempotent Core upsert (same ``call_id`` + twice -> one refreshed row). The ``channel="wa_voice_call"`` value the + scraper stamps distinguishes WA rows from WeChat rows in the shared + ``PhoneCall`` table; the SyncRun row records ``entity_type='wa_voice_call'``. + """ + return await _sync_full_rescrape( + session, + entity_type=ENTITY_WA_VOICE_CALL, + channel=None, + scope_id=None, + scrape=lambda: scrape_wa_voice_calls(client), + upsert=upsert_voice_call, + ) + + async def sync_phone_calls( session: AsyncSession, client: YunkeClient ) -> SyncResult: @@ -560,17 +586,18 @@ async def sync_all_wechat( ) -> list[SyncResult]: """Run the full WeChat sync sequence: org/entities first, then messages. - Order: departments → accounts → friends → groups → voice_calls, then the - per-conversation message sync for every ``(wechat_id, friend_id)`` in - ``conversations``. Each step returns its own :class:`SyncResult` so a single - failed entity does NOT abort the batch (the failure is recorded and the - next entity proceeds). Returns the results in execution order. + Order: departments → accounts → friends → groups → voice_calls → + wa_voice_calls, then the per-conversation message sync for every + ``(wechat_id, friend_id)`` in ``conversations``. Each step returns its own + :class:`SyncResult` so a single failed entity does NOT abort the batch + (the failure is recorded and the next entity proceeds). Returns the + results in execution order. ``conversations`` is the explicit "which direct / room chats to pull messages for" list. Full auto-discovery from the talkers endpoint is a LATER concern (Task 2.5 wires the scraper; wiring it into the coordinator is intentionally deferred to keep this entrypoint bounded and testable). - When ``conversations`` is ``None`` / empty, only the five full-re-scrape + When ``conversations`` is ``None`` / empty, only the six full-re-scrape entities run. """ results: list[SyncResult] = [ @@ -579,6 +606,7 @@ async def sync_all_wechat( await sync_wechat_friends(session, client), await sync_wechat_groups(session, client), await sync_wechat_voice_calls(session, client), + await sync_wa_voice_calls(session, client), ] for wechat_id, friend_id in conversations or []: results.append( diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 68b25fe..fdca23b 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -58,6 +58,7 @@ from app.sync.orchestrator import ( sync_departments, sync_phone_calls, sync_sms, + sync_wa_voice_calls, sync_wechat_accounts, sync_wechat_friends, sync_wechat_groups, @@ -152,6 +153,12 @@ async def _run_voice_calls( return [await sync_wechat_voice_calls(session, client)] +async def _run_wa_voice_calls( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> list[SyncResult]: + return [await sync_wa_voice_calls(session, client)] + + async def _run_phone_calls( session: AsyncSession, client: YunkeClient, store: ObjectStore ) -> list[SyncResult]: @@ -217,6 +224,7 @@ ENTITY_FUNCS: dict[ "friends": _run_friends, "groups": _run_groups, "voice_calls": _run_voice_calls, + "wa_voice_calls": _run_wa_voice_calls, "phone_calls": _run_phone_calls, "sms": _run_sms, "messages": _run_messages, diff --git a/yunke-server/tests/test_orchestrator.py b/yunke-server/tests/test_orchestrator.py index 3ce17fe..b7e6bfe 100644 --- a/yunke-server/tests/test_orchestrator.py +++ b/yunke-server/tests/test_orchestrator.py @@ -471,8 +471,8 @@ async def test_failing_message_scraper_records_failed_run(monkeypatch): async def test_sync_all_wechat_runs_each_entity_then_messages(monkeypatch): """The coordinator runs departments → accounts → friends → groups → - voice_calls, then per-conversation messages for an explicit list. - Returns one result per entity/conversation in order.""" + voice_calls → wa_voice_calls, then per-conversation messages for an + explicit list. Returns one result per entity/conversation in order.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) @@ -490,6 +490,7 @@ async def test_sync_all_wechat_runs_each_entity_then_messages(monkeypatch): monkeypatch.setattr(orchestrator, "scrape_wechat_friends", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_groups", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_voice_calls", empty_async_gen) + monkeypatch.setattr(orchestrator, "scrape_wa_voice_calls", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_messages", empty_messages) # Scrapers yield nothing -> upsert / media loops never execute, so the # real upsert_messages / download_message_media (also imported into the @@ -510,13 +511,14 @@ async def test_sync_all_wechat_runs_each_entity_then_messages(monkeypatch): await session.commit() entities = [r.entity for r in results] - # Five full-re-scrape entities in order, then two message runs. + # Six full-re-scrape entities in order, then two message runs. assert entities == [ "department", "wechat_account", "wechat_friend", "wechat_group", "wechat_voice_call", + "wa_voice_call", "message", "message", ] diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py index 065b077..788b8b1 100644 --- a/yunke-server/tests/test_sync_scheduler.py +++ b/yunke-server/tests/test_sync_scheduler.py @@ -76,6 +76,7 @@ def test_entity_funcs_has_expected_names(): "friends", "groups", "voice_calls", + "wa_voice_calls", "phone_calls", "sms", "messages",