xiaoxia-saas/apps/api/app/tasks/lipsync_tts.py

"""AI 数字人对口型 TTS 异步任务 — 将 TTS 合成从 HTTP 请求移至 Celery 后台执行.

优化目标：将 create_job 的 API 响应时间从 6~35s 降到 <1s。
任务流程：
1. 创建新 DB session，加载 job 记录
2. 调用 CosyVoice 合成音频
3. 下载音频并转存到自家 OSS
4. 更新 job 的 audio_url
5. 签名 URL 并提交到 MediaKit
6. 更新 job 状态为 submitted
7. 异常时标记 job 为 failed

注意：使用 @shared_task 而非绑定到某个 celery_app 实例，
确保任务能被 Worker 侧 celery_app 正确注册，同时 API 侧 send_task/apply_async 仍可正常调用。
"""

import io
import logging
from datetime import datetime, timezone
from urllib.parse import urlparse

from celery import shared_task

logger = logging.getLogger(__name__)

# MediaKit 预签名 URL 有效期（7天，秒），与 LipsyncService._sign_media_url 保持一致
_MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600


def _sign_media_url(url: str) -> str:
    """对自家 OSS 私有桶 URL 重签长有效期预签名.

    - 自家 OSS URL → 重签 7 天有效期
    - 外部临时 URL → 原样透传
    - 任何异常降级原样返回，不阻断主流程
    """
    if not url:
        return url
    try:
        from packages.shared.storage import get_shared_storage_service

        storage = get_shared_storage_service()
        public_base = getattr(storage, "public_url", "")
        if not isinstance(public_base, str) or not public_base:
            return url
        own_host = urlparse(public_base).netloc.lower()
        host = urlparse(url).netloc.lower()
        if not own_host or host != own_host:
            return url
        signed = storage.get_download_url(url, expires_seconds=_MEDIAKIT_URL_TTL_SECONDS)
        return signed or url
    except Exception as exc:  # noqa: BLE001
        logger.warning("[lipsync_tts] URL 重签失败，原样返回: url_prefix=%s err=%s", url[:80], exc)
        return url




def _split_script_into_sentences(script_text: str) -> list[str]:
    """按句号/问号/感叹号/分号/换行分句（与前端 splitScriptIntoSentences 一致）."""
    import re
    text = (script_text or "").strip()
    if not text:
        return []
    parts = re.split(r"[。！？!?；;\n\r]+", text)
    return [p.strip() for p in parts if p.strip()]


def _compute_sentence_timings(audio_data: bytes, script_text: str, total_duration: float) -> list[dict]:
    """基于 TTS 音频的静音检测，精确计算每句文案的起止时间.

    使用 ffmpeg silencedetect 检测静音段，将静音点与句子边界对齐。
    比字数比例估算准确得多。

    Args:
        audio_data: TTS 音频二进制数据（MP3）
        script_text: 文案全文
        total_duration: 音频总时长（秒）

    Returns:
        list[{"index": int, "text": str, "start_time": float, "end_time": float}]
    """
    import re
    import subprocess
    import tempfile

    sentences = _split_script_into_sentences(script_text)
    if not sentences:
        return []

    # 写入临时音频文件
    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp:
        tmp.write(audio_data)
        tmp_path = tmp.name

    try:
        # 用 ffmpeg silencedetect 检测静音段
        result = subprocess.run(
            [
                "ffmpeg", "-i", tmp_path,
                "-af", "silencedetect=noise=-25dB:d=0.3",
                "-f", "null", "-",
            ],
            capture_output=True,
            text=True,
            timeout=30,
        )
        stderr = result.stderr or ""

        # 解析静音结束时间点（silence_end: X.XXX）
        silence_ends = []
        for match in re.finditer(r"silence_end:\s*([\d.]+)", stderr):
            t = float(match.group(1))
            if 0 < t < total_duration:
                silence_ends.append(t)

        # 如果没有检测到足够的静音点，降级为字数比例估算
        if len(silence_ends) < len(sentences) - 1:
            logger.warning(
                "[sentence_timings] 静音点不足(%d < %d)，降级为字数比例估算",
                len(silence_ends), len(sentences) - 1,
            )
            return _estimate_sentence_timings_by_chars(sentences, total_duration)

        # 贪心匹配：N-1 个句子边界对应 N-1 个静音点
        # 按时间均匀分布期望值，选择最近的静音点
        n_boundaries = len(sentences) - 1
        boundaries = []
        used_indices = set()

        for i in range(n_boundaries):
            # 期望的边界位置（按句子数量均匀分布）
            expected_pos = (i + 1) / len(sentences) * total_duration
            # 找最近的未使用静音点
            best_idx = None
            best_dist = float("inf")
            for j, t in enumerate(silence_ends):
                if j in used_indices:
                    continue
                dist = abs(t - expected_pos)
                if dist < best_dist:
                    best_dist = dist
                    best_idx = j
            if best_idx is not None:
                used_indices.add(best_idx)
                boundaries.append(silence_ends[best_idx])

        boundaries.sort()

        # 构建 sentence_timings
        timings = []
        prev_end = 0.0
        for i, sent in enumerate(sentences):
            start = prev_end
            end = boundaries[i] if i < len(boundaries) else total_duration
            timings.append({
                "index": i,
                "text": sent,
                "start_time": round(start, 2),
                "end_time": round(end, 2),
            })
            prev_end = end

        return timings

    except Exception as exc:
        logger.warning("[sentence_timings] 静音检测异常，降级为字数比例估算: %s", exc)
        return _estimate_sentence_timings_by_chars(sentences, total_duration)
    finally:
        import os
        try:
            os.unlink(tmp_path)
        except Exception:
            pass


def _estimate_sentence_timings_by_chars(sentences: list[str], total_duration: float) -> list[dict]:
    """降级方案：按字数比例估算句子时间（与原前端逻辑一致）."""
    if not sentences or total_duration <= 0:
        return []
    total_chars = sum(len(s.replace("\s", "")) for s in sentences)
    if total_chars == 0:
        return []

    timings = []
    acc = 0
    for i, sent in enumerate(sentences):
        chars = len(sent.replace("\s", ""))
        start = (acc / total_chars) * total_duration
        end = ((acc + chars) / total_chars) * total_duration
        timings.append({
            "index": i,
            "text": sent,
            "start_time": round(start, 2),
            "end_time": round(end, 2),
        })
        acc += chars
    return timings

@shared_task(
    bind=True,
    name="lipsync_tts.synthesize_and_submit",
    max_retries=2,
    default_retry_delay=30,
)
def tts_synthesize_and_submit(
    self,
    job_id: str,
    user_id: str,
    voice_id: str,
    script_text: str,
    speed: float,
    emotion: str,
):
    """异步执行 TTS 合成 + OSS 转存 + MediaKit 提交.

    在 Celery worker 中运行，不阻塞 HTTP 请求。
    """
    from app.services.mediakit_client import MediaKitError, get_mediakit_client
    from sqlalchemy.orm import Session as DBSession

    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
    from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
    from packages.shared.url_security import safe_download_bytes

    # SessionLocal 获取：
    # - API 容器：app.db.SessionLocal（环境变量完整，导入即建引擎）
    # - Worker 容器：worker_app.db.SessionLocal（Worker 自己的 settings 初始化引擎）
    # API 侧没有 worker_app 模块 → ImportError 直接回退；
    # Worker 侧 app.db 会因缺少 API 专有环境变量抛 pydantic ValidationError，
    # 此时也要回退到 worker_app.db。
    try:
        from worker_app.db import SessionLocal  # type: ignore
    except Exception:  # noqa: BLE001
        from app.db import SessionLocal  # type: ignore

    db: DBSession = SessionLocal()
    try:
        job = (
            db.query(LipsyncJobModel)
            .filter(
                LipsyncJobModel.id == job_id,
                LipsyncJobModel.user_id == user_id,
            )
            .first()
        )

        if job is None:
            logger.error("[lipsync_tts] Job not found: job_id=%s", job_id)
            return

        # 已取消的任务不再处理
        if job.status == "cancelled":
            logger.info("[lipsync_tts] Job already cancelled, skipping: job_id=%s", job_id)
            return

        # 1. TTS 合成
        try:
            cosyvoice = CosyVoiceService()
            result = cosyvoice.submit_synthesize_task(
                text=script_text,
                voice_id=voice_id,
                speed=speed,
                emotion=emotion,
            )
        except CosyVoiceError as exc:
            logger.error("[lipsync_tts] TTS 合成失败: job_id=%s err=%s", job_id, exc)
            job.status = "failed"
            job.error_message = f"TTS 合成失败: {exc}"
            job.error_code = "TTSSynthesisFailed"
            job.updated_at = datetime.now(timezone.utc)
            db.commit()
            return
        except ValueError as exc:
            logger.error("[lipsync_tts] TTS 参数错误: job_id=%s err=%s", job_id, exc)
            job.status = "failed"
            job.error_message = f"TTS 参数错误: {exc}"
            job.error_code = "TTSInvalidParam"
            job.updated_at = datetime.now(timezone.utc)
            db.commit()
            return

        temp_url = result.get("audio_url", "")
        if not temp_url:
            logger.error("[lipsync_tts] TTS 未返回音频 URL: job_id=%s", job_id)
            job.status = "failed"
            job.error_message = "TTS 未返回音频 URL"
            job.error_code = "TTSNoAudio"
            job.updated_at = datetime.now(timezone.utc)
            db.commit()
            return

        # 2. 下载并转存到自家 OSS
        try:
            audio_data = safe_download_bytes(
                temp_url,
                purpose="lipsync_tts_audio",
                allowed_mime_types=(
                    "audio/mpeg",
                    "audio/mp3",
                    "audio/wav",
                    "audio/x-wav",  # CosyVoice 部分接口返回 audio/x-wav，与 audio/wav 等价（RIFF/WAVE）
                    "audio/mp4",
                    "audio/x-m4a",
                ),
                timeout=60.0,
            )
            from packages.shared.storage import get_shared_storage_service

            storage = get_shared_storage_service()
            storage_key = f"lipsync-tts/{user_id}/{job_id}.mp3"
            permanent_url = storage.upload_file(io.BytesIO(audio_data), storage_key, content_type="audio/mpeg")
            logger.info("[lipsync_tts] TTS 音频已转存 OSS: job_id=%s key=%s", job_id, storage_key)
            job.audio_url = permanent_url
        except Exception as exc:
            logger.warning(
                "[lipsync_tts] TTS 音频转存 OSS 失败，回退临时 URL: job_id=%s err=%s",
                job_id,
                exc,
            )
            job.audio_url = temp_url

        db.commit()

        # 2.5 计算精确句子时间戳（基于 TTS 音频静音检测）
        import os as _os
        _st_tmp_path = None
        try:
            import subprocess as _sp
            import tempfile as _tmpf

            # 下载音频用于探测时长和静音检测
            if isinstance(job.audio_url, str) and job.audio_url:
                from packages.shared.url_security import safe_download_bytes as _sdl
                _audio_bytes = _sdl(job.audio_url, purpose="sentence_timings", timeout=30.0)
            else:
                _audio_bytes = audio_data

            # ffprobe 获取音频时长
            with _tmpf.NamedTemporaryFile(suffix=".mp3", delete=False) as _atmp:
                _atmp.write(_audio_bytes)
                _st_tmp_path = _atmp.name

            _probe_result = _sp.run(
                ["ffprobe", "-v", "error", "-show_entries", "format=duration",
                 "-of", "default=noprint_wrappers=1:nokey=1", _st_tmp_path],
                capture_output=True, text=True, timeout=10,
            )
            _audio_duration = float(_probe_result.stdout.strip()) if _probe_result.stdout.strip() else 0.0

            if _audio_duration > 0:
                _timings = _compute_sentence_timings(_audio_bytes, script_text, _audio_duration)
                if _timings:
                    job.sentence_timings = _timings
                    logger.info(
                        "[lipsync_tts] 句子时间戳已计算: job_id=%s sentences=%d duration=%.1f",
                        job_id, len(_timings), _audio_duration,
                    )
            db.commit()
        except Exception as _st_err:
            logger.warning("[lipsync_tts] 句子时间戳计算失败（不影响主流程）: job_id=%s err=%s", job_id, _st_err)
        finally:
            if _st_tmp_path:
                try:
                    _os.unlink(_st_tmp_path)
                except Exception:
                    pass

        # 3. 签名 URL 并提交到 MediaKit（复用模块内 _sign_media_url，避免对 LipsyncService 的耦合）
        audio_url = _sign_media_url(job.audio_url)
        video_url = _sign_media_url(job.video_url)

        client = get_mediakit_client()
        try:
            mk_result = client.submit_lipsync(
                video_url=video_url,
                audio_url=audio_url,
                enable_video_loop=job.enable_video_loop,
                client_token=job_id,
            )
            job.mediakit_task_id = mk_result["task_id"]
            job.status = "submitted"
            job.submitted_at = datetime.now(timezone.utc)
            logger.info(
                "[lipsync_tts] 已提交 MediaKit: job_id=%s task_id=%s",
                job_id,
                mk_result["task_id"],
            )
        except MediaKitError as exc:
            job.status = "failed"
            job.error_message = str(exc)
            job.error_code = exc.code
            logger.error("[lipsync_tts] 提交 MediaKit 失败: job_id=%s err=%s", job_id, exc)

        db.commit()

    except Exception:
        logger.exception("[lipsync_tts] 未预期的异常: job_id=%s", job_id)
        try:
            job = db.query(LipsyncJobModel).filter(LipsyncJobModel.id == job_id).first()
            if job and job.status not in ("cancelled", "failed", "completed"):
                job.status = "failed"
                job.error_message = "TTS 异步任务执行异常"
                job.error_code = "AsyncTaskError"
                job.updated_at = datetime.now(timezone.utc)
                db.commit()
        except Exception:
            logger.exception("[lipsync_tts] 回写失败状态时异常: job_id=%s", job_id)
    finally:
        db.close()