diff --git a/tests/unit/test_extract_from_douyin_errors.py b/tests/unit/test_extract_from_douyin_errors.py index f806f1e00..ef8d75786 100644 --- a/tests/unit/test_extract_from_douyin_errors.py +++ b/tests/unit/test_extract_from_douyin_errors.py @@ -1,13 +1,16 @@ -"""验证 extract-from-douyin 在各种失败场景返回正确的 HTTP 状态码（绝不能 500）""" +"""验证 extract-from-douyin 在各种失败场景返回正确的 HTTP 状态码（绝不能 500） + +新版架构：douyin_resolver 多源轮询 → MediaKit ASR → 本地下载+ASR → desc 兜底。 +所有外部依赖（resolver、MediaKit、transcribe_to_text）均通过 mock 隔离。 +""" from __future__ import annotations -import sys -import types from unittest import mock import pytest from app.auth import AuthenticatedUser +from app.services.douyin_resolver import ResolveResult from fastapi import HTTPException, status @@ -22,52 +25,21 @@ def fake_user(): return AuthenticatedUser(user=_FakeUser()) -class _FakeYDLBase: - """通用假 yt-dlp 基类（支持上下文管理器 with 语法）""" - - extract_info_result = None - extract_info_raises = None - prepare_filename_result = "/tmp/fake.mp4" - - 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 - return self.__class__.extract_info_result - - def prepare_filename(self, info): - return self.__class__.prepare_filename_result - - -def _install_fake_ytdlp(fake_ydl_class, *, download_error_cls=None): - """把假 yt-dlp 注入 sys.modules，函数内 import yt_dlp 会拿到我们的假版本""" - fake_mod = types.ModuleType("yt_dlp") - fake_mod.YoutubeDL = fake_ydl_class - if download_error_cls is None: - download_error_cls = type("DownloadError", (Exception,), {}) - fake_mod.DownloadError = download_error_cls - utils_mod = types.ModuleType("yt_dlp.utils") - utils_mod.DownloadError = download_error_cls - fake_mod.utils = utils_mod - sys.modules["yt_dlp"] = fake_mod - sys.modules["yt_dlp.utils"] = utils_mod - return fake_mod - - def _import_target(): from app.api.routes import scripts_ai return scripts_ai +def _fake_mk_available(text="识别成功的文案", duration=5.0): + """Mock MediaKitClient 可用并返回指定 ASR 结果。""" + fake_mk = mock.MagicMock() + fake_mk.is_available = True + fake_mk.asr_submit.return_value = "tk1" + fake_mk.asr_poll.return_value = (text, duration) + return mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk) + + def _fake_mk_unavailable(): """Mock MediaKitClient 不可用，强制走下载+本地 ASR 路径。""" fake_mk = mock.MagicMock() @@ -75,362 +47,197 @@ def _fake_mk_unavailable(): 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""" +def _fake_resolver_success(video_url="https://example.com/direct.mp4", desc="", source="app_feed"): + """Mock resolver 返回成功。""" + result = ResolveResult(video_url=video_url, desc=desc, source=source) + return mock.patch("app.api.routes.scripts_ai.resolve_douyin_video", return_value=result) + + +def _fake_resolver_image(desc="图文文案内容", source="app_feed_image"): + """Mock resolver 返回图文视频（video_url 为空）。""" + result = ResolveResult(video_url="", desc=desc, source=source) + return mock.patch("app.api.routes.scripts_ai.resolve_douyin_video", return_value=result) + + +def _fake_resolver_failure(): + """Mock resolver 所有源均失败，返回 None。""" + return mock.patch("app.api.routes.scripts_ai.resolve_douyin_video", return_value=None) + + +# ── 解析阶段失败 ────────────────────────────────────────────────── + + +def test_resolver_all_fail_returns_503_parse(fake_user): + """所有解析源均失败 → 503 解析失败。""" scripts_ai = _import_target() body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/test123/") - class DownloadError(Exception): - pass - - class FailingYDL(_FakeYDLBase): - extract_info_raises = DownloadError("ERROR: Unable to download webpage: HTTP Error 404: Not Found") - - _install_fake_ytdlp(FailingYDL, download_error_cls=DownloadError) - - 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 - ), f"应为400，实际 {exc.value.status_code}: {exc.value.detail}" - assert "无法解析" in exc.value.detail or "抖音" in exc.value.detail - - -def test_download_network_error_returns_502_not_500(fake_user): - """网络错误 / 上游异常 → 502，不能 500""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - class DownloadError(Exception): - pass - - class NetErrYDL(_FakeYDLBase): - # 路径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 _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 - - -def test_info_none_returns_400(fake_user): - """yt-dlp 返回 None info → 400""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - class NoneInfoYDL(_FakeYDLBase): - def extract_info(self, url, download=True): - # 元信息探测返回 None；下载也返回 None - return None - - _install_fake_ytdlp(NoneInfoYDL) - - 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 - - -def test_asr_not_configured_returns_503(fake_user): - scripts_ai = _import_target() - from app.services.script_asr_service import ASRNotConfiguredError - - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - class OkYDL(_FakeYDLBase): - 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 ( - _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("未配置")), - ): + with _fake_resolver_failure(): 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 + assert "解析失败" in exc.value.detail or "链接" in exc.value.detail -def test_asr_failure_returns_502(fake_user): +# ── 图文视频（无需ASR） ───────────────────────────────────────── + + +def test_image_post_returns_desc_directly(fake_user): + """图文视频（resolver返回空video_url有desc）→ 直接返回 desc，不走 ASR。""" scripts_ai = _import_target() - from app.services.script_asr_service import ASRTranscriptionError + body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abcdeFG/") - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - class OkYDL(_FakeYDLBase): - def extract_info(self, url, download=True): - if not download: - return None - return {"id": "x", "duration": 10, "title": "t"} - - _install_fake_ytdlp(OkYDL) - with ( - _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("识别失败")), - ): - 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 + with _fake_resolver_image(desc="这是图文文案"): + resp = scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) + assert resp.text == "这是图文文案" + assert resp.duration_seconds == 0.0 -def test_asr_unexpected_error_returns_502_not_500(fake_user): - """ASR 抛未预期异常也应被兜住，不能 500""" +# ── MediaKit ASR 成功路径 ──────────────────────────────────────── + + +def test_mediakit_asr_success(fake_user): + """正常流程：resolver 成功 + MediaKit ASR 成功 → 返回文本。""" scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") + body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/ 快来看看！") - class OkYDL(_FakeYDLBase): - def extract_info(self, url, download=True): - if not download: - return None - return {"id": "x", "duration": 10, "title": "t"} - - _install_fake_ytdlp(OkYDL) - with ( - _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")), - ): - 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, f"应为502，实际 {exc.value.status_code}" - - -def test_missing_downloaded_file_returns_502_not_500(fake_user): - """yt-dlp 返回 info 但文件未落地（isfile False）→ 502""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - class OkYDL(_FakeYDLBase): - def extract_info(self, url, download=True): - if not download: - return None - return {"id": "x", "duration": 10, "title": "t"} - - _install_fake_ytdlp(OkYDL) - with ( - _fake_mk_unavailable(), - mock.patch.object(scripts_ai.os.path, "isfile", return_value=False), - ): - 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 != 500 - assert "下载异常" in exc.value.detail or "文件" in exc.value.detail - - -def test_any_unexpected_error_does_not_return_500_raw(fake_user): - """兜底：prepare_filename 抛未预期异常也应被捕获，返回500 code但含业务detail""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - 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 _fake_mk_unavailable(): - with pytest.raises(HTTPException) as exc: - scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) - assert "抖音" in exc.value.detail or "失败" in exc.value.detail or exc.value.status_code != 500 - - -# ── cookies 相关测试 ────────────────────────────────────────────── - - -def test_cookies_error_returns_503_friendly_message(fake_user): - """cookies 缺失/过期 → 返回 503 + 友好文案，不暴露原始错误""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/test123/") - - class DownloadError(Exception): - pass - - class CookiesYDL(_FakeYDLBase): - def extract_info(self, url, download=True): - raise DownloadError( - "ERROR: [Douyin] 7623712911260650802: Fresh cookies (not necessarily logged in) are needed" + with _fake_resolver_success(desc="Feed标题"): + with _fake_mk_available(text="这是MediaKit识别的文案", duration=12.5): + result = scripts_ai.extract_from_douyin( + request=body, current_user=fake_user, db=mock.MagicMock() ) - - _install_fake_ytdlp(CookiesYDL, download_error_cls=DownloadError) - - 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}" - assert "Fresh cookies" not in exc.value.detail + assert result.text == "这是MediaKit识别的文案" + assert result.duration_seconds == 12.5 -def test_cookies_error_in_generic_except_also_returns_503(fake_user): - """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): - raise RuntimeError("Fresh cookies are needed to access this video") - - _install_fake_ytdlp(CookieBugYDL) - - 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 指向该路径（在下载分支）""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - captured_opts_download = {} - - class CaptureOptsYDL(_FakeYDLBase): - def __init__(self, opts): - # 下载分支会触发 download=True；元信息探测 download=False - # 元信息也会传 cookiefile，但我们只在下载分支记录（更接近真实） - super().__init__() - self._opts = opts - # 总是记录最后一次的 opts，方便断言 - captured_opts_download.clear() - captured_opts_download.update(opts) - - def extract_info(self, url, download=True): - if not download: - return None # 元信息失败，走下载分支 - return {"id": "x", "duration": 5, "title": "t"} - - _install_fake_ytdlp(CaptureOptsYDL) - - with ( - _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), - mock.patch.object(scripts_ai, "transcribe_to_text", return_value="ok"), - ): - scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) - - 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): - """cookies 文件不存在时，ydl_opts 不应包含 cookiefile 键""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - captured_opts_download = {} - - class CaptureOptsYDL(_FakeYDLBase): - def __init__(self, opts): - super().__init__() - captured_opts_download.clear() - captured_opts_download.update(opts) - - def extract_info(self, url, download=True): - if not download: - return None - return {"id": "x", "duration": 5, "title": "t"} - - _install_fake_ytdlp(CaptureOptsYDL) - - with ( - _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), - mock.patch.object(scripts_ai, "transcribe_to_text", return_value="ok"), - ): - scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock()) - - assert ( - "cookiefile" not in captured_opts_download - ), f"cookies 文件缺失时不应设置 cookiefile，opts={captured_opts_download}" - - -def test_generic_download_error_hides_raw_message(fake_user): - """非 cookies 非 404 的通用下载错误 → 502，且不暴露 yt-dlp 原始错误文本""" - scripts_ai = _import_target() - body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/") - - class DownloadError(Exception): - pass - - class GenErrYDL(_FakeYDLBase): - 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 _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""" + """分享文本（含前后说明文字）应能正确提取 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"} + with _fake_resolver_success(): + with _fake_mk_available(text="识别成功的文案", duration=5.0) as mk_mock: + 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 - _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""" + """粘贴非抖音分享链接 → 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 + 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 + + +# ── ASR 空结果 → desc 兜底 ──────────────────────────────────── + + +def test_asr_empty_falls_back_to_desc(fake_user): + """MediaKit 和本地 ASR 都返回空文本（无旁白视频）→ 使用 desc 兜底。""" + scripts_ai = _import_target() + body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") + + # MediaKit 返回空文本 + fake_mk = mock.MagicMock() + fake_mk.is_available = True + fake_mk.asr_submit.return_value = "tk1" + fake_mk.asr_poll.return_value = ("", 4.5) + + # 本地下载+ASR 也返回空（通过mock _direct_url_download_and_local_asr） + with _fake_resolver_success(desc="Feed描述文案"): + with mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk): + with mock.patch( + "app.api.routes.scripts_ai._direct_url_download_and_local_asr", + return_value=("", 0.0), + ): + resp = scripts_ai.extract_from_douyin( + request=body, current_user=fake_user, db=mock.MagicMock() + ) + assert resp.text == "Feed描述文案" + + +# ── ASR 未配置 → 503 ───────────────────────────────────────── + + +def test_asr_not_configured_returns_503(fake_user): + """本地 ASR 未配置 → 503。""" + from app.services.script_asr_service import ASRNotConfiguredError + + scripts_ai = _import_target() + body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") + + with _fake_resolver_success(): + with _fake_mk_unavailable(): + with mock.patch( + "app.api.routes.scripts_ai._direct_url_download_and_local_asr", + side_effect=HTTPException(status_code=503, detail="ASR未配置"), + ): + 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 + + +# ── ASR 转写失败 → 502 ──────────────────────────────────────── + + +def test_asr_transcription_failure_returns_502(fake_user): + """ASR 转写异常 → 502（被 _direct_url_download_and_local_asr 包装）。""" + from app.services.script_asr_service import ASRTranscriptionError + + scripts_ai = _import_target() + body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") + + # MediaKit 失败 + fake_mk = mock.MagicMock() + fake_mk.is_available = True + from app.services.mediakit_client import MediaKitError + + fake_mk.asr_submit.side_effect = MediaKitError("ASR failed", code="TaskFailed") + + with _fake_resolver_success(desc=""): + with mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk): + with mock.patch( + "app.api.routes.scripts_ai._direct_url_download_and_local_asr", + side_effect=HTTPException(status_code=502, detail="语音识别失败"), + ): + 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 + + +# ── 下载超时 → 504 ──────────────────────────────────────────── + + +def test_download_timeout_returns_504(fake_user): + """视频下载超时 → 504。""" + scripts_ai = _import_target() + body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") + + with _fake_resolver_success(): + with _fake_mk_unavailable(): + with mock.patch( + "app.api.routes.scripts_ai._direct_url_download_and_local_asr", + side_effect=HTTPException(status_code=504, detail="视频下载超时"), + ): + 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_504_GATEWAY_TIMEOUT diff --git a/tests/unit/test_scripts_ai.py b/tests/unit/test_scripts_ai.py index 4629d6b40..6b3e60dc4 100644 --- a/tests/unit/test_scripts_ai.py +++ b/tests/unit/test_scripts_ai.py @@ -42,72 +42,43 @@ def mock_db(): return MagicMock() -def _mock_youtube_dl( - extract_info_return=None, - extract_info_side_effect=None, - prepare_filename_return="/tmp/douyin_extract_abc/abc123.mp4", +def _mock_resolver( + video_url="https://example.com/direct.mp4", + desc="", + source="app_feed", ): - """构造 yt_dlp.YoutubeDL 的 mock. - - 路由中用法: ydl = yt_dlp.YoutubeDL(opts); info = ydl.extract_info(...) - 所以 mock_ydl_cls.return_value 就是 ydl 实例. - """ - mock_ydl_instance = MagicMock() - if extract_info_side_effect is not None: - mock_ydl_instance.extract_info.side_effect = extract_info_side_effect - else: - mock_ydl_instance.extract_info.return_value = extract_info_return or { - "id": "abc123", - "duration": 120.5, - } - mock_ydl_instance.prepare_filename.return_value = prepare_filename_return - return mock_ydl_instance + """Mock resolve_douyin_video 返回 ResolveResult。""" + from app.services.douyin_resolver import ResolveResult + return ResolveResult(video_url=video_url, desc=desc, source=source) # ── extract_from_douyin ────────────────────────────────────────────────────── class TestExtractFromDouyin: - """POST /extract-from-douyin 测试.""" + """POST /extract-from-douyin 测试（新版：resolver 多源轮询架构）。""" @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( + @patch("app.api.routes.scripts_ai._direct_url_download_and_local_asr") + @patch("app.api.routes.scripts_ai.resolve_douyin_video") + def test_extract_from_douyin_success_mediakit( self, - mock_meta, - mock_isfile, - mock_getsize, - mock_ydl_cls, - mock_tempdir, - mock_transcribe, + mock_resolve, + mock_local_asr, mock_get_mk, ): - """正常流程（MediaKit不可用，走本地下载+ASR）：下载视频 + ASR 转写成功.""" + """正常流程（MediaKit ASR 成功）：resolver → MediaKit → 返回文本。""" from app.api.routes.scripts_ai import extract_from_douyin from app.schemas.scripts_ai import ExtractFromDouyinRequest + mock_resolve.return_value = _mock_resolver(desc="Feed标题") + fake_mk = MagicMock() - fake_mk.is_available = False + fake_mk.is_available = True + fake_mk.asr_submit.return_value = "tk1" + fake_mk.asr_poll.return_value = ("这是一段测试文案内容", 120.5) 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") - mock_td.__exit__ = MagicMock(return_value=False) - mock_tempdir.return_value = mock_td - - mock_transcribe.return_value = "这是一段测试文案内容" - req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") auth = _make_auth_user() result = extract_from_douyin(request=req, current_user=auth, db=MagicMock()) @@ -115,7 +86,36 @@ class TestExtractFromDouyin: assert result.text == "这是一段测试文案内容" assert result.duration_seconds == 120.5 assert result.source_url == "https://v.douyin.com/xxxxx/" - mock_transcribe.assert_called_once() + mock_local_asr.assert_not_called() + + @patch("app.api.routes.scripts_ai.get_mediakit_client") + @patch("app.api.routes.scripts_ai._direct_url_download_and_local_asr") + @patch("app.api.routes.scripts_ai.resolve_douyin_video") + def test_extract_from_douyin_success_local_asr( + self, + mock_resolve, + mock_local_asr, + mock_get_mk, + ): + """MediaKit 不可用时走本地下载+ASR。""" + from app.api.routes.scripts_ai import extract_from_douyin + from app.schemas.scripts_ai import ExtractFromDouyinRequest + + mock_resolve.return_value = _mock_resolver() + + fake_mk = MagicMock() + fake_mk.is_available = False + mock_get_mk.return_value = fake_mk + + mock_local_asr.return_value = ("本地ASR识别文案", 30.0) + + req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") + auth = _make_auth_user() + result = extract_from_douyin(request=req, current_user=auth, db=MagicMock()) + + assert result.text == "本地ASR识别文案" + assert result.duration_seconds == 30.0 + mock_local_asr.assert_called_once() @pytest.mark.parametrize( "bad_url", @@ -128,7 +128,7 @@ class TestExtractFromDouyin: ], ) def test_extract_from_douyin_invalid_url(self, bad_url): - """非法 URL 返回 400.""" + """非法 URL 返回 400。""" from app.api.routes.scripts_ai import extract_from_douyin from app.schemas.scripts_ai import ExtractFromDouyinRequest from fastapi import HTTPException @@ -140,111 +140,49 @@ 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, mock_meta, mock_get_mk): - """下载失败返回 502.""" + @patch("app.api.routes.scripts_ai.resolve_douyin_video", return_value=None) + def test_extract_from_douyin_resolver_all_fail(self, mock_resolve): + """所有解析源失败 → 503 解析失败。""" 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") - mock_td.__exit__ = MagicMock(return_value=False) - mock_tempdir.return_value = mock_td - - req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") - auth = _make_auth_user() - - with pytest.raises(HTTPException) as exc_info: - extract_from_douyin(request=req, current_user=auth) - assert exc_info.value.status_code == 502 - - @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) - def test_extract_from_douyin_asr_not_configured( - self, - mock_isfile, - mock_getsize, - mock_ydl_cls, - mock_tempdir, - mock_transcribe, - ): - """ASR 未配置返回 503.""" - from app.api.routes.scripts_ai import extract_from_douyin - from app.schemas.scripts_ai import ExtractFromDouyinRequest - from app.services.script_asr_service import ASRNotConfiguredError - from fastapi import HTTPException - - mock_ydl_cls.return_value = _mock_youtube_dl( - extract_info_return={"id": "abc123", "duration": 60}, - ) - - mock_td = MagicMock() - mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc") - mock_td.__exit__ = MagicMock(return_value=False) - mock_tempdir.return_value = mock_td - - mock_transcribe.side_effect = ASRNotConfiguredError("ASR 服务未配置") - req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") auth = _make_auth_user() with pytest.raises(HTTPException) as exc_info: extract_from_douyin(request=req, current_user=auth) assert exc_info.value.status_code == 503 + assert "解析" in exc_info.value.detail - @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) - def test_extract_from_douyin_asr_failure( + @patch("app.api.routes.scripts_ai.get_mediakit_client") + @patch("app.api.routes.scripts_ai._direct_url_download_and_local_asr") + @patch("app.api.routes.scripts_ai.resolve_douyin_video") + def test_extract_from_douyin_asr_empty_desc_fallback( self, - mock_isfile, - mock_getsize, - mock_ydl_cls, - mock_tempdir, - mock_transcribe, + mock_resolve, + mock_local_asr, + mock_get_mk, ): - """ASR 调用失败返回 502.""" + """ASR 返回空文本时使用 desc 兜底。""" from app.api.routes.scripts_ai import extract_from_douyin from app.schemas.scripts_ai import ExtractFromDouyinRequest - from app.services.script_asr_service import ASRTranscriptionError - from fastapi import HTTPException - mock_ydl_cls.return_value = _mock_youtube_dl( - extract_info_return={"id": "abc123", "duration": 60}, - ) + mock_resolve.return_value = _mock_resolver(desc="Feed描述文案兜底") - mock_td = MagicMock() - mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc") - mock_td.__exit__ = MagicMock(return_value=False) - mock_tempdir.return_value = mock_td + fake_mk = MagicMock() + fake_mk.is_available = True + fake_mk.asr_submit.return_value = "tk1" + fake_mk.asr_poll.return_value = ("", 4.5) # 空结果（无旁白视频） + mock_get_mk.return_value = fake_mk - mock_transcribe.side_effect = ASRTranscriptionError("语音识别失败: timeout") + mock_local_asr.return_value = ("", 0.0) # 本地也返回空 req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/") auth = _make_auth_user() + result = extract_from_douyin(request=req, current_user=auth, db=MagicMock()) - with pytest.raises(HTTPException) as exc_info: - extract_from_douyin(request=req, current_user=auth) - assert exc_info.value.status_code == 502 + assert result.text == "Feed描述文案兜底" # ── ai_rewrite ───────────────────────────────────────────────────────────────