云端粗出
This commit is contained in:
root
@@ -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():
|
||||
|
||||
491
python_core/cli/commands/category.py
Normal file
491
python_core/cli/commands/category.py
Normal file
@@ -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()
|
||||
0
python_core/database/__init__.py
Normal file
0
python_core/database/__init__.py
Normal file
10
python_core/database/db.py
Normal file
10
python_core/database/db.py
Normal file
@@ -0,0 +1,10 @@
|
||||
|
||||
from python_core.kv import kv
|
||||
|
||||
class Db:
|
||||
|
||||
def __init__(self, key: str):
|
||||
self.kv = kv
|
||||
self.key = key
|
||||
pass
|
||||
|
||||
13
python_core/database/template.py
Normal file
13
python_core/database/template.py
Normal file
@@ -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
|
||||
@@ -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
|
||||
raise e
|
||||
|
||||
|
||||
kv = Kv()
|
||||
@@ -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()
|
||||
|
||||
@@ -1,6 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
存储层模块
|
||||
提供统一的存储接口，支持多种存储后端
|
||||
"""
|
||||
__all__ = []
|
||||
116
test_category_cli.py
Normal file
116
test_category_cli.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user