From 538a0dc635fa5bee371192181f8c3f810556a364 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= <email@sample.com>
Date: Tue, 11 Aug 2026 12:06:26 +0800
Subject: [PATCH] =?UTF-8?q?feat(sync):=20=E9=95=BF=E8=AF=AD=E9=9F=B3?=
 =?UTF-8?q?=E9=9F=B3=E9=A2=91=E6=9C=AC=E5=9C=B0=E5=8C=96(voice=5Fclip=20me?=
 =?UTF-8?q?dia=5Fhash+download+local-first=20playback)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Mirror download_call_recordings pattern for voice_clips:
- VoiceClip.media_hash column + Alembic migration (c4d5e6f7a8b9)
- VoiceClipOut.media_hash + frontend VoiceClip.media_hash type
- download_voice_clip_audio in app/scraper/media.py (resilient, sha256-dedup,
  same download_url/upsert_media helpers as call recordings)
- sync_voice_clip_audio orchestrator wrapper (commits own SyncRun, mirrors
  sync_call_recordings semantics)
- scheduler: _run_voice_clip_audio + ENTITY_FUNCS['voice_clip_audio'] +
  appended to daily _run_all job (after voice_clips)
- VoiceClipModal playClip local-first: /api/media/{media_hash} with OSS
  oss_file_name fallback (mirrors CallList.playRecord)

Tested: 376 pytest pass; vue-tsc clean; migration applied; 1523/1533 clips
localized (10 pending are genuine upstream HTTP 404s — resilient skip).

Co-Authored-By: Claude <noreply@anthropic.com>
---
 yunke-server/app/api/schemas.py               |  3 +
 yunke-server/app/db/models.py                 |  6 ++
 yunke-server/app/scraper/media.py             | 83 ++++++++++++++++++-
 yunke-server/app/sync/orchestrator.py         | 63 +++++++++++++-
 yunke-server/app/sync/scheduler.py            | 15 ++++
 .../c4d5e6f7a8b9_voice_clip_media_hash.py     | 34 ++++++++
 yunkes/src/api/types.ts                       |  2 +
 yunkes/src/components/VoiceClipModal.vue      |  8 +-
 8 files changed, 210 insertions(+), 4 deletions(-)
 create mode 100644 yunke-server/migrations/versions/c4d5e6f7a8b9_voice_clip_media_hash.py

diff --git a/yunke-server/app/api/schemas.py b/yunke-server/app/api/schemas.py
index f735217..9ee5f71 100644
--- a/yunke-server/app/api/schemas.py
+++ b/yunke-server/app/api/schemas.py
@@ -275,6 +275,9 @@ class VoiceClipOut(BaseModel):
     # OSS recording URL — surfaced so the modal can render an inline
     # ``<audio controls>`` player (see spec §6).
     oss_file_name: str | None = None
+    # sha256 media_hash of the localized audio (NULL until downloaded). Playback
+    # prefers /api/media/{this} over the OSS oss_file_name URL (local-first).
+    media_hash: str | None = None
     mp3_duration: int | None = None
     duration_format: str | None = None
     # LOCAL-only 手动匹配 (spec §7): the customer.friend_id this clip has been
diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py
index 7d44255..681f737 100644
--- a/yunke-server/app/db/models.py
+++ b/yunke-server/app/db/models.py
@@ -516,6 +516,12 @@ class VoiceClip(Base):
     duration_format: Mapped[str | None] = mapped_column(String(32), nullable=True)
     channel: Mapped[str] = mapped_column(String(16), nullable=False)
 
+    # sha256 media_hash of the localized audio blob (FK -> Media.media_hash);
+    # NULL until sync_voice_clip_audio downloads it. Populated by the
+    # voice_clip_audio sync pass; playback prefers /api/media/{this} over the
+    # OSS oss_file_name URL.
+    media_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
     # LOCAL-only manual match: references customer.friend_id when an operator
     # has associated this clip with a specific customer (好友). NULL = unmatched.
     # The mirror never pushes this back to 云客 — it's our own bookkeeping for
diff --git a/yunke-server/app/scraper/media.py b/yunke-server/app/scraper/media.py
index ff91ed6..bea2761 100644
--- a/yunke-server/app/scraper/media.py
+++ b/yunke-server/app/scraper/media.py
@@ -37,7 +37,7 @@ import httpx
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.db.models import Media, PhoneCall
+from app.db.models import Media, PhoneCall, VoiceClip
 from app.session.yunke_client import YunkeClient
 from app.storage.object_store import ObjectStore
 from app.storage.repos import upsert_media
@@ -46,6 +46,7 @@ __all__ = [
     "download_call_recordings",
     "download_message_media",
     "download_url",
+    "download_voice_clip_audio",
     "guess_mime_type",
     "media_type_from_message",
 ]
@@ -293,3 +294,83 @@ async def download_call_recordings(
             localized += 1
         await session.commit()
     return attempted, localized
+
+
+async def download_voice_clip_audio(
+    client: YunkeClient,
+    store: ObjectStore,
+    session: AsyncSession,
+    *,
+    batch_size: int = 50,
+) -> tuple[int, int]:
+    """Download voice_clip audio (``voice_clip.oss_file_name`` OSS URLs) into
+    the local object store and link them via ``voice_clip.media_hash``.
+
+    Mirrors :func:`download_call_recordings` exactly: iterates ``voice_clip``
+    rows that carry an http(s) ``oss_file_name`` but no ``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 ``media_hash`` back onto the clip.
+
+    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 voice_clip_ids from later batches so a persistently-failing row
+    is NOT re-fetched infinitely (it stays ``media_hash IS NULL`` and is
+    retried on the NEXT run). Rows already carrying ``media_hash`` are skipped;
+    identical bytes are stored once.
+    """
+    attempted = 0
+    localized = 0
+    tried: set[int] = set()
+    while True:
+        stmt = (
+            sa.select(
+                VoiceClip.voice_clip_id,
+                VoiceClip.oss_file_name,
+                VoiceClip.channel,
+            )
+            .where(
+                VoiceClip.oss_file_name.is_not(None),
+                VoiceClip.oss_file_name.like("http%"),
+                VoiceClip.media_hash.is_(None),
+            )
+            .limit(batch_size)
+        )
+        if tried:
+            stmt = stmt.where(VoiceClip.voice_clip_id.not_in(tried))
+        rows = (await session.execute(stmt)).all()
+        if not rows:
+            break
+        for voice_clip_id, oss_file_name, channel in rows:
+            attempted += 1
+            tried.add(voice_clip_id)
+            try:
+                data = await download_url(client, oss_file_name)
+            except (httpx.HTTPError, OSError) as exc:
+                logger.warning(
+                    "voice_clip audio download failed for %s: %s",
+                    voice_clip_id,
+                    exc,
+                )
+                continue
+            ext = _ext_from_url(oss_file_name)
+            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=oss_file_name,
+                local_path=str(path),
+                size_bytes=len(data),
+                mime_type=guess_mime_type(ext),
+            )
+            await session.execute(
+                sa.update(VoiceClip)
+                .where(VoiceClip.voice_clip_id == voice_clip_id)
+                .values(media_hash=sha256_hex)
+            )
+            localized += 1
+        await session.commit()
+    return attempted, localized
diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py
index 0ce14b4..e7eadc8 100644
--- a/yunke-server/app/sync/orchestrator.py
+++ b/yunke-server/app/sync/orchestrator.py
@@ -51,7 +51,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.models import Department, SyncRun, WechatAccount, WhatsappAccount
 from app.scraper.call import scrape_phone_calls, scrape_sms
-from app.scraper.media import download_call_recordings, download_message_media
+from app.scraper.media import (
+    download_call_recordings,
+    download_message_media,
+    download_voice_clip_audio,
+)
 from app.scraper.voice_clip import scrape_voice_clips
 from app.scraper.wechat import (
     scrape_departments,
@@ -87,12 +91,14 @@ logger = logging.getLogger(__name__)
 
 __all__ = [
     "ENTITY_CALL_RECORDINGS",
+    "ENTITY_VOICE_CLIP_AUDIO",
     "SyncResult",
     "sync_all_wechat",
     "sync_call_recordings",
     "sync_departments",
     "sync_phone_calls",
     "sync_sms",
+    "sync_voice_clip_audio",
     "sync_voice_clips",
     "sync_wechat_accounts",
     "sync_wechat_friends",
@@ -138,6 +144,11 @@ ENTITY_CALL_RECORDINGS = "call_recordings"
 # sync walks every distinct user_id across both staff-account tables and stamps
 # the channel locally (WechatAccount -> wechat, WhatsappAccount -> whatsapp).
 ENTITY_VOICE_CLIP = "voice_clip"
+# Voice-clip audio localization pass: not a scrape — reads already-mirrored
+# ``voice_clip`` rows and downloads their OSS ``oss_file_name`` into the local
+# object store (records a SyncRun with this entity_type; no SyncState row
+# because the voice_clip table itself is the cursor). Mirrors call_recordings.
+ENTITY_VOICE_CLIP_AUDIO = "voice_clip_audio"
 
 CHANNEL_WECHAT = "wechat"
 CHANNEL_WHATSAPP = "whatsapp"
@@ -1026,6 +1037,56 @@ async def sync_voice_clips(
     )
 
 
+async def sync_voice_clip_audio(
+    session: AsyncSession, client: YunkeClient, store: ObjectStore
+) -> SyncResult:
+    """Download localized copies of voice_clip audio for already-mirrored clips.
+
+    Not a scrape: reads local ``voice_clip`` rows whose ``oss_file_name`` (OSS
+    URL) has no ``media_hash`` yet, downloads each via the public OSS URL, and
+    links it. Records a ``SyncRun`` (``entity_type='voice_clip_audio'``);
+    ``fetched`` = clips needing download, ``upserted`` = clips localized.
+
+    Mirrors :func:`sync_call_recordings` exactly (same commit semantics):
+    commits its own ``SyncRun`` row because :func:`download_voice_clip_audio`
+    commits per batch internally, 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 audio pass does
+    NOT abort a multi-entity batch.
+    """
+    run = await _start_run(session, ENTITY_VOICE_CLIP_AUDIO)
+    try:
+        attempted, localized = await download_voice_clip_audio(
+            client, store, session
+        )
+        await _finish_run(
+            session, run, status="success", fetched=attempted, upserted=localized
+        )
+        await session.commit()
+        return SyncResult(
+            entity=ENTITY_VOICE_CLIP_AUDIO,
+            status="success",
+            fetched=attempted,
+            upserted=localized,
+        )
+    except Exception as exc:
+        logger.exception("sync_voice_clip_audio failed")
+        await _finish_run(
+            session, run, status="failed", fetched=0, upserted=0, error=repr(exc)
+        )
+        await session.commit()
+        return SyncResult(
+            entity=ENTITY_VOICE_CLIP_AUDIO,
+            status="failed",
+            fetched=0,
+            upserted=0,
+            error=repr(exc),
+        )
+
+
 # ---------------------------------------------------------------------------
 # Coordinator
 # ---------------------------------------------------------------------------
diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py
index 1715f96..1c976ed 100644
--- a/yunke-server/app/sync/scheduler.py
+++ b/yunke-server/app/sync/scheduler.py
@@ -59,6 +59,7 @@ from app.sync.orchestrator import (
     sync_departments,
     sync_phone_calls,
     sync_sms,
+    sync_voice_clip_audio,
     sync_voice_clips,
     sync_wa_voice_calls,
     sync_wechat_accounts,
@@ -203,6 +204,13 @@ async def _run_call_recordings(
     return [await sync_call_recordings(session, client, store)]
 
 
+async def _run_voice_clip_audio(
+    session: AsyncSession, client: YunkeClient, store: ObjectStore
+) -> list[SyncResult]:
+    """Audio-localization pass for already-mirrored ``voice_clip`` rows."""
+    return [await sync_voice_clip_audio(session, client, store)]
+
+
 async def _run_messages(
     session: AsyncSession, client: YunkeClient, store: ObjectStore
 ) -> list[SyncResult]:
@@ -276,6 +284,12 @@ async def _run_all(
     # tables. Must run AFTER wechat_accounts + wa_accounts (which seed the
     # user_id -> channel mapping the voice-clip sync reads).
     results.append(await sync_voice_clips(session, client))
+    # Voice-clip audio localization: download OSS audio for already-mirrored
+    # voice_clip rows. Runs AFTER voice_clips so freshly-scraped rows are
+    # eligible. Commits internally (mirrors call_recordings) — placing it last
+    # in the daily job means no later in-batch entity can be affected by the
+    # mid-batch commit boundary.
+    results.append(await sync_voice_clip_audio(session, client, store))
     return results
 
 
@@ -297,6 +311,7 @@ ENTITY_FUNCS: dict[
     "sms": _run_sms,
     "voice_clips": _run_voice_clips,
     "call_recordings": _run_call_recordings,
+    "voice_clip_audio": _run_voice_clip_audio,
     "messages": _run_messages,
     "wa_messages": _run_wa_messages,
     "all_wechat": _run_all_wechat,
diff --git a/yunke-server/migrations/versions/c4d5e6f7a8b9_voice_clip_media_hash.py b/yunke-server/migrations/versions/c4d5e6f7a8b9_voice_clip_media_hash.py
new file mode 100644
index 0000000..dc3fb32
--- /dev/null
+++ b/yunke-server/migrations/versions/c4d5e6f7a8b9_voice_clip_media_hash.py
@@ -0,0 +1,34 @@
+"""voice_clip media_hash
+
+Revision ID: c4d5e6f7a8b9
+Revises: b3c4d5e6f7a8
+Create Date: 2026-08-11 00:00:00.000000
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'c4d5e6f7a8b9'
+down_revision: str | Sequence[str] | None = 'b3c4d5e6f7a8'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # sha256 media_hash of the localized audio blob (FK -> Media.media_hash).
+    # NULL until sync_voice_clip_audio downloads it. batch_alter_table keeps
+    # this portable across SQLite (dev/test, table-recreate) and MySQL (prod,
+    # direct ALTER). Mirrors the phone_call.record_media_hash migration.
+    with op.batch_alter_table('voice_clip', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('media_hash', sa.String(length=64), nullable=True))
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    with op.batch_alter_table('voice_clip', schema=None) as batch_op:
+        batch_op.drop_column('media_hash')
diff --git a/yunkes/src/api/types.ts b/yunkes/src/api/types.ts
index 9e0f4b7..59a8f7e 100644
--- a/yunkes/src/api/types.ts
+++ b/yunkes/src/api/types.ts
@@ -169,6 +169,8 @@ export interface VoiceClip {
   end_time?: number | null
   /** OSS recording URL (aliyun OSS 直链). */
   oss_file_name?: string | null
+  /** sha256 media_hash of the localized audio (NULL until downloaded). */
+  media_hash?: string | null
   /** seconds */
   mp3_duration?: number | null
   /** `HH:MM:SS` formatted duration from upstream. */
diff --git a/yunkes/src/components/VoiceClipModal.vue b/yunkes/src/components/VoiceClipModal.vue
index 9b337f6..e198f17 100644
--- a/yunkes/src/components/VoiceClipModal.vue
+++ b/yunkes/src/components/VoiceClipModal.vue
@@ -15,7 +15,8 @@ import { formatCallDuration, formatTime } from '@/utils/format'
  * 关联到 customer.friend_id，不回写云客。matched_friend_name 由后端在 GET 时
  * 经 Customer join 解析（remark > nickname > real_name > friend_id）。
  *
- * 录音播放仍用 `oss_file_name` 的 OSS 直链（已知全本地妥协，随 ASR 一起修）。
+ * 录音播放采用 local-first：优先 `/api/media/{media_hash}`（本地化音频），
+ * 回退到 `oss_file_name` 的 OSS 直链（mirrors CallList.playRecord）。
  */
 const props = withDefaults(
   defineProps<{
@@ -146,7 +147,10 @@ const displayItems = computed(() =>
 )
 
 function playClip(row: VoiceClip): void {
-  const url = row.oss_file_name
+  // Local-first: 优先本地化的 /api/media/{media_hash}，回退到 OSS 直链
+  // (mirrors CallList.playRecord)。
+  const local = row.media_hash ? `/api/media/${row.media_hash}` : null
+  const url = local ?? row.oss_file_name
   if (!url) {
     ElMessage.info('该片段无录音文件')
     return