diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 2afd60b..affcf00 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -48,7 +48,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Department, SyncRun from app.scraper.call import scrape_phone_calls, scrape_sms -from app.scraper.media import download_message_media +from app.scraper.media import download_call_recordings, download_message_media from app.scraper.wechat import ( scrape_departments, scrape_wechat_accounts, @@ -75,8 +75,10 @@ from app.sync.watermarks import get_last_timestamp, set_watermark logger = logging.getLogger(__name__) __all__ = [ + "ENTITY_CALL_RECORDINGS", "SyncResult", "sync_all_wechat", + "sync_call_recordings", "sync_departments", "sync_phone_calls", "sync_sms", @@ -106,6 +108,11 @@ ENTITY_MESSAGE = "message" # wechat-family entities; they track their own SyncRun / SyncState rows. ENTITY_PHONE_CALL = "phone_call" ENTITY_SMS = "sms" +# Recording-localization pass: not a scrape — reads already-mirrored +# ``phone_call`` rows and downloads their OSS ``record_file`` into the local +# object store (records a SyncRun with this entity_type; no SyncState row +# because the phone_call table itself is the cursor). +ENTITY_CALL_RECORDINGS = "call_recordings" CHANNEL_WECHAT = "wechat" @@ -588,6 +595,57 @@ async def sync_wa_voice_calls( ) +async def sync_call_recordings( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> SyncResult: + """Download localized copies of call recordings for already-mirrored calls. + + Not a scrape: reads local ``phone_call`` rows whose ``record_file`` (OSS + URL) has no ``record_media_hash`` yet, downloads each via the public OSS + URL, and links it. Records a ``SyncRun`` (``entity_type='call_recordings'``); + ``fetched`` = calls needing download, ``upserted`` = calls localized. + + Unlike the other ``sync_*`` functions (which defer the commit to the + caller), this one commits its own ``SyncRun`` row: the underlying + :func:`download_call_recordings` commits per batch as it advances, so the + run bookkeeping must land in the same transaction boundary to stay + consistent with what was actually persisted. + + Failure policy mirrors every other ``sync_*``: a downloader exception is + recorded as a ``status='failed'`` ``SyncRun`` and a failed + :class:`SyncResult` is returned (no re-raise), so a broken recording pass + does NOT abort a multi-entity batch. + """ + run = await _start_run(session, ENTITY_CALL_RECORDINGS) + try: + attempted, localized = await download_call_recordings( + client, store, session + ) + await _finish_run( + session, run, status="success", fetched=attempted, upserted=localized + ) + await session.commit() + return SyncResult( + entity=ENTITY_CALL_RECORDINGS, + status="success", + fetched=attempted, + upserted=localized, + ) + except Exception as exc: + logger.exception("sync_call_recordings failed") + await _finish_run( + session, run, status="failed", fetched=0, upserted=0, error=str(exc) + ) + await session.commit() + return SyncResult( + entity=ENTITY_CALL_RECORDINGS, + status="failed", + fetched=0, + upserted=0, + error=str(exc), + ) + + async def sync_phone_calls( session: AsyncSession, client: YunkeClient ) -> SyncResult: diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index fdca23b..b6f3205 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -55,6 +55,7 @@ from app.storage.object_store import ObjectStore from app.sync.orchestrator import ( SyncResult, sync_all_wechat, + sync_call_recordings, sync_departments, sync_phone_calls, sync_sms, @@ -171,6 +172,13 @@ async def _run_sms( return [await sync_sms(session, client)] +async def _run_call_recordings( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> list[SyncResult]: + """Recording-localization pass for already-mirrored ``phone_call`` rows.""" + return [await sync_call_recordings(session, client, store)] + + async def _run_messages( session: AsyncSession, client: YunkeClient, store: ObjectStore ) -> list[SyncResult]: @@ -227,6 +235,7 @@ ENTITY_FUNCS: dict[ "wa_voice_calls": _run_wa_voice_calls, "phone_calls": _run_phone_calls, "sms": _run_sms, + "call_recordings": _run_call_recordings, "messages": _run_messages, "all_wechat": _run_all_wechat, "all": _run_all, diff --git a/yunke-server/tests/test_sync_call_recordings.py b/yunke-server/tests/test_sync_call_recordings.py new file mode 100644 index 0000000..9c59d52 --- /dev/null +++ b/yunke-server/tests/test_sync_call_recordings.py @@ -0,0 +1,202 @@ +"""Tests for :func:`app.sync.orchestrator.sync_call_recordings`. + +Covers the orchestrator's call-recording localization pass: it is NOT a scrape +(it reads already-mirrored ``phone_call`` rows), so the contract under test is +that it: + +* opens a ``SyncRun`` (``entity_type='call_recordings'``), calls + :func:`download_call_recordings`, then closes the run with the returned + counts and returns a matching :class:`SyncResult`; +* on success, the ``SyncRun`` row is ``status='success'`` and the underlying + call's ``record_media_hash`` is populated (the download actually happened); +* on a downloader failure, the run is recorded ``status='failed'`` and a + failed :class:`SyncResult` is returned without re-raising. + +The happy path exercises the real :func:`download_call_recordings` against a +respx-mocked OSS URL and a temp-dir :class:`ObjectStore`, mirroring +``test_recording_downloader.py``. The failure path monkeypatches the +downloader so the orchestrator's exception handling is tested in isolation. +""" + +from __future__ import annotations + +import hashlib + +import httpx +import respx +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import PhoneCall, SyncRun +from app.scraper.media import download_call_recordings +from app.session.yunke_client import YunkeClient +from app.storage.object_store import ObjectStore +from app.sync import orchestrator +from app.sync.orchestrator import ( + ENTITY_CALL_RECORDINGS, + SyncResult, + sync_call_recordings, +) + +BASE = "https://crm2.yunkecn.com" +# Realistic OSS recording URL (same shape as the yunke-pcfile call +# recordings): public GET, .mp3 suffix. +OSS_MP3 = ( + "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/wechat-call/8m0atd/" + "x/ossCallback/a_b.mp3" +) + +USERID = "a" * 32 +COMPANY = "8m0atd" + + +# --------------------------------------------------------------------------- +# Shared helpers (mirror tests/test_recording_downloader.py) +# --------------------------------------------------------------------------- + + +def _primed_client() -> YunkeClient: + """A YunkeClient already in a logged-in state (login flow skipped).""" + return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY) + + +async def _make_engine(): + """Create an in-memory async SQLite engine with all tables created.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return engine + + +# --------------------------------------------------------------------------- +# sync_call_recordings — happy path +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_sync_call_recordings_localizes_and_logs_run(tmp_path): + """One call with an OSS mp3 record_file: download happens, a success + ``SyncRun`` (``entity_type='call_recordings'``) is recorded, and a matching + :class:`SyncResult` is returned with ``fetched == upserted == 1``. The + call's ``record_media_hash`` is populated. + + ``fetched`` here means "calls needing download" and ``upserted`` means + "calls localized" — the recording downloader returns a ``(attempted, + localized)`` tuple, which the orchestrator surfaces verbatim. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + blob = b"ID3\x04\x00\x00\x00\x00\x00\x00mp3-payload" + respx.get(OSS_MP3).mock(return_value=httpx.Response(200, content=blob)) + + client = _primed_client() + try: + # Seed one call whose record_file is an OSS URL and no hash yet. + async with SessionLocal() as session: + session.add( + PhoneCall( + call_id="c1", channel="voice_call", record_file=OSS_MP3 + ) + ) + await session.commit() + + async with SessionLocal() as session: + result = await sync_call_recordings(session, client, store) + # The function commits its own SyncRun row (per the plan's + # contract — differs from the other sync_* funcs that defer commit + # to the caller); no extra commit needed here. + + # Returned the right SyncResult shape. + assert isinstance(result, SyncResult) + assert result.entity == ENTITY_CALL_RECORDINGS + assert result.status == "success" + assert result.fetched == 1 + assert result.upserted == 1 + assert result.error is None + + expected_sha = hashlib.sha256(blob).hexdigest() + + # A SyncRun row exists with entity_type == 'call_recordings'. + async with SessionLocal() as session: + run = ( + await session.execute( + sa.select(SyncRun).where( + SyncRun.entity_type == ENTITY_CALL_RECORDINGS + ) + ) + ).scalar_one() + assert run.status == "success" + assert run.records_fetched == 1 + assert run.records_upserted == 1 + assert run.error is None + + # The call's record_media_hash got set (download happened). + call = ( + await session.execute( + sa.select(PhoneCall).where(PhoneCall.call_id == "c1") + ) + ).scalar_one() + assert call.record_media_hash == expected_sha + finally: + await client.aclose() + await engine.dispose() + + +# --------------------------------------------------------------------------- +# sync_call_recordings — failure path (downloader raises) +# --------------------------------------------------------------------------- + + +async def test_sync_call_recordings_records_failed_run_and_does_not_raise( + monkeypatch, tmp_path +): + """When the downloader raises, the orchestrator records a failed SyncRun + (``entity_type='call_recordings'``) and returns a failed SyncResult — + matching the failure policy of every other ``sync_*`` function. The + exception does NOT escape, so a multi-entity batch keeps going. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + async def boom(client, store, session, *, batch_size=50): + raise RuntimeError("oss exploded") + + monkeypatch.setattr(orchestrator, "download_call_recordings", boom) + + async with SessionLocal() as session: + # Seed a candidate so the downloader WOULD have work (it raises before + # touching rows, but its presence makes the test realistic). + session.add( + PhoneCall(call_id="c2", channel="voice_call", record_file=OSS_MP3) + ) + await session.commit() + + async with SessionLocal() as session: + result = await sync_call_recordings( + session, client=_primed_client(), store=store + ) + + assert isinstance(result, SyncResult) + assert result.entity == ENTITY_CALL_RECORDINGS + assert result.status == "failed" + assert result.fetched == 0 + assert result.upserted == 0 + assert "oss exploded" in (result.error or "") + + async with SessionLocal() as session: + run = ( + await session.execute( + sa.select(SyncRun).where( + SyncRun.entity_type == ENTITY_CALL_RECORDINGS + ) + ) + ).scalar_one() + assert run.status == "failed" + assert run.records_fetched == 0 + assert run.records_upserted == 0 + assert "oss exploded" in (run.error or "") + assert run.finished_at is not None