dongzi
556f5b8ab6
Some checks failed
API Parameter Change Check / api-param-check (push) Failing after 3s
115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
"""
|
||
Controller 端点解析模块。
|
||
调用 Java AST 解析器 JAR，将 Java 源码转为结构化的 API 端点列表。
|
||
"""
|
||
|
||
import json
|
||
import subprocess
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
|
||
|
||
@dataclass
|
||
class ApiParameter:
|
||
"""单个接口参数。"""
|
||
|
||
name: str
|
||
type: str
|
||
required: bool = True
|
||
source: str = "query"
|
||
description: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class ApiEndpoint:
|
||
"""单个 Controller 接口端点。"""
|
||
|
||
http_method: str
|
||
uri: str
|
||
controller_class: str
|
||
method_name: str
|
||
source_file: str
|
||
parameters: List[ApiParameter] = field(default_factory=list)
|
||
|
||
@property
|
||
def endpoint_key(self) -> str:
|
||
"""唯一标识：HTTP 方法 + URI，用于跨版本匹配。"""
|
||
return f"{self.http_method} {self.uri}"
|
||
|
||
|
||
def run_java_parser(source_dir: Path, jar_path: Path, output_json: Path) -> List[ApiEndpoint]:
|
||
"""
|
||
调用 JavaParser JAR 扫描源码目录，返回解析结果。
|
||
|
||
:param source_dir: Java 源码根目录
|
||
:param jar_path: controller-parser JAR 路径
|
||
:param output_json: 临时 JSON 输出路径
|
||
:return: ApiEndpoint 列表
|
||
:raises RuntimeError: Java 进程失败或 JAR 不存在
|
||
"""
|
||
if not jar_path.exists():
|
||
raise RuntimeError(
|
||
f"Java 解析器 JAR 不存在: {jar_path}\n"
|
||
"请先在 .gitea/java-parser 目录执行: mvn -q package"
|
||
)
|
||
|
||
cmd = ["java", "-jar", str(jar_path), str(source_dir), str(output_json)]
|
||
result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8")
|
||
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"Java 解析器执行失败:\n{result.stderr}")
|
||
|
||
with open(output_json, "r", encoding="utf-8") as f:
|
||
raw = json.load(f)
|
||
|
||
return [_dict_to_endpoint(item) for item in raw]
|
||
|
||
|
||
def _dict_to_endpoint(data: dict) -> ApiEndpoint:
|
||
"""将 JSON 字典转换为 ApiEndpoint 对象。"""
|
||
params = [
|
||
ApiParameter(
|
||
name=p.get("name", ""),
|
||
type=p.get("type", ""),
|
||
required=p.get("required", True),
|
||
source=p.get("source", "query"),
|
||
description=p.get("description"),
|
||
)
|
||
for p in data.get("parameters", [])
|
||
]
|
||
return ApiEndpoint(
|
||
http_method=data.get("httpMethod", "GET"),
|
||
uri=data.get("uri", "/"),
|
||
controller_class=data.get("controllerClass", ""),
|
||
method_name=data.get("methodName", ""),
|
||
source_file=data.get("sourceFile", ""),
|
||
parameters=params,
|
||
)
|
||
|
||
|
||
def endpoints_to_map(endpoints: List[ApiEndpoint]) -> Dict[str, ApiEndpoint]:
|
||
"""
|
||
将端点列表转为字典，key 为 endpoint_key。
|
||
|
||
:param endpoints: 端点列表
|
||
:return: { "GET /api/users/{id}": ApiEndpoint, ... }
|
||
"""
|
||
return {ep.endpoint_key: ep for ep in endpoints}
|
||
|
||
|
||
def filter_endpoints_by_files(
|
||
endpoints: List[ApiEndpoint], changed_files: List[str]
|
||
) -> List[ApiEndpoint]:
|
||
"""
|
||
仅保留源文件在变更列表中的端点（缩小对比范围）。
|
||
|
||
:param endpoints: 全部端点
|
||
:param changed_files: 变更文件相对路径列表
|
||
:return: 过滤后的端点
|
||
"""
|
||
if not changed_files:
|
||
return endpoints
|
||
changed_set = set(changed_files)
|
||
return [ep for ep in endpoints if ep.source_file in changed_set]
|