"""账本(T-P0:DDL 初始化 + 连接纪律;CRUD/预扣在 T-P1 落地)。 工程纪律: - D-P10:sqlite3 是同步库,所有 DB 调用必须经 asyncio.to_thread(由调用方 routes 层包装;本模块保持同步实现,可测试性好)。连接 check_same_thread=False + threading.Lock 串行化(沿用 ReviewQueue 模式:每操作新连接 + 全局锁)。 - WAL 模式(§3)。 - D-P11:预扣用原子 UPDATE,见 try_hold(T-P1 实现)。 """ from __future__ import annotations import sqlite3 import threading from pathlib import Path from typing import Any, Dict, List, Optional # 四表 + 两索引(§3,一字不改) _SCHEMA = """ CREATE TABLE IF NOT EXISTS students( id INTEGER PRIMARY KEY, name TEXT NOT NULL, class TEXT DEFAULT '', status TEXT NOT NULL DEFAULT 'active', balance_milli INTEGER NOT NULL DEFAULT 0, daily_cap_milli INTEGER NOT NULL DEFAULT 5000, spent_today_milli INTEGER NOT NULL DEFAULT 0, spent_date TEXT DEFAULT ''); CREATE TABLE IF NOT EXISTS proxy_keys( id INTEGER PRIMARY KEY, key_hash TEXT UNIQUE NOT NULL, key_prefix TEXT NOT NULL, student_id INTEGER NOT NULL REFERENCES students(id), created_ts INTEGER NOT NULL, revoked INTEGER NOT NULL DEFAULT 0, rpm_cap INTEGER NOT NULL DEFAULT 10, day_cap_req INTEGER NOT NULL DEFAULT 200, req_today INTEGER NOT NULL DEFAULT 0, req_date TEXT DEFAULT ''); CREATE TABLE IF NOT EXISTS usage_ledger( request_id TEXT PRIMARY KEY, ts INTEGER NOT NULL, key_id INTEGER NOT NULL, model TEXT NOT NULL, bucket TEXT NOT NULL DEFAULT 'default', in_miss_tok INTEGER NOT NULL DEFAULT 0, in_hit_tok INTEGER NOT NULL DEFAULT 0, out_tok INTEGER NOT NULL DEFAULT 0, gateway_cached INTEGER NOT NULL DEFAULT 0, upstream_cost_milli INTEGER NOT NULL DEFAULT 0, charged_milli INTEGER NOT NULL DEFAULT 0, margin_milli INTEGER NOT NULL DEFAULT 0, ttfb_ms INTEGER, total_ms INTEGER, status TEXT NOT NULL); CREATE TABLE IF NOT EXISTS semcache( cache_key TEXT PRIMARY KEY, bucket TEXT NOT NULL, q_norm TEXT NOT NULL, answer TEXT NOT NULL, model TEXT NOT NULL, created_ts INTEGER NOT NULL, ttl_ts INTEGER NOT NULL, doc_version INTEGER NOT NULL DEFAULT 1, hits INTEGER NOT NULL DEFAULT 0); CREATE INDEX IF NOT EXISTS idx_semcache_bucket ON semcache(bucket, ttl_ts); CREATE INDEX IF NOT EXISTS idx_ledger_ts ON usage_ledger(ts); """ _TABLES = ("students", "proxy_keys", "usage_ledger", "semcache") class Ledger: """代理层账本(sqlite WAL;同步实现,调用方负责 to_thread)。""" def __init__(self, db_path: str | Path): self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.Lock() self._init_db() # ---------- 初始化 ---------- @classmethod def init_db(cls, db_path: str | Path) -> "Ledger": """工厂(§6 签名):建库建表(幂等)。""" return cls(db_path) def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path, check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conn def _init_db(self) -> None: with self._lock, self._connect() as conn: conn.executescript(_SCHEMA) # ---------- 自省(T-P0 验收用) ---------- def table_names(self) -> List[str]: """列出已建表名(测试验收)。""" with self._lock, self._connect() as conn: rows = conn.execute( "SELECT name FROM sqlite_master WHERE type='table'").fetchall() return [r["name"] for r in rows] def index_names(self) -> List[str]: with self._lock, self._connect() as conn: rows = conn.execute( "SELECT name FROM sqlite_master WHERE type='index' AND name LIKE 'idx_%'" ).fetchall() return [r["name"] for r in rows]