388 lines
15 KiB
Python
388 lines
15 KiB
Python
|
"""
|
|||
|
|
接口描述解析器
|
|||
|
|
══════════════════════════════════════════════════════════════
|
|||
|
|
支持格式（当前版）：
|
|||
|
|
顶层字段：
|
|||
|
|
- project : 项目名称，用于命名输出目录
|
|||
|
|
- description : 项目描述
|
|||
|
|
- units : 接口/函数描述列表
|
|||
|
|
|
|||
|
|
接口公共字段：
|
|||
|
|
- name : 接口/函数名称
|
|||
|
|
- requirement_id : 需求编号
|
|||
|
|
- description : 接口描述
|
|||
|
|
- type : "function" | "http"
|
|||
|
|
- url : function → Python 源文件路径，如 "create_user.py"
|
|||
|
|
http → base URL，如 "http://127.0.0.1/api"
|
|||
|
|
- parameters : 统一参数字段（HTTP 与 function 两种协议共用）
|
|||
|
|
⚠️ HTTP 接口不再使用 url_parameters / request_parameters
|
|||
|
|
- return : 返回值描述（含 on_success / on_failure）
|
|||
|
|
|
|||
|
|
HTTP 专用字段：
|
|||
|
|
- method : "get" | "post" | "put" | "delete" 等
|
|||
|
|
- 完整请求 URL = url.rstrip("/") + "/" + name.lstrip("/")
|
|||
|
|
|
|||
|
|
Function 专用：
|
|||
|
|
- url 为 .py 文件路径，转换为 Python 模块路径供 import 使用
|
|||
|
|
- 函数名即 name 字段
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
import json
|
|||
|
|
import re
|
|||
|
|
from typing import Any
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
# 基础数据结构
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class ParameterInfo:
|
|||
|
|
name: str
|
|||
|
|
type: str # 支持复合类型 "string|integer"
|
|||
|
|
description: str
|
|||
|
|
required: bool = True
|
|||
|
|
default: Any = None
|
|||
|
|
inout: str = "in" # "in" | "out" | "inout"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class ReturnCase:
|
|||
|
|
"""on_success 或 on_failure 的返回描述"""
|
|||
|
|
value: Any = None
|
|||
|
|
description: str = ""
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def is_exception(self) -> bool:
|
|||
|
|
"""判断是否为抛出异常的场景"""
|
|||
|
|
return (
|
|||
|
|
isinstance(self.value, str)
|
|||
|
|
and "exception" in self.value.lower()
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def value_fields(self) -> dict[str, Any]:
|
|||
|
|
"""若 value 是 dict，返回其字段；否则返回空 dict"""
|
|||
|
|
return self.value if isinstance(self.value, dict) else {}
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def value_summary(self) -> str:
|
|||
|
|
"""返回值的简短文字描述，用于 LLM 提示词"""
|
|||
|
|
if isinstance(self.value, dict):
|
|||
|
|
return json.dumps(self.value, ensure_ascii=False)
|
|||
|
|
if isinstance(self.value, bool):
|
|||
|
|
return str(self.value).lower()
|
|||
|
|
return str(self.value) if self.value is not None else ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class ReturnInfo:
|
|||
|
|
type: str = ""
|
|||
|
|
description: str = ""
|
|||
|
|
on_success: ReturnCase = field(default_factory=ReturnCase)
|
|||
|
|
on_failure: ReturnCase = field(default_factory=ReturnCase)
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def all_fields(self) -> set[str]:
|
|||
|
|
"""汇总 on_success + on_failure 中出现的所有字段名"""
|
|||
|
|
fields: set[str] = set()
|
|||
|
|
for case in (self.on_success, self.on_failure):
|
|||
|
|
fields.update(case.value_fields.keys())
|
|||
|
|
return fields
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
# 接口描述
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class InterfaceInfo:
|
|||
|
|
name: str
|
|||
|
|
description: str
|
|||
|
|
protocol: str # "http" | "function"
|
|||
|
|
url: str = "" # 原始 url 字段
|
|||
|
|
requirement_id: str = ""
|
|||
|
|
|
|||
|
|
# HTTP 专用
|
|||
|
|
method: str = "get"
|
|||
|
|
|
|||
|
|
# 统一参数列表（HTTP 与 function 两种协议共用，均来自 "parameters" 字段）
|
|||
|
|
parameters: list[ParameterInfo] = field(default_factory=list)
|
|||
|
|
|
|||
|
|
# 返回值描述
|
|||
|
|
return_info: ReturnInfo = field(default_factory=ReturnInfo)
|
|||
|
|
|
|||
|
|
# ── 便捷属性：参数分组 ────────────────────────────────────
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def in_parameters(self) -> list[ParameterInfo]:
|
|||
|
|
"""inout = "in" 或 "inout" 的参数（作为输入）"""
|
|||
|
|
return [p for p in self.parameters if p.inout in ("in", "inout")]
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def out_parameters(self) -> list[ParameterInfo]:
|
|||
|
|
"""inout = "out" 或 "inout" 的参数（作为输出）"""
|
|||
|
|
return [p for p in self.parameters if p.inout in ("out", "inout")]
|
|||
|
|
|
|||
|
|
# ── 便捷属性：HTTP ────────────────────────────────────────
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def http_full_url(self) -> str:
|
|||
|
|
"""
|
|||
|
|
HTTP 协议的完整请求 URL：
|
|||
|
|
url (base) + name (path)
|
|||
|
|
例：url="http://127.0.0.1/api", name="/delete_user"
|
|||
|
|
→ "http://127.0.0.1/api/delete_user"
|
|||
|
|
"""
|
|||
|
|
if self.protocol != "http":
|
|||
|
|
return ""
|
|||
|
|
base = self.url.rstrip("/")
|
|||
|
|
path = self.name.lstrip("/")
|
|||
|
|
return f"{base}/{path}" if base else f"/{path}"
|
|||
|
|
|
|||
|
|
# ── 便捷属性：Function ────────────────────────────────────
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def module_path(self) -> str:
|
|||
|
|
"""
|
|||
|
|
将 url（.py 文件路径）转换为 Python 模块导入路径。
|
|||
|
|
规则：
|
|||
|
|
1. 去除 .py 后缀
|
|||
|
|
2. 去除前导 ./ 或 /
|
|||
|
|
3. 将路径分隔符替换为 .
|
|||
|
|
示例：
|
|||
|
|
"create_user.py" → "create_user"
|
|||
|
|
"myapp/services/user_service.py" → "myapp.services.user_service"
|
|||
|
|
"./services/user_service.py" → "services.user_service"
|
|||
|
|
"myapp.services.user_service" → "myapp.services.user_service"
|
|||
|
|
"""
|
|||
|
|
if self.protocol != "function" or not self.url:
|
|||
|
|
return ""
|
|||
|
|
u = self.url.strip()
|
|||
|
|
# 已经是点分模块路径（无路径分隔符且无 .py 后缀）
|
|||
|
|
if re.match(r'^[\w]+(\.[\w]+)*$', u) and not u.endswith(".py"):
|
|||
|
|
return u
|
|||
|
|
# 文件路径 → 模块路径
|
|||
|
|
u = u.replace("\\", "/")
|
|||
|
|
u = re.sub(r'\.py$', '', u)
|
|||
|
|
u = re.sub(r'^\.?/', '', u)
|
|||
|
|
return u.replace("/", ".")
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def source_file(self) -> str:
|
|||
|
|
"""function 协议的源文件路径（原始 url 字段）"""
|
|||
|
|
return self.url if self.protocol == "function" else ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
# 描述文件根结构
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class ApiDescriptor:
|
|||
|
|
project: str
|
|||
|
|
description: str
|
|||
|
|
interfaces: list[InterfaceInfo]
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
# 解析器
|
|||
|
|
# ══════════════════════════════════════════════════════════════
|
|||
|
|
|
|||
|
|
class InterfaceParser:
|
|||
|
|
|
|||
|
|
# ── 公开接口 ──────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def parse_file(self, file_path: str) -> ApiDescriptor:
|
|||
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|||
|
|
data = json.load(f)
|
|||
|
|
return self.parse(data)
|
|||
|
|
|
|||
|
|
def parse(self, data: dict | list) -> ApiDescriptor:
|
|||
|
|
"""
|
|||
|
|
兼容三种格式：
|
|||
|
|
新格式 ：{"project": "...", "description": "...", "units": [...]}
|
|||
|
|
旧格式1：{"project": "...", "units": [...]}
|
|||
|
|
旧格式2：[...] 直接是接口数组
|
|||
|
|
"""
|
|||
|
|
if isinstance(data, list):
|
|||
|
|
return ApiDescriptor(
|
|||
|
|
project="default",
|
|||
|
|
description="",
|
|||
|
|
interfaces=self._parse_units(data),
|
|||
|
|
)
|
|||
|
|
return ApiDescriptor(
|
|||
|
|
project=data.get("project", "default"),
|
|||
|
|
description=data.get("description", ""),
|
|||
|
|
interfaces=self._parse_units(data.get("units", [])),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ── 单元解析 ──────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def _parse_units(self, units: list[dict]) -> list[InterfaceInfo]:
|
|||
|
|
result = []
|
|||
|
|
for item in units:
|
|||
|
|
protocol = (
|
|||
|
|
item.get("protocol") or item.get("type") or "function"
|
|||
|
|
).lower()
|
|||
|
|
if protocol == "http":
|
|||
|
|
result.append(self._parse_http(item))
|
|||
|
|
else:
|
|||
|
|
result.append(self._parse_function(item))
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
def _parse_http(self, item: dict) -> InterfaceInfo:
|
|||
|
|
"""
|
|||
|
|
HTTP 接口解析。
|
|||
|
|
参数统一从 "parameters" 字段读取，
|
|||
|
|
同时兼容旧格式的 "request_parameters" / "url_parameters"。
|
|||
|
|
优先级：parameters > request_parameters + url_parameters
|
|||
|
|
"""
|
|||
|
|
# 优先使用新统一字段 "parameters"
|
|||
|
|
raw_params: dict = item.get("parameters", {})
|
|||
|
|
|
|||
|
|
# 旧格式兼容：合并 url_parameters + request_parameters
|
|||
|
|
if not raw_params:
|
|||
|
|
raw_params = {
|
|||
|
|
**item.get("url_parameters", {}),
|
|||
|
|
**item.get("request_parameters", {}),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return InterfaceInfo(
|
|||
|
|
name=item["name"],
|
|||
|
|
description=item.get("description", ""),
|
|||
|
|
protocol="http",
|
|||
|
|
url=item.get("url", ""),
|
|||
|
|
requirement_id=item.get("requirement_id", ""),
|
|||
|
|
method=item.get("method", "get").lower(),
|
|||
|
|
parameters=self._parse_param_dict(raw_params, with_inout=True),
|
|||
|
|
return_info=self._parse_return(item.get("return", {})),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def _parse_function(self, item: dict) -> InterfaceInfo:
|
|||
|
|
return InterfaceInfo(
|
|||
|
|
name=item["name"],
|
|||
|
|
description=item.get("description", ""),
|
|||
|
|
protocol="function",
|
|||
|
|
url=item.get("url", ""),
|
|||
|
|
requirement_id=item.get("requirement_id", ""),
|
|||
|
|
parameters=self._parse_param_dict(
|
|||
|
|
item.get("parameters", {}), with_inout=True,
|
|||
|
|
),
|
|||
|
|
return_info=self._parse_return(item.get("return", {})),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ── 参数解析 ──────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def _parse_param_dict(
|
|||
|
|
self,
|
|||
|
|
params: dict,
|
|||
|
|
with_inout: bool = False,
|
|||
|
|
) -> list[ParameterInfo]:
|
|||
|
|
result = []
|
|||
|
|
for name, info in params.items():
|
|||
|
|
result.append(ParameterInfo(
|
|||
|
|
name=name,
|
|||
|
|
type=info.get("type", "string"),
|
|||
|
|
description=info.get("description", ""),
|
|||
|
|
required=info.get("required", True),
|
|||
|
|
default=info.get("default", None),
|
|||
|
|
inout=info.get("inout", "in") if with_inout else "in",
|
|||
|
|
))
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# ── 返回值解析 ────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
def _parse_return(self, ret: dict) -> ReturnInfo:
|
|||
|
|
if not ret:
|
|||
|
|
return ReturnInfo()
|
|||
|
|
return ReturnInfo(
|
|||
|
|
type=ret.get("type", ""),
|
|||
|
|
description=ret.get("description", ""),
|
|||
|
|
on_success=self._parse_return_case(ret.get("on_success", {})),
|
|||
|
|
on_failure=self._parse_return_case(ret.get("on_failure", {})),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def _parse_return_case(self, case: dict) -> ReturnCase:
|
|||
|
|
if not case:
|
|||
|
|
return ReturnCase()
|
|||
|
|
return ReturnCase(
|
|||
|
|
value=case.get("value"),
|
|||
|
|
description=case.get("description", ""),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ══════════════════════════════════════════════════════════
|
|||
|
|
# 转换为 LLM 可读摘要
|
|||
|
|
# ══════════════════════════════════════════════════════════
|
|||
|
|
|
|||
|
|
def to_summary_dict(self, interfaces: list[InterfaceInfo]) -> list[dict]:
|
|||
|
|
return [
|
|||
|
|
self._http_summary(i) if i.protocol == "http"
|
|||
|
|
else self._function_summary(i)
|
|||
|
|
for i in interfaces
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
def _http_summary(self, iface: InterfaceInfo) -> dict:
|
|||
|
|
ri = iface.return_info
|
|||
|
|
return {
|
|||
|
|
"name": iface.name,
|
|||
|
|
"requirement_id": iface.requirement_id,
|
|||
|
|
"description": iface.description,
|
|||
|
|
"protocol": "http",
|
|||
|
|
"full_url": iface.http_full_url,
|
|||
|
|
"method": iface.method,
|
|||
|
|
# HTTP 接口统一用 "parameters" 输出给 LLM
|
|||
|
|
"parameters": self._params_to_dict(iface.parameters),
|
|||
|
|
"return": {
|
|||
|
|
"type": ri.type,
|
|||
|
|
"description": ri.description,
|
|||
|
|
"on_success": {
|
|||
|
|
"value": ri.on_success.value,
|
|||
|
|
"description": ri.on_success.description,
|
|||
|
|
},
|
|||
|
|
"on_failure": {
|
|||
|
|
"value": ri.on_failure.value,
|
|||
|
|
"description": ri.on_failure.description,
|
|||
|
|
},
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
def _function_summary(self, iface: InterfaceInfo) -> dict:
|
|||
|
|
ri = iface.return_info
|
|||
|
|
return {
|
|||
|
|
"name": iface.name,
|
|||
|
|
"requirement_id": iface.requirement_id,
|
|||
|
|
"description": iface.description,
|
|||
|
|
"protocol": "function",
|
|||
|
|
"source_file": iface.source_file,
|
|||
|
|
"module_path": iface.module_path,
|
|||
|
|
"parameters": self._params_to_dict(iface.parameters),
|
|||
|
|
"return": {
|
|||
|
|
"type": ri.type,
|
|||
|
|
"description": ri.description,
|
|||
|
|
"on_success": {
|
|||
|
|
"value": ri.on_success.value,
|
|||
|
|
"description": ri.on_success.description,
|
|||
|
|
},
|
|||
|
|
"on_failure": {
|
|||
|
|
"value": ri.on_failure.value,
|
|||
|
|
"description": ri.on_failure.description,
|
|||
|
|
},
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
def _params_to_dict(self, params: list[ParameterInfo]) -> dict:
|
|||
|
|
result = {}
|
|||
|
|
for p in params:
|
|||
|
|
entry: dict = {
|
|||
|
|
"type": p.type,
|
|||
|
|
"inout": p.inout,
|
|||
|
|
"description": p.description,
|
|||
|
|
"required": p.required,
|
|||
|
|
}
|
|||
|
|
if p.default is not None:
|
|||
|
|
entry["default"] = p.default
|
|||
|
|
result[p.name] = entry
|
|||
|
|
return result
|