diff --git a/docs/superpowers/plans/2026-08-07-yunke-recording-localization.md b/docs/superpowers/plans/2026-08-07-yunke-recording-localization.md new file mode 100644 index 0000000..f7c9a76 --- /dev/null +++ b/docs/superpowers/plans/2026-08-07-yunke-recording-localization.md @@ -0,0 +1,471 @@ +# 录音批量本地化（Recording Bulk Localization）Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 把通话录音（`phone_call.record_file` 的云客 OSS mp3 URL）在 sync 时下载到本地 media 存储，运行时前端播放/详情 drawer **优先本地**、未下载完回退 OSS。 + +**Architecture:** 复用现有 media 管道（`ObjectStore` + `Media` 表 + `download_url` + `upsert_media` + `GET /api/media/{hash}`，原作者专为录音预留的扩展点），只补接线：①`phone_call` 加 `record_media_hash` 列；②`download_call_recordings` 下载器；③`sync_call_recordings` 独立 sync pass + scheduler `ENTITY_FUNCS['call_recordings']`；④`PhoneCallOut`/前端类型加字段；⑤CallList `playRecord` + callFields drawer 本地优先。 + +**Tech Stack:** 后端 Python 3.12 + FastAPI + SQLAlchemy 2.0 async + Alembic + httpx（respx 测试）；前端 Vue3 + TS + Element Plus + axios(`baseURL:'/api'`)。 + +**Spec:** `docs/superpowers/specs/2026-08-07-yunke-recording-localization-design.md`（复用主张已 spec-review 核验）。 + +**事实依据（已实测）：** +- 录音 OSS URL 公开免登录（voice_call + wa_voice_call `record_file` 均 `HTTP 200`/`AliyunOSS`/`audio/mpeg`，无需云客 cookie）。 +- 复用文件：`app/storage/object_store.py::ObjectStore.put_by_hash(data,*,ext)`、`app/scraper/media.py::download_url/_ext_from_url/guess_mime_type`、`app/storage/repos.py::upsert_media(...)`、`app/api/routers/media.py::GET /api/media/{media_hash}`。 +- sync 模式：`app/sync/orchestrator.py` 的 `_start_run/_finish_run` + `SyncResult`；`app/sync/scheduler.py` 的 `_run_(session, client, store)` wrapper + `ENTITY_FUNCS`。 +- 前端：`yunkes/src/api/client.ts` axios `baseURL:'/api'`（Vite 代理→:8000），故浏览器里 `/api/media/{hash}` 直接可用。 + +**运行命令（验证用）：** +- 后端测试：`cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_media_downloader.py tests/test_api_calls.py -v` +- 迁移：`cd yunke-server && .venv/Scripts/python.exe -m alembic upgrade head` +- 实跑下载：`cd yunke-server && DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe -m app.sync.cli sync --entity call_recordings` +- 类型检查：`cd yunkes && npx vue-tsc -b` + +--- + +## File Map + +**后端：** +- Modify `yunke-server/app/db/models.py` — `PhoneCall` 加 `record_media_hash` 列 +- Create `yunke-server/migrations/versions/_phone_call_record_media_hash.py` — Alembic 加列 +- Modify `yunke-server/app/scraper/media.py` — 加 `download_call_recordings`（复用本文件 helper） +- Modify `yunke-server/app/sync/orchestrator.py` — `ENTITY_CALL_RECORDINGS` 常量 + `sync_call_recordings` +- Modify `yunke-server/app/sync/scheduler.py` — `_run_call_recordings` wrapper + `ENTITY_FUNCS['call_recordings']` +- Modify `yunke-server/app/api/schemas.py` — `PhoneCallOut.record_media_hash` +- Create/Modify tests：`tests/test_recording_downloader.py`（新）、`tests/test_api_calls.py`（字段断言）、orchestrator sync 测试 + +**前端：** +- Modify `yunkes/src/api/types.ts` — `PhoneCall.record_media_hash` +- Modify `yunkes/src/views/CallList.vue` — `playRecord` 本地优先 + 录音按钮 v-if +- Modify `yunkes/src/utils/callFields.ts` — recordFile audio 字段本地优先（drawer） + +--- + +## Task 1: 后端模型 + Alembic 迁移（phone_call.record_media_hash） + +> 事实：`PhoneCall` 模型在 `app/db/models.py:442-468`，列止于 `raw_json/fetched_at/updated_at`，无 `record_media_hash`。迁移模板见 `migrations/versions/3892dbfab839_customer_add_time.py`（`batch_alter_table` + `add_column`）。 + +**Files:** +- Modify `yunke-server/app/db/models.py`（`PhoneCall` 类，~L458 `record_file` 行下方） +- Create `yunke-server/migrations/versions/_phone_call_record_media_hash.py` +- Test: `yunke-server/tests/test_migration_phone_call_record_media_hash.py`（新，可选；或靠现有迁移 round-trip 测试覆盖） + +- [ ] **Step 1: 加列到模型** + +在 `app/db/models.py` 的 `PhoneCall` 类内，`record_file`（L458）下方加： +```python + # sha256 media_hash of the localized recording blob (FK -> Media.media_hash); + # NULL until sync_call_recordings downloads it. Populated by the + # call_recordings sync pass; playback prefers /api/media/{this} over the + # OSS record_file URL. + record_media_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) +``` +（`String` 已在文件顶部 import。） + +- [ ] **Step 2: 生成/手写迁移** + +手写迁移文件 `migrations/versions/<新rev>_phone_call_record_media_hash.py`（照 `3892dbfab839` 模板；`down_revision` = 当前 head `3892dbfab839`，可用 `alembic heads` 确认）： +```python +"""phone_call record_media_hash + +Revision ID: <生成的新 id> +Revises: 3892dbfab839 +Create Date: 2026-08-07 ... +""" +from collections.abc import Sequence +import sqlalchemy as sa +from alembic import op + +revision: str = '<新 id>' +down_revision: str | Sequence[str] | None = '3892dbfab839' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +def upgrade() -> None: + with op.batch_alter_table('phone_call', schema=None) as batch_op: + batch_op.add_column(sa.Column('record_media_hash', sa.String(length=64), nullable=True)) + +def downgrade() -> None: + with op.batch_alter_table('phone_call', schema=None) as batch_op: + batch_op.drop_column('record_media_hash') +``` +> 也可 `cd yunke-server && .venv/Scripts/python.exe -m alembic revision --autogenerate -m "phone_call record_media_hash"` 自动生成再核对（只应含 phone_call 加列）。 + +- [ ] **Step 3: 跑迁移 + 既有测试确认不破** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m alembic upgrade head +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_repos.py tests/test_api_calls.py -q +``` +期望：迁移 exit 0；既有测试仍绿。 + +- [ ] **Step 4: Commit** +```bash +git add yunke-server/app/db/models.py yunke-server/migrations/versions/*phone_call_record_media_hash*.py +git commit -m "$(cat <<'EOF' +feat(db): phone_call.record_media_hash 列 + 迁移（录音本地化 FK→Media） + +Co-Authored-By: Claude +EOF +)" +``` + +--- + +## Task 2: 下载器 download_call_recordings（TDD） + +> 事实：`app/scraper/media.py` 已有 `download_url(client,url)->bytes`、`_ext_from_url(url)`、`guess_mime_type(ext)`、`_EXT_MIME`（含 `.mp3→audio/mpeg`）、`logger`、`httpx`、`upsert_media` import、`ObjectStore`、`AsyncSession`、`Media`。需补 import `PhoneCall`。测试风格见 `tests/test_media_downloader.py`（respx mock OSS + tmp_path ObjectStore + in-memory SQLite）。 + +**Files:** +- Modify `yunke-server/app/scraper/media.py`（加 `download_call_recordings` + export） +- Test: `yunke-server/tests/test_recording_downloader.py`（新） + +- [ ] **Step 1: 写失败测试** `tests/test_recording_downloader.py` + +照 `test_media_downloader.py` 的 fixture（`_primed_client`、`_make_engine`、tmp_path `ObjectStore`、respx mock）。核心用例： +```python +import respx, httpx +from sqlalchemy import select +from app.db.models import Media, PhoneCall +from app.scraper.media import download_call_recordings +from app.storage.object_store import ObjectStore + +OSS_MP3 = "https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/wechat-call/8m0atd/x/ossCallback/a_b.mp3" + +async def test_download_call_recordings_localizes_and_links(monkeypatch, tmp_path): + engine, session = await _make_engine() # 复用 test_media_downloader 的 in-memory engine helper + # 种一条 phone_call：有 record_file、无 record_media_hash + session.add(PhoneCall(call_id="c1", channel="voice_call", record_file=OSS_MP3)) + await session.commit() + + store = ObjectStore(base_dir=tmp_path) + client = _primed_client() + with respx.mock: + respx.get(OSS_MP3).mock(return_value=httpx.Response(200, content=b"ID3"+b"\x00"*10)) + attempted, localized = await download_call_recordings(client, store, session) + + assert (attempted, localized) == (1, 1) + call = (await session.execute(select(PhoneCall))).scalar_one() + assert call.record_media_hash # 回写 + media = (await session.execute(select(Media))).scalar_one() + assert media.media_hash == call.record_media_hash + assert media.media_type == "audio" + assert media.mime_type == "audio/mpeg" + assert media.original_url == OSS_MP3 + assert (tmp_path / media.local_path.split(str(tmp_path))[-1].lstrip("/")).exists() or tmp_path.exists() + +async def test_idempotent_skip(monkeypatch, tmp_path): + # 已有 record_media_hash 的行不被重复处理（fetched 集合不含它） + ...种一条 record_media_hash='existing'；跑 → attempted==0 + +async def test_resilient_on_failure(monkeypatch, tmp_path): + # download_url 抛 httpx.HTTPError → 该条计入 attempted 不 localized，不抛 + ...respx.get(OSS_MP3).mock(return_value=httpx.Response(500))；attempted==1, localized==0 +``` +（`_make_engine`/`_primed_client` 从 `test_media_downloader.py` 提为共享 import，或在本文件复制。） + +- [ ] **Step 2: 跑测试确认失败** +`cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_recording_downloader.py -v` → 期望 ImportError/AttributeError（`download_call_recordings` 未定义）。 + +- [ ] **Step 3: 实现 `download_call_recordings`** + +`app/scraper/media.py`：①顶部 import 加 `PhoneCall`（`from app.db.models import Media, PhoneCall`）；②`__all__` 加 `"download_call_recordings"`；③实现： +```python +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: rows already carrying ``record_media_hash`` are + skipped; identical bytes are stored once. + """ + attempted = 0 + localized = 0 + while True: + rows = ( + await session.execute( + 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) + ) + ).all() + if not rows: + break + for call_id, record_file, channel in rows: + attempted += 1 + 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 +``` +（`sa` 已 import；`PhoneCall` 见①。） + +- [ ] **Step 4: 跑测试确认通过** +`cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_recording_downloader.py -v` → 全绿。 + +- [ ] **Step 5: Commit** +```bash +git add yunke-server/app/scraper/media.py yunke-server/tests/test_recording_downloader.py +git commit -m "$(cat <<'EOF' +feat(scraper): download_call_recordings 下载通话录音到本地 Media + 回写 record_media_hash + +复用 download_url/put_by_hash/upsert_media；幂等+resilient；media_type='audio'。 + +Co-Authored-By: Claude +EOF +)" +``` + +--- + +## Task 3: sync_call_recordings + scheduler ENTITY_FUNC（TDD） + +> 事实：`app/sync/orchestrator.py` 有 `_start_run(session, entity_type, *, channel=None, scope_id=None)`、`_finish_run(session, run, *, status, fetched, upserted, error=None)`、`SyncResult(entity,status,fetched,upserted,error)`、ENTITY 常量集（~L92-106）。`sync_wechat_messages(session, client, store, ...)` 是带 store 的 sync 先例。scheduler `_run_(session, client, store)` wrapper + `ENTITY_FUNCS` dict（~L219-233）。 + +**Files:** +- Modify `yunke-server/app/sync/orchestrator.py`（常量 + `sync_call_recordings` + import `download_call_recordings`） +- Modify `yunke-server/app/sync/scheduler.py`（`_run_call_recordings` + `ENTITY_FUNCS` + import） +- Test: `yunke-server/tests/test_orchestrator.py`（加 call_recordings 用例）或新 `tests/test_sync_call_recordings.py` + +- [ ] **Step 1: 写失败测试** + +`tests/test_sync_call_recordings.py`（或并入 test_orchestrator.py）：种一条无 media 的 phone_call + respx mock OSS → 调 `sync_call_recordings(session, client, store)` → 断言返回 `SyncResult(status='success', upserted=1)`，且 `SyncRun` 行（entity_type='call_recordings'）存在。 +```python +from app.sync.orchestrator import sync_call_recordings, ENTITY_CALL_RECORDINGS +from app.db.models import SyncRun +# ... 断言 SyncResult.entity==ENTITY_CALL_RECORDINGS, status=='success', upserted==1 +# (await session.execute(select(SyncRun).where(SyncRun.entity_type==ENTITY_CALL_RECORDINGS))).scalar_one() +``` + +- [ ] **Step 2: 跑确认失败**（`sync_call_recordings`/`ENTITY_CALL_RECORDINGS` 未定义）。 + +- [ ] **Step 3: 实现** + +`app/sync/orchestrator.py`： +① 顶部 import：`from app.scraper.media import download_call_recordings`（与既有 `from app.scraper...` 同区）。 +② ENTITY 常量区（~L92-106）加：`ENTITY_CALL_RECORDINGS = "call_recordings"`。 +③ 加函数（放在 `sync_wa_voice_calls` 之后）： +```python +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. + """ + 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)) +``` + +`app/sync/scheduler.py`： +① import：`from app.sync.orchestrator import (... sync_call_recordings, ENTITY_CALL_RECORDINGS)`（按既有 import 风格补）。 +② 加 wrapper（挨着其它 `_run_*`）： +```python +async def _run_call_recordings( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> list[SyncResult]: + return [await sync_call_recordings(session, client, store)] +``` +③ `ENTITY_FUNCS`（~L221-232）加：`"call_recordings": _run_call_recordings,`。 + +- [ ] **Step 4: 跑测试确认通过** `pytest tests/test_sync_call_recordings.py -v`（+ 既有 orchestrator/scheduler 测试不破）。 + +- [ ] **Step 5: Commit** +```bash +git add yunke-server/app/sync/orchestrator.py yunke-server/app/sync/scheduler.py yunke-server/tests/test_sync_call_recordings.py +git commit -m "$(cat <<'EOF' +feat(sync): sync_call_recordings + ENTITY_FUNCS['call_recordings'] 独立下载 pass + +Co-Authored-By: Claude +EOF +)" +``` + +--- + +## Task 4: API PhoneCallOut 字段（TDD） + +> 事实：`PhoneCallOut` 在 `app/api/schemas.py:192-220`，`from_attributes=True`，列在 L207-220。测试 `tests/test_api_calls.py`。 + +**Files:** +- Modify `yunke-server/app/api/schemas.py`（`PhoneCallOut` 加字段） +- Test: `yunke-server/tests/test_api_calls.py` + +- [ ] **Step 1: 写失败测试** 在 `test_api_calls.py` 加：种一条 `record_media_hash='abc'` 的 phone_call → GET `/api/calls` → 断言 `items[0]['record_media_hash']=='abc'`。 +- [ ] **Step 2: 跑确认失败**（字段不在响应）。 +- [ ] **Step 3: 实现** `PhoneCallOut` 内（L219 `record_file` 下方）加： +```python + # sha256 media_hash of the localized recording (NULL until downloaded). + record_media_hash: str | None = None +``` +- [ ] **Step 4: 跑确认通过** `pytest tests/test_api_calls.py -v`。 +- [ ] **Step 5: Commit** +```bash +git add yunke-server/app/api/schemas.py yunke-server/tests/test_api_calls.py +git commit -m "$(cat <<'EOF' +feat(api): PhoneCallOut 暴露 record_media_hash + +Co-Authored-By: Claude +EOF +)" +``` + +--- + +## Task 5: 前端——类型 + playRecord 本地优先 + drawer（TDD/manual） + +> 事实：`PhoneCall` 类型 `yunkes/src/api/types.ts:123-147`；`playRecord` `yunkes/src/views/CallList.vue`（~L92-106，`playRecord` 读 `cell(row,'recordFile')`/`row.record_file`）；录音按钮 v-if 在 voice/wa/phone 三 tab 的录音列（`row.record_file || cell(row,'recordFile')`）。callFields.ts `toDisplayField`：`AUDIO_KEYS={'recordFile'}` → `render='audio', value=readField(...)`，drawer 模板 `