205 lines
7.3 KiB
Python
205 lines
7.3 KiB
Python
"""
|
||
可声明式批量注册的 HTTP API 工具。
|
||
|
||
添加新接口只需在 API_ENDPOINTS 列表中追加一条配置，无需编写函数代码。
|
||
URL 中支持 {param} 路径参数，GET 自动走 query string，POST/PUT 自动走 JSON body。
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
|
||
import httpx
|
||
|
||
from tools.registry import tool_registry
|
||
|
||
# ── 公共默认值（所有接口共享） ─────────────────────────────
|
||
|
||
DEFAULT_BASE_URL = "https://hskuaikuai.com:9000/test" # TODO: 替换为实际服务地址
|
||
|
||
DEFAULT_HEADERS = {
|
||
"Content-Type": "application/json",
|
||
"token": "265a068eb7af455ca97f1a5063561a5d", # TODO: 后续由前端传入
|
||
}
|
||
|
||
DEFAULT_TIMEOUT = 10 # 秒
|
||
|
||
|
||
# ── API 端点声明 ─────────────────────────────────────────
|
||
#
|
||
# 字段说明：
|
||
# name ─ 工具名称（英文，唯一）
|
||
# description ─ 工具描述（LLM 据此判断何时调用）
|
||
# method ─ HTTP 方法：GET / POST / PUT / DELETE
|
||
# path ─ 请求路径（自动拼接 DEFAULT_BASE_URL），可含 {param} 占位符
|
||
# headers ─ 可选，额外请求头（与 DEFAULT_HEADERS 合并）
|
||
# parameters ─ OpenAI function calling parameters schema
|
||
#
|
||
# 路由规则：
|
||
# - path 自动拼接 DEFAULT_BASE_URL
|
||
# - {param} 占位符 → 从参数中取出并替换到 URL
|
||
# - GET → 剩余参数作为 query string
|
||
# - POST / PUT → 剩余参数作为 JSON body
|
||
#
|
||
|
||
API_ENDPOINTS: list[dict] = [
|
||
# ── 示例：签到汇总 ──
|
||
{
|
||
"name": "checkin_sum",
|
||
"description": "查询出勤人数",
|
||
"method": "POST",
|
||
"path": "/checkin/sum",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {},
|
||
"required": [],
|
||
},
|
||
},
|
||
# ── APP获取不同部门考勤打卡信息 ──
|
||
{
|
||
"name": "get_checkin_app_dept",
|
||
"description": "获取不同部门的考勤打卡信息，包括出勤率、总人数、打卡人数等。deptName传'全部'可一次查询所有部门汇总。",
|
||
"method": "POST",
|
||
"path": "/checkin/app/dept",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"deptName": {
|
||
"type": "string",
|
||
"description": "部门名称，传'全部'查所有部门",
|
||
"enum": ["养护部", "工程部", "环卫部", "办公室", "全部"],
|
||
},
|
||
},
|
||
"required": ["deptName"],
|
||
},
|
||
},
|
||
]
|
||
|
||
|
||
def _summarize_checkin_dept(raw: str) -> str:
|
||
"""将 get_checkin_app_dept 的完整返回精简为摘要，大幅减少 token 消耗。"""
|
||
try:
|
||
obj = json.loads(raw)
|
||
data = obj.get("data")
|
||
if not data or "workerList" not in data:
|
||
return raw
|
||
|
||
workers = data["workerList"]
|
||
attendance = data.get("attendance", "")
|
||
worker_num = data.get("workerNum", 0)
|
||
checkin_num = data.get("checkinNum", 0)
|
||
|
||
# 最早打卡
|
||
checked_in = [w for w in workers if w.get("checkin")]
|
||
earliest = None
|
||
if checked_in:
|
||
checked_in.sort(key=lambda w: w["checkin"])
|
||
e = checked_in[0]
|
||
t = e["checkin"].split("T")[1][:5] if "T" in e["checkin"] else e["checkin"]
|
||
earliest = f"{e['name']} {t}({e.get('deptName', '')})"
|
||
|
||
# 未打卡/请假
|
||
un_checked = [
|
||
f"{w['name']}({w.get('checkinState', '未打卡')})"
|
||
for w in workers if not w.get("checkin")
|
||
]
|
||
|
||
# 各子部门打卡人数汇总
|
||
dept_stats = {}
|
||
for w in workers:
|
||
dept = w.get("deptName", "未知")
|
||
if dept not in dept_stats:
|
||
dept_stats[dept] = {"total": 0, "checked": 0}
|
||
dept_stats[dept]["total"] += 1
|
||
if w.get("checkin"):
|
||
dept_stats[dept]["checked"] += 1
|
||
|
||
summary = {
|
||
"出勤率": f"{attendance}%",
|
||
"总人数": worker_num,
|
||
"已打卡": checkin_num,
|
||
"最早到岗": earliest,
|
||
"未到岗": un_checked if un_checked else "全部已打卡",
|
||
"子部门": {k: f"{v['checked']}/{v['total']}" for k, v in dept_stats.items()},
|
||
}
|
||
return json.dumps(summary, ensure_ascii=False)
|
||
except Exception:
|
||
return raw
|
||
|
||
|
||
# 工具名 → 后处理函数
|
||
_RESULT_SUMMARIZERS = {
|
||
"get_checkin_app_dept": _summarize_checkin_dept,
|
||
}
|
||
|
||
|
||
# ── 通用请求执行器 ────────────────────────────────────────
|
||
|
||
_PATH_PARAM_RE = re.compile(r"\{(\w+)\}")
|
||
|
||
|
||
def _call_api(endpoint: dict, **kwargs) -> str:
|
||
method = endpoint["method"].upper()
|
||
headers = {**DEFAULT_HEADERS, **endpoint.get("headers", {})}
|
||
|
||
# "全部"部门：批量查询所有部门，合并摘要后返回
|
||
if endpoint["name"] == "get_checkin_app_dept" and kwargs.get("deptName") == "全部":
|
||
all_depts = ["养护部", "工程部", "环卫部", "办公室"]
|
||
combined = {}
|
||
for dept in all_depts:
|
||
try:
|
||
url = DEFAULT_BASE_URL.rstrip("/") + endpoint["path"]
|
||
resp = httpx.request(method, url, headers=headers, json={"deptName": dept}, timeout=DEFAULT_TIMEOUT)
|
||
resp.raise_for_status()
|
||
summary = _summarize_checkin_dept(resp.text)
|
||
combined[dept] = json.loads(summary)
|
||
except Exception as exc:
|
||
combined[dept] = f"查询失败: {exc}"
|
||
result = json.dumps(combined, ensure_ascii=False)
|
||
print(f"全部部门摘要: {result[:300]}...")
|
||
return result
|
||
|
||
url = DEFAULT_BASE_URL.rstrip("/") + endpoint["path"]
|
||
# 1) 路径参数替换：{user_id} → kwargs["user_id"]
|
||
path_params = set(_PATH_PARAM_RE.findall(url))
|
||
for p in path_params:
|
||
value = kwargs.pop(p, "")
|
||
url = url.replace(f"{{{p}}}", str(value))
|
||
|
||
# 2) 分流：GET → query string；POST/PUT/DELETE → JSON body
|
||
try:
|
||
if method == "GET":
|
||
resp = httpx.request(
|
||
method, url, headers=headers, params=kwargs, timeout=DEFAULT_TIMEOUT
|
||
)
|
||
else:
|
||
resp = httpx.request(
|
||
method, url, headers=headers, json=kwargs or None, timeout=DEFAULT_TIMEOUT
|
||
)
|
||
resp.raise_for_status()
|
||
result = resp.text
|
||
print(f"API 调用成功 ({endpoint['name']}): {result[:200]}...")
|
||
# 对特定工具的返回做摘要，减少 LLM 上下文消耗
|
||
summarizer = _RESULT_SUMMARIZERS.get(endpoint["name"])
|
||
if summarizer:
|
||
result = summarizer(result)
|
||
print(f"摘要后 ({endpoint['name']}): {result}")
|
||
return result
|
||
except Exception as exc:
|
||
return f"API 调用失败 ({endpoint['name']}): {exc}"
|
||
|
||
|
||
# ── 自动注册所有声明的端点 ────────────────────────────────
|
||
|
||
for _ep in API_ENDPOINTS:
|
||
|
||
def _make_handler(ep=_ep):
|
||
def handler(**kw) -> str:
|
||
return _call_api(ep, **kw)
|
||
return handler
|
||
|
||
tool_registry.register(
|
||
name=_ep["name"],
|
||
description=_ep["description"],
|
||
parameters=_ep["parameters"],
|
||
)(_make_handler())
|