""" API 参数变更对比模块。 对比新旧两个版本的 Controller 端点，识别参数的增、删、改、重命名。 """ from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, Set, Tuple from controller_parser import ApiEndpoint, ApiParameter class ChangeType(str, Enum): """参数变更类型。""" ADDED = "added" REMOVED = "removed" MODIFIED = "modified" RENAMED = "renamed" @dataclass class ParameterChange: """单条参数变更记录。""" change_type: ChangeType param_name: str param_type: Optional[str] = None old_name: Optional[str] = None old_type: Optional[str] = None required: Optional[bool] = None old_required: Optional[bool] = None detail: Optional[str] = None def to_display_line(self) -> str: """ 格式化为通知模板中的一行文本。 :return: 如 "删除: Boolean userType" 或 "重命名: String userName -> String accountName" """ if self.change_type == ChangeType.REMOVED: return f" - 删除: {self.param_type} {self.param_name}" if self.change_type == ChangeType.ADDED: req_text = f" (是否必填：{str(self.required).lower()})" if self.required is not None else "" return f" - 新增: {self.param_type} {self.param_name}{req_text}" if self.change_type == ChangeType.RENAMED: return f" - 重命名: {self.old_type} {self.old_name} -> {self.param_type} {self.param_name}" if self.change_type == ChangeType.MODIFIED: parts = [f" - 修改: {self.param_name}"] if self.detail: parts.append(f" ({self.detail})") return "".join(parts) return f" - {self.change_type.value}: {self.param_name}" @dataclass class EndpointChangeReport: """单个接口的变更报告。""" uri: str http_method: str controller_class: str method_name: str parameter_changes: List[ParameterChange] = field(default_factory=list) is_new_endpoint: bool = False is_removed_endpoint: bool = False @property def has_changes(self) -> bool: """是否存在任何变更。""" return ( self.is_new_endpoint or self.is_removed_endpoint or len(self.parameter_changes) > 0 ) @property def endpoint_key(self) -> str: return f"{self.http_method} {self.uri}" def _param_key(p: ApiParameter) -> Tuple[str, str]: """参数匹配键：(source, name)。""" return (p.source, p.name) def compare_parameters( old_params: List[ApiParameter], new_params: List[ApiParameter] ) -> List[ParameterChange]: """ 对比同一接口新旧版本的参数列表，识别增删改及重命名。 重命名启发式：若删除与新增参数类型相同且 source 相同，则视为重命名。 :param old_params: 旧版本参数 :param new_params: 新版本参数 :return: 变更列表 """ changes: List[ParameterChange] = [] old_map: Dict[Tuple[str, str], ApiParameter] = {_param_key(p): p for p in old_params} new_map: Dict[Tuple[str, str], ApiParameter] = {_param_key(p): p for p in new_params} old_keys = set(old_map.keys()) new_keys = set(new_map.keys()) removed_keys = old_keys - new_keys added_keys = new_keys - old_keys common_keys = old_keys & new_keys # 1. 共同参数：检查类型、必填等属性变更 for key in common_keys: old_p = old_map[key] new_p = new_map[key] detail_parts = [] if old_p.type != new_p.type: detail_parts.append(f"类型 {old_p.type} -> {new_p.type}") if old_p.required != new_p.required: detail_parts.append(f"必填 {old_p.required} -> {new_p.required}") if detail_parts: changes.append( ParameterChange( change_type=ChangeType.MODIFIED, param_name=new_p.name, param_type=new_p.type, required=new_p.required, old_required=old_p.required, detail=", ".join(detail_parts), ) ) # 2. 重命名检测：在 removed + added 中找同 type + source 的配对 unmatched_removed: List[Tuple[Tuple[str, str], ApiParameter]] = [] unmatched_added: List[Tuple[Tuple[str, str], ApiParameter]] = [] for key in removed_keys: unmatched_removed.append((key, old_map[key])) for key in added_keys: unmatched_added.append((key, new_map[key])) matched_removed: Set[Tuple[str, str]] = set() matched_added: Set[Tuple[str, str]] = set() for r_key, r_param in unmatched_removed: for a_key, a_param in unmatched_added: if a_key in matched_added: continue if r_param.type == a_param.type and r_param.source == a_param.source: changes.append( ParameterChange( change_type=ChangeType.RENAMED, param_name=a_param.name, param_type=a_param.type, old_name=r_param.name, old_type=r_param.type, required=a_param.required, ) ) matched_removed.add(r_key) matched_added.add(a_key) break # 3. 纯删除 for key, param in unmatched_removed: if key not in matched_removed: changes.append( ParameterChange( change_type=ChangeType.REMOVED, param_name=param.name, param_type=param.type, ) ) # 4. 纯新增 for key, param in unmatched_added: if key not in matched_added: changes.append( ParameterChange( change_type=ChangeType.ADDED, param_name=param.name, param_type=param.type, required=param.required, ) ) return changes def compare_endpoints( old_endpoints: Dict[str, ApiEndpoint], new_endpoints: Dict[str, ApiEndpoint], ) -> List[EndpointChangeReport]: """ 对比新旧两个版本的全部 Controller 端点，生成变更报告列表。 :param old_endpoints: 旧版本 { endpoint_key: ApiEndpoint } :param new_endpoints: 新版本 { endpoint_key: ApiEndpoint } :return: 有变更的接口报告列表 """ reports: List[EndpointChangeReport] = [] all_keys = set(old_endpoints.keys()) | set(new_endpoints.keys()) for key in sorted(all_keys): old_ep = old_endpoints.get(key) new_ep = new_endpoints.get(key) if old_ep is None and new_ep is not None: # 全新接口 reports.append( EndpointChangeReport( uri=new_ep.uri, http_method=new_ep.http_method, controller_class=new_ep.controller_class, method_name=new_ep.method_name, is_new_endpoint=True, parameter_changes=[ ParameterChange( change_type=ChangeType.ADDED, param_name=p.name, param_type=p.type, required=p.required, ) for p in new_ep.parameters ], ) ) continue if new_ep is None and old_ep is not None: # 接口被删除 reports.append( EndpointChangeReport( uri=old_ep.uri, http_method=old_ep.http_method, controller_class=old_ep.controller_class, method_name=old_ep.method_name, is_removed_endpoint=True, ) ) continue # 同 URI 对比参数 param_changes = compare_parameters(old_ep.parameters, new_ep.parameters) if param_changes: reports.append( EndpointChangeReport( uri=new_ep.uri, http_method=new_ep.http_method, controller_class=new_ep.controller_class, method_name=new_ep.method_name, parameter_changes=param_changes, ) ) return [r for r in reports if r.has_changes]