04 · 数据库层：无 ORM 的 SQLite，15 张表与种子数据

CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    username TEXT UNIQUE NOT NULL,
    password_hash TEXT NOT NULL,
    salt TEXT NOT NULL,
    role TEXT NOT NULL DEFAULT 'employee',      -- admin / employee
    department TEXT NOT NULL DEFAULT '',         -- 部门，多个用逗号分隔
    display_name TEXT NOT NULL DEFAULT '',
    must_change_password INTEGER NOT NULL DEFAULT 0,  -- 1=下次登录强制改密
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    conversation_id INTEGER NOT NULL,
    role TEXT NOT NULL,                          -- user / assistant
    content TEXT NOT NULL,
    refs_json TEXT NOT NULL DEFAULT '[]',        -- 引用来源（RAG 命中的知识块）
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS llm_calls (            -- LLM 网关调用审计
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER, assistant_id INTEGER, run_id INTEGER,
    provider TEXT NOT NULL DEFAULT '',
    model TEXT NOT NULL DEFAULT '',
    prompt_tokens INTEGER NOT NULL DEFAULT 0,
    completion_tokens INTEGER NOT NULL DEFAULT 0,
    latency_ms INTEGER NOT NULL DEFAULT 0,
    attempts INTEGER NOT NULL DEFAULT 1,          -- 实际尝试次数（含重试）
    status TEXT NOT NULL DEFAULT 'success',       -- success / degraded / failed
    error TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL
);

def get_conn(db_path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path, timeout=10)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    # 多线程（FastAPI 线程池 + 工作流线程）并发读写：等锁而不是立刻抛 database is locked
    conn.execute("PRAGMA busy_timeout = 5000")
    return conn

conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")

python -c "import sqlite3; c=sqlite3.connect('data/platform.db'); print(c.execute('PRAGMA journal_mode').fetchone())"
# ('wal',)

def init_db(db_path: str) -> None:
    """建表 + 迁移 + 写入种子数据（幂等）。"""
    path = resolve_db_path(db_path)
    conn = get_conn(path)
    try:
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA synchronous = NORMAL")
        conn.executescript(SCHEMA)      # ① 全部 CREATE TABLE IF NOT EXISTS
        _migrate(conn)                  # ② 增量迁移：老库补新列
        _seed(conn)                     # ③ 种子数据（只在空表时写入）
        conn.commit()
    finally:
        conn.close()

def _migrate(conn: sqlite3.Connection) -> None:
    """旧库增量迁移：CREATE TABLE IF NOT EXISTS 覆盖不了的新增列在这里补。"""
    tables = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
    def has_col(pragma: str, col: str) -> bool:
        return any(r["name"] == col for r in conn.execute(pragma))
    if "users" in tables and not has_col("PRAGMA table_info(users)", "must_change_password"):
        conn.execute("ALTER TABLE users ADD COLUMN must_change_password INTEGER NOT NULL DEFAULT 0")
    # ... sessions.expires_at、knowledge_bases.departments 同理

def record_audit(conn, *, user_id, username, action, detail="", ip="") -> None:
    """写一条安全审计日志。调用方负责 commit（多数场景与业务写在同一事务里）。"""
    conn.execute(
        "INSERT INTO audit_logs(user_id,username,action,detail,ip,created_at) VALUES(?,?,?,?,?,?)",
        (user_id, (username or "")[:64], action, (detail or "")[:500], (ip or "")[:64], now()),
    )