From 2b4316ed25928b822b274a0e5f8fccb7802c1d81 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Mon, 10 Aug 2026 17:26:39 +0800 Subject: [PATCH] =?UTF-8?q?refactor(external-systems)!:=20=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=E4=B8=BA=E9=80=9A=E7=94=A8=20API=20provider?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除 factory_mes preset 和兼容包装，新增存量 definition 一次性迁移脚本，并让 OpenAPI 参数直接遵循规格约束。 BREAKING CHANGE: 管理 API 不再接受 factory_mes provider，部署新版前必须先执行 definition 迁移脚本。 --- CHANGELOG.md | 5 + DESIGN.md | 6 +- PROGRESS.md | 4 +- RUN.md | 11 +- core/__init__.py | 2 +- core/external_systems/catalog.py | 37 ++++ core/external_systems/factory.py | 29 ---- core/external_systems/mcp.py | 2 - core/external_systems/openapi.py | 33 ++-- core/external_systems/registry.py | 30 ---- core/external_systems/service.py | 1 - scripts/migrate_factory_mes_definitions.py | 189 ++++++++++++++++++++ tests/test_external_system_migration.py | 62 +++++++ tests/test_external_systems.py | 190 +++++++++------------ tests/test_web_routes_nodb.py | 6 +- tools/external_systems.py | 10 +- web/admin.py | 8 +- web/static/js/admin.js | 56 +++--- 18 files changed, 434 insertions(+), 247 deletions(-) delete mode 100644 core/external_systems/factory.py create mode 100644 scripts/migrate_factory_mes_definitions.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bacf8fc..a6eb539 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,11 @@ > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## 0.63.4 — 2026-08-10 + +- 外部系统管理统一为通用 OpenAPI 和 MCP 配置，不再需要选择特定 MES 类型；已有 MES 定义可通过随版本提供的一次性脚本转换，用户连接和加密凭据保持不变。 +- OpenAPI 调用现在直接执行接口规格声明的数值、长度、数组和枚举约束，不再按分页参数名称套用特定系统规则。 + ## 0.63.3 — 2026-08-10 - PDF 和 PPT 预览恢复连续滚动阅读，无需反复点击上一页、下一页；页码跳转和缩放仍可使用，长文档也会按浏览位置逐页加载。 diff --git a/DESIGN.md b/DESIGN.md index 157fcb4..dfc04fb 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -411,9 +411,9 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) **诉求**:用户用自己的 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 停号/改权后下次调用即生效。 +**Provider 边界**:运行态只保留 `generic_openapi` 与 `generic_mcp`;具体 ERP、MES、LIMS 和 SaaS 都是数据库中的 definition,不再为单个业务系统维护 Python preset。用户名密码换 Token、API Key 和 Bearer Token 由通用认证 strategy 组合,业务查询提示、推荐 operation 和只读 POST policy 全部随 definition 保存。zcbot `user_id` 只能取自己的 connection,远端凭据再判定实际业务权限,不在 zcbot 复制上游 RBAC。 -**通用连接器边界**:`openapi` connector 负责规格发现、operation 解析、安全 URL 拼接、参数校验、执行模式、分页和响应体积限制；`mcp` connector 使用官方 MCP v2 SDK 连接管理员托管的 Streamable HTTP Server，通过 `tools/list` 动态发现、搜索并调用全部远端工具。两者共用认证 strategy、definition/grant/connection、revision、凭据加密、响应额度、大结果缓存与审计。`factory_mes` 只是 OpenAPI preset；MCP 不为 Factory 建专用 provider，Factory 与后续 ERP/LIMS/SaaS 都以 `generic_mcp` definition 接入。标准 OpenAPI/MCP 系统只新增数据库 definition，不需要新增 Python provider；只有 OAuth 回调/签名交换、SOAP、消息队列或私有二进制协议等不符合现有 connector/strategy 契约的系统才新增适配代码。 +**通用连接器边界**:`openapi` connector 负责规格发现、operation 解析、安全 URL 拼接、参数 schema 校验、执行模式和响应体积限制；参数的 `minimum/maximum/enum` 等契约直接以 Swagger/OpenAPI 为事实源,不按 `page/page_size/pageoff` 等名字维护第二套分页语义。`mcp` connector 使用官方 MCP v2 SDK 连接管理员托管的 Streamable HTTP Server，通过 `tools/list` 动态发现、搜索并调用全部远端工具。两者共用认证 strategy、definition/grant/connection、revision、凭据加密、响应额度、大结果缓存与审计。标准 OpenAPI/MCP 系统只新增数据库 definition，不需要新增 Python provider；只有 OAuth 回调/签名交换、SOAP、消息队列或私有二进制协议等不符合现有 connector/strategy 契约的系统才新增适配代码。 **信任边界**: - definition 当前由管理员维护，持久化同时预留 `owner_type/owner_user_id/visibility/trust_level/review_status/egress_policy_id`，未来可开放私有用户定义。Base URL 必须与 OpenAPI URL 或 MCP URL 同源；普通用户不能填任意 URL，避免 SSRF/内网代理。每个 definition 带单调递增 revision：目标地址、期望 MCP Server 身份或认证绑定变化保留密文但要求重新验证，未验证到当前 revision 的连接不挂调用工具。 @@ -426,7 +426,7 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) **大响应**:`max_result_bytes` 是进入模型上下文的单次内联额度,不再用于切断原始 JSON；超额响应完整写入 `.zcbot_cache//external_results/`,工具只返回合法结构化预览、`result_ref`、原始字节数和可继续读取的位置。reader 每次读取都重新校验当前 user 对原 external system 的 active 授权,并与 call 共享本轮 `max_total_result_bytes` 内联额度；export 同样重验授权，并把查询 operation/参数/时间等 provenance 与完整响应一起持久化，导出文件不受缓存 TTL 影响。缓存固定 24h TTL、单响应 10 MiB、单 task 50 MiB、单 user 200 MiB,过期或超额时优先清理最旧缓存；0.62.1 的 `.zcbot_external_results/` 在读取和容量核算上保留兼容窗口。超过响应安全上限的远端结果直接拒绝并要求缩小范围,不产生半截 JSON。这里把“上游响应安全边界”“完整结果保存”“模型上下文额度”“用户明确留存”拆成四层,既不丢数据,也不靠无限提高上下文额度解决大结果问题。 -**明细扫描边界**:单次响应保留安全上限与模型内联额度,每次 agent run 另按外部系统累计内联返回量；Factory connector 将 `page_size` 限在管理员上限,拒绝 `page=0` / `pageoff` 关闭分页。三者防模型通过连续翻日志自行做昂贵聚合,但不改变 Factory 对其他客户端的分页契约。达到边界后工具正向引导回 dataset/聚合接口、`result_ref` 分段读取或缩小查询范围。 +**明细扫描边界**:单次响应保留安全下载上限与模型内联额度,每次 agent run 另按外部系统累计内联返回量；请求参数严格执行 OpenAPI schema 声明的数值、长度、数组和枚举约束。规格没有声明的分页哨兵语义不由 zcbot 猜测,应优先修正上游规格；平台通过响应与累计额度阻止模型连续拉取大量明细。达到边界后工具正向引导回聚合接口、`result_ref` 分段读取或缩小查询范围。 **状态与 UI（三实体）**:`external_system_definitions` 保存可信目录、connector 配置、revision、治理元数据和查询提示；OpenAPI definition 另有执行模式及只读 POST policy，MCP definition 保存 URL、期望 Server 名称与传输响应上限，不保存工具清单。`external_system_grants` 只保存 selected 可见授权；`external_systems` 只保存用户连接、AAD 绑定密文、verified revision 和 `active|invalid|needs_reverify|needs_credentials` 状态。定义更新先对新旧配置做默认值补全后的语义比较：查询提示、推荐入口和响应限额等运行配置变化让 active 连接原子跟随新 revision；目标、登录、认证绑定、期望 Server 名称或 TLS 变化保留密文但置 `needs_reverify`。其余撤权、断开、密文与审计语义不变。 diff --git a/PROGRESS.md b/PROGRESS.md index 11e0cb4..fe0b8d0 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-10(PDF/PPT 连续滚动预览,bump 0.63.3) +最后更新:2026-08-10(外部系统统一为通用 OpenAPI/MCP,bump 0.63.4) --- @@ -23,6 +23,8 @@ ### 2026-08-10 +- **08-10 / 0.63.4 / 移除 Factory preset + 通用 OpenAPI 契约收敛**:运行态和管理端删除 `factory_mes` provider、兼容包装类及业务默认文案，只保留通用 OpenAPI/MCP definition；新增显式 `ZCBOT_MIGRATION_DB_URL`、默认 dry-run 的一次性脚本，将存量 definition 原子转换为 `generic_openapi + query` 并同步 active connection revision，不解密或重写凭据。OpenAPI 参数直接执行规格中的数值、长度、数组和枚举约束，移除按 `page/page_size/pageoff` 名称猜测分页语义及无效 `max_page_size` 配置。完整 524 项 unittest 全绿（17 skip），外部系统/迁移/无 DB 路由专项 80 项、Python 编译、Ruff 致命规则、JavaScript 语法及 diff 检查通过；迁移脚本仅验证缺少显式 URL 时拒绝运行，未连接或写入任何数据库。 + - **08-10 / 0.63.3 / PDF/PPT 连续滚动预览**:PDF.js 展示由上一页/下一页驱动的单页 Canvas 恢复为纵向连续页列表，保留页码跳转、适宽和缩放；页面进入视口前后才渲染，滚远后释放 Canvas，兼顾原有阅读习惯与长文档内存占用。Node 前端预览 10 项、JavaScript 语法及 diff 检查通过；无 schema、migration、HTTP API、依赖或运行方式变化，未连接生产 DB。 - **08-10 / 0.63.2 / App WebView PDF/PPT/HTML 预览兼容**:PDF 与 PPT 转换结果不再通过 blob iframe 依赖浏览器内置 PDF 插件，改为本地固化 PDF.js 后按单页 Canvas 渲染，支持翻页、适宽与缩放，并在关闭或切换预览时释放 loading/render task、worker 和文档资源。HTML 保留 opaque-origin sandbox 安全边界，内容入口由兼容不稳的 `srcdoc` 改为同源静态宿主页 + postMessage，避开原生 URL 白名单误拦 `blob:`；App 对接文档明确白名单仅处理主 frame。Node 前端 18 项、静态资源 unittest 12 项、JavaScript/PDF.js bundle 语法及 diff 检查通过；无 schema、migration、HTTP API 或 Python 依赖变化，未连接生产 DB。 diff --git a/RUN.md b/RUN.md index 3536fa9..366bea2 100644 --- a/RUN.md +++ b/RUN.md @@ -2,7 +2,7 @@ > 怎么把 zcbot 跑起来。env / 常用命令 / 故障兜底。设计看 `DESIGN.md`,进度看 `PROGRESS.md`。 -最后更新:2026-08-07(外部系统定义更新保留凭据并按语义重验) +最后更新:2026-08-10(外部系统统一为通用 OpenAPI/MCP definition) --- @@ -150,7 +150,14 @@ - **未绑定成员发消息 → 回绑定指引**(不再静默):聊天优先布局下新员工第一动作就是打字,回调对未绑定成员的 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)。 -- **OpenAPI / MCP 外部系统**:① `.env` 配置独立的 `ZCBOT_CREDENTIAL_MASTER_KEY`，可选 `ZCBOT_CREDENTIAL_KEY_ID` 标识当前密钥；轮换时把旧 key 以 JSON 对象放入 `ZCBOT_CREDENTIAL_PREVIOUS_KEYS`，待用户凭据完成重写后再移除。② 执行 `main.py db upgrade head`；MCP 复用现有三实体和 JSONB config，不新增 migration。③ admin 进入管理后台「外部系统」，可选择 Factory MES、通用 OpenAPI 或通用 MCP。MCP 填写与登录 Base URL 同源的 Streamable HTTP URL，可选填写期望 Server 名称；连接后以 `tools/list` 为事实源，Server 当前暴露的全部工具均可搜索和调用，不需要在 zcbot 重复列工具。生产部署若 Factory MCP 是独立端口，应先由反向代理统一为与 `/api/auth/token/` 同源的 `/mcp` 地址。④ 普通用户点击左栏 **「外部」**，页面按定义动态显示用户名密码、API Key 或 Bearer Token；连接测试会完成 MCP initialize 和工具发现，目标、Server 身份、登录、认证绑定或 TLS 变化后保留密文并暂停调用，重新测试成功后恢复。OpenAPI spec 和 MCP tool catalog 只在进程内按连接身份有界缓存，登录与业务响应均限长，普通用户和模型不能传任意 URL。 +- **OpenAPI / MCP 外部系统**:① `.env` 配置独立的 `ZCBOT_CREDENTIAL_MASTER_KEY`，可选 `ZCBOT_CREDENTIAL_KEY_ID` 标识当前密钥；轮换时把旧 key 以 JSON 对象放入 `ZCBOT_CREDENTIAL_PREVIOUS_KEYS`，待用户凭据完成重写后再移除。② 执行 `main.py db upgrade head`。③ admin 进入管理后台「外部系统」，选择通用 OpenAPI 或通用 MCP；具体 MES/ERP/LIMS 都作为数据库 definition 配置，不新增专用 provider。MCP 填写与登录 Base URL 同源的 Streamable HTTP URL，可选填写期望 Server 名称；连接后以 `tools/list` 为事实源。④ 普通用户点击左栏 **「外部」**，页面按 definition 动态显示用户名密码、API Key 或 Bearer Token；目标、Server 身份、登录、认证绑定或 TLS 变化后保留密文并暂停调用，重新测试成功后恢复。OpenAPI spec 和 MCP tool catalog 只在进程内按连接身份有界缓存，登录与业务响应均限长，普通用户和模型不能传任意 URL。 + - **旧 `factory_mes` definition 一次性转换**:新版代码不再识别 `factory_mes`；部署时保持旧服务进程运行，先从新代码目录执行数据脚本，转换成功后再重启到新版。脚本不加载 `.env`、不读取 `ZCBOT_DB_URL`，只认显式的 `ZCBOT_MIGRATION_DB_URL`；默认 dry-run，检查同名冲突与配置合法性。确认输出后加 `--apply`，脚本把 definition 转为 `generic_openapi + query`、物化 JWT/提示/只读 POST 配置并同步 active connection revision，不解密或改写用户凭据。 + ```powershell + $env:ZCBOT_MIGRATION_DB_URL="postgresql+psycopg://user:pass@host:5432/zcbot" + .venv/Scripts/python.exe scripts/migrate_factory_mes_definitions.py + .venv/Scripts/python.exe scripts/migrate_factory_mes_definitions.py --apply + ``` + 生产执行 `--apply` 前再次确认目标并先备份相关表；输出 `remaining factory_mes definitions: 0` 后再部署新版。若存在同一 owner 下的同名 `generic_openapi` definition，脚本整批回滚，先在管理后台改名后重跑。 - **测试库(可选,`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 \ diff --git a/core/__init__.py b/core/__init__.py index 407b0d9..44da8dd 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.63.3" +__version__ = "0.63.4" diff --git a/core/external_systems/catalog.py b/core/external_systems/catalog.py index 857e5f4..f98b093 100644 --- a/core/external_systems/catalog.py +++ b/core/external_systems/catalog.py @@ -111,6 +111,43 @@ def validate_json_value( raise ValueError(f"{path} 应为 {expected}") if "enum" in schema and value not in schema.get("enum", []): raise ValueError(f"{path} 不在允许值范围内") + if isinstance(value, (int, float)) and not isinstance(value, bool): + minimum = schema.get("minimum") + maximum = schema.get("maximum") + exclusive_minimum = schema.get("exclusiveMinimum") + exclusive_maximum = schema.get("exclusiveMaximum") + if isinstance(minimum, (int, float)): + if value < minimum or (exclusive_minimum is True and value == minimum): + operator = ">" if exclusive_minimum is True else ">=" + raise ValueError(f"{path} 必须 {operator} {minimum}") + if isinstance(maximum, (int, float)): + if value > maximum or (exclusive_maximum is True and value == maximum): + operator = "<" if exclusive_maximum is True else "<=" + raise ValueError(f"{path} 必须 {operator} {maximum}") + if isinstance(exclusive_minimum, (int, float)) and not isinstance( + exclusive_minimum, bool + ): + if value <= exclusive_minimum: + raise ValueError(f"{path} 必须 > {exclusive_minimum}") + if isinstance(exclusive_maximum, (int, float)) and not isinstance( + exclusive_maximum, bool + ): + if value >= exclusive_maximum: + raise ValueError(f"{path} 必须 < {exclusive_maximum}") + if isinstance(value, str): + minimum_length = schema.get("minLength") + maximum_length = schema.get("maxLength") + if isinstance(minimum_length, int) and len(value) < minimum_length: + raise ValueError(f"{path} 长度不能小于 {minimum_length}") + if isinstance(maximum_length, int) and len(value) > maximum_length: + raise ValueError(f"{path} 长度不能超过 {maximum_length}") + if isinstance(value, list): + minimum_items = schema.get("minItems") + maximum_items = schema.get("maxItems") + if isinstance(minimum_items, int) and len(value) < minimum_items: + raise ValueError(f"{path} 项数不能少于 {minimum_items}") + if isinstance(maximum_items, int) and len(value) > maximum_items: + raise ValueError(f"{path} 项数不能超过 {maximum_items}") if isinstance(value, dict): required = schema.get("required") or [] missing = [str(name) for name in required if name not in value] diff --git a/core/external_systems/factory.py b/core/external_systems/factory.py deleted file mode 100644 index 8228291..0000000 --- a/core/external_systems/factory.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Factory MES 兼容入口。 - -新代码使用 :mod:`core.external_systems.openapi`；保留原类名，避免已有测试和内部引用 -在通用化过程中发生无意义破坏。 -""" - -from __future__ import annotations - -from typing import Any - -from .openapi import OpenApiClient, OpenApiConfig, OpenApiError -from .registry import merged_config - - -FactoryMesError = OpenApiError - - -class FactoryMesConfig(OpenApiConfig): - @classmethod - def from_mapping(cls, data: dict[str, Any]) -> "FactoryMesConfig": - common = OpenApiConfig.from_mapping(merged_config("factory_mes", data)) - return cls(**common.__dict__) - - -class FactoryMesClient(OpenApiClient): - def __init__(self, username: str, password: str, cfg: FactoryMesConfig): - self.username = username - self.password = password - super().__init__({"username": username, "password": password}, cfg) diff --git a/core/external_systems/mcp.py b/core/external_systems/mcp.py index 4475d30..ec38f85 100644 --- a/core/external_systems/mcp.py +++ b/core/external_systems/mcp.py @@ -63,7 +63,6 @@ class McpConfig: 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, ...] @@ -117,7 +116,6 @@ class McpConfig: 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, diff --git a/core/external_systems/openapi.py b/core/external_systems/openapi.py index 0b7eff5..277ded2 100644 --- a/core/external_systems/openapi.py +++ b/core/external_systems/openapi.py @@ -91,7 +91,6 @@ class OpenApiConfig: 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, ...] @@ -172,7 +171,6 @@ class OpenApiConfig: 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, @@ -637,7 +635,7 @@ class OpenApiClient: base.scheme, base.netloc, ): - raise OpenApiError("OpenAPI server 越出 Factory MES 主机") + raise OpenApiError("OpenAPI server 越出管理员配置的外部系统主机") if declared.query or declared.fragment: raise OpenApiError("OpenAPI server URL 不能包含查询或片段") return ("/" + declared.path.lstrip("/")).rstrip("/") @@ -804,7 +802,21 @@ class OpenApiClient: continue value = supplied.pop(name) parameter_schema = param.get("schema") or { - key: param[key] for key in ("type", "enum", "items") if key in param + key: param[key] + for key in ( + "type", + "enum", + "items", + "minimum", + "maximum", + "exclusiveMinimum", + "exclusiveMaximum", + "minLength", + "maxLength", + "minItems", + "maxItems", + ) + if key in param } try: validate_json_value( @@ -818,19 +830,6 @@ class OpenApiClient: 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 或分页查看明细" - ) if isinstance(value, list): raw_collection_format = param.get("collectionFormat") collection_format = ( diff --git a/core/external_systems/registry.py b/core/external_systems/registry.py index f6ba622..253c477 100644 --- a/core/external_systems/registry.py +++ b/core/external_systems/registry.py @@ -10,15 +10,6 @@ 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 @@ -30,27 +21,6 @@ class ProviderSpec: _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), - "operation_mode": "upstream_managed", - "operation_policies": { - "bi_dataset_exec": "read", - }, - }, - ), "generic_openapi": ProviderSpec( provider="generic_openapi", title="通用 OpenAPI 系统", diff --git a/core/external_systems/service.py b/core/external_systems/service.py index 9d0cbc9..22bbe56 100644 --- a/core/external_systems/service.py +++ b/core/external_systems/service.py @@ -72,7 +72,6 @@ def _normalized_config(provider: str, data: dict[str, Any]) -> dict[str, Any]: "timeout_seconds": cfg.timeout_seconds, "max_result_bytes": cfg.max_result_bytes, "max_total_result_bytes": cfg.max_total_result_bytes, - "max_page_size": cfg.max_page_size, "verify_tls": cfg.verify_tls, "query_guidance": cfg.query_guidance, "recommended_operation_ids": list(cfg.recommended_operation_ids), diff --git a/scripts/migrate_factory_mes_definitions.py b/scripts/migrate_factory_mes_definitions.py new file mode 100644 index 0000000..d8a4235 --- /dev/null +++ b/scripts/migrate_factory_mes_definitions.py @@ -0,0 +1,189 @@ +"""把存量 Factory MES definition 一次性转换为通用 OpenAPI definition。 + +脚本默认只预检。数据库地址只从显式的 ``ZCBOT_MIGRATION_DB_URL`` 读取， +不会加载项目 ``.env``，也不会回退到 ``ZCBOT_DB_URL``。 +""" + +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path +from typing import Any + +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from core.external_systems.service import _normalized_config # noqa: E402 +from core.storage.models import ( # noqa: E402 + ExternalSystem, + ExternalSystemDefinition, +) + + +FACTORY_QUERY_GUIDANCE = ( + "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询，" + "统一先调用 BI dataset list，再执行匹配的数据集。日志和业务明细列表用于" + "用户明确要求查看逐条记录、编号或追溯过程的场景。未匹配到 dataset 时，" + "先限定范围或向用户确认明细查询需求。" +) + + +def migrated_config(raw: dict[str, Any] | None) -> dict[str, Any]: + """物化旧 preset，并收敛为 query 模式的通用 OpenAPI 配置。""" + source = dict(raw or {}) + policies = { + str(key).strip(): str(value).strip().lower() + for key, value in (source.get("operation_policies") or {}).items() + if str(key).strip() + } + for operation_id in source.get("allowed_post_operations") or []: + if str(operation_id).strip(): + policies.setdefault(str(operation_id).strip(), "read") + policies.setdefault("bi_dataset_exec", "read") + source.update( + { + "auth_type": source.get("auth_type") or "password_jwt", + "login_path": source.get("login_path") or "/api/auth/token/", + "username_field": source.get("username_field") or "username", + "password_field": source.get("password_field") or "password", + "token_field": source.get("token_field") or "access", + "auth_header_name": source.get("auth_header_name") or "Authorization", + "auth_header_template": source.get("auth_header_template") + or "Bearer {token}", + "operation_mode": "query", + "operation_policies": policies, + "query_guidance": source.get("query_guidance") + or FACTORY_QUERY_GUIDANCE, + "recommended_operation_ids": source.get("recommended_operation_ids") + or ["bi_dataset_list", "bi_dataset_exec"], + } + ) + source.pop("allowed_post_operations", None) + return _normalized_config("generic_openapi", source) + + +def _conflicting_definition( + session: Session, definition: ExternalSystemDefinition +) -> ExternalSystemDefinition | None: + owner_match = ( + ExternalSystemDefinition.owner_type == "platform" + if definition.owner_type == "platform" + else ExternalSystemDefinition.owner_user_id == definition.owner_user_id + ) + return session.execute( + select(ExternalSystemDefinition).where( + ExternalSystemDefinition.provider == "generic_openapi", + ExternalSystemDefinition.name == definition.name, + owner_match, + ) + ).scalar_one_or_none() + + +def migrate(session: Session, *, apply: bool) -> tuple[int, int]: + statement = ( + select(ExternalSystemDefinition) + .where(ExternalSystemDefinition.provider == "factory_mes") + .order_by(ExternalSystemDefinition.name) + ) + if apply: + statement = statement.with_for_update() + definitions = ( + session.execute(statement) + .scalars() + .all() + ) + connection_count = 0 + prepared: list[tuple[ExternalSystemDefinition, dict[str, Any], int]] = [] + for definition in definitions: + conflict = _conflicting_definition(session, definition) + if conflict is not None: + raise RuntimeError( + f"definition name conflict: {definition.name} " + f"({definition.definition_id} vs {conflict.definition_id})" + ) + config = migrated_config(definition.config) + count = len( + session.execute( + select(ExternalSystem.external_system_id).where( + ExternalSystem.definition_id == definition.definition_id + ) + ).all() + ) + connection_count += count + prepared.append((definition, config, count)) + print( + f"[INFO] {definition.definition_id} name={definition.name!r} " + f"connections={count}" + ) + + print(f"[INFO] factory_mes definitions: {len(prepared)}") + print(f"[INFO] affected connections: {connection_count}") + if not apply: + return len(prepared), connection_count + + for definition, config, _ in prepared: + definition.provider = "generic_openapi" + definition.config = config + definition.revision += 1 + active_connections = ( + session.execute( + select(ExternalSystem).where( + ExternalSystem.definition_id == definition.definition_id, + ExternalSystem.status == "active", + ) + ) + .scalars() + .all() + ) + for connection in active_connections: + connection.verified_revision = definition.revision + session.flush() + remaining = session.execute( + select(ExternalSystemDefinition.definition_id).where( + ExternalSystemDefinition.provider == "factory_mes" + ) + ).first() + if remaining is not None: + raise RuntimeError("factory_mes definitions remain after migration") + print("[OK] remaining factory_mes definitions: 0") + return len(prepared), connection_count + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--apply", + action="store_true", + help="执行写入；省略时只做预检并回滚事务", + ) + args = parser.parse_args() + database_url = os.environ.get("ZCBOT_MIGRATION_DB_URL", "").strip() + if not database_url: + print("[ERR] ZCBOT_MIGRATION_DB_URL is required", file=sys.stderr) + return 2 + engine = create_engine(database_url, pool_pre_ping=True, future=True) + try: + with Session(engine, future=True) as session: + try: + definitions, connections = migrate(session, apply=args.apply) + if args.apply: + session.commit() + else: + session.rollback() + except Exception: + session.rollback() + raise + finally: + engine.dispose() + action = "migrated" if args.apply else "validated" + print(f"[OK] {action} definitions: {definitions}") + print(f"[OK] affected connections: {connections}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_external_system_migration.py b/tests/test_external_system_migration.py index 0fa9ad4..3c2aaa3 100644 --- a/tests/test_external_system_migration.py +++ b/tests/test_external_system_migration.py @@ -33,5 +33,67 @@ class ExternalSystemMigrationTests(unittest.TestCase): self.assertIn("operation_mode", rendered) +class FactoryDefinitionDataMigrationTests(unittest.TestCase): + def test_materializes_factory_defaults_as_generic_query_config(self): + from scripts.migrate_factory_mes_definitions import migrated_config + + config = migrated_config( + { + "base_url": "https://factory.invalid", + "openapi_url": "https://factory.invalid/swagger.json", + "allowed_post_operations": ["quality_report"], + } + ) + + self.assertEqual(config["auth_type"], "password_jwt") + self.assertEqual(config["operation_mode"], "query") + self.assertEqual( + config["operation_policies"], + {"bi_dataset_exec": "read", "quality_report": "read"}, + ) + self.assertNotIn("allowed_post_operations", config) + self.assertEqual( + config["recommended_operation_ids"], + ["bi_dataset_list", "bi_dataset_exec"], + ) + + def test_explicit_factory_config_is_preserved_except_execution_mode(self): + from scripts.migrate_factory_mes_definitions import migrated_config + + config = migrated_config( + { + "base_url": "https://factory.invalid/api", + "openapi_url": "https://factory.invalid/openapi.json", + "login_path": "/login", + "token_field": "data.token", + "query_guidance": "使用质量聚合接口", + "recommended_operation_ids": ["quality_summary"], + "operation_mode": "upstream_managed", + "operation_policies": {"quality_summary": "export"}, + } + ) + + self.assertEqual(config["base_url"], "https://factory.invalid/api") + self.assertEqual(config["login_path"], "/login") + self.assertEqual(config["token_field"], "data.token") + self.assertEqual(config["query_guidance"], "使用质量聚合接口") + self.assertEqual(config["recommended_operation_ids"], ["quality_summary"]) + self.assertEqual(config["operation_mode"], "query") + self.assertEqual( + config["operation_policies"], + {"bi_dataset_exec": "read", "quality_summary": "export"}, + ) + + def test_runtime_provider_registry_has_no_factory_preset(self): + from core.external_systems.registry import get_provider, provider_specs + + self.assertEqual( + {spec.provider for spec in provider_specs()}, + {"generic_openapi", "generic_mcp"}, + ) + with self.assertRaisesRegex(ValueError, "不支持"): + get_provider("factory_mes") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_external_systems.py b/tests/test_external_systems.py index 9fa1fe4..d564628 100644 --- a/tests/test_external_systems.py +++ b/tests/test_external_systems.py @@ -17,6 +17,12 @@ from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from core.external_systems.openapi import OpenApiClient, OpenApiConfig, OpenApiError + + +def _openapi_client(username: str, password: str, cfg: OpenApiConfig) -> OpenApiClient: + return OpenApiClient({"username": username, "password": password}, cfg) + class ExternalCredentialCryptoTests(unittest.TestCase): def test_requires_master_key_and_never_falls_back_to_plaintext(self): @@ -128,7 +134,7 @@ class ExternalConnectionRevisionTests(unittest.TestCase): "auth_header_template": "Bearer {token}", } _, _, changed, impact = _classify_definition_config_change( - "factory_mes", legacy, materialized + "generic_openapi", legacy, materialized ) self.assertEqual(changed, frozenset()) self.assertEqual(impact, "none") @@ -141,7 +147,7 @@ class ExternalConnectionRevisionTests(unittest.TestCase): "openapi_url": "https://factory.invalid/swagger.json", } _, _, changed, impact = _classify_definition_config_change( - "factory_mes", + "generic_openapi", config, {**config, "query_guidance": "先查数据集目录"}, ) @@ -160,7 +166,7 @@ class ExternalConnectionRevisionTests(unittest.TestCase): "openapi_url": "https://factory-new.invalid/swagger.json", } _, _, changed, impact = _classify_definition_config_change( - "factory_mes", old, new + "generic_openapi", old, new ) self.assertEqual( changed, frozenset({"base_url", "openapi_url"}) @@ -428,9 +434,7 @@ class GenericMcpConnectorTests(unittest.TestCase): def _cfg(*, allowed=frozenset(), recommended=(), operation_mode="query"): - from core.external_systems.factory import FactoryMesConfig - - return FactoryMesConfig( + return OpenApiConfig( base_url="https://factory.invalid", openapi_url="https://factory.invalid/swagger.json", login_path="/api/auth/token/", @@ -438,7 +442,6 @@ def _cfg(*, allowed=frozenset(), recommended=(), operation_mode="query"): timeout_seconds=5, max_result_bytes=65536, max_total_result_bytes=262144, - max_page_size=200, verify_tls=True, query_guidance="先查数据集目录", recommended_operation_ids=tuple(recommended), @@ -579,16 +582,14 @@ class ExternalRuntimeCacheTests(unittest.TestCase): self.assertTrue(third.closed) -class FactoryOpenApiConnectorTests(unittest.TestCase): +class OpenApiConnectorTests(unittest.TestCase): def setUp(self): from core.external_systems.openapi import _SPEC_CACHE _SPEC_CACHE.clear() def test_admin_mapping_builds_bounded_runtime_config(self): - from core.external_systems.factory import FactoryMesConfig - - cfg = FactoryMesConfig.from_mapping( + cfg = OpenApiConfig.from_mapping( { "base_url": "https://factory.invalid/", "openapi_url": "https://factory.invalid/swagger.json", @@ -605,8 +606,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(cfg.timeout_seconds, 60) self.assertEqual(cfg.max_result_bytes, 4096) self.assertEqual(cfg.max_total_result_bytes, 262144) - self.assertEqual(cfg.max_page_size, 200) - self.assertEqual(cfg.operation_mode, "upstream_managed") + self.assertEqual(cfg.operation_mode, "query") self.assertEqual( cfg.operation_policies, { @@ -614,17 +614,12 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): "report_preview": "export", }, ) - self.assertIn("dataset list", cfg.query_guidance) - self.assertEqual( - cfg.recommended_operation_ids, - ("bi_dataset_list", "bi_dataset_exec"), - ) + self.assertEqual(cfg.query_guidance, "") + self.assertEqual(cfg.recommended_operation_ids, ()) def test_admin_mapping_rejects_embedded_url_credentials(self): - from core.external_systems.factory import FactoryMesConfig, FactoryMesError - - with self.assertRaisesRegex(FactoryMesError, "不能内嵌凭据"): - FactoryMesConfig.from_mapping( + with self.assertRaisesRegex(OpenApiError, "不能内嵌凭据"): + OpenApiConfig.from_mapping( { "base_url": "https://user:secret@factory.invalid", "openapi_url": "https://factory.invalid/swagger.json", @@ -632,10 +627,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) def test_admin_mapping_requires_same_origin_openapi_document(self): - from core.external_systems.factory import FactoryMesConfig, FactoryMesError - - with self.assertRaisesRegex(FactoryMesError, "必须与 base_url 同源"): - FactoryMesConfig.from_mapping( + with self.assertRaisesRegex(OpenApiError, "必须与 base_url 同源"): + OpenApiConfig.from_mapping( { "base_url": "https://factory.invalid", "openapi_url": "https://spec.attacker.invalid/swagger.json", @@ -705,10 +698,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(len(_SPEC_CACHE), 2) def test_search_discovers_operation_without_exposing_credentials(self): - from core.external_systems.factory import FactoryMesClient - http = _Http() - client = FactoryMesClient("mes-user", "mes-password", _cfg()) + client = _openapi_client("mes-user", "mes-password", _cfg()) with patch.object(client, "_client", return_value=http): result = client.search("成品检验") self.assertEqual(result[0]["operation_id"], "qm_ftestwork_read") @@ -717,12 +708,11 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertNotIn("remote-jwt", rendered) def test_runtime_reuses_http_auth_spec_and_compiled_catalog(self): - from core.external_systems.factory import FactoryMesClient from core.external_systems.openapi import compile_operation_catalog http = _Http() - first = FactoryMesClient("mes-user", "mes-password", _cfg()) - second = FactoryMesClient("mes-user", "mes-password", _cfg()) + first = _openapi_client("mes-user", "mes-password", _cfg()) + second = _openapi_client("mes-user", "mes-password", _cfg()) with ( patch.object(first, "_client", return_value=http) as first_factory, patch.object(second, "_client", return_value=_Http()) as second_factory, @@ -765,8 +755,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) def test_concurrent_identical_query_is_singleflight_only(self): - from core.external_systems.factory import FactoryMesClient - class SlowHttp(_Http): def __init__(self): super().__init__() @@ -784,7 +772,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): return super().request(method, url, **kwargs) http = SlowHttp() - client = FactoryMesClient("mes-user", "mes-password", _cfg()) + client = _openapi_client("mes-user", "mes-password", _cfg()) with patch.object(client, "_client", return_value=http): client.search("成品检验") # 预热认证、规格和 catalog，只测业务请求单飞。 with ThreadPoolExecutor(max_workers=2) as executor: @@ -809,8 +797,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertIsNot(first_result, second_result) def test_concurrent_cold_search_coalesces_login_and_spec_fetch(self): - from core.external_systems.factory import FactoryMesClient - class SlowDiscoveryHttp(_Http): def post(self, url, **kwargs): response = super().post(url, **kwargs) @@ -823,8 +809,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): return response http = SlowDiscoveryHttp() - first = FactoryMesClient("mes-user", "mes-password", _cfg()) - second = FactoryMesClient("mes-user", "mes-password", _cfg()) + first = _openapi_client("mes-user", "mes-password", _cfg()) + second = _openapi_client("mes-user", "mes-password", _cfg()) with ( patch.object(first, "_client", return_value=http), patch.object(second, "_client", return_value=http), @@ -853,8 +839,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) def test_cached_password_token_refreshes_once_after_401(self): - from core.external_systems.factory import FactoryMesClient - class RefreshHttp(_Http): def __init__(self): super().__init__() @@ -869,7 +853,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): return super().request(method, url, **kwargs) http = RefreshHttp() - client = FactoryMesClient("mes-user", "mes-password", _cfg()) + client = _openapi_client("mes-user", "mes-password", _cfg()) with patch.object(client, "_client", return_value=http): result = client.call("qm_ftestwork_read", arguments={"batch": "B1"}) @@ -885,10 +869,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) def test_search_pins_callable_admin_recommendations_without_keyword_match(self): - from core.external_systems.factory import FactoryMesClient - http = _Http() - client = FactoryMesClient( + client = _openapi_client( "mes-user", "mes-password", _cfg( @@ -905,10 +887,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertTrue(all(item["recommended"] for item in result[:2])) def test_search_includes_resolved_swagger_body_schema(self): - from core.external_systems.factory import FactoryMesClient - http = _Http() - client = FactoryMesClient( + client = _openapi_client( "mes-user", "mes-password", _cfg(allowed={"bi_dataset_exec"}) ) with patch.object(client, "_client", return_value=http): @@ -969,8 +949,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) def test_catalog_resolves_referenced_header_parameter(self): - from core.external_systems.factory import FactoryMesClient - spec = { "openapi": "3.0.0", "components": { @@ -994,7 +972,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): } cfg = _cfg() http = _Http() - client = FactoryMesClient("u", "p", cfg) + client = _openapi_client("u", "p", cfg) with ( patch.object(client, "_client", return_value=http), patch.object(client, "_fetch_spec", return_value=spec), @@ -1004,8 +982,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(request[2]["headers"]["X-Trace-Id"], "trace-1") def test_swagger_array_query_uses_declared_collection_format(self): - from core.external_systems.factory import FactoryMesClient - spec = { "swagger": "2.0", "paths": { @@ -1027,7 +1003,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): } cfg = _cfg() http = _Http() - client = FactoryMesClient("u", "p", cfg) + client = _openapi_client("u", "p", cfg) with ( patch.object(client, "_client", return_value=http), patch.object(client, "_fetch_spec", return_value=spec), @@ -1037,10 +1013,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(request[2]["params"]["batches"], "B1,B2") def test_get_call_resolves_encoded_path_and_query(self): - from core.external_systems.factory import FactoryMesClient - http = _Http() - client = FactoryMesClient("mes-user", "mes-password", _cfg()) + client = _openapi_client("mes-user", "mes-password", _cfg()) with patch.object(client, "_client", return_value=http): result = client.call( "qm_ftestwork_read", @@ -1054,13 +1028,21 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(kwargs["params"], {"page_size": 50}) self.assertEqual(result["data"]["count"], 1) - def test_get_call_bounds_pagination_for_agent_queries(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError - + def test_query_parameters_follow_openapi_schema_without_name_heuristics(self): spec = deepcopy(_SPEC) + page_size = spec["paths"]["/api/qm/ftestwork/{batch}/"]["get"][ + "parameters" + ][1] + page_size["maximum"] = 200 spec["paths"]["/api/qm/ftestwork/{batch}/"]["get"]["parameters"].extend( [ - {"name": "page", "in": "query", "required": False, "type": "integer"}, + { + "name": "page", + "in": "query", + "required": False, + "type": "integer", + "minimum": 0, + }, { "name": "pageoff", "in": "query", @@ -1070,38 +1052,43 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ] ) http = _Http() - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with ( patch.object(client, "_client", return_value=http), patch.object(client, "_fetch_spec", return_value=spec), ): client.call( "qm_ftestwork_read", - arguments={"batch": "B1", "page": 1, "page_size": 99999}, + arguments={ + "batch": "B1", + "page": 0, + "page_size": 200, + "pageoff": True, + }, ) request = next(call for call in http.calls if call[0] == "GET") self.assertEqual(request[2]["params"]["page_size"], 200) + self.assertEqual(request[2]["params"]["page"], 0) + self.assertTrue(request[2]["params"]["pageoff"]) with ( - patch.object(client, "authenticate", return_value="jwt"), + patch.object(client, "authenticate", return_value={}), patch.object(client, "_fetch_spec", return_value=spec), - self.assertRaisesRegex(FactoryMesError, "不允许 page=0"), + self.assertRaisesRegex(OpenApiError, "page_size 必须 <= 200"), ): client.call( "qm_ftestwork_read", - arguments={"batch": "B1", "page": 0}, + arguments={"batch": "B1", "page_size": 99999}, ) def test_swagger_base_path_is_added_to_operation_url(self): - from core.external_systems.factory import FactoryMesClient - spec = deepcopy(_SPEC) spec["basePath"] = "/api" spec["paths"] = { path.removeprefix("/api"): value for path, value in spec["paths"].items() } http = _Http() - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with ( patch.object(client, "_client", return_value=http), patch.object(client, "_fetch_spec", return_value=spec), @@ -1111,21 +1098,17 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(request[1], "https://factory.invalid/api/qm/ftestwork/B1/") def test_api_base_path_does_not_change_login_url(self): - from core.external_systems.factory import FactoryMesClient - http = _Http() - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with patch.object(client, "_client", return_value=http): client.authenticate() request = next(call for call in http.calls if call[0] == "POST") self.assertEqual(request[1], "https://factory.invalid/api/auth/token/") def test_base_path_is_not_duplicated_when_operation_already_contains_it(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesConfig - spec = {**deepcopy(_SPEC), "basePath": "/api"} http = _Http() - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with ( patch.object(client, "_client", return_value=http), patch.object(client, "_fetch_spec", return_value=spec), @@ -1134,10 +1117,10 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): request = next(call for call in http.calls if call[0] == "GET") self.assertNotIn("/api/api/", request[1]) - configured_prefix = FactoryMesClient( + configured_prefix = _openapi_client( "u", "p", - FactoryMesConfig( + OpenApiConfig( **{**_cfg().__dict__, "base_url": "https://factory.invalid/api"} ), ) @@ -1147,9 +1130,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) def test_openapi_server_path_is_used_but_cross_origin_server_is_rejected(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError - - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) same_origin = {"openapi": "3.0.0", "servers": [{"url": "/v1"}], "paths": {}} self.assertEqual( client._operation_url(same_origin, "/quality/results/"), @@ -1160,31 +1141,27 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): "servers": [{"url": "https://attacker.invalid/v1"}], "paths": {}, } - with self.assertRaisesRegex(FactoryMesError, "越出 Factory MES 主机"): + with self.assertRaisesRegex(OpenApiError, "越出管理员配置的外部系统主机"): client._operation_url(cross_origin, "/quality/results/") def test_no_declared_base_path_keeps_existing_url_behavior(self): - from core.external_systems.factory import FactoryMesClient - - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) self.assertEqual( client._operation_url(_SPEC, "/api/qm/ftestwork/B1/"), "https://factory.invalid/api/qm/ftestwork/B1/", ) def test_post_is_denied_unless_admin_allowlists_operation(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError - - denied = FactoryMesClient("u", "p", _cfg()) + denied = _openapi_client("u", "p", _cfg()) with ( patch.object(denied, "authenticate", return_value="jwt"), patch.object(denied, "_fetch_spec", return_value=_SPEC), ): - with self.assertRaisesRegex(FactoryMesError, "只读调用范围"): + with self.assertRaisesRegex(OpenApiError, "只读调用范围"): denied.call("bi_dataset_exec", arguments={"code": "x", "payload": {}}) http = _Http() - allowed = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"})) + allowed = _openapi_client("u", "p", _cfg(allowed={"bi_dataset_exec"})) with patch.object(allowed, "_client", return_value=http): result = allowed.call( "bi_dataset_exec", @@ -1197,10 +1174,8 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertFalse(result["truncated"]) def test_allowlisted_post_accepts_separate_body_field(self): - from core.external_systems.factory import FactoryMesClient - http = _Http() - allowed = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"})) + allowed = _openapi_client("u", "p", _cfg(allowed={"bi_dataset_exec"})) with patch.object(allowed, "_client", return_value=http): result = allowed.call( "bi_dataset_exec", @@ -1214,8 +1189,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertFalse(result["truncated"]) def test_upstream_managed_mode_allows_declared_write_method(self): - from core.external_systems.factory import FactoryMesClient - spec = deepcopy(_SPEC) spec["paths"]["/api/items/{item_id}/"] = { "put": { @@ -1241,7 +1214,7 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): } } http = _Http() - client = FactoryMesClient("u", "p", _cfg(operation_mode="upstream_managed")) + client = _openapi_client("u", "p", _cfg(operation_mode="upstream_managed")) with ( patch.object(client, "_client", return_value=http), patch.object(client, "authenticate", return_value={}), @@ -1258,8 +1231,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertEqual(result["status_code"], 200) def test_query_mode_still_rejects_declared_write_method(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError - spec = { "swagger": "2.0", "paths": { @@ -1278,23 +1249,21 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): } }, } - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with ( patch.object(client, "authenticate", return_value={}), patch.object(client, "_fetch_spec", return_value=spec), ): - with self.assertRaisesRegex(FactoryMesError, "只读调用范围"): + with self.assertRaisesRegex(OpenApiError, "只读调用范围"): client.call("item_delete", arguments={"item_id": "A-1"}) def test_call_preserves_payload_larger_than_inline_limit(self): - from core.external_systems.factory import FactoryMesClient - class LargeHttp(_Http): def request(self, method, url, **kwargs): self.calls.append((method, url, kwargs)) return _Response(payload={"rows": "x" * 70000}) - client = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"})) + client = _openapi_client("u", "p", _cfg(allowed={"bi_dataset_exec"})) with patch.object(client, "_client", return_value=LargeHttp()): result = client.call( "bi_dataset_exec", @@ -1306,7 +1275,6 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertFalse(result["truncated"]) def test_call_stops_stream_when_download_limit_is_exceeded(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError from core.external_systems.results import MAX_STORED_RESULT_BYTES class OversizedResponse(_Response): @@ -1319,14 +1287,12 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.calls.append((method, url, kwargs)) return OversizedResponse(payload={}) - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with patch.object(client, "_client", return_value=OversizedHttp()): - with self.assertRaisesRegex(FactoryMesError, "安全下载上限"): + with self.assertRaisesRegex(OpenApiError, "安全下载上限"): client.call("qm_ftestwork_read", arguments={"batch": "B1"}) def test_call_surfaces_sanitized_upstream_error_detail(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError - class ErrorHttp(_Http): def request(self, method, url, **kwargs): self.calls.append((method, url, kwargs)) @@ -1338,9 +1304,9 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): }, ) - client = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"})) + client = _openapi_client("u", "p", _cfg(allowed={"bi_dataset_exec"})) with patch.object(client, "_client", return_value=ErrorHttp()): - with self.assertRaises(FactoryMesError) as raised: + with self.assertRaises(OpenApiError) as raised: client.call( "bi_dataset_exec", arguments={"code": "yield"}, @@ -1352,14 +1318,12 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): self.assertNotIn("must-not-leak", message) def test_rejects_unknown_arguments(self): - from core.external_systems.factory import FactoryMesClient, FactoryMesError - - client = FactoryMesClient("u", "p", _cfg()) + client = _openapi_client("u", "p", _cfg()) with ( patch.object(client, "authenticate", return_value="jwt"), patch.object(client, "_fetch_spec", return_value=_SPEC), ): - with self.assertRaisesRegex(FactoryMesError, "接口定义之外"): + with self.assertRaisesRegex(OpenApiError, "接口定义之外"): client.call( "qm_ftestwork_read", arguments={"batch": "B1", "unexpected": "x"}, diff --git a/tests/test_web_routes_nodb.py b/tests/test_web_routes_nodb.py index 16480cb..fbb1af8 100644 --- a/tests/test_web_routes_nodb.py +++ b/tests/test_web_routes_nodb.py @@ -139,8 +139,8 @@ class AuthGateTests(unittest.TestCase): class ExternalSystemRoutesTests(unittest.TestCase): def test_provider_catalog_and_create_are_user_scoped(self): provider = { - "provider": "factory_mes", - "title": "Factory MES", + "provider": "generic_openapi", + "title": "通用 OpenAPI 系统", "configured": True, } with patch("web.routers.external_systems.provider_catalog", return_value=[provider]) as catalog: @@ -157,7 +157,7 @@ class ExternalSystemRoutesTests(unittest.TestCase): headers=_AUTH, json={ "definition_id": str(definition_id), - "name": "Factory MES", + "name": "材料数据平台", "credentials": { "username": "mes-user", "password": "secret", diff --git a/tools/external_systems.py b/tools/external_systems.py index 5ebefba..0e867ce 100644 --- a/tools/external_systems.py +++ b/tools/external_systems.py @@ -10,7 +10,7 @@ from pathlib import Path from uuid import UUID, uuid4 from core.artifacts import ArtifactRef, ToolExecutionResult, resolve_artifact_path -from core.external_systems.factory import FactoryMesError +from core.external_systems.openapi import OpenApiError from core.external_systems.mcp import McpConnectorError from core.external_systems.results import ( ExternalResultError, @@ -110,7 +110,7 @@ class ExternalSystemSearchTool(Tool): "count": len(results), } ) - except (ExternalSystemError, FactoryMesError, McpConnectorError) as exc: + except (ExternalSystemError, OpenApiError, McpConnectorError) as exc: print(f"[WARN] external system search failed: {type(exc).__name__}") return f"[Error] {exc}" @@ -299,7 +299,7 @@ class ExternalSystemCallTool(Tool): return output except ( ExternalSystemError, - FactoryMesError, + OpenApiError, McpConnectorError, ExternalResultError, ) as exc: @@ -410,7 +410,7 @@ class ExternalSystemResultReadTool(Tool): return output except ( ExternalSystemError, - FactoryMesError, + OpenApiError, McpConnectorError, ExternalResultError, TypeError, @@ -508,7 +508,7 @@ class ExternalSystemResultExportTool(Tool): return ToolExecutionResult(content, artifacts=(ArtifactRef(path=rel),)) except ( ExternalSystemError, - FactoryMesError, + OpenApiError, McpConnectorError, ExternalResultError, OSError, diff --git a/web/admin.py b/web/admin.py index 5d49990..d8f79fd 100644 --- a/web/admin.py +++ b/web/admin.py @@ -183,7 +183,7 @@ class SetPlanRequest(BaseModel): class ExternalSystemDefinitionRequest(BaseModel): - provider: str = "factory_mes" + provider: str = "generic_openapi" name: str base_url: str = "" openapi_url: str = "" @@ -201,13 +201,10 @@ class ExternalSystemDefinitionRequest(BaseModel): timeout_seconds: float = 15 max_result_bytes: int = 65536 max_total_result_bytes: int = 262144 - max_page_size: int = 200 max_response_bytes: int = 10485760 verify_tls: bool = True query_guidance: str = "" - recommended_operation_ids: list[str] = Field( - default_factory=lambda: ["bi_dataset_list", "bi_dataset_exec"] - ) + recommended_operation_ids: list[str] = Field(default_factory=list) enabled: bool = True visibility: str = "selected" selected_user_ids: list[UUID] = Field(default_factory=list) @@ -230,7 +227,6 @@ def _external_definition_config(body: ExternalSystemDefinitionRequest) -> dict[s "timeout_seconds": body.timeout_seconds, "max_result_bytes": body.max_result_bytes, "max_total_result_bytes": body.max_total_result_bytes, - "max_page_size": body.max_page_size, "max_response_bytes": body.max_response_bytes, "verify_tls": body.verify_tls, "query_guidance": body.query_guidance, diff --git a/web/static/js/admin.js b/web/static/js/admin.js index 7bc1c48..4db33b0 100644 --- a/web/static/js/admin.js +++ b/web/static/js/admin.js @@ -9,11 +9,6 @@ import { dialogPrompt } from "./dialog.js"; const LS_TOKEN = "zcbot.token"; const REFRESH_MS = 10000; const PAGE_SIZE = 20; -const DEFAULT_EXTERNAL_QUERY_GUIDANCE = - "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询，统一先调用 BI dataset list，再执行匹配的数据集。" - + "日志和业务明细列表用于用户明确要求查看逐条记录、编号或追溯过程的场景。" - + "未匹配到 dataset 时，先限定范围或向用户确认明细查询需求。"; - const RANGE_OPTS = [["all", "全部"], ["7d", "近7天"], ["30d", "近30天"]]; const SORT_OPTS = [["cost", "按成本"], ["tokens", "按用量"]]; const SECTIONS = [ @@ -177,22 +172,22 @@ function renderExternalDefinitions() { $("s-external").innerHTML = `

外部系统目录

` + `OpenAPI 与 Streamable HTTP MCP 系统均可配置；用户只提交该系统要求的凭据
` + `
` - + `` + + `` + `` + `` - + `` - + `` - + `` - + `` - + `` + + `` + + `` + + `` + + `` + + `` + `` + `` + `` + `` + `` + `` - + `` - + `` + + `` + + `` + `