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)