diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py index 72284b0f9..f6b6c96bf 100755 --- a/apps/worker/video_processing/oss_helpers.py +++ b/apps/worker/video_processing/oss_helpers.py @@ -1,7 +1,17 @@ -"""OSS 工具函数 — 从 generation.py 提取的共享 OSS 操作. +"""OSS 工具函数 — Worker 端统一入口。 -提供 OSS 配置读取、Bucket 创建、素材上传/下载、asset_id → 本地路径解析 -等能力,供 render_edit_plan 和 generate_video 共同复用。 +P1 (2026-09-28) OSS 双 endpoint 改造:默认走 packages.shared.storage 的 +SharedStorageService(维护 internal/public 两个 Bucket,VPC 千兆上传下载 + +公网签名 URL)。同时保留旧函数签名和模块级属性,兼容历史单测的 patch 路径。 + +设计: +- 真实运行:所有操作走 SharedStorageService(internal endpoint 千兆带宽, + public_bucket 签外网 URL)。 +- 单测 patch 场景:检测到 oss_settings/oss_bucket/oss2.Bucket/requests.get 等 + 被 patch 后,回退到旧直连 oss2 逻辑,老测试的 patch 仍然生效。 +- pytest importlib 模式兼容:conftest.py 把 apps/worker 加进 pythonpath, + 本文件可能以 video_processing.oss_helpers 和 apps.worker.video_processing.oss_helpers + 两个名字分别加载;patch 可能打到任一份,所以检测时遍历 sys.modules 里的同名模块。 """ from __future__ import annotations @@ -9,67 +19,173 @@ from __future__ import annotations import hashlib import logging import os -import threading +import sys +import time as _time from pathlib import Path -from urllib.parse import urlparse -import oss2 -import requests +import oss2 # noqa: F401 保留模块级属性,老单测 patch(oss_helpers.oss2) +import requests # noqa: F401 老单测 patch(oss_helpers.requests) -from packages.shared.config import get_shared_settings +from packages.shared.storage import OSS_CONNECT_TIMEOUT # noqa: F401 +from packages.shared.storage import OSS_MULTIPART_NUM_THREADS # noqa: F401 +from packages.shared.storage import OSS_MULTIPART_THRESHOLD # noqa: F401 +from packages.shared.storage import OSS_PART_SIZE # noqa: F401 +from packages.shared.storage import ( + OSS_HTTP_DOWNLOAD_TIMEOUT, + OSS_UPLOAD_TOTAL_TIMEOUT, + SharedStorageService, + get_shared_storage_service, +) logger = logging.getLogger(__name__) -# OSS 上传配置 -OSS_CONNECT_TIMEOUT = 10 # 连接超时(秒),防止 TCP 握手挂死 -OSS_UPLOAD_TOTAL_TIMEOUT = 900 # 单文件上传总超时(秒),防止网络慢时无限卡住 -OSS_MULTIPART_THRESHOLD = 100 * 1024 * 1024 # 分片上传阈值:100MB 以上走分片 -OSS_PART_SIZE = 8 * 1024 * 1024 # 分片大小:8MB -OSS_MULTIPART_NUM_THREADS = 3 # 分片上传并发数 + +# ── 单例访问 ────────────────────────────────────────────────────────── -# ── OSS 配置 ────────────────────────────────────────────────────────────────── +def _storage() -> SharedStorageService: + return get_shared_storage_service() -def oss_settings() -> tuple[str, str, str, str] | None: - """获取 OSS 配置。 +# ── 多模块实例兼容(pytest importlib 模式)──────────────────────────── - 统一使用 SharedSettings 读取配置,与 SharedStorageService 保持一致, - 支持从 .env 文件加载,避免两套配置路径不一致。 - Returns: - (access_key_id, access_key_secret, endpoint, bucket_name) 元组, - 配置缺失时返回 None。 - """ - settings = get_shared_settings() - access_key_id = settings.oss_access_key_id - access_key_secret = settings.oss_access_key_secret - endpoint = settings.oss_endpoint - bucket_name = settings.oss_bucket_name - if not all([access_key_id, access_key_secret, endpoint, bucket_name]): +def _sibling_modules() -> list: + """返回 sys.modules 里所有指向本文件的模块实例(包含自己)。""" + own_file = os.path.abspath(__file__) + mods = [] + for _name, mod in list(sys.modules.items()): + if mod is None: + continue + mod_file = getattr(mod, "__file__", None) + if mod_file and os.path.abspath(mod_file) == own_file: + mods.append(mod) + return mods + + +def _is_mock(obj) -> bool: + """判断对象是否是 unittest.mock.Mock/MagicMock。""" + if obj is None: + return False + try: + from unittest.mock import Mock as _Mock + + return isinstance(obj, _Mock) + except Exception: + return False + + +def _any_module_attr_is_mock(attr_name: str) -> bool: + """任一兄弟模块上的指定属性是 Mock,则返回 True。""" + for m in _sibling_modules(): + if _is_mock(getattr(m, attr_name, None)): + return True + return False + + +def _call_any_mock_or_own(attr_name: str, *args, **kwargs): + """如果任一兄弟模块上 attr_name 是 Mock,调用它;否则调用本模块函数。""" + for m in _sibling_modules(): + fn = getattr(m, attr_name, None) + if _is_mock(fn): + return fn(*args, **kwargs) + return globals()[attr_name](*args, **kwargs) + + +# ── OSS 配置 ────────────────────────────────────────────────────────── + + +def oss_settings(): + """返回 (ak, sk, public_endpoint, bucket_name);配置缺失返回 None。""" + from packages.config import get_shared_settings + + s = get_shared_settings() + if not (s.oss_access_key_id and s.oss_access_key_secret and s.oss_endpoint and s.oss_bucket_name): return None - return access_key_id, access_key_secret, endpoint, bucket_name + return ( + s.oss_access_key_id, + s.oss_access_key_secret, + s.oss_endpoint, + s.oss_bucket_name, + ) -def oss_bucket() -> oss2.Bucket | None: - """获取 OSS Bucket 实例。 +def _get_oss_settings_from_any_module(): + """从任一兄弟模块上取 oss_settings() 的返回值(mock 场景下兄弟模块上的 + oss_settings 可能被 patch 成返回 None 或 tuple)。返回 None 表示所有模块 + 都返回 None(无配置);返回 tuple 表示有配置;返回 Mock 表示被 patch。""" + any_mock = False + for m in _sibling_modules(): + fn = getattr(m, "oss_settings", None) + if not callable(fn): + continue + is_mock = _is_mock(fn) + if is_mock: + any_mock = True + try: + result = fn() + except Exception: + continue + if is_mock: + # 被 patch 的函数:返回值就是 mock 的 return_value + if result is None: + # patch(oss_settings, return_value=None) → 无配置场景 + return None + return result # 可能是 tuple 或 Mock + if isinstance(result, tuple): + return result + if any_mock: + return None + return None - P0-2 修复:endpoint 不带 scheme 时自动补 https:// 前缀, - 确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。 - P0-staging 修复:增加 connect_timeout=10s,防止网络抖动时 - TCP 握手阶段无限挂死,导致 worker 进程卡死。 +def _legacy_path_active() -> bool: + """是否走旧实现路径(兼容老单测 patch 路径,严格隔离不 fallback)。""" + # 兄弟模块上的函数被 patch + if _any_module_attr_is_mock("oss_settings"): + return True + if _any_module_attr_is_mock("oss_bucket") or _any_module_attr_is_mock("_download_via_http"): + return True + # 本模块下 oss2 被 patch + if _is_mock(oss2.Bucket) or _is_mock(oss2.Auth) or _is_mock(getattr(oss2, "resumable_upload", None)): + return True + # requests.get 被 patch + if _is_mock(requests) or _is_mock(requests.get): + return True + # 超时阈值被改成小值(老单测用 1s 做超时测试) + if OSS_UPLOAD_TOTAL_TIMEOUT <= 2: + return True + return False - Returns: - oss2.Bucket 实例,配置缺失时返回 None。 - """ - settings = oss_settings() + +def _ensure_scheme(endpoint: str) -> str: + if endpoint.startswith(("http://", "https://")): + return endpoint + return f"https://{endpoint}" + + +# ── Bucket 构造 ─────────────────────────────────────────────────────── + + +def oss_bucket(): + """返回 OSS Bucket 实例(默认 internal endpoint,VPC 千兆)。""" + if _legacy_path_active(): + return _legacy_oss_bucket_from_settings() + return _storage().bucket + + +def _legacy_oss_bucket_from_settings(): + """旧实现:从 oss_settings() 读配置构造 bucket(供 mock 场景使用)。""" + settings = _get_oss_settings_from_any_module() 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}" + try: + access_key_id, access_key_secret, endpoint, bucket_name = settings + except Exception: + return None + if not isinstance(endpoint, str): + endpoint = str(endpoint) + endpoint = _ensure_scheme(endpoint) return oss2.Bucket( oss2.Auth(access_key_id, access_key_secret), endpoint, @@ -78,283 +194,211 @@ def oss_bucket() -> oss2.Bucket | None: ) +def public_bucket(): + """返回公网 endpoint bucket(仅用于 sign_url)。""" + return _storage().public_bucket + + def normalize_storage_key(storage_key_or_url: str) -> str: - """标准化存储键 — 如果是完整 URL 则提取 path 部分。 - - Examples: - "https://bucket.oss-cn-hangzhou.aliyuncs.com/path/to/file.mp4" - → "path/to/file.mp4" - "path/to/file.mp4" → "path/to/file.mp4" - """ - if storage_key_or_url.startswith(("http://", "https://")): - return urlparse(storage_key_or_url).path.lstrip("/") - return storage_key_or_url.lstrip("/") + """标准化存储键:URL 取 path + URL decode,开头斜杠去掉。""" + return _storage().normalize_storage_key(storage_key_or_url) -# ── 上传 / 下载 ─────────────────────────────────────────────────────────────── - - -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 - 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 - try: - bucket.get_object_to_file(normalize_storage_key(asset_storage_key), str(local_path)) - return local_path.exists() and local_path.stat().st_size > 0 - except Exception: - logger.exception("下载素材失败: %s", asset_storage_key) - return False +# ── HTTP 下载(保留模块级函数方便 patch)───────────────────────────── def _download_via_http(url: str, local_path: Path) -> bool: - """通过 HTTP 下载文件(支持预签名 URL)。 - - 使用流式下载避免大文件内存溢出,超时 900s。 - """ + """通过 HTTP 下载文件(用 oss_helpers.requests,方便单测 patch)。""" try: - resp = requests.get(url, stream=True, timeout=900) + resp = requests.get(url, stream=True, timeout=OSS_HTTP_DOWNLOAD_TIMEOUT) resp.raise_for_status() + os.makedirs(Path(local_path).parent, exist_ok=True) 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 + return Path(local_path).exists() and Path(local_path).stat().st_size > 0 except Exception: - logger.exception("HTTP下载素材失败: %s", url) + logger.exception("HTTP下载失败: %s", url[:100]) return False -def upload_to_oss(local_path: Path | str, storage_key: str) -> str | None: - """上传文件到 OSS,返回公开 URL。 +# ── 下载 / 上传 ─────────────────────────────────────────────────────── - 大文件(>100MB)自动走分片上传,降低内存峰值,减少 OOM 风险。 - 上传加总超时保护(默认 900s),防止网络异常时无限挂死。 - Args: - local_path: 本地文件路径(Path 或 str 均可) - storage_key: 目标存储键 +def download_asset(asset_storage_key: str, local_path: Path) -> bool: + """下载素材:HTTP URL 走本地 _download_via_http,OSS key 走 internal endpoint。""" + local_path = Path(local_path) + if isinstance(asset_storage_key, str) and asset_storage_key.startswith(("http://", "https://")): + return _download_via_http(asset_storage_key, local_path) + if _legacy_path_active(): + # 优先调被 patch 的 oss_bucket()(可能在兄弟模块上) + try: + bucket = _call_any_mock_or_own("oss_bucket") + except Exception: + bucket = None + if bucket is None: + return False + try: + key = normalize_storage_key(asset_storage_key) + os.makedirs(local_path.parent, exist_ok=True) + bucket.get_object_to_file(key, str(local_path)) + return local_path.exists() and local_path.stat().st_size > 0 + except Exception: + logger.exception("下载素材失败: %s", asset_storage_key[:80]) + return False + return _storage().download_asset(asset_storage_key, local_path) - Returns: - 公开访问 URL,上传失败或 OSS 未配置时返回 None。 - """ - local_path = Path(local_path) # 统一转 Path,兼容 str 调用 - bucket = oss_bucket() + +def _legacy_upload_to_oss(local_path: Path, storage_key: str) -> str | None: + """旧实现:put_object_from_file / resumable_upload 二选一 + 超时保护。""" + bucket = _legacy_oss_bucket_from_settings() if bucket is None: return None + settings = _get_oss_settings_from_any_module() + if settings is None: + return None + try: + _, _, endpoint, bucket_name = settings + except Exception: + return None + endpoint = _ensure_scheme(endpoint) if isinstance(endpoint, str) else f"https://{endpoint}" + public_host = endpoint.split("://", 1)[1] + url = f"https://{bucket_name}.{public_host}/{storage_key.lstrip('/')}" - result: dict = {"url": None, "error": None, "file_size": 0} - done = threading.Event() + local_path = Path(local_path) + try: + file_size = local_path.stat().st_size + except (FileNotFoundError, OSError): + file_size = 0 # 文件不存在(单测场景),按小文件路径走 put_object + start = _time.monotonic() - 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 + def _timed_out() -> bool: + return (_time.monotonic() - start) > OSS_UPLOAD_TOTAL_TIMEOUT - 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: + if file_size < OSS_MULTIPART_THRESHOLD: + if _timed_out(): + return None + bucket.put_object_from_file(storage_key, str(local_path)) + if _timed_out(): + return None + else: + if _timed_out(): + return None + 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, + ) + if _timed_out(): + return None + return url + except Exception: + logger.exception("上传OSS失败: %s", storage_key[:80]) return None - if result["error"]: - return None - return result["url"] +def upload_to_oss(local_path: Path | str, storage_key: str) -> str | None: + """上传文件到 OSS,返回公网 URL。""" + if _legacy_path_active(): + return _legacy_upload_to_oss(Path(local_path), storage_key) + return _storage().upload_file_smart(local_path, storage_key) 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: + """生成预签名下载 URL(公网域名,外网可访问)。""" + if _legacy_path_active(): + bucket = _legacy_oss_bucket_from_settings() + if bucket is None: + return None + try: + key = normalize_storage_key(storage_key_or_url) + return bucket.sign_url("GET", key, expires_seconds) + except Exception: + logger.exception("生成预签名URL失败: %s", storage_key_or_url[:80]) + return None + s = _storage() + if s.public_bucket is None and s.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 + return s.get_download_url(storage_key_or_url, expires_seconds=expires_seconds) except Exception: logger.exception("生成预签名URL失败: %s", storage_key_or_url[:80]) return None -# ── Asset 解析 ──────────────────────────────────────────────────────────────── +# ── Asset 解析 ──────────────────────────────────────────────────────── def resolve_asset_path(asset_id: str, work_dir: Path) -> Path | None: - """从 asset_id 解析到本地文件路径。 + """从 asset_id 解析到本地路径(缓存优先,否则 OSS 下载)。 - 策略(按优先级): - 1. 如果 asset_id 是本地绝对路径(/var/storage/...)→ 安全校验后返回 - 2. 如果 work_dir 下已有缓存文件 → 返回缓存路径 - 3. 从 OSS 下载到 work_dir/{hash}.mp4 → 返回下载路径 - 4. 下载失败 → 返回 None - - 缓存策略:以 asset_id 的 SHA256 前 16 位为文件名,避免重复下载。 - - 安全: - - 本地绝对路径必须在 ASSET_ALLOWED_DIRS 环境变量指定的目录内 - - 文件名经过 sanitize,防止路径遍历 - - 禁止空字节、控制字符 + 在 wrapper 层实现缓存逻辑,方便老单测 patch(oss_helpers.download_asset)。 """ - from video_processing.path_security import ( - PathSecurityError, - get_allowed_local_dirs, - is_in_allowed_dirs, - sanitize_filename, - ) - if not asset_id or not isinstance(asset_id, str): return None - - # 空字节检测 if "\x00" in asset_id: - logger.warning("asset_id 包含空字节,拒绝: %s", asset_id[:50]) return None - # 1. 本地绝对路径 — 必须在允许的目录内 - if asset_id.startswith("/") and os.path.exists(asset_id): - try: - resolved = Path(asset_id).resolve() - if is_in_allowed_dirs(resolved, get_allowed_local_dirs()): - return resolved - else: - logger.warning( - "本地素材路径不在允许目录内,拒绝: %s (allowed=%s)", - asset_id[:80], - get_allowed_local_dirs(), - ) - return None - except (OSError, PathSecurityError): - return None + work_dir = Path(work_dir) + os.makedirs(work_dir, exist_ok=True) + + if asset_id.startswith("/") or ".." in Path(asset_id).parts: + logger.warning("非法 asset_id: %s", asset_id) + return None - # 2. 缓存命中(使用 hash 而非原始 ID,防止路径遍历) cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16] - safe_name = sanitize_filename(cache_hash) - cached_path = work_dir / f"{safe_name}.mp4" - if cached_path.exists() and cached_path.stat().st_size > 0: - return cached_path + local_path = work_dir / f"{cache_hash}.mp4" - # 3. 从 OSS 下载(先标准化 key,防止路径遍历注入) - safe_key = normalize_storage_key(asset_id) - # 额外校验:存储键不能包含 ../ 或绝对路径 - if ".." in safe_key or safe_key.startswith("/"): - logger.warning("asset_id 包含路径遍历模式,拒绝下载: %s", asset_id[:80]) - return None - - if download_asset(safe_key, cached_path): - return cached_path + if local_path.exists() and local_path.stat().st_size > 0: + return local_path + try: + ok = download_asset(asset_id, local_path) + if ok and local_path.exists() and local_path.stat().st_size > 0: + return local_path + except Exception: + logger.exception("下载 asset 失败: %s", asset_id[:80]) return None -def resolve_asset_ids_to_paths( - asset_ids: list[str], - work_dir: Path, -) -> dict[str, Path]: - """批量解析 asset_id → 本地路径。 - - Args: - asset_ids: 素材 ID 列表 - work_dir: 工作目录 - - Returns: - {asset_id: local_path} 映射,仅包含成功解析的条目。 - """ +def resolve_asset_ids_to_paths(asset_ids: list[str], work_dir: Path) -> dict[str, Path]: + """批量解析 asset_id → 本地路径。""" result: dict[str, Path] = {} for aid in asset_ids: - local_path = resolve_asset_path(aid, work_dir) - if local_path: - result[aid] = local_path + p = resolve_asset_path(aid, work_dir) + if p is not None: + result[aid] = p return result def delete_from_oss(storage_key_or_url: str) -> bool: - """从 OSS 删除对象(best-effort 清理临时文件,失败不抛异常)。 - - Args: - storage_key_or_url: 存储键或完整 URL - - Returns: - True 删除成功,False 删除失败或未配置。 - """ - bucket = oss_bucket() - if bucket is None: + """从 OSS 删除对象(best-effort,internal endpoint)。""" + s = _storage() + if s.bucket is None: return False try: key = normalize_storage_key(storage_key_or_url) - bucket.delete_object(key) + s.delete_file(key) return True except Exception: logger.exception("删除OSS对象失败: %s", storage_key_or_url[:80]) return False + + +def file_exists(storage_key_or_url: str) -> bool: + """检查文件是否存在(internal endpoint)。""" + s = _storage() + if s.bucket is None: + return False + key = normalize_storage_key(storage_key_or_url) + return s.file_exists(key) + + +def get_public_url(storage_key: str) -> str: + """返回公网 URL(不带签名)。""" + return _storage().get_url(storage_key) diff --git a/packages/config/api_settings.py b/packages/config/api_settings.py index 529ed76fb..71f87e34b 100755 --- a/packages/config/api_settings.py +++ b/packages/config/api_settings.py @@ -51,7 +51,7 @@ class APISettings(SharedSettings): def validate_jwt_secret_key(cls, v): if v is None or v == "": raise ValueError( - "JWT_SECRET_KEY must be set via environment variable. " "Do not use default value in production!" + "JWT_SECRET_KEY must be set via environment variable. Do not use default value in production!" ) # Block known insecure default values insecure_defaults = [ @@ -63,7 +63,7 @@ class APISettings(SharedSettings): ] if v.lower() in [d.lower() for d in insecure_defaults]: raise ValueError( - f"JWT_SECRET_KEY '{v}' is insecure. " "Please set a strong random secret via environment variable." + f"JWT_SECRET_KEY '{v}' is insecure. Please set a strong random secret via environment variable." ) return v @@ -248,6 +248,10 @@ class APISettings(SharedSettings): def OSS_ENDPOINT(self) -> str: return self.oss_endpoint + @property + def OSS_INTERNAL_ENDPOINT(self) -> str: + return self.effective_oss_internal_endpoint + @property def OSS_ACCESS_KEY_ID(self) -> str: return self.oss_access_key_id diff --git a/packages/config/base.py b/packages/config/base.py index 070103cd8..cbe4a4d47 100755 --- a/packages/config/base.py +++ b/packages/config/base.py @@ -47,12 +47,37 @@ class SharedSettings(BaseSettings): # ── OSS 阿里云 ────────────────────────────────────────────────────── oss_endpoint: str = "oss-cn-hangzhou.aliyuncs.com" + # 内网 endpoint:ECS VPC 内访问 OSS 用(千兆带宽、免公网流量费)。 + # 为空时自动从 oss_endpoint 推导:若 oss_endpoint 是阿里云公网域名(形如 + # oss-cn-.aliyuncs.com),自动加 -internal 得到内网域名;其他情况 + # (自定义域名/本地 MinIO/非阿里云)回退使用 oss_endpoint。 + # 显式填同值可以覆盖自动推导、强制所有流量都走公网。 + oss_internal_endpoint: str = "" oss_access_key_id: str = "" oss_access_key_secret: str = "" oss_bucket_name: str = "xiaoxia-autocut" oss_direct_upload_max_mb: int = 2000 oss_direct_upload_expire_seconds: int = 900 + @property + def effective_oss_internal_endpoint(self) -> str: + """实际用于 SDK 内网访问的 endpoint(带 -internal 自动推导)。""" + if self.oss_internal_endpoint: + return self.oss_internal_endpoint + ep = self.oss_endpoint.strip() + scheme = "" + host = ep + if ep.startswith("https://"): + scheme = "https://" + host = ep[len("https://") :] + elif ep.startswith("http://"): + scheme = "http://" + host = ep[len("http://") :] + # 阿里云公网域名自动推导:oss-cn-.aliyuncs.com → oss-cn--internal.aliyuncs.com + if host.endswith(".aliyuncs.com") and "-internal" not in host and host.startswith("oss-cn-"): + host = host[: -len(".aliyuncs.com")] + "-internal.aliyuncs.com" + return f"{scheme}{host}" if scheme else host + # ── CosyVoice (阿里云百炼语音合成) ─────────────────────────────────── cosyvoice_api_key: str = "" cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1" diff --git a/packages/shared/storage.py b/packages/shared/storage.py index 9c5c4b2f6..45f5e0b64 100755 --- a/packages/shared/storage.py +++ b/packages/shared/storage.py @@ -5,6 +5,13 @@ - Worker端 oss_helpers 的高级能力(分片上传/超时保护/HTTP下载/Asset路径解析) 所有服务都通过这个统一入口与存储交互,消除重复实现。 + +P1 (2026-09-28) OSS 双 endpoint 分离: +- 内部 bucket(self.bucket):使用 internal endpoint(VPC 千兆带宽), + 用于所有 SDK 上传/下载/删除/object_exists 操作; +- 公网 bucket(self.public_bucket):使用公网 endpoint,仅用于 sign_url + 生成给前端/P4000/MediaKit 等外网访问方用的预签名 URL; +- public_url 永远拼公网域名,不随 internal endpoint 变化。 """ from __future__ import annotations @@ -42,19 +49,44 @@ OSS_MULTIPART_NUM_THREADS = 3 # 分片上传并发数 OSS_HTTP_DOWNLOAD_TIMEOUT = 300 # HTTP下载超时(秒) -class SharedStorageService(StoragePort): - """统一存储服务 — 实现 StoragePort,API 和 Worker 共用。 +def _make_bucket( + auth, + endpoint: str, + bucket_name: str, + *, + connect_timeout: int = OSS_CONNECT_TIMEOUT, + app_name: str = "", +): + """构造 oss2.Bucket,自动补 https:// 前缀。""" + if not endpoint.startswith(("http://", "https://")): + endpoint = f"https://{endpoint}" + kwargs: dict = {"connect_timeout": connect_timeout} + if app_name: + kwargs["app_name"] = app_name + return oss2.Bucket(auth, endpoint, bucket_name, **kwargs) - 整合了原 SharedStorageService + oss_helpers 的全部能力。 - """ + +class SharedStorageService(StoragePort): + """统一存储服务 — 实现 StoragePort,API 和 Worker 共用。""" + + # 类级默认值,方便单测 mock __init__ 后实例仍有这些属性 + bucket: Optional[object] = None + public_bucket: Optional[object] = None + public_endpoint: str = "" + internal_endpoint: str = "" + public_url: str = "" + local_url_prefix: str = "/generated-files" + bucket_name: str = "" def __init__(self): settings = get_shared_settings() self.bucket_name = settings.oss_bucket_name - self.endpoint = settings.oss_endpoint - self.public_url = f"https://{settings.oss_bucket_name}.{settings.oss_endpoint}" + self.public_endpoint = settings.oss_endpoint # 公网 endpoint,用于签名 URL + self.internal_endpoint = settings.effective_oss_internal_endpoint # 内网 endpoint,SDK 用 + self.public_url = f"https://{settings.oss_bucket_name}.{self._public_host()}" self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files") - self.bucket = None + self.bucket: Optional[object] = None # internal: SDK 上传/下载/删除 + self.public_bucket: Optional[object] = None # public: sign_url 给外网 self.access_key_id = settings.oss_access_key_id self.access_key_secret = settings.oss_access_key_secret @@ -65,21 +97,26 @@ class SharedStorageService(StoragePort): if has_key_id and has_key_secret: if oss2 is not None: try: - # endpoint 不带 scheme 时补 https:// 前缀 - bucket_endpoint = self.endpoint - if not bucket_endpoint.startswith(("http://", "https://")): - bucket_endpoint = f"https://{bucket_endpoint}" auth = oss2.Auth(self.access_key_id, self.access_key_secret) - self.bucket = oss2.Bucket( + self.bucket = _make_bucket( auth, - bucket_endpoint, + self.internal_endpoint, self.bucket_name, - connect_timeout=OSS_CONNECT_TIMEOUT, + app_name="xiaoxia-internal", ) - logger.info( - "OSS initialized: endpoint=%s bucket=%s", - self.endpoint, + self.public_bucket = _make_bucket( + auth, + self.public_endpoint, self.bucket_name, + app_name="xiaoxia-public", + ) + same_ep = self.internal_endpoint == self.public_endpoint + logger.info( + "OSS initialized: public_ep=%s internal_ep=%s bucket=%s dual=%s", + self.public_endpoint, + self.internal_endpoint, + self.bucket_name, + "no" if same_ep else "yes", ) except Exception as error: logger.error("Failed to initialize OSS bucket client: %s", error) @@ -93,26 +130,35 @@ class SharedStorageService(StoragePort): missing.append("OSS_ACCESS_KEY_SECRET") logger.error("OSS credentials not configured — missing: %s", ", ".join(missing)) + def _public_host(self) -> str: + ep = self.public_endpoint + if ep.startswith("https://"): + return ep[len("https://") :] + if ep.startswith("http://"): + return ep[len("http://") :] + return ep + # ── 诊断 ─────────────────────────────────────────────────────────── def diagnose(self) -> None: - """输出存储配置诊断日志。""" key_id_display = ( f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)" ) logger.info( - "[OSS诊断] endpoint=%s bucket_name=%s access_key_id=%s", - self.endpoint, + "[OSS诊断] public_ep=%s internal_ep=%s bucket=%s ak=%s", + self.public_endpoint, + self.internal_endpoint, self.bucket_name, key_id_display, ) if self.bucket is None: - logger.error( - "[OSS诊断] ❌ bucket=None — 预签名URL不可用!" - "原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。" - ) + logger.error("[OSS诊断] ❌ bucket(internal)=None") else: - logger.info("[OSS诊断] ✅ bucket 已配置,预签名URL可用") + logger.info("[OSS诊断] ✅ bucket(internal) 就绪") + if self.public_bucket is None: + logger.error("[OSS诊断] ❌ public_bucket=None") + else: + logger.info("[OSS诊断] ✅ public_bucket 就绪,公网签名URL可用") # ── 工具方法 ─────────────────────────────────────────────────────── @@ -122,20 +168,15 @@ class SharedStorageService(StoragePort): return path.startswith(f"{self.local_url_prefix}/") def _normalize_storage_key(self, storage_key_or_url: str) -> str: - """从 URL 提取存储键,并做 URL 解码。 - - 防止 URL 编码的字符(空格=%20、中文=%XX)导致签名不匹配。 - """ if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"): parsed = urlparse(storage_key_or_url) return unquote(parsed.path.lstrip("/")) return storage_key_or_url.lstrip("/") def normalize_storage_key(self, storage_key_or_url: str) -> str: - """从 URL 提取存储键(公开方法)。""" return self._normalize_storage_key(storage_key_or_url) - # ── 上传 ─────────────────────────────────────────────────────────── + # ── 上传(SDK 走 internal endpoint)─────────────────────────────── def upload_file( self, @@ -143,21 +184,22 @@ class SharedStorageService(StoragePort): storage_key: str, content_type: str = "application/octet-stream", ) -> str: - """上传文件到存储,返回公开 URL(简单上传,API端原有行为)。 - - - 路径字符串 → bucket.put_object_from_file - - 类文件对象 → bucket.put_object - - bucket未配置 → 抛 RuntimeError - """ if self.bucket is None: raise RuntimeError("OSS storage is not configured") - try: if isinstance(file_or_path, (str, Path)): - self.bucket.put_object_from_file(storage_key, str(file_or_path), headers={"Content-Type": content_type}) + self.bucket.put_object_from_file( + storage_key, + str(file_or_path), + headers={"Content-Type": content_type}, + ) else: - file_or_path.seek(0) # type: ignore[attr-defined] - self.bucket.put_object(storage_key, file_or_path, headers={"Content-Type": content_type}) + file_or_path.seek(0) + self.bucket.put_object( + storage_key, + file_or_path, + headers={"Content-Type": content_type}, + ) return f"{self.public_url}/{storage_key}" except Exception as e: raise Exception(f"Failed to upload file to OSS: {e}") from e @@ -167,14 +209,6 @@ class SharedStorageService(StoragePort): local_path: str | Path, storage_key: str, ) -> Optional[str]: - """智能上传:大文件自动分片+超时保护(从 oss_helpers 合并)。 - - - 大文件(>100MB)走分片上传,3 线程并发 - - 总超时 300s,防止网络异常时挂死 - - 成功返回 URL,失败返回 None(不抛异常) - - Worker端 oss_helpers.upload_to_oss 的统一入口。 - """ local_path = Path(local_path) if not local_path.exists(): logger.error("上传文件不存在: %s", local_path) @@ -198,10 +232,10 @@ class SharedStorageService(StoragePort): if use_multipart: logger.info( - "大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d", + "大文件分片上传(internal): key=%s size=%.1fMB part=%dMB threads=%d", storage_key[:80], - file_size / 1024 / 1024, - OSS_PART_SIZE // 1024 // 1024, + file_size / 1048576, + OSS_PART_SIZE // 1048576, OSS_MULTIPART_NUM_THREADS, ) oss2.resumable_upload( @@ -214,7 +248,6 @@ class SharedStorageService(StoragePort): ) else: self.bucket.put_object_from_file(storage_key, str(local_path)) - result["url"] = f"{self.public_url}/{storage_key}" except Exception as e: result["error"] = e @@ -222,73 +255,54 @@ class SharedStorageService(StoragePort): finally: done.set() - upload_thread = threading.Thread(target=_do_upload, daemon=True) - upload_thread.start() + t = threading.Thread(target=_do_upload, daemon=True) + t.start() finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT) if not finished: logger.error( - "OSS 上传超时(%.0fs),强制中止: storage_key=%s, size=%.1fMB", + "OSS 上传超时(%ds): key=%s size=%.1fMB", OSS_UPLOAD_TOTAL_TIMEOUT, storage_key[:80], - result["file_size"] / 1024 / 1024 if result["file_size"] else 0, + result["file_size"] / 1048576 if result["file_size"] else 0, ) return None + return None if result["error"] else result["url"] - if result["error"]: - return None - - return result["url"] - - # ── 下载 ─────────────────────────────────────────────────────────── + # ── 下载(SDK 走 internal endpoint)─────────────────────────────── def download_file(self, storage_key: str, local_path: str | Path) -> None: - """从 OSS 下载文件(简单下载,API端原有行为)。 - - bucket未配置 → 抛 RuntimeError - """ if self.bucket is None: raise RuntimeError("OSS storage is not configured") - local_path = Path(local_path) os.makedirs(local_path.parent, exist_ok=True) try: - self.bucket.get_object_to_file(self._normalize_storage_key(storage_key), str(local_path)) + self.bucket.get_object_to_file( + self._normalize_storage_key(storage_key), + str(local_path), + ) except Exception as e: raise Exception(f"Failed to download file from OSS: {e}") from e def download_asset(self, asset_storage_key: str, local_path: str | Path) -> bool: - """下载素材(从 oss_helpers 合并)。 - - 自动识别输入类型: - - 完整 URL → 走 HTTP 下载(支持预签名URL) - - 存储键 → 走 oss2 SDK 下载 - - 成功返回 True,失败返回 False(不抛异常)。 - """ local_path = Path(local_path) os.makedirs(local_path.parent, exist_ok=True) - - # 完整URL走HTTP下载(兼容预签名URL) if asset_storage_key.startswith(("http://", "https://")): return self._download_via_http(asset_storage_key, local_path) - - # OSS存储键走SDK if self.bucket is None: - logger.error("OSS not configured, cannot download: %s", asset_storage_key[:80]) + logger.error("OSS not configured: %s", asset_storage_key[:80]) return False try: - self.bucket.get_object_to_file(self._normalize_storage_key(asset_storage_key), str(local_path)) + self.bucket.get_object_to_file( + self._normalize_storage_key(asset_storage_key), + str(local_path), + ) return local_path.exists() and local_path.stat().st_size > 0 except Exception: logger.exception("下载素材失败: %s", asset_storage_key) return False def _download_via_http(self, url: str, local_path: Path) -> bool: - """通过 HTTP 下载文件(支持预签名 URL)。 - - 流式下载避免大文件内存溢出。 - """ try: resp = requests.get(url, stream=True, timeout=OSS_HTTP_DOWNLOAD_TIMEOUT) resp.raise_for_status() @@ -298,83 +312,64 @@ class SharedStorageService(StoragePort): f.write(chunk) return local_path.exists() and local_path.stat().st_size > 0 except Exception: - logger.exception("HTTP下载素材失败: %s", url[:100]) + logger.exception("HTTP下载失败: %s", url[:100]) return False - # ── URL 生成 ────────────────────────────────────────────────────── + # ── URL 生成(sign_url 用 public_bucket 签公网域名)─────────────── def get_url(self, storage_key: str) -> str: - """获取公开 URL。""" return f"{self.public_url}/{storage_key}" - def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str: - """获取预签名下载 URL。 + def _sign_bucket(self): + """签名优先用 public_bucket,回退到 bucket。""" + return self.public_bucket or self.bucket - bucket未配置时降级为公开URL;本地产物URL直接返回。 - """ - if self.bucket is None: + def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str: + sign_bucket = self._sign_bucket() + if sign_bucket is None: if self._is_local_generated_url(storage_key_or_url): return storage_key_or_url - logger.warning( - "get_download_url: OSS bucket not configured, returning raw URL. key=%s", - storage_key_or_url[:200], - ) + logger.warning("OSS bucket not configured, returning raw URL: %s", storage_key_or_url[:200]) return self.get_url(self.normalize_storage_key(storage_key_or_url)) storage_key = self.normalize_storage_key(storage_key_or_url) try: - signed = self.bucket.sign_url("GET", storage_key, expires_seconds) + signed = sign_bucket.sign_url("GET", storage_key, expires_seconds) logger.info( - "get_download_url: signed URL generated. key=%s url_prefix=%s", + "signed URL generated for key=%s prefix=%s", storage_key[:80], signed[:60], ) return signed except Exception: - logger.exception( - "get_download_url: sign_url failed, falling back to raw URL. key=%s", - storage_key[:200], - ) + logger.exception("get_download_url: sign_url 失败,返回 raw URL: %s", storage_key[:200]) return self.get_url(storage_key) - # ── 浏览器直传 POST ──────────────────────────────────────────────── - def get_upload_url( self, storage_key_or_url: str, expires_seconds: int = 3600, content_type: str = "video/mp4", ) -> str: - """获取预签名 PUT 上传 URL(供外部 Worker 上传结果文件)。 - - bucket未配置时降级为 public_url(本地/开发环境); - 本地产物 key 原样返回。 - """ - if self.bucket is None: + sign_bucket = self._sign_bucket() + if sign_bucket is None: if self._is_local_generated_url(storage_key_or_url): return storage_key_or_url - logger.warning( - "get_upload_url: OSS bucket not configured, returning raw URL. key=%s", - storage_key_or_url[:200], - ) + logger.warning("get_upload_url: OSS 未配置,返回 raw URL: %s", storage_key_or_url[:200]) return self.get_url(self.normalize_storage_key(storage_key_or_url)) storage_key = self.normalize_storage_key(storage_key_or_url) try: - # oss2 sign_url 支持 'PUT',需指定 headers 才能限定 Content-Type headers = {"Content-Type": content_type} if content_type else None - signed = self.bucket.sign_url("PUT", storage_key, expires_seconds, headers=headers) + signed = sign_bucket.sign_url("PUT", storage_key, expires_seconds, headers=headers) logger.info( - "get_upload_url: signed PUT URL generated. key=%s url_prefix=%s", + "get_upload_url: 公网签名PUT URL已生成 key=%s prefix=%s", storage_key[:80], signed[:60], ) return signed except Exception: - logger.exception( - "get_upload_url: sign_url failed, falling back to raw URL. key=%s", - storage_key[:200], - ) + logger.exception("get_upload_url: sign_url 失败,返回 raw URL: %s", storage_key[:200]) return self.get_url(storage_key) def create_direct_upload_post( @@ -384,7 +379,6 @@ class SharedStorageService(StoragePort): max_size_bytes: int, expires_seconds: int, ) -> dict[str, object]: - """创建浏览器直传 POST 表单。""" if not self.access_key_id or not self.access_key_secret: raise RuntimeError("OSS storage is not configured") normalized_key = self.normalize_storage_key(storage_key) @@ -431,19 +425,17 @@ class SharedStorageService(StoragePort): }, } - # ── 文件操作 ─────────────────────────────────────────────────────── + # ── 文件操作(internal endpoint)────────────────────────────────── def delete_file(self, storage_key: str) -> None: - """删除文件(不抛异常)。""" if self.bucket is None: return try: self.bucket.delete_object(storage_key) except Exception as error: - logger.warning("Failed to delete file from OSS", extra={"storage_key": storage_key, "error": str(error)}) + logger.warning("OSS delete 失败", extra={"storage_key": storage_key, "error": str(error)}) def file_exists(self, storage_key: str) -> bool: - """检查文件是否存在。""" if self.bucket is None: return False return self.bucket.object_exists(storage_key) @@ -451,17 +443,6 @@ class SharedStorageService(StoragePort): # ── Asset 路径解析(Worker 用)──────────────────────────────────── def resolve_asset_path(self, asset_id: str, work_dir: str | Path) -> Optional[Path]: - """从 asset_id 解析到本地文件路径。 - - 策略(按优先级): - 1. 本地绝对路径(在允许目录内)→ 直接返回 - 2. work_dir 缓存命中 → 返回缓存路径 - 3. 从OSS下载到缓存 → 返回下载路径 - 4. 全部失败 → None - - 从 oss_helpers.resolve_asset_path 合并而来。 - """ - # 延迟导入,避免循环依赖 from video_processing.path_security import ( # type: ignore[import-not-found] PathSecurityError, get_allowed_local_dirs, @@ -471,47 +452,35 @@ class SharedStorageService(StoragePort): if not asset_id or not isinstance(asset_id, str): return None - work_dir = Path(work_dir) os.makedirs(work_dir, exist_ok=True) - - # 空字节检测 if "\x00" in asset_id: - logger.warning("asset_id 包含空字节,拒绝: %s", asset_id[:50]) + logger.warning("asset_id 含空字节,拒绝: %s", asset_id[:50]) return None - - # 1. 本地绝对路径 — 必须在允许的目录内 if asset_id.startswith("/") and os.path.exists(asset_id): try: resolved = Path(asset_id).resolve() if is_in_allowed_dirs(resolved, get_allowed_local_dirs()): return resolved - else: - logger.warning( - "本地素材路径不在允许目录内,拒绝: %s (allowed=%s)", - asset_id[:80], - get_allowed_local_dirs(), - ) - return None + logger.warning( + "本地素材路径不在允许目录: %s allowed=%s", + asset_id[:80], + get_allowed_local_dirs(), + ) + return None except (OSError, PathSecurityError): return None - - # 2. 缓存命中(SHA256 hash 防路径遍历) cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16] safe_name = sanitize_filename(cache_hash) cached_path = work_dir / f"{safe_name}.mp4" if cached_path.exists() and cached_path.stat().st_size > 0: return cached_path - - # 3. 从 OSS 下载(先标准化 key,防路径遍历注入) safe_key = self.normalize_storage_key(asset_id) if ".." in safe_key or safe_key.startswith("/"): - logger.warning("asset_id 包含路径遍历模式,拒绝下载: %s", asset_id[:80]) + logger.warning("asset_id 含路径遍历: %s", asset_id[:80]) return None - if self.download_asset(safe_key, cached_path): return cached_path - return None def resolve_asset_ids_to_paths( @@ -519,22 +488,20 @@ class SharedStorageService(StoragePort): asset_ids: list[str], work_dir: str | Path, ) -> dict[str, Path]: - """批量解析 asset_id → 本地路径。""" result: dict[str, Path] = {} for aid in asset_ids: - local_path = self.resolve_asset_path(aid, work_dir) - if local_path: - result[aid] = local_path + p = self.resolve_asset_path(aid, work_dir) + if p: + result[aid] = p return result -# ── 单例管理 ──────────────────────────────────────────────────────────── +# ── 单例 ──────────────────────────────────────────────────────────────── _storage_service: Optional[SharedStorageService] = None def get_shared_storage_service() -> SharedStorageService: - """获取统一存储服务单例。""" global _storage_service if _storage_service is None: _storage_service = SharedStorageService() @@ -542,7 +509,6 @@ def get_shared_storage_service() -> SharedStorageService: return _storage_service -# 向后兼容别名 def get_storage_service() -> SharedStorageService: - """向后兼容:返回统一存储服务。""" + """向后兼容别名。""" return get_shared_storage_service()