algorithm/sdk/cli/algorithm-showcase-cli.py

#!/usr/bin/env python3
"""
智能算法展示平台命令行工具
"""

import argparse
import json
import sys
from typing import Optional

from algorithm_showcase import AlgorithmShowcaseClient


def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(description='智能算法展示平台命令行工具')
    parser.add_argument('--base-url', default='http://localhost:8001/api/v1', help='API基础URL')
    parser.add_argument('--api-key', default=None, help='API密钥')
    
    subparsers = parser.add_subparsers(dest='command', help='子命令')
    
    # 算法相关命令
    algo_parser = subparsers.add_parser('algorithm', help='算法相关命令')
    algo_subparsers = algo_parser.add_subparsers(dest='algo_command', help='算法子命令')
    
    # 获取算法列表
    algo_subparsers.add_parser('list', help='获取算法列表')
    
    # 获取算法详情
    algo_detail_parser = algo_subparsers.add_parser('detail', help='获取算法详情')
    algo_detail_parser.add_argument('algorithm_id', help='算法ID')
    
    # 获取算法版本列表
    algo_versions_parser = algo_subparsers.add_parser('versions', help='获取算法版本列表')
    algo_versions_parser.add_argument('algorithm_id', help='算法ID')
    
    # 调用算法
    algo_call_parser = algo_subparsers.add_parser('call', help='调用算法')
    algo_call_parser.add_argument('algorithm_id', help='算法ID')
    algo_call_parser.add_argument('version_id', help='版本ID')
    algo_call_parser.add_argument('--input', required=True, help='输入数据JSON文件路径')
    algo_call_parser.add_argument('--params', default=None, help='算法参数JSON文件路径')
    
    # 获取调用结果
    algo_result_parser = algo_subparsers.add_parser('result', help='获取调用结果')
    algo_result_parser.add_argument('call_id', help='调用ID')
    
    # OpenAI相关命令
    openai_parser = subparsers.add_parser('openai', help='OpenAI相关命令')
    openai_subparsers = openai_parser.add_subparsers(dest='openai_command', help='OpenAI子命令')
    
    # 生成仿真数据
    openai_generate_parser = openai_subparsers.add_parser('generate', help='生成仿真输入数据')
    openai_generate_parser.add_argument('prompt', help='描述')
    openai_generate_parser.add_argument('--type', default='text', choices=['text', 'image', 'structured'], help='数据类型')
    
    # 历史记录相关命令
    history_parser = subparsers.add_parser('history', help='历史记录相关命令')
    history_subparsers = history_parser.add_subparsers(dest='history_command', help='历史记录子命令')
    
    # 获取调用历史
    history_list_parser = history_subparsers.add_parser('list', help='获取调用历史')
    history_list_parser.add_argument('--algorithm-id', help='算法ID')
    history_list_parser.add_argument('--status', help='调用状态')
    history_list_parser.add_argument('--start-date', help='开始日期')
    history_list_parser.add_argument('--end-date', help='结束日期')
    
    # 获取调用统计
    history_stats_parser = history_subparsers.add_parser('stats', help='获取调用统计')
    history_stats_parser.add_argument('--algorithm-id', help='算法ID')
    
    # 导出历史记录
    history_export_parser = history_subparsers.add_parser('export', help='导出历史记录')
    history_export_parser.add_argument('--algorithm-id', help='算法ID')
    history_export_parser.add_argument('--start-date', help='开始日期')
    history_export_parser.add_argument('--end-date', help='结束日期')
    history_export_parser.add_argument('--format', default='json', help='导出格式')
    
    # 监控相关命令
    monitor_parser = subparsers.add_parser('monitor', help='监控相关命令')
    monitor_subparsers = monitor_parser.add_subparsers(dest='monitor_command', help='监控子命令')
    
    # 获取系统健康状况
    monitor_subparsers.add_parser('health', help='获取系统健康状况')
    
    # 获取算法性能指标
    perf_parser = monitor_subparsers.add_parser('performance', help='获取算法性能指标')
    perf_parser.add_argument('algorithm_id', help='算法ID')
    perf_parser.add_argument('--days', type=int, default=7, help='统计天数')
    
    # 权限相关命令
    perm_parser = subparsers.add_parser('permission', help='权限相关命令')
    perm_parser.add_argument('algorithm_id', help='算法ID')
    perm_parser.add_argument('--type', default='execute', choices=['read', 'execute', 'write', 'admin'], help='权限类型')
    
    return parser.parse_args()


def load_json_file(file_path: str) -> dict:
    """加载JSON文件"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        print(f"加载文件失败: {e}")
        sys.exit(1)


def main():
    """主函数"""
    args = parse_args()
    
    # 初始化客户端
    client = AlgorithmShowcaseClient(
        base_url=args.base_url,
        api_key=args.api_key
    )
    
    # 处理命令
    if args.command == 'algorithm':
        if args.algo_command == 'list':
            # 获取算法列表
            algorithms = client.get_algorithms()
            print(json.dumps(algorithms, ensure_ascii=False, indent=2))
        
        elif args.algo_command == 'detail':
            # 获取算法详情
            algorithm = client.get_algorithm(args.algorithm_id)
            print(json.dumps(algorithm.__dict__, ensure_ascii=False, indent=2))
        
        elif args.algo_command == 'versions':
            # 获取算法版本列表
            versions = client.get_algorithm_versions(args.algorithm_id)
            versions_dict = [v.__dict__ for v in versions]
            print(json.dumps(versions_dict, ensure_ascii=False, indent=2))
        
        elif args.algo_command == 'call':
            # 加载输入数据
            input_data = load_json_file(args.input)
            
            # 加载算法参数
            params = {}
            if args.params:
                params = load_json_file(args.params)
            
            # 调用算法
            result = client.call_algorithm(
                algorithm_id=args.algorithm_id,
                version_id=args.version_id,
                input_data=input_data,
                params=params
            )
            print(json.dumps(result.__dict__, ensure_ascii=False, indent=2))
        
        elif args.algo_command == 'result':
            # 获取调用结果
            result = client.get_call_result(args.call_id)
            print(json.dumps(result.__dict__, ensure_ascii=False, indent=2))
    
    elif args.command == 'openai':
        if args.openai_command == 'generate':
            # 生成仿真数据
            result = client.generate_simulation_data(
                prompt=args.prompt,
                data_type=args.type
            )
            print(json.dumps(result, ensure_ascii=False, indent=2))
    
    elif args.command == 'history':
        if args.history_command == 'list':
            # 获取调用历史
            history = client.get_call_history(
                algorithm_id=args.algorithm_id,
                status=args.status,
                start_date=args.start_date,
                end_date=args.end_date
            )
            print(json.dumps(history, ensure_ascii=False, indent=2))
        
        elif args.history_command == 'stats':
            # 获取调用统计
            stats = client.get_call_statistics(
                algorithm_id=args.algorithm_id
            )
            print(json.dumps(stats, ensure_ascii=False, indent=2))
        
        elif args.history_command == 'export':
            # 导出历史记录
            result = client.export_history(
                algorithm_id=args.algorithm_id,
                start_date=args.start_date,
                end_date=args.end_date,
                format_type=args.format
            )
            print(json.dumps(result, ensure_ascii=False, indent=2))
    
    elif args.command == 'monitor':
        if args.monitor_command == 'health':
            # 获取系统健康状况
            health = client.get_system_health()
            print(json.dumps(health, ensure_ascii=False, indent=2))
        
        elif args.monitor_command == 'performance':
            # 获取算法性能指标
            performance = client.get_algorithm_performance(
                algorithm_id=args.algorithm_id,
                days=args.days
            )
            print(json.dumps(performance, ensure_ascii=False, indent=2))
    
    elif args.command == 'permission':
        # 检查权限
        has_permission = client.check_permission(
            algorithm_id=args.algorithm_id,
            permission_type=args.type
        )
        print(json.dumps({'has_permission': has_permission}, ensure_ascii=False, indent=2))
    
    else:
        print('请指定命令')
        sys.exit(1)


if __name__ == '__main__':
    main()