""" 系统管理 CLI 命令 """ import json import uuid from pathlib import Path from typing import Optional import typer from python_core.utils.jsonrpc_enhanced import create_response_handler from python_core.database.user_postgres import user_table from python_core.database.resource_category_postgres import resource_category_table from python_core.config import settings from python_core.utils.logger import setup_logger logger = setup_logger(__name__) # 创建系统应用 system_app = typer.Typer(help="系统管理命令") @system_app.command("init") def init_system( force: bool = typer.Option(False, "--force", help="强制重新初始化"), verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"), json_output: bool = typer.Option(True, "--json", help="JSON格式输出") ): """初始化系统，创建默认数据""" response = create_response_handler() try: # 检查是否已经初始化 session_file = Path("session.json") if session_file.exists() and not force: with open(session_file, 'r', encoding='utf-8') as f: session_data = json.load(f) response.success({ 'already_initialized': True, 'session_data': session_data, 'message': '系统已经初始化，使用 --force 强制重新初始化' }) return # 创建默认用户 default_user_id = str(uuid.uuid4()) default_access_token = str(uuid.uuid4()) # 创建 session.json session_data = { 'user_id': default_user_id, 'access_token': default_access_token, 'username': 'default', 'email': 'default@mixvideo.com', 'initialized_at': '2025-07-13T11:43:00.000000', 'app_version': settings.app_version } with open(session_file, 'w', encoding='utf-8') as f: json.dump(session_data, f, indent=2, ensure_ascii=False) # 创建默认资源分类 default_categories = [ { 'title': '开场介绍', 'ai_prompt': '视频开头的介绍部分，包括标题、logo等', 'color': '#FF6B6B' }, { 'title': '主要内容', 'ai_prompt': '视频的核心内容部分', 'color': '#4ECDC4' }, { 'title': '转场过渡', 'ai_prompt': '视频片段之间的过渡效果', 'color': '#45B7D1' }, { 'title': '结尾总结', 'ai_prompt': '视频结尾的总结和结束语', 'color': '#96CEB4' }, { 'title': '背景音乐', 'ai_prompt': '视频的背景音乐和音效', 'color': '#FFEAA7' }, { 'title': '字幕文本', 'ai_prompt': '视频中的字幕和文字说明', 'color': '#DDA0DD' }, { 'title': '特效动画', 'ai_prompt': '视频中的特效和动画效果', 'color': '#98D8C8' }, { 'title': '产品展示', 'ai_prompt': '产品或服务的展示片段', 'color': '#F7DC6F' } ] created_categories = [] for category_data in default_categories: try: from python_core.database.types import ResourceCategory category = ResourceCategory( id="", # 将由数据库生成 title=category_data['title'], ai_prompt=category_data['ai_prompt'], color=category_data['color'], is_active=True, is_cloud=False, user_id='default', created_at="", updated_at="" ) category_id = resource_category_table.create_category(category) if category_id: created_categories.append(category_data['title']) logger.info(f"Created default category: {category_data['title']}") except Exception as e: logger.warning(f"Failed to create category {category_data['title']}: {e}") # 创建必要的目录 directories = [ 'templates', 'projects', 'uploads', 'temp', 'logs' ] created_dirs = [] for dir_name in directories: dir_path = Path(dir_name) if not dir_path.exists(): dir_path.mkdir(parents=True, exist_ok=True) created_dirs.append(str(dir_path)) logger.info(f"Created directory: {dir_path}") response.success({ 'initialized': True, 'session_file': str(session_file), 'session_data': session_data, 'created_categories': created_categories, 'created_directories': created_dirs, 'message': f'系统初始化成功，创建了 {len(created_categories)} 个默认分类和 {len(created_dirs)} 个目录' }) except Exception as e: logger.error(f"系统初始化失败: {e}") response.error(-32603, f"系统初始化失败: {str(e)}") @system_app.command("status") def system_status( verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"), json_output: bool = typer.Option(True, "--json", help="JSON格式输出") ): """检查系统状态""" response = create_response_handler() try: # 检查数据库连接 db_status = "connected" try: # 测试各个表的连接 user_count = user_table.get_user_count() category_count = resource_category_table.get_category_count() except Exception as e: db_status = f"error: {str(e)}" user_count = 0 category_count = 0 # 检查 session.json session_file = Path("session.json") session_status = "exists" if session_file.exists() else "missing" session_data = None if session_file.exists(): try: with open(session_file, 'r', encoding='utf-8') as f: session_data = json.load(f) except Exception as e: session_status = f"error: {str(e)}" # 检查目录 directories = ['templates', 'projects', 'uploads', 'temp', 'logs'] dir_status = {} for dir_name in directories: dir_path = Path(dir_name) dir_status[dir_name] = { 'exists': dir_path.exists(), 'is_dir': dir_path.is_dir() if dir_path.exists() else False, 'path': str(dir_path.absolute()) } response.success({ 'database': { 'status': db_status, 'user_count': user_count, 'category_count': category_count }, 'session': { 'status': session_status, 'data': session_data }, 'directories': dir_status, 'app_version': settings.app_version, 'message': f'系统状态检查完成 - 数据库: {db_status}, 会话: {session_status}' }) except Exception as e: logger.error(f"系统状态检查失败: {e}") response.error(-32603, f"系统状态检查失败: {str(e)}") @system_app.command("reset") def reset_system( confirm: bool = typer.Option(False, "--confirm", help="确认重置系统"), keep_users: bool = typer.Option(True, "--keep-users", help="保留用户数据"), verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"), json_output: bool = typer.Option(True, "--json", help="JSON格式输出") ): """重置系统（危险操作）""" response = create_response_handler() if not confirm: response.error(-32602, "请使用 --confirm 参数确认重置操作") return try: reset_actions = [] # 删除 session.json session_file = Path("session.json") if session_file.exists(): session_file.unlink() reset_actions.append("删除 session.json") # 清理分类数据 try: categories = resource_category_table.get_all_resource_categories() for category in categories: resource_category_table.delete_resource_category(category.id, hard_delete=True) reset_actions.append(f"删除 {len(categories)} 个分类") except Exception as e: logger.warning(f"清理分类数据失败: {e}") # 可选：清理用户数据 if not keep_users: try: # 这里可以添加清理用户数据的逻辑 reset_actions.append("清理用户数据") except Exception as e: logger.warning(f"清理用户数据失败: {e}") response.success({ 'reset': True, 'actions': reset_actions, 'keep_users': keep_users, 'message': f'系统重置完成，执行了 {len(reset_actions)} 个操作' }) except Exception as e: logger.error(f"系统重置失败: {e}") response.error(-32603, f"系统重置失败: {str(e)}") if __name__ == "__main__": system_app()