xiaoxia-saas/packages/ports/storage_port.py

"""Storage 端口接口 — 统一存储服务的抽象定义。

所有存储实现（OSS、本地、S3等）都必须实现这个端口。
API 和 Worker 都通过这个端口与存储交互，消除两套独立实现。
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from pathlib import Path
from typing import Optional


class StoragePort(ABC):
    """统一存储服务端口。

    定义所有存储后端必须实现的核心能力。
    具体实现见 packages.shared.storage.SharedStorageService。
    """

    # ── 基础上传 / 下载 ────────────────────────────────────────────────

    @abstractmethod
    def upload_file(
        self,
        file_or_path: 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: 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: 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:
        """输出存储配置诊断日志。"""
        ...