feat(scraper): download_call_recordings 下载通话录音到本地 Media + 回写 record_media_hash

复用 download_url/put_by_hash/upsert_media；幂等+resilient（tried 集合防死循环）；media_type='audio'。

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

yunke-server/app/scraper/media.py

@@ -37,12 +37,13 @@ import httpx
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.db.models import Media
+from app.db.models import Media, PhoneCall
 from app.session.yunke_client import YunkeClient
 from app.storage.object_store import ObjectStore
 from app.storage.repos import upsert_media
 
 __all__ = [
+    "download_call_recordings",
     "download_message_media",
     "download_url",
     "guess_mime_type",
@@ -220,3 +221,75 @@ async def download_message_media(
         raw_json=msg.get("raw_json"),
     )
     return media_hash
+
+
+async def download_call_recordings(
+    client: YunkeClient,
+    store: ObjectStore,
+    session: AsyncSession,
+    *,
+    batch_size: int = 50,
+) -> tuple[int, int]:
+    """Download call recordings (``phone_call.record_file`` OSS URLs) into the
+    local object store and link them via ``phone_call.record_media_hash``.
+
+    Iterates ``phone_call`` rows that carry an http(s) ``record_file`` but no
+    ``record_media_hash`` yet. For each: fetch bytes (:func:`download_url`),
+    store content-addressed (:meth:`ObjectStore.put_by_hash`, sha256 dedup),
+    upsert a ``Media`` row (``media_type='audio'``), and write
+    ``record_media_hash`` back onto the call.
+
+    Returns ``(attempted, localized)``. Resilient: a per-row download/store
+    failure is logged and counted as attempted-but-not-localized; no exception
+    escapes. Idempotent within a run: a per-run ``tried`` set excludes already-
+    attempted call_ids from later batches so a persistently-failing row is NOT
+    re-fetched infinitely (it stays ``record_media_hash IS NULL`` and is retried
+    on the NEXT run, which is desirable). Rows already carrying
+    ``record_media_hash`` are skipped; identical bytes are stored once.
+    """
+    attempted = 0
+    localized = 0
+    tried: set[str] = set()
+    while True:
+        stmt = (
+            sa.select(PhoneCall.call_id, PhoneCall.record_file, PhoneCall.channel)
+            .where(
+                PhoneCall.record_file.is_not(None),
+                PhoneCall.record_file.like("http%"),
+                PhoneCall.record_media_hash.is_(None),
+            )
+            .limit(batch_size)
+        )
+        if tried:
+            stmt = stmt.where(PhoneCall.call_id.not_in(tried))
+        rows = (await session.execute(stmt)).all()
+        if not rows:
+            break
+        for call_id, record_file, channel in rows:
+            attempted += 1
+            tried.add(call_id)
+            try:
+                data = await download_url(client, record_file)
+            except (httpx.HTTPError, OSError) as exc:
+                logger.warning("recording download failed for %s: %s", call_id, exc)
+                continue
+            ext = _ext_from_url(record_file)
+            sha256_hex, path = await store.put_by_hash(data, ext=ext)
+            await upsert_media(
+                session,
+                media_hash=sha256_hex,
+                channel=channel or "wechat",
+                media_type="audio",
+                original_url=record_file,
+                local_path=str(path),
+                size_bytes=len(data),
+                mime_type=guess_mime_type(ext),
+            )
+            await session.execute(
+                sa.update(PhoneCall)
+                .where(PhoneCall.call_id == call_id)
+                .values(record_media_hash=sha256_hex)
+            )
+            localized += 1
+        await session.commit()
+    return attempted, localized

yunke-server/tests/test_recording_downloader.py

@@ -0,0 +1,210 @@
+"""Tests for :func:`app.scraper.media.download_call_recordings`.
+
+Covers the call-recording batch downloader that localizes
+``phone_call.record_file`` OSS URLs into the local object store, writes a
+``Media`` row per blob, and links the call back via ``record_media_hash``.
+
+* Happy path: one call -> downloaded, stored, upserted, linked.
+* Idempotent: a call already carrying ``record_media_hash`` is excluded from
+  the candidate set (``attempted == 0``).
+* Resilient: a persistent 500 does NOT hang the loop (the per-run ``tried``
+  set excludes the failing call_id from later batches) and returns
+  ``(attempted=1, localized=0)`` with no exception escaping.
+"""
+
+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 Media, PhoneCall
+from app.scraper.media import download_call_recordings
+from app.session.yunke_client import YunkeClient
+from app.storage.object_store import ObjectStore
+
+BASE = "https://crm2.yunkecn.com"
+# A 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_media_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
+
+
+# ---------------------------------------------------------------------------
+# download_call_recordings
+# ---------------------------------------------------------------------------
+
+
+@respx.mock
+async def test_download_call_recordings_localizes_and_links(tmp_path):
+    """One call with an OSS mp3 record_file: downloaded, stored, linked.
+
+    Returns ``(1, 1)``; the call's ``record_media_hash`` is the sha256 of the
+    fetched bytes; a ``Media`` row exists with ``media_type='audio'``,
+    ``mime_type='audio/mpeg'``, ``original_url == OSS_MP3``; the file exists
+    under ``tmp_path`` at the sha256-content-addressed ``.mp3`` path.
+    """
+    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:
+        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:
+            attempted, localized = await download_call_recordings(
+                client, store, session
+            )
+            await session.commit()
+
+        assert (attempted, localized) == (1, 1)
+
+        # File landed on disk at the sha256-content-addressed path with .mp3.
+        expected_sha = hashlib.sha256(blob).hexdigest()
+        expected_path = tmp_path / expected_sha[:2] / f"{expected_sha}.mp3"
+        assert expected_path.is_file()
+        assert expected_path.read_bytes() == blob
+
+        async with SessionLocal() as session:
+            call = (
+                await session.execute(
+                    sa.select(PhoneCall).where(PhoneCall.call_id == "c1")
+                )
+            ).scalar_one()
+            assert call.record_media_hash == expected_sha
+
+            media = (
+                await session.execute(
+                    sa.select(Media).where(Media.media_hash == expected_sha)
+                )
+            ).scalar_one()
+            assert media.media_type == "audio"
+            assert media.mime_type == "audio/mpeg"
+            assert media.original_url == OSS_MP3
+            assert media.local_path == str(expected_path)
+            assert media.size_bytes == len(blob)
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_idempotent_skip(tmp_path):
+    """A call already carrying ``record_media_hash`` is excluded -> attempted 0.
+
+    The WHERE clause filters ``record_media_hash IS NULL``, so a previously
+    localized call is never re-fetched.
+    """
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            session.add(
+                PhoneCall(
+                    call_id="c2",
+                    channel="voice_call",
+                    record_file=OSS_MP3,
+                    record_media_hash="already",
+                )
+            )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            attempted, localized = await download_call_recordings(
+                client, store, session
+            )
+            await session.commit()
+
+        assert (attempted, localized) == (0, 0)
+        # No network was touched: the candidate set was empty.
+        assert respx.calls.call_count == 0
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_resilient_on_failure_and_no_infinite_loop(tmp_path):
+    """A persistent 500: no exception, loop TERMINATES, ``(1, 0)`` returned.
+
+    This is the key guard for the per-run ``tried`` set: without it, the
+    failing row would stay ``record_media_hash IS NULL`` and be re-selected
+    by every subsequent batch forever, hanging the run. The ``tried`` set
+    excludes the attempted call_id from later batches, so the next SELECT
+    returns an empty page and the loop breaks.
+    """
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+    respx.get(OSS_MP3).mock(return_value=httpx.Response(500))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            session.add(
+                PhoneCall(
+                    call_id="c3", channel="voice_call", record_file=OSS_MP3
+                )
+            )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            attempted, localized = await download_call_recordings(
+                client, store, session
+            )
+            await session.commit()
+
+        assert (attempted, localized) == (1, 0)
+
+        # The failing call is left un-localized (retried on the NEXT run).
+        async with SessionLocal() as session:
+            call = (
+                await session.execute(
+                    sa.select(PhoneCall).where(PhoneCall.call_id == "c3")
+                )
+            ).scalar_one()
+            assert call.record_media_hash is None
+    finally:
+        await client.aclose()
+        await engine.dispose()