diff --git a/python_core/cli/commands/template.py b/python_core/cli/commands/template.py index aa44a21..8c1f068 100644 --- a/python_core/cli/commands/template.py +++ b/python_core/cli/commands/template.py @@ -152,10 +152,15 @@ def get_template_detail( # 获取 draft_content draft_content = template_table.get_draft_content(template_id) - if not draft_content: + if draft_content is None: response.error(-32603, f"模板详细信息不存在: {template_id}") return + # 确保 draft_content 是字典类型 + if not isinstance(draft_content, dict): + logger.warning(f"draft_content is not a dict for template {template_id}: {type(draft_content)}") + draft_content = {} + # 构建详细信息 detail = { 'id': template.id, diff --git a/test_draft_content_parsing.py b/test_draft_content_parsing.py new file mode 100644 index 0000000..d8c9fae --- /dev/null +++ b/test_draft_content_parsing.py @@ -0,0 +1,283 @@ +#!/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()