257 lines
8.5 KiB
Python
257 lines
8.5 KiB
Python
"""
|
|
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]
|