"""TTS 朗读端点 提供两种调用方式: 1. 阻塞式:POST /v1/xtc/tts/speech —— 直接返回 MP3 二进制(适合短文本 / 非手表端) 2. 伪流式(防手表 30s 超时): - POST /v1/xtc/tts/pseudo/start —— 立即返回 session_id,后台生成 - GET /v1/xtc/tts/pseudo/poll?session_id= —— 轮询状态 - GET /v1/xtc/tts/audio/{session_id} —— 下载生成的 MP3(capability URL) 调用微软 Edge TTS(经 edge-tts 库),免费、无需 API Key。 认证、限流、请求日志均由既有中间件/依赖自动处理。 """ from __future__ import annotations import asyncio import hashlib import time import uuid from typing import Optional from fastapi import APIRouter, Depends, Request from fastapi.responses import Response from ..adapters.tts import DEFAULT_VOICE, MAX_TEXT_LEN, clean_text_for_tts, synthesize from ..auth import require_access_key from ..errors import HttpError from ._common import CORS_HEADERS, ok_with_cors router = APIRouter(prefix="/v1/xtc/tts", tags=["tts"]) async def _parse_tts_body(request: Request) -> dict: """解析 TTS 请求体。 不依赖 content-type:小天才平台可能剥离/篡改 content-type,导致 FastAPI Pydantic model 参数走非 JSON 解析路径而报 422 model_attributes_type。 这里用 request.json() 手动解析(直接 json.loads body,不检查 content-type), 与 chat 伪流式 /pseudo/start 端点保持一致。 """ try: body = await request.json() except Exception: # 兜底:少数情况下 body 是 bytes/str,尝试手动 loads raw = await request.body() if isinstance(raw, (bytes, bytearray)): raw = raw.decode("utf-8", errors="ignore") import json as _json try: body = _json.loads(raw) if raw else {} except Exception: body = {} if not isinstance(body, dict): raise HttpError("invalid body: expected JSON object", status=400, code="bad_request") return body @router.post("/speech") async def tts_speech( request: Request, _key: str = Depends(require_access_key), ) -> Response: """文本转语音,阻塞式返回 MP3 音频字节。 适合短文本或非手表端直接调用。手表端长文本请走 /pseudo/start 伪流式。 """ body = await _parse_tts_body(request) text = body.get("text") voice = body.get("voice") rate = body.get("rate") volume = body.get("volume") pitch = body.get("pitch") # 查询音频缓存:相同参数 5 分钟内已生成过则直接复用 cache_key = _tts_cache_key(text, voice, rate, volume, pitch) audio = _lookup_tts_cache(cache_key) if audio is None: audio = await synthesize( text=text, voice=voice, rate=rate, volume=volume, pitch=pitch, ) _record_tts_cache(cache_key, audio, voice or DEFAULT_VOICE) return Response( content=audio, media_type="audio/mpeg", headers={ **CORS_HEADERS, "Cache-Control": "no-store", "x-xtc-tts-voice": voice or DEFAULT_VOICE, }, ) @router.get("/voices") async def tts_voices( _key: str = Depends(require_access_key), ) -> dict: """返回常用中文音色列表,供前端选择。""" from ..adapters.tts import KNOWN_VOICES voices = [ {"id": "zh-CN-XiaoxiaoNeural", "name": "晓晓(女,常用)"}, {"id": "zh-CN-XiaoyiNeural", "name": "晓伊(女)"}, {"id": "zh-CN-YunxiNeural", "name": "云希(男)"}, {"id": "zh-CN-YunyangNeural", "name": "云扬(男)"}, {"id": "zh-CN-YunjianNeural", "name": "云健(男)"}, {"id": "zh-CN-YunxiaNeural", "name": "云夏(男)"}, {"id": "zh-CN-liaoning-XiaobeiNeural", "name": "晓贝(女,东北话)"}, {"id": "zh-CN-shaanxi-XiaoniNeural", "name": "晓妮(女,陕西话)"}, ] listed = [v for v in voices if v["id"] in KNOWN_VOICES] return {"ok": True, "voices": listed, "default": DEFAULT_VOICE} # ===== 伪流式(防手表 30s fetch 超时)===== # 流程:start 立即返回 session_id → 后台 edge-tts 生成 → poll 查状态 → audio 下载 _TTS_SESSION_TTL = 600 # 会话存活 10 分钟 # 内存存储:session_id -> {status, audio, error, created_at, voice} _tts_sessions: dict[str, dict] = {} def _cleanup_tts_sessions() -> None: """清理过期 TTS 会话,释放内存。""" now = time.time() expired = [k for k, v in _tts_sessions.items() if now - v["created_at"] > _TTS_SESSION_TTL] for k in expired: _tts_sessions.pop(k, None) # ===== TTS 音频缓存(按参数复用,5 分钟 TTL)===== # 缓存 key 由 text+voice+rate+volume+pitch 组合哈希得到, # 与前端 repo.ttsCacheKey 用相同的字段组合,保证“相同请求命中同一缓存”。 # 缓存命中时跳过 edge-tts 生成(最耗时的一步),直接复用已生成的音频字节。 # 同时受益于 /speech 阻塞端点与 /pseudo/start 伪流式端点。 _TTS_CACHE_TTL = 300 # 5 分钟,与前端 TTS_CACHE_TTL_MS 对齐 _TTS_CACHE_MAX = 64 # 最大条目数,超过则淘汰最老的(HF Space 内存有限) _tts_cache: dict[str, dict] = {} # key -> {audio, ts, voice} def _tts_cache_key(text, voice, rate, volume, pitch) -> str: """由请求参数计算缓存 key(sha256 短截)。""" h = hashlib.sha256() h.update(b"t=") h.update(str(text or "").encode("utf-8", errors="ignore")) h.update(b"|v=") h.update(str(voice or "").encode("utf-8", errors="ignore")) h.update(b"|r=") h.update(str(rate or "").encode("utf-8", errors="ignore")) h.update(b"|vol=") h.update(str(volume or "").encode("utf-8", errors="ignore")) h.update(b"|p=") h.update(str(pitch or "").encode("utf-8", errors="ignore")) return h.hexdigest()[:16] def _cleanup_tts_cache() -> None: """清理过期缓存条目,并在超限时淘汰最老的。""" now = time.time() # 1. 过期淘汰 expired = [k for k, v in _tts_cache.items() if now - v["ts"] > _TTS_CACHE_TTL] for k in expired: _tts_cache.pop(k, None) # 2. 超限淘汰最老的 if len(_tts_cache) > _TTS_CACHE_MAX: # 按 ts 升序,淘汰到上限以下 sorted_keys = sorted(_tts_cache.items(), key=lambda kv: kv[1]["ts"]) for k, _ in sorted_keys[: len(_tts_cache) - _TTS_CACHE_MAX]: _tts_cache.pop(k, None) def _lookup_tts_cache(key: str) -> Optional[bytes]: """查询缓存:命中返回 audio bytes,未命中返回 None。""" entry = _tts_cache.get(key) if not entry: return None if time.time() - entry["ts"] > _TTS_CACHE_TTL: _tts_cache.pop(key, None) return None return entry.get("audio") def _record_tts_cache(key: str, audio: bytes, voice: str) -> None: """记录缓存。失败静默跳过(缓存只是加速,不影响正确性)。""" if not audio: return _cleanup_tts_cache() _tts_cache[key] = {"audio": audio, "ts": time.time(), "voice": voice or DEFAULT_VOICE} async def _run_tts_background(session_id: str, body: dict, cache_key: str = "") -> None: """后台任务:调用 edge-tts 生成音频,结果写入内存会话,并登记到缓存。""" sess = _tts_sessions.get(session_id) if not sess: return try: audio = await synthesize( text=body.get("text"), voice=body.get("voice"), rate=body.get("rate"), volume=body.get("volume"), pitch=body.get("pitch"), ) sess["audio"] = audio sess["status"] = "done" # 生成成功后登记缓存,供后续相同参数请求复用,跳过 edge-tts 调用 if cache_key: _record_tts_cache(cache_key, audio, sess.get("voice") or DEFAULT_VOICE) except HttpError as e: sess["status"] = "error" sess["error"] = e.hint or e.message except Exception as e: sess["status"] = "error" sess["error"] = str(e) @router.post("/pseudo/start") async def tts_pseudo_start( request: Request, _key: str = Depends(require_access_key), ) -> dict: """启动伪流式 TTS 会话:立即返回 session_id,后台异步生成音频。 解决手表平台 fetch.fetch 强制 30s 超时的问题:长文本生成耗时长, 阻塞式 /speech 会被掐断报 999。伪流式立即返回,前端轮询状态, 完成后用 request.download(无 30s 限制)下载 MP3。 """ # 手动解析 body(不依赖 content-type,小天才平台可能篡改 content-type) body = await _parse_tts_body(request) text = body.get("text") voice = body.get("voice") # 参数校验(与 synthesize 一致,提前拦截避免无意义的后台任务) if not text or not str(text).strip(): raise HttpError("text is required", status=400, code="bad_request") if len(str(text)) > MAX_TEXT_LEN: raise HttpError( f"text too long: {len(str(text))} > {MAX_TEXT_LEN}", status=413, code="too_large", ) _cleanup_tts_sessions() # 查询音频缓存:相同 text+voice+rate+volume+pitch 在 5 分钟内已生成过, # 直接复用音频字节,跳过 edge-tts 生成(最耗时的一步)。 cache_key = _tts_cache_key( text, voice, body.get("rate"), body.get("volume"), body.get("pitch") ) cached_audio = _lookup_tts_cache(cache_key) if cached_audio: # 缓存命中:创建一个“已完成”的会话,前端首次 poll 即得 done session_id = uuid.uuid4().hex _tts_sessions[session_id] = { "status": "done", "audio": cached_audio, "error": None, "created_at": time.time(), "voice": voice or DEFAULT_VOICE, } return ok_with_cors({ "session_id": session_id, "poll_after_ms": 200, "cached": True, }) session_id = uuid.uuid4().hex _tts_sessions[session_id] = { "status": "running", "audio": None, "error": None, "created_at": time.time(), "voice": voice or DEFAULT_VOICE, } # 后台异步生成,不阻塞响应;生成成功后会登记到缓存 asyncio.create_task(_run_tts_background(session_id, body, cache_key)) return ok_with_cors({ "session_id": session_id, "poll_after_ms": 1000, }) @router.get("/pseudo/poll") async def tts_pseudo_poll( session_id: str, _key: str = Depends(require_access_key), ) -> dict: """轮询 TTS 会话状态。 返回 status: running / done / error。 done 时前端可从 /audio/{session_id} 下载 MP3。 """ sess = _tts_sessions.get(session_id) if not sess: raise HttpError("session not found or expired", status=404, code="not_found") return ok_with_cors({ "session_id": session_id, "status": sess["status"], "done": sess["status"] == "done", "error": sess["error"], "poll_after_ms": 1500, }) @router.get("/audio/{session_id}") async def tts_audio(session_id: str) -> Response: """下载已生成的 TTS MP3 音频。 capability URL:session_id 为 128bit 随机 hex,不可猜测,TTL 10 分钟。 因此无需 access_key 校验(手表 request.download 不便携带自定义 header)。 下载后保留会话直至 TTL 过期,允许重播。 """ sess = _tts_sessions.get(session_id) if not sess: raise HttpError("session not found or expired", status=404, code="not_found") if sess["status"] != "done" or not sess["audio"]: if sess["status"] == "error": raise HttpError( sess["error"] or "generation failed", status=502, code="upstream_error", ) raise HttpError("audio not ready yet", status=409, code="not_ready") return Response( content=sess["audio"], media_type="audio/mpeg", headers={ **CORS_HEADERS, "Cache-Control": "no-store", "x-xtc-tts-voice": sess.get("voice") or DEFAULT_VOICE, }, ) # ===== 片段式 TTS(一次会话多片段,后端并行生成)===== # 解决旧“前端切多段 + 每段各自 /pseudo/start 轮询”导致的卡死问题: # 一次会话管全部片段,后端切分 + 并行生成,前端单条轮询得知各片段就绪情况, # 按序下载/播放,下载与播放重叠。仅一条 fetch.fetch 轮询在途,规避平台并发不稳定性。 # # 流程: # POST /seg/start —— 后端切分文本为 N 段并并行生成,立即返回 session_id + total # GET /seg/poll?session_id= —— 一次性返回所有片段状态(哪些就绪可下载) # GET /seg/audio/{session_id}/{seg_idx} —— 下载某片段 MP3(capability URL,无 auth) _SEG_SESSION_TTL = 600 # 片段会话存活 10 分钟 _seg_sessions: dict[str, dict] = {} # 后端并行生成并发上限:edge-tts 是免费服务,适度并发即可,避免被限流 _SEG_GEN_CONCURRENCY = 4 _seg_gen_sem: Optional[asyncio.Semaphore] = None def _get_seg_gen_sem() -> asyncio.Semaphore: """惰性创建全局信号量(避免在模块加载时绑定事件循环,兼容旧 Python)。""" global _seg_gen_sem if _seg_gen_sem is None: _seg_gen_sem = asyncio.Semaphore(_SEG_GEN_CONCURRENCY) return _seg_gen_sem # 默认 / 最小 / 最大片段大小(字符) DEFAULT_SEG_SIZE = 500 MIN_SEG_SIZE = 200 MAX_SEG_SIZE = 1500 def _split_text_to_segments(text: str, target_size: int) -> list[str]: """按句末标点智能切分文本,目标长度 target_size 字符。 在 [target_size*0.5, target_size*1.5] 范围内找最近的句末标点切分, 保持句子完整以保听感自然;找不到则按 target_size 硬切。最后一段可能较短。 与前端 repo.splitTextToSegments 算法一致,保证前后端切分结果相同。 """ t = (text or "").replace("\r", "").strip() if not t: return [] limit = max(MIN_SEG_SIZE, min(MAX_SEG_SIZE, int(target_size or DEFAULT_SEG_SIZE))) if len(t) <= limit: return [t] out: list[str] = [] i = 0 sentence_end = set("。!?!?;;\n") while i < len(t): if len(t) - i <= limit: out.append(t[i:]) break search_start = i + limit // 2 search_end = min(len(t), i + limit + limit // 2) cut = -1 for j in range(search_end - 1, search_start - 1, -1): if t[j] in sentence_end: cut = j + 1 break if cut < 0: cut = i + limit # 找不到标点硬切 out.append(t[i:cut]) i = cut return [s for s in out if s.strip()] def _cleanup_seg_sessions() -> None: """清理过期片段会话,释放内存。""" now = time.time() expired = [k for k, v in _seg_sessions.items() if now - v["created_at"] > _SEG_SESSION_TTL] for k in expired: _seg_sessions.pop(k, None) async def _gen_one_segment( session_id: str, seg_idx: int, text: str, voice: Optional[str], rate: Optional[str], volume: Optional[str], pitch: Optional[str], ) -> None: """生成单个片段音频并写入会话状态。带全局并发信号量。 优先查音频缓存(与 /speech、/pseudo/start 共享同一缓存),命中则跳过 edge-tts。 传 clean=False:父流程已在切分前统一清洗过整段文本,避免重复清洗。 """ sess = _seg_sessions.get(session_id) if not sess: return seg = sess["segments"][seg_idx] cache_key = _tts_cache_key(text, voice, rate, volume, pitch) cached = _lookup_tts_cache(cache_key) if cached is not None: seg["status"] = "done" seg["audio"] = cached seg["cache_key"] = cache_key sess["done_count"] = int(sess.get("done_count", 0)) + 1 return seg["status"] = "running" async with _get_seg_gen_sem(): try: audio = await synthesize( text=text, voice=voice, rate=rate, volume=volume, pitch=pitch, clean=False ) seg["audio"] = audio seg["status"] = "done" seg["cache_key"] = cache_key _record_tts_cache(cache_key, audio, voice or DEFAULT_VOICE) sess["done_count"] = int(sess.get("done_count", 0)) + 1 except HttpError as e: seg["status"] = "error" seg["error"] = e.hint or e.message except Exception as e: seg["status"] = "error" seg["error"] = str(e) @router.post("/seg/start") async def tts_seg_start( request: Request, _key: str = Depends(require_access_key), ) -> dict: """启动片段式 TTS 会话:后端切分文本 + 并行生成,立即返回 session_id 与片段总数。 解决旧“前端切多段 + 每段各自 /pseudo/start 轮询”在手表上的卡死问题: 一次会话管全部片段,前端只需一条轮询即可得知各片段就绪情况。 """ body = await _parse_tts_body(request) text = body.get("text") if not text or not str(text).strip(): raise HttpError("text is required", status=400, code="bad_request") if len(str(text)) > MAX_TEXT_LEN: raise HttpError( f"text too long: {len(str(text))} > {MAX_TEXT_LEN}", status=413, code="too_large", ) voice = body.get("voice") rate = body.get("rate") volume = body.get("volume") pitch = body.get("pitch") seg_size_raw = body.get("segment_size") if seg_size_raw is None: seg_size_raw = body.get("seg_size") try: seg_size = int(seg_size_raw) if seg_size_raw is not None else DEFAULT_SEG_SIZE except (TypeError, ValueError): seg_size = DEFAULT_SEG_SIZE seg_size = max(MIN_SEG_SIZE, min(MAX_SEG_SIZE, seg_size)) # 先清洗整段文本再切分,避免切在 markdown 符号中间; # 后续 _gen_one_segment 调 synthesize 时传 clean=False 复用此结果 cleaned = clean_text_for_tts(str(text)) if not cleaned: raise HttpError( "text is empty after cleaning (no readable content)", status=400, code="bad_request", ) segments_text = _split_text_to_segments(cleaned, seg_size) if not segments_text: raise HttpError("no segments after split", status=400, code="bad_request") _cleanup_seg_sessions() session_id = uuid.uuid4().hex segments = [ {"text": s, "status": "pending", "audio": None, "error": None, "cache_key": ""} for s in segments_text ] _seg_sessions[session_id] = { "segments": segments, "done_count": 0, "created_at": time.time(), "voice": voice or DEFAULT_VOICE, "total": len(segments), } # 并行启动所有片段生成(fire-and-forget);信号量内部限并发 for idx, seg_text in enumerate(segments_text): asyncio.create_task( _gen_one_segment(session_id, idx, seg_text, voice, rate, volume, pitch) ) return ok_with_cors({ "session_id": session_id, "total": len(segments), "segment_size": seg_size, "poll_after_ms": 800, }) @router.get("/seg/poll") async def tts_seg_poll( session_id: str, _key: str = Depends(require_access_key), ) -> dict: """轮询片段式会话状态:一次性返回所有片段的就绪情况。 前端据此决定下载哪段:status=done 即可下载。单条轮询替代旧的“每段各自轮询”, 规避手表平台多条 fetch.fetch 并发的不稳定性。 """ sess = _seg_sessions.get(session_id) if not sess: raise HttpError("session not found or expired", status=404, code="not_found") segs = sess["segments"] statuses = [] ready_count = 0 error_count = 0 for s in segs: st = s["status"] ready = st == "done" if ready: ready_count += 1 if st == "error": error_count += 1 statuses.append({"status": st, "ready": ready, "error": s.get("error")}) return ok_with_cors({ "session_id": session_id, "total": len(segs), "segments": statuses, "ready_count": ready_count, "error_count": error_count, "all_done": ready_count == len(segs), "poll_after_ms": 1000, }) @router.get("/seg/audio/{session_id}/{seg_idx}") async def tts_seg_audio(session_id: str, seg_idx: int) -> Response: """下载某片段的 MP3 音频。 capability URL:session_id 为 128bit 随机 hex,不可猜测,TTL 10 分钟。 因此无需 access_key 校验(手表 request.download 不便携带自定义 header), 与 /audio/{session_id} 保持一致。 """ sess = _seg_sessions.get(session_id) if not sess: raise HttpError("session not found or expired", status=404, code="not_found") if seg_idx < 0 or seg_idx >= len(sess["segments"]): raise HttpError("invalid segment index", status=400, code="bad_request") seg = sess["segments"][seg_idx] if seg["status"] == "error": raise HttpError(seg["error"] or "generation failed", status=502, code="upstream_error") if seg["status"] != "done" or not seg["audio"]: raise HttpError("segment not ready yet", status=409, code="not_ready") return Response( content=seg["audio"], media_type="audio/mpeg", headers={ **CORS_HEADERS, "Cache-Control": "no-store", "x-xtc-tts-voice": sess.get("voice") or DEFAULT_VOICE, }, )