EAI-deepface/upload_file.py

import oss2
import os

oss_bucket_name = 'eai-files'
oss_endpoint = 'oss-cn-shanghai.aliyuncs.com'
access_key_id = 'LTAI5tAbkw8rkreiJbfcY1jZ'
access_key_secret = '2yvtVYE1qAZEgm0ChvdMqhp0LhB5Lj'


class OSSUploader:
    def __init__(self):

        self.bucket_name = 'eai-files'
        self.endpoint = 'oss-cn-shanghai.aliyuncs.com'
        self.access_key_id = 'LTAI5tAbkw8rkreiJbfcY1jZ'
        self.access_key_secret = '2yvtVYE1qAZEgm0ChvdMqhp0LhB5Lj'
        # 初始化 OSS 认证和 Bucket 对象
        self.auth = oss2.Auth(self.access_key_id, self.access_key_secret)
        self.bucket = oss2.Bucket(self.auth, self.endpoint, self.bucket_name)

    def upload_video_to_oss_local(self, local_video_path):
        """
        上传本地视频文件到 OSS
        :param local_video_path: 本地视频文件路径
        :return: OSS 文件 URL（如果上传成功），否则返回 None
        """
        try:
            # 提取本地视频文件名
            video_filename = os.path.basename(local_video_path)

            # 上传视频文件到 OSS
            result = self.bucket.put_object_from_file(video_filename, local_video_path)

            if result.status == 200:
                # 生成公共读的访问 URL
                oss_url = f"https://{self.bucket_name}.{self.endpoint}/{video_filename}"

                # 删除本地视频文件
                os.remove(local_video_path)
                print(f"已成功删除本地视频文件: {local_video_path}")
                return oss_url
            else:
                print(f"上传失败，状态码: {result.status}")
                return None
        except Exception as e:
            print(f"上传过程中出现错误: {e}")
            return None

    def upload_video_to_oss_stream(self, data_stream, object_name):
        """
        上传数据流到 OSS
        :param data_stream: 数据流（字节流或文件流）
        :param object_name: OSS 中的对象名称
        :return: OSS 文件 URL（如果上传成功），否则返回 None
        """
        try:
            # 上传数据流到 OSS
            result = self.bucket.put_object(object_name, data_stream)
            if result.status == 200:
                # 生成公共读的访问 URL
                oss_url = f"https://{self.bucket_name}.{self.endpoint}/{object_name}"
                return oss_url
            else:
                print(f"上传失败，状态码: {result.status}")
                return None
        except Exception as e:
            print(f"上传过程中出现错误: {e}")
            return None