shanxia2.0/backend/app/services/bx/storage.py

"""报销附件存储适配 — 复用门户 app.core.storage，对外保持 bx 协议。"""
from __future__ import annotations

from io import BytesIO
from pathlib import Path

from app.core.config import settings
from app.core.storage import LocalStorage, MinIOStorage, get_storage, storage


def normalize_storage_path(file_url: str) -> str:
    """将 upload 返回的完整 URL 归一化为存储相对路径（与 admin 附件 upload 一致）。"""
    u = (file_url or "").strip()
    if not u:
        return ""
    if "?" in u:
        u = u.split("?", 1)[0]
    base_url = (settings.MINIO_BASE_URL or "").rstrip("/")
    if base_url and u.startswith(base_url + "/"):
        return u[len(base_url) + 1 :]
    local_base = settings.LOCAL_STORAGE_URL.rstrip("/")
    if u.startswith(local_base + "/"):
        return u[len(local_base) + 1 :]
    if u.startswith("http://") or u.startswith("https://"):
        store = get_storage()
        if isinstance(store, MinIOStorage):
            return store._extract_object_key(u)
    return u.lstrip("/")


def upload_bx_file(file_content: bytes, filename: str, *, biz_type: str = "") -> tuple[str, str]:
    """
    上传文件到对象存储。
    返回 (relative_path, access_url)。
    """
    folder = "bx"
    if biz_type:
        folder = f"bx/{biz_type.strip('/')}"
    full_url = storage.upload_file(file_content, filename, folder)
    relative_path = normalize_storage_path(full_url)
    return relative_path, full_url


def get_access_url(file_path: str) -> str:
    """获取可访问 URL（本地/MinIO 公开地址）。"""
    store = get_storage()
    key = normalize_storage_path(file_path)
    if not key:
        return ""
    return store.get_url(key)


def get_preview_url(file_path: str, *, expires_seconds: int = 3600) -> str:
    """图片/附件预览 URL：MinIO 预签名 inline，本地/其他存储用 get_url。"""
    store = get_storage()
    key = normalize_storage_path(file_path)
    if not key:
        return ""
    if isinstance(store, MinIOStorage):
        # 走 MinIOStorage.presigned_preview_url（按 MINIO_BASE_URL 公网域名签发）
        return store.presigned_preview_url(key, expires_seconds=expires_seconds)
    return store.get_url(key)


def get_download_url(file_path: str, file_name: str, *, expires_seconds: int = 3600) -> str:
    """带文件名的下载 URL（对齐门户 attachment/download-url）。"""
    store = get_storage()
    key = normalize_storage_path(file_path)
    return store.presigned_download_url(key, file_name, expires_seconds)


def download_file_bytes(file_path: str) -> bytes:
    """读取文件二进制，供 StreamingResponse 下载。"""
    store = get_storage()
    key = normalize_storage_path(file_path)
    if not key:
        raise ValueError("无效的文件路径")
    if isinstance(store, MinIOStorage):
        response = store.client.get_object(store.bucket_name, key)
        try:
            return response.read()
        finally:
            response.close()
            response.release_conn()
    if isinstance(store, LocalStorage):
        file_full_path = Path(store.upload_dir) / key
        if not file_full_path.is_file():
            raise FileNotFoundError(f"文件不存在: {key}")
        return file_full_path.read_bytes()
    # 七牛/OSS 等：回退 HTTP 拉取（开发环境较少用到）
    import urllib.request

    url = store.get_url(key)
    with urllib.request.urlopen(url, timeout=60) as resp:
        return resp.read()


def file_stream(file_path: str) -> BytesIO:
    return BytesIO(download_file_bytes(file_path))