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):