app/api/tts.py

"""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,
        },
    )