#!/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:8000/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()