chat-python/utils/upload_file_to_oss.py

from minio import Minio
from minio.error import S3Error
from minio.deleteobjects import DeleteObject
from config import minio_config
from utils.get_logger import setup_logger

logger = setup_logger(__name__)

class UploadMinio():
    def __init__(self):
        self.minio_endpoint = minio_config.get("minio_endpoint")
        self.minio_access_key = minio_config.get("minio_access_key")
        self.minio_secret_key = minio_config.get("minio_secret_key")
        self.minio_bucket = minio_config.get("minio_bucket")
        self.flag = minio_config.get("flag")
        self.minio_client = Minio(self.minio_endpoint,
                    access_key=self.minio_access_key,
                    secret_key=self.minio_secret_key,
                    cert_check=False,
                    # http_client=http_client,
                    secure=self.flag)  # 如果使用https，请将secure设置为True

    def upload_file(self, local_file_path, minio_bucket_file):
        try:
            # 使用put_object方法将文件上传到MinIO
            self.minio_client.fput_object(self.minio_bucket, minio_bucket_file, local_file_path)
        
            logger.info(f"File {local_file_path} uploaded to MinIO successfully.")
            return True
        
        except S3Error as e:
            logger.error(f"Error uploading file to MinIO: {e}")
            return False

        except Exception as e:
            logger.error(f"报错信息：{e}")
            return False
        
    def upload_io_bytes(self, minio_bucket_file, bytes_obj):
        try:
            # 使用put_object方法将文件上传到MinIO
            self.minio_client.put_object(self.minio_bucket,minio_bucket_file, bytes_obj, bytes_obj.tell())
        
            logger.info(f"File {minio_bucket_file} uploaded to MinIO successfully.")
            return True
        
        except S3Error as e:
            logger.error(f"Error uploading file to MinIO: {e}")
            return False

        except Exception as e:
            logger.error(f"报错信息：{e}")
            return False
        
    def delete_doc_id_images(self, knowledge_id, doc_id):
        try:
            objects = self.minio_client.list_objects(self.minio_bucket, prefix=f"pdf/{knowledge_id}/{doc_id}", recursive=True)

            objects_to_delete = []
            for obj in objects:
                objects_to_delete.append(DeleteObject(obj.object_name))
                # self.minio_client.remove_object(self.minio_bucket, obj.object_name)
            error_responses = self.minio_client.remove_objects(self.minio_bucket, objects_to_delete)
            for error in error_responses:
                if error:
                    delete_flag = False 
                    logger.info("Deletion Error: {}".format(error))
                    break
            else:
                logger.info("Sussess")
                delete_flag = True
            return delete_flag
        except Exception as e:
            logger.error(f"删除minio图片出错：{e}")
            return False