395 lines
16 KiB
Python
395 lines
16 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
|
||
from config import config
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 基础数据结构
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
@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),
|
||
)
|
||
|
||
units = []
|
||
for unit in config.UNIT_KEYWORDS:
|
||
if unit not in data:
|
||
continue
|
||
units = data.get(unit, [])
|
||
|
||
return ApiDescriptor(
|
||
project=data.get("project", "default"),
|
||
description=data.get("description", ""),
|
||
interfaces=self._parse_units(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 |