AI_design_agent/scr/upload_tos.py

"""
上传服务模块
处理图像上传到TOS云存储
"""


from datetime import datetime
import io
import os
from PIL import Image
import PIL
import requests
import tos
from logger_setup import logger

table_name="ai_video_info"

# 从环境变量获取 AK 和 SK 信息
ak ='AKLTNDNlMTFlNzk4OTNmNGU1YTlhMzQ0MmJjZjViMzc0YzQ'
sk = 'T0dReU1USm1PVEkzT0RSa05EY3hZamt6Wm1SalpqUmlOamhqTnpZMVlUVQ=='

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


def upload_image(image):
    try:
        if isinstance(image,str):
            if image.startswith("http"):
                temp_img = requests.get(image, timeout=30)
                temp_img.raise_for_status()
                temp_img = Image.open(io.BytesIO(temp_img.content)).convert("RGB")
                
            else:
                temp_img = Image.open(image).convert("RGB")
            image_url=process_cropped_upload(temp_img)
        elif isinstance(image, Image.Image):
            image_url=process_cropped_upload(image)
        logger.info(f'上传图片成功: {image_url}')
        return image_url
    except Exception as e:
        error_msg = f'上传失败，未知错误: {str(e)}'
        raise Exception(error_msg)


def process_cropped_upload(pil_image) :
                    # 保存到 temp（时间戳命名）
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
    filename = f"cropped_{timestamp}.png"
    temp_dir = os.path.join(os.path.dirname(__file__), "temp")
    os.makedirs(temp_dir, exist_ok=True)
    local_path = os.path.join(temp_dir, filename)
    pil_image.save(local_path, format="PNG")

    # 上传到 TOS，成功后删除临时文件并替换最终链接
    uploaded_url = upload_file_to_tos(local_path)
    if uploaded_url:
        # try:
        #     os.remove(local_path)
        #     logger.info(f"已上传并删除裁剪后的临时文件: {local_path}")
        # except Exception as del_err:
        #     logger.warning(f"删除临时文件失败: {del_err}")
        return uploaded_url
    else:
        logger.warning(f"TOS 上传失败，保留临时文件: {local_path}")
        return ''



def upload_file_to_tos(file_path: str, dir_path="design_image") -> str:
    """上传文件到TOS云存储"""
    try:
        # 从文件路径中提取文件名作为object_key
        object_key = dir_path+'/'+os.path.basename(file_path)
        logger.info(f'开始上传文件: {file_path}')
        logger.info(f'文件将保存为: {object_key}')
        
        # 创建客户端并上传文件
        client = tos.TosClientV2(ak, sk, endpoint, region)
        client.put_object_from_file(bucket_name, object_key, file_path)
        
        # 生成访问URL
        # object_url = f"https://{bucket_name}.{endpoint.split('//')[1]}/{object_key}"
        object_url = f"https://testdgxcx-oss.gloria.com.cn/{object_key}"
        logger.info(f'文件上传成功，访问URL: {object_url}')
        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__":
    upload_file_to_tos(file_path=r"D:\AI模特流程\历史.xlsx")