diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py
index 6ba88ac7a..20fff48d4 100755
--- a/apps/worker/video_processing/oss_helpers.py
+++ b/apps/worker/video_processing/oss_helpers.py
@@ -14,6 +14,7 @@ from pathlib import Path
 from urllib.parse import urlparse
 
 import oss2
+import requests
 
 from packages.shared.config import get_shared_settings
 
@@ -96,13 +97,22 @@ def normalize_storage_key(storage_key_or_url: str) -> str:
 def download_asset(asset_storage_key: str, local_path: Path) -> bool:
     """从 OSS 下载素材文件到本地路径。
 
+    自动识别输入类型：
+    - 完整 URL（http:// 或 https:// 开头）→ 走 HTTP 下载（支持预签名URL）
+    - OSS 存储键 → 走 oss2 SDK 下载
+
     Args:
-        asset_storage_key: 素材的存储键（或完整 URL）
+        asset_storage_key: 素材的存储键或完整 URL
         local_path: 本地保存路径
 
     Returns:
         True 表示下载成功，False 表示失败。
     """
+    # 完整URL走HTTP下载（兼容预签名URL）
+    if asset_storage_key.startswith(("http://", "https://")):
+        return _download_via_http(asset_storage_key, local_path)
+
+    # OSS存储键走SDK
     bucket = oss_bucket()
     if bucket is None:
         return False
@@ -114,6 +124,24 @@ def download_asset(asset_storage_key: str, local_path: Path) -> bool:
         return False
 
 
+def _download_via_http(url: str, local_path: Path) -> bool:
+    """通过 HTTP 下载文件（支持预签名 URL）。
+
+    使用流式下载避免大文件内存溢出，超时 300s。
+    """
+    try:
+        resp = requests.get(url, stream=True, timeout=300)
+        resp.raise_for_status()
+        with open(local_path, "wb") as f:
+            for chunk in resp.iter_content(chunk_size=8 * 1024 * 1024):
+                if chunk:
+                    f.write(chunk)
+        return local_path.exists() and local_path.stat().st_size > 0
+    except Exception:
+        logger.exception("HTTP下载素材失败: %s", url)
+        return False
+
+
 def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
     """上传文件到 OSS，返回公开 URL。
 
diff --git a/tests/unit/test_oss_helpers_download.py b/tests/unit/test_oss_helpers_download.py
new file mode 100755
index 000000000..06374797a
--- /dev/null
+++ b/tests/unit/test_oss_helpers_download.py
@@ -0,0 +1,189 @@
+"""oss_helpers.download_asset 单元测试 — HTTP URL / OSS key 双路径覆盖."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def tmp_asset_dir(tmp_path: Path) -> Path:
+    d = tmp_path / "assets"
+    d.mkdir()
+    return d
+
+
+# ── HTTP URL 下载路径 ──────────────────────────────────────────────────────
+
+
+class TestDownloadAssetHttpUrl:
+    """传入完整 https:// URL 时走 requests HTTP 下载，不走 oss2 SDK."""
+
+    @patch("apps.worker.video_processing.oss_helpers.requests.get")
+    def test_http_url_download_success(self, mock_get, tmp_asset_dir):
+        """HTTP URL 下载成功."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.iter_content.return_value = [b"fake-video-data" * 100]
+        mock_get.return_value = mock_resp
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("https://oss.example.com/assets/test.mp4?sign=xxx", local_path)
+
+        assert result is True
+        assert local_path.exists()
+        assert local_path.stat().st_size > 0
+        mock_get.assert_called_once()
+        # 不走 oss2 bucket
+        assert "oss_bucket" not in str(mock_get.call_args)
+
+    @patch("apps.worker.video_processing.oss_helpers.requests.get")
+    def test_http_url_download_403_returns_false(self, mock_get, tmp_asset_dir):
+        """HTTP URL 返回 403，下载失败返回 False."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_resp = MagicMock()
+        mock_resp.status_code = 403
+        mock_resp.raise_for_status.side_effect = Exception("403 Forbidden")
+        mock_get.return_value = mock_resp
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("https://oss.example.com/assets/expired.mp4?sign=old", local_path)
+
+        assert result is False
+
+    @patch("apps.worker.video_processing.oss_helpers.requests.get")
+    def test_http_url_connection_error_returns_false(self, mock_get, tmp_asset_dir):
+        """HTTP URL 连接超时，返回 False."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_get.side_effect = Exception("Connection timeout")
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("https://unreachable.example.com/test.mp4", local_path)
+
+        assert result is False
+
+    @patch("apps.worker.video_processing.oss_helpers.requests.get")
+    def test_http_url_streams_in_chunks(self, mock_get, tmp_asset_dir):
+        """HTTP 下载使用流式 + 8MB chunk，避免大文件内存溢出."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.iter_content.return_value = [b"chunk1", b"chunk2", b"chunk3"]
+        mock_get.return_value = mock_resp
+
+        local_path = tmp_asset_dir / "test.mp4"
+        download_asset("https://oss.example.com/assets/big.mp4", local_path)
+
+        mock_resp.iter_content.assert_called_once()
+        call_kwargs = mock_resp.iter_content.call_args
+        assert call_kwargs.kwargs.get("chunk_size") == 8 * 1024 * 1024
+
+
+# ── OSS key 下载路径 ──────────────────────────────────────────────────────
+
+
+class TestDownloadAssetOssKey:
+    """传入 OSS 存储键时走 oss2 SDK，不走 HTTP."""
+
+    @patch("apps.worker.video_processing.oss_helpers.requests.get")
+    @patch("apps.worker.video_processing.oss_helpers.oss_bucket")
+    def test_oss_key_uses_sdk_not_http(self, mock_oss_bucket, mock_requests_get, tmp_asset_dir):
+        """OSS key 路径走 oss2 SDK，不调用 requests.get."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_bucket = MagicMock()
+        mock_bucket.get_object_to_file.side_effect = lambda key, path: Path(path).write_bytes(b"oss-data")
+        mock_oss_bucket.return_value = mock_bucket
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("assets/user123/test.mp4", local_path)
+
+        assert result is True
+        mock_bucket.get_object_to_file.assert_called_once()
+        mock_requests_get.assert_not_called()
+
+    @patch("apps.worker.video_processing.oss_helpers.oss_bucket")
+    def test_oss_bucket_none_returns_false(self, mock_oss_bucket, tmp_asset_dir):
+        """OSS 未配置时返回 False."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_oss_bucket.return_value = None
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("assets/test.mp4", local_path)
+
+        assert result is False
+
+    @patch("apps.worker.video_processing.oss_helpers.oss_bucket")
+    def test_oss_sdk_exception_returns_false(self, mock_oss_bucket, tmp_asset_dir):
+        """OSS SDK 抛异常时返回 False，不崩溃."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_bucket = MagicMock()
+        mock_bucket.get_object_to_file.side_effect = Exception("NoSuchKey")
+        mock_oss_bucket.return_value = mock_bucket
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("assets/nonexistent.mp4", local_path)
+
+        assert result is False
+
+
+# ── 路径选择逻辑 ──────────────────────────────────────────────────────────
+
+
+class TestDownloadAssetRouteSelection:
+    """URL vs OSS key 的路由选择."""
+
+    @patch("apps.worker.video_processing.oss_helpers._download_via_http")
+    @patch("apps.worker.video_processing.oss_helpers.oss_bucket")
+    def test_https_url_routes_to_http(self, mock_oss_bucket, mock_http_download, tmp_asset_dir):
+        """https:// 开头走 HTTP 下载."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_http_download.return_value = True
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("https://oss.example.com/a/b.mp4", local_path)
+
+        assert result is True
+        mock_http_download.assert_called_once()
+        mock_oss_bucket.assert_not_called()
+
+    @patch("apps.worker.video_processing.oss_helpers._download_via_http")
+    @patch("apps.worker.video_processing.oss_helpers.oss_bucket")
+    def test_http_url_routes_to_http(self, mock_oss_bucket, mock_http_download, tmp_asset_dir):
+        """http:// 开头也走 HTTP 下载."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_http_download.return_value = True
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("http://oss.example.com/a/b.mp4", local_path)
+
+        assert result is True
+        mock_http_download.assert_called_once()
+        mock_oss_bucket.assert_not_called()
+
+    @patch("apps.worker.video_processing.oss_helpers._download_via_http")
+    @patch("apps.worker.video_processing.oss_helpers.oss_bucket")
+    def test_plain_key_routes_to_oss_sdk(self, mock_oss_bucket, mock_http_download, tmp_asset_dir):
+        """普通路径走 oss2 SDK."""
+        from apps.worker.video_processing.oss_helpers import download_asset
+
+        mock_bucket = MagicMock()
+        mock_bucket.get_object_to_file.side_effect = lambda key, path: Path(path).write_bytes(b"x")
+        mock_oss_bucket.return_value = mock_bucket
+
+        local_path = tmp_asset_dir / "test.mp4"
+        result = download_asset("user/123/asset.mp4", local_path)
+
+        assert result is True
+        mock_http_download.assert_not_called()