xiaoxia-saas/apps/api/app/api/routes/scripts_ai.py

"""Scripts AI 能力路由 — Issue #1893.

三个 AI 工具接口（均挂载在 /api/v1/scripts 前缀下）：
- POST /extract-from-douyin  从抖音视频提取文案（yt-dlp 下载 + ASR 转写）
- POST /ai-rewrite           AI 文案改写（复用豆包 LLM）
- POST /ai-generate-titles   AI 标题生成（复用 generate_smart_titles）
"""

from __future__ import annotations

import logging
import re
import tempfile

from app.auth import AuthenticatedUser, get_current_user
from app.schemas.scripts_ai import (
    AiGenerateTitlesRequest,
    AiGenerateTitlesResponse,
    AiRewriteRequest,
    AiRewriteResponse,
    ExtractFromDouyinRequest,
    ExtractFromDouyinResponse,
)
from app.services.script_asr_service import (
    ASRNotConfiguredError,
    ASRTranscriptionError,
    transcribe_to_text,
)
from fastapi import APIRouter, Depends, HTTPException, status

from packages.shared.ai_client import get_doubao_client

logger = logging.getLogger(__name__)

router = APIRouter()

# 抖音 URL 校验：支持短链 v.douyin.com 和长链 www.douyin.com/video/
_DOUYIN_URL_RE = re.compile(
    r"^(https?://)?(v\.douyin\.com/\S+|www\.douyin\.com/video/\S+)$",
    re.IGNORECASE,
)


def _validate_douyin_url(url: str) -> None:
    """校验抖音 URL 格式，不合法时抛 HTTPException(400)."""
    if not url or not url.strip():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="链接不能为空",
        )
    if not _DOUYIN_URL_RE.match(url.strip()):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="无效的抖音链接，仅支持 v.douyin.com 短链或 www.douyin.com/video/ 长链",
        )


# ── 1. 从抖音视频提取文案 ─────────────────────────────────────────────────────


@router.post(
    "/extract-from-douyin",
    response_model=ExtractFromDouyinResponse,
)
def extract_from_douyin(
    request: ExtractFromDouyinRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
) -> ExtractFromDouyinResponse:
    """从抖音视频下载无水印视频并通过 ASR 提取文案."""
    source_url = request.url.strip()
    _validate_douyin_url(source_url)

    # 确保 URL 有 scheme（yt-dlp 需要完整 URL）
    url_for_download = source_url
    if not re.match(r"^https?://", url_for_download, re.IGNORECASE):
        url_for_download = "https://" + url_for_download

    # 使用临时目录下载视频，退出时自动清理
    try:
        with tempfile.TemporaryDirectory(prefix="douyin_extract_") as temp_dir:
            import yt_dlp

            ydl_opts = {
                "format": "best[ext=mp4]/best",
                "outtmpl": f"{temp_dir}/%(id)s.%(ext)s",
                "quiet": True,
                "no_warnings": True,
                "noplaylist": True,
            }

            try:
                ydl = yt_dlp.YoutubeDL(ydl_opts)
                info = ydl.extract_info(url_for_download, download=True)
            except Exception as exc:
                logger.error("抖音视频下载失败: url=%s error=%s", source_url, exc)
                raise HTTPException(
                    status_code=status.HTTP_502_BAD_GATEWAY,
                    detail=f"视频下载失败: {exc}",
                ) from exc

            if info is None:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="无法解析该抖音链接",
                )

            video_path = ydl.prepare_filename(info)
            duration = float(info.get("duration") or 0)

            # ASR 转写
            try:
                text = transcribe_to_text(video_path)
            except ASRNotConfiguredError as exc:
                raise HTTPException(
                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                    detail=str(exc),
                ) from exc
            except ASRTranscriptionError as exc:
                raise HTTPException(
                    status_code=status.HTTP_502_BAD_GATEWAY,
                    detail=str(exc),
                ) from exc

    except HTTPException:
        raise

    return ExtractFromDouyinResponse(
        text=text,
        duration_seconds=duration,
        source_url=source_url,
    )


# ── 2. AI 文案改写 ───────────────────────────────────────────────────────────


@router.post(
    "/ai-rewrite",
    response_model=AiRewriteResponse,
)
def ai_rewrite(
    request: AiRewriteRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
) -> AiRewriteResponse:
    """使用豆包大模型改写文案."""
    content = (request.content or "").strip()
    if not content:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="文案内容不能为空",
        )

    style = request.style or "口语化"

    client = get_doubao_client()
    if not client.is_available:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="AI 服务不可用，请联系管理员配置豆包大模型 API Key",
        )

    system_prompt = (
        "你是一个专业的短视频文案改写专家。请对以下文案进行改写，"
        "要求：保留原意、口语化、适合短视频口播、调整语序避免查重。"
    )
    if style:
        system_prompt += f"\n风格要求：{style}"

    user_prompt = f"请改写以下文案：\n\n{content}"

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ]

    try:
        rewritten = client.chat_completion(
            messages=messages,
            temperature=0.8,
            max_tokens=2048,
        )
    except Exception as exc:
        logger.error("AI 改写调用失败: %s", exc)
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"AI 改写失败: {exc}",
        ) from exc

    if not rewritten:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="AI 改写未返回有效结果",
        )

    return AiRewriteResponse(
        original=content,
        rewritten=rewritten.strip(),
        style=style,
    )


# ── 3. AI 标题生成 ───────────────────────────────────────────────────────────


@router.post(
    "/ai-generate-titles",
    response_model=AiGenerateTitlesResponse,
)
def ai_generate_titles(
    request: AiGenerateTitlesRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
) -> AiGenerateTitlesResponse:
    """使用现有 generate_smart_titles 生成标题."""
    content = (request.content or "").strip()
    if not content:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="文案内容不能为空",
        )

    # count 限制在 1-5（Pydantic ge=1 le=5 已校验），但为兼容直接调用场景截断
    count = max(1, min(5, request.count))

    from app.services.ai_service import generate_smart_titles

    result = generate_smart_titles(
        description=content,
        style="viral",
        count=count,
    )

    titles = result.get("titles", [])[:count]

    return AiGenerateTitlesResponse(titles=titles)