feat(api): 长语音 voice_clip 后端(model+migration+scraper+sync+endpoint)

- VoiceClip model (BigInteger PK voice_clip_id, structured cols + raw_json)
- Alembic migration a1b2c3d4e5f6 (chains from 8c4f1d2a9e6b)
- scrape_voice_clips(client, user_id, *, channel, page_size) per-employee scraper
  over POST /wechatweb/call/list {pageNum, pageSize, userId}; channel stamped
  locally (wechat/whatsapp), startTime/endTime ms-epoch string -> int
- upsert_voice_clip repo + start_time/end_time/mp3_duration int coercion
- sync_voice_clips orchestrator: iterates distinct user_id across
  WechatAccount + WhatsappAccount, stamps channel per account table
- scheduler ENTITY_FUNCS['voice_clips'] + added to daily 'all' job
- GET /api/voice-clips (user_id exact beats user_real_name fuzzy; channel;
  company_code; start/end window on end_time; ordered end_time desc)
- VoiceClipOut schema + router registered in main.py
- tests: scraper (5) + API (11) = 16 new; full suite 370 green

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-10 18:12:28 +08:00
parent a0db4535f1
commit d9060b6920
12 changed files with 1271 additions and 1 deletions
+138 -1
View File
@@ -49,9 +49,10 @@ from typing import Any
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Department, SyncRun
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.voice_clip import scrape_voice_clips
from app.scraper.wechat import (
scrape_departments,
scrape_wechat_accounts,
@@ -76,6 +77,7 @@ from app.storage.repos import (
upsert_phone_call,
upsert_sms,
upsert_voice_call,
upsert_voice_clip,
upsert_wechat_account,
upsert_whatsapp_account,
)
@@ -91,6 +93,7 @@ __all__ = [
"sync_departments",
"sync_phone_calls",
"sync_sms",
"sync_voice_clips",
"sync_wechat_accounts",
"sync_wechat_friends",
"sync_wechat_groups",
@@ -131,6 +134,10 @@ ENTITY_SMS = "sms"
# object store (records a SyncRun with this entity_type; no SyncState row
# because the phone_call table itself is the cursor).
ENTITY_CALL_RECORDINGS = "call_recordings"
# 长语音 (voice_clip) — per-employee scrape of ``/wechatweb/call/list``. The
# 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"
CHANNEL_WECHAT = "wechat"
CHANNEL_WHATSAPP = "whatsapp"
@@ -889,6 +896,136 @@ async def sync_sms(
)
# ---------------------------------------------------------------------------
# VoiceClip (长语音) — per-employee scrape across both staff-account tables
# ---------------------------------------------------------------------------
async def _collect_voice_clip_staff(
session: AsyncSession,
) -> list[tuple[str, str]]:
"""Return ``(user_id, channel)`` pairs for voice-clip scraping.
One pair per DISTINCT ``user_id`` across both staff-account tables. The
``call/list`` endpoint is scoped per employee (``userId`` request field)
and is shared by WeChat + WhatsApp staff (distinct ``userId`` ranges);
each employee is scraped exactly once. WeChat accounts take precedence on
collision (deterministic — in practice the two ``userId`` ranges are
disjoint, so collisions are not expected).
The returned ``channel`` is the LOCAL messaging-channel label stamped on
every scraped record: ``wechat`` for ids found in :class:`WechatAccount`,
``whatsapp`` for ids found only in :class:`WhatsappAccount`. The endpoint
response does NOT carry the channel, so it must be resolved here from the
owning account table.
"""
pairs: list[tuple[str, str]] = []
seen: set[str] = set()
wechat_rows = (
await session.execute(
sa.select(WechatAccount.user_id)
.where(WechatAccount.user_id.is_not(None))
.where(WechatAccount.user_id != "")
.distinct()
)
).all()
for (uid,) in wechat_rows:
if uid and uid not in seen:
seen.add(uid)
pairs.append((uid, CHANNEL_WECHAT))
wa_rows = (
await session.execute(
sa.select(WhatsappAccount.user_id)
.where(WhatsappAccount.user_id.is_not(None))
.where(WhatsappAccount.user_id != "")
.distinct()
)
).all()
for (uid,) in wa_rows:
if uid and uid not in seen:
seen.add(uid)
pairs.append((uid, CHANNEL_WHATSAPP))
return pairs
async def sync_voice_clips(
session: AsyncSession, client: YunkeClient
) -> SyncResult:
"""Full re-scrape of all 长语音 (voice_clip) records across all staff.
Walks every distinct ``user_id`` in the local staff-account tables
(:class:`WechatAccount` + :class:`WhatsappAccount`) and, per employee,
scrapes ``POST /wechatweb/call/list`` (body ``{pageNum, pageSize,
userId}``), upserting each record into :class:`VoiceClip`. The channel is
stamped locally (``wechat`` / ``whatsapp``) based on which account table
owns the ``user_id`` — the shared endpoint does not carry it.
Mirrors :func:`sync_wechat_voice_calls`'s failure policy + SyncRun
bookkeeping, but iterates MULTIPLE per-employee scrape streams (rather than
one department-scoped stream). A scraper failure on one employee is caught,
recorded as ``status='failed'`` for the whole entity, and a failed
:class:`SyncResult` is returned (no re-raise) — the caller decides whether
to surface / retry.
When no staff ``user_id`` is present (e.g. accounts not yet synced), the
scrape is SKIPPED with a logged warning and a ``status='skipped'``
:class:`SyncResult` (no ``SyncRun`` recorded), so a fresh DB does not abort
a larger batch.
"""
pairs = await _collect_voice_clip_staff(session)
if not pairs:
logger.warning(
"sync_voice_clips: no staff user_ids found in wechat_account /"
" whatsapp_account (sync accounts first?); skipping call/list scrape"
)
return SyncResult(
entity=ENTITY_VOICE_CLIP,
status="skipped",
fetched=0,
upserted=0,
error="no staff user_ids available",
)
run = await _start_run(session, ENTITY_VOICE_CLIP)
fetched = 0
upserted = 0
try:
for uid, channel in pairs:
async for rec in scrape_voice_clips(client, uid, channel=channel):
await upsert_voice_clip(session, rec)
fetched += 1
upserted += 1
except Exception as exc:
logger.exception("scrape_voice_clips failed")
error = repr(exc)
await _finish_run(
session, run, status="failed",
fetched=fetched, upserted=upserted, error=error,
)
return SyncResult(
entity=ENTITY_VOICE_CLIP,
status="failed",
fetched=fetched,
upserted=upserted,
error=error,
)
await set_watermark(session, ENTITY_VOICE_CLIP)
await _finish_run(
session, run, status="success", fetched=fetched, upserted=upserted
)
return SyncResult(
entity=ENTITY_VOICE_CLIP,
status="success",
fetched=fetched,
upserted=upserted,
)
# ---------------------------------------------------------------------------
# Coordinator
# ---------------------------------------------------------------------------