diff --git a/packages/ports/storage_port.py b/packages/ports/storage_port.py new file mode 100755 index 000000000..b02fa4010 --- /dev/null +++ b/packages/ports/storage_port.py @@ -0,0 +1,118 @@ +"""Storage 端口接口 — 统一存储服务的抽象定义。 + +所有存储实现（OSS、本地、S3等）都必须实现这个端口。 +API 和 Worker 都通过这个端口与存储交互，消除两套独立实现。 +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Optional, Union + + +class StoragePort(ABC): + """统一存储服务端口。 + + 定义所有存储后端必须实现的核心能力。 + 具体实现见 packages.shared.storage.SharedStorageService。 + """ + + # ── 基础上传 / 下载 ──────────────────────────────────────────────── + + @abstractmethod + def upload_file( + self, + file_or_path: Union[str, Path, object], + storage_key: str, + content_type: str = "application/octet-stream", + ) -> str: + """上传文件到存储，返回公开 URL。 + + Args: + file_or_path: 本地文件路径（str/Path）或类文件对象 + storage_key: 目标存储键 + content_type: MIME 类型 + + Returns: + 公开访问 URL + """ + ... + + @abstractmethod + def download_file(self, storage_key_or_url: str, local_path: Union[str, Path]) -> bool: + """从存储下载文件到本地。 + + 自动识别输入：完整URL走HTTP下载（支持预签名），存储键走SDK下载。 + + Args: + storage_key_or_url: 存储键或完整 URL + local_path: 本地保存路径 + + Returns: + True 成功，False 失败 + """ + ... + + # ── URL 生成 ────────────────────────────────────────────────────── + + @abstractmethod + def get_url(self, storage_key: str) -> str: + """获取公开 URL。""" + ... + + @abstractmethod + def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str: + """获取预签名下载 URL（私有 bucket 用）。 + + 未配置OSS时降级为公开URL。 + """ + ... + + # ── 文件操作 ────────────────────────────────────────────────────── + + @abstractmethod + def delete_file(self, storage_key: str) -> None: + """删除文件（不抛异常）。""" + ... + + @abstractmethod + def file_exists(self, storage_key: str) -> bool: + """检查文件是否存在。""" + ... + + # ── 浏览器直传 ──────────────────────────────────────────────────── + + @abstractmethod + def create_direct_upload_post( + self, + storage_key: str, + content_type: str, + max_size_bytes: int, + expires_seconds: int, + ) -> dict[str, object]: + """创建浏览器直传 POST 表单（用于前端直传OSS）。""" + ... + + # ── Asset 解析（Worker 用）──────────────────────────────────────── + + @abstractmethod + def resolve_asset_path(self, asset_id: str, work_dir: Union[str, Path]) -> Optional[Path]: + """从 asset_id 解析到本地文件路径。 + + 策略：本地路径 → 缓存命中 → OSS下载 → None + 缓存：SHA256(asset_id)[:16] 为文件名，避免重复下载 + """ + ... + + # ── 工具方法 ────────────────────────────────────────────────────── + + @abstractmethod + def normalize_storage_key(self, storage_key_or_url: str) -> str: + """从 URL 提取存储键，URL decode 处理。""" + ... + + @abstractmethod + def diagnose(self) -> None: + """输出存储配置诊断日志。""" + ... diff --git a/packages/shared/storage.py b/packages/shared/storage.py index 92cec6156..fcf9997f0 100755 --- a/packages/shared/storage.py +++ b/packages/shared/storage.py @@ -1,4 +1,13 @@ -"""Shared OSS storage service for API and Worker.""" +"""统一存储服务 — API 和 Worker 共用的唯一存储入口。 + +实现 StoragePort 端口接口，整合原来分散在各处的存储能力： +- API端 SharedStorageService 的全部能力（上传/下载/签名URL/直传POST） +- Worker端 oss_helpers 的高级能力（分片上传/超时保护/HTTP下载/Asset路径解析） + +所有服务都通过这个统一入口与存储交互，消除重复实现。 +""" + +from __future__ import annotations import base64 import datetime as dt @@ -7,53 +16,70 @@ import hmac import json import logging import os -from typing import Optional +import threading +from pathlib import Path +from typing import Optional, Union from urllib.parse import unquote, urlparse +import requests + try: import oss2 except ImportError: # pragma: no cover oss2 = None -from packages.shared.config import get_shared_settings +from packages.config import get_shared_settings +from packages.ports.storage_port import StoragePort logger = logging.getLogger(__name__) +# ── OSS 高级配置（从 oss_helpers 合并）───────────────────────────────── +OSS_CONNECT_TIMEOUT = 10 # 连接超时（秒） +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_HTTP_DOWNLOAD_TIMEOUT = 300 # HTTP下载超时（秒） -class SharedStorageService: - """Shared OSS storage service.""" + +class SharedStorageService(StoragePort): + """统一存储服务 — 实现 StoragePort，API 和 Worker 共用。 + + 整合了原 SharedStorageService + oss_helpers 的全部能力。 + """ 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.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files") self.bucket = None - has_key_id = bool(settings.oss_access_key_id) - has_key_secret = bool(settings.oss_access_key_secret) + self.access_key_id = settings.oss_access_key_id + self.access_key_secret = settings.oss_access_key_secret + + has_key_id = bool(self.access_key_id) + has_key_secret = bool(self.access_key_secret) if has_key_id and has_key_secret: if oss2 is not None: try: - # P0-2 修复：oss2.Bucket 的 endpoint 必须带 https:// 前缀， - # 否则 sign_url 默认生成 HTTP URL。 - bucket_endpoint = settings.oss_endpoint + # endpoint 不带 scheme 时补 https:// 前缀 + bucket_endpoint = self.endpoint if not bucket_endpoint.startswith(("http://", "https://")): bucket_endpoint = f"https://{bucket_endpoint}" - auth = oss2.Auth( - settings.oss_access_key_id, - settings.oss_access_key_secret, - ) + auth = oss2.Auth(self.access_key_id, self.access_key_secret) self.bucket = oss2.Bucket( auth, bucket_endpoint, - settings.oss_bucket_name, + self.bucket_name, + connect_timeout=OSS_CONNECT_TIMEOUT, ) logger.info( "OSS initialized: endpoint=%s bucket=%s", - settings.oss_endpoint, - settings.oss_bucket_name, + self.endpoint, + self.bucket_name, ) except Exception as error: logger.error("Failed to initialize OSS bucket client: %s", error) @@ -67,14 +93,14 @@ class SharedStorageService: missing.append("OSS_ACCESS_KEY_SECRET") logger.error("OSS credentials not configured — missing: %s", ", ".join(missing)) - self.access_key_id = settings.oss_access_key_id - self.access_key_secret = settings.oss_access_key_secret - self.endpoint = settings.oss_endpoint + # ── 诊断 ─────────────────────────────────────────────────────────── def diagnose(self) -> None: - """启动诊断：输出 OSS 配置状态，帮助排查预签名 URL 问题。""" + """输出存储配置诊断日志。""" key_id_display = ( - f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)" + 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", @@ -86,89 +112,238 @@ class SharedStorageService: logger.error( "[OSS诊断] ❌ bucket=None — 预签名URL不可用！" "原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。" - "请检查服务器 .env 文件（如 /var/lib/xiaoxia-saas-staging/.env）" ) else: logger.info("[OSS诊断] ✅ bucket 已配置，预签名URL可用") + # ── 工具方法 ─────────────────────────────────────────────────────── + def _is_local_generated_url(self, storage_key_or_url: str) -> bool: parsed = urlparse(storage_key_or_url) path = parsed.path if parsed.scheme else storage_key_or_url 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) + + # ── 上传 ─────────────────────────────────────────────────────────── + def upload_file( self, - file_or_path, + file_or_path: Union[str, Path, object], storage_key: str, content_type: str = "application/octet-stream", ) -> str: - """Upload file to OSS.""" + """上传文件到存储，返回公开 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): - self.bucket.put_object_from_file(storage_key, file_or_path, headers={"Content-Type": content_type}) + if isinstance(file_or_path, (str, Path)): + self.bucket.put_object_from_file( + storage_key, str(file_or_path), headers={"Content-Type": content_type} + ) else: - file_or_path.seek(0) + file_or_path.seek(0) # type: ignore[attr-defined] 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 - def get_url(self, storage_key: str) -> str: - """Get public URL for a file.""" - return f"{self.public_url}/{storage_key}" + def upload_file_smart( + self, + local_path: Union[str, Path], + storage_key: str, + ) -> Optional[str]: + """智能上传：大文件自动分片+超时保护（从 oss_helpers 合并）。 - def download_file(self, storage_key: str, local_path: str): - """Download file from OSS to local path.""" + - 大文件（>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) + return None + if self.bucket is None: + logger.error("OSS未配置，无法上传: %s", storage_key[:80]) + return None + + result: dict = {"url": None, "error": None, "file_size": 0} + done = threading.Event() + + def _do_upload(): + try: + 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: + 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( + self.bucket, + storage_key, + str(local_path), + multipart_threshold=OSS_MULTIPART_THRESHOLD, + part_size=OSS_PART_SIZE, + num_threads=OSS_MULTIPART_NUM_THREADS, + ) + 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 + 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, + ) + return None + + if result["error"]: + return None + + return result["url"] + + # ── 下载 ─────────────────────────────────────────────────────────── + + def download_file(self, storage_key: str, local_path: Union[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: - os.makedirs(os.path.dirname(local_path), exist_ok=True) - self.bucket.get_object_to_file(storage_key, 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: Union[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]) + return False + try: + 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() + 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[:100]) + return False + + # ── URL 生成 ────────────────────────────────────────────────────── + + 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: - """Get signed download URL.""" + """获取预签名下载 URL。 + + bucket未配置时降级为公开URL；本地产物URL直接返回。 + """ if self.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. storage_key_or_url=%s", + "get_download_url: OSS bucket not configured, returning raw URL. key=%s", storage_key_or_url[:200], ) - return self.get_url(self._normalize_storage_key(storage_key_or_url)) + return self.get_url(self.normalize_storage_key(storage_key_or_url)) - storage_key = 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) logger.info( - "get_download_url: signed URL generated. storage_key=%s url_prefix=%s", + "get_download_url: signed URL generated. key=%s url_prefix=%s", storage_key[:80], signed[:60], ) return signed except Exception: logger.exception( - "get_download_url: sign_url failed, falling back to raw URL. storage_key=%s", + "get_download_url: sign_url failed, falling back to raw URL. key=%s", storage_key[:200], ) return self.get_url(storage_key) - def _normalize_storage_key(self, storage_key_or_url: str) -> str: - """Extract storage key from URL. - - 从完整 URL 提取 OSS 存储键，并做 URL 解码 — 否则 URL 编码的字符 - （如空格=%20、中文=%XX）会导致 sign_url 计算的签名与 OSS 服务端 - 不匹配（SignatureDoesNotMatch）。原始 key 传入时直接返回。 - """ - 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("/") + # ── 浏览器直传 POST ──────────────────────────────────────────────── def create_direct_upload_post( self, @@ -177,10 +352,10 @@ class SharedStorageService: max_size_bytes: int, expires_seconds: int, ) -> dict[str, object]: - """Create browser direct upload POST form.""" + """创建浏览器直传 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) + normalized_key = self.normalize_storage_key(storage_key) if not normalized_key.startswith("uploads/"): raise ValueError("direct upload key must be under uploads/") @@ -193,12 +368,22 @@ class SharedStorageService: {"bucket": self.bucket_name}, {"key": normalized_key}, ["content-length-range", 1, max_size_bytes], - ["starts-with", "$Content-Type", content_type.split("/", 1)[0] + "/" if "/" in content_type else ""], + [ + "starts-with", + "$Content-Type", + content_type.split("/", 1)[0] + "/" if "/" in content_type else "", + ], ], } - encoded_policy = base64.b64encode(json.dumps(policy, separators=(",", ":")).encode("utf-8")).decode("ascii") + encoded_policy = base64.b64encode( + json.dumps(policy, separators=(",", ":")).encode("utf-8") + ).decode("ascii") signature = base64.b64encode( - hmac.new(self.access_key_secret.encode("utf-8"), encoded_policy.encode("utf-8"), hashlib.sha1).digest() + hmac.new( + self.access_key_secret.encode("utf-8"), + encoded_policy.encode("utf-8"), + hashlib.sha1, + ).digest() ).decode("ascii") return { @@ -216,8 +401,10 @@ class SharedStorageService: }, } - def delete_file(self, storage_key: str): - """Delete file from OSS.""" + # ── 文件操作 ─────────────────────────────────────────────────────── + + def delete_file(self, storage_key: str) -> None: + """删除文件（不抛异常）。""" if self.bucket is None: return try: @@ -226,17 +413,98 @@ class SharedStorageService: logger.warning("Failed to delete file from OSS", extra={"storage_key": storage_key, "error": str(error)}) def file_exists(self, storage_key: str) -> bool: - """Check if file exists.""" + """检查文件是否存在。""" if self.bucket is None: return False return self.bucket.object_exists(storage_key) + # ── Asset 路径解析（Worker 用）──────────────────────────────────── + + def resolve_asset_path(self, asset_id: str, work_dir: Union[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, + is_in_allowed_dirs, + sanitize_filename, + ) + + 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]) + 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 + + # 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]) + return None + + if self.download_asset(safe_key, cached_path): + return cached_path + + return None + + def resolve_asset_ids_to_paths( + self, + asset_ids: list[str], + work_dir: Union[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 + return result + + +# ── 单例管理 ──────────────────────────────────────────────────────────── _storage_service: Optional[SharedStorageService] = None def get_shared_storage_service() -> SharedStorageService: - """Get shared storage service instance (global singleton).""" + """获取统一存储服务单例。""" global _storage_service if _storage_service is None: _storage_service = SharedStorageService() @@ -244,7 +512,7 @@ def get_shared_storage_service() -> SharedStorageService: return _storage_service -# Backward compatibility alias +# 向后兼容别名 def get_storage_service() -> SharedStorageService: - """Backward compatibility: returns shared storage service.""" + """向后兼容：返回统一存储服务。""" return get_shared_storage_service()