comi-logic/mxivideo
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: bug

Browse Source
This commit is contained in:
root
2025-07-12 20:46:30 +08:00
parent fe9597eb36
commit 7ee8bcab54
6 changed files with 733 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,468 +1,309 @@
"""
模板管理CLI命令
模板管理CLI命令 - 简化版本，只传输JSON RPC
"""
from pathlib import Path
from typing import Optional, List
from dataclasses import asdict
import typer
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
from rich.live import Live
import json
from python_core.utils.jsonrpc_enhanced import create_response_handler, create_progress_reporter
from python_core.services.template_manager_cloud import TemplateManagerCloud, TemplateInfo
from python_core.utils.logger import logger
from uuid import uuid4
console = Console()
template_app = typer.Typer(name="template", help="模板管理命令")
@template_app.command("import")
@template_app.command("batch-import")
def batch_import(
source_folder: str = typer.Argument(..., help="包含模板子文件夹的源文件夹"),
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
show_progress: bool = typer.Option(True, "--progress", "-p", help="显示进度条")
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""批量导入模板"""
response = create_progress_reporter(str(uuid4()))
try:
console.print(f"📦 [bold blue]批量导入模板[/bold blue]")
console.print(f"📁 源文件夹: {source_folder}")
# 验证源文件夹
source_path = Path(source_folder)
if not source_path.exists():
console.print(f"[red]❌ 源文件夹不存在: {source_folder}[/red]")
raise typer.Exit(1)
response.error(-32603, f"源文件夹不存在: {source_folder}")
return
if not source_path.is_dir():
console.print(f"[red]❌ 路径不是文件夹: {source_folder}[/red]")
raise typer.Exit(1)
response.error(-32603, f"路径不是文件夹: {source_folder}")
return
# 创建模板管理器
manager = TemplateManagerCloud()
manager = TemplateManagerCloud(user_id=user_id or "default")
# 进度回调函数
progress_messages = []
def progress_callback(message: str):
progress_messages.append(message)
if verbose:
console.print(f" {message}")
def progress_callback(current: int, total: int, message: str):
response.progress(current, total, message)
# 执行批量导入
if show_progress and not verbose:
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TaskProgressColumn(),
console=console
) as progress:
task = progress.add_task("正在导入模板...", total=None)
def progress_with_bar(message: str):
progress.update(task, description=message)
progress_callback(message)
result = manager.batch_import_templates(source_folder, progress_with_bar)
else:
result = manager.batch_import_templates(source_folder, progress_callback)
result = manager.batch_import_templates(source_folder, progress_callback)
# 显示结果
# 返回结果
if result['status']:
console.print(f"\n✅ [bold green]导入完成![/bold green]")
console.print(f"📊 成功导入: {result['imported_count']} 个模板")
console.print(f"❌ 导入失败: {result['failed_count']} 个模板")
# 显示成功导入的模板
if result['imported_templates'] and verbose:
console.print(f"\n📋 [bold green]成功导入的模板:[/bold green]")
for template in result['imported_templates']:
if isinstance(template, dict):
console.print(f" ✅ {template.get('name', 'Unknown')} (ID: {template.get('id', 'Unknown')})")
else:
console.print(f" ✅ {template.name} (ID: {template.id})")
# 显示失败的模板
if result['failed_templates']:
console.print(f"\n❌ [bold red]导入失败的模板:[/bold red]")
for failed in result['failed_templates']:
console.print(f" ❌ {failed['name']}: {failed['error']}")
response.success({
"imported_count": result['imported_count'],
"failed_count": result['failed_count'],
"imported_templates": result['imported_templates'],
"failed_templates": result['failed_templates'],
"message": result['msg']
})
else:
console.print(f"[red]❌ 导入失败: {result['msg']}[/red]")
raise typer.Exit(1)
response.error(-32603, result['msg'])
except Exception as e:
console.print(f"[red]❌ 批量导入失败: {str(e)}[/red]")
raise typer.Exit(1)
response.error(-32603, f"批量导入失败: {str(e)}")
@template_app.command("list")
def list_templates(
limit: int = typer.Option(20, "--limit", "-l", help="显示数量限制"),
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模板"),
limit: int = typer.Option(100, "--limit", "-l", help="显示数量限制"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
format_output: str = typer.Option("table", "--format", "-f", help="输出格式: table, json")
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""列出所有模板"""
response = create_response_handler(str(uuid4()))
try:
console.print(f"📋 [bold blue]模板列表[/bold blue]")
# 创建模板管理器
manager = TemplateManagerCloud()
manager = TemplateManagerCloud(user_id=user_id or "default")
# 获取模板列表
templates = manager.get_templates()
if not templates:
console.print("📭 没有模板")
return
templates = manager.get_templates(include_cloud=include_cloud)
# 限制显示数量
total_count = len(templates)
if len(templates) > limit:
templates = templates[:limit]
console.print(f"📊 显示前 {limit} 个模板（共 {total_count} 个）")
else:
console.print(f"📊 共 {total_count} 个模板")
if format_output == "json":
# JSON格式输出
templates_data = []
for template in templates:
if isinstance(template, TemplateInfo):
from dataclasses import asdict
templates_data.append(asdict(template))
else:
templates_data.append(template)
console.print(json.dumps(templates_data, ensure_ascii=False, indent=2))
return
# 表格格式输出
table = Table(title="模板列表")
table.add_column("ID", style="cyan", width=12)
table.add_column("名称", style="green")
table.add_column("描述", style="yellow")
table.add_column("时长", style="magenta")
table.add_column("素材数", style="blue")
table.add_column("轨道数", style="red")
table.add_column("创建时间", style="dim")
if verbose:
table.add_column("标签", style="cyan")
table.add_column("资源路径", style="dim")
# 转换为字典格式
templates_data = []
for template in templates:
# 格式化时长
duration = getattr(template, 'duration', 0)
if duration > 60:
duration_str = f"{duration // 60}m{duration % 60}s"
if isinstance(template, TemplateInfo):
templates_data.append(asdict(template))
else:
duration_str = f"{duration}s"
# 格式化创建时间
created_at = getattr(template, 'created_at', '')
if 'T' in created_at:
created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8]
row = [
getattr(template, 'id', '')[:12],
getattr(template, 'name', ''),
getattr(template, 'description', '')[:50] + ('...' if len(getattr(template, 'description', '')) > 50 else ''),
duration_str,
str(getattr(template, 'material_count', 0)),
str(getattr(template, 'track_count', 0)),
created_at
]
if verbose:
tags = getattr(template, 'tags', [])
tags_str = ', '.join(tags) if tags else '-'
resources_path = getattr(template, 'resources_path', '')
row.extend([
tags_str,
resources_path
])
table.add_row(*row)
templates_data.append(template)
console.print(table)
response.success({
"templates": templates_data,
"total_count": len(templates_data),
"limit": limit
})
except Exception as e:
console.print(f"[red]❌ 获取模板列表失败: {str(e)}[/red]")
raise typer.Exit(1)
response.error(-32603, f"获取模板列表失败: {str(e)}")
@template_app.command("get")
def get_template(
template_id: str = typer.Argument(..., help="模板ID"),
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
format_output: str = typer.Option("panel", "--format", "-f", help="输出格式: panel, json")
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""获取模板详情"""
response = create_response_handler(str(uuid4()))
try:
console.print(f"🔍 [bold blue]获取模板详情[/bold blue]")
console.print(f"模板ID: {template_id}")
# 创建模板管理器
manager = TemplateManagerCloud()
manager = TemplateManagerCloud(user_id=user_id or "default")
# 获取模板
template = manager.get_template(template_id)
if not template:
console.print(f"[red]❌ 未找到模板: {template_id}[/red]")
raise typer.Exit(1)
if format_output == "json":
# JSON格式输出
if isinstance(template, TemplateInfo):
from dataclasses import asdict
template_data = asdict(template)
else:
template_data = template
console.print(json.dumps(template_data, ensure_ascii=False, indent=2))
response.error(-32604, f"未找到模板: {template_id}")
return
# 面板格式输出
console.print(f"\n✅ [bold green]模板详情[/bold green]")
# 格式化时长
duration = getattr(template, 'duration', 0)
if duration > 60:
duration_str = f"{duration // 60}分{duration % 60}秒"
# 转换为字典格式
if isinstance(template, TemplateInfo):
template_data = asdict(template)
else:
duration_str = f"{duration}秒"
template_data = template
# 基本信息
basic_info = f"""
🆔 模板ID: {getattr(template, 'id', '')}
📝 名称: {getattr(template, 'name', '')}
📄 描述: {getattr(template, 'description', '')}
⏱️ 时长: {duration_str}
📦 素材数: {getattr(template, 'material_count', 0)}
🎬 轨道数: {getattr(template, 'track_count', 0)}
📅 创建时间: {getattr(template, 'created_at', '')}
🔄 更新时间: {getattr(template, 'updated_at', '')}
"""
panel = Panel(basic_info.strip(), title="模板基本信息", border_style="green")
console.print(panel)
if verbose:
# 详细信息
tags = getattr(template, 'tags', [])
tags_str = ', '.join(tags) if tags else '无'
detail_info = f"""
🏷️ 标签: {tags_str}
📁 草稿路径: {getattr(template, 'draft_content_path', '')}
📂 资源路径: {getattr(template, 'resources_path', '')}
🖼️ 缩略图: {getattr(template, 'thumbnail_path', '') or '无'}
"""
detail_panel = Panel(detail_info.strip(), title="详细信息", border_style="blue")
console.print(detail_panel)
# 画布配置
canvas_config = getattr(template, 'canvas_config', {})
if canvas_config:
canvas_info = f"""
📐 画布配置:
宽度: {canvas_config.get('width', 'Unknown')}
高度: {canvas_config.get('height', 'Unknown')}
帧率: {canvas_config.get('fps', 'Unknown')}
"""
canvas_panel = Panel(canvas_info.strip(), title="画布配置", border_style="yellow")
console.print(canvas_panel)
response.success(template_data)
except Exception as e:
console.print(f"[red]❌ 获取模板详情失败: {str(e)}[/red]")
raise typer.Exit(1)
@template_app.command("detail")
def get_template_detail(
template_id: str = typer.Argument(..., help="模板ID"),
format_output: str = typer.Option("json", "--format", "-f", help="输出格式: json, summary")
):
"""获取模板详细信息（包含轨道和片段）"""
try:
console.print(f"🔍 [bold blue]获取模板详细信息[/bold blue]")
console.print(f"模板ID: {template_id}")
# 创建模板管理器
manager = TemplateManagerCloud()
# 获取模板详细信息
detail = manager.get_template_detail(template_id)
if not detail:
console.print(f"[red]❌ 未找到模板详细信息: {template_id}[/red]")
raise typer.Exit(1)
if format_output == "json":
# JSON格式输出
console.print(json.dumps(detail, ensure_ascii=False, indent=2))
return
# 摘要格式输出
console.print(f"\n✅ [bold green]模板详细信息[/bold green]")
# 基本信息
template_info = detail.get('template', {})
console.print(f"📝 名称: {template_info.get('name', 'Unknown')}")
console.print(f"📄 描述: {template_info.get('description', 'Unknown')}")
console.print(f"⏱️ 时长: {template_info.get('duration', 0)}秒")
# 轨道信息
tracks = detail.get('tracks', [])
console.print(f"\n🎬 [bold green]轨道信息 ({len(tracks)} 个轨道)[/bold green]")
for i, track in enumerate(tracks, 1):
track_type = track.get('type', 'unknown')
segments_count = len(track.get('segments', []))
console.print(f" 轨道 {i}: {track_type} ({segments_count} 个片段)")
# 显示片段信息
for j, segment in enumerate(track.get('segments', []), 1):
segment_name = segment.get('name', f'片段{j}')
start_time = segment.get('start_time', 0)
end_time = segment.get('end_time', 0)
console.print(f" 片段 {j}: {segment_name} ({start_time}s - {end_time}s)")
except Exception as e:
console.print(f"[red]❌ 获取模板详细信息失败: {str(e)}[/red]")
raise typer.Exit(1)
response.error(-32603, f"获取模板详情失败: {str(e)}")
@template_app.command("delete")
def delete_template(
template_id: str = typer.Argument(..., help="模板ID"),
force: bool = typer.Option(False, "--force", "-f", help="强制删除，不询问确认")
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
force: bool = typer.Option(False, "--force", "-f", help="强制删除，不询问确认"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""删除模板"""
response = create_response_handler(str(uuid4()))
try:
console.print(f"🗑️ [bold red]删除模板[/bold red]")
console.print(f"模板ID: {template_id}")
# 创建模板管理器
manager = TemplateManagerCloud()
manager = TemplateManagerCloud(user_id=user_id or "default")
# 获取模板信息
template = manager.get_template(template_id)
if not template:
console.print(f"[red]❌ 未找到模板: {template_id}[/red]")
raise typer.Exit(1)
console.print(f"模板名称: {getattr(template, 'name', 'Unknown')}")
console.print(f"模板描述: {getattr(template, 'description', 'Unknown')}")
# 确认删除
if not force:
confirm = typer.confirm("确定要删除这个模板吗？此操作不可恢复。")
if not confirm:
console.print("❌ 操作已取消")
return
response.error(-32604, f"未找到模板: {template_id}")
return
# 删除模板
success = manager.delete_template(template_id)
if success:
console.print(f"✅ [bold green]模板删除成功[/bold green]")
response.success({
"deleted": True,
"template_id": template_id,
"template_name": getattr(template, 'name', 'Unknown')
})
else:
console.print(f"[red]❌ 模板删除失败[/red]")
raise typer.Exit(1)
response.error(-32603, "模板删除失败")
except Exception as e:
console.print(f"[red]❌ 删除模板失败: {str(e)}[/red]")
raise typer.Exit(1)
response.error(-32603, f"删除模板失败: {str(e)}")
@template_app.command("search")
def search_templates(
query: str = typer.Argument(..., help="搜索关键词"),
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模板"),
limit: int = typer.Option(50, "--limit", "-l", help="显示数量限制"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""搜索模板"""
response = create_response_handler(str(uuid4()))
try:
# 创建模板管理器
manager = TemplateManagerCloud(user_id=user_id or "default")
# 搜索模板
templates = manager.search_templates(query, include_cloud=include_cloud, limit=limit)
# 转换为字典格式
templates_data = []
for template in templates:
if isinstance(template, TemplateInfo):
templates_data.append(asdict(template))
else:
templates_data.append(template)
response.success({
"templates": templates_data,
"query": query,
"total_count": len(templates_data),
"limit": limit
})
except Exception as e:
response.error(-32603, f"搜索模板失败: {str(e)}")
@template_app.command("stats")
def show_stats():
"""显示模板统计信息"""
def get_template_stats(
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""获取模板统计信息"""
response = create_response_handler(str(uuid4()))
try:
console.print(f"📊 [bold blue]模板统计信息[/bold blue]")
# 创建模板管理器
manager = TemplateManagerCloud()
manager = TemplateManagerCloud(user_id=user_id or "default")
# 获取所有模板
templates = manager.get_templates()
# 获取模板数量
template_count = manager.get_template_count(include_cloud=True)
user_template_count = manager.get_template_count(include_cloud=False)
if not templates:
console.print("📭 没有模板")
return
# 获取所有模板进行统计
templates = manager.get_templates(include_cloud=True)
# 计算统计信息
total_templates = len(templates)
total_materials = sum(getattr(template, 'material_count', 0) for template in templates)
total_tracks = sum(getattr(template, 'track_count', 0) for template in templates)
total_duration = sum(getattr(template, 'duration', 0) for template in templates)
# 格式化总时长
if total_duration > 3600:
duration_str = f"{total_duration // 3600}小时{(total_duration % 3600) // 60}分{total_duration % 60}秒"
elif total_duration > 60:
duration_str = f"{total_duration // 60}分{total_duration % 60}秒"
else:
duration_str = f"{total_duration}秒"
# 创建统计面板
stats_content = f"""
📈 [bold green]总体统计[/bold green]
模板总数: {total_templates}
素材总数: {total_materials}
轨道总数: {total_tracks}
总时长: {duration_str}
平均时长: {total_duration // total_templates if total_templates > 0 else 0}秒
"""
stats_panel = Panel(stats_content.strip(), title="模板统计", border_style="green")
console.print(stats_panel)
# 显示最近的模板
recent_templates = sorted(templates, key=lambda x: getattr(x, 'created_at', ''), reverse=True)[:5]
if recent_templates:
console.print(f"\n🕒 [bold green]最近创建的模板[/bold green]")
recent_table = Table()
recent_table.add_column("名称", style="green")
recent_table.add_column("时长", style="magenta")
recent_table.add_column("创建时间", style="dim")
for template in recent_templates:
duration = getattr(template, 'duration', 0)
duration_str = f"{duration}s" if duration < 60 else f"{duration // 60}m{duration % 60}s"
created_at = getattr(template, 'created_at', '')
if 'T' in created_at:
created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8]
recent_table.add_row(
getattr(template, 'name', 'Unknown'),
duration_str,
created_at
)
console.print(recent_table)
response.success({
"total_templates": template_count,
"user_templates": user_template_count,
"cloud_templates": template_count - user_template_count,
"total_materials": total_materials,
"total_tracks": total_tracks,
"total_duration": total_duration,
"average_duration": total_duration // template_count if template_count > 0 else 0
})
except Exception as e:
console.print(f"[red]❌ 获取统计信息失败: {str(e)}[/red]")
raise typer.Exit(1)
response.error(-32603, f"获取统计信息失败: {str(e)}")
@template_app.command("tags")
def get_popular_tags(
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
limit: int = typer.Option(20, "--limit", "-l", help="显示数量限制"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""获取热门标签"""
response = create_response_handler(str(uuid4()))
try:
# 创建模板管理器
manager = TemplateManagerCloud(user_id=user_id or "default")
# 获取热门标签
tags = manager.get_popular_tags(limit=limit)
response.success({
"tags": tags,
"total_count": len(tags),
"limit": limit
})
except Exception as e:
response.error(-32603, f"获取热门标签失败: {str(e)}")
@template_app.command("by-tag")
def get_templates_by_tag(
tag: str = typer.Argument(..., help="标签名称"),
user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模板"),
limit: int = typer.Option(50, "--limit", "-l", help="显示数量限制"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
"""根据标签获取模板"""
response = create_response_handler(str(uuid4()))
try:
# 创建模板管理器
manager = TemplateManagerCloud(user_id=user_id or "default")
# 根据标签获取模板
templates = manager.get_templates_by_tag(tag, include_cloud=include_cloud, limit=limit)
# 转换为字典格式
templates_data = []
for template in templates:
if isinstance(template, TemplateInfo):
templates_data.append(asdict(template))
else:
templates_data.append(template)
response.success({
"templates": templates_data,
"tag": tag,
"total_count": len(templates_data),
"limit": limit
})
except Exception as e:
response.error(-32603, f"根据标签获取模板失败: {str(e)}")
if __name__ == "__main__":