feat(sync): sync_wa_voice_calls + scheduler ENTITY_FUNCS wa_voice_calls

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-07 09:36:14 +08:00
parent 169a64ee29
commit 18620870c6
4 changed files with 48 additions and 9 deletions
+34 -6
View File
@@ -57,6 +57,7 @@ from app.scraper.wechat import (
scrape_wechat_messages, scrape_wechat_messages,
scrape_wechat_voice_calls, scrape_wechat_voice_calls,
) )
from app.scraper.whatsapp import scrape_wa_voice_calls
from app.session.yunke_client import YunkeClient from app.session.yunke_client import YunkeClient
from app.storage.object_store import ObjectStore from app.storage.object_store import ObjectStore
from app.storage.repos import ( from app.storage.repos import (
@@ -84,6 +85,7 @@ __all__ = [
"sync_wechat_groups", "sync_wechat_groups",
"sync_wechat_messages", "sync_wechat_messages",
"sync_wechat_voice_calls", "sync_wechat_voice_calls",
"sync_wa_voice_calls",
] ]
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -98,6 +100,7 @@ ENTITY_WECHAT_ACCOUNT = "wechat_account"
ENTITY_WECHAT_FRIEND = "wechat_friend" ENTITY_WECHAT_FRIEND = "wechat_friend"
ENTITY_WECHAT_GROUP = "wechat_group" ENTITY_WECHAT_GROUP = "wechat_group"
ENTITY_WECHAT_VOICE_CALL = "wechat_voice_call" ENTITY_WECHAT_VOICE_CALL = "wechat_voice_call"
ENTITY_WA_VOICE_CALL = "wa_voice_call"
ENTITY_MESSAGE = "message" ENTITY_MESSAGE = "message"
# M4 (/pc/ CRM module) — phone calls + SMS share no channel column with the # M4 (/pc/ CRM module) — phone calls + SMS share no channel column with the
# wechat-family entities; they track their own SyncRun / SyncState rows. # 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( async def sync_phone_calls(
session: AsyncSession, client: YunkeClient session: AsyncSession, client: YunkeClient
) -> SyncResult: ) -> SyncResult:
@@ -560,17 +586,18 @@ async def sync_all_wechat(
) -> list[SyncResult]: ) -> list[SyncResult]:
"""Run the full WeChat sync sequence: org/entities first, then messages. """Run the full WeChat sync sequence: org/entities first, then messages.
Order: departments → accounts → friends → groups → voice_calls, then the Order: departments → accounts → friends → groups → voice_calls →
per-conversation message sync for every ``(wechat_id, friend_id)`` in wa_voice_calls, then the per-conversation message sync for every
``conversations``. Each step returns its own :class:`SyncResult` so a single ``(wechat_id, friend_id)`` in ``conversations``. Each step returns its own
failed entity does NOT abort the batch (the failure is recorded and the :class:`SyncResult` so a single failed entity does NOT abort the batch
next entity proceeds). Returns the results in execution order. (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 ``conversations`` is the explicit "which direct / room chats to pull
messages for" list. Full auto-discovery from the talkers endpoint is a 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 LATER concern (Task 2.5 wires the scraper; wiring it into the coordinator
is intentionally deferred to keep this entrypoint bounded and testable). 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. entities run.
""" """
results: list[SyncResult] = [ results: list[SyncResult] = [
@@ -579,6 +606,7 @@ async def sync_all_wechat(
await sync_wechat_friends(session, client), await sync_wechat_friends(session, client),
await sync_wechat_groups(session, client), await sync_wechat_groups(session, client),
await sync_wechat_voice_calls(session, client), await sync_wechat_voice_calls(session, client),
await sync_wa_voice_calls(session, client),
] ]
for wechat_id, friend_id in conversations or []: for wechat_id, friend_id in conversations or []:
results.append( results.append(
+8
View File
@@ -58,6 +58,7 @@ from app.sync.orchestrator import (
sync_departments, sync_departments,
sync_phone_calls, sync_phone_calls,
sync_sms, sync_sms,
sync_wa_voice_calls,
sync_wechat_accounts, sync_wechat_accounts,
sync_wechat_friends, sync_wechat_friends,
sync_wechat_groups, sync_wechat_groups,
@@ -152,6 +153,12 @@ async def _run_voice_calls(
return [await sync_wechat_voice_calls(session, client)] 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( async def _run_phone_calls(
session: AsyncSession, client: YunkeClient, store: ObjectStore session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> list[SyncResult]: ) -> list[SyncResult]:
@@ -217,6 +224,7 @@ ENTITY_FUNCS: dict[
"friends": _run_friends, "friends": _run_friends,
"groups": _run_groups, "groups": _run_groups,
"voice_calls": _run_voice_calls, "voice_calls": _run_voice_calls,
"wa_voice_calls": _run_wa_voice_calls,
"phone_calls": _run_phone_calls, "phone_calls": _run_phone_calls,
"sms": _run_sms, "sms": _run_sms,
"messages": _run_messages, "messages": _run_messages,
+5 -3
View File
@@ -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): async def test_sync_all_wechat_runs_each_entity_then_messages(monkeypatch):
"""The coordinator runs departments → accounts → friends → groups → """The coordinator runs departments → accounts → friends → groups →
voice_calls, then per-conversation messages for an explicit list. voice_calls → wa_voice_calls, then per-conversation messages for an
Returns one result per entity/conversation in order.""" explicit list. Returns one result per entity/conversation in order."""
engine = await _make_engine() engine = await _make_engine()
try: try:
SessionLocal = async_sessionmaker(engine, expire_on_commit=False) 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_friends", empty_async_gen)
monkeypatch.setattr(orchestrator, "scrape_wechat_groups", 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_wechat_voice_calls", empty_async_gen)
monkeypatch.setattr(orchestrator, "scrape_wa_voice_calls", empty_async_gen)
monkeypatch.setattr(orchestrator, "scrape_wechat_messages", empty_messages) monkeypatch.setattr(orchestrator, "scrape_wechat_messages", empty_messages)
# Scrapers yield nothing -> upsert / media loops never execute, so the # Scrapers yield nothing -> upsert / media loops never execute, so the
# real upsert_messages / download_message_media (also imported into 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() await session.commit()
entities = [r.entity for r in results] 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 == [ assert entities == [
"department", "department",
"wechat_account", "wechat_account",
"wechat_friend", "wechat_friend",
"wechat_group", "wechat_group",
"wechat_voice_call", "wechat_voice_call",
"wa_voice_call",
"message", "message",
"message", "message",
] ]
@@ -76,6 +76,7 @@ def test_entity_funcs_has_expected_names():
"friends", "friends",
"groups", "groups",
"voice_calls", "voice_calls",
"wa_voice_calls",
"phone_calls", "phone_calls",
"sms", "sms",
"messages", "messages",