From 5f1fc7d9c81818a9c0902658363526b2f4498128 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 12 Jul 2025 17:07:14 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BA=91=E7=AB=AF=E7=B2=97=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python_core/cli/cli.py | 3 + python_core/cli/commands/category.py | 491 +++++++++++++++++++++++ python_core/database/__init__.py | 0 python_core/database/db.py | 10 + python_core/database/template.py | 13 + python_core/{storage => }/kv.py | 55 ++- python_core/services/template_manager.py | 130 ------ python_core/storage/__init__.py | 6 - test_category_cli.py | 116 ++++++ 9 files changed, 659 insertions(+), 165 deletions(-) create mode 100644 python_core/cli/commands/category.py create mode 100644 python_core/database/__init__.py create mode 100644 python_core/database/db.py create mode 100644 python_core/database/template.py rename python_core/{storage => }/kv.py (88%) delete mode 100644 python_core/storage/__init__.py create mode 100644 test_category_cli.py diff --git a/python_core/cli/cli.py b/python_core/cli/cli.py index 7b8e074..8b286a8 100644 --- a/python_core/cli/cli.py +++ b/python_core/cli/cli.py @@ -10,6 +10,7 @@ import typer # 导入命令模块 from python_core.cli.commands import scene_detect from python_core.cli.commands.template import template_app +from python_core.cli.commands.category import category_app app = typer.Typer( name="mixvideo", @@ -19,6 +20,7 @@ app = typer.Typer( 功能完整的视频处理和管理工具套件： • 🎯 场景检测 - 智能识别视频场景变化 • 📋 模板管理 - 视频模板批量导入、列表查看、详情获取 + • 🏷️ 分类管理 - 资源分类创建、更新、删除、搜索 • 📤 媒体管理 - 上传、处理、组织视频文件 • ⚙️ 系统管理 - 配置、状态、存储管理 """, @@ -29,6 +31,7 @@ app = typer.Typer( # 添加命令组到主应用 app.add_typer(scene_detect, name="scene") app.add_typer(template_app, name="template") +app.add_typer(category_app, name="category") @app.command() def init(): diff --git a/python_core/cli/commands/category.py b/python_core/cli/commands/category.py new file mode 100644 index 0000000..5f719c3 --- /dev/null +++ b/python_core/cli/commands/category.py @@ -0,0 +1,491 @@ +""" +资源分类管理CLI命令 +""" + +from typing import Optional +import typer +from rich.console import Console +from rich.table import Table +from rich.panel import Panel +import json +import re + +from python_core.services.resource_category_manager import ResourceCategoryManager, ResourceCategory +from python_core.utils.logger import logger + +console = Console() +category_app = typer.Typer(name="category", help="资源分类管理命令") + + +def validate_color(color: str) -> bool: + """验证颜色格式（hex格式）""" + pattern = r'^#[0-9A-Fa-f]{6}$' + return bool(re.match(pattern, color)) + + +@category_app.command("list") +def list_categories( + include_inactive: bool = typer.Option(False, "--include-inactive", "-i", help="包含已禁用的分类"), + limit: int = typer.Option(20, "--limit", "-l", help="显示数量限制"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"), + format_output: str = typer.Option("table", "--format", "-f", help="输出格式: table, json") +): + """列出所有资源分类""" + + try: + console.print(f"📋 [bold blue]资源分类列表[/bold blue]") + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 获取分类列表 + categories = manager.get_all_categories() + + # 过滤非活跃分类 + if not include_inactive: + categories = [cat for cat in categories if cat.get('is_active', True)] + + if not categories: + console.print("📭 没有分类") + return + + # 限制显示数量 + total_count = len(categories) + if len(categories) > limit: + categories = categories[:limit] + console.print(f"📊 显示前 {limit} 个分类（共 {total_count} 个）") + else: + console.print(f"📊 共 {total_count} 个分类") + + if format_output == "json": + # JSON格式输出 + console.print(json.dumps(categories, 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("AI提示词", style="yellow") + table.add_column("颜色", style="magenta") + table.add_column("状态", style="blue") + table.add_column("创建时间", style="dim") + + if verbose: + table.add_column("更新时间", style="dim") + + for category in categories: + # 状态 + status = "✅ 启用" if category.get('is_active', True) else "❌ 禁用" + + # 格式化创建时间 + created_at = category.get('created_at', '') + if 'T' in created_at: + created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8] + + # 颜色显示 + color = category.get('color', '') + color_display = f"[{color}]●[/{color}] {color}" if color else color + + row = [ + category.get('id', '')[:12], + category.get('title', ''), + category.get('ai_prompt', '')[:30] + ('...' if len(category.get('ai_prompt', '')) > 30 else ''), + color_display, + status, + created_at + ] + + if verbose: + updated_at = category.get('updated_at', '') + if 'T' in updated_at: + updated_at = updated_at.split('T')[0] + ' ' + updated_at.split('T')[1][:8] + row.append(updated_at) + + table.add_row(*row) + + console.print(table) + + except Exception as e: + console.print(f"[red]❌ 获取分类列表失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@category_app.command("create") +def create_category( + title: str = typer.Argument(..., help="分类标题"), + ai_prompt: str = typer.Argument(..., help="AI识别提示词"), + color: str = typer.Argument(..., help="展示颜色（hex格式，如 #FF5733）"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """创建新的资源分类""" + + try: + console.print(f"🆕 [bold blue]创建资源分类[/bold blue]") + console.print(f"标题: {title}") + console.print(f"AI提示词: {ai_prompt}") + console.print(f"颜色: {color}") + + # 验证颜色格式 + if not validate_color(color): + console.print(f"[red]❌ 颜色格式无效: {color}[/red]") + console.print("颜色必须是hex格式，如: #FF5733") + raise typer.Exit(1) + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 创建分类 + result = manager.create_category(title, ai_prompt, color) + + console.print(f"\n✅ [bold green]分类创建成功![/bold green]") + console.print(f"分类ID: {result['id']}") + console.print(f"创建时间: {result['created_at']}") + + if verbose: + # 显示分类详细信息 + category_info = f""" +🆔 分类ID: {result['id']} +📝 标题: {result['title']} +🤖 AI提示词: {result['ai_prompt']} +🎨 颜色: {result['color']} +📅 创建时间: {result['created_at']} +🔄 更新时间: {result['updated_at']} +✅ 状态: {'启用' if result['is_active'] else '禁用'} + """ + + panel = Panel(category_info.strip(), title="分类详情", border_style="green") + console.print(panel) + + except Exception as e: + console.print(f"[red]❌ 创建分类失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@category_app.command("get") +def get_category( + category_id: str = typer.Argument(..., help="分类ID"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"), + format_output: str = typer.Option("panel", "--format", "-f", help="输出格式: panel, json") +): + """获取分类详情""" + + try: + console.print(f"🔍 [bold blue]获取分类详情[/bold blue]") + console.print(f"分类ID: {category_id}") + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 获取分类 + category = manager.get_category_by_id(category_id) + + if not category: + console.print(f"[red]❌ 未找到分类: {category_id}[/red]") + raise typer.Exit(1) + + if format_output == "json": + # JSON格式输出 + console.print(json.dumps(category, ensure_ascii=False, indent=2)) + return + + # 面板格式输出 + console.print(f"\n✅ [bold green]分类详情[/bold green]") + + # 基本信息 + basic_info = f""" +🆔 分类ID: {category.get('id', '')} +📝 标题: {category.get('title', '')} +🤖 AI提示词: {category.get('ai_prompt', '')} +🎨 颜色: {category.get('color', '')} +📅 创建时间: {category.get('created_at', '')} +🔄 更新时间: {category.get('updated_at', '')} +✅ 状态: {'启用' if category.get('is_active', True) else '禁用'} + """ + + panel = Panel(basic_info.strip(), title="分类基本信息", border_style="green") + console.print(panel) + + if verbose: + # 显示颜色预览 + color = category.get('color', '') + if color: + color_info = f""" +🎨 颜色预览: [{color}]████████[/{color}] {color} + """ + color_panel = Panel(color_info.strip(), title="颜色预览", border_style="blue") + console.print(color_panel) + + except Exception as e: + console.print(f"[red]❌ 获取分类详情失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@category_app.command("update") +def update_category( + category_id: str = typer.Argument(..., help="分类ID"), + title: Optional[str] = typer.Option(None, "--title", "-t", help="新的标题"), + ai_prompt: Optional[str] = typer.Option(None, "--ai-prompt", "-p", help="新的AI提示词"), + color: Optional[str] = typer.Option(None, "--color", "-c", help="新的颜色（hex格式）"), + active: Optional[bool] = typer.Option(None, "--active", "-a", help="是否启用"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """更新分类信息""" + + try: + console.print(f"🔄 [bold blue]更新分类[/bold blue]") + console.print(f"分类ID: {category_id}") + + # 验证至少有一个更新参数 + if not any([title, ai_prompt, color, active is not None]): + console.print(f"[red]❌ 请至少提供一个要更新的参数[/red]") + raise typer.Exit(1) + + # 验证颜色格式（如果提供） + if color and not validate_color(color): + console.print(f"[red]❌ 颜色格式无效: {color}[/red]") + console.print("颜色必须是hex格式，如: #FF5733") + raise typer.Exit(1) + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 更新分类 + result = manager.update_category( + category_id=category_id, + title=title, + ai_prompt=ai_prompt, + color=color, + is_active=active + ) + + if not result: + console.print(f"[red]❌ 未找到分类或更新失败: {category_id}[/red]") + raise typer.Exit(1) + + console.print(f"\n✅ [bold green]分类更新成功![/bold green]") + console.print(f"更新时间: {result['updated_at']}") + + if verbose: + # 显示更新后的分类信息 + category_info = f""" +🆔 分类ID: {result['id']} +📝 标题: {result['title']} +🤖 AI提示词: {result['ai_prompt']} +🎨 颜色: {result['color']} +📅 创建时间: {result['created_at']} +🔄 更新时间: {result['updated_at']} +✅ 状态: {'启用' if result['is_active'] else '禁用'} + """ + + panel = Panel(category_info.strip(), title="更新后的分类信息", border_style="green") + console.print(panel) + + except Exception as e: + console.print(f"[red]❌ 更新分类失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@category_app.command("delete") +def delete_category( + category_id: str = typer.Argument(..., help="分类ID"), + force: bool = typer.Option(False, "--force", "-f", help="强制删除，不询问确认") +): + """删除分类""" + + try: + console.print(f"🗑️ [bold red]删除分类[/bold red]") + console.print(f"分类ID: {category_id}") + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 获取分类信息 + category = manager.get_category_by_id(category_id) + if not category: + console.print(f"[red]❌ 未找到分类: {category_id}[/red]") + raise typer.Exit(1) + + console.print(f"分类标题: {category.get('title', 'Unknown')}") + console.print(f"AI提示词: {category.get('ai_prompt', 'Unknown')}") + + # 确认删除 + if not force: + confirm = typer.confirm("确定要删除这个分类吗？此操作不可恢复。") + if not confirm: + console.print("❌ 操作已取消") + return + + # 删除分类 + success = manager.delete_category(category_id) + + if success: + console.print(f"✅ [bold green]分类删除成功[/bold green]") + else: + console.print(f"[red]❌ 分类删除失败[/red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"[red]❌ 删除分类失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@category_app.command("search") +def search_categories( + keyword: str = typer.Argument(..., help="搜索关键词"), + limit: int = typer.Option(10, "--limit", "-l", help="显示数量限制"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """搜索分类""" + + try: + console.print(f"🔍 [bold blue]搜索分类[/bold blue]") + console.print(f"关键词: {keyword}") + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 搜索分类 + categories = manager.search_categories(keyword) + + if not categories: + console.print(f"📭 没有找到匹配 '{keyword}' 的分类") + return + + console.print(f"✅ 找到 {len(categories)} 个匹配的分类") + + # 限制显示数量 + if len(categories) > limit: + categories = categories[:limit] + console.print(f"📊 显示前 {limit} 个结果") + + # 创建表格 + table = Table(title=f"搜索结果: {keyword}") + table.add_column("ID", style="cyan", width=12) + table.add_column("标题", style="green") + table.add_column("AI提示词", style="yellow") + table.add_column("颜色", style="magenta") + table.add_column("状态", style="blue") + + if verbose: + table.add_column("创建时间", style="dim") + + for category in categories: + status = "✅ 启用" if category.get('is_active', True) else "❌ 禁用" + + # 颜色显示 + color = category.get('color', '') + color_display = f"[{color}]●[/{color}] {color}" if color else color + + row = [ + category.get('id', '')[:12], + category.get('title', ''), + category.get('ai_prompt', '')[:30] + ('...' if len(category.get('ai_prompt', '')) > 30 else ''), + color_display, + status + ] + + if verbose: + created_at = category.get('created_at', '') + if 'T' in created_at: + created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8] + row.append(created_at) + + table.add_row(*row) + + console.print(table) + + except Exception as e: + console.print(f"[red]❌ 搜索分类失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@category_app.command("stats") +def show_stats(): + """显示分类统计信息""" + + try: + console.print(f"📊 [bold blue]分类统计信息[/bold blue]") + + # 创建分类管理器 + manager = ResourceCategoryManager() + + # 获取所有分类 + categories = manager.get_all_categories() + + if not categories: + console.print("📭 没有分类") + return + + # 计算统计信息 + total_categories = len(categories) + active_categories = len([cat for cat in categories if cat.get('is_active', True)]) + inactive_categories = total_categories - active_categories + + # 颜色统计 + colors = {} + for category in categories: + color = category.get('color', '') + if color: + colors[color] = colors.get(color, 0) + 1 + + # 创建统计面板 + stats_content = f""" +📈 [bold green]总体统计[/bold green] + 分类总数: {total_categories} + 启用分类: {active_categories} + 禁用分类: {inactive_categories} + 使用的颜色: {len(colors)} + """ + + stats_panel = Panel(stats_content.strip(), title="分类统计", border_style="green") + console.print(stats_panel) + + # 显示最近的分类 + recent_categories = sorted(categories, key=lambda x: x.get('created_at', ''), reverse=True)[:5] + + if recent_categories: + 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 category in recent_categories: + color = category.get('color', '') + color_display = f"[{color}]●[/{color}] {color}" if color else color + + created_at = category.get('created_at', '') + if 'T' in created_at: + created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8] + + recent_table.add_row( + category.get('title', 'Unknown'), + color_display, + created_at + ) + + console.print(recent_table) + + # 显示颜色使用统计 + if colors: + console.print(f"\n🎨 [bold green]颜色使用统计[/bold green]") + + color_table = Table() + color_table.add_column("颜色", style="magenta") + color_table.add_column("使用次数", style="cyan") + + for color, count in sorted(colors.items(), key=lambda x: x[1], reverse=True): + color_display = f"[{color}]●[/{color}] {color}" + color_table.add_row(color_display, str(count)) + + console.print(color_table) + + except Exception as e: + console.print(f"[red]❌ 获取统计信息失败: {str(e)}[/red]") + raise typer.Exit(1) + + +if __name__ == "__main__": + category_app() diff --git a/python_core/database/__init__.py b/python_core/database/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/python_core/database/db.py b/python_core/database/db.py new file mode 100644 index 0000000..a27f74c --- /dev/null +++ b/python_core/database/db.py @@ -0,0 +1,10 @@ + +from python_core.kv import kv + +class Db: + + def __init__(self, key: str): + self.kv = kv + self.key = key + pass + \ No newline at end of file diff --git a/python_core/database/template.py b/python_core/database/template.py new file mode 100644 index 0000000..c92186d --- /dev/null +++ b/python_core/database/template.py @@ -0,0 +1,13 @@ + +from python_core.kv import kv +from .db import Db + + +# 模板管理 +# 实现模板的增删改查 +# 主键 是模板的ID uuid +class TemplateDb(Db): + + def __init__(self): + self.key = "templates" + pass \ No newline at end of file diff --git a/python_core/storage/kv.py b/python_core/kv.py similarity index 88% rename from python_core/storage/kv.py rename to python_core/kv.py index f6af204..78bf0a7 100644 --- a/python_core/storage/kv.py +++ b/python_core/kv.py @@ -13,8 +13,27 @@ class Kv: self.cf_kv_api_token = settings.cloudflare_api_key self.cf_kv_id = settings.cloudflare_kv_id - def get(self, key: str): - ... + def gets(self, keys: list[str]): + try: + with httpx.Client() as client: + response = client.post( + f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk/get", + headers={"Authorization": f"Bearer {self.cf_kv_api_token}"}, + json={ + "keys": keys, + "type": "json" + } + ) + response.raise_for_status() + except httpx.RequestError as e: + logger.error(f"An error occurred while put kv to cloudflare") + raise e + except httpx.HTTPStatusError as e: + logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}") + raise e + except Exception as e: + logger.error(f"An unexpected error occurred: {str(e)}") + raise e def sets(self, caches: Dict[str, str]): try: @@ -41,32 +60,7 @@ class Kv: except Exception as e: logger.error(f"An unexpected error occurred: {str(e)}") raise e - - def set(self, key: str, value: str): - try: - with httpx.Client() as client: - response = client.put( - f"https://api.cloudflare.com/client/v4/accounts/{self.cf_account_id}/storage/kv/namespaces/{self.cf_kv_id}/bulk", - headers={"Authorization": f"Bearer {self.cf_kv_api_token}"}, - json=[ - { - "based64": False, - "key": key, - "value": value, - } - ] - ) - response.raise_for_status() - except httpx.RequestError as e: - logger.error(f"An error occurred while put kv to cloudflare") - raise e - except httpx.HTTPStatusError as e: - logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}") - raise e - except Exception as e: - logger.error(f"An unexpected error occurred: {str(e)}") - raise e - + def remove(self, keys: list[str]): with httpx.Client() as client: try: @@ -84,4 +78,7 @@ class Kv: raise e except Exception as e: logger.error(f"An unexpected error occurred: {str(e)}") - raise e \ No newline at end of file + raise e + + +kv = Kv() \ No newline at end of file diff --git a/python_core/services/template_manager.py b/python_core/services/template_manager.py index b310fb0..e3a61b7 100644 --- a/python_core/services/template_manager.py +++ b/python_core/services/template_manager.py @@ -13,7 +13,6 @@ from dataclasses import dataclass, asdict from datetime import datetime from ..utils.logger import setup_logger -from ..utils.jsonrpc import create_response_handler, create_progress_reporter, JSONRPCError from ..config import settings logger = setup_logger(__name__) @@ -548,132 +547,3 @@ class TemplateManager: except Exception as e: logger.error(f"Failed to get template detail for {template_id}: {e}") return None - - -def main(): - """Command line interface for template management.""" - import argparse - import json - import sys - - parser = argparse.ArgumentParser(description="Template Manager") - parser.add_argument("--action", required=True, help="Action to perform") - parser.add_argument("--source_folder", help="Source folder for batch import") - parser.add_argument("--template_id", help="Template ID for operations") - - args = parser.parse_args() - - # Create JSON-RPC response handler - rpc = create_response_handler("template_manager") - progress = create_progress_reporter() - - try: - manager = TemplateManager() - - if args.action == "batch_import": - if not args.source_folder: - rpc.error(JSONRPCError.INVALID_PARAMS, "Source folder is required for batch import") - return - - def progress_callback(message): - # Parse progress information from message - if "Processing template" in message: - # Extract template info from message like "Processing template 1/3: Template_Name" - parts = message.split(":") - if len(parts) >= 2: - template_name = parts[1].strip() - # Extract progress numbers - if "/" in parts[0]: - progress_part = parts[0].split("Processing template")[1].strip() - if "/" in progress_part: - current, total = progress_part.split("/") - try: - current_num = int(current.strip()) - total_num = int(total.strip()) - progress_percent = (current_num / total_num) * 100 - - progress.report( - step="import", - progress=progress_percent, - message=message, - details={ - "current_template": template_name, - "total_templates": total_num, - "processed_templates": current_num - } - ) - return - except ValueError: - pass - - # Default progress step - progress.step("import", message) - - result = manager.batch_import_templates(args.source_folder, progress_callback) - # Convert TemplateInfo objects to dictionaries for JSON serialization - if 'imported_templates' in result: - result['imported_templates'] = [asdict(template) for template in result['imported_templates']] - - # Send successful result via JSON-RPC - rpc.success(result) - - elif args.action == "get_templates": - templates = manager.get_templates() - result = { - "status": True, - "templates": [asdict(template) for template in templates] - } - rpc.success(result) - - elif args.action == "get_template": - if not args.template_id: - rpc.error(JSONRPCError.INVALID_PARAMS, "Template ID is required") - return - - template = manager.get_template(args.template_id) - if template: - result = { - "status": True, - "template": asdict(template) - } - rpc.success(result) - else: - rpc.error(JSONRPCError.TEMPLATE_NOT_FOUND, f"Template not found: {args.template_id}") - - elif args.action == "delete_template": - if not args.template_id: - rpc.error(JSONRPCError.INVALID_PARAMS, "Template ID is required") - return - - success = manager.delete_template(args.template_id) - if success: - result = { - "status": True, - "msg": "Template deleted successfully" - } - rpc.success(result) - else: - rpc.error(JSONRPCError.TEMPLATE_NOT_FOUND, "Failed to delete template") - - elif args.action == "get_template_detail": - if not args.template_id: - rpc.error(JSONRPCError.INVALID_PARAMS, "Template ID is required") - return - - detail = manager.get_template_detail(args.template_id) - if detail: - rpc.success(detail) - else: - rpc.error(JSONRPCError.TEMPLATE_NOT_FOUND, f"Template detail not found: {args.template_id}") - - else: - rpc.error(JSONRPCError.METHOD_NOT_FOUND, f"Unknown action: {args.action}") - - except Exception as e: - # Send error via JSON-RPC - rpc.error(JSONRPCError.INTERNAL_ERROR, f"Template manager error: {str(e)}", str(e)) - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/python_core/storage/__init__.py b/python_core/storage/__init__.py deleted file mode 100644 index 8580345..0000000 --- a/python_core/storage/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -#!/usr/bin/env python3 -""" -存储层模块 -提供统一的存储接口，支持多种存储后端 -""" -__all__ = [] diff --git a/test_category_cli.py b/test_category_cli.py new file mode 100644 index 0000000..c0a93ed --- /dev/null +++ b/test_category_cli.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +测试资源分类管理CLI功能 +""" + +import sys +import os +sys.path.insert(0, '/root/projects/mixvideo_v2') + +from python_core.services.resource_category_manager import ResourceCategoryManager + +def test_category_manager(): + """测试分类管理器基本功能""" + print("🧪 测试资源分类管理器...") + + try: + # 创建管理器 + manager = ResourceCategoryManager() + print("✅ ResourceCategoryManager 初始化成功") + + # 创建测试分类 + result1 = manager.create_category('商业', '商业相关的视频内容', '#FF6B6B') + print(f"✅ 创建分类1: {result1['title']} (ID: {result1['id'][:8]})") + + result2 = manager.create_category('教育', '教育培训相关内容', '#4ECDC4') + print(f"✅ 创建分类2: {result2['title']} (ID: {result2['id'][:8]})") + + result3 = manager.create_category('娱乐', '娱乐休闲相关内容', '#45B7D1') + print(f"✅ 创建分类3: {result3['title']} (ID: {result3['id'][:8]})") + + # 获取所有分类 + categories = manager.get_all_categories() + print(f"✅ 获取到 {len(categories)} 个分类") + + # 搜索分类 + search_results = manager.search_categories('商业') + print(f"✅ 搜索'商业'找到 {len(search_results)} 个结果") + + # 更新分类 + updated = manager.update_category(result1['id'], title='商业推广', color='#FF5555') + if updated: + print(f"✅ 更新分类成功: {updated['title']}") + + # 获取单个分类 + category = manager.get_category_by_id(result2['id']) + if category: + print(f"✅ 获取分类详情: {category['title']}") + + print("\n📋 所有分类列表:") + for i, cat in enumerate(categories, 1): + print(f" {i}. {cat['title']} - {cat['color']} - {cat['ai_prompt'][:20]}...") + + print("\n🎉 所有测试通过！") + return True + + except Exception as e: + print(f"❌ 测试失败: {e}") + return False + +def test_cli_import(): + """测试CLI模块导入""" + print("\n🧪 测试CLI模块导入...") + + try: + from python_core.cli.commands.category import category_app + print("✅ category CLI模块导入成功") + + from python_core.cli.cli import app + print("✅ 主CLI应用导入成功") + + # 检查命令是否注册 + commands = [cmd.name for cmd in app.registered_commands.values()] + groups = [group.name for group in app.registered_groups.values()] + + print(f"✅ 注册的命令: {commands}") + print(f"✅ 注册的命令组: {groups}") + + if 'category' in groups: + print("✅ category命令组已正确注册") + else: + print("❌ category命令组未注册") + return False + + print("🎉 CLI导入测试通过！") + return True + + except Exception as e: + print(f"❌ CLI导入测试失败: {e}") + return False + +def main(): + """主测试函数""" + print("🚀 开始测试资源分类管理CLI集成...") + + # 测试分类管理器 + manager_ok = test_category_manager() + + # 测试CLI导入 + cli_ok = test_cli_import() + + if manager_ok and cli_ok: + print("\n🎉 所有测试通过！资源分类管理CLI集成成功！") + + print("\n📖 使用方法:") + print(" python3 -m python_core.cli category --help") + print(" python3 -m python_core.cli category list") + print(" python3 -m python_core.cli category create '标题' 'AI提示词' '#颜色'") + print(" python3 -m python_core.cli category search '关键词'") + print(" python3 -m python_core.cli category stats") + + else: + print("\n❌ 测试失败！") + sys.exit(1) + +if __name__ == "__main__": + main()