From a127f7049c8d66d220be26df026c0d37cd360955 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Thu, 17 Sep 2026 11:44:22 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9C=9F=E5=AE=9E=20Oracle=20=E7=94=9F?= =?UTF-8?q?=E4=BA=A7=E5=BA=93=E8=B7=91=E9=80=9A=EF=BC=8C=E4=BF=AE=E6=8E=89?= =?UTF-8?q?=E4=B8=89=E5=A4=84=E5=8F=AA=E6=9C=89=E8=BF=9E=E7=9C=9F=E5=BA=93?= =?UTF-8?q?=E6=89=8D=E6=9A=B4=E9=9C=B2=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在一个企业 BI 库（SAP 物料/采购/凭证 + MES 生产 + IOT 设备）上端到端验证。 连接 3 秒握手，thin 模式，确实不需要 Instant Client。三处问题只有连真库才看得见： ## options.schema 被当成驱动参数拼进了连接串 我自己引入的：build_url 把 options 里所有剩余的 key 都拼成 query 参数， 而 schema 是探查配置不是连接参数，于是 ?service_name=BI&schema=ANALYTICS 直接 被驱动拒掉。options 里混着两类东西，现在显式区分。 ## 只读账号名下一张表都没有 这是企业环境的常态而非例外：TAIKE 名下只有 2 个 SYNONYM，数据在 ANALYTICS (51 个视图) 和 STAGING(9 张表)。默认探查当前用户 schema 得到 0 张表，而原来 只会安静地返回一个空壳——用户完全不知道下一步该干什么。 现在：options.schema 可指定探查目标；探不到任何对象时自动列出可访问的非系统 schema（排掉 MDSYS/CTXSYS/XDB 那一屏 Oracle 内部对象），摘要里直接告诉 Copilot"该账号名下没有对象，数据可能在这些 schema"。 ## 表名不带 schema 前缀直接 ORA-00942 只读账号访问的是别人 schema 下的对象，SELECT * FROM V_TRIP_FACT 必然失败。 schema_cache 里为每个对象记 qualified 全名，summary / table_names / db_query 的工具描述统一用全名——模型照着写就不会漏前缀。describe_table 则 反过来放宽：全名、裸表名、大写形态都认（Oracle 惯用大写，SQLAlchemy 返回小写）。 ## 实测 探查 53 个对象 7.8 秒，摘要约 2000 token（含一个 273 字段的宽视图），在预算内。 Oracle 的中文表/列注释正确读出（"不合格数据明细表"、"工厂代码，复合主键之一"）。 agent 自主查询 8 轮：列对象 → 挑 3 个 MES 视图看结构 → 逐个 COUNT + MAX(load_time) → 补查业务时间字段，最后报出 v_demo_table 有 365,633 行、最新数据 2026-09-17 06:53。全程自己带对了 schema 前缀。 安全边界在生产库上复验： - UPDATE / DELETE / DROP 全被拒，数据未变 - 19.87 亿行的 TG_ECC_MSEG 裸查 SELECT *：0.7 秒返回 200 行即断开， 流式读取没有把表拉回来 - 203 列宽表的字节上限在第 37 行触发 凭据只进加密库（Fernet 密文，已核验库中无明文），不在仓库、提交历史或任何输出里。 --- backend/app/api/datasources.py | 4 +- backend/app/data/engine.py | 9 ++++ backend/app/data/introspect.py | 90 ++++++++++++++++++++++++++++---- backend/tests/test_datasource.py | 10 +++- 4 files changed, 100 insertions(+), 13 deletions(-) diff --git a/backend/app/api/datasources.py b/backend/app/api/datasources.py index 0fd56e73..7d2d2454 100644 --- a/backend/app/api/datasources.py +++ b/backend/app/api/datasources.py @@ -114,7 +114,9 @@ async def list_kinds() -> dict[str, Any]: {"value": "oracle", "label": "Oracle", "default_port": 1521, "needs": ["host", "username", "password"], "hint": "库名填 service_name，或在 options 里给 service_name / sid。" - "驱动走 thin 模式，不需要装 Instant Client"}, + "驱动走 thin 模式，不需要装 Instant Client。" + "只读账号名下通常没有对象——数据在别的 schema 里，" + "用 options.schema 指定（如 ANALYTICS），否则探查结果是空的"}, {"value": "sqlite", "label": "SQLite（文件）", "default_port": None, "needs": ["database"], "hint": "库名填数据库文件的绝对路径"}, ], diff --git a/backend/app/data/engine.py b/backend/app/data/engine.py index a3f82b7f..79c75e1d 100644 --- a/backend/app/data/engine.py +++ b/backend/app/data/engine.py @@ -36,6 +36,10 @@ _DEFAULT_PORTS = { SUPPORTED_KINDS = tuple(sorted(set(_DRIVERS))) +# options 里这些 key 是 AgentLab 自己的配置，不是驱动参数，拼 URL 时要摘掉。 +# schema：探查哪个 schema（企业库里只读账号名下常常什么都没有，数据在别处） +_NON_DRIVER_OPTIONS = frozenset({"schema"}) + @dataclass class QueryResult: @@ -69,6 +73,11 @@ def build_url(source: Any, *, reveal: bool = False) -> str: raise ValueError(f"不支持的数据库类型：{source.kind}（支持 {', '.join(SUPPORTED_KINDS)}）") options: dict[str, Any] = dict(source.options or {}) + # options 混着两类东西：驱动参数（charset、service_name…）和我们自己的配置 + # （schema 指探查哪个 schema）。后者不能拼进连接串，否则驱动会把它当成 + # 未知的连接参数直接拒掉。 + for key in _NON_DRIVER_OPTIONS: + options.pop(key, None) if kind == "sqlite": # SQLite 没有主机和账号，database 就是文件路径 diff --git a/backend/app/data/introspect.py b/backend/app/data/introspect.py index 3c14045f..63dc05b4 100644 --- a/backend/app/data/introspect.py +++ b/backend/app/data/introspect.py @@ -27,13 +27,53 @@ _MAX_TABLES = 200 _PREVIEW_COLUMNS = 6 +# 各家的系统 schema，发现可用 schema 时要排掉——否则用户面对的是一屏 +# MDSYS / CTXSYS / XDB 这类内部对象，真正的业务库反而埋在里面。 +_SYSTEM_SCHEMAS = frozenset({ + # Oracle + "sys", "system", "sysaux", "mdsys", "ctxsys", "xdb", "olapsys", "wmsys", + "lbacsys", "ordsys", "orddata", "outln", "dbsnmp", "appqossys", "audsys", + "gsmadmin_internal", "dvsys", "ojvmsys", "dbsfwuser", "remote_scheduler_agent", + # PostgreSQL / MySQL + "information_schema", "pg_catalog", "pg_toast", "performance_schema", "mysql", "sys", +}) + + +async def list_schemas(source: Any) -> list[str]: + """列出可访问的非系统 schema。 + + 企业环境里只读账号名下往往一张表都没有——数据在别的 schema，靠跨库授权或 + synonym 访问。直接返回"0 张表"等于把人晾在那儿，得告诉他数据可能在哪。 + """ + engine = await engines.get(source) + + def _collect(sync_conn: Any) -> list[str]: + try: + names = sa_inspect(sync_conn).get_schema_names() + except Exception: # noqa: BLE001 - 有些方言不支持 + return [] + return [n for n in names if n.lower() not in _SYSTEM_SCHEMAS] + + async with engine.connect() as conn: + return await conn.run_sync(_collect) + + +def _target_schema(source: Any, explicit: str | None) -> str | None: + """探查哪个 schema：显式参数 > 数据源配置 > 当前用户默认。""" + if explicit: + return explicit + configured = (source.options or {}).get("schema") + return str(configured) if configured else None + + async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any]: """探查整个库的结构，结果直接存进 DataSource.schema_cache。""" engine = await engines.get(source) + target = _target_schema(source, schema) def _collect(sync_conn: Any) -> dict[str, Any]: inspector = sa_inspect(sync_conn) - target_schema = schema or None + target_schema = target names = inspector.get_table_names(schema=target_schema) views = inspector.get_view_names(schema=target_schema) truncated = len(names) + len(views) > _MAX_TABLES @@ -57,6 +97,10 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any comment = None tables[name] = { + # 写 SQL 要用的全名。没有 schema 前缀时 Oracle 直接 ORA-00942—— + # 只读账号名下没有这些对象，全靠跨 schema 授权 + "qualified": f"{target_schema}.{name}" if target_schema else name, + "schema": target_schema, "columns": [ { "name": c["name"], @@ -76,7 +120,12 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any payload = await conn.run_sync(_collect) payload["synced_at"] = datetime.now(timezone.utc).isoformat() - payload["schema"] = schema + payload["schema"] = target + + # 一张表都没探到时，别只回一个空壳——多半是数据在别的 schema 里， + # 把候选列出来，用户才知道下一步该填什么 + if not payload["tables"]: + payload["available_schemas"] = await list_schemas(source) return payload @@ -85,20 +134,31 @@ def summary(source: Any, *, max_tables: int = 40) -> str: cache = source.schema_cache or {} tables = cache.get("tables") or {} if not tables: - return f"- {source.name}（{source.kind}）：{source.description or '未填说明'}｜尚未探查结构" + hint = "" + candidates = cache.get("available_schemas") or [] + if candidates: + hint = f"｜该账号名下没有对象，数据可能在这些 schema：{'、'.join(candidates[:8])}" + return ( + f"- {source.name}（{source.kind}）：{source.description or '未填说明'}" + f"｜尚未探查结构{hint}" + ) + schema = cache.get("schema") lines = [ f"- {source.name}（{source.kind}，{'只读' if source.readonly else '可写'}）：" f"{source.description or '未填说明'}" + + (f"｜schema：{schema}" if schema else "") ] for name, meta in list(tables.items())[:max_tables]: cols = meta.get("columns") or [] preview = "、".join(c["name"] for c in cols[:_PREVIEW_COLUMNS]) more = f" 等 {len(cols)} 字段" if len(cols) > _PREVIEW_COLUMNS else "" note = f"（{meta['comment']}）" if meta.get("comment") else "" - lines.append(f" · {name}{note}: {preview}{more}") + kind = "视图" if meta.get("is_view") else "表" + # 用全名：模型照着写 SQL 时少了 schema 前缀在 Oracle 上直接报 ORA-00942 + lines.append(f" · {meta.get('qualified', name)}[{kind}]{note}: {preview}{more}") if len(tables) > max_tables: - lines.append(f" · …另有 {len(tables) - max_tables} 张表，用 db_schema 工具查看") + lines.append(f" · …另有 {len(tables) - max_tables} 个对象，用 db_schema 工具查看") return "\n".join(lines) @@ -106,16 +166,22 @@ def describe_table(source: Any, table: str) -> str: """单表的完整字段说明。agent 调 db_schema 工具时返回这个。""" cache = source.schema_cache or {} tables = cache.get("tables") or {} + # 模型可能传全名（ANALYTICS.V_TRIP_FACT），也可能只传表名，两种都认 meta = tables.get(table) + if meta is None and "." in table: + meta = tables.get(table.rsplit(".", 1)[-1]) + if meta is None: + lowered = table.rsplit(".", 1)[-1].lower() + meta = next((m for n, m in tables.items() if n.lower() == lowered), None) if not meta: - available = "、".join(list(tables)[:30]) or "（还没探查过结构）" - return f"数据源「{source.name}」里没有表 {table}。现有的表：{available}" + available = "、".join( + m.get("qualified", n) for n, m in list(tables.items())[:30] + ) or "（还没探查过结构）" + return f"数据源「{source.name}」里没有 {table}。现有的对象：{available}" - head = f"表 {table}" + head = f"{'视图' if meta.get('is_view') else '表'} {meta.get('qualified', table)}" if meta.get("comment"): head += f"（{meta['comment']}）" - if meta.get("is_view"): - head += " [视图]" lines = [head] pk = set(meta.get("primary_key") or []) for col in meta.get("columns") or []: @@ -132,4 +198,6 @@ def describe_table(source: Any, table: str) -> str: def table_names(source: Any) -> list[str]: - return list((source.schema_cache or {}).get("tables") or {}) + """带 schema 前缀的对象全名。模型照着这个写 SQL 才不会漏掉前缀。""" + tables = (source.schema_cache or {}).get("tables") or {} + return [meta.get("qualified", name) for name, meta in tables.items()] diff --git a/backend/tests/test_datasource.py b/backend/tests/test_datasource.py index c1eb236f..34612297 100644 --- a/backend/tests/test_datasource.py +++ b/backend/tests/test_datasource.py @@ -75,10 +75,18 @@ async def test_summary_says_so_when_never_introspected(source): assert "尚未探查" in summary(source) +async def test_describe_table_accepts_qualified_name(source): + """模型可能传 SCHEMA.TABLE 全名，也可能只传表名，两种都要认。""" + source.schema_cache = await introspect(source) + assert "amount" in describe_table(source, "orders") + assert "amount" in describe_table(source, "main.orders") + assert "amount" in describe_table(source, "ORDERS") # Oracle 惯用大写 + + async def test_describe_table_lists_unknown_table_options(source): source.schema_cache = await introspect(source) out = describe_table(source, "没有这张表") - assert "现有的表" in out and "orders" in out + assert "现有的对象" in out and "orders" in out async def test_query_returns_rows(source):