diff --git a/python_core/cli/commands/template.py b/python_core/cli/commands/template.py index 8c1f068..5e9871c 100644 --- a/python_core/cli/commands/template.py +++ b/python_core/cli/commands/template.py @@ -177,15 +177,34 @@ def get_template_detail( tracks_data = draft_content.get('tracks', []) materials_data = draft_content.get('materials', []) + # 确保 tracks_data 和 materials_data 是列表 + if not isinstance(tracks_data, list): + logger.warning(f"tracks_data is not a list: {type(tracks_data)}") + tracks_data = [] + + if not isinstance(materials_data, list): + logger.warning(f"materials_data is not a list: {type(materials_data)}") + materials_data = [] + # 创建素材查找表 materials_lookup = {} for material in materials_data: + # 确保 material 是字典 + if not isinstance(material, dict): + logger.warning(f"material is not a dict: {type(material)}") + continue + material_id = material.get('id', '') if material_id: materials_lookup[material_id] = material # 处理轨道 for idx, track_data in enumerate(tracks_data): + # 确保 track_data 是字典 + if not isinstance(track_data, dict): + logger.warning(f"track_data is not a dict: {type(track_data)}") + continue + track = { 'id': track_data.get('id', f'track_{idx}'), 'name': track_data.get('name', f'轨道 {idx + 1}'), @@ -197,7 +216,18 @@ def get_template_detail( # 处理片段 segments_data = track_data.get('segments', []) + + # 确保 segments_data 是列表 + if not isinstance(segments_data, list): + logger.warning(f"segments_data is not a list: {type(segments_data)}") + segments_data = [] + for segment_data in segments_data: + # 确保 segment_data 是字典 + if not isinstance(segment_data, dict): + logger.warning(f"segment_data is not a dict: {type(segment_data)}") + continue + # 获取时间信息 start_time = segment_data.get('start', 0) / 1000.0 # 转换为秒 end_time = segment_data.get('end', 0) / 1000.0 diff --git a/test_draft_content_parsing.py b/test_draft_content_parsing.py deleted file mode 100644 index d8c9fae..0000000 --- a/test_draft_content_parsing.py +++ /dev/null @@ -1,283 +0,0 @@ -#!/usr/bin/env python3 -""" -测试 draft_content 解析修复 -验证字符串类型的 draft_content 能正确解析为字典 -""" - -import sys -import os -import json -import uuid -from datetime import datetime -sys.path.append(os.path.dirname(os.path.abspath(__file__))) - -from python_core.database.template_postgres import template_table -from python_core.database.types import TemplateInfo - -def test_draft_content_parsing(): - """测试 draft_content 解析功能""" - print("=== 测试 draft_content 解析修复 ===\n") - - try: - # 创建测试数据 - test_draft = { - "canvas_config": {"width": 1920, "height": 1080, "fps": 30}, - "tracks": [ - {"id": "track1", "type": "video", "name": "视频轨道"} - ], - "materials": [ - {"id": "mat1", "name": "test.mp4", "type": "video"} - ], - "duration": 10000 - } - - template_id = str(uuid.uuid4()) - - # 1. 测试正常的字典类型 draft_content - print("1. 测试字典类型 draft_content...") - template_info = TemplateInfo( - id=template_id, - name=f"测试模板-字典-{template_id[:8]}", - description="测试字典类型的 draft_content", - thumbnail_path="", - draft_content_path="", - resources_path="", - created_at=datetime.now().isoformat(), - updated_at=datetime.now().isoformat(), - canvas_config=test_draft.get('canvas_config', {}), - duration=test_draft.get('duration', 0), - material_count=len(test_draft.get('materials', [])), - track_count=len(test_draft.get('tracks', [])), - tags=["测试"], - user_id="test_user", - draft_content=test_draft # 字典类型 - ) - - # 存储到数据库 - created_id = template_table.create_template(template_info) - print(f" ✅ 模板创建成功: {created_id}") - - # 读取并验证 - stored_template = template_table.get_template_by_id(created_id) - if stored_template and isinstance(stored_template.draft_content, dict): - print(f" ✅ draft_content 类型正确: {type(stored_template.draft_content)}") - print(f" ✅ draft_content 键数量: {len(stored_template.draft_content)}") - else: - print(f" ❌ draft_content 类型错误: {type(stored_template.draft_content) if stored_template else 'None'}") - return False - - # 清理 - template_table.delete_template(created_id) - - # 2. 测试直接在数据库中插入 JSON 字符串 - print("\n2. 测试 JSON 字符串类型 draft_content...") - - # 直接操作数据库，插入 JSON 字符串 - import psycopg2.extras - from python_core.database.template_postgres import TemplateTablePostgres - - template_id2 = str(uuid.uuid4()) - draft_json_str = json.dumps(test_draft) - - with template_table._get_connection() as conn: - with conn.cursor() as cursor: - insert_sql = """ - INSERT INTO templates ( - id, name, description, thumbnail_path, draft_content_path, - draft_content, resources_path, canvas_config, duration, material_count, - track_count, tags, is_cloud, user_id, created_at, updated_at - ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) - """ - - now = datetime.now() - cursor.execute(insert_sql, ( - template_id2, - f"测试模板-字符串-{template_id2[:8]}", - "测试 JSON 字符串类型的 draft_content", - "", - "", - draft_json_str, # 直接插入 JSON 字符串 - "", - json.dumps({}), - 10000, - 1, - 1, - json.dumps(["测试"]), - False, - "test_user", - now, - now - )) - conn.commit() - - print(f" ✅ 直接插入 JSON 字符串成功: {template_id2}") - - # 读取并验证解析 - stored_template2 = template_table.get_template_by_id(template_id2) - if stored_template2: - print(f" draft_content 类型: {type(stored_template2.draft_content)}") - if isinstance(stored_template2.draft_content, dict): - print(f" ✅ JSON 字符串正确解析为字典") - print(f" ✅ draft_content 键数量: {len(stored_template2.draft_content)}") - print(f" ✅ 包含 tracks: {'tracks' in stored_template2.draft_content}") - print(f" ✅ 包含 materials: {'materials' in stored_template2.draft_content}") - else: - print(f" ❌ JSON 字符串解析失败") - return False - else: - print(f" ❌ 模板读取失败") - return False - - # 测试 get_draft_content 方法 - print("\n3. 测试 get_draft_content 方法...") - draft_content = template_table.get_draft_content(template_id2) - if draft_content is not None and isinstance(draft_content, dict): - print(f" ✅ get_draft_content 返回字典类型") - print(f" ✅ 键数量: {len(draft_content)}") - else: - print(f" ❌ get_draft_content 返回类型错误: {type(draft_content)}") - return False - - # 清理 - template_table.delete_template(template_id2) - - print("\n✅ 所有测试通过！") - return True - - except Exception as e: - print(f"❌ 测试失败: {e}") - import traceback - traceback.print_exc() - return False - -def test_cli_command(): - """测试 CLI 命令""" - print("\n=== 测试 CLI 命令 ===\n") - - try: - # 创建一个测试模板 - test_draft = { - "canvas_config": {"width": 1920, "height": 1080, "fps": 30}, - "tracks": [ - { - "id": "track1", - "type": "video", - "name": "视频轨道", - "segments": [ - { - "id": "seg1", - "start": 0, - "end": 5000, - "material_id": "mat1" - } - ] - } - ], - "materials": [ - { - "id": "mat1", - "name": "test.mp4", - "type": "video", - "path": "test.mp4" - } - ], - "duration": 5000 - } - - template_id = str(uuid.uuid4()) - template_info = TemplateInfo( - id=template_id, - name=f"CLI测试模板-{template_id[:8]}", - description="用于测试 CLI 命令的模板", - thumbnail_path="", - draft_content_path="", - resources_path="", - created_at=datetime.now().isoformat(), - updated_at=datetime.now().isoformat(), - canvas_config=test_draft.get('canvas_config', {}), - duration=test_draft.get('duration', 0), - material_count=len(test_draft.get('materials', [])), - track_count=len(test_draft.get('tracks', [])), - tags=["CLI测试"], - user_id="test_user", - draft_content=test_draft - ) - - # 创建模板 - created_id = template_table.create_template(template_info) - print(f"✅ 创建测试模板: {created_id}") - - # 测试 CLI 命令 - import subprocess - - cmd = [ - "python3", "-m", "python_core.cli", - "template", "detail", created_id, - "--json" - ] - - print(f"执行命令: {' '.join(cmd)}") - - result = subprocess.run( - cmd, - capture_output=True, - text=True, - cwd="/root/projects/mixvideo_v2" - ) - - print(f"返回码: {result.returncode}") - if result.stdout: - print(f"标准输出: {result.stdout}") - if result.stderr: - print(f"标准错误: {result.stderr}") - - if result.returncode == 0: - # 检查输出是否包含 JSONRPC 成功响应 - if "JSONRPC:" in result.stdout and '"error"' not in result.stdout: - print("✅ CLI 命令执行成功，没有错误") - success = True - else: - print("❌ CLI 命令返回错误") - success = False - else: - print("❌ CLI 命令执行失败") - success = False - - # 清理 - template_table.delete_template(created_id) - print(f"✅ 清理测试模板: {created_id}") - - return success - - except Exception as e: - print(f"❌ CLI 测试失败: {e}") - import traceback - traceback.print_exc() - return False - -def main(): - """主测试函数""" - print("开始测试 draft_content 解析修复...\n") - - # 测试1：draft_content 解析 - test1_success = test_draft_content_parsing() - - # 测试2：CLI 命令 - test2_success = test_cli_command() - - print(f"\n=== 测试结果 ===") - print(f"测试1 (draft_content解析): {'✅ 通过' if test1_success else '❌ 失败'}") - print(f"测试2 (CLI命令): {'✅ 通过' if test2_success else '❌ 失败'}") - - if test1_success and test2_success: - print("\n🎉 所有测试通过！draft_content 解析修复成功！") - print("\n修复内容:") - print("1. ✅ 修复了数据库中 JSON 字符串的解析问题") - print("2. ✅ 改进了 CLI 命令的类型检查") - print("3. ✅ 确保 draft_content 始终为字典类型") - print("4. ✅ 解决了 'str' object has no attribute 'get' 错误") - else: - print("\n❌ 部分测试失败，需要进一步检查") - -if __name__ == "__main__": - main()