""" 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_markdown_line(self, *, plain: bool = False) -> str: """ 格式化为企微 Markdown 行。 plain=True 时用于新增接口,直接列出参数,不加「新增」前缀。 """ req_optional = self.required is False req_required = self.required is True if plain and self.change_type == ChangeType.ADDED: tag = ( '必填' if req_required else '可选' ) return f'> `{self.param_type}` **{self.param_name}** · {tag}' if self.change_type == ChangeType.REMOVED: return ( f'【删除】 ' f'`{self.param_type}` ~~{self.param_name}~~' ) if self.change_type == ChangeType.ADDED: tag = ( '必填' if req_required else '可选' ) return ( f'【新增】 ' f'`{self.param_type}` **{self.param_name}** · {tag}' ) if self.change_type == ChangeType.RENAMED: return ( f'【重命名】 ' f'`{self.old_type}` {self.old_name} → ' f'`{self.param_type}` **{self.param_name}**' ) if self.change_type == ChangeType.MODIFIED: detail = f' · {self.detail}' if self.detail else "" return f'【修改】 **{self.param_name}**{detail}' return f'- {self.param_name}' @dataclass class EndpointChangeReport: """单个接口的变更报告。""" uri: str http_method: str controller_class: str method_name: str source_file: str = "" parameter_changes: List[ParameterChange] = field(default_factory=list) is_new_endpoint: bool = False is_removed_endpoint: bool = False is_renamed_endpoint: bool = False old_uri: Optional[str] = None is_method_changed: bool = False old_http_method: Optional[str] = None @property def has_changes(self) -> bool: """是否存在任何变更。""" return ( self.is_new_endpoint or self.is_removed_endpoint or self.is_renamed_endpoint or self.is_method_changed 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 端点,生成变更报告列表。 支持以下变更类型检测: - HTTP 方法变更(GET → POST 等) - URI 路径变更(路径重命名) - 新增 / 删除接口 - 参数变更 """ reports: List[EndpointChangeReport] = [] old_keys = set(old_endpoints.keys()) new_keys = set(new_endpoints.keys()) removed_keys = old_keys - new_keys added_keys = new_keys - old_keys common_keys = old_keys & new_keys # 收集未匹配的 removed / added unmatched_removed: List[Tuple[str, ApiEndpoint]] = [] unmatched_added: List[Tuple[str, ApiEndpoint]] = [] for key in removed_keys: unmatched_removed.append((key, old_endpoints[key])) for key in added_keys: unmatched_added.append((key, new_endpoints[key])) matched_removed: Set[str] = set() matched_added: Set[str] = set() # 1. HTTP 方法变更检测(uri + controller + method_name 相同,但 method 不同) for r_key, r_ep in unmatched_removed: for a_key, a_ep in unmatched_added: if a_key in matched_added: continue if ( r_ep.uri == a_ep.uri and r_ep.controller_class == a_ep.controller_class and r_ep.method_name == a_ep.method_name and r_ep.http_method != a_ep.http_method ): reports.append( EndpointChangeReport( uri=a_ep.uri, http_method=a_ep.http_method, controller_class=a_ep.controller_class, method_name=a_ep.method_name, source_file=a_ep.source_file, is_method_changed=True, old_http_method=r_ep.http_method, ) ) matched_removed.add(r_key) matched_added.add(a_key) break # 2. URI 路径变更检测(method + controller + method_name 相同,但 uri 不同) # 同时检测参数变更,即使 URI 改变也记录参数差异 for r_key, r_ep in unmatched_removed: if r_key in matched_removed: continue for a_key, a_ep in unmatched_added: if a_key in matched_added: continue if ( r_ep.http_method == a_ep.http_method and r_ep.controller_class == a_ep.controller_class and r_ep.method_name == a_ep.method_name and r_ep.uri != a_ep.uri ): param_changes = compare_parameters(r_ep.parameters, a_ep.parameters) reports.append( EndpointChangeReport( uri=a_ep.uri, http_method=a_ep.http_method, controller_class=a_ep.controller_class, method_name=a_ep.method_name, source_file=a_ep.source_file, is_renamed_endpoint=True, old_uri=r_ep.uri, parameter_changes=param_changes, ) ) matched_removed.add(r_key) matched_added.add(a_key) break # 3. 剩余未匹配的 removed → 删除接口 for key, ep in unmatched_removed: if key not in matched_removed: reports.append( EndpointChangeReport( uri=ep.uri, http_method=ep.http_method, controller_class=ep.controller_class, method_name=ep.method_name, source_file=ep.source_file, is_removed_endpoint=True, ) ) # 4. 剩余未匹配的 added → 新增接口 for key, ep in unmatched_added: if key not in matched_added: reports.append( EndpointChangeReport( uri=ep.uri, http_method=ep.http_method, controller_class=ep.controller_class, method_name=ep.method_name, source_file=ep.source_file, 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 ep.parameters ], ) ) # 5. 共同 URI:对比参数变更 for key in common_keys: old_ep = old_endpoints[key] new_ep = new_endpoints[key] 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, source_file=new_ep.source_file, parameter_changes=param_changes, ) ) return [r for r in reports if r.has_changes]