diff --git a/yunke-server/app/scraper/media.py b/yunke-server/app/scraper/media.py index b9432fa..ff91ed6 100644 --- a/yunke-server/app/scraper/media.py +++ b/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 diff --git a/yunke-server/tests/test_recording_downloader.py b/yunke-server/tests/test_recording_downloader.py new file mode 100644 index 0000000..40cc7a9 --- /dev/null +++ b/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()