Upload minio_uploader.py

minio_uploader.py

@@ -0,0 +1,102 @@
+# core/minio_uploader.py
+import os
+from minio import Minio
+from minio.error import S3Error
+
+# 从环境变量读取 MinIO 配置（推荐）
+# 如果环境变量不存在，则使用默认值
+MINIO_ENDPOINT = os.environ.get("MINIO_ENDPOINT", "minioapi.auton.one")  # 不包含协议
+MINIO_ROOT_USER = os.environ.get("MINIO_ROOT_USER", "arvin")  # 管理员 Access Key
+MINIO_ROOT_PASSWORD = os.environ.get("MINIO_ROOT_PASSWORD", "LinJia1996")  # 管理员 Secret Key
+BUCKET_NAME = os.environ.get("MINIO_BUCKET", "video-demo")
+MINIO_SECURE = os.environ.get("MINIO_SECURE", "True").lower() == "true"  # 是否使用 HTTPS
+
+# 初始化 MinIO 客户端（使用管理员账户）
+minio_client = Minio(
+    MINIO_ENDPOINT,
+    access_key=MINIO_ROOT_USER,
+    secret_key=MINIO_ROOT_PASSWORD,
+    secure=MINIO_SECURE  # 根据配置使用 HTTP 或 HTTPS
+)
+
+
+# 确保存储桶存在
+try:
+    if not minio_client.bucket_exists(BUCKET_NAME):
+        minio_client.make_bucket(BUCKET_NAME)
+        print(f"[MinIO] ✅ 创建存储桶: {BUCKET_NAME}")
+    else:
+        print(f"[MinIO] ℹ️ 存储桶已存在: {BUCKET_NAME}")
+except S3Error as e:
+    print(f"[MinIO] ⚠️ 检查存储桶失败: {e.code} - {e.message}")
+
+
+def upload_video_to_minio(video_file_path: str):
+    """
+    上传视频到 MinIO，返回 (video_path, public_url_or_error_message)
+    兼容 Gradio 输出格式：(video, link)
+    
+    返回预签名 URL，有效期 7 天，供 Gemini API 访问
+    """
+    if not video_file_path or not os.path.isfile(video_file_path):
+        return None, "❌ 视频文件无效或不存在"
+
+    try:
+        from datetime import timedelta
+        
+        original_name = os.path.basename(video_file_path)
+        file_ext = os.path.splitext(original_name)[1].lower()
+        allowed_video_exts = {".mp4", ".mov", ".avi", ".mkv", ".flv", ".wmv"}
+        
+        if file_ext not in allowed_video_exts:
+            return None, f"❌ 不支持的视频格式: {file_ext}，仅支持: {', '.join(allowed_video_exts)}"
+
+        print(f"[MinIO] 开始上传: {original_name}")
+        
+        # 根据文件扩展名确定正确的 MIME 类型
+        mime_types = {
+            ".mp4": "video/mp4",
+            ".mov": "video/quicktime",
+            ".avi": "video/x-msvideo",
+            ".mkv": "video/x-matroska",
+            ".flv": "video/x-flv",
+            ".wmv": "video/x-ms-wmv"
+        }
+        content_type = mime_types.get(file_ext, "video/mp4")  # 默认使用 video/mp4
+        
+        print(f"[MinIO] 设置 Content-Type: {content_type}")
+        
+        # 上传到 MinIO，使用 metadata 明确指定 Content-Type
+        # MinIO Python 客户端需要通过 metadata 设置 Content-Type
+        minio_client.fput_object(
+            bucket_name=BUCKET_NAME,
+            object_name=original_name,
+            file_path=video_file_path,
+            content_type=content_type,  # 直接指定 content_type
+            metadata={
+                "Content-Type": content_type  # 也在 metadata 中设置
+            }
+        )
+        
+        print(f"[MinIO] ✅ 上传成功: {original_name}")
+        print(f"[MinIO] ✅ Content-Type 已设置为: {content_type}")
+
+
+
+        # 生成直链（格式：协议://MINIO_ENDPOINT/MINIO_BUCKET/original_name）
+        # 注意：这要求 MinIO 对象必须是公开可访问的
+        protocol = "https" if MINIO_SECURE else "http"
+        public_url = f"{protocol}://{MINIO_ENDPOINT}/{BUCKET_NAME}/{original_name}"
+        
+        print(f"[MinIO] 🔗 生成直链: {public_url}")
+        
+        return video_file_path, public_url  # ✅ 返回直链
+
+    except S3Error as e:
+        error_msg = f"❌ MinIO 上传失败: {e.code} - {e.message}"
+        print(f"[MinIO] {error_msg}")
+        return None, error_msg
+    except Exception as e:
+        error_msg = f"❌ 上传出错: {type(e).__name__} - {str(e)}"
+        print(f"[MinIO] {error_msg}")
+        return None, error_msg