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)}")