VSeesResourceWeb/background/Object.py

# -*- encoding: utf-8 -*-
"""
@File    : AmazonS3Util.py
@Time    : 2022/8/11 14:00
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import traceback

import boto3
import botocore
from boto3.session import Session
from botocore import client
from botocore.exceptions import ClientError


class AmazonS3Util:
    def __init__(self):
        self.access_id = 'AKIA2E67UIMD45Y3HL53'
        self.access_secret = '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'
        self.region_name = 'us-east-1'
        session = Session(
            aws_access_key_id=self.access_id,
            aws_secret_access_key=self.access_secret,
            region_name=self.region_name
        )
        self.client_conn = boto3.client(
            's3',
            aws_access_key_id=self.access_id,
            aws_secret_access_key=self.access_secret,
            config=botocore.client.Config(signature_version='s3v4'),
            region_name=self.region_name
        )
        self.session_conn = session.resource('s3')

    def upload_file_obj(self, bucket, file_key, file_obj, extra_args=None):
        """
        对象上传至S3存储桶
        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Object.upload_file
        @param bucket: 存储桶名称-必须
        @param file_key: 需要上传文件路径+文件名称
        @param file_obj: 文件对象
        @param extra_args: 额外参数 如ACL配置
        @return: 当上传成功时为True;否则,False
        """
        try:
            session = self.session_conn
            bucket = session.Bucket(bucket)
            obj = bucket.Object(file_key)
            obj.upload_fileobj(file_obj, ExtraArgs=extra_args)
            return True
        except Exception as e:
            print(e.args)
            ex = traceback.format_exc()
            print('具体错误{}'.format(ex))
            return False

    def generate_file_obj_url(self, bucket, file_key):
        """
        生成对象URL
        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.generate_presigned_url
        @param bucket: 存储桶名称
        @param file_key: 文件名称
        @return: url
        """
        response_url = self.client_conn.generate_presigned_url(
            ClientMethod='get_object',
            Params={
                'Bucket': bucket,
                'Key': file_key
            }
        )
        return response_url

    def delete_obj(self, bucket, file_key):
        """
        删除对象
        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Object.delete
        @param bucket: 存储桶
        @param file_key: 文件名称
        @return: 当删除成功时为True;否则,False
        """
        try:
            bucket = self.session_conn.Bucket(bucket)
            obj = bucket.Object(file_key)
            obj.delete()
            return True
        except Exception as e:
            print(e.args)
            ex = traceback.format_exc()
            print('具体错误{}'.format(ex))
            return False

    def bucket_exists(self, bucket_name):
        """
        判断桶是否存在，是否有访问权限
        @return: 当bucket存在时为True;否则,假的
        """
        try:
            self.client_conn.head_bucket(Bucket=bucket_name)
            exists = True
        except ClientError:
            exists = False
        return exists

    def get_object(self, bucket, key):
        """
        获取对象
        @param bucket: 存储桶
        @param key: 文件
        @return : boolean
        """
        try:
            self.client_conn.get_object(Bucket=bucket, Key=key)
            return True
        except self.client_conn.exceptions.NoSuchKey:
            return False

    def download_object(self, bucket, key, file_name):
        """
        下载对象至本地
        @param file_name: 保存位置以及名称
        @param bucket: 存储桶
        @param key: 文件
        @return : boolean
        """
        try:
            self.client_conn.download_file(bucket, key, file_name)
            return True
        except Exception as e:
            return e.args

    def copy_obj(self, source_bucket, to_bucket, file_key):
        """
        复制对象
        @param source_bucket: 原存储桶
        @param file_key: 文件名称
        @param to_bucket: 新存储桶
        """
        source_dict = {
            'Bucket': source_bucket,
            'Key': file_key
        }
        self.session_conn.meta.client.copy(source_dict, to_bucket, file_key)

    def copy_single_obj(self, source_bucket, source_object, target_bucket, target_object, StorageClass=None):
        """
        单个对象复制
        @param source_bucket:源存储桶
        @param source_object:源对象
        @param target_bucket:目标存储桶
        @param target_object:目标对象
        @param StorageClass:存储类
        @return: None
        """
        s3 = self.session_conn
        copy_source = {
            'Bucket': source_bucket,
            'Key': source_object
        }
        target_object = s3.Object(target_bucket, target_object)
        if StorageClass:
            target_object.copy_from(CopySource=copy_source, StorageClass=StorageClass)
        else:
            target_object.copy_from(CopySource=copy_source)

    def generate_put_obj_url(self, bucket_name, obj_key, storage_class=None):
        """
        生成预签名对象URL
        @param bucket_name: 存储桶名称
        @param obj_key: 对象key
        @param storage_class: 存储类 例
        @return: 对象URL
        """
        params = {
            'Bucket': bucket_name,
            'Key': obj_key,
        }
        if storage_class:
            params['StorageClass'] = storage_class
        return self.session_conn.meta.client.generate_presigned_url('put_object',
                                                                    Params=params,
                                                                    ExpiresIn=7200)

    def batch_copy_obj(self, source_bucket, target_bucket, prefix, target_prefix, storage_class=None):
        """
        批量拷贝对象
        @param source_bucket: 源存储桶
        @param target_bucket: 目标存储桶
        @param prefix: 需要搜索的对象前缀 例：AUS000247LTCLY/vod1/1686043996
        @param target_prefix: 目标对象前缀 例：app/algorithm-shop/1686043996
        @param storage_class: 存储类
        @return: None
        """
        s3 = self.session_conn
        # 遍历源存储桶中指定前缀下的所有对象，依次进行复制操作
        for obj in s3.Bucket(source_bucket).objects.filter(Prefix=prefix):
            key = obj.key  # 对象键名
            target_key = f'{target_prefix}/' + key.split('/')[-1]  # 新的对象键名，此处为 "new_path/" + 原有文件名
            copy_source = {
                'Bucket': source_bucket,
                'Key': key
            }
            # 将对象复制到目标存储桶，并设置存储类型和新的对象键名
            if storage_class:
                s3.Object(target_bucket, target_key).copy_from(CopySource=copy_source, StorageClass=storage_class)
            else:
                s3.Object(target_bucket, target_key).copy_from(CopySource=copy_source)

    def get_object_size(self, bucket_name, object_key):
        """
        获取存储桶中指定对象的大小

        :param bucket_name: string，存储桶名称
        :param object_key: string，对象键名
        :return: int，指定对象的大小，单位为字节
        """
        s3 = self.session_conn
        obj = s3.Object(bucket_name, object_key)
        try:
            return obj.content_length
        except Exception as e:
            return 0

    def get_object_list(self, bucket_name, prefix):
        """
        获取指定路径所有对象

        :param bucket_name: string，存储桶名称
        :param prefix: string，路径
        :return: int，指定对象的大小，单位为字节
        """
        try:
            s3 = self.client_conn
            obj = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
            return obj['Contents']
        except Exception as e:
            return []