feat(sync): sync_call_recordings + ENTITY_FUNCS['call_recordings'] 独立下载 pass

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

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:

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,

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