deepdata/tools/minio-file-server/main.py

import base64
import uuid
import io
import urllib.parse
from typing import Optional
from fastapi import FastAPI, UploadFile, File, Query, HTTPException, Body
from fastapi.responses import StreamingResponse
from pydantic_settings import BaseSettings, SettingsConfigDict
from minio import Minio
from minio.error import S3Error
import argparse

import base64
import uuid
import io
import urllib.parse
import mimetypes
import os
from typing import Optional
from fastapi import FastAPI, UploadFile, File, Query, HTTPException, Body
from fastapi.responses import StreamingResponse
from pydantic_settings import BaseSettings, SettingsConfigDict
from minio import Minio
from minio.error import S3Error

# 加载环境变量


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8")
    minio_endpoint: str
    minio_access_key: str
    minio_secret_key: str
    minio_secure: bool
    minio_bucket: str
    public_prefix: str


settings = Settings()

# MinIO Client
minio_client = Minio(
    endpoint=settings.minio_endpoint,
    access_key=settings.minio_access_key,
    secret_key=settings.minio_secret_key,
    secure=settings.minio_secure
)

# 初始化FastAPI
app = FastAPI(
    title="MinIO 文件服务",
    description="原始文件名存入MinIO对象元数据；上传仅返回MD图片相对路径，访问时自动读取元数据作为下载名称",
    version="1.0.0"
)


def get_ext(filename: str) -> str:
    if "." in filename:
        return filename.split(".")[-1]
    return ""


def build_relative_path(obj_key: str) -> str:
    """生成用于Markdown图片的相对路径"""
    return f"/file/{obj_key}"


@app.post("/upload/form", summary="Form表单上传，仅返回相对路径，原始文件名存入MinIO元数据")
async def upload_form_file(
    file: UploadFile = File(..., description="文件"),
    filename: Optional[str] = Query(
        default=None, description="自定义文件名，不传使用原始文件名")
):
    use_filename = filename or file.filename or "file"
    ext = get_ext(use_filename)
    obj_key = f"{settings.public_prefix}/{uuid.uuid4()}.{ext}"
    # ✅ URL编码中文文件名，解决header latin1编码报错
    metadata = {
        "X-Amz-Meta-Original-Name": urllib.parse.quote(use_filename)
    }
    try:
        minio_client.put_object(
            bucket_name=settings.minio_bucket,
            object_name=obj_key,
            data=file.file,
            length=-1,
            part_size=10*1024*1024,
            content_type=file.content_type,
            metadata=metadata
        )
        path = build_relative_path(obj_key)
        return {"path": path}
    except S3Error as e:
        raise HTTPException(status_code=500, detail=f"MinIO上传失败: {str(e)}")


@app.post("/upload/base64", summary="Base64上传，仅返回相对路径，原始文件名存入MinIO元数据")
async def upload_base64(
    base64_data: str = Body(),
    filename: str = Body()
):
    if base64_data.startswith("data:"):
        base64_data = base64_data.split(",")[-1]
    try:
        raw_bytes = base64.b64decode(base64_data)
    except Exception:
        raise HTTPException(status_code=400, detail="base64数据解码失败")

    ext = get_ext(filename)
    obj_key = f"{settings.public_prefix}/{uuid.uuid4()}.{ext}"
    # ✅ URL编码中文文件名
    metadata = {
        "X-Amz-Meta-Original-Name": urllib.parse.quote(filename)
    }
    # 猜测content-type
    content_type = "application/octet-stream"
    if ext.lower() in ["jpg", "jpeg"]:
        content_type = "image/jpeg"
    elif ext.lower() == "png":
        content_type = "image/png"
    elif ext.lower() == "gif":
        content_type = "image/gif"
    elif ext.lower() == "pdf":
        content_type = "application/pdf"
    elif ext.lower() == "txt":
        content_type = "text/plain"

    stream = io.BytesIO(raw_bytes)
    try:
        minio_client.put_object(
            bucket_name=settings.minio_bucket,
            object_name=obj_key,
            data=stream,
            length=len(raw_bytes),
            content_type=content_type,
            metadata=metadata
        )
        path = build_relative_path(obj_key)
        return {"path": path}
    except S3Error as e:
        raise HTTPException(status_code=500, detail=f"MinIO上传失败: {str(e)}")


@app.get("/file/{object_key:path}", summary="文件访问：默认预览，dl=1强制下载，自动读取MinIO元数据中的原始文件名")
def file_access(
    object_key: str,
    dl: int = Query(default=0, description="dl=1强制下载；默认0在线预览"),
    name: str = Query(default="", description="手动指定下载名称，优先级高于元数据")
):
    content_type = "application/octet-stream"
    original_name = ""
    stream = None

    # 1. 尝试使用 HEAD 请求获取元数据 (快，不消耗流量)
    try:
        stat = minio_client.stat_object(settings.minio_bucket, object_key)
        content_type = stat.content_type

        # 提取原始文件名 (SDK 通常会去掉 x-amz-meta- 前缀并转小写)
        original_name = stat.metadata.get("original-name", "")

        stream = minio_client.get_object(settings.minio_bucket, object_key)
    except S3Error as e:
        # 如果是文件不存在，直接 404
        if e.code == "NoSuchKey":
            raise HTTPException(status_code=404, detail=f"文件不存在: {str(e)}")

        # 2. 降级策略：如果 HEAD 失败 (如 Nginx 代理丢弃鉴权头导致 AccessDenied)，尝试 GET 请求
        # 这能解决 stat_object 报错但实际有权限读取文件的问题
        try:
            response = minio_client.get_object(
                settings.minio_bucket, object_key)
            stream = response

            # 从 HTTP Header 中提取信息
            if response.headers.get("content-type"):
                content_type = response.headers["content-type"]

            # 提取原始文件名 (Header 中通常包含完整前缀 x-amz-meta-*)
            # 兼容大小写
            meta_val = response.headers.get("x-amz-meta-original-name") or \
                response.headers.get("X-Amz-Meta-Original-Name")
            if meta_val:
                original_name = meta_val

        except S3Error as get_err:
            # 如果 GET 也失败，抛出错误
            raise HTTPException(status_code=404 if get_err.code == "NoSuchKey" else 500,
                                detail=f"S3 操作失败: {str(get_err)}")
        except Exception as get_err:
            raise HTTPException(
                status_code=500, detail=f"读取文件流失败: {str(get_err)}")

    # 3. Content-Type 兜底推断 (处理未设置 Content-Type 的文件)
    if content_type == "application/octet-stream":
        guessed_type, _ = mimetypes.guess_type(object_key)
        if guessed_type:
            content_type = guessed_type
        else:
            # 针对常见后缀手动映射，防止 mimetypes 未覆盖
            ext = object_key.split(".")[-1].lower()
            type_map = {
                'mp4': 'video/mp4', 'webm': 'video/webm', 'mov': 'video/quicktime',
                'zip': 'application/zip', 'rar': 'application/x-rar-compressed',
                '7z': 'application/x-7z-compressed', 'tar': 'application/x-tar',
                'gz': 'application/gzip', 'pdf': 'application/pdf'
            }
            content_type = type_map.get(ext, 'application/octet-stream')

    # 4. 设置响应头
    resp_headers = {}

    # 确定最终下载文件名
    # 优先级：URL参数 name > 元数据/HTTP Header > 对象 Key 中的文件名
    final_name = name or urllib.parse.unquote(
        original_name) or object_key.split("/")[-1]

    if dl == 1:
        # 强制下载
        encoded_name = urllib.parse.quote(final_name)
        resp_headers["Content-Disposition"] = f'attachment; filename="{final_name}"; filename*=utf-8\'\'{encoded_name}'
    else:
        # 在线预览 (浏览器默认行为，显式设置 inline 可避免某些浏览器强制下载)
        pass

    return StreamingResponse(
        stream,
        media_type=content_type,
        headers=resp_headers
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='')
    parser.add_argument("--port", type=int,
                        default=8000, help="缺省: 8000")
    args = parser.parse_args()

    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=args.port)