From 068e97d6fce21b88541e68f3ed88c6b5448b3196 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 19:57:27 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20P0-2=20=E6=B7=B1=E5=BA=A6=E6=A0=B9?= =?UTF-8?q?=E5=9B=A0=E4=BF=AE=E5=A4=8D=20=E2=80=94=20Worker=E7=AB=AFURL?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C403=20+=20endpoint=20HTTPS=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 根因分析 ### 问题 1（P0 紧急）：Worker 端上传后 URL 校验永远失败 **现象**：所有生成任务在 OSS 上传后失败，报 "OSS 上传后 URL 不可访问"。 **根因**： 用裸 URL 直接发起 HEAD 请求， 私有 bucket 下永远返回 403，导致所有生成任务失败。 **修复**： - 校验前先调用 生成预签名 URL - 预签名 URL 校验也失败时，降级用 确认上传成功 （上传成功本身就是最可靠的凭证） ### 问题 2：Worker 端 oss_bucket() endpoint 无 scheme **根因**： 的 直接使用环境变量中的 endpoint（如 ），不传 scheme。 oss2 SDK 在这种情况下 sign_url 会默认生成 HTTP URL。 **修复**：endpoint 不带 scheme 时自动补 ，与 API 端 storage.py 的修复保持一致。 ### 问题 3：API 端 results 接口返回裸 URL？ **排查结论：API 端代码逻辑正确，不是后端 bug。** 接口确实调用了 ，预签名 URL 放在 字段返回。 字段保留原始 URL（设计如此）。 如果前端拿不到预签名 URL，可能原因： 1. 前端使用了 而非 字段 2. staging API 尚未部署含 HTTPS 修复的最新代码 3. OSS 凭证未配置导致 bucket=None，走 fallback 返回裸 URL （日志中会有 warning） ## 新增函数 - Worker 端生成预签名 URL 的统一入口 - 自动从完整 URL 提取 storage key - 失败时返回 None（不抛异常） ## 测试 新增 10 个单元测试： - oss_bucket endpoint scheme 修复（4 个） - get_signed_download_url 功能（4 个） - upload_to_oss 返回 HTTPS URL（2 个） 测试结果：10/10 passed --- apps/worker/video_processing/oss_helpers.py | 32 ++- apps/worker/worker_app/tasks/generation.py | 22 +- tests/unit/test_p02_worker_oss_fix.py | 268 ++++++++++++++++++++ 3 files changed, 318 insertions(+), 4 deletions(-) mode change 100644 => 100755 apps/worker/video_processing/oss_helpers.py create mode 100755 tests/unit/test_p02_worker_oss_fix.py diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py old mode 100644 new mode 100755 index 18e6a1674..0e988f6a3 --- a/apps/worker/video_processing/oss_helpers.py +++ b/apps/worker/video_processing/oss_helpers.py @@ -40,6 +40,9 @@ def oss_settings() -> tuple[str, str, str, str] | None: def oss_bucket() -> oss2.Bucket | None: """获取 OSS Bucket 实例。 + P0-2 修复：endpoint 不带 scheme 时自动补 https:// 前缀， + 确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。 + Returns: oss2.Bucket 实例，配置缺失时返回 None。 """ @@ -47,6 +50,9 @@ def oss_bucket() -> oss2.Bucket | None: if settings is None: return None access_key_id, access_key_secret, endpoint, bucket_name = settings + # 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) @@ -105,13 +111,37 @@ def upload_to_oss(local_path: Path, storage_key: str) -> str | None: settings = oss_settings() if settings: _, _, endpoint, bucket_name = settings - return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}" + endpoint_clean = endpoint.replace("https://", "").replace("http://", "") + return f"https://{bucket_name}.{endpoint_clean}/{storage_key}" return None except Exception: logger.exception("上传 OSS 失败: %s", storage_key) return None +def get_signed_download_url(storage_key_or_url: str, expires_seconds: int = 3600) -> str | None: + """生成预签名下载 URL（用于私有 bucket 的 URL 校验或临时下载）。 + + Args: + storage_key_or_url: 存储键或完整 URL（URL 会自动提取 path） + expires_seconds: 签名有效期（秒） + + Returns: + 预签名 URL，失败或 OSS 未配置时返回 None。 + """ + bucket = oss_bucket() + if bucket is None: + return None + try: + storage_key = normalize_storage_key(storage_key_or_url) + signed = bucket.sign_url("GET", storage_key, expires_seconds) + logger.info("生成预签名URL: key=%s url_prefix=%s", storage_key[:80], signed[:60]) + return signed + except Exception: + logger.exception("生成预签名URL失败: %s", storage_key_or_url[:80]) + return None + + # ── Asset 解析 ──────────────────────────────────────────────────────────────── diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 1e6716aa1..3c84f45c8 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -87,6 +87,7 @@ from video_processing.dedup_helpers import create_video_record_and_dedup from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg from video_processing.oss_helpers import ( download_asset, + get_signed_download_url, upload_to_oss, ) from video_processing.unified_render_service import UnifiedRenderService @@ -623,9 +624,24 @@ def generate_video(self, task_id: str) -> dict: f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}, " f"output_path={output_path}" ) - # HEAD 校验 URL 可访问 - if not _verify_url_accessible(file_url): - raise RuntimeError(f"OSS 上传后 URL 不可访问: file_url={file_url}, " f"storage_key={storage_key}") + # P0-2 修复：私有 bucket 下裸 URL 永远 403，改用预签名 URL 校验 + # 先用预签名 URL 校验，失败则降级为检查文件是否存在（object_exists） + verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url + if not _verify_url_accessible(verify_url): + # 预签名 URL 也访问失败时，退一步用 object_exists 确认上传成功 + from video_processing.oss_helpers import oss_bucket, normalize_storage_key + + bucket = oss_bucket() + key = normalize_storage_key(file_url) + if bucket and bucket.object_exists(key): + logger.info( + "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key + ) + else: + raise RuntimeError( + f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, " + f"storage_key={storage_key}" + ) logger.info("OSS 上传成功: file_url=%s", file_url) diff --git a/tests/unit/test_p02_worker_oss_fix.py b/tests/unit/test_p02_worker_oss_fix.py new file mode 100755 index 000000000..3be633bfc --- /dev/null +++ b/tests/unit/test_p02_worker_oss_fix.py @@ -0,0 +1,268 @@ +"""P0-2 深度修复：Worker 端 OSS 工具函数测试. + +测试： +1. oss_bucket() endpoint 自动补 https:// 前缀 +2. get_signed_download_url() 生成预签名 URL +3. upload_to_oss() 返回 HTTPS URL +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +import pytest + + +# ── oss_bucket endpoint scheme 修复 ────────────────────────────────────────── + + +class TestOSSBucketEndpointScheme: + """测试 oss_bucket() 自动为 endpoint 补 https:// 前缀.""" + + def test_endpoint_without_scheme_adds_https(self): + """endpoint 不带 scheme 时，自动补 https://.""" + 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") as mock_auth, patch( + "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance + ) as mock_bucket_cls: + # 清除缓存，确保重新创建 + import video_processing.oss_helpers as oss_mod + + bucket = oss_bucket() + + assert bucket is mock_bucket_instance + # 验证 endpoint 传的是带 https:// 的 + call_args = mock_bucket_cls.call_args + endpoint_arg = call_args[0][1] # 第 2 个位置参数是 endpoint + assert endpoint_arg.startswith("https://"), ( + f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}" + ) + assert "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg + + def test_endpoint_with_https_keeps_as_is(self): + """endpoint 已有 https:// 时，不重复添加.""" + 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": "https://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: + import video_processing.oss_helpers as oss_mod + + bucket = oss_bucket() + + call_args = mock_bucket_cls.call_args + endpoint_arg = call_args[0][1] + # 不应该出现 https://https:// 这种双重前缀 + assert endpoint_arg.count("https://") == 1 + assert endpoint_arg == "https://oss-cn-hangzhou.aliyuncs.com" + + def test_endpoint_with_http_keeps_as_is(self): + """endpoint 已有 http:// 时，不修改（保留用户选择）.""" + 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": "http://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: + import video_processing.oss_helpers as oss_mod + + bucket = oss_bucket() + + call_args = mock_bucket_cls.call_args + endpoint_arg = call_args[0][1] + assert endpoint_arg == "http://oss-cn-hangzhou.aliyuncs.com" + + def test_missing_credentials_returns_none(self): + """凭证缺失时返回 None.""" + from video_processing.oss_helpers import oss_bucket + + with patch.dict( + os.environ, + { + "OSS_ACCESS_KEY_ID": "", + "OSS_ACCESS_KEY_SECRET": "", + "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", + "OSS_BUCKET_NAME": "test-bucket", + }, + clear=True, + ): + import video_processing.oss_helpers as oss_mod + + bucket = oss_bucket() + assert bucket is None + + +# ── get_signed_download_url ────────────────────────────────────────────────── + + +class TestGetSignedDownloadUrl: + """测试 get_signed_download_url() 预签名 URL 生成.""" + + def test_returns_signed_url_with_storage_key(self): + """传入 storage key 时，调用 sign_url 并返回结果.""" + from video_processing.oss_helpers import get_signed_download_url + + mock_bucket = MagicMock() + mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx" + + 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 + ): + result = get_signed_download_url("generated/test.mp4", expires_seconds=3600) + + assert result is not None + assert "Signature=" in result + mock_bucket.sign_url.assert_called_once_with("GET", "generated/test.mp4", 3600) + + def test_normalizes_full_url_to_storage_key(self): + """传入完整 URL 时，提取 storage key 再生成签名.""" + from video_processing.oss_helpers import get_signed_download_url + + mock_bucket = MagicMock() + mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx" + + 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 + ): + result = get_signed_download_url( + "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" + ) + + mock_bucket.sign_url.assert_called_once() + # 验证传给 sign_url 的是纯 storage key，不是完整 URL + call_key = mock_bucket.sign_url.call_args[0][1] + assert not call_key.startswith("http") + assert call_key == "generated/test.mp4" + + def test_returns_none_when_bucket_none(self): + """bucket 为 None 时返回 None（不抛异常）.""" + from video_processing.oss_helpers import get_signed_download_url + + with patch.dict(os.environ, {}, clear=True): + result = get_signed_download_url("generated/test.mp4") + assert result is None + + def test_sign_url_exception_returns_none(self): + """sign_url 抛异常时，返回 None（不向上抛出）.""" + from video_processing.oss_helpers import get_signed_download_url + + mock_bucket = MagicMock() + mock_bucket.sign_url.side_effect = Exception("sign failed") + + 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 + ): + result = get_signed_download_url("generated/test.mp4") + assert result is None + + +# ── upload_to_oss 返回 HTTPS URL ───────────────────────────────────────────── + + +class TestUploadToOSSReturnsHTTPS: + """测试 upload_to_oss() 返回的 URL 始终是 HTTPS.""" + + def test_endpoint_without_scheme_returns_https_url(self): + """endpoint 不带 scheme 时，返回 HTTPS URL.""" + from video_processing.oss_helpers import upload_to_oss + + mock_bucket = MagicMock() + mock_bucket.put_object_from_file = MagicMock() + + from pathlib import Path + + 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 + ): + result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4") + + assert result is not None + assert result.startswith("https://") + assert "test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" in result + + def test_endpoint_with_https_returns_clean_url(self): + """endpoint 带 https:// 时，URL 里不会有双重 https.""" + from video_processing.oss_helpers import upload_to_oss + + mock_bucket = MagicMock() + mock_bucket.put_object_from_file = MagicMock() + + from pathlib import Path + + with patch.dict( + os.environ, + { + "OSS_ACCESS_KEY_ID": "test-key", + "OSS_ACCESS_KEY_SECRET": "test-secret", + "OSS_ENDPOINT": "https://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 + ): + result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4") + + assert result is not None + assert result.startswith("https://") + # 不应该出现 https://https:// + assert result.count("https://") == 1 -- 2.54.0