Merge pull request #446 from JefferyHcool/codex/release-v2-4-5

chore(release): v2.4.5

.env.example

@@ -43,3 +43,12 @@ TRANSCRIBER_TYPE=fast-whisper
 WHISPER_MODEL_SIZE=tiny
 
 GROQ_TRANSCRIBER_MODEL=whisper-large-v3-turbo # groq提供的faster-whisper 默认为 whisper-large-v3-turbo
+
+# whisper 本地模型从 HuggingFace 下载。镜像默认 HF_ENDPOINT=https://hf-mirror.com（国内友好）。
+# 若下载失败（容器连不上镜像站），可在此覆盖为官方源或其它镜像，例如：
+#   HF_ENDPOINT=https://huggingface.co
+# 注意：宿主机的 VPN/代理默认不会进入容器。若要让模型下载走代理，请在前端「设置」里
+# 配置代理（会自动应用到 HuggingFace 下载），或在下面用标准环境变量指定：
+#   HTTP_PROXY=http://host.docker.internal:7890
+#   HTTPS_PROXY=http://host.docker.internal:7890
+# HF_ENDPOINT=https://hf-mirror.com

BillNote_frontend/src-tauri/tauri.conf.json

@@ -1,7 +1,7 @@
 {
   "$schema": "../node_modules/@tauri-apps/cli/config.schema.json",
   "productName": "BiliNote",
-  "version": "2.4.4",
+  "version": "2.4.5",
   "identifier": "com.jefferyhuang.bilinote",
   "build": {
     "frontendDist": "../dist",
@@ -43,4 +43,4 @@
       "icons/icon.png"
     ]
   }
-}
+}

BillNote_frontend/src/pages/HomePage/components/NoteForm.tsx

@@ -42,6 +42,9 @@ import { useNavigate } from 'react-router-dom'
 import toast from 'react-hot-toast'
 
 /* -------------------- 校验 Schema -------------------- */
+/** 用户粘贴的链接常缺协议头（如 bilibili.com/...），无任何 scheme 时自动补 https:// */
+const withScheme = (url: string) => (/^[a-z][a-z0-9+.-]*:\/\//i.test(url) ? url : `https://${url}`)
+
 const formSchema = z
   .object({
     video_url: z.string().optional(),
@@ -72,7 +75,7 @@ const formSchema = z
       }
       else {
         try {
-          const url = new URL(video_url)
+          const url = new URL(withScheme(video_url))
           if (!['http:', 'https:'].includes(url.protocol))
             throw new Error()
         }
@@ -221,6 +224,8 @@ const NoteForm = () => {
     console.log('Not even go here')
     const payload: NoteFormValues = {
       ...values,
+      video_url:
+        values.platform === 'local' ? values.video_url : withScheme(values.video_url || ''),
       provider_id: modelList.find(m => m.model_name === values.model_name)!.provider_id,
       task_id: currentTaskId || '',
     }

README.md

@@ -46,6 +46,23 @@ BiliNote 是一个开源的 AI 视频笔记助手，支持通过哔哩哔哩、Y
 
 直接访问 **[www.bilinote.app](https://www.bilinote.app/)** 即可使用 BiliNote Pro 在线版，无需本地部署。
 
+## 🌟 搭配使用：KaCutAI
+
+做 BiliNote 的时候，我发现不少用户不只是看别人的视频做笔记，自己也在拍、在剪、在攒素材。当本地素材越堆越多，找个画面翻半天——这个问题 BiliNote 解决不了。
+
+所以我做了 **[KaCutAI](https://www.kacut.app)**：一个跑在 Mac 本地的 AI 视频素材搜索引擎。用中文自然语言搜片段，不用上传云端，不按月付费。
+
+> 💡 BiliNote 帮你「看视频做笔记」，KaCutAI 帮你「找自己的视频素材」。一个看别人的，一个找自己的，搭配使用更香。
+
+**核心能力：**
+
+- 🇨🇳 中文原生：说人话就能搜，「穿白衬衫的人在说话」「有鸟叫的黄昏海滩」直接出结果
+- 🔒 全本地：素材不上云，AI 在你 Mac 上跑，隐私零泄露
+- 🧠 六模联动：画面 + 人物 + 动作 + 对话 + OCR + 音频，复合查询是它的强项
+- 💰 买断制：一次付费，永久使用
+
+**适合谁：** 有大量视频素材需要管理的创作者、剪辑师、自媒体、影像工作者。
+
 ## 📝 使用文档
 详细文档可以查看[这里](https://docs.bilinote.app/)
 ## 📦 桌面版下载

backend/app/db/builtin_providers.json

@@ -13,7 +13,8 @@
     "type": "built-in",
     "logo": "DeepSeek",
     "api_key": "",
-    "base_url": "https://api.deepseek.com"
+    "base_url": "https://api.deepseek.com",
+    "models": ["deepseek-chat", "deepseek-reasoner"]
   },
   {
     "id": "qwen",
@@ -21,7 +22,8 @@
     "type": "built-in",
     "logo": "Qwen",
     "api_key": "",
-    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1"
+    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
+    "models": ["qwen-plus", "qwen-turbo", "qwen-max", "qwen-long"]
   },
     {
     "id": "Claude",

backend/app/downloaders/base.py

@@ -13,6 +13,19 @@ QUALITY_MAP = {
     "slow": "128"
 }
 
+# yt-dlp 的 `retries` 默认值(10)是命令行参数解析器给的，Python API 不套用它：
+# 不显式设置时 HttpFD 拿到的是 `self.params.get('retries')` == None，而
+# `RetryManager.__init__` 做的是 `self.retries = _retries or 0`——也就是
+# 一次都不重试。任何一次网络抖动（例如 B 站 CDN
+# upos-sz-mirror*.bilivideo.com 读超时）都会让整个笔记任务直接失败。
+#
+# 这里的值偏保守：笔记任务是用户在前台等的，重试太多不如早点失败让用户重来。
+YDL_RETRY_OPTS = {
+    "retries": 3,
+    "fragment_retries": 3,
+    "socket_timeout": 30,
+}
+
 
 class Downloader(ABC):
     def __init__(self):

backend/app/downloaders/bilibili_dm_patch.py

@@ -58,12 +58,15 @@ def apply_bilibili_dm_img_patch() -> bool:
     if getattr(original, '_bili_dm_patched', False):
         return True
 
-    def _patched_download_playinfo(self, bvid, cid, headers=None, query=None):
+    def _patched_download_playinfo(self, bvid, cid, headers=None, query=None, **kwargs):
         # dm_* are merged into the query that the original method signs via
         # _sign_wbi; caller-supplied query params (e.g. try_look/qn) take
         # precedence over the injected dummies.
+        # **kwargs stays open on purpose: yt-dlp keeps adding parameters to
+        # _download_playinfo (2026.x added `fatal`), and a wrapper that pins the
+        # signature turns every such addition into a TypeError at download time.
         merged_query = {**build_dm_img_params(), **(query or {})}
-        return original(self, bvid, cid, headers=headers, query=merged_query)
+        return original(self, bvid, cid, headers=headers, query=merged_query, **kwargs)
 
     _patched_download_playinfo._bili_dm_patched = True
     BilibiliBaseIE._download_playinfo = _patched_download_playinfo

backend/app/downloaders/bilibili_downloader.py

@@ -7,7 +7,7 @@ from typing import Union, Optional, List
 
 import yt_dlp
 
-from app.downloaders.base import Downloader, DownloadQuality, QUALITY_MAP
+from app.downloaders.base import Downloader, DownloadQuality, QUALITY_MAP, YDL_RETRY_OPTS
 from app.downloaders.bilibili_dm_patch import apply_bilibili_dm_img_patch
 from app.downloaders.bilibili_subtitle import BilibiliSubtitleFetcher
 from app.models.notes_model import AudioDownloadResult
@@ -63,6 +63,7 @@ class BilibiliDownloader(Downloader, ABC):
         output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
 
         ydl_opts = {
+            **YDL_RETRY_OPTS,
             'format': 'bestaudio[ext=m4a]/bestaudio/best',
             'outtmpl': output_path,
             'http_headers': {'Referer': 'https://www.bilibili.com'},
@@ -122,6 +123,7 @@ class BilibiliDownloader(Downloader, ABC):
         output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
 
         ydl_opts = {
+            **YDL_RETRY_OPTS,
             'format': 'bv*[ext=mp4]/bestvideo+bestaudio/best',
             'outtmpl': output_path,
             'http_headers': {'Referer': 'https://www.bilibili.com'},
@@ -183,6 +185,7 @@ class BilibiliDownloader(Downloader, ABC):
         video_id = extract_video_id(video_url, "bilibili")
 
         ydl_opts = {
+            **YDL_RETRY_OPTS,
             'writesubtitles': True,
             'writeautomaticsub': True,
             'subtitleslangs': langs,

backend/app/downloaders/youtube_downloader.py

@@ -5,7 +5,7 @@ from typing import Union, Optional, List
 
 import yt_dlp
 
-from app.downloaders.base import Downloader, DownloadQuality
+from app.downloaders.base import Downloader, DownloadQuality, YDL_RETRY_OPTS
 from app.downloaders.youtube_subtitle import YouTubeSubtitleFetcher
 from app.models.notes_model import AudioDownloadResult
 from app.models.transcriber_model import TranscriptResult
@@ -47,6 +47,7 @@ class YoutubeDownloader(Downloader, ABC):
         output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
 
         ydl_opts = {
+            **YDL_RETRY_OPTS,
             'format': 'bestaudio[ext=m4a]/bestaudio/best',
             'outtmpl': output_path,
             'noplaylist': True,
@@ -55,6 +56,10 @@ class YoutubeDownloader(Downloader, ABC):
 
         if skip_download:
             ydl_opts['skip_download'] = True
+            # 只取元信息时并不需要媒体流。yt-dlp 版本落后于 YouTube player 时，
+            # nsig 解析失败会导致所有音视频格式被丢弃，此时格式选择会抛
+            # "Requested format is not available"，把一个已经拿到字幕的任务带崩。
+            ydl_opts['ignore_no_formats_error'] = True
 
         _apply_proxy(ydl_opts)
         with yt_dlp.YoutubeDL(ydl_opts) as ydl:
@@ -63,7 +68,8 @@ class YoutubeDownloader(Downloader, ABC):
             title = info.get("title")
             duration = info.get("duration", 0)
             cover_url = info.get("thumbnail")
-            ext = info.get("ext", "m4a")
+            # skip_download 时 yt-dlp 返回 ext=None，默认值不会生效，避免拼出 "xxx.None"
+            ext = info.get("ext") or "m4a"
             audio_path = os.path.join(output_dir, f"{video_id}.{ext}")
 
         return AudioDownloadResult(
@@ -95,6 +101,7 @@ class YoutubeDownloader(Downloader, ABC):
         output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
 
         ydl_opts = {
+            **YDL_RETRY_OPTS,
             'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]',
             'outtmpl': output_path,
             'noplaylist': True,

backend/app/routers/config.py

@@ -240,6 +240,38 @@ class ModelDownloadRequest(BaseModel):
     transcriber_type: str = "fast-whisper"  # "fast-whisper" 或 "mlx-whisper"
 
 
+def _friendly_download_error(e: Exception) -> str:
+    """把 HuggingFace 的网络类报错翻译成用户能照着做的提示（issue #417）。
+
+    典型原文：'An error happened while trying to locate the file on the Hub and we
+    cannot find the requested files in the local cache...' —— 本质是连不上 Hub。
+    用户大概率不知道：默认走 hf-mirror.com 镜像，可配代理或改 HF_ENDPOINT。
+    """
+    raw = str(e)
+    lowered = raw.lower()
+    network_markers = (
+        "locate the file on the hub",
+        "couldn't connect",
+        "connection error",
+        "connecttimeout",
+        "read timed out",
+        "max retries exceeded",
+        "failed to establish",
+        "name or service not known",
+        "temporary failure in name resolution",
+    )
+    if any(m in lowered for m in network_markers):
+        endpoint = os.getenv("HF_ENDPOINT", "https://huggingface.co")
+        return (
+            f"{raw}\n"
+            f"——连不上模型仓库（当前 HF_ENDPOINT={endpoint}）。可尝试："
+            f"1) 在「设置」里配置可用代理；"
+            f"2) 设置环境变量 HF_ENDPOINT 切换镜像（国内可用 https://hf-mirror.com）；"
+            f"3) 确认容器能访问外网/镜像站后重试。"
+        )
+    return raw
+
+
 def _do_download_whisper(model_size: str):
     """后台下载 faster-whisper 模型（支持内置 size / 自定义 repo_id / 本地路径）。
 
@@ -250,9 +282,14 @@ def _do_download_whisper(model_size: str):
     """
     from huggingface_hub import snapshot_download
     from app.transcriber.whisper_models import resolve_whisper_model, is_local_target
+    from app.services.proxy_config_manager import ProxyConfigManager
 
     try:
         dl_state.mark_downloading(model_size)
+        # 让 UI 配的代理对 HuggingFace 下载也生效（issue #417：容器里代理没生效）
+        proxy = ProxyConfigManager().apply_to_env()
+        if proxy:
+            logger.info(f"whisper 下载走代理: {proxy}")
         model_dir = get_model_dir("whisper")
 
         # 已经下好就不重复下
@@ -289,8 +326,9 @@ def _do_download_whisper(model_size: str):
         logger.info(f"whisper 模型下载完成: {model_size}")
         dl_state.mark_done(model_size)
     except Exception as e:
+        msg = _friendly_download_error(e)
         logger.error(f"whisper 模型下载失败: {model_size}, {e}")
-        dl_state.mark_failed(model_size, str(e))
+        dl_state.mark_failed(model_size, msg)
 
 
 def _do_download_mlx_whisper(model_size: str):
@@ -300,6 +338,12 @@ def _do_download_mlx_whisper(model_size: str):
         dl_state.mark_downloading(key)
         from huggingface_hub import snapshot_download as hf_download
         from app.transcriber.mlx_whisper_transcriber import resolve_mlx_repo_id
+        from app.services.proxy_config_manager import ProxyConfigManager
+
+        # 让 UI 配的代理对 HuggingFace 下载也生效（issue #417）
+        proxy = ProxyConfigManager().apply_to_env()
+        if proxy:
+            logger.info(f"mlx-whisper 下载走代理: {proxy}")
 
         try:
             repo_id = resolve_mlx_repo_id(model_size)
@@ -319,8 +363,9 @@ def _do_download_mlx_whisper(model_size: str):
         logger.info(f"mlx-whisper 模型下载完成: {model_size}")
         dl_state.mark_done(key)
     except Exception as e:
+        msg = _friendly_download_error(e)
         logger.error(f"mlx-whisper 模型下载失败: {model_size}, {e}")
-        dl_state.mark_failed(key, str(e))
+        dl_state.mark_failed(key, msg)
 
 
 @router.post("/transcriber_download")

backend/app/routers/note.py

@@ -7,7 +7,7 @@ from typing import Optional
 from urllib.parse import urlparse
 
 from fastapi import APIRouter, HTTPException, BackgroundTasks, UploadFile, File
-from pydantic import BaseModel, validator, field_validator
+from pydantic import BaseModel, validator, field_validator, model_validator
 from dataclasses import asdict
 
 from app.db.video_task_dao import get_task_by_video
@@ -17,7 +17,7 @@ from app.exceptions.note import NoteError
 from app.services.note import NoteGenerator, logger
 from app.services.task_serial_executor import task_serial_executor
 from app.utils.response import ResponseWrapper as R
-from app.utils.url_parser import extract_video_id
+from app.utils.url_parser import extract_video_id, normalize_video_url
 from app.validators.video_url_validator import is_supported_video_url
 from fastapi import APIRouter, Request, HTTPException
 from fastapi.responses import StreamingResponse
@@ -55,6 +55,15 @@ class VideoRequest(BaseModel):
     #   {"language": "zh", "full_text": "...", "segments": [{"start","end","text"}, ...]}
     prefetched_transcript: Optional[dict] = None
 
+    @model_validator(mode="before")
+    @classmethod
+    def normalize_url(cls, data):
+        # 稍后再看/收藏夹/带追踪参数的 B 站链接先规范化成标准 /video/BVxxx 形式，
+        # 后续校验和 yt-dlp 下载拿到的都是干净链接
+        if isinstance(data, dict) and data.get("platform") == "bilibili" and data.get("video_url"):
+            data["video_url"] = normalize_video_url(str(data["video_url"]))
+        return data
+
     @field_validator("video_url")
     def validate_supported_url(cls, v):
         url = str(v)

backend/app/services/model.py

@@ -83,22 +83,56 @@ class ModelService:
         return enabled_models
     @staticmethod
     def get_all_models_by_id(provider_id: str, verbose: bool = False):
+        """拉取某供应商的可选模型列表，用于设置页下拉。
+
+        历史坑（issue #417）：旧实现对 get_model_list 的返回值直接取 `.data`，但
+        get_model_list 在 /models 调用失败时会吞掉异常返回 `[]`，于是 `[].data`
+        触发 AttributeError，又被这里的 except 吞成 `[]` —— 最终接口返回
+        `{"code":0,"msg":"success","data":[]}`，把「DeepSeek /models 取不到」伪装成
+        成功的空列表，用户完全看不到原因。
+
+        现在：
+          1. 直接捕获 /models 的真实异常（不再二次吞）；
+          2. normalize_models 兼容 SyncPage / list / dict，绝不再 `.data` 崩；
+          3. 动态拿不到（失败或空）时退回内置已知清单，保证下拉非空；
+          4. 仍然为空且确有报错时，把报错带回去（前端可提示，不再假装成功）。
+        """
+        from app.services.model_fallback import (
+            builtin_fallback_models,
+            normalize_models,
+            as_model_dicts,
+        )
+
+        provider = ProviderService.get_provider_by_id(provider_id)
+        if not provider:
+            logger.warning(f"[{provider_id}] 供应商不存在")
+            return {"models": []}
+
+        models: list = []
+        error: str | None = None
         try:
-            provider = ProviderService.get_provider_by_id(provider_id)
-
-            models = ModelService.get_model_list(provider["id"], verbose=verbose)
-            print(type(models))
-            serializable_models = [m.dict() for m in models.data]
-            model_list = {
-                "models": serializable_models
-            }
-
-            logger.info(f"[{provider['name']}] 获取模型成功")
-            return model_list
+            config = ModelService._build_model_config(provider)
+            gpt = GPTFactory().from_config(config)
+            models = normalize_models(gpt.list_models())
+            if verbose:
+                print(f"[{provider['name']}] 动态模型列表: {models}")
         except Exception as e:
-            # print(f"[{provider_id}] 获取模型失败: {e}")
-            logger.error(f"[{provider_id}] 获取模型失败: {e}")
-            return []
+            error = str(e)
+            logger.warning(f"[{provider['name']}] 动态获取模型失败，尝试回退内置清单: {e}")
+
+        if not models:
+            fallback = builtin_fallback_models(provider)
+            if fallback:
+                logger.info(f"[{provider['name']}] /models 为空，回退内置清单: {fallback}")
+                models = as_model_dicts(fallback, owned_by=provider.get("name", ""))
+
+        result = {"models": models}
+        if not models and error:
+            # 既没动态结果也没兜底清单：把真实报错带回去，别再伪装成功
+            result["error"] = error
+        else:
+            logger.info(f"[{provider['name']}] 获取模型成功，共 {len(models)} 个")
+        return result
     @staticmethod
     def connect_test(id: str, model: str | None = None) -> bool:
         """连通性测试：发一条最小化 chat completion。

backend/app/services/model_fallback.py

@@ -0,0 +1,86 @@
+"""内置供应商的回退模型清单 + 模型对象归一化（issue #417）。
+
+背景：设置页的「模型下拉」依赖 provider 的 `/v1/models` 动态列表。但这个接口
+并不可靠——
+
+  - DeepSeek 的 `/models` 在部分账号/网络下取不到，下拉直接空白；
+  - 不少自建 OpenAI 兼容网关压根不实现 `/models`；
+  - key 没有 inference 权限时也可能返回异常。
+
+（`OpenAI_compatible_provider.test_connection` 的注释里已经记录过这个不可靠性。）
+
+所以对**内置供应商**额外维护一份已知可用清单兜底：动态拿不到时退回这份清单，
+保证下拉永远有内容，用户不至于卡在空列表。清单数据写在
+`app/db/builtin_providers.json` 的 `models` 字段里，单一数据源，方便维护。
+
+本模块只依赖标准库，便于单测隔离加载（不触发 app 包的重依赖导入链）。
+"""
+import json
+from pathlib import Path
+from typing import Any, List, Optional
+
+# builtin_providers.json 与本文件同属 backend/app 下：app/services/ -> app/db/
+_BUILTIN_JSON = Path(__file__).resolve().parent.parent / "db" / "builtin_providers.json"
+
+
+def _load_builtin() -> List[dict]:
+    try:
+        return json.loads(_BUILTIN_JSON.read_text(encoding="utf-8"))
+    except Exception:
+        return []
+
+
+def builtin_fallback_models(provider: Optional[dict]) -> List[str]:
+    """按 provider 的 id 或 name（忽略大小写）匹配内置清单里的 models 字段。
+
+    自定义供应商（DB 里 id 是 uuid）通常 name 也对得上内置名，所以 id / name 都试。
+    匹配不到或没配 models 返回空列表。
+    """
+    if not provider:
+        return []
+    keys = {str(provider.get("id", "")).strip().lower(), str(provider.get("name", "")).strip().lower()}
+    keys.discard("")
+    if not keys:
+        return []
+    for p in _load_builtin():
+        candidate = {str(p.get("id", "")).strip().lower(), str(p.get("name", "")).strip().lower()}
+        if keys & candidate:
+            models = p.get("models") or []
+            return [str(m) for m in models if m]
+    return []
+
+
+def normalize_models(raw: Any) -> List[dict]:
+    """把 SDK 返回值统一成 [{'id', 'object', 'owned_by', ...}] 列表。
+
+    兼容三种形态：
+      - openai SDK 的 SyncPage（取 .data）
+      - 普通 list（含旧代码失败时返回的 []，绝不能再 .data）
+      - list 里既可能是 pydantic Model 也可能是 dict
+    """
+    if raw is None:
+        return []
+    data = getattr(raw, "data", raw)  # SyncPage -> .data；list/tuple 原样
+    if not isinstance(data, (list, tuple)):
+        return []
+    out: List[dict] = []
+    for m in data:
+        if isinstance(m, dict):
+            d = m
+        elif hasattr(m, "model_dump"):
+            d = m.model_dump()
+        elif hasattr(m, "dict"):
+            d = m.dict()
+        else:
+            d = {"id": getattr(m, "id", None)}
+        if d.get("id"):
+            out.append(d)
+    return out
+
+
+def as_model_dicts(model_ids: List[str], owned_by: str = "") -> List[dict]:
+    """把模型名列表包成与 SDK Model 一致的 dict，前端下拉直接复用同一套渲染。"""
+    return [
+        {"id": mid, "object": "model", "created": None, "owned_by": owned_by}
+        for mid in model_ids
+    ]

backend/app/services/note.py

@@ -262,7 +262,10 @@ class NoteGenerator:
             raise Exception(f"不支持的转写器：{self.transcriber_type}")
 
         logger.info(f"使用转写器：{self.transcriber_type}")
-        return get_transcriber(transcriber_type=self.transcriber_type)
+        return get_transcriber(
+            transcriber_type=self.transcriber_type,
+            model_size=self.model_size,
+        )
 
     def _get_gpt(self, model_name: Optional[str], provider_id: Optional[str]) -> GPT:
         """

backend/app/services/proxy_config_manager.py

@@ -58,3 +58,22 @@ class ProxyConfigManager:
             if val:
                 return val
         return None
+
+    def apply_to_env(self) -> Optional[str]:
+        """把当前生效的代理 URL 写进进程环境变量，返回生效的 url（无则 None）。
+
+        为什么需要（issue #417）：huggingface_hub / requests 这类库**只认**环境变量
+        HTTP_PROXY / HTTPS_PROXY / ALL_PROXY，不读我们 UI 配置文件。whisper 模型用
+        snapshot_download 从 HuggingFace 拉取，如果用户只在设置页填了代理，下载根本
+        不走代理 —— 就是用户说的「Docker 容器里代理没生效」。在下载前/启动时调用本
+        方法，把 UI 配的代理 export 到环境变量，HF 下载就能复用同一个代理。
+
+        大小写别名都写，覆盖不同库的读取习惯。
+        """
+        url = self.get_proxy_url()
+        if not url:
+            return None
+        for key in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
+                    "http_proxy", "https_proxy", "all_proxy"):
+            os.environ[key] = url
+        return url

backend/app/transcriber/bcut.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import threading
 import time
 from typing import Optional, List, Dict, Union
 
@@ -40,6 +41,10 @@ class BcutTranscriber(Transcriber):
         self.session = requests.Session()
         self.task_id = None
         self.__etags = []
+        # 实例被 transcriber_provider 缓存复用，并发任务会交错读写上传会话状态
+        # （etags/upload_id/task_id），必须整段串行化
+        # ponytail: 全局实例锁，并发转写会排队；如需吞吐量应改为每任务独立实例
+        self._lock = threading.Lock()
 
         self.__in_boss_key: Optional[str] = None
         self.__resource_id: Optional[str] = None
@@ -59,6 +64,15 @@ class BcutTranscriber(Transcriber):
 
     def _upload(self, file_path: str) -> None:
         """申请上传"""
+        # 实例被 transcriber_provider 缓存复用，必须清掉上一次上传的会话状态，
+        # 否则 __etags 跨上传残留，提交的 etag 数与本次分片数不符，B 站会拒绝合并
+        self.__etags = []
+        self.__in_boss_key = None
+        self.__resource_id = None
+        self.__upload_id = None
+        self.__upload_urls = []
+        self.__download_url = None
+
         file_binary = self._load_file(file_path)
         if not file_binary:
             raise ValueError("无法读取文件数据")
@@ -169,6 +183,10 @@ class BcutTranscriber(Transcriber):
     @timeit
     def transcript(self, file_path: str) -> TranscriptResult:
         """执行识别过程，符合 Transcriber 接口"""
+        with self._lock:
+            return self._transcript_locked(file_path)
+
+    def _transcript_locked(self, file_path: str) -> TranscriptResult:
         try:
             logger.info(f"开始处理文件: {file_path}")
             

backend/app/transcriber/transcriber_provider.py

@@ -1,5 +1,6 @@
 import os
 import platform
+import threading
 from enum import Enum
 
 from app.transcriber.groq import GroqTranscriber
@@ -38,17 +39,29 @@ _transcribers = {
     TranscriberType.GROQ: None,
 }
 
+# Cache instances together with their constructor configuration. The
+# transcriber choice and Whisper model size can be changed from the frontend,
+# so caching by transcriber type alone would keep using the first loaded model.
+_transcriber_configs = {key: None for key in _transcribers}
+_transcriber_init_lock = threading.Lock()
+
 # 公共实例初始化函数
 def _init_transcriber(key: TranscriberType, cls, *args, **kwargs):
-    if _transcribers[key] is None:
-        logger.info(f'创建 {cls.__name__} 实例: {key}')
-        try:
-            _transcribers[key] = cls(*args, **kwargs)
+    init_config = (args, tuple(sorted(kwargs.items())))
+    with _transcriber_init_lock:
+        instance = _transcribers[key]
+        if instance is None or _transcriber_configs[key] != init_config:
+            action = "创建" if instance is None else "按新配置重新创建"
+            logger.info(f'{action} {cls.__name__} 实例: {key}')
+            try:
+                new_instance = cls(*args, **kwargs)
+            except Exception as e:
+                logger.error(f"{cls.__name__} 创建失败: {e}")
+                raise
+            _transcribers[key] = new_instance
+            _transcriber_configs[key] = init_config
             logger.info(f'{cls.__name__} 创建成功')
-        except Exception as e:
-            logger.error(f"{cls.__name__} 创建失败: {e}")
-            raise
-    return _transcribers[key]
+        return _transcribers[key]
 
 # 各类型获取方法
 def get_groq_transcriber():
@@ -70,13 +83,13 @@ def get_mlx_whisper_transcriber(model_size="base"):
     return _init_transcriber(TranscriberType.MLX_WHISPER, MLXWhisperTranscriber, model_size=model_size)
 
 # 通用入口
-def get_transcriber(transcriber_type="fast-whisper", model_size="base", device="cuda"):
+def get_transcriber(transcriber_type="fast-whisper", model_size=None, device="cuda"):
     """
     获取指定类型的转录器实例
 
     参数:
         transcriber_type: 支持 "fast-whisper", "mlx-whisper", "bcut", "kuaishou", "groq"
-        model_size: 模型大小，适用于 whisper 类
+        model_size: 模型大小，适用于 whisper 类；未提供时才读取环境变量默认值
         device: 设备类型（如 cuda / cpu），仅 whisper 使用
 
     返回:
@@ -90,7 +103,9 @@ def get_transcriber(transcriber_type="fast-whisper", model_size="base", device="
         logger.warning(f'未知转录器类型 "{transcriber_type}"，默认使用 fast-whisper')
         transcriber_enum = TranscriberType.FAST_WHISPER
 
-    whisper_model_size = os.environ.get("WHISPER_MODEL_SIZE", model_size)
+    # The explicit value normally comes from the persisted frontend setting and
+    # must take precedence over Docker's startup default.
+    whisper_model_size = model_size or os.environ.get("WHISPER_MODEL_SIZE", "base")
 
     if transcriber_enum == TranscriberType.FAST_WHISPER:
         return get_whisper_transcriber(whisper_model_size, device=device)

backend/app/utils/url_parser.py

@@ -35,6 +35,29 @@ def extract_video_id(url: str, platform: str) -> Optional[str]:
     return None
 
 
+def normalize_video_url(url: str) -> str:
+    """
+    将任意包含 BV 号的 B 站链接规范化为标准视频链接。
+
+    支持稍后再看（/list/watchlater/?bvid=BV...）、收藏夹播放页（/list/mlXXX?bvid=BV...）、
+    带追踪参数的分享链接等。保留分 P 参数，丢弃其余查询参数。
+
+    b23.tv 短链与无 BV 号的链接原样返回（后者交由校验器拒绝）。
+    """
+    if "b23.tv" in url:
+        return url
+
+    match = re.search(r"BV([0-9A-Za-z]+)", url)
+    if not match:
+        return url
+
+    normalized = f"https://www.bilibili.com/video/BV{match.group(1)}"
+    p = extract_bilibili_p_number(url)
+    if p:
+        normalized += f"?p={p}"
+    return normalized
+
+
 def resolve_bilibili_short_url(short_url: str) -> Optional[str]:
     """
     解析哔哩哔哩短链接以获取真实视频链接

backend/app/validators/video_url_validator.py

@@ -1,7 +1,9 @@
-from pydantic import AnyUrl, validator, BaseModel, field_validator
+from pydantic import AnyUrl, validator, BaseModel, field_validator, model_validator
 import re
 from urllib.parse import urlparse
 
+from app.utils.url_parser import normalize_video_url
+
 SUPPORTED_PLATFORMS = {
     "bilibili": r"(https?://)?(www\.)?bilibili\.com/video/[a-zA-Z0-9]+",
     "youtube": r"(https?://)?(www\.)?(youtube\.com/(watch\?v=|shorts/)|youtu\.be/)[\w\-]+",
@@ -31,6 +33,13 @@ class VideoRequest(BaseModel):
     url: AnyUrl
     platform: str
 
+    @model_validator(mode="before")
+    @classmethod
+    def normalize_url(cls, data):
+        if isinstance(data, dict) and data.get("platform") == "bilibili" and data.get("url"):
+            data["url"] = normalize_video_url(str(data["url"]))
+        return data
+
     @field_validator("url")
     def validate_video_url(cls, v):
         if not is_supported_video_url(str(v)):

backend/main.py

@@ -60,6 +60,13 @@ async def lifespan(app: FastAPI):
         logger.info("[startup 4/5] seed_default_providers() — 初始化默认 LLM 供应商")
         seed_default_providers()
 
+        # 把已配置的代理 export 到环境变量，让 huggingface_hub（whisper 模型下载）
+        # 也能走代理——含转写时的按需下载（issue #417）。
+        from app.services.proxy_config_manager import ProxyConfigManager
+        _proxy = ProxyConfigManager().apply_to_env()
+        if _proxy:
+            logger.info(f"           已应用全局代理到环境变量: {_proxy}")
+
         logger.info("[startup 5/5] 启动完成，等待请求")
     except Exception:
         logger.exception("[startup FAILED] 后端启动期异常，详见堆栈；容器会退出并由 restart 策略决定是否重试")

backend/requirements.txt

@@ -95,7 +95,7 @@ pytz==2025.2
 pywin32-ctypes==0.2.3
 PyYAML==6.0.2
 redis==5.2.1
-requests==2.32.3
+requests==2.32.4
 rich==14.0.0
 rich-toolkit==0.14.1
 shellingham==1.5.4
@@ -125,5 +125,5 @@ webencodings==0.5.1
 websockets==15.0.1
 yarl==1.19.0
 youtube-transcript-api>=1.0.0
-yt-dlp==2025.3.31
+yt-dlp>=2026.7.4
 zopfli==0.2.3.post1

backend/tests/test_bcut_state_reset.py

@@ -0,0 +1,74 @@
+import json
+
+from app.transcriber import bcut as bcut_module
+from app.transcriber.bcut import BcutTranscriber
+
+
+class _FakeResp:
+    def __init__(self, payload=None, etag=""):
+        self._payload = payload or {}
+        self.headers = {"Etag": etag}
+        self.url = ""
+
+    def raise_for_status(self):
+        pass
+
+    def json(self):
+        return self._payload
+
+
+class _FakeSession:
+    """模拟必剪上传三步接口：申请上传 / PUT 分片 / 提交合并"""
+
+    def __init__(self, per_size, upload_urls_count):
+        self.per_size = per_size
+        self.upload_urls_count = upload_urls_count
+        self.commit_payloads = []
+
+    def post(self, url, data=None, headers=None):
+        if url == bcut_module.API_REQ_UPLOAD:
+            body = json.loads(data)
+            return _FakeResp({
+                "data": {
+                    "in_boss_key": "boss",
+                    "resource_id": "res",
+                    "upload_id": "up",
+                    "upload_urls": [f"http://fake/{i}" for i in range(self.upload_urls_count)],
+                    "per_size": self.per_size,
+                    "size": body["size"],
+                }
+            })
+        if url == bcut_module.API_COMMIT_UPLOAD:
+            self.commit_payloads.append(json.loads(data))
+            return _FakeResp({"code": 0, "data": {"download_url": "http://fake/dl"}})
+        raise AssertionError(f"unexpected post: {url}")
+
+    def put(self, url, data=None, headers=None):
+        return _FakeResp(etag=f"etag-{url[-1]}")
+
+
+def _upload_file(transcriber, tmp_path, name, content, chunks):
+    f = tmp_path / name
+    f.write_bytes(content)
+    transcriber.session = _FakeSession(per_size=5, upload_urls_count=chunks)
+    transcriber._upload(str(f))
+    return transcriber.session.commit_payloads
+
+
+def test_second_upload_commits_only_its_own_etags(tmp_path):
+    t = BcutTranscriber()
+
+    # 第一次：3 字节，1 分片
+    commits = _upload_file(t, tmp_path, "a.mp3", b"aaa", chunks=1)
+    assert len(commits[0]["Etags"].split(",")) == 1
+
+    # 第二次（同一实例）：8 字节，2 分片——修复前会提交 3 个 etag
+    commits = _upload_file(t, tmp_path, "b.mp3", b"aaaaaaaa", chunks=2)
+    assert len(commits[0]["Etags"].split(",")) == 2
+
+
+def test_first_upload_unchanged(tmp_path):
+    t = BcutTranscriber()
+    commits = _upload_file(t, tmp_path, "a.mp3", b"aaa", chunks=1)
+    assert commits[0]["Etags"] == "etag-0"
+    assert commits[0]["UploadId"] == "up"

backend/tests/test_bilibili_dm_patch.py

@@ -71,7 +71,9 @@ class ApplyPatchTest(unittest.TestCase):
 
         def fake_download_json(url, video_id, **kwargs):
             # Avoid any network; the real playurl call would 412 without dm_*.
-            return {"data": {"ok": True}}
+            # yt-dlp >= 2026.x checks the envelope's `code` before returning
+            # `data`, so the fake has to look like a real playurl response.
+            return {"code": 0, "data": {"ok": True}}
 
         ie = BilibiliBaseIE(YoutubeDL({"quiet": True}))
         ie._sign_wbi = fake_sign_wbi
@@ -89,6 +91,32 @@ class ApplyPatchTest(unittest.TestCase):
         # the original method still builds its base params
         self.assertEqual(captured["bvid"], "BV1X9L16oEgB")
 
+    def test_patch_forwards_unknown_kwargs_to_original(self):
+        """
+        yt-dlp's real call site passes kwargs the wrapper never declared —
+        `_real_extract` calls `_download_playinfo(..., fatal=False)` since
+        2026.x. A wrapper with a pinned signature raises TypeError there and
+        breaks every Bilibili download, so unknown kwargs must pass through.
+        """
+        from yt_dlp import YoutubeDL
+        from yt_dlp.extractor.bilibili import BilibiliBaseIE
+
+        bilibili_dm_patch.apply_bilibili_dm_img_patch()
+
+        seen = {}
+
+        def fake_download_json(url, video_id, **kwargs):
+            return {"code": 0, "data": {"ok": True}}
+
+        ie = BilibiliBaseIE(YoutubeDL({"quiet": True}))
+        ie._sign_wbi = lambda params, video_id: seen.update(params) or params
+        ie._download_json = fake_download_json
+
+        # Must not raise TypeError on a kwarg the wrapper does not name.
+        ie._download_playinfo("BV1X9L16oEgB", 4242, headers={}, query={}, fatal=False)
+
+        self.assertTrue(REQUIRED_KEYS.issubset(seen.keys()))
+
 
 if __name__ == "__main__":
     unittest.main()

backend/tests/test_model_fallback.py

@@ -0,0 +1,89 @@
+import importlib.util
+import pathlib
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+MODULE_PATH = ROOT / "app" / "services" / "model_fallback.py"
+spec = importlib.util.spec_from_file_location("model_fallback", MODULE_PATH)
+if spec is None or spec.loader is None:
+    raise ImportError("model_fallback module spec not found")
+model_fallback = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(model_fallback)
+
+builtin_fallback_models = model_fallback.builtin_fallback_models
+normalize_models = model_fallback.normalize_models
+as_model_dicts = model_fallback.as_model_dicts
+
+
+class FakeModel:
+    """模拟 openai SDK 的 Model（pydantic）对象。"""
+
+    def __init__(self, mid, created=None, owned_by="x"):
+        self.id = mid
+        self.created = created
+        self.object = "model"
+        self.owned_by = owned_by
+
+    def dict(self):
+        return {"id": self.id, "created": self.created, "object": self.object, "owned_by": self.owned_by}
+
+
+class FakeSyncPage:
+    def __init__(self, data):
+        self.data = data
+
+
+class TestBuiltinFallbackModels(unittest.TestCase):
+    def test_deepseek_by_id_has_fallback(self):
+        # issue #417：DeepSeek /models 不稳定，必须有兜底清单
+        models = builtin_fallback_models({"id": "deepseek", "name": "DeepSeek"})
+        self.assertIn("deepseek-chat", models)
+        self.assertIn("deepseek-reasoner", models)
+
+    def test_match_by_name_case_insensitive(self):
+        models = builtin_fallback_models({"id": "whatever-uuid", "name": "deepseek"})
+        self.assertIn("deepseek-chat", models)
+
+    def test_unknown_provider_returns_empty(self):
+        self.assertEqual(builtin_fallback_models({"id": "nope", "name": "nope"}), [])
+
+    def test_none_provider_returns_empty(self):
+        self.assertEqual(builtin_fallback_models(None), [])
+
+
+class TestNormalizeModels(unittest.TestCase):
+    def test_syncpage_with_models(self):
+        page = FakeSyncPage([FakeModel("deepseek-chat"), FakeModel("deepseek-reasoner")])
+        out = normalize_models(page)
+        self.assertEqual([m["id"] for m in out], ["deepseek-chat", "deepseek-reasoner"])
+
+    def test_plain_list_of_models(self):
+        out = normalize_models([FakeModel("a"), FakeModel("b")])
+        self.assertEqual([m["id"] for m in out], ["a", "b"])
+
+    def test_empty_list_does_not_raise(self):
+        # 关键回归：旧代码对 [] 取 .data 会 AttributeError
+        self.assertEqual(normalize_models([]), [])
+
+    def test_list_of_dicts(self):
+        out = normalize_models([{"id": "x", "object": "model"}])
+        self.assertEqual(out[0]["id"], "x")
+
+    def test_drops_entries_without_id(self):
+        out = normalize_models([{"object": "model"}, {"id": "ok"}])
+        self.assertEqual([m["id"] for m in out], ["ok"])
+
+    def test_none_returns_empty(self):
+        self.assertEqual(normalize_models(None), [])
+
+
+class TestAsModelDicts(unittest.TestCase):
+    def test_shape_matches_sdk(self):
+        out = as_model_dicts(["deepseek-chat"], owned_by="DeepSeek")
+        self.assertEqual(out[0]["id"], "deepseek-chat")
+        self.assertEqual(out[0]["object"], "model")
+        self.assertEqual(out[0]["owned_by"], "DeepSeek")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_proxy_apply_env.py

@@ -0,0 +1,64 @@
+import importlib.util
+import os
+import pathlib
+import tempfile
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+MODULE_PATH = ROOT / "app" / "services" / "proxy_config_manager.py"
+spec = importlib.util.spec_from_file_location("proxy_config_manager", MODULE_PATH)
+if spec is None or spec.loader is None:
+    raise ImportError("proxy_config_manager module spec not found")
+pcm = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(pcm)
+ProxyConfigManager = pcm.ProxyConfigManager
+
+PROXY_ENV_KEYS = (
+    "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
+    "http_proxy", "https_proxy", "all_proxy",
+)
+
+
+class TestApplyToEnv(unittest.TestCase):
+    def setUp(self):
+        self._saved = {k: os.environ.get(k) for k in PROXY_ENV_KEYS}
+        for k in PROXY_ENV_KEYS:
+            os.environ.pop(k, None)
+        self._tmp = tempfile.TemporaryDirectory()
+        self.cfg_path = os.path.join(self._tmp.name, "proxy.json")
+
+    def tearDown(self):
+        for k, v in self._saved.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        self._tmp.cleanup()
+
+    def test_enabled_proxy_exported_to_env(self):
+        mgr = ProxyConfigManager(filepath=self.cfg_path)
+        mgr.update_config(enabled=True, url="http://127.0.0.1:7890")
+        returned = mgr.apply_to_env()
+        self.assertEqual(returned, "http://127.0.0.1:7890")
+        # huggingface_hub / requests 只认环境变量，必须 export 进去
+        for k in PROXY_ENV_KEYS:
+            self.assertEqual(os.environ.get(k), "http://127.0.0.1:7890", k)
+
+    def test_no_proxy_returns_none_and_no_env(self):
+        mgr = ProxyConfigManager(filepath=self.cfg_path)
+        returned = mgr.apply_to_env()
+        self.assertIsNone(returned)
+        for k in PROXY_ENV_KEYS:
+            self.assertIsNone(os.environ.get(k), k)
+
+    def test_env_fallback_is_idempotent(self):
+        # 没配文件代理但环境已有代理：apply 应把它补全到所有别名（含小写）
+        os.environ["HTTPS_PROXY"] = "http://10.0.0.1:1080"
+        mgr = ProxyConfigManager(filepath=self.cfg_path)
+        returned = mgr.apply_to_env()
+        self.assertEqual(returned, "http://10.0.0.1:1080")
+        self.assertEqual(os.environ.get("https_proxy"), "http://10.0.0.1:1080")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_transcriber_provider_model_selection.py

@@ -0,0 +1,76 @@
+"""Model-selection contract tests without loading the Whisper runtime."""
+import importlib.util
+import pathlib
+import sys
+import types
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+MODULE_PATH = ROOT / "app" / "transcriber" / "transcriber_provider.py"
+
+
+class _Logger:
+    def info(self, *_args, **_kwargs):
+        pass
+
+    def warning(self, *_args, **_kwargs):
+        pass
+
+    def error(self, *_args, **_kwargs):
+        pass
+
+
+class _Whisper:
+    def __init__(self, model_size, device):
+        self.model_size = model_size
+        self.device = device
+
+
+def _stub(monkeypatch, name, **attrs):
+    module = types.ModuleType(name)
+    for key, value in attrs.items():
+        setattr(module, key, value)
+    monkeypatch.setitem(sys.modules, name, module)
+
+
+def _load_provider(monkeypatch):
+    _stub(monkeypatch, "app")
+    _stub(monkeypatch, "app.transcriber")
+    _stub(monkeypatch, "app.utils")
+    _stub(monkeypatch, "app.transcriber.groq", GroqTranscriber=object)
+    _stub(monkeypatch, "app.transcriber.whisper", WhisperTranscriber=_Whisper)
+    _stub(monkeypatch, "app.transcriber.bcut", BcutTranscriber=object)
+    _stub(monkeypatch, "app.transcriber.kuaishou", KuaishouTranscriber=object)
+    _stub(monkeypatch, "app.utils.logger", get_logger=lambda _name: _Logger())
+
+    spec = importlib.util.spec_from_file_location("transcriber_provider_under_test", MODULE_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_explicit_whisper_model_overrides_docker_default(monkeypatch):
+    """A saved UI choice must not be replaced by WHISPER_MODEL_SIZE."""
+    provider = _load_provider(monkeypatch)
+    monkeypatch.setenv("WHISPER_MODEL_SIZE", "tiny")
+
+    transcriber = provider.get_transcriber(
+        transcriber_type="fast-whisper",
+        model_size="large-v3-turbo",
+        device="cpu",
+    )
+
+    assert transcriber.model_size == "large-v3-turbo"
+
+
+def test_switching_whisper_models_rebuilds_the_cached_instance(monkeypatch):
+    """Caching only by transcriber type would keep returning the first model."""
+    provider = _load_provider(monkeypatch)
+    monkeypatch.delenv("WHISPER_MODEL_SIZE", raising=False)
+
+    base = provider.get_transcriber("fast-whisper", model_size="base", device="cpu")
+    turbo = provider.get_transcriber("fast-whisper", model_size="large-v3-turbo", device="cpu")
+
+    assert turbo is not base
+    assert turbo.model_size == "large-v3-turbo"

backend/tests/test_url_normalize.py

@@ -0,0 +1,67 @@
+import pytest
+
+from app.utils.url_parser import normalize_video_url
+from app.validators.video_url_validator import VideoRequest
+
+
+def test_watchlater_url():
+    url = ("https://www.bilibili.com/list/watchlater/?bvid=BV1CPXpBYEui"
+           "&oid=116294762371214&spm_id_from=333.881.0.0&vd_source=abc")
+    assert normalize_video_url(url) == "https://www.bilibili.com/video/BV1CPXpBYEui"
+
+
+def test_favlist_url():
+    url = "https://www.bilibili.com/list/ml123456?bvid=BV1xx411c7mD&oid=987"
+    assert normalize_video_url(url) == "https://www.bilibili.com/video/BV1xx411c7mD"
+
+
+def test_tracking_params_stripped():
+    url = "https://www.bilibili.com/video/BV1xx411c7mD/?spm_id_from=333.881&vd_source=abc"
+    assert normalize_video_url(url) == "https://www.bilibili.com/video/BV1xx411c7mD"
+
+
+def test_p_number_preserved():
+    url = "https://www.bilibili.com/video/BV1xx411c7mD?p=36&spm_id_from=333.881"
+    assert normalize_video_url(url) == "https://www.bilibili.com/video/BV1xx411c7mD?p=36"
+
+
+def test_no_bv_returned_unchanged():
+    url = "https://www.bilibili.com/anime/timeline"
+    assert normalize_video_url(url) == url
+
+
+def test_b23_short_url_unchanged():
+    url = "https://b23.tv/abc123"
+    assert normalize_video_url(url) == url
+
+
+def test_video_request_accepts_watchlater():
+    req = VideoRequest(
+        url="https://www.bilibili.com/list/watchlater/?bvid=BV1CPXpBYEui&oid=116294762371214",
+        platform="bilibili",
+    )
+    assert str(req.url) == "https://www.bilibili.com/video/BV1CPXpBYEui"
+
+
+def test_video_request_rejects_no_bv():
+    with pytest.raises(ValueError):
+        VideoRequest(url="https://www.bilibili.com/anime/timeline", platform="bilibili")
+
+
+def test_note_router_request_accepts_watchlater():
+    # note.py 里的 VideoRequest 才是 /generate_note 实际使用的请求模型
+    from app.routers.note import VideoRequest as NoteVideoRequest
+
+    req = NoteVideoRequest(
+        video_url="https://www.bilibili.com/list/watchlater/?bvid=BV1CPXpBYEui&oid=116294762371214&spm_id_from=333.881.0.0",
+        platform="bilibili",
+        quality="fast",
+        model_name="test-model",
+        provider_id="test-provider",
+    )
+    assert req.video_url == "https://www.bilibili.com/video/BV1CPXpBYEui"
+
+
+def test_video_request_youtube_unaffected():
+    req = VideoRequest(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ", platform="youtube")
+    assert str(req.url) == "https://www.youtube.com/watch?v=dQw4w9WgXcQ"

backend/tests/test_whisper_config_forwarding.py

@@ -0,0 +1,120 @@
+import os
+import pathlib
+import subprocess
+import sys
+import textwrap
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+
+
+def _run_isolated(script: str) -> None:
+    env = os.environ.copy()
+    env["PYTHONPATH"] = os.pathsep.join(
+        filter(None, [str(ROOT), env.get("PYTHONPATH")])
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", textwrap.dedent(script)],
+        cwd=ROOT,
+        env=env,
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
+def test_note_generator_forwards_configured_whisper_model_size():
+    _run_isolated(
+        """
+        from app.services import note as note_service
+
+        calls = []
+
+        def fake_get_transcriber(**kwargs):
+            calls.append(kwargs)
+            return object()
+
+        note_service.get_transcriber = fake_get_transcriber
+
+        generator = note_service.NoteGenerator.__new__(note_service.NoteGenerator)
+        generator.transcriber_type = "fast-whisper"
+        generator.model_size = "large-v3-turbo"
+
+        generator._init_transcriber()
+
+        assert calls == [
+            {
+                "transcriber_type": "fast-whisper",
+                "model_size": "large-v3-turbo",
+            }
+        ]
+        """
+    )
+
+
+def test_whisper_cache_is_rebuilt_when_model_size_changes():
+    _run_isolated(
+        """
+        from app.transcriber import transcriber_provider as provider
+
+        class FakeWhisperTranscriber:
+            def __init__(self, model_size, device):
+                self.model_size = model_size
+                self.device = device
+
+        provider._transcribers = {key: None for key in provider._transcribers}
+        provider._transcriber_configs = {
+            key: None for key in provider._transcribers
+        }
+        provider.WhisperTranscriber = FakeWhisperTranscriber
+
+        base = provider.get_whisper_transcriber("base", device="cpu")
+        turbo = provider.get_whisper_transcriber("large-v3-turbo", device="cpu")
+        turbo_again = provider.get_whisper_transcriber(
+            "large-v3-turbo", device="cpu"
+        )
+
+        assert base.model_size == "base"
+        assert turbo.model_size == "large-v3-turbo"
+        assert turbo is not base
+        assert turbo_again is turbo
+        """
+    )
+
+
+def test_explicit_model_size_wins_over_environment_default():
+    _run_isolated(
+        """
+        import os
+
+        from app.transcriber import transcriber_provider as provider
+
+        class FakeWhisperTranscriber:
+            def __init__(self, model_size, device):
+                self.model_size = model_size
+                self.device = device
+
+        os.environ["WHISPER_MODEL_SIZE"] = "tiny"
+        provider._transcribers = {key: None for key in provider._transcribers}
+        provider._transcriber_configs = {
+            key: None for key in provider._transcribers
+        }
+        provider.WhisperTranscriber = FakeWhisperTranscriber
+
+        transcriber = provider.get_transcriber(
+            transcriber_type="fast-whisper",
+            model_size="large-v3-turbo",
+            device="cpu",
+        )
+
+        assert transcriber.model_size == "large-v3-turbo"
+
+        fallback = provider.get_transcriber(
+            transcriber_type="fast-whisper",
+            model_size=None,
+            device="cpu",
+        )
+
+        assert fallback.model_size == "tiny"
+        """
+    )

backend/tests/test_ydl_retry_behavior.py

@@ -0,0 +1,97 @@
+"""Regression coverage for the retry contract at the yt-dlp boundary."""
+import importlib.util
+import pathlib
+import sys
+import types
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+MODULE_PATH = ROOT / "app" / "downloaders" / "youtube_downloader.py"
+
+
+def _stub(monkeypatch, name, **attrs):
+    module = types.ModuleType(name)
+    for key, value in attrs.items():
+        setattr(module, key, value)
+    monkeypatch.setitem(sys.modules, name, module)
+    return module
+
+
+class _Downloader:
+    def __init__(self):
+        self.cache_data = "/tmp"
+
+
+class _AudioDownloadResult:
+    def __init__(self, **kwargs):
+        self.__dict__.update(kwargs)
+
+
+def _load_downloader_module(monkeypatch):
+    _stub(monkeypatch, "app")
+    _stub(monkeypatch, "app.downloaders")
+    _stub(monkeypatch, "app.models")
+    _stub(monkeypatch, "app.services")
+    _stub(monkeypatch, "app.utils")
+    _stub(
+        monkeypatch,
+        "app.downloaders.base",
+        Downloader=_Downloader,
+        DownloadQuality=str,
+        YDL_RETRY_OPTS={"retries": 3, "fragment_retries": 3, "socket_timeout": 30},
+    )
+    _stub(monkeypatch, "app.downloaders.youtube_subtitle", YouTubeSubtitleFetcher=object)
+    _stub(monkeypatch, "app.models.notes_model", AudioDownloadResult=_AudioDownloadResult)
+    _stub(monkeypatch, "app.models.transcriber_model", TranscriptResult=object)
+    _stub(
+        monkeypatch,
+        "app.services.proxy_config_manager",
+        ProxyConfigManager=type("ProxyConfigManager", (), {"get_proxy_url": lambda self: None}),
+    )
+    _stub(monkeypatch, "app.utils.path_helper", get_data_dir=lambda: "/tmp")
+    _stub(monkeypatch, "app.utils.url_parser", extract_video_id=lambda url, platform: "video-id")
+
+    spec = importlib.util.spec_from_file_location("youtube_downloader_under_test", MODULE_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+class _CapturingYoutubeDL:
+    options = None
+
+    def __init__(self, options):
+        type(self).options = options
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        return False
+
+    def extract_info(self, _url, download=True):
+        return {
+            "id": "video-id",
+            "title": "Example",
+            "duration": 1,
+            "thumbnail": None,
+            "ext": "m4a",
+            "tags": [],
+        }
+
+
+def test_audio_download_configures_a_nonzero_retry_budget(tmp_path, monkeypatch):
+    """Removing retry options must make transient download errors unprotected."""
+    module = _load_downloader_module(monkeypatch)
+    original_youtube_dl = module.yt_dlp.YoutubeDL
+    module.yt_dlp.YoutubeDL = _CapturingYoutubeDL
+    try:
+        module.YoutubeDownloader().download("https://youtu.be/example", str(tmp_path))
+    finally:
+        module.yt_dlp.YoutubeDL = original_youtube_dl
+
+    options = _CapturingYoutubeDL.options
+    assert options["retries"] > 0
+    assert options["fragment_retries"] > 0
+    assert options["socket_timeout"] > 0

backend/tests/test_ydl_retry_opts.py

@@ -0,0 +1,125 @@
+"""
+Coverage for the yt-dlp retry settings shared by the downloaders.
+
+Background: yt-dlp's documented `retries` default of 10 comes from its *command
+line* option parser. Nothing applies that default to the Python API, so a
+`YoutubeDL({...})` built without `retries` ends up in:
+
+    # yt_dlp/downloader/http.py
+    for retry in RetryManager(self.params.get('retries'), ...)   # -> None
+    # yt_dlp/utils/_utils.py
+    self.retries = _retries or 0                                 # -> 0
+
+i.e. exactly one attempt and no retries. A single transient network hiccup
+(observed: read timeout from upos-sz-mirrorcosov.bilivideo.com) then fails the
+whole note task, even though an immediate re-run succeeds.
+
+These tests pin both halves of the fix: the constant produces a real retry
+budget, and every yt-dlp options dict in the downloaders actually carries it.
+"""
+import ast
+import importlib.util
+import pathlib
+import sys
+import types
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+DOWNLOADERS = ROOT / "app" / "downloaders"
+DOWNLOADER_SOURCES = ["youtube_downloader.py", "bilibili_downloader.py"]
+
+
+def _load_base():
+    """Load app/downloaders/base.py with its app-level imports stubbed out."""
+    for name, attrs in {
+        "app": {},
+        "app.enmus": {},
+        "app.models": {},
+        "app.enmus.note_enums": {"DownloadQuality": str},
+        "app.models.notes_model": {"AudioDownloadResult": object},
+        "app.models.transcriber_model": {"TranscriptResult": object},
+    }.items():
+        module = types.ModuleType(name)
+        for key, value in attrs.items():
+            setattr(module, key, value)
+        sys.modules.setdefault(name, module)
+
+    spec = importlib.util.spec_from_file_location("dl_base", DOWNLOADERS / "base.py")
+    if spec is None or spec.loader is None:
+        raise ImportError("base module spec not found")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class RetryOptsValueTest(unittest.TestCase):
+    def setUp(self):
+        self.base = _load_base()
+
+    def test_retry_budget_is_not_zero(self):
+        try:
+            from yt_dlp.utils import RetryManager
+        except Exception as exc:  # pragma: no cover - env without yt-dlp
+            self.skipTest(f"yt-dlp not importable: {exc}")
+
+        retries = self.base.YDL_RETRY_OPTS["retries"]
+        budget = RetryManager(retries, lambda *a, **k: None).retries
+        self.assertGreater(budget, 0)
+
+    def test_documents_the_zero_default_being_guarded_against(self):
+        """The bug this guards: an unset `retries` collapses to a 0 budget."""
+        try:
+            from yt_dlp.utils import RetryManager
+        except Exception as exc:  # pragma: no cover - env without yt-dlp
+            self.skipTest(f"yt-dlp not importable: {exc}")
+
+        self.assertEqual(RetryManager(None, lambda *a, **k: None).retries, 0)
+
+    def test_socket_timeout_is_bounded(self):
+        # Without a bound, a stalled read can hang a task instead of failing
+        # fast enough for the retries above to be useful.
+        timeout = self.base.YDL_RETRY_OPTS["socket_timeout"]
+        self.assertGreater(timeout, 0)
+
+
+class RetryOptsAreAppliedTest(unittest.TestCase):
+    """
+    Structural check: every `ydl_opts = {...}` literal in the downloaders must
+    unpack YDL_RETRY_OPTS. Catches a newly added download path that silently
+    goes back to the zero-retry default.
+    """
+
+    def _ydl_opts_dicts(self, path):
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        found = []
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Dict):
+                continue
+            names = [t.id for t in node.targets if isinstance(t, ast.Name)]
+            if "ydl_opts" in names:
+                found.append(node.value)
+        return found
+
+    def test_every_ydl_opts_dict_unpacks_retry_opts(self):
+        for filename in DOWNLOADER_SOURCES:
+            path = DOWNLOADERS / filename
+            dicts = self._ydl_opts_dicts(path)
+            self.assertTrue(dicts, f"no ydl_opts dict found in {filename}")
+
+            for index, node in enumerate(dicts):
+                with self.subTest(file=filename, dict_index=index, line=node.lineno):
+                    unpacked = {
+                        value.id
+                        for key, value in zip(node.keys, node.values)
+                        if key is None and isinstance(value, ast.Name)
+                    }
+                    self.assertIn(
+                        "YDL_RETRY_OPTS",
+                        unpacked,
+                        f"{filename}:{node.lineno} builds yt-dlp options without "
+                        f"YDL_RETRY_OPTS, so it gets zero retries",
+                    )
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_youtube_metadata_only.py

@@ -0,0 +1,155 @@
+"""
+Coverage for the YouTube "metadata only" download path.
+
+Background: when a YouTube video already has subtitles, NoteGenerator skips the
+audio download and calls `YoutubeDownloader.download(skip_download=True)` purely
+to read title/duration/cover. That call used to still request
+`format='bestaudio[ext=m4a]/bestaudio/best'`.
+
+Whenever the installed yt-dlp lags behind YouTube's player, nsig extraction
+fails, every audio/video format is dropped (only storyboard images remain) and
+format selection raises "Requested format is not available" — killing a task
+whose transcript had already been fetched successfully.
+
+These tests pin the two guarantees of that path:
+  1. skip_download implies ignore_no_formats_error, so a formatless extraction
+     degrades to "no audio" instead of failing the whole note.
+  2. ext falls back to m4a, since yt-dlp reports ext=None when skipping the
+     download (dict.get's default does not fire on an explicit None).
+"""
+import importlib.util
+import pathlib
+import sys
+import types
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+MODULE_PATH = ROOT / "app" / "downloaders" / "youtube_downloader.py"
+
+
+def _stub(name, **attrs):
+    module = types.ModuleType(name)
+    for key, value in attrs.items():
+        setattr(module, key, value)
+    # Tests in this repository load downloader modules with lightweight app
+    # stubs. Replace a previous test's stub instead of inheriting it; otherwise
+    # a prior import can leave AudioDownloadResult as ``object`` and turn this
+    # metadata-only test into a TypeError unrelated to the behavior under test.
+    sys.modules[name] = module
+    return module
+
+
+class _Downloader:
+    def __init__(self):
+        self.cache_data = "/tmp"
+
+
+class _AudioDownloadResult:
+    def __init__(self, **kwargs):
+        self.__dict__.update(kwargs)
+
+
+def _load_youtube_downloader():
+    """Load the module with its app-level dependencies stubbed out."""
+    _stub("app")
+    _stub("app.downloaders")
+    _stub("app.models")
+    _stub("app.services")
+    _stub("app.utils")
+    _stub(
+        "app.downloaders.base",
+        Downloader=_Downloader,
+        DownloadQuality=str,
+        YDL_RETRY_OPTS={"retries": 3, "fragment_retries": 3, "socket_timeout": 30},
+    )
+    _stub("app.downloaders.youtube_subtitle", YouTubeSubtitleFetcher=object)
+    _stub("app.models.notes_model", AudioDownloadResult=_AudioDownloadResult)
+    _stub("app.models.transcriber_model", TranscriptResult=object)
+    _stub(
+        "app.services.proxy_config_manager",
+        ProxyConfigManager=type(
+            "ProxyConfigManager", (), {"get_proxy_url": lambda self: None}
+        ),
+    )
+    _stub("app.utils.path_helper", get_data_dir=lambda: "/tmp")
+    _stub("app.utils.url_parser", extract_video_id=lambda url, platform: "vid")
+
+    spec = importlib.util.spec_from_file_location("youtube_downloader", MODULE_PATH)
+    if spec is None or spec.loader is None:
+        raise ImportError("youtube_downloader module spec not found")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class _FakeYoutubeDL:
+    """Records the opts it was constructed with; mimics a formatless extraction."""
+
+    captured_opts = None
+
+    def __init__(self, opts):
+        type(self).captured_opts = opts
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        return False
+
+    def extract_info(self, url, download=True):
+        # What yt-dlp yields for a metadata-only extraction: no media, so no ext.
+        return {
+            "id": "CJ4ndXv3CkY",
+            "title": "example",
+            "duration": 2231,
+            "thumbnail": "https://example.invalid/t.jpg",
+            "ext": None,
+            "tags": [],
+        }
+
+
+class YoutubeMetadataOnlyTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        try:
+            cls.module = _load_youtube_downloader()
+        except Exception as exc:  # pragma: no cover - env without yt-dlp
+            raise unittest.SkipTest(f"youtube_downloader not importable: {exc}")
+
+    def setUp(self):
+        self._real_ydl = self.module.yt_dlp.YoutubeDL
+        self.module.yt_dlp.YoutubeDL = _FakeYoutubeDL
+        _FakeYoutubeDL.captured_opts = None
+
+    def tearDown(self):
+        self.module.yt_dlp.YoutubeDL = self._real_ydl
+
+    def test_skip_download_tolerates_missing_formats(self):
+        self.module.YoutubeDownloader().download(
+            "https://www.youtube.com/watch?v=CJ4ndXv3CkY",
+            output_dir="/tmp",
+            skip_download=True,
+        )
+        self.assertTrue(_FakeYoutubeDL.captured_opts.get("ignore_no_formats_error"))
+
+    def test_missing_ext_falls_back_to_m4a(self):
+        result = self.module.YoutubeDownloader().download(
+            "https://www.youtube.com/watch?v=CJ4ndXv3CkY",
+            output_dir="/tmp",
+            skip_download=True,
+        )
+        self.assertTrue(result.file_path.endswith(".m4a"), result.file_path)
+        self.assertNotIn("None", result.file_path)
+
+    def test_full_download_still_selects_an_audio_format(self):
+        self.module.YoutubeDownloader().download(
+            "https://www.youtube.com/watch?v=CJ4ndXv3CkY",
+            output_dir="/tmp",
+        )
+        opts = _FakeYoutubeDL.captured_opts
+        self.assertIn("bestaudio", opts.get("format", ""))
+        self.assertNotIn("ignore_no_formats_error", opts)
+
+
+if __name__ == "__main__":
+    unittest.main()