"""Input Parser Module""" import json import os from typing import Any, Dict, Optional, List, Union from pydantic import BaseModel # For defining the structure of parsed inputs from dataclasses import dataclass, field import logging logger = logging.getLogger("InputParser") class ParsedOpenAPISpec(BaseModel): # Placeholder for OpenAPI spec details relevant to the compliance suite spec: Dict[str, Any] info: Dict[str, Any] # Swagger 'info' object with title, version, etc. paths: Dict[str, Dict[str, Any]] # API paths and their operations tags: Optional[List[Dict[str, str]]] = None # API tags basePath: Optional[str] = None # Base path for all APIs swagger_version: str # Swagger specification version @dataclass class YAPIEndpoint: """YAPI API端点信息""" path: str method: str title: str = "" description: str = "" category_name: str = "" req_params: List[Dict[str, Any]] = field(default_factory=list) req_query: List[Dict[str, Any]] = field(default_factory=list) req_headers: List[Dict[str, Any]] = field(default_factory=list) req_body_type: str = "" req_body_other: str = "" res_body_type: str = "" res_body: str = "" @dataclass class ParsedYAPISpec: """解析后的YAPI规范""" endpoints: List[YAPIEndpoint] categories: List[Dict[str, Any]] total_count: int @dataclass class SwaggerEndpoint: """Swagger API端点信息""" path: str method: str summary: str = "" description: str = "" operation_id: str = "" tags: List[str] = field(default_factory=list) parameters: List[Dict[str, Any]] = field(default_factory=list) responses: Dict[str, Any] = field(default_factory=dict) consumes: List[str] = field(default_factory=list) produces: List[str] = field(default_factory=list) request_body: Dict[str, Any] = field(default_factory=dict) @dataclass class ParsedSwaggerSpec: """解析后的Swagger规范""" endpoints: List[SwaggerEndpoint] info: Dict[str, Any] swagger_version: str host: str = "" base_path: str = "" schemes: List[str] = field(default_factory=list) tags: List[Dict[str, Any]] = field(default_factory=list) categories: List[Dict[str, Any]] = field(default_factory=list) class ParsedBusinessLogic(BaseModel): # Placeholder for parsed business logic flow name: str steps: list # List of steps, each could be another Pydantic model class InputParser: """ Responsible for parsing DDMS supplier's input materials like API specs, etc. """ def __init__(self): pass def parse_openapi_spec(self, spec_path: str) -> Optional[ParsedOpenAPISpec]: """ Parses an OpenAPI specification from a file path. Args: spec_path: The file path of the OpenAPI specification. Returns: A ParsedOpenAPISpec object containing the parsed specification, or None if parsing fails. """ try: # Check if file exists if not os.path.exists(spec_path): print(f"Error: File not found: {spec_path}") return None # Read and parse JSON file with open(spec_path, 'r', encoding='utf-8') as f: swagger_data = json.load(f) # Extract basic information swagger_version = swagger_data.get('swagger', swagger_data.get('openapi', 'Unknown')) info = swagger_data.get('info', {}) paths = swagger_data.get('paths', {}) tags = swagger_data.get('tags', []) base_path = swagger_data.get('basePath', '') # Create and return ParsedOpenAPISpec return ParsedOpenAPISpec( spec=swagger_data, info=info, paths=paths, tags=tags, basePath=base_path, swagger_version=swagger_version ) except FileNotFoundError: print(f"File not found: {spec_path}") return None except json.JSONDecodeError as e: print(f"Error parsing JSON from {spec_path}: {e}") return None except Exception as e: print(f"Error parsing OpenAPI spec from {spec_path}: {e}") return None def parse_yapi_spec(self, file_path: str) -> Optional[ParsedYAPISpec]: """ 解析YAPI规范文件 Args: file_path: YAPI JSON文件路径 Returns: Optional[ParsedYAPISpec]: 解析后的YAPI规范，如果解析失败则返回None """ if not os.path.isfile(file_path): logger.error(f"文件不存在: {file_path}") return None try: with open(file_path, 'r', encoding='utf-8') as f: yapi_data = json.load(f) if not isinstance(yapi_data, list): logger.error(f"无效的YAPI文件格式: 顶层元素应该是数组") return None endpoints = [] categories = [] # 处理分类 for category_data in yapi_data: if not isinstance(category_data, dict): logger.warning(f"YAPI 分类条目格式不正确，应为字典类型，已跳过: {category_data}") continue category_name = category_data.get('name', '') category_desc = category_data.get('desc', '') # 添加到分类列表 categories.append({ 'name': category_name, 'desc': category_desc }) # 处理API接口 api_list = category_data.get('list', []) if not isinstance(api_list, list): logger.warning(f"分类 '{category_name}' 中的 API列表 (list) 格式不正确，应为数组类型，已跳过。") continue for api_item in api_list: if not isinstance(api_item, dict): logger.warning(f"分类 '{category_name}' 中的 API条目格式不正确，应为字典类型，已跳过: {api_item}") continue # 提取API信息 path = api_item.get('path', '') if not path: logger.info(f"分类 '{category_name}' 中的 API条目缺少 'path'，使用空字符串。 API: {api_item.get('title', '未命名')}") method = api_item.get('method', 'GET') if api_item.get('method') is None: # 仅当原始数据中完全没有 method 字段时记录 logger.info(f"分类 '{category_name}' 中的 API条目 '{path}' 缺少 'method'，使用默认值 'GET'。") title = api_item.get('title', '') if not title: logger.info(f"分类 '{category_name}' 中的 API条目 '{path}' ({method}) 缺少 'title'，使用空字符串。") description = api_item.get('desc', '') # 提取请求参数 req_params = api_item.get('req_params', []) req_query = api_item.get('req_query', []) req_headers = api_item.get('req_headers', []) # 提取请求体信息 req_body_type = api_item.get('req_body_type', '') req_body_other = api_item.get('req_body_other', '') # 提取响应体信息 res_body_type = api_item.get('res_body_type', '') res_body = api_item.get('res_body', '') # 创建端点对象 endpoint = YAPIEndpoint( path=path, method=method, title=title, description=description, category_name=category_name, req_params=req_params, req_query=req_query, req_headers=req_headers, req_body_type=req_body_type, req_body_other=req_body_other, res_body_type=res_body_type, res_body=res_body ) endpoints.append(endpoint) return ParsedYAPISpec( endpoints=endpoints, categories=categories, total_count=len(endpoints) ) except Exception as e: logger.error(f"解析YAPI文件时出错: {str(e)}") return None def parse_swagger_spec(self, file_path: str) -> Optional[ParsedSwaggerSpec]: """ 解析Swagger规范文件 Args: file_path: Swagger JSON文件路径 Returns: Optional[ParsedSwaggerSpec]: 解析后的Swagger规范，如果解析失败则返回None """ if not os.path.isfile(file_path): logger.error(f"文件不存在: {file_path}") return None try: with open(file_path, 'r', encoding='utf-8') as f: swagger_data = json.load(f) if not isinstance(swagger_data, dict): logger.error(f"无效的Swagger文件格式: 顶层元素应该是对象") return None # 提取基本信息 swagger_version = swagger_data.get('swagger', swagger_data.get('openapi', '')) info = swagger_data.get('info', {}) host = swagger_data.get('host', '') base_path = swagger_data.get('basePath', '') schemes = swagger_data.get('schemes', []) tags = swagger_data.get('tags', []) # 创建分类列表 categories = [] for tag in tags: categories.append({ 'name': tag.get('name', ''), 'desc': tag.get('description', '') }) # 处理API路径 paths = swagger_data.get('paths', {}) endpoints = [] for path, path_item in paths.items(): if not isinstance(path_item, dict): continue # 处理每个HTTP方法 (GET, POST, PUT, DELETE等) for method, operation in path_item.items(): if method in ['get', 'post', 'put', 'delete', 'patch', 'options', 'head', 'trace']: if not isinstance(operation, dict): continue # 提取操作信息 summary = operation.get('summary', '') description = operation.get('description', '') operation_id = operation.get('operationId', '') operation_tags = operation.get('tags', []) # 提取参数信息 parameters = operation.get('parameters', []) # 提取响应信息 responses = operation.get('responses', {}) # 提取请求和响应的内容类型 consumes = operation.get('consumes', swagger_data.get('consumes', [])) produces = operation.get('produces', swagger_data.get('produces', [])) # 提取请求体信息 (OpenAPI 3.0 格式) request_body = operation.get('requestBody', {}) # 创建端点对象 endpoint = SwaggerEndpoint( path=path, method=method.upper(), summary=summary, description=description, operation_id=operation_id, tags=operation_tags, parameters=parameters, responses=responses, consumes=consumes, produces=produces, request_body=request_body ) endpoints.append(endpoint) # 创建返回对象 return ParsedSwaggerSpec( endpoints=endpoints, info=info, swagger_version=swagger_version, host=host, base_path=base_path, schemes=schemes, tags=tags, categories=categories ) except Exception as e: logger.error(f"解析Swagger文件时出错: {str(e)}") return None def parse_business_logic_flow(self, flow_description: str) -> Optional[ParsedBusinessLogic]: """ Parses a business logic flow description. The format of this description is TBD and this parser would need to be built accordingly. Args: flow_description: The string content describing the business logic flow. Returns: A ParsedBusinessLogic object or None if parsing fails. """ print(f"[InputParser] Placeholder: Parsing business logic flow. Content: {flow_description[:100]}...") # Placeholder: Actual parsing logic will depend on the defined format. return ParsedBusinessLogic(name="Example Flow", steps=["Step 1 API call", "Step 2 Validate Response"]) # Add other parsers as needed (e.g., for data object definitions)