video-create/utils/upload.py

import os
import tos

from dotenv import load_dotenv
from utils.logger_config import setup_logger

logger = setup_logger(__name__)

# 加载环境变量
load_dotenv()


# 从环境变量获取 AK 和 SK 信息
ak = os.getenv('TOS_ACCESS_KEY')
sk = os.getenv('TOS_SECRET_KEY')


# 存储桶配置信息
endpoint = "https://tos-cn-guangzhou.volces.com"
region = "cn-guangzhou" 
bucket_name = "guide-material"


def upload_file_to_tos(file_name: str) -> str:
    """
    上传文件到TOS存储桶并返回访问URL
    
    Args:
        file_name (str): 本地文件的完整路径
        
    Returns:
        str: 上传文件的访问URL
        
    Raises:
        Exception: 上传过程中的任何错误
    """
    try:
        # 检查文件是否存在
        if not os.path.exists(file_name):
            raise FileNotFoundError(f"文件不存在: {file_name}")
        
        # 从文件路径中提取文件名作为object_key，确保使用正斜杠
        filename = os.path.basename(file_name)
        object_key = f"video-create/{filename}".replace("\\", "/")
        logger.info(f'开始上传文件: {file_name}')
        logger.info(f'文件将保存为: {object_key}')
        
        # 创建客户端并上传文件
        client = tos.TosClientV2(ak, sk, endpoint, region)
        client.put_object_from_file(bucket_name, object_key, file_name)
        
        # 验证上传是否成功 - 尝试获取对象元数据
        try:
            head_response = client.head_object(bucket_name, object_key)
            if head_response is None:
                raise Exception("无法验证文件是否上传成功：head_object返回None")
            logger.info(f'文件上传验证成功，ETag: {getattr(head_response, "etag", "N/A")}')
        except Exception as verify_error:
            logger.warning(f'验证上传状态时出现警告: {str(verify_error)}')
            # 不抛出异常，因为上传可能已经成功，只是验证失败
        
        # 生成访问URL，确保object_key使用正斜杠
        object_key_normalized = object_key.replace("\\", "/")
        object_url = f"https://testdgxcx-oss.gloria.com.cn/{object_key_normalized}"
        logger.info(f'文件上传成功，访问URL: {object_url}')
        logger.info(f'Object Key (用于调试): {object_key_normalized}')
        return object_url
        
    except tos.exceptions.TosClientError as e:
        error_msg = f'上传失败，客户端错误: message={e.message}, cause={e.cause}'
        logger.error(error_msg)
        raise Exception(error_msg)
    except tos.exceptions.TosServerError as e:
        error_msg = f'上传失败，服务端错误: code={e.code}, request_id={e.request_id}, message={e.message}, status_code={e.status_code}, ec={e.ec}, request_url={e.request_url}'
        logger.error(error_msg)
        raise Exception(error_msg)
    except Exception as e:
        error_msg = f'上传失败，未知错误: {str(e)}'
        logger.error(error_msg)
        raise Exception(error_msg)

# 使用示例
if __name__ == "__main__":

    # python -m utils.upload
    test_file = "./output/new_frame_scene0_camera0_shot1.png"
    # test_file = "/data/data/luosy/project/iclip/output/rawvideo/rawvideo_a/oral_show/final-final-rawvideo-0-0.mp4"
    # 
    # test_file = "010.jpg"
    try:
        url = upload_file_to_tos(test_file)
        print(f"文件上传成功，访问URL: {url}")
    except Exception as e:
        print(f"上传失败: {str(e)}")