movie-create/examples/video_create/utils/tools.py

import os
import re 
import toon
import json
import requests
from taskflow import get_logger
import tos
from dotenv import load_dotenv

logger = get_logger("examples.video_create.utils.tools")

def string_to_json(markdown_string):
    try:
        json_content = re.sub(r'^```json|\n```$', '', markdown_string, flags=re.MULTILINE).strip()
        if not json_content:
            json_content = markdown_string
            # raise ValueError("字符串中未找到有效的JSON内容")
            
        # 解析JSON内容
        json_data = json.loads(json_content)
        
        return json_data
        
    except Exception as e:
        logger.error(f"生成结果解析失败：\n{markdown_string}")
        raise e

def download_video(video_url, output_path):
    """
    根据视频URL下载视频到本地指定路径
    
    参数:
        video_url (str): 视频的URL地址
        output_path (str): 本地保存路径（包含文件名和扩展名）
        
    返回:
        bool: 下载成功返回True，失败返回False
    """
    try:
        # 创建目录（如果不存在）
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        
        # 发送HTTP GET请求
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
        }
        response = requests.get(video_url, headers=headers, stream=True)
        response.raise_for_status()  # 检查请求是否成功
        
        # 获取文件总大小（用于进度显示）
        total_size = int(response.headers.get('content-length', 0))
        
        # 以二进制写入模式保存视频
        with open(output_path, 'wb') as f:
            downloaded_size = 0
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:  # 过滤掉保持连接的新块
                    f.write(chunk)
                    downloaded_size += len(chunk)            
                        
        logger.info(f"\n视频已成功保存到: {output_path}")
        return True
        
    except requests.exceptions.RequestException as e:
        logger.info(f"下载视频时出错: {e}")
    except IOError as e:
        logger.info(f"保存视频时出错: {e}")
    except Exception as e:
        logger.info(f"发生未知错误: {e}")
        
    return False

def download_image(image_url, output_path):
    """
    根据图片URL下载图片到本地指定路径
    
    参数:
        image_url (str): 图片的URL地址
        output_path (str): 本地保存路径（包含文件名和扩展名）
    """
    try:
        # 创建目录（如果不存在）
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        
        # 发送HTTP GET请求
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
        }
        response = requests.get(image_url, headers=headers, stream=True)
        response.raise_for_status()  # 检查请求是否成功
        
        # 以二进制写入模式保存图片
        with open(output_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:  # 过滤掉保持连接的新块
                    f.write(chunk)
                    
        logger.info(f"\n图片已成功保存到: {output_path}")
        return True
        
    except requests.exceptions.RequestException as e:
        logger.info(f"下载图片时出错: {e}")
    except IOError as e:
        logger.info(f"保存图片时出错: {e}")
    except Exception as e:
        logger.info(f"发生未知错误: {e}")
        
    return False

def efficient_sort(a, b):
    """
    高效排序算法，通过贪心策略
    """
    n = len(a)
    selected = [False] * n
    result = []
    available = set()
    
    # 统计每个元素的依赖
    dependencies = []
    for i in range(n):
        # 如果a[i]在b[i]中，它可以立即被处理
        immediate = a[i] in b[i]
        dependencies.append((immediate, i))
    
    # 先处理可以立即处理的元素
    for immediate, i in dependencies:
        if immediate and not selected[i]:
            result.append(i)
            selected[i] = True
            available.update(b[i])
    
    # 然后处理其他元素
    while len(result) < n:
        progress = False
        
        for i in range(n):
            if not selected[i] and a[i] in available:
                result.append(i)
                selected[i] = True
                available.update(b[i])
                progress = True
        
        # 如果没有进展，选择第一个未处理的元素
        if not progress:
            for i in range(n):
                if not selected[i]:
                    result.append(i)
                    selected[i] = True
                    available.update(b[i])
                    break
    
    return result

# 加载环境变量
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__":
    # data = {
    #     "name": "John",
    #     "age": 30,
    #     "city": "New York"
    # }
    # toon_str = toon.encode(data)
    # print(type(toon_str), toon_str)
    # decoded_data = toon.decode(toon_str)
    # print(type(decoded_data), decoded_data)

    # python -m utils.upload
    test_file = "./image.jpg"
    try:
        url = upload_file_to_tos(test_file)
        print(f"文件上传成功，访问URL: {url}")
    except Exception as e:
        print(f"上传失败: {str(e)}")