office-pc-plug/tools/oss_client.py

import datetime
import hashlib
import json
import os

from minio import Minio, S3Error

from tools.config import get_config
from tools.logger_handle import logger


class MinioClient:
    client = None

    def __new__(cls, *args, **kwargs):
        if not cls.client:
            cls.client = object.__new__(cls)
        return cls.client

    def __init__(self, service, access_key, secret_key, secure=False):
        self.service = service
        self.access_key = access_key
        self.secret_key = secret_key
        self.client = Minio(service, access_key=access_key, secret_key=secret_key, secure=secure)

    @staticmethod
    def load_metadata(local_path):
        with open(f'{local_path}.metadata', 'r', encoding='utf-8') as f:
            file_metadata = json.loads(f.read())
            return file_metadata

    def get_bucket_list(self):
        buckets = self.client.list_buckets()
        bucket_list = []
        for bucket in buckets:
            bucket_list.append(
                {"bucket_name": bucket.name, "create_time": bucket.creation_date}
            )
        return bucket_list

    def upload_file(self, file_path):
        """
        上传文件 + 写入
        :param file_path: 本地文件路径
        :return:
        """
        try:
            with open(f'{file_path}.metadata', 'r', encoding='utf-8') as f:
                file_metadata = json.loads(f.read())
            oss_path = file_metadata['oss_path']

            self.client.fput_object(file_metadata['bucket_name'], oss_path, file_path)
            return True
        except S3Error as e:
            logger.exception(e)
            return False

    def fget_file(self, bucket_name, oss_path, file_path):
        """
        下载保存文件保存本地
        :param bucket_name:
        :param oss_path:服务端文件名
        :param file_path:要写入的本地路径
        :return:
        """
        self.client.fget_object(bucket_name, oss_path, file_path)
        with open(f'{file_path}.metadata', 'w', encoding='utf-8') as f:
            f.write(json.dumps({
                'serve': self.service,
                'access_key': self.access_key,
                'secret_key': self.secret_key,
                'bucket_name': bucket_name,
                'md5': self.get_cloud_file_md5(bucket_name, oss_path),
                'oss_path': oss_path,
                'last_modified': self.get_cloud_last_modified(bucket_name, oss_path)
            }))

    def get_cloud_file_md5(self, bucket_name, file):
        return self.client.stat_object(bucket_name, file).etag

    def get_cloud_last_modified(self, bucket_name, file):
        return self.client.stat_object(bucket_name, file).last_modified.strftime("%Y-%m-%d %H:%M:%S")

    @staticmethod
    def get_file_md5(file_path):
        md5 = hashlib.md5()
        with open(file_path, 'rb') as f:
            # 分块读取以支持大文件
            while chunk := f.read(8192):
                md5.update(chunk)
        return md5.hexdigest()


oss_config = get_config('oss')
oss_handle = MinioClient(oss_config['endpoint'],
                         access_key=oss_config['access_key'],
                         secret_key=oss_config['secret_key'])

if __name__ == '__main__':
    client = MinioClient('172.10.3.36:9000', access_key='miniominio', secret_key='miniominio')
    # print(client.get_cloud_file_md5('document-mgr', '接口文档.txt'))
    # print(client.get_file_md5('接口文档.txt'))
    print(client.get_cloud_file_md5('document-mgr', 'test.docx'))

    print(client.upload_file(r'E:\PycharmProjects\office_plugin\tools\storage\admin\document-mgr_test.docx'))
    oss_path = 'test.docx'
    print(client.get_cloud_file_md5('document-mgr', 'test.docx'))
    # client.fget_file('document-mgr',
    #                  'document-mgr/ten_1/2_2/document-mgr/document-mgr/2025/05/27/2025-05-271112065750681423872-新建文件.docx',
    #                  '接口文档.docx')