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(r"\s", "")) for s in sentences)
    if total_chars == 0:
        return []

    timings = []
    acc = 0
    for i, sent in enumerate(sentences):
        chars = len(sent.replace(r"\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=5,  # 事务竞态重试3次（job not found）+ TTS偶发错误2次
    default_retry_delay=30,
    autoretry_for=(OSError, ConnectionError),  # 网络/连接错误自动重试
    retry_backoff=True,
    retry_backoff_max=30,
    soft_time_limit=180,
    time_limit=200,
)
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:
            # 事务竞态防御：API 在 commit 前投递了任务，worker 消费时事务尚未提交。
            # Celery 内置 autoretry_for 不支持"业务条件重试"，这里手动 retry 3 次，
            # 间隔递增（1s/3s/7s），让 API 事务有时间提交。
            # max_retries 由 self.request(retries) 维护；默认 self.max_retries=3 由装饰器 soft_time_limit 下方指定。
            retries = getattr(self.request, "retries", 0)
            max_retries = 3
            if retries < max_retries:
                backoff = (2 ** retries) + (retries * 1)  # 1s, 3s, 7s
                logger.warning(
                    "[lipsync_tts] Job not found yet (retry %d/%d, backoff %ds): job_id=%s",
                    retries + 1,
                    max_retries,
                    backoff,
                    job_id,
                )
                self.db.close()
                raise self.retry(countdown=backoff, max_retries=max_retries)
            logger.error(
                "[lipsync_tts] Job not found after %d retries, giving up: job_id=%s",
                max_retries,
                job_id,
            )
            return

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

        # 1. TTS 合成
        logger.info(
            "[lipsync_tts] 开始 TTS 合成: job_id=%s voice_id=%s text_len=%d speed=%.2f",
            job_id,
            voice_id,
            len(script_text),
            speed,
        )
        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. 下载 TTS 音频到内存（用于 2.5 静音检测；不转存自家 OSS，直接使用 CosyVoice 临时 URL）
        audio_data: bytes | None = None
        _st_tmp_path: str | None = None
        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,
            )
            logger.info(
                "[lipsync_tts] TTS 音频已下载到内存: job_id=%s size=%d",
                job_id,
                len(audio_data) if audio_data else 0,
            )
        except Exception as exc:
            # 下载失败：audio_data 保持 None，2.5 静音检测会跳过；后续仍用 temp_url 提交 MediaKit
            logger.warning(
                "[lipsync_tts] TTS 音频下载失败，跳过静音检测，直接使用临时 URL 提交: job_id=%s err=%s",
                job_id,
                exc,
            )
        # TTS 音频使用 CosyVoice 临时 URL，跳过自家 OSS 转存（加速，步骤⑥）
        job.audio_url = temp_url
        logger.info("[lipsync_tts] TTS 音频使用 CosyVoice 临时 URL（跳过 OSS 转存）: job_id=%s", job_id)

        db.commit()

        # 2.5 计算精确句子时间戳（基于 TTS 音频静音检测）
        # 直接复用步骤 2 已下载到内存的 audio_data，避免重新下载
        import os as _os

        try:
            import subprocess as _sp
            import tempfile as _tmpf

            if not audio_data:
                logger.warning("[lipsync_tts] 无音频数据，跳过句子时间戳计算: job_id=%s", job_id)
            else:
                # 写入临时文件供 ffprobe/ffmpeg 使用
                with _tmpf.NamedTemporaryFile(suffix=".mp3", delete=False) as _atmp:
                    _atmp.write(audio_data)
                    _st_tmp_path = _atmp.name

                # ffprobe 获取音频时长
                _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
                logger.info(
                    "[lipsync_tts] 音频时长探测: job_id=%s duration=%.2f probe_stdout=%s probe_stderr=%s",
                    job_id,
                    _audio_duration,
                    _probe_result.stdout.strip()[:50],
                    _probe_result.stderr.strip()[:100] if _probe_result.stderr else "",
                )

                if _audio_duration > 0:
                    _timings = _compute_sentence_timings(audio_data, 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,
                        )
                    else:
                        logger.warning("[lipsync_tts] 句子时间戳计算返回空结果: job_id=%s", job_id)
                else:
                    logger.warning(
                        "[lipsync_tts] ffprobe 未获取到有效时长，跳过句子时间戳: job_id=%s stdout=%s stderr=%s",
                        job_id,
                        _probe_result.stdout.strip()[:100],
                        _probe_result.stderr.strip()[:200] if _probe_result.stderr else "",
                    )
                db.commit()
        except Exception as _st_err:
            logger.warning(
                "[lipsync_tts] 句子时间戳计算失败（不影响主流程）: job_id=%s err=%s", job_id, _st_err, exc_info=True
            )
        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()


@shared_task(
    name="lipsync_tts.persist_output_video",
    max_retries=2,
    default_retry_delay=30,
)
def persist_output_video_task(job_id: str, user_id: str, temp_url: str):
    """异步转存对口型输出视频到自家 OSS（步骤⑦ — 将同步阻塞挪到后台，加速前端响应）.

    - MediaKit 返回 completed 后先以 temp_url 回前端（前端可立即播放临时 URL）
    - Celery 后台下载 temp_url 并转存 OSS，成功后更新 job.output_video_url 为永久 URL
    - 失败则保留 temp_url，不阻断主流程
    """

    try:
        from worker_app.db import SessionLocal  # type: ignore
    except Exception:  # noqa: BLE001
        from app.db import SessionLocal  # type: ignore

    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
    from packages.shared.storage import get_shared_storage_service

    db = 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.persist] Job not found: job_id=%s", job_id)
            return

        if not temp_url:
            logger.warning("[lipsync_tts.persist] temp_url 为空，跳过转存: job_id=%s", job_id)
            return

        try:
            import httpx

            with httpx.Client(timeout=180.0, follow_redirects=True) as client:
                resp = client.get(temp_url)
                resp.raise_for_status()
                data = resp.content

            storage = get_shared_storage_service()
            storage_key = f"lipsync-outputs/{user_id}/{job_id}.mp4"
            permanent_url = storage.upload_file(io.BytesIO(data), storage_key, content_type="video/mp4")
            # 对自家 OSS URL 重签 7 天有效期预签名，供前端播放
            final_url = _sign_media_url(permanent_url) if permanent_url else temp_url
            job.output_video_url = final_url
            job.updated_at = datetime.now(timezone.utc)
            db.commit()
            logger.info("[lipsync_tts.persist] 输出视频已转存 OSS: job_id=%s key=%s", job_id, storage_key)
        except Exception as exc:
            logger.warning(
                "[lipsync_tts.persist] 输出视频转存失败，保留临时 URL: job_id=%s err=%s",
                job_id,
                exc,
            )
    except Exception:
        logger.exception("[lipsync_tts.persist] 未预期异常: job_id=%s", job_id)
    finally:
        db.close()