""" 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 description: Optional[str] = None old_description: Optional[str] = None source: str = "query" def _change_tag(self) -> str: """变更类型标签(企微颜色)。""" tags = { ChangeType.ADDED: '**新增**', ChangeType.REMOVED: '**删除**', ChangeType.RENAMED: '**重命名**', ChangeType.MODIFIED: '**修改**', } return tags.get(self.change_type, "") def _required_tag(self) -> str: """必填/可选标签。""" if self.required is True: return '必填' if self.required is False: return '可选' return "" def to_markdown_block(self, index: int = 1) -> str: """格式化为企微友好的参数变更卡片(列表式,非表格)。""" lines: List[str] = [] desc = self.description or self.old_description if self.change_type == ChangeType.RENAMED: lines.append(f"**{index}. `{self.param_name}`** {self._change_tag()}") lines.append(f"> `{self.old_name}` → `{self.param_name}`") if desc: lines.append(f"> 说明:{desc}") return "\n".join(lines) if self.change_type == ChangeType.ADDED: type_part = f" · `{self.param_type}`" if self.param_type else "" req_part = f" · {self._required_tag()}" if self._required_tag() else "" lines.append( f"**{index}. `{self.param_name}`**{type_part}{req_part} {self._change_tag()}" ) if desc: lines.append(f"> 说明:{desc}") return "\n".join(lines) if self.change_type == ChangeType.REMOVED: type_part = f" · `{self.param_type}`" if self.param_type else "" lines.append( f"**{index}. `{self.param_name}`**{type_part} {self._change_tag()}" ) if desc: lines.append(f"> 说明:{desc}") return "\n".join(lines) # MODIFIED lines.append(f"**{index}. `{self.param_name}`** {self._change_tag()}") if desc: lines.append(f"> 说明:{desc}") if self.detail: lines.append(f"> 变更:{self.detail}") return "\n".join(lines) def to_table_row(self) -> str: """兼容旧调用,委托至卡片块。""" return self.to_markdown_block(1) @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 _format_type_change(old_type: str, new_type: str) -> str: """类型变更文案。""" return f"类型由{old_type}改为{new_type}" 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(_format_type_change(old_p.type, new_p.type)) if old_p.required != new_p.required: req_label = lambda r: "必填" if r else "可选" detail_parts.append( f"必填性由{req_label(old_p.required)}改为{req_label(new_p.required)}" ) if old_p.description != new_p.description: detail_parts.append( f"说明由{old_p.description or '-'}改为{new_p.description or '-'}" ) 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), description=new_p.description, old_description=old_p.description, source=new_p.source, ) ) # 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, description=a_param.description, old_description=r_param.description, source=a_param.source, ) ) 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, description=param.description, source=param.source, ) ) # 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, description=param.description, source=param.source, ) ) 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 不同) # 放宽匹配条件:只要同一个 Controller 的同一个 URI 请求方式改变,就识别为「修改请求方式」 # 不再要求 method_name 相同(允许方法重命名场景) # 如果同时有参数变更,生成两条独立报告(方法变更 + 参数变更),互不干扰 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.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, ) ) # 再检测参数变更,如果有则额外生成参数报告 param_changes = compare_parameters(r_ep.parameters, a_ep.parameters) if param_changes: 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, parameter_changes=param_changes, ) ) matched_removed.add(r_key) matched_added.add(a_key) break # 2. URI 路径变更检测(method + controller + method_name 相同,但 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 ): # 先生成纯路径变更报告 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, ) ) # 再检测参数变更,如果有则额外生成参数报告 param_changes = compare_parameters(r_ep.parameters, a_ep.parameters) if param_changes: 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, 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, description=p.description, source=p.source, ) 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]