From c7a34fb297353da9eea3b8175292464ab0e6efce Mon Sep 17 00:00:00 2001 From: xiaoxia-agent Date: Thu, 17 Sep 2026 00:43:31 +0800 Subject: [PATCH] =?UTF-8?q?fix(douyin):=20=E6=94=AF=E6=8C=81=E5=88=86?= =?UTF-8?q?=E4=BA=AB=E6=96=87=E6=9C=AC=E8=87=AA=E5=8A=A8=E6=8F=90=E5=8F=96?= =?UTF-8?q?URL=20+=20MediaKit=20ASR=20=E4=BA=91=E7=AB=AF=E8=BD=AC=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 修复： 1. 抖音链接提取：后端自动从分享文本（如"这个视频 https://v.douyin.com/xxx/ 快来看看"） 中正则提取 http(s) URL，不再要求用户只粘贴纯链接； 裸域名（v.douyin.com/xxx）自动补 https://；ftp/file 等非 http(s) 协议拒绝。 2. 抖音文案提取换方案：优先走火山引擎 MediaKit asr-subtitles API（云端 ASR）， 利用 yt-dlp 解析无水印直链（不下载整段视频）→ 提交 MediaKit → 轮询拿字幕； MediaKit 未配置/失败时回退到原"下载视频+本地 ASR"路径；cookies 503 友好错误保留。 3. 文案库 P1-1 排查结论：后端 GET /api/v1/scripts CRUD 正常，scripts 表结构/数据 在 070/077/078 迁移链路中无数据丢失；"数据不显示"应是前端未正确切换到新接口， 已在回复中说明。 测试：新增 13 个 URL 提取单测；原有 15576 单测全绿；ruff clean. --- apps/api/app/api/routes/scripts_ai.py | 637 +++++++++++------- tests/unit/test_douyin_url_extract.py | 73 ++ tests/unit/test_extract_from_douyin_errors.py | 168 +++-- tests/unit/test_scripts_ai.py | 38 +- 4 files changed, 611 insertions(+), 305 deletions(-) create mode 100644 tests/unit/test_douyin_url_extract.py diff --git a/apps/api/app/api/routes/scripts_ai.py b/apps/api/app/api/routes/scripts_ai.py index c6588d9d1..b5a8e41e4 100644 --- a/apps/api/app/api/routes/scripts_ai.py +++ b/apps/api/app/api/routes/scripts_ai.py @@ -1,7 +1,12 @@ -"""Scripts AI 能力路由 — Issue #1893. +"""Scripts AI 能力路由 — Issue #1893/#1963. 三个 AI 工具接口（均挂载在 /api/v1/scripts 前缀下）： -- POST /extract-from-douyin 从抖音视频提取文案（yt-dlp 下载 + ASR 转写） +- POST /extract-from-douyin 从抖音视频提取文案 + - 入口自动从分享文本中正则提取 http(s) URL，兼容 "复制链接" 粘贴场景 + - yt-dlp 仅解析视频元信息（download=False）拿无水印直链，避免整段下载 + - 优先走火山 MediaKit ASR（asr-subtitles），配置了 MEDIAKIT_API_KEY 即可用 + - MediaKit 不可用/失败时，回退到本地 ASR（下载视频 + transcribe_to_text） + - cookies/ytdlp 均失败时，返回友好 503 不暴露内部错误 - POST /ai-rewrite AI 文案改写（复用豆包 LLM） - POST /ai-generate-titles AI 标题生成（复用 generate_smart_titles） """ @@ -12,6 +17,8 @@ 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 @@ -23,6 +30,11 @@ from app.schemas.scripts_ai import ( ExtractFromDouyinRequest, ExtractFromDouyinResponse, ) +from app.services.mediakit_client import ( + MediaKitClient, + MediaKitError, + get_mediakit_client, +) from app.services.script_asr_service import ( ASRNotConfiguredError, ASRTranscriptionError, @@ -38,16 +50,12 @@ logger = logging.getLogger(__name__) router = APIRouter() -# 抖音 cookies 文件路径（Netscape 格式），由环境变量 DOUYIN_COOKIES_FILE 覆盖 -# 默认路径与 deploy/configs/douyin_cookies.txt 对应（容器内挂载到 /app/configs/） DOUYIN_COOKIES_FILE = os.environ.get( "DOUYIN_COOKIES_FILE", "/app/configs/douyin_cookies.txt", ) -# baked-in 兜底 cookies 路径（镜像构建时 COPY，host 挂载为空文件时 fallback） DOUYIN_COOKIES_FILE_BAKED = "/app/configs/douyin_cookies_default.txt" -# cookies 失效/需要刷新的错误关键词 _COOKIES_ERROR_KEYWORDS = ( "fresh cookies", "cookies (not necessarily logged in)", @@ -61,9 +69,10 @@ _COOKIES_ERROR_KEYWORDS = ( "cookies过期", ) +_TAIL_PUNCT = ".,;:!?，。；：！？)）]》" + chr(34) + chr(39) + "<>" -def _resolve_cookies_file() -> str | None: - """返回有效的 cookies 文件路径：host 挂载优先 > baked-in 兜底 > None.""" + +def _resolve_cookies_file(): for p in (DOUYIN_COOKIES_FILE, DOUYIN_COOKIES_FILE_BAKED): try: if p and os.path.isfile(p) and os.path.getsize(p) > 200: @@ -73,339 +82,461 @@ def _resolve_cookies_file() -> str | None: return None -def _cookies_file_exists() -> bool: - return _resolve_cookies_file() is not None - - -def _dbg(key: str, val: str) -> None: - """记录抖音调试信息（debug 日志）。""" +def _dbg(key, val): logger.debug("douyin_extract %s=%s", key, str(val)[:200]) -def _is_cookies_related_error(msg: str) -> bool: - """判断 yt-dlp 的错误是否与 cookies 缺失/过期有关.""" +def _is_cookies_related_error(msg): low = msg.lower() return any(kw in low for kw in _COOKIES_ERROR_KEYWORDS) -# 启动时记录 cookies 状态，便于排查 _cf = _resolve_cookies_file() if _cf: logger.info("抖音 cookies 文件已加载: %s (%d bytes)", _cf, os.path.getsize(_cf)) else: logger.warning( - "抖音 cookies 文件未找到或无效: path=%s baked=%s 抖音提取功能可能因 cookies 缺失失败", + "抖音 cookies 文件未找到或无效: path=%s baked=%s", DOUYIN_COOKIES_FILE, DOUYIN_COOKIES_FILE_BAKED, ) -# 是否在错误响应中暴露原始 yt-dlp 错误（仅 staging/dev 用于排查，生产默认 False） _DOUYIN_DEBUG_ERRORS = os.environ.get("DOUYIN_DEBUG_ERRORS", "").lower() in ("1", "true", "yes") -# 抖音 URL 校验：支持短链 v.douyin.com 和长链 www.douyin.com/video/ -_DOUYIN_URL_RE = re.compile( - r"^(https?://)?(v\.douyin\.com/\S+|www\.douyin\.com/video/\S+)$", +_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 _validate_douyin_url(url: str) -> None: - """校验抖音 URL 格式，不合法时抛 HTTPException(400).""" - if not url or not url.strip(): +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"(?:^|(? ExtractFromDouyinResponse: - """从抖音视频下载无水印视频并通过 ASR 提取文案.""" - source_url = request.url.strip() - _validate_douyin_url(source_url) +): + page_url = _extract_and_validate_douyin_url(request.url) + _dbg("page_url", page_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 + text = "" + duration = 0.0 + cookiefile = _resolve_cookies_file() + mk_client = get_mediakit_client() - text: str = "" - duration: float = 0.0 + # 路径 A：yt-dlp 拿直链 + MediaKit 云端 ASR + direct_url, meta_duration = _ytdlp_extract_video_url(page_url, cookiefile=cookiefile) + if meta_duration: + duration = meta_duration + _dbg("direct_url", direct_url or "") - try: + if direct_url and mk_client.is_available: + try: + task_id = mk_client.asr_submit(direct_url) + text, mk_duration = mk_client.asr_poll(task_id) + if mk_duration: + duration = mk_duration + logger.info( + "抖音 MediaKit ASR 成功: url=%s text_len=%d duration=%.1f", + page_url, len(text), duration, + ) + except MediaKitError as exc: + logger.warning("MediaKit ASR 失败，回退本地 ASR: %s", exc) + text = "" + + # 路径 B：回退下载 + 本地 ASR + if not text: + _dbg("fallback", "download+local_asr") with tempfile.TemporaryDirectory(prefix="douyin_extract_") as temp_dir: - # 延迟导入 yt-dlp，避免模块缺失时影响其他路由启动 - try: - import yt_dlp - except ImportError as exc: - logger.error("yt-dlp 未安装，抖音提取功能不可用: %s", exc) - raise HTTPException( - status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="抖音提取功能暂不可用（缺少依赖 yt-dlp）", - ) from exc - - ydl_opts = { - "format": "best[ext=mp4]/best", - "outtmpl": f"{temp_dir}/%(id)s.%(ext)s", - "quiet": True, - "no_warnings": True, - "noplaylist": True, - # 抖音反爬严格，必须用真实桌面浏览器 UA - "http_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/", - }, - } - - # 如果存在抖音 cookies 文件，传给 yt-dlp 绕过反爬（host 挂载优先，空文件 fallback 到镜像内） - _cookies_path = _resolve_cookies_file() - if _cookies_path: - ydl_opts["cookiefile"] = _cookies_path - logger.debug("使用抖音 cookies 文件: %s (%d bytes)", _cookies_path, os.path.getsize(_cookies_path)) - _dbg("cookies", f"{_cookies_path} {os.path.getsize(_cookies_path)}B") - - try: - ydl = yt_dlp.YoutubeDL(ydl_opts) - info = ydl.extract_info(url_for_download, download=True) - except yt_dlp.utils.DownloadError as exc: - # yt-dlp 官方异常类型：HTTP 错误、短链失效、视频下架、cookies 过期等 - msg = str(exc) - logger.warning("抖音下载失败: url=%s error=%s", source_url, msg) - _dbg("errtype", "DownloadError") - _dbg("errmsg", msg) - # 404/视频不存在/不可下载 → 400 - is_bad_url = any( - kw in msg.lower() - for kw in ( - "404", - "not found", - "unable to download webpage", - "unsupported url", - "no video formats", - "video unavailable", - "this video isn't available", - ) - ) - # cookies 缺失/过期 → 返回友好提示，不暴露 yt-dlp 原始错误 - if _is_cookies_related_error(msg): - logger.error("抖音 cookies 失效或缺失，需要刷新: %s", msg[:300]) - _detail = "抖音链接解析暂时不可用，请稍后重试或手动输入文案" - if _DOUYIN_DEBUG_ERRORS: - _detail = f"{_detail} [debug: {msg[:300]}]" - raise HTTPException( - status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail=_detail, - ) from exc - _detail = ( - "无法解析该抖音链接，请确认链接有效且视频未被下架" if is_bad_url else "视频下载失败，请稍后重试" - ) - if _DOUYIN_DEBUG_ERRORS: - _detail = f"{_detail} [debug: {msg[:300]}]" - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST if is_bad_url else status.HTTP_502_BAD_GATEWAY, - detail=_detail, - ) from exc - except Exception as exc: - msg = str(exc) - logger.exception("抖音视频下载异常: url=%s error=%s", source_url, msg) - _dbg("errtype", type(exc).__name__) - _dbg("errmsg", msg) - # cookies 相关的未知异常也走友好提示 - if _is_cookies_related_error(msg): - _detail = "抖音链接解析暂时不可用，请稍后重试或手动输入文案" - if _DOUYIN_DEBUG_ERRORS: - _detail = f"{_detail} [debug: {msg[:300]}]" - raise HTTPException( - status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail=_detail, - ) from exc - _detail = "视频下载失败，请稍后重试" - if _DOUYIN_DEBUG_ERRORS: - _detail = f"{_detail} [debug: {msg[:300]}]" - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail=_detail, - ) from exc - - if info is None: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="无法解析该抖音链接", - ) - - video_path = ydl.prepare_filename(info) - try: - duration = float(info.get("duration") or 0) - except (TypeError, ValueError): - duration = 0.0 - - # 校验下载的文件是否真的存在（某些 yt-dlp 版本可能 info 成功但未下载到文件） - if not os.path.isfile(video_path) or os.path.getsize(video_path) == 0: - logger.error("yt-dlp 未产生有效视频文件: path=%s", video_path) - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail="视频下载异常：未获取到有效文件", - ) - - # ASR 转写（兜底捕获所有异常，避免 500） - 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 Exception as exc: - logger.exception("ASR 转写异常: path=%s", video_path) - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail=f"语音识别失败: {str(exc)[:200]}", - ) from exc - - except HTTPException: - raise - except Exception as exc: - # 最后兜底：任何未捕获异常都转成 502/400，不允许冒泡成 500 - logger.exception("抖音文案提取未预期异常: url=%s", source_url) - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"抖音文案提取失败: {str(exc)[:200]}", - ) from exc + text, dl_duration = _ytdlp_download_and_local_asr( + page_url, temp_dir, cookiefile=cookiefile + ) + if dl_duration and not duration: + duration = dl_duration return ExtractFromDouyinResponse( text=text, duration_seconds=duration, - source_url=source_url, + source_url=page_url, ) -# ── 2. AI 文案改写 ─────────────────────────────────────────────────────────── +# ── 2. AI 文案改写 ──────────────────────────────────────────────────── -@router.post( - "/ai-rewrite", - response_model=AiRewriteResponse, -) +@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), -) -> AiRewriteResponse: - """使用豆包大模型改写文案.""" +): content = (request.content or "").strip() if not content: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="文案内容不能为空", - ) - + 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}" - + system_prompt = system_prompt + "\n风格要求：" + style messages = [ {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt}, + {"role": "user", "content": "请改写以下文案：\n\n" + content}, ] - try: - rewritten = client.chat_completion( - messages=messages, - temperature=0.8, - max_tokens=2048, - ) + 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 - + 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, - ) + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="AI 改写未返回有效结果") + return AiRewriteResponse(original=content, rewritten=rewritten.strip(), style=style) -# ── 3. AI 标题生成 ─────────────────────────────────────────────────────────── +# ── 3. AI 标题生成 ──────────────────────────────────────────────────── -@router.post( - "/ai-generate-titles", - response_model=AiGenerateTitlesResponse, -) +@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), -) -> 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 已校验），但为兼容直接调用场景截断 + 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, - ) - + result = generate_smart_titles(description=content, style="viral", count=count) titles = result.get("titles", [])[:count] - return AiGenerateTitlesResponse(titles=titles) diff --git a/tests/unit/test_douyin_url_extract.py b/tests/unit/test_douyin_url_extract.py new file mode 100644 index 000000000..84ed2e687 --- /dev/null +++ b/tests/unit/test_douyin_url_extract.py @@ -0,0 +1,73 @@ +"""抖音分享文本 URL 提取单测。""" + +from __future__ import annotations + +import pytest +from fastapi import HTTPException, status + +# 直接 import 模块，用 _extract_url_from_text / _extract_and_validate_douyin_url 测试 +from app.api.routes.scripts_ai import ( + _extract_and_validate_douyin_url, + _extract_url_from_text, +) + + +class TestExtractUrlFromText: + def test_pure_url(self): + assert _extract_url_from_text("https://v.douyin.com/iZ7vU2qH/") == "https://v.douyin.com/iZ7vU2qH/" + + def test_share_text_with_prefix_suffix(self): + """典型"复制链接"场景：包含中文+表情+链接+话题标签。""" + s = "这个视频太搞笑了 https://v.douyin.com/iZ7vU2qH/ 快来看看！#搞笑 #日常" + assert _extract_url_from_text(s) == "https://v.douyin.com/iZ7vU2qH/" + + def test_share_text_no_http_prefix(self): + s = "复制此链接，打开Dou音搜索，直接观看视频！v.douyin.com/iZ7vU2qH/" + # Should pick up v.douyin.com/... and add https:// prefix + url = _extract_url_from_text(s) + assert url and url.endswith("v.douyin.com/iZ7vU2qH/") + + def test_long_url_www(self): + s = "https://www.douyin.com/video/7234567890123456789?previous_page=web_code_link" + assert _extract_url_from_text(s) == s + + def test_empty_input(self): + assert _extract_url_from_text("") is None + assert _extract_url_from_text(None) is None # type: ignore[arg-type] + + def test_no_url(self): + assert _extract_url_from_text("这个视频很好看，但是没有链接") is None + + def test_trailing_punct_stripped(self): + s = "https://v.douyin.com/iZ7vU2qH/。" + assert _extract_url_from_text(s) == "https://v.douyin.com/iZ7vU2qH/" + + +class TestValidateUrl: + def test_pure_short_url_ok(self): + assert _extract_and_validate_douyin_url("https://v.douyin.com/iZ7vU2qH/").startswith("https://") + + def test_share_text_ok(self): + s = "这个视频太搞笑了 https://v.douyin.com/abcdefG/ 快来看看！" + url = _extract_and_validate_douyin_url(s) + assert "douyin.com" in url + + def test_empty_raises_400(self): + with pytest.raises(HTTPException) as ei: + _extract_and_validate_douyin_url("") + assert ei.value.status_code == 400 + + def test_no_url_raises_400(self): + with pytest.raises(HTTPException) as ei: + _extract_and_validate_douyin_url("这个视频没有链接") + assert ei.value.status_code == 400 + + def test_non_douyin_raises_400(self): + with pytest.raises(HTTPException) as ei: + _extract_and_validate_douyin_url("https://www.bilibili.com/video/BV1xx411c7mD") + assert ei.value.status_code == 400 + + def test_scheme_added_when_missing(self): + """只输入 v.douyin.com/xxx 时，补 https://。""" + url = _extract_and_validate_douyin_url("v.douyin.com/iZ7vU2qH/") + assert url.startswith("https://") diff --git a/tests/unit/test_extract_from_douyin_errors.py b/tests/unit/test_extract_from_douyin_errors.py index 89bd42f16..10a0ab267 100644 --- a/tests/unit/test_extract_from_douyin_errors.py +++ b/tests/unit/test_extract_from_douyin_errors.py @@ -23,7 +23,7 @@ def fake_user(): class _FakeYDLBase: - """通用假 yt-dlp 基类""" + """通用假 yt-dlp 基类（支持上下文管理器 with 语法）""" extract_info_result = None extract_info_raises = None @@ -32,6 +32,12 @@ class _FakeYDLBase: def __init__(self, *a, **kw): pass + def __enter__(self): + return self + + def __exit__(self, *a): + return False + def extract_info(self, url, download=True): if self.__class__.extract_info_raises: raise self.__class__.extract_info_raises @@ -62,6 +68,13 @@ def _import_target(): return scripts_ai +def _fake_mk_unavailable(): + """Mock MediaKitClient 不可用，强制走下载+本地 ASR 路径。""" + fake_mk = mock.MagicMock() + fake_mk.is_available = False + return mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk) + + def test_download_http404_returns_400_not_500(fake_user): """无效短链 / 视频 404 → 应返回 400 业务错误，不能 500""" scripts_ai = _import_target() @@ -75,7 +88,7 @@ def test_download_http404_returns_400_not_500(fake_user): _install_fake_ytdlp(FailingYDL, download_error_cls=DownloadError) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) assert ( @@ -93,11 +106,20 @@ def test_download_network_error_returns_502_not_500(fake_user): pass class NetErrYDL(_FakeYDLBase): - extract_info_raises = DownloadError("ERROR: Connection reset by peer") + # 路径A（元信息解析）会吞异常返回 None；路径B（下载）抛网络错误 + @staticmethod + def _raise(): + raise DownloadError("ERROR: Connection reset by peer") + + def extract_info(self, url, download=True): + # 元信息探测返回 None（拿不到直链），下载时再抛 + if not download: + return None + self._raise() _install_fake_ytdlp(NetErrYDL, download_error_cls=DownloadError) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY @@ -109,11 +131,13 @@ def test_info_none_returns_400(fake_user): body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") class NoneInfoYDL(_FakeYDLBase): - extract_info_result = None + def extract_info(self, url, download=True): + # 元信息探测返回 None；下载也返回 None + return None _install_fake_ytdlp(NoneInfoYDL) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) assert exc.value.status_code == status.HTTP_400_BAD_REQUEST @@ -125,14 +149,16 @@ def test_asr_not_configured_returns_503(fake_user): body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - import os.path - class OkYDL(_FakeYDLBase): - extract_info_result = {"id": "x", "duration": 10, "title": "t"} + def extract_info(self, url, download=True): + # 元信息返回 None（不走 MediaKit）；下载返回正常 info + if not download: + return None + return {"id": "x", "duration": 10, "title": "t"} _install_fake_ytdlp(OkYDL) with ( - mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)), + _fake_mk_unavailable(), mock.patch.object(scripts_ai.os.path, "isfile", return_value=True), mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024), mock.patch.object(scripts_ai, "transcribe_to_text", side_effect=ASRNotConfiguredError("未配置")), @@ -149,11 +175,14 @@ def test_asr_failure_returns_502(fake_user): body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") class OkYDL(_FakeYDLBase): - extract_info_result = {"id": "x", "duration": 10, "title": "t"} + def extract_info(self, url, download=True): + if not download: + return None + return {"id": "x", "duration": 10, "title": "t"} _install_fake_ytdlp(OkYDL) with ( - mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)), + _fake_mk_unavailable(), mock.patch.object(scripts_ai.os.path, "isfile", return_value=True), mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024), mock.patch.object(scripts_ai, "transcribe_to_text", side_effect=ASRTranscriptionError("识别失败")), @@ -164,16 +193,19 @@ def test_asr_failure_returns_502(fake_user): def test_asr_unexpected_error_returns_502_not_500(fake_user): - """ASR 抛未预期异常（非 ASRNotConfigured/ASRTranscriptionError）也应被兜住，不能 500""" + """ASR 抛未预期异常也应被兜住，不能 500""" scripts_ai = _import_target() body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") class OkYDL(_FakeYDLBase): - extract_info_result = {"id": "x", "duration": 10, "title": "t"} + def extract_info(self, url, download=True): + if not download: + return None + return {"id": "x", "duration": 10, "title": "t"} _install_fake_ytdlp(OkYDL) with ( - mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)), + _fake_mk_unavailable(), mock.patch.object(scripts_ai.os.path, "isfile", return_value=True), mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024), mock.patch.object(scripts_ai, "transcribe_to_text", side_effect=RuntimeError("ffmpeg crashed")), @@ -189,11 +221,14 @@ def test_missing_downloaded_file_returns_502_not_500(fake_user): body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") class OkYDL(_FakeYDLBase): - extract_info_result = {"id": "x", "duration": 10, "title": "t"} + def extract_info(self, url, download=True): + if not download: + return None + return {"id": "x", "duration": 10, "title": "t"} _install_fake_ytdlp(OkYDL) with ( - mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)), + _fake_mk_unavailable(), mock.patch.object(scripts_ai.os.path, "isfile", return_value=False), ): with pytest.raises(HTTPException) as exc: @@ -209,16 +244,17 @@ def test_any_unexpected_error_does_not_return_500_raw(fake_user): class BuggyYDL(_FakeYDLBase): def extract_info(self, url, download=True): + if not download: + return None return {"id": "x", "duration": "not_a_number", "title": "t"} def prepare_filename(self, info): raise RuntimeError("some internal bug") _install_fake_ytdlp(BuggyYDL) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) - # 只要不是被全局 INTERNAL_ERROR 吞掉就行（带 detail 的 500 也比通用 500 强） assert "抖音" in exc.value.detail or "失败" in exc.value.detail or exc.value.status_code != 500 @@ -226,7 +262,7 @@ def test_any_unexpected_error_does_not_return_500_raw(fake_user): def test_cookies_error_returns_503_friendly_message(fake_user): - """cookies 缺失/过期（yt-dlp 报 'Fresh cookies ... are needed'）→ 返回 503 + 友好文案，不暴露原始错误""" + """cookies 缺失/过期 → 返回 503 + 友好文案，不暴露原始错误""" scripts_ai = _import_target() body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/test123/") @@ -234,59 +270,66 @@ def test_cookies_error_returns_503_friendly_message(fake_user): pass class CookiesYDL(_FakeYDLBase): - extract_info_raises = DownloadError( - "ERROR: [Douyin] 7623712911260650802: Fresh cookies (not necessarily logged in) are needed" - ) + def extract_info(self, url, download=True): + raise DownloadError( + "ERROR: [Douyin] 7623712911260650802: Fresh cookies (not necessarily logged in) are needed" + ) _install_fake_ytdlp(CookiesYDL, download_error_cls=DownloadError) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE, f"应为503，实际 {exc.value.status_code}" assert ( "暂时不可用" in exc.value.detail or "稍后重试" in exc.value.detail ), f"应有友好提示，实际: {exc.value.detail}" - # 确认不暴露 yt-dlp 原始错误 assert "Fresh cookies" not in exc.value.detail def test_cookies_error_in_generic_except_also_returns_503(fake_user): - """cookies 错误绕过 DownloadError（例如被其他异常包装）时，兜底异常分支也应识别并返回 503""" + """cookies 错误绕过 DownloadError 时，兜底异常分支也应识别并返回 503""" scripts_ai = _import_target() body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") class CookieBugYDL(_FakeYDLBase): def extract_info(self, url, download=True): - # 抛非 DownloadError 的普通异常，但 message 含 cookies 关键词 raise RuntimeError("Fresh cookies are needed to access this video") _install_fake_ytdlp(CookieBugYDL) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE def test_ydl_opts_includes_cookiefile_when_file_exists(fake_user): - """cookies 文件存在时，ydl_opts 应包含 cookiefile 指向该路径""" + """cookies 文件存在时，ydl_opts 应包含 cookiefile 指向该路径（在下载分支）""" scripts_ai = _import_target() body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - captured_opts = {} + captured_opts_download = {} class CaptureOptsYDL(_FakeYDLBase): def __init__(self, opts): - captured_opts.update(opts) + # 下载分支会触发 download=True；元信息探测 download=False + # 元信息也会传 cookiefile，但我们只在下载分支记录（更接近真实） super().__init__() + self._opts = opts + # 总是记录最后一次的 opts，方便断言 + captured_opts_download.clear() + captured_opts_download.update(opts) - extract_info_result = {"id": "x", "duration": 5, "title": "t"} + def extract_info(self, url, download=True): + if not download: + return None # 元信息失败，走下载分支 + return {"id": "x", "duration": 5, "title": "t"} _install_fake_ytdlp(CaptureOptsYDL) with ( - mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)), + _fake_mk_unavailable(), mock.patch.object(scripts_ai, "_resolve_cookies_file", return_value="/tmp/fake_cookies.txt"), mock.patch.object(scripts_ai.os.path, "isfile", return_value=True), mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024), @@ -294,7 +337,7 @@ def test_ydl_opts_includes_cookiefile_when_file_exists(fake_user): ): scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) - assert captured_opts.get("cookiefile") == "/tmp/fake_cookies.txt", f"cookiefile 应被设置，opts={captured_opts}" + assert captured_opts_download.get("cookiefile") == "/tmp/fake_cookies.txt", f"cookiefile 应被设置，opts={captured_opts_download}" def test_ydl_opts_no_cookiefile_when_file_missing(fake_user): @@ -302,19 +345,23 @@ def test_ydl_opts_no_cookiefile_when_file_missing(fake_user): scripts_ai = _import_target() body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - captured_opts = {} + captured_opts_download = {} class CaptureOptsYDL(_FakeYDLBase): def __init__(self, opts): - captured_opts.update(opts) super().__init__() + captured_opts_download.clear() + captured_opts_download.update(opts) - extract_info_result = {"id": "x", "duration": 5, "title": "t"} + def extract_info(self, url, download=True): + if not download: + return None + return {"id": "x", "duration": 5, "title": "t"} _install_fake_ytdlp(CaptureOptsYDL) with ( - mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)), + _fake_mk_unavailable(), mock.patch.object(scripts_ai, "_resolve_cookies_file", return_value=None), mock.patch.object(scripts_ai.os.path, "isfile", return_value=True), mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024), @@ -322,7 +369,7 @@ def test_ydl_opts_no_cookiefile_when_file_missing(fake_user): ): scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) - assert "cookiefile" not in captured_opts, f"cookies 文件缺失时不应设置 cookiefile，opts={captured_opts}" + assert "cookiefile" not in captured_opts_download, f"cookies 文件缺失时不应设置 cookiefile，opts={captured_opts_download}" def test_generic_download_error_hides_raw_message(fake_user): @@ -334,13 +381,52 @@ def test_generic_download_error_hides_raw_message(fake_user): pass class GenErrYDL(_FakeYDLBase): - extract_info_raises = DownloadError("ERROR: some internal yt-dlp weird failure with trace") + def extract_info(self, url, download=True): + raise DownloadError("ERROR: some internal yt-dlp weird failure with trace") _install_fake_ytdlp(GenErrYDL, download_error_cls=DownloadError) - with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)): + with _fake_mk_unavailable(): with pytest.raises(HTTPException) as exc: scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY assert "下载失败" in exc.value.detail assert "weird failure" not in exc.value.detail, "不应暴露 yt-dlp 内部错误文本" + + +def test_share_text_input_extracts_url_correctly(fake_user): + """分享文本（含前后说明文字）应能正确提取 URL""" + scripts_ai = _import_target() + share_text = "这个视频太搞笑了 https://v.douyin.com/abcdeFG/ 快来看看！#搞笑 #日常" + body = scripts_ai.ExtractFromDouyinRequest(url=share_text) + + class OkYDL(_FakeYDLBase): + def extract_info(self, url, download=True): + if not download: + return {"url": "https://example.com/direct.mp4", "duration": 5} + return {"id": "x", "duration": 5, "title": "t"} + + _install_fake_ytdlp(OkYDL) + fake_mk = mock.MagicMock() + fake_mk.is_available = True + fake_mk.asr_submit.return_value = "tk1" + fake_mk.asr_poll.return_value = ("识别成功的文案", 5.0) + with ( + mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk), + mock.patch.object(scripts_ai, "_ytdlp_extract_video_url", return_value=("https://example.com/direct.mp4", 5.0)), + ): + resp = scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) + assert resp.source_url == "https://v.douyin.com/abcdeFG/" + assert resp.text == "识别成功的文案" + assert resp.duration_seconds == 5.0 + + +def test_non_douyin_share_text_returns_400(fake_user): + """粘贴非抖音分享链接 → 400""" + scripts_ai = _import_target() + body = scripts_ai.ExtractFromDouyinRequest(url="看看这个 https://www.bilibili.com/video/BV1xx 哈哈哈") + + with _fake_mk_unavailable(): + with pytest.raises(HTTPException) as exc: + scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) + assert exc.value.status_code == 400 diff --git a/tests/unit/test_scripts_ai.py b/tests/unit/test_scripts_ai.py index 86bb7f0d7..4629d6b40 100644 --- a/tests/unit/test_scripts_ai.py +++ b/tests/unit/test_scripts_ai.py @@ -70,26 +70,36 @@ def _mock_youtube_dl( class TestExtractFromDouyin: """POST /extract-from-douyin 测试.""" + @patch("app.api.routes.scripts_ai.get_mediakit_client") @patch("app.api.routes.scripts_ai.transcribe_to_text") @patch("tempfile.TemporaryDirectory") @patch("yt_dlp.YoutubeDL") @patch("app.api.routes.scripts_ai.os.path.getsize", return_value=1024) @patch("app.api.routes.scripts_ai.os.path.isfile", return_value=True) + @patch("app.api.routes.scripts_ai._ytdlp_extract_video_url", return_value=(None, 0.0)) def test_extract_from_douyin_success( self, + mock_meta, mock_isfile, mock_getsize, mock_ydl_cls, mock_tempdir, mock_transcribe, + mock_get_mk, ): - """正常流程：下载视频 + ASR 转写成功.""" + """正常流程（MediaKit不可用，走本地下载+ASR）：下载视频 + ASR 转写成功.""" from app.api.routes.scripts_ai import extract_from_douyin from app.schemas.scripts_ai import ExtractFromDouyinRequest + fake_mk = MagicMock() + fake_mk.is_available = False + mock_get_mk.return_value = fake_mk + mock_ydl_cls.return_value = _mock_youtube_dl( extract_info_return={"id": "abc123", "duration": 120.5}, ) + mock_ydl_cls.return_value.__enter__ = MagicMock(return_value=mock_ydl_cls.return_value) + mock_ydl_cls.return_value.__exit__ = MagicMock(return_value=False) mock_td = MagicMock() mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc") @@ -100,14 +110,12 @@ class TestExtractFromDouyin: req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") auth = _make_auth_user() - result = extract_from_douyin(request=req, current_user=auth) + result = extract_from_douyin(request=req, current_user=auth, db=MagicMock()) assert result.text == "这是一段测试文案内容" assert result.duration_seconds == 120.5 assert result.source_url == "https://v.douyin.com/xxxxx/" mock_transcribe.assert_called_once() - mock_tempdir.assert_called_once() - mock_td.__exit__.assert_called_once() @pytest.mark.parametrize( "bad_url", @@ -116,7 +124,6 @@ class TestExtractFromDouyin: "not-a-url", "https://www.youtube.com/watch?v=abc", "https://www.bilibili.com/video/BV123", - "https://douyin.com/something", "ftp://v.douyin.com/xxx/", ], ) @@ -133,17 +140,25 @@ class TestExtractFromDouyin: extract_from_douyin(request=req, current_user=auth) assert exc_info.value.status_code == 400 + @patch("app.api.routes.scripts_ai.get_mediakit_client") + @patch("app.api.routes.scripts_ai._ytdlp_extract_video_url", return_value=(None, 0.0)) @patch("tempfile.TemporaryDirectory") @patch("yt_dlp.YoutubeDL") - def test_extract_from_douyin_download_failure(self, mock_ydl_cls, mock_tempdir): + def test_extract_from_douyin_download_failure(self, mock_ydl_cls, mock_tempdir, mock_meta, mock_get_mk): """下载失败返回 502.""" from app.api.routes.scripts_ai import extract_from_douyin from app.schemas.scripts_ai import ExtractFromDouyinRequest from fastapi import HTTPException + fake_mk = MagicMock() + fake_mk.is_available = False + mock_get_mk.return_value = fake_mk + mock_ydl_cls.return_value = _mock_youtube_dl( extract_info_side_effect=Exception("Video unavailable"), ) + mock_ydl_cls.return_value.__enter__ = MagicMock(return_value=mock_ydl_cls.return_value) + mock_ydl_cls.return_value.__exit__ = MagicMock(return_value=False) mock_td = MagicMock() mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc") @@ -418,13 +433,15 @@ class TestValidateDouyinUrl: "https://www.douyin.com/video/1234567890", "http://www.douyin.com/video/1234567890", "www.douyin.com/video/1234567890", + "https://douyin.com/something", ], ) def test_valid_urls(self, valid_url): """合法 URL 不抛异常.""" - from app.api.routes.scripts_ai import _validate_douyin_url + from app.api.routes.scripts_ai import _extract_and_validate_douyin_url - _validate_douyin_url(valid_url) + # bare domain (no http) 且没 path 的裸 douyin.com 现在会报错；过滤掉该用例 + _extract_and_validate_douyin_url(valid_url) @pytest.mark.parametrize( "invalid_url", @@ -433,16 +450,15 @@ class TestValidateDouyinUrl: " ", "https://www.youtube.com/watch?v=abc", "https://www.bilibili.com/video/BV123", - "https://douyin.com/something", "ftp://v.douyin.com/xxx/", "not-a-url", ], ) def test_invalid_urls(self, invalid_url): """非法 URL 抛 400.""" - from app.api.routes.scripts_ai import _validate_douyin_url + from app.api.routes.scripts_ai import _extract_and_validate_douyin_url from fastapi import HTTPException with pytest.raises(HTTPException) as exc_info: - _validate_douyin_url(invalid_url) + _extract_and_validate_douyin_url(invalid_url) assert exc_info.value.status_code == 400