mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 07:51:57 +00:00
ThearchyHelios
03c18c4d78
一、文案（前端约 1600 处，后端传到界面的约 880 处） - 术语统一：工作流层面用「运行」（继续运行、重新运行、试运行），节点和工具层面用「执行」，不再用「跑」； 总时长 / 执行时长 / 等待审批；服务端；助手；Agent；token；工作流；固定版本；模型接入；系统提示；批准 / 驳回；受管级别 - 口语改书面：取不到、没取回来、读不懂、连不上、多半、照样、兜底、揉成这类说法改成「无法 / 未能 / 失败 / 请」；去掉拟人写法 - 不再露出实现细节：界面上不写 checkpoint、同一拍、哈希（凭证处除外），不写改版历史（「以前…」「老数据」）； 报错、校验、门禁、自动修复里裸写的配置键名和枚举值换成界面上的中文标签，由新的 backend/app/engine/labels.py 统一提供； 发布弹窗里结论句的选项不再是 off / require_citation - 同一件事只留一种说法：生成阶段文案、出具档位说明、画布锁定提示各合成一份 - 去掉界面上的工业场景示例（MES、ERP、「张工」「王工」）和真实产品名，换成通用示例 二、说错的提示（都有测试证明） - 数据库测连接失败时的专用提示以前从不出现（按报错文字全等比较，永远比不上）：改为按错误类型判断，不再拼出重复的前缀 - 「设置 → 供应商」改指「设置 → 模型接入」；「在设置里建数据源」改指「数据」页 - 缺少 API Key、模型接入已停用的报错不再夹英文，写明去哪里填 - 两处原因说错（「这种引用在后续版本支持」「这一版还不支持」）改成真实原因 - 节点说明里不准确的说法改正：代码节点在 microVM 或系统沙箱中执行；报告撰写、口径卡的说明不再说过头 - 导出的证据 CSV 里两列同名，第三列改为「片段编号」 三、前后端对齐 - 前端按后端原文识别日志、报错的地方，新旧两种写法都认得（历史运行里存的是旧文字），逐条补了新旧原文的核对 - 后端交回模型的原指令放进单独的 for_model 字段，前端在解析响应时去掉，不会显示到界面上 四、检查 - 检查脚本里按原文匹配的断言改成新文案；改完会变成永远成立的反向断言逐条修正，并做了变异验证 - 新增 check-copy（静态扫描前端界面字符串和后端上界面的文案）：再出现「跑」「后端」「多半」「令牌」、 界面上的 Copilot、裸写的配置键名、半角标点等就报错；确需保留的写进 scripts/copy-allow.json 并注明理由
676 lines
29 KiB
Python
676 lines
29 KiB
Python
"""数据源的增删改查、连接测试、结构探查。
|
||
|
||
密码的处理规则和 Provider 完全一致（app/api/settings.py）：入库加密、
|
||
出站只给掩码、PATCH 时不传表示不动、传空串表示清空。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.api import health
|
||
from app.core.errors import AUTH, NETWORK, TIMEOUT, classify, explain, first_line, raw
|
||
from app.core.config import settings
|
||
from app.core.crypto import encrypt, mask
|
||
from app.data import introspect as introspect_mod
|
||
from app.data.engine import (
|
||
MASK_COLUMNS_OPTION, MAX_QUERY_TIMEOUT_S, QUERY_TIMEOUT_OPTION, SUPPORTED_KINDS, build_url, engine_args,
|
||
engines, mask_columns_problem, query_timeout_problem,
|
||
)
|
||
from app.db.base import get_session
|
||
from app.db.models import DataSource
|
||
from app.tools.datasource import tool_names
|
||
|
||
router = APIRouter(prefix="/api/datasources", tags=["datasources"])
|
||
|
||
# 名字会成为工具名的一部分（db_query__<name>），而模型的 function name 只认
|
||
# ASCII。所以这里收紧成 slug，中文说明放 description 字段。
|
||
_NAME_PATTERN = r"^[a-z][a-z0-9_]{0,40}$"
|
||
|
||
|
||
class DataSourceIn(BaseModel):
|
||
name: str = Field(pattern=_NAME_PATTERN,
|
||
description="英文标识，会成为工具名的一部分，如 sales")
|
||
kind: str
|
||
host: str | None = None
|
||
port: int | None = None
|
||
database: str | None = None
|
||
username: str | None = None
|
||
password: str | None = None
|
||
options: dict[str, Any] = Field(default_factory=dict)
|
||
readonly: bool = True
|
||
description: str = ""
|
||
enabled: bool = True
|
||
|
||
|
||
class DataSourcePatch(BaseModel):
|
||
kind: str | None = None
|
||
host: str | None = None
|
||
port: int | None = None
|
||
database: str | None = None
|
||
username: str | None = None
|
||
password: str | None = None # 空串=清空；不传=不动
|
||
options: dict[str, Any] | None = None
|
||
readonly: bool | None = None
|
||
description: str | None = None
|
||
enabled: bool | None = None
|
||
|
||
|
||
class DataSourceOut(BaseModel):
|
||
id: str
|
||
name: str
|
||
kind: str
|
||
host: str | None
|
||
port: int | None
|
||
database: str | None
|
||
username: str | None
|
||
options: dict[str, Any]
|
||
readonly: bool
|
||
description: str
|
||
enabled: bool
|
||
password_masked: str = ""
|
||
has_password: bool = False
|
||
table_count: int = 0
|
||
schema_synced_at: str | None = None
|
||
#: 上次探查为什么没拿到表。"还没探查"和"探查失败了"是两回事：
|
||
#: 前者去点一下按钮就行，后者点了也没用，得先解决超时或者换 schema
|
||
schema_error: str = ""
|
||
#: 这台服务器上还有哪些库/schema 可选。探不到表时它就是下一步的线索
|
||
available_schemas: list[str] = Field(default_factory=list)
|
||
tools: list[str] = Field(default_factory=list)
|
||
#: 缓存是按哪个 schema 探的："" 是默认 schema，None 是没有缓存（或老缓存没记）。
|
||
#: 和 options.schema 对不上时，助手看到的结构不是配置里那个 schema 的
|
||
cached_schema: str | None = None
|
||
#: 最近一次测连接（见 app/api/health.py）。没测过、或者连接配置改过之后都是 None
|
||
last_checked_at: str | None = None
|
||
last_check_ok: bool | None = None
|
||
last_latency_ms: int | None = None
|
||
last_error: str | None = None
|
||
|
||
|
||
def _cached_schema(cache: dict[str, Any]) -> str | None:
|
||
if not cache or "schema" not in cache:
|
||
return None
|
||
return str(cache.get("schema") or "")
|
||
|
||
|
||
def _to_out(row: DataSource) -> DataSourceOut:
|
||
cache = row.schema_cache or {}
|
||
tables = cache.get("tables") or {}
|
||
return DataSourceOut(
|
||
cached_schema=_cached_schema(cache),
|
||
schema_error=str(cache.get("error") or "") if cache.get("failed") else "",
|
||
available_schemas=list(cache.get("available_schemas") or []),
|
||
id=row.id, name=row.name, kind=row.kind, host=row.host, port=row.port,
|
||
database=row.database, username=row.username, options=row.options or {},
|
||
readonly=row.readonly, description=row.description, enabled=row.enabled,
|
||
password_masked=mask(row.password),
|
||
has_password=bool(row.password),
|
||
table_count=len(tables),
|
||
schema_synced_at=row.schema_synced_at.isoformat() if row.schema_synced_at else None,
|
||
tools=tool_names(row),
|
||
**health.fields(row.last_check),
|
||
)
|
||
|
||
|
||
def _connection(source: Any) -> str | None:
|
||
"""连接配置的指纹：驱动实际拿到的连接串和参数。配置拼不出连接串时为 None。
|
||
|
||
按连接串比而不是按字段比：Oracle 的服务名填在「数据库」还是 options 里，
|
||
连的是同一个库；探查用的 schema 不进连接串，改它也不影响连不连得上。
|
||
"""
|
||
try:
|
||
url, extra = engine_args(source)
|
||
except Exception: # noqa: BLE001 - 拼不出来就谈不上「测的是同一份」
|
||
return None
|
||
return health.fingerprint("datasource", url, extra)
|
||
|
||
|
||
def _probe_record(result: dict[str, Any]) -> dict[str, Any]:
|
||
return health.record(result.get("ok"), result.get("elapsed_ms"), result.get("error"))
|
||
|
||
|
||
async def _get_or_404(session: AsyncSession, source_id: str) -> DataSource:
|
||
row = await session.get(DataSource, source_id)
|
||
if not row:
|
||
raise HTTPException(404, "数据源不存在，可能已被删除")
|
||
return row
|
||
|
||
|
||
@router.get("", response_model=list[DataSourceOut])
|
||
async def list_sources(session: AsyncSession = Depends(get_session)) -> list[DataSourceOut]:
|
||
rows = (await session.execute(select(DataSource).order_by(DataSource.name))).scalars()
|
||
return [_to_out(r) for r in rows]
|
||
|
||
|
||
#: 每种库都有的「查询时限」。数据库按它自己停下语句，不只是后端不再等
|
||
_TIMEOUT_FIELD = {
|
||
"key": QUERY_TIMEOUT_OPTION, "label": "查询时限（秒）", "placeholder": "30",
|
||
"help": f"单条查询的最长执行时间，超时后由数据库终止；留空为 30 秒，最多 {MAX_QUERY_TIMEOUT_S} 秒",
|
||
}
|
||
|
||
|
||
#: 证据面板展示原始行时遮掉的列。每种库都有：遮罩只看查询结果的列名，和方言无关
|
||
_MASK_FIELD = {
|
||
"key": MASK_COLUMNS_OPTION, "label": "遮罩的列", "placeholder": "phone, email",
|
||
"help": "证据面板展示查询结果的原始行时，这些列显示为「已遮罩」。"
|
||
"遮罩只减少暴露，不是安全边界：完整快照仍可通过工件获取，SQL 中给列起别名也可绕过",
|
||
}
|
||
|
||
|
||
def _refuse_bad_options(options: dict[str, Any] | None) -> None:
|
||
problem = query_timeout_problem(options) or mask_columns_problem(options)
|
||
if problem:
|
||
raise HTTPException(422, problem)
|
||
|
||
|
||
@router.get("/kinds")
|
||
async def list_kinds() -> dict[str, Any]:
|
||
"""支持哪些数据库，以及各自需要填什么——前端表单据此渲染。
|
||
|
||
hint 只用表单上看得见的说法。以前写的是「在 options 里给 service_name」
|
||
「options 可填 charset」，表单上根本没有叫 options 的地方。advanced 列出
|
||
「高级连接参数」里常用的键，hint 提到的每个键都在那里找得到。
|
||
"""
|
||
return {
|
||
"kinds": [
|
||
{"value": "mysql", "label": "MySQL / MariaDB", "default_port": 3306,
|
||
"needs": ["host", "database", "username", "password"],
|
||
"hint": "字符集默认为 utf8mb4；如需更改，请在「高级连接参数」中添加 charset",
|
||
"advanced": [{"key": "charset", "label": "字符集", "placeholder": "utf8mb4"},
|
||
_TIMEOUT_FIELD, _MASK_FIELD]},
|
||
{"value": "postgres", "label": "PostgreSQL", "default_port": 5432,
|
||
"needs": ["host", "database", "username", "password"],
|
||
"hint": "「schema」留空时使用 public",
|
||
"advanced": [_TIMEOUT_FIELD, _MASK_FIELD]},
|
||
{"value": "oracle", "label": "Oracle", "default_port": 1521,
|
||
"needs": ["host", "username", "password"],
|
||
"hint": "service_name 与 SID 二选一，通常填写 service_name；无需安装 Instant Client。"
|
||
"若只读账号名下没有对象，请在「schema」中填写数据所在的 schema（如 ANALYTICS）",
|
||
"advanced": [
|
||
{"key": "service_name", "label": "service_name", "placeholder": "ORCLPDB1",
|
||
"help": "与 SID 二选一"},
|
||
{"key": "sid", "label": "SID", "placeholder": "ORCL", "help": "与 service_name 二选一"},
|
||
_TIMEOUT_FIELD,
|
||
_MASK_FIELD,
|
||
]},
|
||
{"value": "sqlite", "label": "SQLite（文件）", "default_port": None,
|
||
"needs": ["database"], "hint": "「数据库文件路径」请填写 .db 文件的绝对路径",
|
||
"advanced": [_TIMEOUT_FIELD, _MASK_FIELD]},
|
||
],
|
||
"supported": list(SUPPORTED_KINDS),
|
||
}
|
||
|
||
|
||
def _oracle_target(options: dict[str, Any] | None) -> tuple[str, str]:
|
||
opts = options or {}
|
||
return str(opts.get("service_name") or "").strip(), str(opts.get("sid") or "").strip()
|
||
|
||
|
||
def _settle_oracle(
|
||
row: DataSource, old_options: dict[str, Any] | None, old_database: str | None
|
||
) -> None:
|
||
"""Oracle 的 service_name / SID 只存一处：options 里，database 清空。
|
||
|
||
引擎按 `options.service_name or database` 取服务名，两者都空才看 sid。老数据
|
||
把服务名存在 database，于是新表单切到 SID 存下去会被它压住、悄悄无视；旧表单
|
||
把 service_name 框绑在 database 上、保存时又原样带回 options，新填的值同样
|
||
被压住。两处都能存，就总有一处在暗中说了算，所以存的时候收成一处：这次
|
||
改了哪一处就听哪一处；都没改就照引擎原来的取法，连的还是同一个库。
|
||
"""
|
||
if (row.kind or "").lower() != "oracle":
|
||
return
|
||
opts = {k: v for k, v in (row.options or {}).items()
|
||
if k not in ("service_name", "sid") or str(v or "").strip()}
|
||
database = (row.database or "").strip()
|
||
picked = any(_oracle_target(opts)) and _oracle_target(opts) != _oracle_target(old_options)
|
||
typed = bool(database) and database != (old_database or "").strip()
|
||
if not picked and (typed or (database and "service_name" not in opts)):
|
||
opts["service_name"] = database
|
||
opts.pop("sid", None)
|
||
row.options = opts
|
||
row.database = None
|
||
|
||
|
||
@router.post("", response_model=DataSourceOut, status_code=201)
|
||
async def create_source(
|
||
payload: DataSourceIn, session: AsyncSession = Depends(get_session)
|
||
) -> DataSourceOut:
|
||
if payload.kind not in SUPPORTED_KINDS:
|
||
raise HTTPException(400, f"不支持「{payload.kind}」类型的数据库，目前支持：{'、'.join(SUPPORTED_KINDS)}")
|
||
exists = (await session.execute(
|
||
select(DataSource).where(DataSource.name == payload.name)
|
||
)).scalar_one_or_none()
|
||
if exists:
|
||
raise HTTPException(409, f"已存在名为「{payload.name}」的数据源，请换一个标识")
|
||
_refuse_bad_options(payload.options)
|
||
|
||
row = DataSource(
|
||
**payload.model_dump(exclude={"password"}),
|
||
password=encrypt(payload.password),
|
||
)
|
||
_settle_oracle(row, None, None)
|
||
# 表单里刚测过这份配置：保存下来就带着那次结果，不必再测一遍
|
||
row.last_check = health.draft_result(_connection(row))
|
||
session.add(row)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
return _to_out(row)
|
||
|
||
|
||
@router.patch("/{source_id}", response_model=DataSourceOut)
|
||
async def update_source(
|
||
source_id: str, payload: DataSourcePatch, session: AsyncSession = Depends(get_session)
|
||
) -> DataSourceOut:
|
||
row = await _get_or_404(session, source_id)
|
||
data = payload.model_dump(exclude_unset=True)
|
||
if "options" in data:
|
||
_refuse_bad_options(data["options"])
|
||
if "password" in data:
|
||
# 空串表示清空，非空表示换新的
|
||
data["password"] = encrypt(data["password"]) if data["password"] else None
|
||
old_options, old_database = dict(row.options or {}), row.database
|
||
before = _connection(row)
|
||
for key, value in data.items():
|
||
setattr(row, key, value)
|
||
_settle_oracle(row, old_options, old_database)
|
||
after = _connection(row)
|
||
if after != before:
|
||
# 连的已经不是测过的那个了：旧结果作废，表单里测过这一份的话换成那一次
|
||
row.last_check = health.draft_result(after)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
# 连接参数可能变了，旧连接池不能再用——否则改完密码还在用旧连接，很难排查
|
||
await engines.invalidate(source_id)
|
||
return _to_out(row)
|
||
|
||
|
||
@router.delete("/{source_id}", status_code=204)
|
||
async def delete_source(source_id: str, session: AsyncSession = Depends(get_session)) -> None:
|
||
row = await _get_or_404(session, source_id)
|
||
await session.delete(row)
|
||
await session.commit()
|
||
await engines.invalidate(source_id)
|
||
|
||
|
||
#: 测连接最多等多久。内网库防火墙丢包时驱动默认要等一分钟，弹窗里干转一分钟
|
||
#: 等于没有反馈
|
||
_TEST_TIMEOUT = 15
|
||
|
||
|
||
_WRONG_PASSWORD = ("用户名或密码错误", "请核对「用户名」和「密码」；编辑时密码留空表示沿用已保存的密码")
|
||
_CHECK_HOST = "请检查「主机」和「端口」；内网数据库请确认本机可以访问（防火墙、VPN）"
|
||
|
||
|
||
def _explain_connect(e: BaseException, kind: str) -> tuple[str, str]:
|
||
"""连库失败的常见原因。驱动的原文九成能定位问题，但得先翻译成表单上的说法。"""
|
||
low = str(e).lower()
|
||
if any(s in low for s in ("password authentication failed", "access denied", "ora-01017",
|
||
"invalid username/password", "(1045")):
|
||
return _WRONG_PASSWORD
|
||
if "ora-12514" in low:
|
||
return "Oracle 无法识别该 service_name", "请核对 service_name；部分旧版数据库只提供 SID，可改用 SID 后重试"
|
||
if "ora-12505" in low:
|
||
return "Oracle 无法识别该 SID", "请核对 SID，或改用 service_name"
|
||
if "service_name 或 sid" in low:
|
||
return "未填写 service_name 或 SID", "请在「service_name」中填写，或改用 SID"
|
||
if "unknown database" in low or "(1049" in low or (
|
||
"does not exist" in low and "database" in low
|
||
):
|
||
return "服务器上不存在该数据库", "请核对「数据库」一栏"
|
||
if "unable to open database file" in low:
|
||
return "无法打开该数据库文件", "请确认路径是绝对路径、文件存在，且服务端进程有读取权限"
|
||
if "file is not a database" in low:
|
||
return "该文件不是 SQLite 数据库", "请确认路径指向的是 .db 文件"
|
||
if "no module named" in low:
|
||
return "服务端未安装该数据库的驱动", "请执行 pip install 'agentlab-backend[db]'，然后重启服务端"
|
||
# 下面按错误类别换成数据库的说法：通用的那几句说的是 API（API Key、Base URL），这里是数据库。
|
||
# 按类别判断，不比较 explain 的文字——那句话改一个字，这里就再也对不上
|
||
category = classify(e)
|
||
if category == AUTH:
|
||
return _WRONG_PASSWORD
|
||
if category == NETWORK:
|
||
return "网络不通，或数据库服务未启动", _CHECK_HOST
|
||
if category == TIMEOUT:
|
||
return "数据库长时间没有响应", _CHECK_HOST
|
||
reason, hint = explain(e)
|
||
if kind == "sqlite" and not hint:
|
||
hint = "请检查数据库文件的路径"
|
||
return reason, hint
|
||
|
||
|
||
def _safe_url(source: Any) -> str:
|
||
try:
|
||
return build_url(source) # 遮掉密码的版本
|
||
except Exception: # noqa: BLE001 - 连串都拼不出来时，原因已经在 error 里了
|
||
return ""
|
||
|
||
|
||
def _missing_sqlite_file(source: Any) -> tuple[str, str, str] | None:
|
||
"""SQLite 的路径指向不存在的文件时，返回 (reason, hint, detail)。
|
||
|
||
可写模式下 SQLite 连一个不存在的路径会当场建一个空库：测试说「连得上」，
|
||
连的却是个空文件，「只测不存」的接口还往磁盘上写了东西。没填路径时连的
|
||
是内存库，一样是假的「连得上」。按 SQLite 自己的解析方式判断（不展开 ~）。
|
||
"""
|
||
from pathlib import Path
|
||
|
||
if (source.kind or "").lower() != "sqlite":
|
||
return None
|
||
path = source.database or ""
|
||
if not path.strip():
|
||
return "未填写数据库文件路径", "请在「数据库文件路径」中填写 .db 文件的绝对路径", "database 为空"
|
||
if path == ":memory:" or Path(path).is_file():
|
||
return None
|
||
return ("无法打开该数据库文件",
|
||
"该路径下没有文件。请确认路径是绝对路径且文件存在（~ 不会被展开）",
|
||
f"文件不存在：{path}")
|
||
|
||
|
||
async def _probe(source: Any, *, cached: bool) -> dict[str, Any]:
|
||
"""真连一次。cached=False 用一次性的 engine，测完就扔——草稿不该进连接池缓存，
|
||
否则改完配置再测，拿到的还是上一版的连接。"""
|
||
import asyncio
|
||
import time
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
from sqlalchemy.pool import NullPool
|
||
|
||
started = time.perf_counter()
|
||
missing = _missing_sqlite_file(source)
|
||
if missing:
|
||
reason, hint, detail = missing
|
||
return {"ok": False, "error": f"连接失败：{reason}", "hint": hint, "detail": detail,
|
||
"elapsed_ms": 0, "url": _safe_url(source)}
|
||
engine = None
|
||
try:
|
||
if cached:
|
||
engine = await engines.get(source)
|
||
else:
|
||
url, extra = engine_args(source)
|
||
engine = create_async_engine(url, poolclass=NullPool, **extra)
|
||
async with asyncio.timeout(_TEST_TIMEOUT):
|
||
async with engine.connect() as conn:
|
||
oracle = (source.kind or "").lower() == "oracle"
|
||
await conn.execute(text("SELECT 1 FROM DUAL" if oracle else "SELECT 1"))
|
||
return {
|
||
"ok": True,
|
||
"elapsed_ms": int((time.perf_counter() - started) * 1000),
|
||
"url": _safe_url(source),
|
||
}
|
||
except Exception as e: # noqa: BLE001
|
||
reason, hint = _explain_connect(e, (source.kind or "").lower())
|
||
return {
|
||
"ok": False,
|
||
"error": f"连接失败：{reason}",
|
||
"hint": hint,
|
||
"detail": raw(e),
|
||
"elapsed_ms": int((time.perf_counter() - started) * 1000),
|
||
"url": _safe_url(source),
|
||
}
|
||
finally:
|
||
if engine is not None and not cached:
|
||
await engine.dispose()
|
||
|
||
|
||
class DataSourceTestIn(BaseModel):
|
||
"""一份还没保存（或改了还没保存）的配置。带 id 表示在编辑已有的那个。"""
|
||
|
||
id: str | None = None
|
||
name: str = ""
|
||
kind: str
|
||
host: str | None = None
|
||
port: int | None = None
|
||
database: str | None = None
|
||
username: str | None = None
|
||
password: str | None = None # 编辑时留空=沿用已保存的
|
||
options: dict[str, Any] = Field(default_factory=dict)
|
||
readonly: bool = True
|
||
|
||
|
||
def _draft_source(payload: DataSourceTestIn, saved: DataSource | None) -> DataSource:
|
||
"""拼一个不进会话的临时数据源，只拿来连一下。
|
||
|
||
编辑时密码框留空表示不改——测连接也得用存着的那个，否则改个端口想测一下，
|
||
只会得到一句"密码不对"。
|
||
"""
|
||
if payload.password:
|
||
password = encrypt(payload.password)
|
||
else:
|
||
password = saved.password if saved else None
|
||
return DataSource(
|
||
name=payload.name or (saved.name if saved else "draft"),
|
||
kind=payload.kind, host=payload.host, port=payload.port, database=payload.database,
|
||
username=payload.username, password=password, options=dict(payload.options or {}),
|
||
readonly=payload.readonly,
|
||
)
|
||
|
||
|
||
@router.post("/test")
|
||
async def test_draft(
|
||
payload: DataSourceTestIn, session: AsyncSession = Depends(get_session)
|
||
) -> dict[str, Any]:
|
||
"""测一份没保存的配置，不落库。
|
||
|
||
以前只能测已存在的数据源：填错一项要保存→关弹窗→点测试→看提示→再打开改，
|
||
接内网 Oracle 这类填错一个字段就连不上的东西，来回四五趟。
|
||
"""
|
||
if payload.kind not in SUPPORTED_KINDS:
|
||
return {"ok": False, "error": f"不支持「{payload.kind}」类型的数据库",
|
||
"hint": f"目前支持：{'、'.join(SUPPORTED_KINDS)}", "detail": "", "elapsed_ms": 0, "url": ""}
|
||
saved = await session.get(DataSource, payload.id) if payload.id else None
|
||
draft = _draft_source(payload, saved)
|
||
result = await _probe(draft, cached=False)
|
||
last, key = _probe_record(result), _connection(draft)
|
||
health.remember_draft(key, last)
|
||
if saved is not None and key is not None and key == _connection(saved):
|
||
# 编辑框里没改连接、只是再测一次：测的就是已保存的那份
|
||
saved.last_check = last
|
||
await session.commit()
|
||
return result
|
||
|
||
|
||
@router.post("/{source_id}/test")
|
||
async def test_source(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||
"""测连接。失败时说清原因和怎么办，驱动的原始报错放在 detail——这类问题九成靠它定位。"""
|
||
row = await _get_or_404(session, source_id)
|
||
result = await _probe(row, cached=True)
|
||
row.last_check = _probe_record(result)
|
||
await session.commit()
|
||
return result
|
||
|
||
|
||
class IntrospectPreview(BaseModel):
|
||
"""只看不存的探查结果。缓存、同步时间都没动。"""
|
||
|
||
dry_run: bool = True
|
||
#: 这次探的是哪个 schema，"" 是默认 schema
|
||
schema_: str = Field(default="", serialization_alias="schema")
|
||
table_count: int = 0
|
||
#: 带 schema 前缀的对象全名，最多 200 个（和真探查的上限一样）
|
||
tables: list[str] = Field(default_factory=list)
|
||
truncated: bool = False
|
||
total: int = 0
|
||
schema_error: str = ""
|
||
available_schemas: list[str] = Field(default_factory=list)
|
||
|
||
|
||
@router.post("/{source_id}/introspect", response_model=DataSourceOut | IntrospectPreview)
|
||
async def introspect_source(
|
||
source_id: str,
|
||
schema: str | None = None,
|
||
dry_run: bool = Query(default=False, description="只返回探到的结构，不写缓存"),
|
||
session: AsyncSession = Depends(get_session),
|
||
) -> DataSourceOut | IntrospectPreview:
|
||
"""探查结构并缓存。显式动作，不做后台轮询——生产库不该被实验工具定时扫。
|
||
|
||
dry_run=true 只看不存：「换个 schema 看看」以前一探就把缓存换掉，用户随后
|
||
选了「不改」，助手此刻看到的也已经是另一套表了。
|
||
"""
|
||
row = await _get_or_404(session, source_id)
|
||
try:
|
||
cache = await introspect_mod.introspect(row, schema=schema)
|
||
except Exception as e: # noqa: BLE001
|
||
reason, hint = _explain_connect(e, (row.kind or "").lower())
|
||
raise HTTPException(400, f"探查表结构失败：{reason}" + (f"。{hint}" if hint else "")) from e
|
||
if dry_run:
|
||
tables = cache.get("tables") or {}
|
||
return IntrospectPreview(
|
||
schema_=str(cache.get("schema") or ""),
|
||
table_count=len(tables),
|
||
tables=[m.get("qualified", n) for n, m in tables.items()],
|
||
truncated=bool(cache.get("truncated")),
|
||
total=int(cache.get("total") or 0),
|
||
schema_error=str(cache.get("error") or "") if cache.get("failed") else "",
|
||
available_schemas=list(cache.get("available_schemas") or []),
|
||
)
|
||
row.schema_cache = cache
|
||
# 失败也要把缓存存下来——里面的 available_schemas 正是下一步的线索——
|
||
# 但**不能盖上"已同步"的戳**。以前盖了，于是界面显示同步过、工具却说
|
||
# "还没探查过"，两边都不说真话（run 554a0f92）
|
||
if not row.schema_cache.get("failed"):
|
||
row.schema_synced_at = datetime.now(timezone.utc)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
return _to_out(row)
|
||
|
||
|
||
@router.get("/{source_id}/schema")
|
||
async def get_schema(source_id: str, table: str | None = None,
|
||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||
row = await _get_or_404(session, source_id)
|
||
if table:
|
||
meta = introspect_mod.find_table(row, table)
|
||
return {
|
||
"table": table,
|
||
# 给模型看的那段文本照旧；界面画表格用下面的结构化字段
|
||
"detail": introspect_mod.describe_table(row, table),
|
||
"found": meta is not None,
|
||
"qualified": meta.get("qualified", table) if meta else None,
|
||
"kind": ("view" if meta.get("is_view") else "table") if meta else None,
|
||
"comment": meta.get("comment") if meta else None,
|
||
"columns": introspect_mod.table_columns(meta) if meta else [],
|
||
}
|
||
return {
|
||
"tables": introspect_mod.table_names(row),
|
||
"summary": introspect_mod.summary(row),
|
||
"synced_at": row.schema_synced_at.isoformat() if row.schema_synced_at else None,
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 上传表格
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
@router.post("/upload", response_model=dict, status_code=201)
|
||
async def upload_table(
|
||
file: UploadFile = File(...),
|
||
name: str = Form(...),
|
||
description: str = Form(""),
|
||
header_row: int = Form(1),
|
||
session: AsyncSession = Depends(get_session),
|
||
) -> dict[str, Any]:
|
||
"""把 Excel / CSV 变成一个可以用 SQL 查的数据源。
|
||
|
||
走 SQLite：落成一个 .db 文件，再建一条 kind='sqlite' 的数据源指过去。
|
||
下游一行都不用改——SQL 守卫、结构探查、db_query__<name> 工具、查询快照进
|
||
工件库、Copilot 的数据源清单，全都白拿。
|
||
|
||
这条路存在的理由不是"多支持一种格式"：表格传进知识库只能被切块检索，
|
||
数字就成了模型从片段里读出来的，而这个项目的地基是"所有算术下沉到
|
||
SQL 或口径卡"。表格必须变成表。
|
||
|
||
**同名就地替换**，不新建。数据源名字会成为工具名（db_query__sales），
|
||
而工具名写进了保存过的工作流——重传时另起一个 sales_2 等于悄悄让那些图失效。
|
||
"""
|
||
import re as _re
|
||
|
||
if not _re.match(_NAME_PATTERN, name or ""):
|
||
raise HTTPException(
|
||
400, "名称必须以小写字母开头，只能包含字母、数字和下划线（名称会成为工具名的一部分）"
|
||
)
|
||
|
||
# 边读边数，超了立刻停。和知识库上传同一套：那个检查拦的是"入库"，
|
||
# 不是"占内存"，读完再判等于先把内存吃掉
|
||
limit = settings.max_upload_mb * 1024 * 1024
|
||
pieces: list[bytes] = []
|
||
total = 0
|
||
while piece := await file.read(1024 * 1024):
|
||
total += len(piece)
|
||
if total > limit:
|
||
raise HTTPException(413, f"文件超过 {settings.max_upload_mb} MB")
|
||
pieces.append(piece)
|
||
raw = b"".join(pieces)
|
||
|
||
tables_dir = settings.uploads_dir / "tables"
|
||
tables_dir.mkdir(parents=True, exist_ok=True)
|
||
db_path = tables_dir / f"{name}.db"
|
||
|
||
from app.data.tabular import UnsupportedTable, load_into
|
||
|
||
existing = (await session.execute(
|
||
select(DataSource).where(DataSource.name == name)
|
||
)).scalar_one_or_none()
|
||
if existing and existing.kind != "sqlite":
|
||
raise HTTPException(
|
||
409, f"已存在名为「{name}」的数据源（非上传表格），请换一个名称"
|
||
)
|
||
|
||
try:
|
||
report = load_into(str(db_path), raw, file.filename or name, header_row=header_row)
|
||
except UnsupportedTable as e:
|
||
raise HTTPException(400, str(e)) from e
|
||
except Exception as e: # noqa: BLE001
|
||
raise HTTPException(
|
||
400, f"导入失败：{first_line(e) if isinstance(e, ValueError) else explain(e)[0]}。"
|
||
"请确认文件是未加密的 Excel 或 CSV，且表头行号填写正确",
|
||
) from e
|
||
|
||
row = existing
|
||
if row is None:
|
||
row = DataSource(
|
||
name=name, kind="sqlite", database=str(db_path),
|
||
readonly=True, description=description,
|
||
)
|
||
session.add(row)
|
||
else:
|
||
row.database = str(db_path)
|
||
if description:
|
||
row.description = description
|
||
# 换了文件内容，缓存的连接还指着旧的那个 engine
|
||
await engines.invalidate(row.id)
|
||
|
||
await session.flush()
|
||
# 建完立刻探查：不探的话 db_query 工具的 description 里没有表清单，
|
||
# 模型得先花一步去问"有哪些表"
|
||
try:
|
||
row.schema_cache = await introspect_mod.introspect(row)
|
||
row.schema_synced_at = datetime.now(timezone.utc)
|
||
except Exception: # noqa: BLE001 - 探查失败不该让导入白做，用户可以手动再探
|
||
pass
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
|
||
return {
|
||
"source": _to_out(row).model_dump(),
|
||
"replaced": existing is not None,
|
||
# 推断出的列名和类型原样回显：表头行取错了（比如文件前两行是标题），
|
||
# 这里当场就能看出来——列名会变成一行数据
|
||
"tables": [
|
||
{
|
||
"name": t.name, "sheet": t.source_name, "rows": t.rows,
|
||
"columns": [
|
||
{"name": c, "type": ty} for c, ty in zip(t.columns, t.types)
|
||
],
|
||
}
|
||
for t in report.tables
|
||
],
|
||
}
|