mxivideo/python_core/cli/commands/model.py

"""
模特管理 CLI 命令
"""

from pathlib import Path
from typing import Optional, List
from dataclasses import asdict
from datetime import datetime
import typer
from python_core.utils.jsonrpc_enhanced import create_response_handler, create_progress_reporter
from python_core.database.model_postgres import model_table
from python_core.utils.logger import logger
from uuid import uuid4

# 创建模特应用
model_app = typer.Typer(help="模特管理命令")


@model_app.command("create")
def create_model(
    model_number: str = typer.Argument(..., help="模特编号"),
    model_image: str = typer.Argument(..., help="模特图片路径"),
    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
    is_cloud: bool = typer.Option(False, "--cloud", help="是否为云端模特"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """创建模特"""
    response = create_response_handler()
    try:
        # 创建模特
        model = model_table.create_model(
            model_number=model_number,
            model_image=model_image,
            user_id=user_id or "default",
            is_cloud=is_cloud
        )
        
        if model:
            response.success({
                'model': asdict(model),
                'message': f'模特创建成功: {model.model_number}'
            })
        else:
            response.error(-32603, "创建模特失败")
            
    except ValueError as e:
        response.error(-32602, str(e))
    except Exception as e:
        logger.error(f"创建模特失败: {e}")
        response.error(-32603, f"创建模特失败: {str(e)}")


@model_app.command("list")
def list_models(
    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
    include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模特"),
    include_inactive: bool = typer.Option(False, "--include-inactive", help="包含已禁用模特"),
    limit: int = typer.Option(100, "--limit", "-l", help="显示数量限制"),
    offset: int = typer.Option(0, "--offset", help="偏移量"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """获取模特列表"""
    response = create_response_handler()
    try:
        # 获取模特列表
        models = model_table.get_all_models(
            user_id=user_id or "default",
            include_cloud=include_cloud,
            include_inactive=include_inactive,
            limit=limit,
            offset=offset
        )
        
        # 获取总数
        total_count = model_table.get_model_count(
            user_id=user_id or "default",
            include_cloud=include_cloud,
            include_inactive=include_inactive
        )
        
        response.success({
            'models': [asdict(model) for model in models],
            'total_count': total_count,
            'limit': limit,
            'offset': offset,
            'message': f'获取到 {len(models)} 个模特'
        })
        
    except Exception as e:
        logger.error(f"获取模特列表失败: {e}")
        response.error(-32603, f"获取模特列表失败: {str(e)}")


@model_app.command("get")
def get_model(
    model_id: str = typer.Argument(..., help="模特ID"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """获取模特详情"""
    response = create_response_handler()
    try:
        model = model_table.get_model_by_id(model_id)
        
        if model:
            response.success({
                'model': asdict(model),
                'message': f'获取模特成功: {model.model_number}'
            })
        else:
            response.error(-32604, f"模特不存在: {model_id}")
            
    except Exception as e:
        logger.error(f"获取模特失败: {e}")
        response.error(-32603, f"获取模特失败: {str(e)}")


@model_app.command("get-by-number")
def get_model_by_number(
    model_number: str = typer.Argument(..., help="模特编号"),
    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()
    try:
        model = model_table.get_model_by_number(
            model_number=model_number,
            user_id=user_id or "default"
        )
        
        if model:
            response.success({
                'model': asdict(model),
                'message': f'获取模特成功: {model.model_number}'
            })
        else:
            response.error(-32604, f"模特不存在: {model_number}")
            
    except Exception as e:
        logger.error(f"获取模特失败: {e}")
        response.error(-32603, f"获取模特失败: {str(e)}")


@model_app.command("update")
def update_model(
    model_id: str = typer.Argument(..., help="模特ID"),
    model_number: Optional[str] = typer.Option(None, "--model-number", help="模特编号"),
    model_image: Optional[str] = typer.Option(None, "--model-image", help="模特图片路径"),
    is_active: Optional[bool] = typer.Option(None, "--active", help="是否激活"),
    is_cloud: Optional[bool] = typer.Option(None, "--cloud", help="是否为云端模特"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """更新模特"""
    response = create_response_handler()
    try:
        # 构建更新数据
        updates = {}
        if model_number is not None:
            updates['model_number'] = model_number
        if model_image is not None:
            updates['model_image'] = model_image
        if is_active is not None:
            updates['is_active'] = is_active
        if is_cloud is not None:
            updates['is_cloud'] = is_cloud
        
        if not updates:
            response.error(-32602, "没有提供更新字段")
            return
        
        success = model_table.update_model(model_id, updates)
        
        if success:
            response.success({
                'model_id': model_id,
                'updates': updates,
                'message': '模特更新成功'
            })
        else:
            response.error(-32604, f"模特不存在: {model_id}")
            
    except ValueError as e:
        response.error(-32602, str(e))
    except Exception as e:
        logger.error(f"更新模特失败: {e}")
        response.error(-32603, f"更新模特失败: {str(e)}")


@model_app.command("delete")
def delete_model(
    model_id: str = typer.Argument(..., help="模特ID"),
    hard_delete: bool = typer.Option(False, "--hard", help="硬删除（永久删除）"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """删除模特"""
    response = create_response_handler()
    try:
        success = model_table.delete_model(model_id, hard_delete=hard_delete)
        
        if success:
            action = "删除" if hard_delete else "禁用"
            response.success({
                'model_id': model_id,
                'hard_delete': hard_delete,
                'message': f'模特{action}成功'
            })
        else:
            response.error(-32604, f"模特不存在: {model_id}")
            
    except Exception as e:
        logger.error(f"删除模特失败: {e}")
        response.error(-32603, f"删除模特失败: {str(e)}")


@model_app.command("search")
def search_models(
    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()
    try:
        models = model_table.search_models(
            query=query,
            user_id=user_id or "default",
            include_cloud=include_cloud,
            limit=limit
        )
        
        response.success({
            'models': [asdict(model) for model in models],
            'query': query,
            'count': len(models),
            'message': f'搜索到 {len(models)} 个模特'
        })
        
    except Exception as e:
        logger.error(f"搜索模特失败: {e}")
        response.error(-32603, f"搜索模特失败: {str(e)}")


@model_app.command("toggle")
def toggle_model_status(
    model_id: str = typer.Argument(..., help="模特ID"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """切换模特状态"""
    response = create_response_handler()
    try:
        success = model_table.toggle_model_status(model_id)
        
        if success:
            response.success({
                'model_id': model_id,
                'message': '模特状态切换成功'
            })
        else:
            response.error(-32604, f"模特不存在: {model_id}")
            
    except Exception as e:
        logger.error(f"切换模特状态失败: {e}")
        response.error(-32603, f"切换模特状态失败: {str(e)}")


@model_app.command("count")
def get_model_count(
    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
    include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模特"),
    include_inactive: bool = typer.Option(False, "--include-inactive", help="包含已禁用模特"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
):
    """获取模特数量"""
    response = create_response_handler()
    try:
        count = model_table.get_model_count(
            user_id=user_id or "default",
            include_cloud=include_cloud,
            include_inactive=include_inactive
        )
        
        response.success({
            'count': count,
            'user_id': user_id or "default",
            'include_cloud': include_cloud,
            'include_inactive': include_inactive,
            'message': f'共有 {count} 个模特'
        })
        
    except Exception as e:
        logger.error(f"获取模特数量失败: {e}")
        response.error(-32603, f"获取模特数量失败: {str(e)}")


if __name__ == "__main__":
    model_app()