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

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

三个 AI 工具接口（均挂载在 /api/v1/scripts 前缀下）：
- POST /extract-from-douyin  从抖音视频提取文案
  - 入口自动从分享文本中正则提取 http(s) URL，兼容 "复制链接" 粘贴场景
  - 多源轮询解析（douyin_resolver）：App Feed API → TikHub → apizero
  - 拿到 MP4 直链后优先走火山 MediaKit ASR，失败回退下载+本地 ASR
  - ASR 空结果时使用 Feed desc 兜底，图文视频直接返回 desc
  - 所有源均失败时返回具体错误信息（不暴露内部细节）
- POST /ai-rewrite           AI 文案改写（复用豆包 LLM）
- POST /ai-generate-titles   AI 标题生成（复用 generate_smart_titles）
"""

from __future__ import annotations

import logging
import os
import re
import tempfile
import time
from urllib.parse import urlparse

from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session
from app.schemas.scripts_ai import (
    AiGenerateTitlesRequest,
    AiGenerateTitlesResponse,
    AiRewriteRequest,
    AiRewriteResponse,
    ExtractFromDouyinRequest,
    ExtractFromDouyinResponse,
)
from app.services.douyin_resolver import available_providers, resolve_douyin_video
from app.services.mediakit_client import (
    MediaKitClient,
    MediaKitError,
    get_mediakit_client,
)
from app.services.script_asr_service import (
    ASRNotConfiguredError,
    ASRTranscriptionError,
    transcribe_to_text,
)
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from packages.middleware.points_gate import points_gate
from packages.shared.ai_client import get_doubao_client

logger = logging.getLogger(__name__)

router = APIRouter()

_DOUYIN_DEBUG_ERRORS = os.environ.get("DOUYIN_DEBUG_ERRORS", "").lower() in (
    "1",
    "true",
    "yes",
) or os.environ.get(
    "APP_ENV", ""
).lower() in ("staging", "dev", "development", "test")

_TAIL_PUNCT = ".,;:!?，。；：！？)）]》" + chr(34) + chr(39) + "<>"
_URL_EXTRACT_RE = re.compile(r"https?://\S+", re.IGNORECASE)
_DOUYIN_HOST_RE = re.compile(
    r"(^|\.)(douyin\.com|iesdouyin\.com|amemv\.com)$",
    re.IGNORECASE,
)
_ANY_SCHEME_RE = re.compile(r"^[a-z][a-z0-9+.-]*://\S+", re.IGNORECASE)


def _dbg(key, val):
    logger.debug("douyin_extract %s=%s", key, str(val)[:200])


def _extract_url_from_text(raw):
    if not raw:
        return None
    m = _URL_EXTRACT_RE.search(raw)
    if m:
        return m.group(0).rstrip(_TAIL_PUNCT)
    short = re.search(
        r"(?:^|(?<![a-z0-9/:]))((?:v|www)\.douyin\.com/\S+|douyin\.com/(?:video|note)/\S+)",
        raw,
        re.IGNORECASE,
    )
    if short:
        return "https://" + short.group(1).rstrip(_TAIL_PUNCT)
    return None


def _extract_and_validate_douyin_url(raw_input):
    raw = (raw_input or "").strip()
    if not raw:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="链接不能为空")

    url = _extract_url_from_text(raw)

    if not url:
        if _ANY_SCHEME_RE.search(raw):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="无效的抖音链接，仅支持 http(s) 协议",
            )
        short = re.search(
            r"(?:^|(?<![a-z0-9]))((?:v|www)\.douyin\.com/\S+|douyin\.com/(?:video|note)/\S+)",
            raw,
            re.IGNORECASE,
        )
        if short:
            url = "https://" + short.group(1).rstrip(_TAIL_PUNCT)
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="未在输入中找到有效抖音链接，请粘贴包含 v.douyin.com 或 www.douyin.com 的分享文本",
            )

    if not re.match(r"^https?://", url, re.IGNORECASE):
        url = "https://" + url

    try:
        parsed = urlparse(url)
        host = parsed.hostname or ""
        scheme = (parsed.scheme or "").lower()
    except Exception:
        host = ""
        scheme = ""
    if scheme not in ("http", "https"):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="无效的抖音链接，仅支持 http(s) 协议",
        )
    if not _DOUYIN_HOST_RE.search(host):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="无效的抖音链接，仅支持 douyin.com 域名（v.douyin.com 短链或 www.douyin.com 长链）",
        )
    return url


# ── MediaKitClient ASR 扩展（monkey patch） ────────────────────────────


def _mk_post_json(self, path, payload):
    import httpx

    if not self.is_available:
        raise MediaKitError("MediaKit API Key 未配置", code="NotConfigured")
    url = self._base_url + path
    try:
        with httpx.Client(timeout=self._timeout) as http:
            resp = http.post(url, headers=self._headers(), json=payload)
            resp.raise_for_status()
            data = resp.json()
    except httpx.TimeoutException as exc:
        raise MediaKitError("MediaKit API 超时 (%ss)" % self._timeout, code="Timeout") from exc
    except httpx.HTTPStatusError as exc:
        raise MediaKitError(
            "MediaKit API HTTP %s: %s" % (exc.response.status_code, exc.response.text[:300]),
            code="HttpError",
        ) from exc
    except httpx.RequestError as exc:
        raise MediaKitError("MediaKit API 网络错误: %s" % exc, code="NetworkError") from exc
    if data.get("success") is False and data.get("error"):
        err = data["error"] if isinstance(data["error"], dict) else {"message": str(data["error"])}
        raise MediaKitError(
            err.get("message", "请求失败"),
            code=err.get("code", "RequestFailed"),
        )
    return data


def _mk_get_json(self, path):
    import httpx

    if not self.is_available:
        raise MediaKitError("MediaKit API Key 未配置", code="NotConfigured")
    url = self._base_url + path
    try:
        with httpx.Client(timeout=self._timeout) as http:
            resp = http.get(url, headers=self._headers())
            resp.raise_for_status()
            return resp.json()
    except httpx.TimeoutException as exc:
        raise MediaKitError("MediaKit API 超时 (%ss)" % self._timeout, code="Timeout") from exc
    except httpx.HTTPStatusError as exc:
        raise MediaKitError(
            "MediaKit API HTTP %s: %s" % (exc.response.status_code, exc.response.text[:300]),
            code="HttpError",
        ) from exc
    except httpx.RequestError as exc:
        raise MediaKitError("MediaKit API 网络错误: %s" % exc, code="NetworkError") from exc


def _mediakit_asr_submit(self, video_url):
    """提交语音转字幕任务（POST /tools/asr-subtitles）。返回 task_id。"""
    data = self._post_json(
        "/tools/asr-subtitles",
        {"video_url": video_url, "language": "cmn-Hans-CN"},
    )
    task_id = data.get("task_id")
    if not task_id:
        raise MediaKitError("MediaKit ASR 提交响应缺少 task_id: %s" % str(data)[:200])
    return task_id


def _mediakit_asr_poll(self, task_id, poll_interval=2.0, max_attempts=90):
    """轮询 ASR 任务直到 completed/failed。返回 (text, duration)。"""
    for attempt in range(max_attempts):
        time.sleep(poll_interval)
        try:
            data = self._get_json("/tasks/" + task_id)
        except MediaKitError as exc:
            if attempt < max_attempts - 1 and getattr(exc, "code", "") in ("Timeout", "NetworkError"):
                logger.warning("MediaKit ASR 轮询异常（第%d次），将重试: %s", attempt + 1, exc)
                continue
            raise
        st = data.get("status")
        if st in ("completed", "success"):
            result = data.get("result") or {}
            subs = result.get("subtitles") or []
            text = "".join(s.get("subtitle_text", "") for s in subs if isinstance(s, dict))
            duration = float(result.get("duration") or 0.0)
            return text.strip(), duration
        if st == "failed":
            err = data.get("error")
            if isinstance(err, dict):
                msg = err.get("message") or "unknown"
                code = err.get("code") or "TaskFailed"
            elif isinstance(err, str):
                msg, code = err, "TaskFailed"
            else:
                msg, code = "unknown", "TaskFailed"
            raise MediaKitError("MediaKit ASR 任务失败: %s" % msg, code=code)
    raise MediaKitError(
        "MediaKit ASR 超时（%ss 未完成）" % int(poll_interval * max_attempts),
        code="Timeout",
    )


# 绑定到类（零侵入）
if not hasattr(MediaKitClient, "_post_json"):
    MediaKitClient._post_json = _mk_post_json
if not hasattr(MediaKitClient, "_get_json"):
    MediaKitClient._get_json = _mk_get_json
if not hasattr(MediaKitClient, "asr_submit"):
    MediaKitClient.asr_submit = _mediakit_asr_submit
if not hasattr(MediaKitClient, "asr_poll"):
    MediaKitClient.asr_poll = _mediakit_asr_poll


# ── 下载 + 本地 ASR 兜底 ──────────────────────────────────────────────


def _direct_url_download_and_local_asr(direct_url, page_url, temp_dir):
    """通过直链下载 MP4，再做本地 ASR。返回 (text, duration)。"""
    import os

    import httpx

    video_path = os.path.join(temp_dir, "video.mp4")
    try:
        with httpx.Client(timeout=90, follow_redirects=True, verify=False) as http:
            with http.stream(
                "GET",
                direct_url,
                headers={
                    "User-Agent": (
                        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                        "AppleWebKit/537.36 (KHTML, like Gecko) "
                        "Chrome/128.0.0.0 Safari/537.36"
                    ),
                    "Referer": "https://www.douyin.com/",
                    "Accept": "*/*",
                    "Accept-Language": "zh-CN,zh;q=0.9",
                },
            ) as resp:
                resp.raise_for_status()
                downloaded = 0
                with open(video_path, "wb") as f:
                    for chunk in resp.iter_bytes(chunk_size=65536):
                        if chunk:
                            f.write(chunk)
                            downloaded += len(chunk)
                if downloaded == 0:
                    raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="直链下载为空")
    except HTTPException:
        raise
    except httpx.TimeoutException:
        logger.warning("直链下载超时: %s", page_url)
        raise HTTPException(status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail="视频下载超时，请稍后重试") from None
    except Exception as exc:  # noqa: BLE001
        logger.exception("直链下载失败: url=%s err=%s", page_url, exc)
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="视频下载失败: " + str(exc)[:200]) from exc

    try:
        text = transcribe_to_text(video_path)
        return text.strip(), 0.0
    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 Exception as exc:
        logger.exception("直链下载后 ASR 转写异常: path=%s", video_path)
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="语音识别失败: " + str(exc)[:200],
        ) from exc


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


@router.get("/douyin/__debug_diag")
def douyin_diag():
    """[Staging/Dev only] 抖音解析源诊断。"""
    import time as _t

    import httpx as _httpx
    from app.services.douyin_resolver import APIZERO_API_KEY as _api_key_apizero
    from app.services.douyin_resolver import TIKHUB_API_KEY as _api_key_tikhub

    results = {
        "providers": available_providers(),
        "env": {
            "APP_ENV": os.environ.get("APP_ENV", ""),
            "MEDIAKIT_CONFIGURED": bool(os.environ.get("MEDIAKIT_API_KEY", "")),
        },
    }

    test_url = "https://v.douyin.com/hb-giW8cC1Q/"

    t0 = _t.time()
    try:
        r = resolve_douyin_video(test_url)
        results["resolver"] = {
            "ok": bool(r),
            "source": r.source if r else None,
            "desc_len": len(r.desc) if r else 0,
            "has_video_url": bool(r.video_url) if r else False,
            "url_domain": r.video_url.split("/")[2] if r and r.video_url and "/" in r.video_url else None,
            "time": round(_t.time() - t0, 2),
        }
    except Exception as e:
        results["resolver"] = {"ok": False, "error": str(e)[:200], "time": round(_t.time() - t0, 2)}

    if _api_key_apizero:
        t0 = _t.time()
        try:
            with _httpx.Client(timeout=8, verify=False) as c:
                r = c.get(
                    "https://v1.apizero.cn/api/video-parse",
                    params={"url": test_url, "flat": 2},
                    headers={"Authorization": f"Bearer {_api_key_apizero}"},
                )
                results["apizero"] = {"status": r.status_code, "prefix": r.text[:200], "time": round(_t.time() - t0, 2)}
        except Exception as e:
            results["apizero"] = {"error": str(e)[:200], "time": round(_t.time() - t0, 2)}

    if _api_key_tikhub:
        t0 = _t.time()
        try:
            with _httpx.Client(timeout=8, verify=False) as c:
                r = c.get(
                    "https://api.tikhub.io/api/v1/douyin/web/get_aweme_id",
                    params={"url": test_url},
                    headers={"Authorization": f"Bearer {_api_key_tikhub}"},
                )
                results["tikhub"] = {"status": r.status_code, "prefix": r.text[:200], "time": round(_t.time() - t0, 2)}
        except Exception as e:
            results["tikhub"] = {"error": str(e)[:200], "time": round(_t.time() - t0, 2)}

    return results


@router.post("/extract-from-douyin", response_model=ExtractFromDouyinResponse)
@points_gate("douyin_extract")
def extract_from_douyin(
    request: ExtractFromDouyinRequest,
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
):
    page_url = _extract_and_validate_douyin_url(request.url)
    _dbg("page_url", page_url)

    # ── Phase A：多源轮询解析 MP4 直链 ──
    last_err_stage = "parse"
    t0 = time.time()
    result = resolve_douyin_video(page_url)
    resolve_elapsed = time.time() - t0
    logger.info("抖音解析耗时: %.2fs providers=%s", resolve_elapsed, available_providers())

    direct_url = result.video_url if result else None
    feed_desc = (result.desc or "").strip() if result else ""

    # 图文视频（无 video_url 但有 desc）直接返回文案，跳过 ASR
    if result and not direct_url and feed_desc:
        logger.info("图文视频直接返回文案: source=%s desc_len=%d", result.source, len(feed_desc))
        return ExtractFromDouyinResponse(
            text=feed_desc,
            duration_seconds=0.0,
            source_url=page_url,
        )

    if not direct_url:
        if _DOUYIN_DEBUG_ERRORS:
            detail = f"抖音视频链接解析失败，请检查链接是否正确或稍后重试 [debug: providers={available_providers()}]"
        else:
            detail = "抖音视频链接解析失败，请检查链接是否正确或稍后重试"
        logger.warning("抖音解析全部失败: url=%s providers=%s", page_url, available_providers())
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=detail)

    # ── Phase B：ASR 转文字 ──
    mk_client = get_mediakit_client()
    text = ""
    duration = 0.0

    # B1：MediaKit 云端 ASR（不下载视频，最快）
    if mk_client.is_available:
        last_err_stage = "asr"
        try:
            task_id = mk_client.asr_submit(direct_url)
            text, duration = mk_client.asr_poll(task_id)
            text = text.strip()
            if text:
                logger.info(
                    "抖音 MediaKit ASR 成功: source=%s text_len=%d duration=%.1f total_time=%.1fs",
                    result.source,
                    len(text),
                    duration,
                    time.time() - t0,
                )
            else:
                logger.info("抖音 MediaKit ASR 返回空文本（无旁白/BGM视频）")
        except MediaKitError as exc:
            logger.warning("MediaKit ASR 失败，回退本地 ASR: %s", exc)
            text = ""

    # B2：回退下载 + 本地 ASR
    if not text:
        last_err_stage = "download"
        try:
            with tempfile.TemporaryDirectory(prefix="douyin_extract_") as temp_dir:
                text, dl_duration = _direct_url_download_and_local_asr(direct_url, page_url, temp_dir)
                text = (text or "").strip()
                if dl_duration and not duration:
                    duration = dl_duration
                if text:
                    logger.info(
                        "抖音本地 ASR 成功: source=%s text_len=%d total_time=%.1fs",
                        result.source,
                        len(text),
                        time.time() - t0,
                    )
                    last_err_stage = "asr"
        except HTTPException as exc:
            # 下载超时（504）是明确的网络错误，直接抛出
            if exc.status_code == status.HTTP_504_GATEWAY_TIMEOUT:
                raise
            # 本地 ASR 不可用/失败（502/503）时记录后继续走 desc 兜底，
            # 不直接抛 502，避免 API 镜像缺 worker 模块时整条链路挂掉
            logger.warning("本地 ASR 链路失败(status=%d): %s", exc.status_code, exc.detail)
            text = ""
            # 如果是下载失败（非ASR错误），保持stage为download
            if "语音识别" in str(exc.detail) or "ASR" in str(exc.detail):
                last_err_stage = "asr"
        except Exception as exc:  # noqa: BLE001
            logger.warning("本地 ASR 链路异常: %s", exc)
            text = ""

    # ── Phase C：结果判定 & 兜底 ──

    # ASR 空结果（无旁白视频）→ 使用解析源 desc 兜底
    if not text and feed_desc:
        text = feed_desc
        logger.info("抖音 ASR 空结果，使用解析源 desc 兜底: desc_len=%d", len(text))

    if not text:
        stage_msg = {
            "parse": "抖音视频链接解析失败，请检查链接是否正确或稍后重试",
            "download": "抖音视频下载失败，请检查网络或稍后重试",
            "asr": "抖音语音识别失败，请稍后重试或手动输入文案",
        }
        user_msg = stage_msg.get(last_err_stage, "抖音链接解析暂时不可用，请稍后重试或手动输入文案")
        if _DOUYIN_DEBUG_ERRORS:
            user_msg = user_msg + f" [debug: stage={last_err_stage} source={result.source}]"
        logger.warning("抖音文案提取失败: url=%s stage=%s source=%s", page_url, last_err_stage, result.source)
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=user_msg)

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


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


@router.post("/ai-rewrite", response_model=AiRewriteResponse)
@points_gate("ai_rewrite")
def ai_rewrite(
    request: AiRewriteRequest,
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
):
    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 = system_prompt + "\n风格要求：" + style
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": "请改写以下文案：\n\n" + content},
    ]
    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="AI 改写失败: " + str(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)
@points_gate("ai_title")
def ai_generate_titles(
    request: AiGenerateTitlesRequest,
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
):
    content = (request.content or "").strip()
    if not content:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="文案内容不能为空")
    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)