fix(worker): 修复 dedup fingerprint JSON 序列化失败 - np.float32 转原生 float

根因：cv2.normalize() 返回 np.float32 数组，color_histograms 直接存进 dict 后
SQLAlchemy JSON 序列化时报 'float32 is not JSON serializable'

修复：VideoFingerprint.to_dict() 中统一转 Python 原生类型
- color_histograms: np.float32 → float
- duration: 统一转 float
- resolution: np.int32 → int

apps/api/app/api/router.py

@@ -13,7 +13,6 @@ from app.api.routes.generated_videos import router as generated_videos_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.health import router as health_check_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
-from app.api.routes.internal_render import router as internal_render_router
 from app.api.routes.jobs import router as jobs_router
 from app.api.routes.projects import router as projects_router
 from app.api.routes.recipes import router as recipes_router
@@ -157,7 +156,3 @@ api_router.include_router(
     feature_flags_router,
     tags=["Internal"],
 )
-api_router.include_router(
-    internal_render_router,
-    tags=["Internal"],
-)

apps/api/app/api/routes/internal_render.py

@@ -1,120 +0,0 @@
-"""渲染结果内部下载接口。
-
-通过内部 API Key 鉴权，为灰度对比工具等内部系统提供渲染结果下载能力。
-
-API:
-  GET  /api/v1/internal/render/videos/{video_id}/download-url  - 获取单个视频下载URL
-  GET  /api/v1/internal/render/tasks/{task_id}/videos          - 获取任务下所有视频及下载URL
-
-鉴权：X-API-Key header，走内部 API Key 验证
-"""
-
-from __future__ import annotations
-
-import logging
-from typing import Any
-
-from app.api.routes.auth import _verify_internal_api_key
-from app.core.storage import OSSStorageService, get_storage_service
-from app.dependencies import get_generated_video_repository
-from fastapi import APIRouter, Depends, HTTPException, Query
-from pydantic import BaseModel
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter(prefix="/internal/render", tags=["Internal"])
-
-
-class InternalRenderVideoItem(BaseModel):
-    """内部渲染视频项。"""
-
-    video_id: str
-    generation_task_id: str
-    project_id: str
-    name: str
-    file_url: str
-    file_size: int | None = None
-    duration: float | None = None
-    width: int | None = None
-    height: int | None = None
-    fps: float | None = None
-    status: str
-    download_url: str
-
-
-class InternalRenderTaskVideosResponse(BaseModel):
-    """任务下所有渲染视频响应。"""
-
-    task_id: str
-    count: int
-    videos: list[InternalRenderVideoItem]
-
-
-class InternalRenderDownloadUrlResponse(BaseModel):
-    """单个视频下载URL响应。"""
-
-    video_id: str
-    download_url: str
-
-
-def _video_to_item(video: Any, download_url: str) -> InternalRenderVideoItem:
-    """将 GeneratedVideo 领域对象转为响应项。"""
-    return InternalRenderVideoItem(
-        video_id=video.id,
-        generation_task_id=video.generation_task_id,
-        project_id=video.project_id,
-        name=video.name,
-        file_url=video.file_url,
-        file_size=getattr(video, "file_size", None),
-        duration=getattr(video, "duration", None),
-        width=getattr(video, "width", None),
-        height=getattr(video, "height", None),
-        fps=getattr(video, "fps", None),
-        status=video.status,
-        download_url=download_url,
-    )
-
-
-@router.get("/videos/{video_id}/download-url", response_model=InternalRenderDownloadUrlResponse)
-def get_render_video_download_url(
-    video_id: str,
-    _: bool = Depends(_verify_internal_api_key),
-    generated_video_repository: Any = Depends(get_generated_video_repository),
-    storage_service: OSSStorageService = Depends(get_storage_service),
-) -> InternalRenderDownloadUrlResponse:
-    """获取单个渲染视频的下载URL（预签名）。"""
-    video = generated_video_repository.get(video_id)
-    if video is None:
-        raise HTTPException(status_code=404, detail=f"GeneratedVideo {video_id} not found")
-
-    download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
-    logger.info("内部渲染下载URL生成: video_id=%s", video_id)
-    return InternalRenderDownloadUrlResponse(video_id=video_id, download_url=download_url)
-
-
-@router.get("/tasks/{task_id}/videos", response_model=InternalRenderTaskVideosResponse)
-def get_render_task_videos(
-    task_id: str,
-    status: str | None = Query(None, description="按状态筛选，如 completed/failed"),
-    _: bool = Depends(_verify_internal_api_key),
-    generated_video_repository: Any = Depends(get_generated_video_repository),
-    storage_service: OSSStorageService = Depends(get_storage_service),
-) -> InternalRenderTaskVideosResponse:
-    """获取生成任务下所有渲染视频及下载URL。"""
-    videos = generated_video_repository.list_by_generation_task(task_id)
-
-    # 状态筛选
-    if status:
-        videos = [v for v in videos if v.status == status]
-
-    items = []
-    for video in videos:
-        download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
-        items.append(_video_to_item(video, download_url))
-
-    logger.info("内部渲染任务视频查询: task_id=%s count=%d", task_id, len(items))
-    return InternalRenderTaskVideosResponse(
-        task_id=task_id,
-        count=len(items),
-        videos=items,
-    )

apps/worker/video_processing/oss_helpers.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 import hashlib
 import logging
 import os
-import threading
 from pathlib import Path
 from typing import Optional
 from urllib.parse import urlparse
@@ -18,13 +17,6 @@ import oss2
 
 logger = logging.getLogger(__name__)
 
-# OSS 上传配置
-OSS_CONNECT_TIMEOUT = 10  # 连接超时（秒），防止 TCP 握手挂死
-OSS_UPLOAD_TOTAL_TIMEOUT = 300  # 单文件上传总超时（秒），防止网络慢时无限卡住
-OSS_MULTIPART_THRESHOLD = 100 * 1024 * 1024  # 分片上传阈值：100MB 以上走分片
-OSS_PART_SIZE = 8 * 1024 * 1024  # 分片大小：8MB
-OSS_MULTIPART_NUM_THREADS = 3  # 分片上传并发数
-
 
 # ── OSS 配置 ──────────────────────────────────────────────────────────────────
 
@@ -51,9 +43,6 @@ def oss_bucket() -> oss2.Bucket | None:
     P0-2 修复：endpoint 不带 scheme 时自动补 https:// 前缀，
     确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。
 
-    P0-staging 修复：增加 connect_timeout=10s，防止网络抖动时
-    TCP 握手阶段无限挂死，导致 worker 进程卡死。
-
     Returns:
         oss2.Bucket 实例，配置缺失时返回 None。
     """
@@ -64,12 +53,7 @@ def oss_bucket() -> oss2.Bucket | None:
     # endpoint 无 scheme 时补 https://，与 API 端 storage.py 保持一致
     if not endpoint.startswith(("http://", "https://")):
         endpoint = f"https://{endpoint}"
-    return oss2.Bucket(
-        oss2.Auth(access_key_id, access_key_secret),
-        endpoint,
-        bucket_name,
-        connect_timeout=OSS_CONNECT_TIMEOUT,
-    )
+    return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
 
 
 def normalize_storage_key(storage_key_or_url: str) -> str:
@@ -112,9 +96,6 @@ def download_asset(asset_storage_key: str, local_path: Path) -> bool:
 def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     """上传文件到 OSS，返回公开 URL。
 
-    大文件（>100MB）自动走分片上传，降低内存峰值，减少 OOM 风险。
-    上传加总超时保护（默认 300s），防止网络异常时无限挂死。
-
     Args:
         local_path: 本地文件路径
         storage_key: 目标存储键
@@ -125,71 +106,18 @@ def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     bucket = oss_bucket()
     if bucket is None:
         return None
-
-    result: dict = {"url": None, "error": None, "file_size": 0}
-    done = threading.Event()
-
-    def _do_upload():
-        try:
-            # 尝试获取文件大小，用于分片判断和日志；stat 失败时 fallback 走普通上传
-            try:
-                file_size = local_path.stat().st_size
-                result["file_size"] = file_size
-                use_multipart = file_size >= OSS_MULTIPART_THRESHOLD
-            except OSError:
-                use_multipart = False
-                file_size = 0
-
-            if use_multipart:
-                # 分片上传：降低内存峰值，每片 8MB，3 线程并发
-                logger.info(
-                    "大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d",
-                    storage_key[:80],
-                    file_size / 1024 / 1024,
-                    OSS_PART_SIZE // 1024 // 1024,
-                    OSS_MULTIPART_NUM_THREADS,
-                )
-                oss2.resumable_upload(
-                    bucket,
-                    storage_key,
-                    str(local_path),
-                    multipart_threshold=OSS_MULTIPART_THRESHOLD,
-                    part_size=OSS_PART_SIZE,
-                    num_threads=OSS_MULTIPART_NUM_THREADS,
-                )
-            else:
-                bucket.put_object_from_file(storage_key, str(local_path))
-
-            # 构造返回 URL
-            settings = oss_settings()
-            if settings:
-                _, _, endpoint, bucket_name = settings
-                endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
-                result["url"] = f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
-        except Exception as e:
-            result["error"] = e
-            logger.exception("上传 OSS 失败: %s", storage_key)
-        finally:
-            done.set()
-
-    upload_thread = threading.Thread(target=_do_upload, daemon=True)
-    upload_thread.start()
-    finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT)
-
-    if not finished:
-        logger.error(
-            "OSS 上传超时（%.0fs），强制中止: storage_key=%s, size=%.1fMB",
-            OSS_UPLOAD_TOTAL_TIMEOUT,
-            storage_key[:80],
-            result["file_size"] / 1024 / 1024 if result["file_size"] else 0,
-        )
+    try:
+        bucket.put_object_from_file(storage_key, str(local_path))
+        settings = oss_settings()
+        if settings:
+            _, _, endpoint, bucket_name = settings
+            endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
+            return f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
         return None
-
-    if result["error"]:
+    except Exception:
+        logger.exception("上传 OSS 失败: %s", storage_key)
         return None
 
-    return result["url"]
-
 
 def get_signed_download_url(storage_key_or_url: str, expires_seconds: int = 3600) -> str | None:
     """生成预签名下载 URL（用于私有 bucket 的 URL 校验或临时下载）。

tests/unit/test_internal_render_download.py

@@ -1,190 +0,0 @@
-"""渲染结果内部下载接口单元测试。
-
-测试 internal_render 路由的核心逻辑，mock 掉 repository 和 storage 依赖。
-"""
-
-from __future__ import annotations
-
-from unittest.mock import MagicMock
-
-import pytest
-from app.api.routes.internal_render import (
-    InternalRenderDownloadUrlResponse,
-    InternalRenderTaskVideosResponse,
-    _video_to_item,
-    get_render_task_videos,
-    get_render_video_download_url,
-)
-
-# ── Helpers ────────────────────────────────────────────────────────────────
-
-
-class MockVideo:
-    """模拟 GeneratedVideo 领域对象。"""
-
-    def __init__(self, **kwargs):
-        self.id = kwargs.get("id", "video-1")
-        self.generation_task_id = kwargs.get("generation_task_id", "task-1")
-        self.project_id = kwargs.get("project_id", "proj-1")
-        self.name = kwargs.get("name", "test_video.mp4")
-        self.file_url = kwargs.get("file_url", "videos/test/output.mp4")
-        self.file_size = kwargs.get("file_size", 1024000)
-        self.duration = kwargs.get("duration", 30.5)
-        self.width = kwargs.get("width", 1080)
-        self.height = kwargs.get("height", 1920)
-        self.fps = kwargs.get("fps", 30.0)
-        self.status = kwargs.get("status", "completed")
-
-
-# ── _video_to_item 测试 ────────────────────────────────────────────────────
-
-
-class TestVideoToItem:
-    """测试视频对象转响应项。"""
-
-    def test_basic_conversion(self):
-        video = MockVideo(id="v1", generation_task_id="t1", status="completed")
-        item = _video_to_item(video, "https://oss.example.com/download?v1")
-        assert item.video_id == "v1"
-        assert item.generation_task_id == "t1"
-        assert item.status == "completed"
-        assert item.download_url == "https://oss.example.com/download?v1"
-
-    def test_missing_optional_fields(self):
-        """缺可选字段时返回 None。"""
-        video = MockVideo()
-        # 去掉可选字段
-        del video.file_size
-        del video.duration
-        item = _video_to_item(video, "https://example.com/dl")
-        assert item.file_size is None
-        assert item.duration is None
-        assert item.width == 1080  # 还在
-
-
-# ── 路由函数测试 ────────────────────────────────────────────────────────────
-
-
-class TestGetRenderVideoDownloadUrl:
-    """测试单个视频下载URL接口。"""
-
-    def test_video_exists(self):
-        video = MockVideo(id="v-abc", file_url="videos/abc/out.mp4")
-        mock_repo = MagicMock()
-        mock_repo.get.return_value = video
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://oss.test/signed?v=abc"
-
-        result = get_render_video_download_url(
-            video_id="v-abc",
-            _=True,
-            generated_video_repository=mock_repo,
-            storage_service=mock_storage,
-        )
-
-        assert isinstance(result, InternalRenderDownloadUrlResponse)
-        assert result.video_id == "v-abc"
-        assert result.download_url == "https://oss.test/signed?v=abc"
-        mock_repo.get.assert_called_once_with("v-abc")
-        mock_storage.get_download_url.assert_called_once()
-
-    def test_video_not_found_raises_404(self):
-        from fastapi import HTTPException
-
-        mock_repo = MagicMock()
-        mock_repo.get.return_value = None
-        mock_storage = MagicMock()
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_render_video_download_url(
-                video_id="nonexistent",
-                _=True,
-                generated_video_repository=mock_repo,
-                storage_service=mock_storage,
-            )
-        assert exc_info.value.status_code == 404
-
-    def test_download_url_long_expiry(self):
-        """过期时间应为 24 小时（86400s）。"""
-        video = MockVideo(id="v1")
-        mock_repo = MagicMock()
-        mock_repo.get.return_value = video
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://oss.test/signed"
-
-        get_render_video_download_url(
-            video_id="v1",
-            _=True,
-            generated_video_repository=mock_repo,
-            storage_service=mock_storage,
-        )
-
-        # 验证 expires_seconds=86400
-        call_kwargs = mock_storage.get_download_url.call_args
-        assert call_kwargs.kwargs.get("expires_seconds") == 86400 or call_kwargs[1].get("expires_seconds") == 86400
-
-
-class TestGetRenderTaskVideos:
-    """测试任务视频列表接口。"""
-
-    def test_list_multiple_videos(self):
-        videos = [
-            MockVideo(id="v1", status="completed"),
-            MockVideo(id="v2", status="completed"),
-            MockVideo(id="v3", status="failed"),
-        ]
-        mock_repo = MagicMock()
-        mock_repo.list_by_generation_task.return_value = videos
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://oss.test/signed"
-
-        result = get_render_task_videos(
-            task_id="task-1",
-            status=None,
-            _=True,
-            generated_video_repository=mock_repo,
-            storage_service=mock_storage,
-        )
-
-        assert isinstance(result, InternalRenderTaskVideosResponse)
-        assert result.task_id == "task-1"
-        assert result.count == 3
-        assert len(result.videos) == 3
-
-    def test_filter_by_status(self):
-        videos = [
-            MockVideo(id="v1", status="completed"),
-            MockVideo(id="v2", status="completed"),
-            MockVideo(id="v3", status="failed"),
-        ]
-        mock_repo = MagicMock()
-        mock_repo.list_by_generation_task.return_value = videos
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://oss.test/signed"
-
-        result = get_render_task_videos(
-            task_id="task-1",
-            status="completed",
-            _=True,
-            generated_video_repository=mock_repo,
-            storage_service=mock_storage,
-        )
-
-        assert result.count == 2
-        assert all(v.status == "completed" for v in result.videos)
-
-    def test_empty_task(self):
-        mock_repo = MagicMock()
-        mock_repo.list_by_generation_task.return_value = []
-        mock_storage = MagicMock()
-
-        result = get_render_task_videos(
-            task_id="empty-task",
-            status=None,
-            _=True,
-            generated_video_repository=mock_repo,
-            storage_service=mock_storage,
-        )
-
-        assert result.count == 0
-        assert result.videos == []

tests/unit/test_oss_upload_crash_fix.py

@@ -1,236 +0,0 @@
-"""P0-staging：OSS 上传崩溃修复测试.
-
-测试：
-1. oss_bucket() 传递 connect_timeout 参数
-2. upload_to_oss() 小文件走 put_object_from_file，大文件走分片上传
-3. upload_to_oss() 超时保护（超过总超时返回 None）
-4. upload_to_oss() 异常时返回 None
-"""
-
-from __future__ import annotations
-
-import os
-import tempfile
-import time
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-# ── oss_bucket connect_timeout 测试 ───────────────────────────────────────────
-
-
-class TestOSSBucketConnectTimeout:
-    """测试 oss_bucket() 传递 connect_timeout 参数."""
-
-    def test_oss_bucket_has_connect_timeout(self):
-        """oss_bucket 应传递 connect_timeout=10s 参数."""
-        from video_processing.oss_helpers import oss_bucket
-
-        mock_bucket_instance = MagicMock()
-        with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
-            ),
-            patch("video_processing.oss_helpers.oss2.Auth"),
-            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
-        ):
-            bucket = oss_bucket()
-
-            assert bucket is mock_bucket_instance
-            # 验证 connect_timeout 关键字参数
-            call_kwargs = mock_bucket_cls.call_args[1]
-            assert "connect_timeout" in call_kwargs, "oss_bucket 应传递 connect_timeout 参数"
-            assert (
-                call_kwargs["connect_timeout"] == 10
-            ), f"connect_timeout 应为 10，实际为 {call_kwargs['connect_timeout']}"
-
-    def test_oss_bucket_no_config_returns_none(self):
-        """OSS 配置缺失时返回 None."""
-        from video_processing.oss_helpers import oss_bucket
-
-        with patch.dict(os.environ, {}, clear=True):
-            bucket = oss_bucket()
-            assert bucket is None
-
-
-# ── upload_to_oss 分片上传测试 ────────────────────────────────────────────────
-
-
-class TestUploadToOSSMultipart:
-    """测试 upload_to_oss() 根据文件大小选择上传方式."""
-
-    def _create_temp_file(self, size_bytes: int) -> Path:
-        """创建指定大小的临时文件."""
-        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
-        tmp.write(b"x" * size_bytes)
-        tmp.close()
-        return Path(tmp.name)
-
-    def test_small_file_uses_put_object(self):
-        """小文件（<100MB）走 put_object_from_file."""
-        from video_processing.oss_helpers import upload_to_oss
-
-        small_file = self._create_temp_file(10 * 1024 * 1024)  # 10MB
-        try:
-            mock_bucket = MagicMock()
-
-            with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
-                ),
-                patch("video_processing.oss_helpers.oss2.Auth"),
-                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
-                patch("video_processing.oss_helpers.oss2.resumable_upload") as mock_resumable,
-            ):
-                url = upload_to_oss(small_file, "test/small.mp4")
-
-                # 验证调用了 put_object_from_file
-                mock_bucket.put_object_from_file.assert_called_once()
-                # 验证没调用分片上传
-                mock_resumable.assert_not_called()
-                # 验证返回 URL
-                assert url == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/test/small.mp4"
-        finally:
-            small_file.unlink()
-
-    def test_large_file_uses_resumable_upload(self):
-        """大文件（>=100MB）走 resumable_upload 分片上传."""
-        from video_processing.oss_helpers import upload_to_oss
-
-        large_file = self._create_temp_file(100 * 1024 * 1024)  # 100MB
-        try:
-            mock_bucket = MagicMock()
-
-            with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
-                ),
-                patch("video_processing.oss_helpers.oss2.Auth"),
-                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
-                patch("video_processing.oss_helpers.oss2.resumable_upload") as mock_resumable,
-            ):
-                url = upload_to_oss(large_file, "test/large.mp4")
-
-                # 验证调用了分片上传
-                mock_resumable.assert_called_once()
-                # 验证没调用 put_object_from_file
-                mock_bucket.put_object_from_file.assert_not_called()
-                # 验证分片参数
-                call_kwargs = mock_resumable.call_args[1]
-                assert call_kwargs["multipart_threshold"] == 100 * 1024 * 1024
-                assert call_kwargs["part_size"] == 8 * 1024 * 1024
-                assert call_kwargs["num_threads"] == 3
-                # 验证返回 URL
-                assert url == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/test/large.mp4"
-        finally:
-            large_file.unlink()
-
-
-# ── upload_to_oss 超时测试 ────────────────────────────────────────────────────
-
-
-class TestUploadToOSSTimeout:
-    """测试 upload_to_oss() 超时保护."""
-
-    def test_upload_timeout_returns_none(self):
-        """上传超过总超时时返回 None."""
-        from video_processing.oss_helpers import upload_to_oss
-
-        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
-        small_file.write(b"x" * 1024)  # 1KB
-        small_file.close()
-        file_path = Path(small_file.name)
-
-        def slow_upload(*args, **kwargs):
-            """模拟慢速上传，超过超时时间."""
-            time.sleep(2)
-
-        mock_bucket = MagicMock()
-        mock_bucket.put_object_from_file.side_effect = slow_upload
-
-        try:
-            with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
-                ),
-                patch("video_processing.oss_helpers.oss2.Auth"),
-                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
-                patch("video_processing.oss_helpers.OSS_UPLOAD_TOTAL_TIMEOUT", 1),  # 1秒超时
-            ):
-                url = upload_to_oss(file_path, "test/slow.mp4")
-                # 超时应返回 None
-                assert url is None, "上传超时应返回 None"
-        finally:
-            file_path.unlink()
-
-    def test_upload_exception_returns_none(self):
-        """上传异常时返回 None."""
-        from video_processing.oss_helpers import upload_to_oss
-
-        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
-        small_file.write(b"x" * 1024)
-        small_file.close()
-        file_path = Path(small_file.name)
-
-        mock_bucket = MagicMock()
-        mock_bucket.put_object_from_file.side_effect = RuntimeError("Network error")
-
-        try:
-            with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
-                ),
-                patch("video_processing.oss_helpers.oss2.Auth"),
-                patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
-            ):
-                url = upload_to_oss(file_path, "test/error.mp4")
-                assert url is None, "上传异常应返回 None"
-        finally:
-            file_path.unlink()
-
-    def test_upload_no_bucket_returns_none(self):
-        """OSS 未配置时返回 None."""
-        from video_processing.oss_helpers import upload_to_oss
-
-        small_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
-        small_file.write(b"x" * 1024)
-        small_file.close()
-        file_path = Path(small_file.name)
-
-        try:
-            with patch.dict(os.environ, {}, clear=True):
-                url = upload_to_oss(file_path, "test/noconfig.mp4")
-                assert url is None
-        finally:
-            file_path.unlink()