feat(sync): 长语音音频本地化(voice_clip media_hash+download+local-first playback)

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>
This commit is contained in:
浅念
2026-08-11 12:06:26 +08:00
parent 8b19c299bb
commit 538a0dc635
8 changed files with 210 additions and 4 deletions
+3
View File
@@ -275,6 +275,9 @@ class VoiceClipOut(BaseModel):
# OSS recording URL — surfaced so the modal can render an inline # OSS recording URL — surfaced so the modal can render an inline
# ``<audio controls>`` player (see spec §6). # ``<audio controls>`` player (see spec §6).
oss_file_name: str | None = None 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 mp3_duration: int | None = None
duration_format: str | None = None duration_format: str | None = None
# LOCAL-only 手动匹配 (spec §7): the customer.friend_id this clip has been # LOCAL-only 手动匹配 (spec §7): the customer.friend_id this clip has been
+6
View File
@@ -516,6 +516,12 @@ class VoiceClip(Base):
duration_format: Mapped[str | None] = mapped_column(String(32), nullable=True) duration_format: Mapped[str | None] = mapped_column(String(32), nullable=True)
channel: Mapped[str] = mapped_column(String(16), nullable=False) 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 # LOCAL-only manual match: references customer.friend_id when an operator
# has associated this clip with a specific customer (好友). NULL = unmatched. # has associated this clip with a specific customer (好友). NULL = unmatched.
# The mirror never pushes this back to 云客 — it's our own bookkeeping for # The mirror never pushes this back to 云客 — it's our own bookkeeping for
+82 -1
View File
@@ -37,7 +37,7 @@ import httpx
import sqlalchemy as sa import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession 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.session.yunke_client import YunkeClient
from app.storage.object_store import ObjectStore from app.storage.object_store import ObjectStore
from app.storage.repos import upsert_media from app.storage.repos import upsert_media
@@ -46,6 +46,7 @@ __all__ = [
"download_call_recordings", "download_call_recordings",
"download_message_media", "download_message_media",
"download_url", "download_url",
"download_voice_clip_audio",
"guess_mime_type", "guess_mime_type",
"media_type_from_message", "media_type_from_message",
] ]
@@ -293,3 +294,83 @@ async def download_call_recordings(
localized += 1 localized += 1
await session.commit() await session.commit()
return attempted, localized 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
+62 -1
View File
@@ -51,7 +51,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Department, SyncRun, WechatAccount, WhatsappAccount from app.db.models import Department, SyncRun, WechatAccount, WhatsappAccount
from app.scraper.call import scrape_phone_calls, scrape_sms 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.voice_clip import scrape_voice_clips
from app.scraper.wechat import ( from app.scraper.wechat import (
scrape_departments, scrape_departments,
@@ -87,12 +91,14 @@ logger = logging.getLogger(__name__)
__all__ = [ __all__ = [
"ENTITY_CALL_RECORDINGS", "ENTITY_CALL_RECORDINGS",
"ENTITY_VOICE_CLIP_AUDIO",
"SyncResult", "SyncResult",
"sync_all_wechat", "sync_all_wechat",
"sync_call_recordings", "sync_call_recordings",
"sync_departments", "sync_departments",
"sync_phone_calls", "sync_phone_calls",
"sync_sms", "sync_sms",
"sync_voice_clip_audio",
"sync_voice_clips", "sync_voice_clips",
"sync_wechat_accounts", "sync_wechat_accounts",
"sync_wechat_friends", "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 # sync walks every distinct user_id across both staff-account tables and stamps
# the channel locally (WechatAccount -> wechat, WhatsappAccount -> whatsapp). # the channel locally (WechatAccount -> wechat, WhatsappAccount -> whatsapp).
ENTITY_VOICE_CLIP = "voice_clip" 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_WECHAT = "wechat"
CHANNEL_WHATSAPP = "whatsapp" 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 # Coordinator
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+15
View File
@@ -59,6 +59,7 @@ from app.sync.orchestrator import (
sync_departments, sync_departments,
sync_phone_calls, sync_phone_calls,
sync_sms, sync_sms,
sync_voice_clip_audio,
sync_voice_clips, sync_voice_clips,
sync_wa_voice_calls, sync_wa_voice_calls,
sync_wechat_accounts, sync_wechat_accounts,
@@ -203,6 +204,13 @@ async def _run_call_recordings(
return [await sync_call_recordings(session, client, store)] 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( async def _run_messages(
session: AsyncSession, client: YunkeClient, store: ObjectStore session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> list[SyncResult]: ) -> list[SyncResult]:
@@ -276,6 +284,12 @@ async def _run_all(
# tables. Must run AFTER wechat_accounts + wa_accounts (which seed the # tables. Must run AFTER wechat_accounts + wa_accounts (which seed the
# user_id -> channel mapping the voice-clip sync reads). # user_id -> channel mapping the voice-clip sync reads).
results.append(await sync_voice_clips(session, client)) 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 return results
@@ -297,6 +311,7 @@ ENTITY_FUNCS: dict[
"sms": _run_sms, "sms": _run_sms,
"voice_clips": _run_voice_clips, "voice_clips": _run_voice_clips,
"call_recordings": _run_call_recordings, "call_recordings": _run_call_recordings,
"voice_clip_audio": _run_voice_clip_audio,
"messages": _run_messages, "messages": _run_messages,
"wa_messages": _run_wa_messages, "wa_messages": _run_wa_messages,
"all_wechat": _run_all_wechat, "all_wechat": _run_all_wechat,
@@ -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')
+2
View File
@@ -169,6 +169,8 @@ export interface VoiceClip {
end_time?: number | null end_time?: number | null
/** OSS recording URL (aliyun OSS 直链). */ /** OSS recording URL (aliyun OSS 直链). */
oss_file_name?: string | null oss_file_name?: string | null
/** sha256 media_hash of the localized audio (NULL until downloaded). */
media_hash?: string | null
/** seconds */ /** seconds */
mp3_duration?: number | null mp3_duration?: number | null
/** `HH:MM:SS` formatted duration from upstream. */ /** `HH:MM:SS` formatted duration from upstream. */
+6 -2
View File
@@ -15,7 +15,8 @@ import { formatCallDuration, formatTime } from '@/utils/format'
* 关联到 customer.friend_id,不回写云客。matched_friend_name 由后端在 GET 时 * 关联到 customer.friend_id,不回写云客。matched_friend_name 由后端在 GET 时
* 经 Customer join 解析(remark > nickname > real_name > friend_id)。 * 经 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( const props = withDefaults(
defineProps<{ defineProps<{
@@ -146,7 +147,10 @@ const displayItems = computed(() =>
) )
function playClip(row: VoiceClip): void { 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) { if (!url) {
ElMessage.info('该片段无录音文件') ElMessage.info('该片段无录音文件')
return return