Important: All scripts/ paths are relative to this skill directory. Run with: cd {skill_dir} && python scripts/sql_linker.py ... Or use the cwd parameter of execute_shell_command.

SQL-Linker — Multi-Database CRUD + Audit / 多数据库增删查改 + 审计

概述 / Overview

（中文） SQL-Linker 提供跨数据库的 CRUD 操作能力，支持 MySQL、PostgreSQL、SQLite 三种主流数据库。内置审计日志模块，每次操作自动记录操作人身份、IP、SQL 语句、操作时间，确保数据可溯源、安全可控。

（English） SQL-Linker provides cross-database CRUD operations, supporting MySQL, PostgreSQL, and SQLite, with a built-in audit trail module that automatically records operator identity, IP, SQL statements, and timestamps for full traceability and compliance.

支持的数据库 / Supported Databases

配置 / Configuration

（中文） 配置文件位于工作区 .sql_linker/ 目录下：

.sql_linker/
├── config.yaml         # 数据库连接配置（不含密码）
├── audit_config.json   # 审计配置 + 用户名（发布时排除）
└── .env                # 敏感凭据（发布时排除）

（English） Configuration lives in .sql_linker/ at the workspace root:

.sql_linker/
├── config.yaml         # Database connection (no passwords)
├── audit_config.json   # Audit config + username (excluded from publishing)
└── .env                # Sensitive credentials (excluded from publishing)

config.yaml

（中文） type 字段指定数据库类型：

type: mysql                  # 数据库类型：mysql | postgres | sqlite
host: localhost
port: 3306
database: mydb
user: admin001
password_env: DB_PASSWORD   # .env 中的键名
read_only: false            # true = 禁止所有写操作
max_rows: 1000             # 查询结果行数上限
timeout: 30                 # 连接超时（秒）

（English） The type field specifies the database type:

type: mysql                  # mysql | postgres | sqlite
host: localhost
port: 3306
database: mydb
user: admin001
password_env: DB_PASSWORD   # Key name in .env
read_only: false           # true = block all write operations
max_rows: 1000             # Query result limit
timeout: 30                # Connection timeout (seconds)

audit_config.json

（中文） 审计配置 + 用户名。首次使用前需运行 init_user("你的姓名") 注册用户名：

{
  "username": "HR",
  "audit": {
    "enabled": true,
    "log_table": "sql_audit_log",
    "log_select": false,
    "mask_values": true
  }
}

（English） Audit config + username. Run init_user("YourName") on first use to register the username:

{
  "username": "HR",
  "audit": {
    "enabled": true,
    "log_table": "sql_audit_log",
    "log_select": false,
    "mask_values": true
  }
}

SQLite 配置示例 / SQLite Config Example

（中文）

type: sqlite
database: ./data/mydb.sqlite
read_only: false
max_rows: 1000
timeout: 30

（English）

type: sqlite
database: ./data/mydb.sqlite
read_only: false
max_rows: 1000
timeout: 30

.env

（中文）

DB_PASSWORD=your_password_here

（English）

DB_PASSWORD=your_password_here

身份上下文 / Identity Context

（中文） 审计日志需要以下四个字段，优先级如下：

（English） Four fields are required for audit logging, with the following priority:

⚠️ （中文）CLI flag 优先级最高，确保审计可溯源时应使用 --user-label 和 --session-id。 ⚠️ CLI flags take highest priority — always use --user-label and --session-id for real traceability.

Python API

初始化 + 注册用户名 / Initialize + Register Username

（中文）

from sql_linker import SQLLinker

linker = SQLLinker()
linker.connect()

# 首次使用：注册用户名（写入 audit_config.json，持久化）
linker.init_user("HR")

（English）

from sql_linker import SQLLinker

linker = SQLLinker()
linker.connect()

# First use: register username (persisted to audit_config.json)
linker.init_user("HR")

自动身份上下文 / Auto Identity Context

（中文）

# 自动获取：IP 从 socket 检测，username 从 audit_config.json 读取
linker.set_user_context_auto()

# 或手动完整传入（从消息 metadata 获取 label 和 session_id）
linker.set_user_context(
    user_name=linker._username,           # 从配置读
    user_label="openclaw-control-ui",    # 从消息 metadata
    ip_address="192.168.1.48",          # socket 自动检测
    session_id="agent:hr:dashboard:..."  # 从消息 context
)

（English）

# Auto: IP from socket, username from audit_config.json
linker.set_user_context_auto()

# Or manual full context (from message metadata)
linker.set_user_context(
    user_name=linker._username,           # from config
    user_label="openclaw-control-ui",    # from message metadata
    ip_address="192.168.1.48",          # socket auto-detect
    session_id="agent:hr:dashboard:..."  # from message context
)

查询 SELECT / Query

（中文）

results = linker.query(
    "SELECT id, name FROM users WHERE status = %s",
    ("active",)
)
# 返回字典列表；失败时返回空列表
# SELECT 操作是否记录取决于 config 中 audit.log_select

（English）

results = linker.query(
    "SELECT id, name FROM users WHERE status = %s",
    ("active",)
)
# Returns list of dicts; empty list on failure
# SELECT is logged only if audit.log_select is true in audit_config.json

插入 INSERT / Insert

（中文）

row_id = linker.insert("users", {
    "name": "张三",
    "email": "[email protected]",
    "status": "active"
})
# 返回自增 ID 或影响行数；自动写审计日志

（English）

row_id = linker.insert("users", {
    "name": "张三",
    "email": "[email protected]",
    "status": "active"
})
# Returns auto-increment ID or rowcount; audit log written automatically

批量插入 / Batch Insert

（中文）

data = [
    {"name": "李四", "email": "[email protected]", "status": "active"},
    {"name": "王五", "email": "[email protected]", "status": "active"},
]
count = linker.batch_insert("users", data)

（English）

data = [
    {"name": "李四", "email": "[email protected]", "status": "active"},
    {"name": "王五", "email": "[email protected]", "status": "active"},
]
count = linker.batch_insert("users", data)

更新 UPDATE / Update

（中文）

count = linker.update(
    "users",
    {"status": "inactive", "updated_at": "2024-01-01"},
    "id = %s AND status = %s",
    (1, "active")
)
# WHERE 条件（不含 WHERE 关键字）；自动写审计日志

（English）

count = linker.update(
    "users",
    {"status": "inactive", "updated_at": "2024-01-01"},
    "id = %s AND status = %s",
    (1, "active")
)
# WHERE clause (no WHERE keyword); audit log written automatically

删除 DELETE / Delete

（中文）

count = linker.delete("users", "id = %s AND status = %s", (1, "active"))

（English）

count = linker.delete("users", "id = %s AND status = %s", (1, "active"))

事务控制 / Transaction Control

（中文）

linker.begin()
linker.insert("users", {"name": "test"})
linker.update("products", {"stock": 50}, "id = %s", (1,))
linker.commit()    # 提交 — 或：
# linker.rollback()  # 出错时回滚

（English）

linker.begin()
linker.insert("users", {"name": "test"})
linker.update("products", {"stock": 50}, "id = %s", (1,))
linker.commit()    # Commit — or:
# linker.rollback()  # Roll back on error

只读模式 / Read-Only Mode

（中文） 当 config.yaml 中 read_only: true 时，所有写操作返回 0 并打印警告：

linker.insert("users", {...})   # 只读模式禁止插入
linker.update("users", {...})   # 只读模式禁止更新
linker.delete("users", ...)     # 只读模式禁止删除

（English） When config.yaml has read_only: true, all write operations return 0 and print a warning:

linker.insert("users", {...})   # 只读模式禁止插入 / Read-only mode: insert blocked
linker.update("users", {...})   # 只读模式禁止更新 / Read-only mode: update blocked
linker.delete("users", ...)     # 只读模式禁止删除 / Read-only mode: delete blocked

关闭连接 / Close

（中文）

linker.close()

（English）

linker.close()

审计日志 / Audit Trail

审计表结构 / Audit Table Schema

审计表会在首次写入时自动创建（如果不存在）。结构如下：

CREATE TABLE sql_audit_log (
    id            BIGINT AUTO_INCREMENT PRIMARY KEY,
    log_time      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    user_name     VARCHAR(128)     -- 操作人姓名
    user_label    VARCHAR(128)     -- 会话标签（来源系统）
    ip_address    VARCHAR(64)      -- 客户端 IP
    session_id    VARCHAR(128)     -- OpenClaw session key
    db_type       VARCHAR(32)      -- mysql / postgres / sqlite
    operation     VARCHAR(16)      -- SELECT / INSERT / UPDATE / DELETE / BATCH_INSERT
    table_name    VARCHAR(128)     -- 操作的目标表
    sql_statement TEXT             -- 脱敏后的 SQL
    rows_affected INT              -- 影响行数
    status        VARCHAR(16)      -- SUCCESS / FAILED
    error_msg     TEXT             -- 失败时的错误信息
);

查询审计日志 / Query Audit Logs

（中文）

# 查询某人的所有操作记录
logs = linker._audit.query_logs(
    user_name="HR",
    start_time="2026-05-01 00:00:00",
    end_time="2026-05-31 23:59:59",
    limit=100
)
for row in logs:
    print(row["log_time"], row["operation"], row["table_name"], row["status"])

# 查询某张表的所有操作
logs = linker._audit.query_logs(table_name="users", limit=50)

（English）

# Query all operations by a specific user
logs = linker._audit.query_logs(
    user_name="HR",
    start_time="2026-05-01 00:00:00",
    end_time="2026-05-31 23:59:59",
    limit=100
)
for row in logs:
    print(row["log_time"], row["operation"], row["table_name"], row["status"])

# Query all operations on a specific table
logs = linker._audit.query_logs(table_name="users", limit=50)

初始化审计表 / Initialize Audit Table

（中文）

python scripts/sql_linker.py audit-setup

python scripts/sql_linker.py audit-logs               # 最近 100 条
python scripts/sql_linker.py audit-logs HR           # 按操作人过滤
python scripts/sql_linker.py audit-logs HR 50        # 指定数量

（English）

python scripts/sql_linker.py audit-setup

python scripts/sql_linker.py audit-logs               # Last 100 records
python scripts/sql_linker.py audit-logs HR           # Filter by user
python scripts/sql_linker.py audit-logs HR 50        # With limit

命令行接口 / Command-Line Interface

python scripts/sql_linker.py [flags] \x3Caction> [options]

Flags:
  --user-label \x3Clabel>    Override user_label (for real audit traceability)
  --session-id \x3Cid>       Override session_id  (for real audit traceability)

Actions: query, insert, update, delete, audit-setup, audit-logs

（中文）完整示例（含身份上下文）/ Full example with identity context

# 查询（身份自动检测 / identity auto-detected）
python scripts/sql_linker.py query "SELECT * FROM users LIMIT 10"

# 插入 — 必须传入真实 user_label 和 session_id
python scripts/sql_linker.py \
  --user-label "openclaw-control-ui" \
  --session-id "agent:hr:dashboard:2454324d-1bd5-4e04-9746-1e5df62bb9a7" \
  insert "users" '{"name":"张三","email":"[email protected]"}'

# 更新
python scripts/sql_linker.py \
  --user-label "openclaw-control-ui" \
  --session-id "agent:hr:dashboard:2454324d-..." \
  update "users" '{"status":"inactive"}' "id = 1"

# 删除
python scripts/sql_linker.py \
  --user-label "openclaw-control-ui" \
  --session-id "agent:hr:dashboard:2454324d-..." \
  delete "users" "id = 1"

（English）Full example with identity context

# Query (identity auto-detected)
python scripts/sql_linker.py query "SELECT * FROM users LIMIT 10"

# Insert — always pass real user_label and session_id
python scripts/sql_linker.py \
  --user-label "openclaw-control-ui" \
  --session-id "agent:hr:dashboard:2454324d-1bd5-4e04-9746-1e5df62bb9a7" \
  insert "users" '{"name":"Zhang San","email":"[email protected]"}'

# Update
python scripts/sql_linker.py \
  --user-label "openclaw-control-ui" \
  --session-id "agent:hr:dashboard:2454324d-..." \
  update "users" '{"status":"inactive"}' "id = 1"

# Delete
python scripts/sql_linker.py \
  --user-label "openclaw-control-ui" \
  --session-id "agent:hr:dashboard:2454324d-..." \
  delete "users" "id = 1"

Note / 注意: On Windows, set $env:PYTHONIOENCODING='utf-8' before running queries that return Chinese characters to avoid GBK encoding errors. Windows PowerShell: JSON arguments MUST use double quotes — single quotes cause escape loss.

错误处理 / Error Handling

（中文）

（English）

安全建议 / Security Notes

（中文）

（English）