diff --git a/CHANGELOG.md b/CHANGELOG.md index 9031459..b1851d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## 0.62.0 — 2026-08-05 + +- 外部系统连接不再局限于 Factory MES：管理员现在可以直接配置标准 OpenAPI 系统，并选择用户名密码换取 Token、API Key 或 Bearer Token；用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。 + ## 0.61.1 — 2026-08-04 - 修复部分 MES 已成功连接、也能搜索接口，但实际查询始终返回 404 的问题；现在会自动识别接口规范声明的 `/api`、`/v1` 等业务路径前缀，无需修改现有连接配置。 diff --git a/DESIGN.md b/DESIGN.md index 7534766..ce8289d 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -395,17 +395,20 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) **不选**:Celery/RQ(多机分发/任务序列化/框架重试——单机 + 模型现写脚本的场景一个都用不上,还多两个常驻组件的部署/蓝绿适配);工具层 async 化 run 内等待(run 不结束,409 照旧,重启照丢);DB 表 + 守护(文件已是事实源,detach 进程写 PG 还得给它凭证)。升级触发:要跨机器跑计算集群时,①②的工具接口不变,只换执行后端。 -### 8.14 外部系统:用户身份连接 + 受控接口调用(implementation,2026-08-04) +### 8.14 外部系统:用户身份连接 + 受控接口调用(implementation,2026-08-05) **诉求**:用户用自己的 MES/ERP/LIMS 账号让 zcbot 做信息查询,并把稳定的问法沉淀成私有 skill。**心智模型**:外部系统负责「连接与身份」,工具负责「受控访问」,skill 负责「业务流程与经验」。它有独立于会话的持久凭据和连接状态,因此是与 skill/知识库/记忆并列的**平台机制**,不是 skill。 **首个 provider=`factory_mes`**:Factory 已有 JWT + RBAC + 部分部门数据权限,zcbot 用每位用户自己的 Factory 账密换 JWT,调用时继承 MES 原生权限;不在 zcbot 里复制第二套 MES RBAC。两层门控:zcbot `user_id` 只能取自己的 `external_systems` 行;远端 JWT 再判定实际业务数据范围。MES 停号/改权后下次调用即生效。 +**通用连接器边界**:`openapi` connector 负责规格发现、operation 解析、安全 URL 拼接、参数校验、只读 allowlist、分页和响应体积限制；认证由独立 strategy 负责。`factory_mes` 只是带 JWT 字段映射、dataset 推荐入口和查询规划提示的内置 preset，`generic_openapi` 可由管理员直接选择用户名密码换 Token、API Key 或 Bearer Token。标准 OpenAPI 系统以后只新增数据库 definition，不需要再写 Python 文件；只有 OAuth 回调/签名交换、SOAP、消息队列或私有二进制协议等不符合现有 connector/strategy 契约的系统才新增适配代码。provider 注册表维护可选能力和安全默认值，不为每个业务系统复制 connector。 + **信任边界**: -- provider 公共定义由管理员在管理后台维护并存入 `external_system_definitions`:Base URL、OpenAPI URL、登录方式和只读 POST allowlist；普通用户只选择已启用的目录项并填写自己的 MES 账密。不允许普通用户填任意 URL,避免 SSRF/内网代理。凭据主密钥仍只来自宿主环境，不进入数据库或管理页面。 +- provider 公共定义由管理员在管理后台维护并存入 `external_system_definitions`:Base URL、OpenAPI URL、认证 strategy/字段映射和只读 POST allowlist；普通用户只选择已启用的目录项，凭据表单按 definition 声明动态生成。不允许普通用户填任意 URL,避免 SSRF/内网代理。凭据主密钥仍只来自宿主环境，不进入数据库或管理页面。 - 凭据用独立的 `ZCBOT_CREDENTIAL_MASTER_KEY` 在 host control plane 加密入 PG，不与 `JWT_SECRET` 复用，以隔离泄漏半径和轮换生命周期；缺 key 则拒绝新建/调用,不像早期微信绑定那样降级明文。API 只返回脱敏账号和 `credential_configured`,不返密码/Token;凭据绝不进 prompt/messages/memory/skill/用户 FS/日志/沙箱。 -- 调用工具不接受完整 URL,只接受 OpenAPI `operation_id`;服务端从受信规格解析 path/method,校验 path/query/body 后附加 JWT。默认只开 GET/HEAD,语义只读但使用 POST 的 BI 查询必须进运维 `operation_id` allowlist。 +- 调用工具不接受完整 URL,只接受 OpenAPI `operation_id`;服务端从受信规格解析 path/method,校验 path/query/body 后附加认证 strategy 生成的 Header。默认只开 GET/HEAD,语义只读但使用 POST 的 BI 查询必须进运维 `operation_id` allowlist。 - Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。 +- Swagger/OpenAPI JSON 不持久化入数据库或文件，连接器按 `definition_id + user_id` 隔离后放在进程内存中缓存 5 分钟；重启自动失效。这样保留实时契约发现，又避免不同身份可见的规格互相污染。 **工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂三个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要 + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST；管理员在 definition JSONB 配置 `query_guidance` 与 `recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯；Swagger 业务文本仍是不可信数据。 diff --git a/PROGRESS.md b/PROGRESS.md index 3896f0a..b7dfe7b 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-08-04(MES OpenAPI 业务路径前缀兼容,bump 0.61.1) +最后更新:2026-08-05(通用 OpenAPI 外部系统与可配置认证,bump 0.62.0) --- @@ -21,6 +21,10 @@ ## 已完成关键能力 +### 2026-08-05 + +- **08-05 / 0.62.0 / 通用 OpenAPI 外部系统 + 可配置认证**:将 Factory 专用运行态拆为通用 OpenAPI connector、认证 strategy 和 provider preset 注册表；标准 ERP/LIMS 等系统可直接在管理后台选择用户名密码换 Token、API Key 或 Bearer Token，并配置 Token 字段与认证 Header，无需再新增 Python connector。Factory MES 保持内置 preset、既有配置和旧 `{username,password}` HTTP 请求兼容；用户凭据表单按 definition 动态生成，Swagger JSON 不落盘并按 `definition_id+user_id` 隔离缓存 5 分钟。完整 475 项 unittest 全绿(17 skip)，相关 41 项回归、Python/JavaScript 语法及 diff 检查通过；当前环境无可用浏览器实例，真实页面点击留部署后冒烟。无 schema、migration 或依赖变化，未连接生产 DB。 + ### 2026-08-04 - **08-04 / 0.61.1 / MES OpenAPI 业务路径前缀兼容**:生产 task `1ade8062` 显示光芯 MES 登录与 OpenAPI 搜索均成功，但 operation path `/mtm/...` 直接拼到主机根路径后被 nginx 404；真实业务路由位于 `/api/mtm/...`。Factory 连接器现读取 Swagger 2 `basePath` 与 OpenAPI 3 `servers` 的同源路径前缀，兼容规范 path 或管理员 Base URL 已含前缀的情况且不重复拼接；登录路径继续独立解析，跨主机 server 仍拒绝。外部系统与无 DB Web 路由共 34 项 unittest、Ruff 致命规则及 diff 检查通过；mypy 仍报告既有 `core/external_systems/service.py:242` SQLAlchemy `rowcount` 类型问题，本次未改该文件；无 schema、migration、配置或依赖变化，未写生产 DB。 diff --git a/RUN.md b/RUN.md index 582c279..9525ca1 100644 --- a/RUN.md +++ b/RUN.md @@ -2,7 +2,7 @@ > 怎么把 zcbot 跑起来。env / 常用命令 / 故障兜底。设计看 `DESIGN.md`,进度看 `PROGRESS.md`。 -最后更新:2026-08-04(新增 Factory MES 外部系统连接的管理员配置、用户绑定和只读接口调用说明) +最后更新:2026-08-05(外部系统支持通用 OpenAPI 配置与动态认证凭据) --- @@ -131,8 +131,8 @@ # 对外品牌名:zcbot 是内部代号,所有用户可见文案(页面标题/顶栏/登录卡/微信·企微推送与 # 提示页)统一用品牌名。/healthz 返回 brand 字段,前端 boot 拉取覆盖静态页默认值。 # ZCBOT_BRAND_NAME=总院科研辅助助手 # 可选,默认即此值 - # Factory MES 外部系统(DESIGN §8.14):公共地址在管理后台配置，用户在「外部」 - # 入口提交自己的 MES 用户名/密码；凭据仅密文入库，agent 只能按 operationId 调用接口。 + # OpenAPI 外部系统(DESIGN §8.14):公共地址和认证方式在管理后台配置，用户在「外部」 + # 入口提交系统要求的凭据；凭据仅密文入库，agent 只能按 operationId 调用接口。 # MASTER_KEY 应使用独立随机值，不与 JWT_SECRET 共用。 # ZCBOT_CREDENTIAL_MASTER_KEY=<至少 32 字符随机串> ``` @@ -150,7 +150,7 @@ - **未绑定成员发消息 → 回绑定指引**(不再静默):聊天优先布局下新员工第一动作就是打字,回调对未绑定成员的 text/图片/文件消息每条回一句"先去控制台绑定"(事件不回)。未绑定成员点菜单「工作台」则落在绑定提示页(不自动建号)。 - **channel 长会话上下文(微信/企业微信通用,0019)**:常驻会话不再无限膨胀。① **自动分段**——入站时距上次消息超过 `config.json` 的 `channel.session_gap_hours`(默 **6** 小时,设 `<=0` 关闭)→ 软重置:只把「最后一条 user 消息起」喂模型(保留上一轮做续聊锚点),之前的历史仍全留 DB,网页端照旧翻完整记录;② **手动新话题**——用户在微信/企业微信里直接发「新话题 / 新会话 / `/new` / 清空上下文」→ 硬重置,彻底从零(回执提示已归档)。两者都**不删任何消息**,只移动「喂给模型的窗口起点」`tasks.context_base_idx`。网页端「清空对话」(`POST /v1/tasks/{id}/clear`)仍整清并把 base 归 0。需 `main.py db upgrade head` 带上 `0019`。 - **PG**:`ZCBOT_DB_URL` 必填。本地 docker compose / 远端 dev / 生产任选;未设置时启动清晰报错,不引导 docker(§7.4)。 -- **Factory MES 外部系统**:① `.env` 只配置独立的 `ZCBOT_CREDENTIAL_MASTER_KEY`；② 执行 `main.py db upgrade head` 创建系统目录和用户连接两张表；③ 重启 web；④ admin 进入管理后台「外部系统」，配置可信 Base URL、Swagger URL、只读 POST operationId、推荐查询入口 operationId 和查询规划提示，并选择“全部用户”或指定用户；提示词按钮复用页内弹框多行编辑，`Ctrl/Command+Enter` 应用；⑤ 普通用户点击左栏 **「外部」**，只会看到自己获权的 MES，再填写个人账号密码。工具下一轮对话开始挂载；管理员配置的推荐入口会在接口搜索中置顶。Factory 默认提示统计聚合先走 `bi_dataset_list` → `bi_dataset_exec`，生产日志用于逐条追溯；单轮累计返回量、`page_size<=200` 及禁止关闭分页共同约束明细扫描。管理员撤权立即停止调用并删除该用户密文凭据。配置复用既有 JSONB，无新 migration；不读取 Gitea 代码、不直接连 MES 数据库，也不允许普通用户或模型传任意 URL。 +- **OpenAPI 外部系统**:① `.env` 只配置独立的 `ZCBOT_CREDENTIAL_MASTER_KEY`；② 首次启用执行 `main.py db upgrade head`；③ admin 进入管理后台「外部系统」，选择 Factory MES preset 或通用 OpenAPI，配置可信 Base URL、Swagger URL、认证方式、只读 POST operationId、推荐查询入口和查询规划提示，再选择“全部用户”或指定用户。通用类型支持“用户名密码换取 Token”“API Key”“Bearer Token”；用户名、密码、Token 和 Header 字段映射由管理员维护。④ 普通用户点击左栏 **「外部」**，页面按定义动态显示所需凭据。工具下一轮对话开始挂载；推荐入口会在接口搜索中置顶。Factory preset 默认把统计聚合路由到 `bi_dataset_list` → `bi_dataset_exec`，生产日志只用于逐条追溯；单轮累计返回量、`page_size<=200` 及禁止关闭分页共同约束明细扫描。配置复用既有 JSONB，无新 migration。Swagger JSON 只在进程内按系统定义和用户缓存 5 分钟，不写文件/数据库。普通用户和模型仍不能传任意 URL。 - **测试库(可选,`ZCBOT_TEST_DB_URL`)**:DB 级单测(`tests/test_usage_report.py` / `tests/test_scheduler.py` / `tests/test_web_routes_db.py`)**只认这个显式变量、绝不回退 `.env` 的 `ZCBOT_DB_URL`**——后者可能经隧道指向生产库,测试插入的到点 job 会被生产实例调度守护真跑一次(2026-07-23 实锤)。未设则这几组自动 skip。一键起库(docker,端口 5433 避开本地 5432): ```bash docker run -d --name zcbot-test-pg -e POSTGRES_PASSWORD=zcbot_test \ @@ -346,9 +346,9 @@ $env:ZCBOT_EVAL_TOKEN = "" | `GET /v1/skills` | 列当前 user 可用 skill(内置 + 自己的);每项带 `source`(builtin/user)/`overrides_builtin`;另返 `load_errors`(用户 skill 因 frontmatter 坏未加载的) | 必填 | | `GET /v1/skills/{name}` | 返某 skill 完整 SKILL.md 正文(前端「技能」modal 点开查看);同名按 user wins | 必填 | | `DELETE /v1/skills/{name}` | 删当前 user 私有 skill(`.skills//` 整目录);只删 user 源,内置不可删 → 404;`.skills` 文件面板隐藏,这是 UI 上删自己 skill 的唯一入口 | 必填 | -| `GET /v1/external-system-providers` | 列管理员已启用的外部系统目录；只返回目录 ID、名称、可用性和主机名，不返回完整配置或密钥 | 必填 | -| `GET/POST /v1/external-systems` | 列当前用户连接 / 新建并在线验证 Factory MES 连接；创建 body `{provider,name,username,password}`，响应仅含脱敏用户名 | 必填 | -| `PUT /v1/external-systems/{id}/credentials` | 重新提交并在线验证当前用户连接的用户名/密码；凭据不提供读取接口 | 必填 | +| `GET /v1/external-system-providers` | 列管理员已启用的外部系统目录和安全的动态凭据字段声明；不返回完整配置或密钥 | 必填 | +| `GET/POST /v1/external-systems` | 列当前用户连接 / 新建并在线验证连接；创建 body `{definition_id,name,credentials}`，旧 Factory `{username,password}` 请求继续兼容 | 必填 | +| `PUT /v1/external-systems/{id}/credentials` | 用 `{credentials}` 重新提交并在线验证当前用户连接；凭据不提供读取接口，旧用户名/密码格式继续兼容 | 必填 | | `POST /v1/external-systems/{id}/test` | 用已保存密文凭据测试登录和 Swagger 可读性，并更新连接状态 | 必填 | | `DELETE /v1/external-systems/{id}` | 清除当前用户连接及密文凭据；指定用户模式保留管理员授予的可见权 | 必填 | | `GET/POST /v1/admin/external-system-definitions` | 管理员列出或新增可信外部系统目录 | admin | diff --git a/core/__init__.py b/core/__init__.py index 3a72a5e..ef76fa7 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.61.1" +__version__ = "0.62.0" diff --git a/core/external_systems/auth.py b/core/external_systems/auth.py new file mode 100644 index 0000000..4d59ea5 --- /dev/null +++ b/core/external_systems/auth.py @@ -0,0 +1,156 @@ +"""外部系统认证策略。 + +认证只消费管理员保存的可信配置和用户加密保存的字段，不允许模型指定认证地址或请求头。 +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Protocol +from urllib.parse import urljoin + +import httpx + + +class ExternalAuthError(RuntimeError): + pass + + +@dataclass(frozen=True) +class CredentialField: + name: str + label: str + secret: bool = True + autocomplete: str = "off" + + def as_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "label": self.label, + "secret": self.secret, + "autocomplete": self.autocomplete, + } + + +class AuthStrategy(Protocol): + credential_fields: tuple[CredentialField, ...] + + def headers( + self, + *, + client: httpx.Client, + base_url: str, + credentials: dict[str, str], + config: dict[str, Any], + ) -> dict[str, str]: ... + + +def _required(credentials: dict[str, str], fields: tuple[CredentialField, ...]) -> None: + missing = [field.label for field in fields if not credentials.get(field.name, "").strip()] + if missing: + raise ExternalAuthError("请填写" + "、".join(missing)) + + +def _nested_value(payload: Any, path: str) -> Any: + current = payload + for part in path.split("."): + if not isinstance(current, dict): + return None + current = current.get(part) + return current + + +def _auth_header(config: dict[str, Any], token: str, *, default_name: str, default_template: str) -> dict[str, str]: + name = str(config.get("auth_header_name") or default_name).strip() + template = str(config.get("auth_header_template") or default_template) + if any(char in name for char in "\r\n:") or any(char in template for char in "\r\n"): + raise ExternalAuthError("认证 Header 配置非法") + if "{token}" not in template: + raise ExternalAuthError("认证 Header 模板必须包含 {token}") + return {name: template.replace("{token}", token)} + + +class PasswordJwtAuth: + credential_fields = ( + CredentialField("username", "用户名", secret=False, autocomplete="username"), + CredentialField("password", "密码", autocomplete="current-password"), + ) + + def headers(self, *, client, base_url, credentials, config) -> dict[str, str]: + _required(credentials, self.credential_fields) + login_path = str(config.get("login_path") or "/api/auth/token/").strip() + if not login_path.startswith("/") or "://" in login_path: + raise ExternalAuthError("login_path 必须是站内绝对路径") + username_field = str(config.get("username_field") or "username").strip() + password_field = str(config.get("password_field") or "password").strip() + token_field = str(config.get("token_field") or "access").strip() + try: + response = client.post( + urljoin(base_url + "/", login_path.lstrip("/")), + json={ + username_field: credentials["username"], + password_field: credentials["password"], + }, + ) + except httpx.HTTPError as exc: + raise ExternalAuthError(f"外部系统登录连接失败: {type(exc).__name__}") from exc + if response.status_code >= 400: + raise ExternalAuthError(f"外部系统登录失败(HTTP {response.status_code})") + try: + token = _nested_value(response.json(), token_field) + except ValueError: + token = None + if not isinstance(token, str) or not token: + raise ExternalAuthError(f"外部系统登录响应缺少 {token_field}") + return _auth_header( + config, token, default_name="Authorization", default_template="Bearer {token}" + ) + + +class ApiKeyAuth: + credential_fields = (CredentialField("api_key", "API Key"),) + + def headers(self, *, client, base_url, credentials, config) -> dict[str, str]: + _required(credentials, self.credential_fields) + return _auth_header( + config, credentials["api_key"], default_name="X-API-Key", default_template="{token}" + ) + + +class BearerTokenAuth: + credential_fields = (CredentialField("token", "Bearer Token"),) + + def headers(self, *, client, base_url, credentials, config) -> dict[str, str]: + _required(credentials, self.credential_fields) + return _auth_header( + config, credentials["token"], default_name="Authorization", default_template="Bearer {token}" + ) + + +_AUTH_STRATEGIES: dict[str, AuthStrategy] = { + "password_jwt": PasswordJwtAuth(), + "api_key": ApiKeyAuth(), + "bearer_token": BearerTokenAuth(), +} + + +def get_auth_strategy(auth_type: str) -> AuthStrategy: + strategy = _AUTH_STRATEGIES.get((auth_type or "").strip()) + if strategy is None: + raise ExternalAuthError(f"不支持的认证方式: {auth_type}") + return strategy + + +def auth_catalog() -> list[dict[str, Any]]: + titles = { + "password_jwt": "用户名密码换取 Token", + "api_key": "API Key", + "bearer_token": "Bearer Token", + } + return [ + { + "auth_type": key, + "title": titles[key], + "credential_fields": [field.as_dict() for field in strategy.credential_fields], + } + for key, strategy in _AUTH_STRATEGIES.items() + ] diff --git a/core/external_systems/factory.py b/core/external_systems/factory.py index 3412440..a122fad 100644 --- a/core/external_systems/factory.py +++ b/core/external_systems/factory.py @@ -1,419 +1,28 @@ -"""Factory MES OpenAPI connector。 +"""Factory MES 兼容入口。 -目标地址全部来自管理员维护的可信系统目录；模型和普通用户只能传 operation_id -与结构化参数，不能传 URL。 +新代码使用 :mod:`core.external_systems.openapi`；保留原类名，避免已有测试和内部引用 +在通用化过程中发生无意义破坏。 """ from __future__ import annotations -import json -import re -import time -from dataclasses import dataclass -from threading import Lock -from typing import Any, Optional -from urllib.parse import quote, urljoin, urlparse +from typing import Any -import httpx +from .openapi import OpenApiClient, OpenApiConfig, OpenApiError, _SPEC_CACHE +from .registry import merged_config -class FactoryMesError(RuntimeError): - pass +FactoryMesError = OpenApiError -_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete") -_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} -_SPEC_LOCK = Lock() - -DEFAULT_QUERY_GUIDANCE = ( - "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询，" - "统一先调用 BI dataset list，再执行匹配的数据集。日志和业务明细列表用于" - "用户明确要求查看逐条记录、编号或追溯过程的场景。未匹配到 dataset 时，" - "先限定范围或向用户确认明细查询需求。" -) -DEFAULT_RECOMMENDED_OPERATIONS = ("bi_dataset_list", "bi_dataset_exec") - - -def _bool_value(value: Any, default: bool) -> bool: - raw = str(value if value is not None else "").strip().lower() - if not raw: - return default - return raw in {"1", "true", "yes", "on"} - - -def _validated_http_url(raw: str, label: str) -> str: - value = (raw or "").strip().rstrip("/") - parsed = urlparse(value) - if parsed.scheme not in {"http", "https"} or not parsed.netloc: - raise FactoryMesError(f"{label} 必须是有效的 http(s) URL") - if parsed.username or parsed.password: - raise FactoryMesError(f"{label} 不能内嵌凭据") - return value - - -@dataclass(frozen=True) -class FactoryMesConfig: - base_url: str - openapi_url: str - login_path: str - allowed_post_operations: frozenset[str] - timeout_seconds: float - max_result_bytes: int - max_total_result_bytes: int - max_page_size: int - verify_tls: bool - query_guidance: str - recommended_operation_ids: tuple[str, ...] - +class FactoryMesConfig(OpenApiConfig): @classmethod def from_mapping(cls, data: dict[str, Any]) -> "FactoryMesConfig": - """从管理员保存的可信目录配置构建运行态配置。""" - base = _validated_http_url(str(data.get("base_url") or ""), "base_url") - spec = _validated_http_url( - str(data.get("openapi_url") or ""), "openapi_url" - ) - login_path = str(data.get("login_path") or "/api/auth/token/").strip() - if not login_path.startswith("/") or "://" in login_path: - raise FactoryMesError("login_path 必须是站内绝对路径") - raw_allowed = data.get("allowed_post_operations") or [] - if isinstance(raw_allowed, str): - raw_allowed = raw_allowed.split(",") - if not isinstance(raw_allowed, (list, tuple, set)): - raise FactoryMesError("allowed_post_operations 必须是字符串数组") - allowed = frozenset(str(item).strip() for item in raw_allowed if str(item).strip()) - guidance = str(data.get("query_guidance") or DEFAULT_QUERY_GUIDANCE).strip() - if len(guidance) > 4000: - raise FactoryMesError("query_guidance 不能超过 4000 字符") - raw_recommended = data.get( - "recommended_operation_ids", DEFAULT_RECOMMENDED_OPERATIONS - ) - if isinstance(raw_recommended, str): - raw_recommended = raw_recommended.split(",") - if not isinstance(raw_recommended, (list, tuple, set)): - raise FactoryMesError("recommended_operation_ids 必须是字符串数组") - recommended = tuple(dict.fromkeys( - str(item).strip() for item in raw_recommended if str(item).strip() - )) - if len(recommended) > 30 or any(len(item) > 200 for item in recommended): - raise FactoryMesError("recommended_operation_ids 最多 30 项且每项不超过 200 字符") - max_result = max(4096, min(int(data.get("max_result_bytes", 65536)), 1048576)) - return cls( - base_url=base, - openapi_url=spec, - login_path=login_path, - allowed_post_operations=allowed, - timeout_seconds=max(1.0, min(float(data.get("timeout_seconds", 15)), 60.0)), - max_result_bytes=max_result, - max_total_result_bytes=max( - max_result, - min(int(data.get("max_total_result_bytes", 262144)), 4194304), - ), - max_page_size=max(1, min(int(data.get("max_page_size", 200)), 1000)), - verify_tls=_bool_value(data.get("verify_tls"), True), - query_guidance=guidance, - recommended_operation_ids=recommended, - ) + common = OpenApiConfig.from_mapping(merged_config("factory_mes", data)) + return cls(**common.__dict__) -class FactoryMesClient: +class FactoryMesClient(OpenApiClient): def __init__(self, username: str, password: str, cfg: FactoryMesConfig): self.username = username self.password = password - self.cfg = cfg - - def _client(self) -> httpx.Client: - return httpx.Client( - timeout=self.cfg.timeout_seconds, - verify=self.cfg.verify_tls, - follow_redirects=False, - ) - - def authenticate(self) -> str: - url = urljoin(self.cfg.base_url + "/", self.cfg.login_path.lstrip("/")) - try: - with self._client() as client: - response = client.post( - url, - json={"username": self.username, "password": self.password}, - ) - except httpx.HTTPError as exc: - raise FactoryMesError(f"Factory MES 登录连接失败: {type(exc).__name__}") from exc - if response.status_code >= 400: - raise FactoryMesError(f"Factory MES 登录失败(HTTP {response.status_code})") - try: - token = response.json().get("access", "") - except (ValueError, AttributeError): - token = "" - if not isinstance(token, str) or not token: - raise FactoryMesError("Factory MES 登录响应缺少 access token") - return token - - def _fetch_spec(self, token: str) -> dict[str, Any]: - now = time.monotonic() - with _SPEC_LOCK: - hit = _SPEC_CACHE.get(self.cfg.openapi_url) - if hit and now - hit[0] < 300: - return hit[1] - try: - with self._client() as client: - response = client.get( - self.cfg.openapi_url, - headers={"Authorization": f"Bearer {token}"}, - ) - except httpx.HTTPError as exc: - raise FactoryMesError(f"Factory OpenAPI 获取失败: {type(exc).__name__}") from exc - if response.status_code >= 400: - raise FactoryMesError(f"Factory OpenAPI 获取失败(HTTP {response.status_code})") - try: - spec = response.json() - except ValueError as exc: - raise FactoryMesError("Factory OpenAPI 不是有效 JSON") from exc - if not isinstance(spec, dict) or not isinstance(spec.get("paths"), dict): - raise FactoryMesError("Factory OpenAPI 缺少 paths") - with _SPEC_LOCK: - _SPEC_CACHE[self.cfg.openapi_url] = (now, spec) - return spec - - @staticmethod - def _operation_id(method: str, path: str, operation: dict[str, Any]) -> str: - explicit = operation.get("operationId") - if isinstance(explicit, str) and explicit.strip(): - return explicit.strip() - safe_path = re.sub(r"[^a-zA-Z0-9]+", "_", path).strip("_") - return f"{method}_{safe_path}" - - @classmethod - def _operations(cls, spec: dict[str, Any]) -> list[dict[str, Any]]: - results: list[dict[str, Any]] = [] - for path, path_item in (spec.get("paths") or {}).items(): - if not isinstance(path_item, dict): - continue - common = path_item.get("parameters") or [] - for method in _HTTP_METHODS: - operation = path_item.get(method) - if not isinstance(operation, dict): - continue - params = list(common) + list(operation.get("parameters") or []) - results.append({ - "operation_id": cls._operation_id(method, path, operation), - "method": method.upper(), - "path": path, - "summary": operation.get("summary") or "", - "description": operation.get("description") or "", - "tags": operation.get("tags") or [], - "parameters": params, - "request_body": operation.get("requestBody"), - }) - return results - - def _spec_base_path(self, spec: dict[str, Any]) -> str: - """Return the API path prefix declared by Swagger 2 / OpenAPI 3. - - The remote specification may describe a different host, but an external - system definition is the only authority allowed to choose the target - origin. OpenAPI ``servers`` therefore contributes only a same-origin - path prefix. - """ - raw_base_path = spec.get("basePath") - if raw_base_path is not None: - if not isinstance(raw_base_path, str) or not raw_base_path.startswith("/"): - raise FactoryMesError("Swagger basePath 必须是站内绝对路径") - parsed = urlparse(raw_base_path) - if parsed.netloc or parsed.query or parsed.fragment or "://" in raw_base_path: - raise FactoryMesError("Swagger basePath 非法") - return parsed.path.rstrip("/") - - servers = spec.get("servers") - if not isinstance(servers, list) or not servers: - return "" - server = servers[0] - raw_url = server.get("url") if isinstance(server, dict) else None - if not isinstance(raw_url, str) or not raw_url.strip(): - raise FactoryMesError("OpenAPI server URL 无效") - raw_url = raw_url.strip() - if "{" in raw_url or "}" in raw_url: - raise FactoryMesError("OpenAPI server URL 包含未解析变量") - declared = urlparse(raw_url) - base = urlparse(self.cfg.base_url) - if declared.netloc and (declared.scheme, declared.netloc) != ( - base.scheme, - base.netloc, - ): - raise FactoryMesError("OpenAPI server 越出 Factory MES 主机") - if declared.query or declared.fragment: - raise FactoryMesError("OpenAPI server URL 不能包含查询或片段") - return ("/" + declared.path.lstrip("/")).rstrip("/") - - def _operation_url(self, spec: dict[str, Any], operation_path: str) -> str: - prefix = self._spec_base_path(spec) - base = urlparse(self.cfg.base_url) - configured_path = base.path.rstrip("/") - path = operation_path - configured_has_prefix = bool(prefix) and ( - configured_path == prefix or configured_path.endswith(prefix) - ) - operation_has_prefix = bool(prefix) and ( - path == prefix or path.startswith(prefix + "/") - ) - if configured_has_prefix and operation_has_prefix: - path = path[len(prefix):] or "/" - elif prefix and not configured_has_prefix and not operation_has_prefix: - path = prefix + "/" + path.lstrip("/") - combined_path = "/".join( - part.strip("/") for part in (configured_path, path) if part.strip("/") - ) - if operation_path.endswith("/") and combined_path: - combined_path += "/" - url = urljoin(f"{base.scheme}://{base.netloc}/", combined_path) - call_origin = urlparse(url) - if (call_origin.scheme, call_origin.netloc) != (base.scheme, base.netloc): - raise FactoryMesError("接口目标越出 Factory MES 主机") - return url - - def test_connection(self) -> dict[str, Any]: - token = self.authenticate() - spec = self._fetch_spec(token) - return {"operation_count": len(self._operations(spec))} - - def search(self, query: str, limit: int = 12) -> list[dict[str, Any]]: - query = (query or "").strip().lower() - if not query: - raise FactoryMesError("query 不能为空") - token = self.authenticate() - spec = self._fetch_spec(token) - terms = list(dict.fromkeys( - [query] + [x for x in re.split(r"[\s,，。/]+", query) if len(x) >= 2] - )) - recommended_order = { - operation_id: index - for index, operation_id in enumerate(self.cfg.recommended_operation_ids) - } - scored: list[tuple[int, int, dict[str, Any]]] = [] - for op in self._operations(spec): - method = op["method"].lower() - if method not in {"get", "head"} and not ( - method == "post" - and op["operation_id"] in self.cfg.allowed_post_operations - ): - continue - hay = " ".join([ - op["operation_id"], op["path"], op["summary"], op["description"], - " ".join(str(x) for x in op["tags"]), - ]).lower() - score = sum(5 if term == query and term in hay else 1 for term in terms if term in hay) - recommended = op["operation_id"] in recommended_order - if score or recommended: - compact = dict(op) - compact["parameters"] = [ - { - "name": p.get("name"), - "in": p.get("in"), - "required": bool(p.get("required")), - "type": p.get("type") or (p.get("schema") or {}).get("type"), - "description": p.get("description") or "", - } - for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p - ] - compact.pop("request_body", None) - compact["recommended"] = recommended - scored.append(( - 0 if recommended else 1, - recommended_order.get(op["operation_id"], -score), - compact, - )) - scored.sort(key=lambda item: (item[0], item[1], item[2]["operation_id"])) - return [item[2] for item in scored[: max(1, min(int(limit), 30))]] - - def call( - self, - operation_id: str, - arguments: Optional[dict[str, Any]] = None, - body: Any = None, - ) -> dict[str, Any]: - token = self.authenticate() - spec = self._fetch_spec(token) - matches = [op for op in self._operations(spec) if op["operation_id"] == operation_id] - if len(matches) != 1: - raise FactoryMesError("operation_id 不存在或不唯一，请先搜索接口") - op = matches[0] - if not op["path"].startswith("/") or "://" in op["path"]: - raise FactoryMesError("OpenAPI operation path 非法") - method = op["method"].lower() - if method not in {"get", "head"} and not ( - method == "post" and operation_id in self.cfg.allowed_post_operations - ): - raise FactoryMesError(f"operation {operation_id} 未列入只读调用范围") - - supplied = dict(arguments or {}) - path = op["path"] - query: dict[str, Any] = {} - headers = {"Authorization": f"Bearer {token}"} - request_body = body - for param in op["parameters"]: - if not isinstance(param, dict) or "$ref" in param: - continue - name = param.get("name") - location = param.get("in") - if not isinstance(name, str): - continue - # Swagger 2 的 body 参数既可按搜索结果中的参数名放在 arguments， - # 也可使用元工具独立的 body 字段；两者只取一个。 - present = name in supplied or (location == "body" and request_body is not None) - if param.get("required") and not present: - raise FactoryMesError(f"缺少必填参数: {name}") - if name not in supplied: - continue - value = supplied.pop(name) - if location == "path": - path = path.replace("{" + name + "}", quote(str(value), safe="")) - elif location == "query": - if name == "page_size": - try: - value = max(1, min(int(value), self.cfg.max_page_size)) - except (TypeError, ValueError) as exc: - raise FactoryMesError("page_size 必须是整数") from exc - elif name == "page" and str(value).strip() == "0": - raise FactoryMesError( - "外部系统查询不允许 page=0 关闭分页，请使用 dataset 或分页查看明细" - ) - elif name == "pageoff" and _bool_value(value, False): - raise FactoryMesError( - "外部系统查询不允许关闭分页，请使用 dataset 或分页查看明细" - ) - query[name] = value - elif location == "body" and request_body is None: - request_body = value - if supplied: - raise FactoryMesError("存在接口定义之外的参数: " + ", ".join(sorted(supplied))) - if "{" in path or "}" in path: - raise FactoryMesError("路径参数未完整提供") - - url = self._operation_url(spec, path) - try: - with self._client() as client: - response = client.request( - method.upper(), - url, - params=query, - json=request_body if method == "post" else None, - headers=headers, - ) - except httpx.HTTPError as exc: - raise FactoryMesError(f"Factory 接口调用失败: {type(exc).__name__}") from exc - if response.status_code >= 400: - raise FactoryMesError(f"Factory 接口返回 HTTP {response.status_code}") - content_type = response.headers.get("content-type", "") - try: - payload: Any = response.json() if "json" in content_type else response.text - except ValueError: - payload = response.text - encoded = json.dumps(payload, ensure_ascii=False, default=str) - truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes - if truncated: - encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore") - payload = encoded - return { - "operation_id": operation_id, - "status_code": response.status_code, - "truncated": truncated, - "data": payload, - } + super().__init__({"username": username, "password": password}, cfg) diff --git a/core/external_systems/openapi.py b/core/external_systems/openapi.py new file mode 100644 index 0000000..c7477a3 --- /dev/null +++ b/core/external_systems/openapi.py @@ -0,0 +1,430 @@ +"""通用 OpenAPI 外部系统连接器。 + +目标地址全部来自管理员维护的可信系统目录；模型和普通用户只能传 operation_id +与结构化参数，不能传 URL。 +""" +from __future__ import annotations + +import json +import hashlib +import re +import time +from dataclasses import dataclass, field +from threading import Lock +from typing import Any, Optional +from urllib.parse import quote, urljoin, urlparse + +import httpx + +from .auth import ExternalAuthError, get_auth_strategy + + +class OpenApiError(RuntimeError): + pass + + +_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete") +_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} +_SPEC_LOCK = Lock() + +def _bool_value(value: Any, default: bool) -> bool: + raw = str(value if value is not None else "").strip().lower() + if not raw: + return default + return raw in {"1", "true", "yes", "on"} + + +def _validated_http_url(raw: str, label: str) -> str: + value = (raw or "").strip().rstrip("/") + parsed = urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise OpenApiError(f"{label} 必须是有效的 http(s) URL") + if parsed.username or parsed.password: + raise OpenApiError(f"{label} 不能内嵌凭据") + return value + + +@dataclass(frozen=True) +class OpenApiConfig: + base_url: str + openapi_url: str + login_path: str + allowed_post_operations: frozenset[str] + timeout_seconds: float + max_result_bytes: int + max_total_result_bytes: int + max_page_size: int + verify_tls: bool + query_guidance: str + recommended_operation_ids: tuple[str, ...] + auth_type: str = "password_jwt" + auth_config: dict[str, Any] = field(default_factory=lambda: { + "login_path": "/api/auth/token/", + "username_field": "username", + "password_field": "password", + "token_field": "access", + "auth_header_name": "Authorization", + "auth_header_template": "Bearer {token}", + }) + + @classmethod + def from_mapping(cls, data: dict[str, Any]) -> "OpenApiConfig": + """从管理员保存的可信目录配置构建运行态配置。""" + base = _validated_http_url(str(data.get("base_url") or ""), "base_url") + spec = _validated_http_url( + str(data.get("openapi_url") or ""), "openapi_url" + ) + login_path = str(data.get("login_path") or "/api/auth/token/").strip() + if not login_path.startswith("/") or "://" in login_path: + raise OpenApiError("login_path 必须是站内绝对路径") + raw_allowed = data.get("allowed_post_operations") or [] + if isinstance(raw_allowed, str): + raw_allowed = raw_allowed.split(",") + if not isinstance(raw_allowed, (list, tuple, set)): + raise OpenApiError("allowed_post_operations 必须是字符串数组") + allowed = frozenset(str(item).strip() for item in raw_allowed if str(item).strip()) + guidance = str(data.get("query_guidance") or "").strip() + if len(guidance) > 4000: + raise OpenApiError("query_guidance 不能超过 4000 字符") + raw_recommended = data.get( + "recommended_operation_ids", [] + ) + if isinstance(raw_recommended, str): + raw_recommended = raw_recommended.split(",") + if not isinstance(raw_recommended, (list, tuple, set)): + raise OpenApiError("recommended_operation_ids 必须是字符串数组") + recommended = tuple(dict.fromkeys( + str(item).strip() for item in raw_recommended if str(item).strip() + )) + if len(recommended) > 30 or any(len(item) > 200 for item in recommended): + raise OpenApiError("recommended_operation_ids 最多 30 项且每项不超过 200 字符") + max_result = max(4096, min(int(data.get("max_result_bytes", 65536)), 1048576)) + return cls( + base_url=base, + openapi_url=spec, + login_path=login_path, + allowed_post_operations=allowed, + timeout_seconds=max(1.0, min(float(data.get("timeout_seconds", 15)), 60.0)), + max_result_bytes=max_result, + max_total_result_bytes=max( + max_result, + min(int(data.get("max_total_result_bytes", 262144)), 4194304), + ), + max_page_size=max(1, min(int(data.get("max_page_size", 200)), 1000)), + verify_tls=_bool_value(data.get("verify_tls"), True), + query_guidance=guidance, + recommended_operation_ids=recommended, + auth_type=str(data.get("auth_type") or "password_jwt").strip(), + auth_config={ + key: data[key] + for key in ( + "login_path", "username_field", "password_field", "token_field", + "auth_header_name", "auth_header_template", + ) + if key in data + }, + ) + + +class OpenApiClient: + def __init__( + self, + credentials: dict[str, str], + cfg: OpenApiConfig, + *, + cache_namespace: str = "", + ): + self.credentials = credentials + self.cfg = cfg + identity = cache_namespace or json.dumps(credentials, sort_keys=True, ensure_ascii=False) + digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() + self._spec_cache_key = f"{cfg.openapi_url}:{digest}" + + def _client(self) -> httpx.Client: + return httpx.Client( + timeout=self.cfg.timeout_seconds, + verify=self.cfg.verify_tls, + follow_redirects=False, + ) + + def authenticate(self) -> dict[str, str]: + try: + with self._client() as client: + return get_auth_strategy(self.cfg.auth_type).headers( + client=client, + base_url=self.cfg.base_url, + credentials=self.credentials, + config=self.cfg.auth_config, + ) + except ExternalAuthError as exc: + raise OpenApiError(str(exc)) from exc + + def _fetch_spec(self, headers: dict[str, str]) -> dict[str, Any]: + now = time.monotonic() + with _SPEC_LOCK: + hit = _SPEC_CACHE.get(self._spec_cache_key) + if hit and now - hit[0] < 300: + return hit[1] + try: + with self._client() as client: + response = client.get( + self.cfg.openapi_url, + headers=headers, + ) + except httpx.HTTPError as exc: + raise OpenApiError(f"OpenAPI 获取失败: {type(exc).__name__}") from exc + if response.status_code >= 400: + raise OpenApiError(f"OpenAPI 获取失败(HTTP {response.status_code})") + try: + spec = response.json() + except ValueError as exc: + raise OpenApiError("OpenAPI 文档不是有效 JSON") from exc + if not isinstance(spec, dict) or not isinstance(spec.get("paths"), dict): + raise OpenApiError("OpenAPI 文档缺少 paths") + with _SPEC_LOCK: + _SPEC_CACHE[self._spec_cache_key] = (now, spec) + return spec + + @staticmethod + def _operation_id(method: str, path: str, operation: dict[str, Any]) -> str: + explicit = operation.get("operationId") + if isinstance(explicit, str) and explicit.strip(): + return explicit.strip() + safe_path = re.sub(r"[^a-zA-Z0-9]+", "_", path).strip("_") + return f"{method}_{safe_path}" + + @classmethod + def _operations(cls, spec: dict[str, Any]) -> list[dict[str, Any]]: + results: list[dict[str, Any]] = [] + for path, path_item in (spec.get("paths") or {}).items(): + if not isinstance(path_item, dict): + continue + common = path_item.get("parameters") or [] + for method in _HTTP_METHODS: + operation = path_item.get(method) + if not isinstance(operation, dict): + continue + params = list(common) + list(operation.get("parameters") or []) + results.append({ + "operation_id": cls._operation_id(method, path, operation), + "method": method.upper(), + "path": path, + "summary": operation.get("summary") or "", + "description": operation.get("description") or "", + "tags": operation.get("tags") or [], + "parameters": params, + "request_body": operation.get("requestBody"), + }) + return results + + def _spec_base_path(self, spec: dict[str, Any]) -> str: + """Return the API path prefix declared by Swagger 2 / OpenAPI 3. + + The remote specification may describe a different host, but an external + system definition is the only authority allowed to choose the target + origin. OpenAPI ``servers`` therefore contributes only a same-origin + path prefix. + """ + raw_base_path = spec.get("basePath") + if raw_base_path is not None: + if not isinstance(raw_base_path, str) or not raw_base_path.startswith("/"): + raise OpenApiError("Swagger basePath 必须是站内绝对路径") + parsed = urlparse(raw_base_path) + if parsed.netloc or parsed.query or parsed.fragment or "://" in raw_base_path: + raise OpenApiError("Swagger basePath 非法") + return parsed.path.rstrip("/") + + servers = spec.get("servers") + if not isinstance(servers, list) or not servers: + return "" + server = servers[0] + raw_url = server.get("url") if isinstance(server, dict) else None + if not isinstance(raw_url, str) or not raw_url.strip(): + raise OpenApiError("OpenAPI server URL 无效") + raw_url = raw_url.strip() + if "{" in raw_url or "}" in raw_url: + raise OpenApiError("OpenAPI server URL 包含未解析变量") + declared = urlparse(raw_url) + base = urlparse(self.cfg.base_url) + if declared.netloc and (declared.scheme, declared.netloc) != ( + base.scheme, + base.netloc, + ): + raise OpenApiError("OpenAPI server 越出 Factory MES 主机") + if declared.query or declared.fragment: + raise OpenApiError("OpenAPI server URL 不能包含查询或片段") + return ("/" + declared.path.lstrip("/")).rstrip("/") + + def _operation_url(self, spec: dict[str, Any], operation_path: str) -> str: + prefix = self._spec_base_path(spec) + base = urlparse(self.cfg.base_url) + configured_path = base.path.rstrip("/") + path = operation_path + configured_has_prefix = bool(prefix) and ( + configured_path == prefix or configured_path.endswith(prefix) + ) + operation_has_prefix = bool(prefix) and ( + path == prefix or path.startswith(prefix + "/") + ) + if configured_has_prefix and operation_has_prefix: + path = path[len(prefix):] or "/" + elif prefix and not configured_has_prefix and not operation_has_prefix: + path = prefix + "/" + path.lstrip("/") + combined_path = "/".join( + part.strip("/") for part in (configured_path, path) if part.strip("/") + ) + if operation_path.endswith("/") and combined_path: + combined_path += "/" + url = urljoin(f"{base.scheme}://{base.netloc}/", combined_path) + call_origin = urlparse(url) + if (call_origin.scheme, call_origin.netloc) != (base.scheme, base.netloc): + raise OpenApiError("接口目标越出管理员配置的主机") + return url + + def test_connection(self) -> dict[str, Any]: + headers = self.authenticate() + spec = self._fetch_spec(headers) + return {"operation_count": len(self._operations(spec))} + + def search(self, query: str, limit: int = 12) -> list[dict[str, Any]]: + query = (query or "").strip().lower() + if not query: + raise OpenApiError("query 不能为空") + headers = self.authenticate() + spec = self._fetch_spec(headers) + terms = list(dict.fromkeys( + [query] + [x for x in re.split(r"[\s,，。/]+", query) if len(x) >= 2] + )) + recommended_order = { + operation_id: index + for index, operation_id in enumerate(self.cfg.recommended_operation_ids) + } + scored: list[tuple[int, int, dict[str, Any]]] = [] + for op in self._operations(spec): + method = op["method"].lower() + if method not in {"get", "head"} and not ( + method == "post" + and op["operation_id"] in self.cfg.allowed_post_operations + ): + continue + hay = " ".join([ + op["operation_id"], op["path"], op["summary"], op["description"], + " ".join(str(x) for x in op["tags"]), + ]).lower() + score = sum(5 if term == query and term in hay else 1 for term in terms if term in hay) + recommended = op["operation_id"] in recommended_order + if score or recommended: + compact = dict(op) + compact["parameters"] = [ + { + "name": p.get("name"), + "in": p.get("in"), + "required": bool(p.get("required")), + "type": p.get("type") or (p.get("schema") or {}).get("type"), + "description": p.get("description") or "", + } + for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p + ] + compact.pop("request_body", None) + compact["recommended"] = recommended + scored.append(( + 0 if recommended else 1, + recommended_order.get(op["operation_id"], -score), + compact, + )) + scored.sort(key=lambda item: (item[0], item[1], item[2]["operation_id"])) + return [item[2] for item in scored[: max(1, min(int(limit), 30))]] + + def call( + self, + operation_id: str, + arguments: Optional[dict[str, Any]] = None, + body: Any = None, + ) -> dict[str, Any]: + headers = self.authenticate() + spec = self._fetch_spec(headers) + matches = [op for op in self._operations(spec) if op["operation_id"] == operation_id] + if len(matches) != 1: + raise OpenApiError("operation_id 不存在或不唯一，请先搜索接口") + op = matches[0] + if not op["path"].startswith("/") or "://" in op["path"]: + raise OpenApiError("OpenAPI operation path 非法") + method = op["method"].lower() + if method not in {"get", "head"} and not ( + method == "post" and operation_id in self.cfg.allowed_post_operations + ): + raise OpenApiError(f"operation {operation_id} 未列入只读调用范围") + + supplied = dict(arguments or {}) + path = op["path"] + query: dict[str, Any] = {} + request_body = body + for param in op["parameters"]: + if not isinstance(param, dict) or "$ref" in param: + continue + name = param.get("name") + location = param.get("in") + if not isinstance(name, str): + continue + # Swagger 2 的 body 参数既可按搜索结果中的参数名放在 arguments， + # 也可使用元工具独立的 body 字段；两者只取一个。 + present = name in supplied or (location == "body" and request_body is not None) + if param.get("required") and not present: + raise OpenApiError(f"缺少必填参数: {name}") + if name not in supplied: + continue + value = supplied.pop(name) + if location == "path": + path = path.replace("{" + name + "}", quote(str(value), safe="")) + elif location == "query": + if name == "page_size": + try: + value = max(1, min(int(value), self.cfg.max_page_size)) + except (TypeError, ValueError) as exc: + raise OpenApiError("page_size 必须是整数") from exc + elif name == "page" and str(value).strip() == "0": + raise OpenApiError( + "外部系统查询不允许 page=0 关闭分页，请使用 dataset 或分页查看明细" + ) + elif name == "pageoff" and _bool_value(value, False): + raise OpenApiError( + "外部系统查询不允许关闭分页，请使用 dataset 或分页查看明细" + ) + query[name] = value + elif location == "body" and request_body is None: + request_body = value + if supplied: + raise OpenApiError("存在接口定义之外的参数: " + ", ".join(sorted(supplied))) + if "{" in path or "}" in path: + raise OpenApiError("路径参数未完整提供") + + url = self._operation_url(spec, path) + try: + with self._client() as client: + response = client.request( + method.upper(), + url, + params=query, + json=request_body if method == "post" else None, + headers=headers, + ) + except httpx.HTTPError as exc: + raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc + if response.status_code >= 400: + raise OpenApiError(f"外部系统接口返回 HTTP {response.status_code}") + content_type = response.headers.get("content-type", "") + try: + payload: Any = response.json() if "json" in content_type else response.text + except ValueError: + payload = response.text + encoded = json.dumps(payload, ensure_ascii=False, default=str) + truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes + if truncated: + encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore") + payload = encoded + return { + "operation_id": operation_id, + "status_code": response.status_code, + "truncated": truncated, + "data": payload, + } diff --git a/core/external_systems/registry.py b/core/external_systems/registry.py new file mode 100644 index 0000000..bf6221d --- /dev/null +++ b/core/external_systems/registry.py @@ -0,0 +1,97 @@ +"""外部系统 provider 注册表。 + +标准 OpenAPI 系统通过数据库配置接入；只有非 OpenAPI 协议才需要新增 connector 文件。 +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .auth import ExternalAuthError, get_auth_strategy + + +FACTORY_QUERY_GUIDANCE = ( + "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询，" + "统一先调用 BI dataset list，再执行匹配的数据集。日志和业务明细列表用于" + "用户明确要求查看逐条记录、编号或追溯过程的场景。未匹配到 dataset 时，" + "先限定范围或向用户确认明细查询需求。" +) +FACTORY_RECOMMENDED_OPERATIONS = ("bi_dataset_list", "bi_dataset_exec") + + +@dataclass(frozen=True) +class ProviderSpec: + provider: str + title: str + connector: str + default_auth_type: str + allowed_auth_types: tuple[str, ...] + defaults: dict[str, Any] + + +_PROVIDERS = { + "factory_mes": ProviderSpec( + provider="factory_mes", + title="Factory MES", + connector="openapi", + default_auth_type="password_jwt", + allowed_auth_types=("password_jwt",), + defaults={ + "login_path": "/api/auth/token/", + "username_field": "username", + "password_field": "password", + "token_field": "access", + "auth_header_name": "Authorization", + "auth_header_template": "Bearer {token}", + "query_guidance": FACTORY_QUERY_GUIDANCE, + "recommended_operation_ids": list(FACTORY_RECOMMENDED_OPERATIONS), + }, + ), + "generic_openapi": ProviderSpec( + provider="generic_openapi", + title="通用 OpenAPI 系统", + connector="openapi", + default_auth_type="password_jwt", + allowed_auth_types=("password_jwt", "api_key", "bearer_token"), + defaults={ + "login_path": "/api/auth/token/", + "username_field": "username", + "password_field": "password", + "token_field": "access", + "auth_header_name": "Authorization", + "auth_header_template": "Bearer {token}", + "query_guidance": "", + "recommended_operation_ids": [], + }, + ), +} + + +def get_provider(provider: str) -> ProviderSpec: + result = _PROVIDERS.get((provider or "").strip()) + if result is None: + raise ValueError(f"不支持的外部系统 provider: {provider}") + return result + + +def provider_specs() -> tuple[ProviderSpec, ...]: + return tuple(_PROVIDERS.values()) + + +def merged_config(provider: str, config: dict[str, Any]) -> dict[str, Any]: + spec = get_provider(provider) + result = {**spec.defaults, **(config or {})} + auth_type = str(result.get("auth_type") or spec.default_auth_type).strip() + if auth_type not in spec.allowed_auth_types: + raise ValueError(f"{spec.title} 不支持认证方式 {auth_type}") + result["auth_type"] = auth_type + return result + + +def credential_fields(provider: str, config: dict[str, Any]) -> list[dict[str, Any]]: + merged = merged_config(provider, config) + try: + strategy = get_auth_strategy(merged["auth_type"]) + except ExternalAuthError as exc: + raise ValueError(str(exc)) from exc + return [field.as_dict() for field in strategy.credential_fields] diff --git a/core/external_systems/service.py b/core/external_systems/service.py index 08257b1..9945313 100644 --- a/core/external_systems/service.py +++ b/core/external_systems/service.py @@ -14,22 +14,23 @@ from core.storage.models import ExternalSystem, ExternalSystemDefinition, User from .crypto import configured as crypto_configured from .crypto import decrypt_secret, encrypt_secret, mask_username -from .factory import FactoryMesClient, FactoryMesConfig, FactoryMesError +from .openapi import OpenApiClient, OpenApiConfig, OpenApiError +from .registry import credential_fields, get_provider, merged_config, provider_specs class ExternalSystemError(RuntimeError): pass -def _factory_config(data: dict[str, Any]) -> FactoryMesConfig: +def _runtime_config(provider: str, data: dict[str, Any]) -> OpenApiConfig: try: - return FactoryMesConfig.from_mapping(data) - except (FactoryMesError, TypeError, ValueError) as exc: + return OpenApiConfig.from_mapping(merged_config(provider, data)) + except (OpenApiError, TypeError, ValueError) as exc: raise ExternalSystemError(str(exc)) from exc -def _normalized_config(data: dict[str, Any]) -> dict[str, Any]: - cfg = _factory_config(data) +def _normalized_config(provider: str, data: dict[str, Any]) -> dict[str, Any]: + cfg = _runtime_config(provider, data) return { "base_url": cfg.base_url, "openapi_url": cfg.openapi_url, @@ -42,6 +43,8 @@ def _normalized_config(data: dict[str, Any]) -> dict[str, Any]: "verify_tls": cfg.verify_tls, "query_guidance": cfg.query_guidance, "recommended_operation_ids": list(cfg.recommended_operation_ids), + "auth_type": cfg.auth_type, + **cfg.auth_config, } @@ -50,12 +53,15 @@ def _definition_view(row: ExternalSystemDefinition, *, include_config: bool) -> result = { "definition_id": str(row.definition_id), "provider": row.provider, + "provider_title": get_provider(row.provider).title, + "connector": get_provider(row.provider).connector, "name": row.name, "enabled": row.enabled, "access_mode": row.access_mode, "host": urlparse(str(config.get("base_url") or "")).hostname or "", "created_at": row.created_at.isoformat() if row.created_at else None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, + "credential_fields": credential_fields(row.provider, config), } if include_config: result["config"] = config @@ -106,7 +112,7 @@ def _sync_selected_users( user_id=uid, definition_id=definition.definition_id, provider=definition.provider, - connector="openapi", + connector=get_provider(definition.provider).connector, name=definition.name, credentials={}, config={}, @@ -120,7 +126,6 @@ def provider_catalog(user_id: UUID) -> list[dict[str, Any]]: rows = s.execute( select(ExternalSystemDefinition) .where( - ExternalSystemDefinition.provider == "factory_mes", ExternalSystemDefinition.enabled.is_(True), or_( ExternalSystemDefinition.access_mode == "all", @@ -135,18 +140,27 @@ def provider_catalog(user_id: UUID) -> list[dict[str, Any]]: ) .order_by(ExternalSystemDefinition.name) ).scalars().all() - definitions = [_definition_view(row, include_config=False) for row in rows] + definitions_by_provider: dict[str, list[dict[str, Any]]] = {} + for row in rows: + definitions_by_provider.setdefault(row.provider, []).append( + _definition_view(row, include_config=False) + ) except Exception: - definitions = [] + definitions_by_provider = {} key_ok = crypto_configured() - return [{ - "provider": "factory_mes", - "title": "Factory MES", - "connector": "openapi", - "configured": bool(definitions and key_ok), - "reason": "" if key_ok else "ZCBOT_CREDENTIAL_MASTER_KEY 未配置或少于 32 字符", - "definitions": definitions, - }] + return [ + { + "provider": spec.provider, + "title": spec.title, + "connector": spec.connector, + "default_auth_type": spec.default_auth_type, + "allowed_auth_types": list(spec.allowed_auth_types), + "configured": bool(definitions_by_provider.get(spec.provider) and key_ok), + "reason": "" if key_ok else "ZCBOT_CREDENTIAL_MASTER_KEY 未配置或少于 32 字符", + "definitions": definitions_by_provider.get(spec.provider, []), + } + for spec in provider_specs() + ] def list_external_system_definitions() -> list[dict[str, Any]]: @@ -174,14 +188,16 @@ def create_external_system_definition( ) -> dict[str, Any]: provider = (provider or "").strip() name = (name or "").strip() - if provider != "factory_mes": - raise ExternalSystemError("首版只支持 factory_mes") + try: + get_provider(provider) + except ValueError as exc: + raise ExternalSystemError(str(exc)) from exc if not name or len(name) > 80: raise ExternalSystemError("系统名称不能为空且不能超过 80 字符") row = ExternalSystemDefinition( provider=provider, name=name, - config=_normalized_config(config), + config=_normalized_config(provider, config), enabled=bool(enabled), access_mode=_validate_access_mode(access_mode), created_by=admin_user_id, @@ -222,7 +238,7 @@ def update_external_system_definition( if row is None: raise ExternalSystemError("external system definition not found") row.name = name - row.config = _normalized_config(config) + row.config = _normalized_config(row.provider, config) row.enabled = bool(enabled) row.access_mode = _validate_access_mode(access_mode) if row.access_mode == "selected": @@ -286,48 +302,73 @@ def get_definition_for_user(user_id: UUID, definition_id: UUID) -> ExternalSyste def _client( - provider: str, username: str, password: str, config: dict[str, Any] -) -> FactoryMesClient: - if provider != "factory_mes": - raise ExternalSystemError(f"unsupported external system provider: {provider}") - return FactoryMesClient(username, password, _factory_config(config)) + provider: str, + credentials: dict[str, str], + config: dict[str, Any], + *, + cache_namespace: str = "", +) -> OpenApiClient: + spec = get_provider(provider) + if spec.connector != "openapi": + raise ExternalSystemError(f"unsupported external system connector: {spec.connector}") + return OpenApiClient( + credentials, + _runtime_config(provider, config), + cache_namespace=cache_namespace, + ) -def _credentials(username: str, password: str) -> dict[str, str]: - username = (username or "").strip() - if not username or not password: - raise ExternalSystemError("用户名和密码不能为空") +def _credential_values( + provider: str, config: dict[str, Any], credentials: dict[str, str] +) -> dict[str, str]: + fields = credential_fields(provider, config) + normalized = { + field["name"]: str(credentials.get(field["name"]) or "").strip() + for field in fields + } + missing = [field["label"] for field in fields if not normalized[field["name"]]] + if missing: + raise ExternalSystemError("请填写" + "、".join(missing)) + return normalized + + +def _credentials( + provider: str, config: dict[str, Any], credentials: dict[str, str] +) -> dict[str, str]: + normalized = _credential_values(provider, config, credentials) try: - return {"username": encrypt_secret(username), "password": encrypt_secret(password)} + return {name: encrypt_secret(value) for name, value in normalized.items()} except (RuntimeError, ValueError) as exc: raise ExternalSystemError(str(exc)) from exc -def credentials_for(row: ExternalSystem) -> tuple[str, str]: +def credentials_for(row: ExternalSystem) -> dict[str, str]: try: - return ( - decrypt_secret(row.credentials["username"]), - decrypt_secret(row.credentials["password"]), - ) - except (KeyError, RuntimeError) as exc: + return {name: decrypt_secret(value) for name, value in row.credentials.items()} + except (AttributeError, RuntimeError) as exc: raise ExternalSystemError(str(exc)) from exc -def client_for_external_system(row: ExternalSystem) -> FactoryMesClient: +def client_for_external_system(row: ExternalSystem) -> OpenApiClient: definition = get_definition_for_user(row.user_id, row.definition_id) - username, password = credentials_for(row) - return _client(definition.provider, username, password, definition.config or {}) + return _client( + definition.provider, + credentials_for(row), + definition.config or {}, + cache_namespace=f"{definition.definition_id}:{row.user_id}", + ) def _view(row: ExternalSystem, definition: ExternalSystemDefinition) -> dict[str, Any]: try: - username, _ = credentials_for(row) - masked = mask_username(username) + credentials = credentials_for(row) + identity = credentials.get("username") or next(iter(credentials.values())) + masked = mask_username(identity) if credentials.get("username") else "***" credential_ok = True - except ExternalSystemError: + except (ExternalSystemError, StopIteration): masked = "***" credential_ok = False - runtime_config = _factory_config(definition.config or {}) + runtime_config = _runtime_config(definition.provider, definition.config or {}) return { "external_system_id": str(row.external_system_id), "definition_id": str(row.definition_id), @@ -338,6 +379,7 @@ def _view(row: ExternalSystem, definition: ExternalSystemDefinition) -> dict[str "status": row.status if definition.enabled else "disabled", "username_masked": masked, "credential_configured": credential_ok, + "credential_fields": credential_fields(definition.provider, definition.config or {}), "query_guidance": runtime_config.query_guidance, "recommended_operation_ids": list(runtime_config.recommended_operation_ids), "last_verified_at": row.last_verified_at.isoformat() if row.last_verified_at else None, @@ -381,8 +423,9 @@ def create_external_system( *, definition_id: UUID, name: str, - username: str, - password: str, + credentials: Optional[dict[str, str]] = None, + username: str = "", + password: str = "", ) -> dict[str, Any]: if not crypto_configured(): raise ExternalSystemError("ZCBOT_CREDENTIAL_MASTER_KEY 未配置或少于 32 字符") @@ -390,9 +433,19 @@ def create_external_system( name = (name or definition.name).strip() if not name or len(name) > 80: raise ExternalSystemError("连接名称不能为空且不能超过 80 字符") + plain = _credential_values( + definition.provider, + definition.config or {}, + credentials or {"username": username, "password": password}, + ) try: - probe = _client(definition.provider, username.strip(), password, definition.config).test_connection() - except FactoryMesError as exc: + probe = _client( + definition.provider, + plain, + definition.config, + cache_namespace=f"{definition.definition_id}:{user_id}", + ).test_connection() + except OpenApiError as exc: raise ExternalSystemError(str(exc)) from exc try: with session_scope() as s: @@ -403,34 +456,49 @@ def create_external_system( ) ).scalar_one_or_none() if row is not None and row.status != "pending": - raise ExternalSystemError("该 MES 已连接，请使用更新凭据") + raise ExternalSystemError("该外部系统已连接，请使用更新凭据") if row is None: row = ExternalSystem( user_id=user_id, definition_id=definition.definition_id, provider=definition.provider, - connector="openapi", + connector=get_provider(definition.provider).connector, ) s.add(row) row.name = name - row.credentials = _credentials(username, password) + row.credentials = _credentials(definition.provider, definition.config or {}, plain) row.config = {"operation_count": probe.get("operation_count", 0)} row.status = "active" row.last_verified_at = datetime.now(timezone.utc) s.flush() return _view(row, definition) except IntegrityError as exc: - raise ExternalSystemError("同名 MES 连接已存在") from exc + raise ExternalSystemError("同名外部系统连接已存在") from exc def update_external_system_credentials( - user_id: UUID, system_id: UUID, *, username: str, password: str + user_id: UUID, + system_id: UUID, + *, + credentials: Optional[dict[str, str]] = None, + username: str = "", + password: str = "", ) -> dict[str, Any]: row = get_external_system(user_id, system_id) definition = get_definition(row.definition_id, enabled_only=True) + plain = _credential_values( + definition.provider, + definition.config or {}, + credentials or {"username": username, "password": password}, + ) try: - probe = _client(definition.provider, username.strip(), password, definition.config).test_connection() - except FactoryMesError as exc: + probe = _client( + definition.provider, + plain, + definition.config, + cache_namespace=f"{definition.definition_id}:{user_id}", + ).test_connection() + except OpenApiError as exc: raise ExternalSystemError(str(exc)) from exc with session_scope() as s: current = s.execute( @@ -439,7 +507,7 @@ def update_external_system_credentials( ExternalSystem.user_id == user_id, ) ).scalar_one() - current.credentials = _credentials(username, password) + current.credentials = _credentials(definition.provider, definition.config or {}, plain) current.config = {**(current.config or {}), "operation_count": probe.get("operation_count", 0)} current.status = "active" current.last_verified_at = datetime.now(timezone.utc) @@ -453,7 +521,7 @@ def test_external_system(user_id: UUID, system_id: UUID) -> dict[str, Any]: try: probe = client_for_external_system(row).test_connection() ok = True - except (ExternalSystemError, FactoryMesError) as exc: + except (ExternalSystemError, OpenApiError) as exc: error = str(exc) with session_scope() as s: current = s.execute( diff --git a/tests/test_external_systems.py b/tests/test_external_systems.py index 1ae383e..f017758 100644 --- a/tests/test_external_systems.py +++ b/tests/test_external_systems.py @@ -164,6 +164,46 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): "openapi_url": "https://factory.invalid/swagger.json", }) + def test_generic_api_key_auth_uses_declared_header_without_login(self): + from core.external_systems.openapi import OpenApiClient, OpenApiConfig, _SPEC_CACHE + from core.external_systems.registry import credential_fields, merged_config + + _SPEC_CACHE.clear() + config = merged_config("generic_openapi", { + "base_url": "https://erp.invalid", + "openapi_url": "https://erp.invalid/openapi.json", + "auth_type": "api_key", + "auth_header_name": "X-ERP-Key", + "auth_header_template": "Key {token}", + }) + client = OpenApiClient({"api_key": "private-key"}, OpenApiConfig.from_mapping(config)) + http = _Http() + with patch.object(client, "_client", return_value=http): + result = client.test_connection() + self.assertGreater(result["operation_count"], 0) + self.assertFalse(any(call[0] == "POST" for call in http.calls)) + get_call = next(call for call in http.calls if call[0] == "GET") + self.assertEqual(get_call[2]["headers"], {"X-ERP-Key": "Key private-key"}) + self.assertEqual(credential_fields("generic_openapi", config)[0]["name"], "api_key") + + def test_openapi_spec_cache_is_isolated_by_connection_namespace(self): + from core.external_systems.openapi import OpenApiClient, OpenApiConfig, _SPEC_CACHE + from core.external_systems.registry import merged_config + + _SPEC_CACHE.clear() + config = OpenApiConfig.from_mapping(merged_config("generic_openapi", { + "base_url": "https://erp.invalid", + "openapi_url": "https://erp.invalid/openapi.json", + "auth_type": "bearer_token", + })) + for namespace in ("definition:user-a", "definition:user-b"): + client = OpenApiClient({"token": namespace}, config, cache_namespace=namespace) + http = _Http() + with patch.object(client, "_client", return_value=http): + client.test_connection() + self.assertTrue(any(call[0] == "GET" for call in http.calls)) + self.assertEqual(len(_SPEC_CACHE), 2) + def test_search_discovers_operation_without_exposing_credentials(self): from core.external_systems.factory import FactoryMesClient diff --git a/tests/test_web_routes_nodb.py b/tests/test_web_routes_nodb.py index 191bae8..4748c17 100644 --- a/tests/test_web_routes_nodb.py +++ b/tests/test_web_routes_nodb.py @@ -167,6 +167,22 @@ class ExternalSystemRoutesTests(unittest.TestCase): self.assertEqual(create.call_args.args[0], _UID) self.assertEqual(create.call_args.kwargs["definition_id"], definition_id) + def test_create_accepts_dynamic_credentials(self): + created = {"external_system_id": str(uuid.uuid4()), "username_masked": "***"} + definition_id = uuid.uuid4() + with patch("web.routers.external_systems.create_external_system", return_value=created) as create: + response = _client.post( + "/v1/external-systems", + headers=_AUTH, + json={ + "definition_id": str(definition_id), + "name": "LIMS", + "credentials": {"api_key": "secret-key"}, + }, + ) + self.assertEqual(response.status_code, 201) + self.assertEqual(create.call_args.kwargs["credentials"], {"api_key": "secret-key"}) + def test_invalid_connection_id_is_not_forwarded_to_service(self): with patch("web.routers.external_systems.test_external_system") as test_connection: r = _client.post("/v1/external-systems/not-a-uuid/test", headers=_AUTH) diff --git a/web/admin.py b/web/admin.py index f63a708..f579625 100644 --- a/web/admin.py +++ b/web/admin.py @@ -188,6 +188,12 @@ class ExternalSystemDefinitionRequest(BaseModel): base_url: str openapi_url: str login_path: str = "/api/auth/token/" + auth_type: str = "password_jwt" + username_field: str = "username" + password_field: str = "password" + token_field: str = "access" + auth_header_name: str = "Authorization" + auth_header_template: str = "Bearer {token}" allowed_post_operations: list[str] = Field(default_factory=list) timeout_seconds: float = 15 max_result_bytes: int = 65536 @@ -208,6 +214,12 @@ def _external_definition_config(body: ExternalSystemDefinitionRequest) -> dict[s "base_url": body.base_url, "openapi_url": body.openapi_url, "login_path": body.login_path, + "auth_type": body.auth_type, + "username_field": body.username_field, + "password_field": body.password_field, + "token_field": body.token_field, + "auth_header_name": body.auth_header_name, + "auth_header_template": body.auth_header_template, "allowed_post_operations": body.allowed_post_operations, "timeout_seconds": body.timeout_seconds, "max_result_bytes": body.max_result_bytes, diff --git a/web/routers/external_systems.py b/web/routers/external_systems.py index 8a029fd..7532244 100644 --- a/web/routers/external_systems.py +++ b/web/routers/external_systems.py @@ -54,6 +54,7 @@ def register_external_system_routes(app, *, require_user) -> None: user_id, definition_id=body.definition_id, name=body.name, + credentials=body.credentials or None, username=body.username, password=body.password, ) @@ -70,6 +71,7 @@ def register_external_system_routes(app, *, require_user) -> None: return update_external_system_credentials( user_id, _uuid(system_id), + credentials=body.credentials or None, username=body.username, password=body.password, ) diff --git a/web/schemas.py b/web/schemas.py index 427e306..45d970d 100644 --- a/web/schemas.py +++ b/web/schemas.py @@ -4,7 +4,7 @@ from __future__ import annotations from typing import Optional from uuid import UUID -from pydantic import BaseModel +from pydantic import BaseModel, Field class TaskCreateRequest(BaseModel): @@ -97,10 +97,12 @@ class KbCreateRequest(BaseModel): class ExternalSystemCreateRequest(BaseModel): definition_id: UUID name: str = "" - username: str - password: str + credentials: dict[str, str] = Field(default_factory=dict) + username: str = "" # deprecated: 兼容旧版 Factory 客户端 + password: str = "" # deprecated: 兼容旧版 Factory 客户端 class ExternalSystemCredentialsRequest(BaseModel): - username: str - password: str + credentials: dict[str, str] = Field(default_factory=dict) + username: str = "" # deprecated + password: str = "" # deprecated diff --git a/web/static/dev.html b/web/static/dev.html index 47f0d46..bb80d3c 100644 --- a/web/static/dev.html +++ b/web/static/dev.html @@ -1685,12 +1685,11 @@
加载中…
加载中…
-
连接 Factory MES
+
连接外部系统
- - - - + + +
diff --git a/web/static/js/admin.js b/web/static/js/admin.js index f91320e..9e31d88 100644 --- a/web/static/js/admin.js +++ b/web/static/js/admin.js @@ -167,7 +167,7 @@ function renderExternalDefinitions() { const rows = externalDefinitions.map(r => { const cfg = r.config || {}; return `` - + `${escapeHtml(r.name)}${r.enabled ? "" : ' 停用'}` + + `${escapeHtml(r.name)} ${escapeHtml(r.provider_title || r.provider)}${r.enabled ? "" : ' 停用'}` + ` ${r.access_mode === "all" ? "全部用户" : `指定 ${((r.selected_user_ids || []).length)} 人`}` + `${escapeHtml(r.host || cfg.base_url || "—")}` + `${(cfg.allowed_post_operations || []).length}` @@ -175,12 +175,19 @@ function renderExternalDefinitions() { + ``; }).join("") || `尚未配置外部系统`; $("s-external").innerHTML = `

外部系统目录

` - + `公共地址由管理员维护；用户只提交自己的账号密码
` + + `标准 OpenAPI 系统可直接配置；用户只提交该系统要求的凭据
` + `` + + `` + + `` + `` + `` + `` + `` + + `` + + `` + + `` + + `` + + `` + `` + `` + `