""" API 参数变更对比模块。 对比新旧两个版本的 Controller 端点，识别参数的增、删、改、重命名。 """ from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, Set, Tuple from models 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]