model_script/client/OSSClient.py

import logging

import oss2
from oss2.credentials import StaticCredentialsProvider

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


class OSSClient(object):
    def __init__(self, access_key, access_secret, endpoint, region):
        print(f"oss.version: {oss2.__version__}")
        self.access_key = access_key
        self.access_secret = access_secret
        self.endpoint = endpoint
        self.region = region
        self.auth = oss2.ProviderAuthV4(StaticCredentialsProvider(self.access_key, self.access_secret))

    def download_file(self, bucket_name, oss_file_path, local_file_path):
        bucket = self._get_bucket(bucket_name)
        bucket.get_object_to_file(oss_file_path, local_file_path)

    def file_is_exist(self, bucket_name, oss_file_path) -> bool:
        """
        检查阿里云 OSS 文件是否存在。
        :param bucket_name: OSS Bucket 名称
        :param oss_file_path: 文件路径，例如 'folder/file.txt'
        :return: 布尔值，表示文件是否存在
        """
        try:
            bucket = self._get_bucket(bucket_name)
            exists = bucket.object_exists(oss_file_path)
            return exists
        except Exception as e:
            return False

    def _get_bucket(self, bucket_name: str):
        return oss2.Bucket(auth=self.auth, endpoint=self.endpoint, bucket_name=bucket_name, region=self.region)