#!/usr/bin/env python3 """ 慢剧工作台命令行工具 用于管理项目、监控Agent状态、查看系统信息 """ import argparse import json import sqlite3 import os import sys from datetime import datetime from pathlib import Path import subprocess class ManhuaStudioCLI: def __init__(self): # 使用硬编码配置，避免配置文件读取问题 self.config = { "version": "2.0.0", "system_name": "慢剧生成工作台", "database_config": { "engines": { "projects": { "path": "/root/data/disk/system/database/projects.db" }, "agents": { "path": "/root/data/disk/system/database/agents.db" }, "analytics": { "path": "/root/data/disk/system/database/analytics.db" } } }, "storage_config": { "root_path": "/root/data/disk", "projects_path": "/root/data/disk/projects", "temp_path": "/root/data/disk/temp" }, "agent_config": { "agents": { "cto": { "pid": 1738491, "status": "running" }, "collector": { "pid": 1738514, "status": "running" }, "writer": { "pid": 1738507, "status": "running" }, "producer": { "pid": 1738498, "status": "running" }, "publisher": { "pid": 1738521, "status": "running" }, "analyzer": { "pid": 1738528, "status": "running" } } }, "skill_config": { "enabled_skills": [ "db", "sqlite", "automation-workflows", "api-gateway", "monitor", "analytics", "ai-drama-prompt-factory", "seedance2-skill", "hot-topics", "bilibili-all-in-one", "seedance-creator", "seedance-shot-design" ] } } self.db_config = self.config['database_config']['engines'] def load_config(self): """加载配置文件（已弃用，使用硬编码配置）""" # 保持接口兼容性 self.db_config = self.config['database_config']['engines'] def connect_db(self, db_name): """连接到指定数据库""" db_path = self.db_config[db_name]['path'] conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn def list_projects(self, args): """列出所有项目""" conn = self.connect_db('projects') cursor = conn.cursor() query = "SELECT project_id, name, status, created_at FROM projects" if args.status: query += f" WHERE status = '{args.status}'" cursor.execute(query) projects = cursor.fetchall() print(f"\n📋 项目列表 (共 {len(projects)} 个):") print("=" * 80) print(f"{'ID':<20} {'名称':<30} {'状态':<15} {'创建时间':<20}") print("-" * 80) for project in projects: print(f"{project['project_id']:<20} {project['name'][:28]:<30} {project['status']:<15} {project['created_at'][:19]:<20}") conn.close() def create_project(self, args): """创建新项目""" project_id = datetime.now().strftime("%Y%m%d%H%M%S") project_name = args.name project_desc = args.description or "" # 1. 创建项目目录 project_dir = f"/root/data/disk/projects/{project_id}" os.makedirs(project_dir, exist_ok=True) # 复制模板结构 template_dir = "/root/data/disk/projects/template" self.copy_template_structure(template_dir, project_dir) # 2. 更新项目信息文件 project_info_path = f"{project_dir}/00_metadata/project_info.json" with open(project_info_path, 'r', encoding='utf-8') as f: project_info = json.load(f) project_info['project_template']['name'] = project_name project_info['project_template']['description'] = project_desc with open(project_info_path, 'w', encoding='utf-8') as f: json.dump(project_info, f, ensure_ascii=False, indent=2) # 3. 在数据库中创建项目记录 conn = self.connect_db('projects') cursor = conn.cursor() cursor.execute(''' INSERT INTO projects (project_id, name, description, status, created_by) VALUES (?, ?, ?, 'draft', ?) ''', (project_id, project_name, project_desc, 'cli')) conn.commit() conn.close() print(f"\n🎉 项目创建成功!") print(f"📁 项目ID: {project_id}") print(f"📁 项目目录: {project_dir}") print(f"📝 项目名称: {project_name}") if args.start: self.start_project_workflow(project_id) def copy_template_structure(self, src_dir, dst_dir): """复制模板目录结构""" for root, dirs, files in os.walk(src_dir): # 计算相对路径 rel_path = os.path.relpath(root, src_dir) dst_path = os.path.join(dst_dir, rel_path) # 创建目录 if rel_path != '.': os.makedirs(dst_path, exist_ok=True) # 复制文件 for file in files: src_file = os.path.join(root, file) dst_file = os.path.join(dst_path, file) # 跳过某些特定文件 if file == 'workflow_status.json': continue # 复制文件 with open(src_file, 'r', encoding='utf-8') as f: content = f.read() with open(dst_file, 'w', encoding='utf-8') as f: f.write(content) def start_project_workflow(self, project_id): """启动项目工作流""" print(f"\n🚀 启动项目工作流: {project_id}") # 更新项目状态 conn = self.connect_db('projects') cursor = conn.cursor() cursor.execute(''' UPDATE projects SET status = 'collecting', workflow_step = 'collection' WHERE project_id = ? ''', (project_id,)) conn.commit() conn.close() # 创建Agent任务 self.create_agent_task(project_id, 'collector', 'collect_hot_topics') print(f"✅ 工作流已启动，采集官Agent开始收集素材...") print(f"📊 使用 'manhua status {project_id}' 查看进度") def create_agent_task(self, project_id, agent_name, task_type): """创建Agent任务""" conn = self.connect_db('agents') cursor = conn.cursor() task_id = f"{project_id}_{agent_name}_{task_type}_{datetime.now().strftime('%H%M%S')}" cursor.execute(''' INSERT INTO agent_tasks (task_id, project_id, agent_name, task_type, status, created_at) VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP) ''', (task_id, project_id, agent_name, task_type)) conn.commit() conn.close() return task_id def show_project_status(self, args): """显示项目状态""" project_id = args.project_id # 获取项目信息 conn = self.connect_db('projects') cursor = conn.cursor() cursor.execute(''' SELECT * FROM projects WHERE project_id = ? ''', (project_id,)) project = cursor.fetchone() conn.close() if not project: print(f"❌ 项目不存在: {project_id}") return print(f"\n📊 项目状态: {project['name']}") print("=" * 60) print(f"📁 项目ID: {project['project_id']}") print(f"📝 描述: {project['description'] or '无描述'}") print(f"📈 状态: {project['status']}") print(f"🔧 当前步骤: {project['workflow_step'] or '未开始'}") print(f"🕐 创建时间: {project['created_at']}") print(f"🕐 更新时间: {project['updated_at']}") # 获取Agent任务 conn = self.connect_db('agents') cursor = conn.cursor() cursor.execute(''' SELECT agent_name, task_type, status, created_at, completed_at FROM agent_tasks WHERE project_id = ? ORDER BY created_at DESC LIMIT 10 ''', (project_id,)) tasks = cursor.fetchall() conn.close() if tasks: print(f"\n📋 最近任务 (共 {len(tasks)} 个):") print("-" * 60) for task in tasks: status_icon = "✅" if task['status'] == 'completed' else "🔄" if task['status'] == 'running' else "⏳" print(f"{status_icon} {task['agent_name']}: {task['task_type']} ({task['status']})") if task['completed_at']: print(f" 完成于: {task['completed_at']}") # 检查项目目录 project_dir = f"/root/data/disk/projects/{project_id}" if os.path.exists(project_dir): print(f"\n📁 项目目录: {project_dir}") # 统计文件 file_count = 0 for root, dirs, files in os.walk(project_dir): file_count += len(files) print(f"📄 文件数量: {file_count}") else: print(f"\n⚠️ 项目目录不存在: {project_dir}") def list_agents(self, args): """列出Agent状态""" conn = self.connect_db('agents') cursor = conn.cursor() cursor.execute(''' SELECT * FROM agent_status ORDER BY agent_name ''') agents = cursor.fetchall() conn.close() print(f"\n🤖 Agent状态 (共 {len(agents)} 个):") print("=" * 60) for agent in agents: status_icon = { 'idle': '🟢', 'busy': '🟡', 'error': '🔴', 'offline': '⚫' }.get(agent['status'], '❓') print(f"{status_icon} {agent['agent_name']}: {agent['status']}") if agent['current_task_id']: print(f" 当前任务: {agent['current_task_id']}") print(f" 最后心跳: {agent['last_heartbeat']}") def system_info(self, args): """显示系统信息""" print(f"\n🖥️ 慢剧工作台系统信息") print("=" * 60) # 配置文件信息 print(f"📋 系统版本: {self.config['version']}") print(f"📋 系统名称: {self.config['system_name']}") # 存储信息 storage = self.config['storage_config'] print(f"\n💾 存储配置:") print(f" 根目录: {storage['root_path']}") print(f" 项目目录: {storage['projects_path']}") print(f" 临时目录: {storage['temp_path']}") # Agent信息 agents = self.config['agent_config']['agents'] print(f"\n🤖 Agent配置 (共 {len(agents)} 个):") for agent_name, agent_config in agents.items(): print(f" {agent_name}: PID={agent_config.get('pid', 'N/A')}, 状态={agent_config['status']}") # 技能信息 skills = self.config['skill_config']['enabled_skills'] print(f"\n🔧 已启用技能 (共 {len(skills)} 个):") for i in range(0, len(skills), 4): print(f" {', '.join(skills[i:i+4])}") # 数据库信息 print(f"\n🗄️ 数据库:") for db_name, db_config in self.db_config.items(): db_path = db_config['path'] if os.path.exists(db_path): size = os.path.getsize(db_path) print(f" {db_name}: {db_path} ({size/1024:.1f} KB)") else: print(f" {db_name}: {db_path} (不存在)") def cleanup_temp(self, args): """清理临时文件""" temp_dir = "/root/data/disk/temp" if not os.path.exists(temp_dir): print(f"❌ 临时目录不存在: {temp_dir}") return # 统计清理前的文件 file_count = 0 total_size = 0 for root, dirs, files in os.walk(temp_dir): file_count += len(files) for file in files: file_path = os.path.join(root, file) total_size += os.path.getsize(file_path) print(f"\n🧹 清理临时文件") print(f" 目录: {temp_dir}") print(f" 文件数量: {file_count}") print(f" 总大小: {total_size/1024/1024:.2f} MB") if args.dry_run: print(f" 模拟运行，不实际删除文件") return if not args.force: confirm = input(f"\n⚠️ 确定要删除 {file_count} 个临时文件吗? (y/N): ") if confirm.lower() != 'y': print("❌ 取消清理") return # 清理文件 deleted_count = 0 deleted_size = 0 for root, dirs, files in os.walk(temp_dir): for file in files: file_path = os.path.join(root, file) try: file_size = os.path.getsize(file_path) os.remove(file_path) deleted_count += 1 deleted_size += file_size except Exception as e: print(f" 删除失败 {file_path}: {e}") print(f"\n✅ 清理完成") print(f" 删除文件: {deleted_count} 个") print(f" 释放空间: {deleted_size/1024/1024:.2f} MB") def main(): """主函数""" parser = argparse.ArgumentParser( description="慢剧工作台命令行工具", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: %(prog)s projects list # 列出所有项目 %(prog)s projects create "项目名称" # 创建新项目 %(prog)s status 20260428150000 # 查看项目状态 %(prog)s agents list # 查看Agent状态 %(prog)s system info # 查看系统信息 %(prog)s cleanup --dry-run # 模拟清理临时文件 """ ) subparsers = parser.add_subparsers(dest="command", help="命令") # 项目管理命令 projects_parser = subparsers.add_parser("projects", help="项目管理") projects_subparsers = projects_parser.add_subparsers(dest="subcommand") # 列出项目 list_parser = projects_subparsers.add_parser("list", help="列出所有项目") list_parser.add_argument("--status", help="按状态筛选") list_parser.set_defaults(func=lambda args: ManhuaStudioCLI().list_projects(args)) # 创建项目 create_parser = projects_subparsers.add_parser("create", help="创建新项目") create_parser.add_argument("name", help="项目名称") create_parser.add_argument("--description", "-d", help="项目描述") create_parser.add_argument("--start", "-s", action="store_true", help="立即启动工作流") create_parser.set_defaults(func=lambda args: ManhuaStudioCLI().create_project(args)) # 项目状态命令 status_parser = subparsers.add_parser("status", help="查看项目状态") status_parser.add_argument("project_id", help="项目ID") status_parser.set_defaults(func=lambda args: ManhuaStudioCLI().show_project_status(args)) # Agent管理命令 agents_parser = subparsers.add_parser("agents", help="Agent管理") agents_subparsers = agents_parser.add_subparsers(dest="subcommand") # 列出Agent agents_list_parser = agents_subparsers.add_parser("list", help="列出Agent状态") agents_list_parser.set_defaults(func=lambda args: ManhuaStudioCLI().list_agents(args)) # 系统信息命令 system_parser = subparsers.add_parser("system", help="系统管理") system_subparsers = system_parser.add_subparsers(dest="subcommand") # 系统信息 info_parser = system_subparsers.add_parser("info", help="显示系统信息") info_parser.set_defaults(func=lambda args: ManhuaStudioCLI().system_info(args)) # 清理命令 cleanup_parser = subparsers.add_parser("cleanup", help="清理临时文件") cleanup_parser.add_argument("--dry-run", "-n", action="store_true", help="模拟运行") cleanup_parser.add_argument("--force", "-f", action="store_true", help="强制清理") cleanup_parser.set_defaults(func=lambda args: ManhuaStudioCLI().cleanup_temp(args)) # 解析参数 args = parser.parse_args() if not args.command: parser.print_help() return 1 try: # 执行命令 args.func(args) return 0 except AttributeError: parser.print_help() return 1 except Exception as e: print(f"❌ 命令执行失败: {e}") return 1 if __name__ == "__main__": sys.exit(main())