bowong
/
mxivideo
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: template

This commit is contained in:
root 2025-07-12 23:48:24 +08:00
parent d301fca72c
commit bfe9f1e425
2 changed files with 30 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
python_core/cli/commands/template.py
View File

@ -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

283
test_draft_content_parsing.py
View File

@ -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()