refactor(#779): 消除重复枚举定义，统一到classification模块

- AssetLibraryKind: 从entities.py+classification.py双份定义，统一到classification.py
- IngestJobStatus: 同上
- ClassificationStatus/ClassificationJobStatus: 合并为ClassificationJobStatus，保留_missing_兼容方法，ClassificationStatus作为别名
- entities.py改为re-export兼容层，107处import不受影响
- 补齐classification.py中AssetLibraryKind缺失的IMAGE值
- 净减8行

packages/domain/classification.py

@@ -19,6 +19,7 @@ from uuid import uuid4
 class AssetLibraryKind(StrEnum):
     VIDEO = "video"
     VOICE = "voice"
+    IMAGE = "image"
 
 
 class IngestJobStatus(StrEnum):
@@ -34,6 +35,27 @@ class ClassificationJobStatus(StrEnum):
     COMPLETED = "completed"
     FAILED = "failed"
 
+    @classmethod
+    def _missing_(cls, value: object) -> "ClassificationJobStatus":
+        """兼容历史数据，避免枚举转换失败导致500。
+
+        - done → COMPLETED（早期版本用 done 表示完成）
+        - 其他未知值 → PENDING（兜底，不阻塞业务）
+        """
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("done", "success", "finished", "complete"):
+                return cls.COMPLETED
+            if normalized in ("fail", "error", "err"):
+                return cls.FAILED
+            if normalized in ("process", "processing", "running", "run"):
+                return cls.PROCESSING
+        return cls.PENDING
+
+
+# 向后兼容别名
+ClassificationStatus = ClassificationJobStatus
+
 
 class AssetClassification(StrEnum):
     """Asset classification categories."""

packages/domain/entities.py

@@ -16,18 +16,12 @@ else:
 from typing import Any
 from uuid import uuid4
 
-
-class AssetLibraryKind(StrEnum):
-    VIDEO = "video"
-    VOICE = "voice"
-    IMAGE = "image"
-
-
-class IngestJobStatus(StrEnum):
-    PENDING = "pending"
-    PROCESSING = "processing"
-    COMPLETED = "completed"
-    FAILED = "failed"
+# 枚举统一从 classification 模块导入，消除重复定义
+from packages.domain.classification import (
+    AssetLibraryKind,
+    ClassificationStatus,
+    IngestJobStatus,
+)
 
 
 @dataclass(slots=True)
@@ -160,30 +154,6 @@ class AssetStatus(StrEnum):
         return cls.READY
 
 
-class ClassificationStatus(StrEnum):
-    PENDING = "pending"
-    PROCESSING = "processing"
-    COMPLETED = "completed"
-    FAILED = "failed"
-
-    @classmethod
-    def _missing_(cls, value: object) -> "ClassificationStatus":
-        """兼容历史数据，避免枚举转换失败导致500。
-
-        - done → COMPLETED（早期版本用 done 表示完成）
-        - 其他未知值 → PENDING（兜底，不阻塞业务）
-        """
-        if isinstance(value, str):
-            normalized = value.strip().lower()
-            if normalized in ("done", "success", "finished", "complete"):
-                return cls.COMPLETED
-            if normalized in ("fail", "error", "err"):
-                return cls.FAILED
-            if normalized in ("process", "processing", "running", "run"):
-                return cls.PROCESSING
-        return cls.PENDING
-
-
 @dataclass(slots=True)
 class Asset:
     id: str

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:
+        """输出存储配置诊断日志。"""
+        ...

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()