#!/usr/bin/env python3 # -*- coding: utf-8 -*- """SQLite3 存储层 —— 账号 / 设置 / 登录记录 / 运行日志 / 运行状态 / 缓存 / 页面快照 为什么用 SQLite 而不是散落的文件 * 远程管理：Docker 里只挂一个 data 卷，所有状态都在这一个文件里，备份 = 拷一个文件 * 并发：running 时有几十个线程写日志，文件追加要自己加锁，SQLite WAL 直接解决 * 查询：/api/log/history 能按账号取最近 N 行，不需要自己维护文件轮转 表 settings 系统配置（name -> JSON），目前只有 llm kv 通用缓存（任意 key -> JSON），例如 StudyClient 试出来的接口前缀 accounts 账号 / 密码 / 会话 cookie / 课程缓存 / 勾选 / 开关 login_log 每次登录尝试（平台原话） run_log 每个账号（key）的运行日志 run_state 运行状态快照，进程重启后 /api/runs 还能看到上次跑到哪 snapshots 课程页原始 HTML（站点改版时定位选择器用） 并发模型 * WAL + busy_timeout=15s：读写互不阻塞，写锁竞争自动重试 * 每个线程一条连接（threading.local）：ThreadPoolExecutor 里多线程写日志不会串 * 本进程内的写操作再用一把 RLock 串起来，避免多线程 commit 抖动（读不加锁） * 首次调用自动建库建表，并把旧的 data/settings.json、data/logs/*.log 导入后改名 （原文件保留为 *.imported，回滚旧版本时改名回去即可） """ import json import os import re import sqlite3 import threading import time HERE = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.environ.get("DATA_DIR") or os.path.join(HERE, "data") DB_PATH = os.environ.get("DB_PATH") or os.path.join(DATA_DIR, "study.db") # 旧的文件式存储位置（只在首次启动时导入一次） LEGACY_SETTINGS = os.path.join(DATA_DIR, "settings.json") LEGACY_LOGDIR = os.path.join(DATA_DIR, "logs") # 每个账号保留的日志行数，超出就丢最旧的（远程长期挂机，不能无限涨） RUN_LOG_KEEP = int(os.environ.get("RUN_LOG_KEEP", "2000") or 2000) # 页面快照保留条数 SNAPSHOT_KEEP = int(os.environ.get("SNAPSHOT_KEEP", "200") or 200) SCHEMA = """ CREATE TABLE IF NOT EXISTS settings ( name TEXT PRIMARY KEY, value TEXT NOT NULL, updated REAL NOT NULL ); CREATE TABLE IF NOT EXISTS kv ( key TEXT PRIMARY KEY, value TEXT, updated REAL NOT NULL ); CREATE TABLE IF NOT EXISTS accounts ( id TEXT PRIMARY KEY, username TEXT, password TEXT, note TEXT, user TEXT, cookies TEXT, courses TEXT, selected TEXT, status TEXT, last_error TEXT, last_error_at REAL, opt_quiz INTEGER DEFAULT 0, opt_feedback INTEGER DEFAULT 1, login_at REAL, created REAL, updated REAL ); CREATE TABLE IF NOT EXISTS login_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, username TEXT, ok INTEGER, code TEXT, elapsed REAL, stage TEXT, error TEXT ); CREATE TABLE IF NOT EXISTS run_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, key TEXT NOT NULL, ts REAL, line TEXT ); CREATE INDEX IF NOT EXISTS idx_run_log_key ON run_log(key, id); CREATE TABLE IF NOT EXISTS run_state ( key TEXT PRIMARY KEY, state TEXT, updated REAL ); CREATE TABLE IF NOT EXISTS snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, plan_id TEXT, project_id TEXT, url TEXT, html TEXT, created REAL ); """ _local = threading.local() _lock = threading.RLock() # 只保护“写”和一次性初始化 _ready = False _write_count = {} # ------------------------------------------------------------------ 基础 def dumps(o): return json.dumps(o, ensure_ascii=False, separators=(",", ":"), default=str) def loads(s, default=None): if not s: return default try: return json.loads(s) except Exception: return default def conn(): """取当前线程的连接（没有就建一条），顺带保证表结构已建好。""" global _ready c = getattr(_local, "conn", None) if c is None: os.makedirs(os.path.dirname(os.path.abspath(DB_PATH)), exist_ok=True) # check_same_thread=False：连接跟着线程走，线程退出后由 GC 关闭， # 关闭动作可能发生在别的线程里，默认的 same-thread 检查会直接报错。 c = sqlite3.connect(DB_PATH, timeout=15, check_same_thread=False) c.row_factory = sqlite3.Row c.execute("PRAGMA journal_mode=WAL") c.execute("PRAGMA synchronous=NORMAL") c.execute("PRAGMA busy_timeout=15000") _local.conn = c if not _ready: with _lock: if not _ready: c.executescript(SCHEMA) c.commit() _migrate(c) _ready = True return c def query(sql, params=()): return conn().execute(sql, params).fetchall() def query_one(sql, params=()): rows = query(sql, params) return rows[0] if rows else None def write(sql, params=()): with _lock: c = conn() cur = c.execute(sql, params) c.commit() return cur def init(): """显式初始化（启动时调用一次，确保库文件已建好）。""" conn() def db_info(): try: size = os.path.getsize(DB_PATH) if os.path.exists(DB_PATH) else 0 except Exception: size = 0 counts = {} for t in ("accounts", "run_log", "login_log", "snapshots", "run_state", "kv", "settings"): try: counts[t] = (query_one("SELECT COUNT(*) AS n FROM %s" % t) or {"n": 0})["n"] except Exception: counts[t] = -1 return {"path": DB_PATH, "size": size, "counts": counts} # ------------------------------------------------------------------ 迁移 def _migrate(c): """把旧的文件式存储导入 SQLite；成功后把原文件改名成 *.imported。""" # 1) data/settings.json -> settings if os.path.exists(LEGACY_SETTINGS): try: with open(LEGACY_SETTINGS, encoding="utf-8") as f: data = json.load(f) or {} if isinstance(data.get("llm"), dict) and get_setting("llm") is None: set_setting("llm", data["llm"]) os.replace(LEGACY_SETTINGS, LEGACY_SETTINGS + ".imported") except Exception: pass # 2) data/logs/login.log -> login_log if os.path.isdir(LEGACY_LOGDIR): for name in sorted(os.listdir(LEGACY_LOGDIR)): if not name.endswith(".log"): continue path = os.path.join(LEGACY_LOGDIR, name) try: with open(path, encoding="utf-8", errors="replace") as f: lines = [x.rstrip("\n") for x in f if x.strip()] except Exception: continue if name == "login.log": for ln in lines: p = ln.split("\t") if len(p) < 4: continue el = re.sub(r"[^0-9.]", "", p[4]) if len(p) > 4 else "0" login_log_add(p[1], p[2] == "OK", p[3], float(el or 0), p[5] if len(p) > 5 else "", p[6] if len(p) > 6 else "", ts=p[0]) else: # data/logs/.log -> run_log（key 是文件名去掉 .log） key = name[:-4] for ln in lines: run_log_add(key, ln, ts=None) try: os.replace(path, path + ".imported") except Exception: pass # 3) data/study_session.json（study_video.py 的会话）-> kv legacy_session = os.path.join(DATA_DIR, "study_session.json") if os.path.exists(legacy_session) and kv_get("study_session") is None: try: with open(legacy_session, encoding="utf-8") as f: kv_set("study_session", json.load(f)) os.replace(legacy_session, legacy_session + ".imported") except Exception: pass # ------------------------------------------------------------------ settings def get_setting(name, default=None): r = query_one("SELECT value FROM settings WHERE name=?", (name,)) return loads(r["value"], default) if r else default def set_setting(name, value): write("INSERT INTO settings(name, value, updated) VALUES(?,?,?) " "ON CONFLICT(name) DO UPDATE SET value=excluded.value, updated=excluded.updated", (name, dumps(value), time.time())) # ------------------------------------------------------------------ kv 缓存 def kv_get(key, default=None): r = query_one("SELECT value FROM kv WHERE key=?", (key,)) return loads(r["value"], default) if r else default def kv_set(key, value): write("INSERT INTO kv(key, value, updated) VALUES(?,?,?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated=excluded.updated", (key, dumps(value), time.time())) def kv_del(key): write("DELETE FROM kv WHERE key=?", (key,)) def kv_all(): return {r["key"]: loads(r["value"]) for r in query("SELECT key, value FROM kv")} # ------------------------------------------------------------------ accounts def _acc_row(r): """DB 行 -> 前端用的记录（不含密码）。""" return { "id": r["id"], "username": r["username"] or "", "note": r["note"] or "", "user": r["user"] or "", "cookies": loads(r["cookies"], []) or [], "courses": loads(r["courses"], []) or [], "selected": loads(r["selected"], []) or [], "status": r["status"] or "idle", "lastError": r["last_error"] or "", "lastErrorAt": r["last_error_at"] or 0, "optQuiz": bool(r["opt_quiz"]), "optFeedback": bool(r["opt_feedback"]), "loginAt": r["login_at"] or 0, "created": r["created"] or 0, "updated": r["updated"] or 0, "hasPassword": bool(r["password"]), } def account_list(): return [_acc_row(r) for r in query("SELECT * FROM accounts ORDER BY created, id")] def account_get(account_id): if not account_id: return None r = query_one("SELECT * FROM accounts WHERE id=?", (str(account_id),)) return _acc_row(r) if r else None def account_password(account_id): if not account_id: return "" r = query_one("SELECT password FROM accounts WHERE id=?", (str(account_id),)) return (r["password"] or "") if r else "" def account_exists(account_id): return bool(account_id) and query_one("SELECT 1 FROM accounts WHERE id=?", (str(account_id),)) is not None def account_upsert(rec): """新增/更新账号。rec['password'] 为空(None)时保留库里原有密码。 明文密码要留在库里：会话失效时后台要能自己重登，不能等用户回来输密码。 """ now = time.time() aid = str(rec.get("id") or "").strip() if not aid: aid = "a%s%04d" % (int(now * 1000), int(time.time() * 10000) % 10000) pwd = rec.get("password") if pwd is not None: pwd = str(pwd) selected = loads(dumps(rec.get("selected") or []), []) or [] write( """INSERT INTO accounts (id, username, password, note, user, cookies, courses, selected, status, last_error, last_error_at, opt_quiz, opt_feedback, login_at, created, updated) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET username = excluded.username, password = COALESCE(excluded.password, accounts.password), note = excluded.note, user = excluded.user, cookies = excluded.cookies, courses = excluded.courses, selected = excluded.selected, status = excluded.status, last_error = excluded.last_error, last_error_at = excluded.last_error_at, opt_quiz = excluded.opt_quiz, opt_feedback = excluded.opt_feedback, login_at = excluded.login_at, updated = excluded.updated""", (aid, rec.get("username") or "", pwd, rec.get("note") or "", rec.get("user") or "", dumps(rec.get("cookies") or []), dumps(rec.get("courses") or []), dumps(selected), rec.get("status") or "idle", rec.get("lastError") or "", float(rec.get("lastErrorAt") or 0), 1 if rec.get("optQuiz") is True else 0, 0 if rec.get("optFeedback") is False else 1, float(rec.get("loginAt") or 0), float(rec.get("created") or now), now)) return aid def account_save_login(account_id, user, cookies, courses): """登录成功后把新的会话写回账号（服务端自己记，不依赖前端回传）。""" rec = account_get(account_id) if not rec: return courses = courses or [] plans = {c.get("plan_id") for c in courses} rec.update({ "user": user or rec.get("user") or "", "cookies": cookies or [], "courses": courses, "selected": [p for p in (rec.get("selected") or []) if p in plans], "status": "logged_in", "lastError": "", "lastErrorAt": 0, "loginAt": time.time(), }) rec.pop("hasPassword", None) account_upsert(rec) def account_delete(account_id): aid = str(account_id or "") write("DELETE FROM accounts WHERE id=?", (aid,)) write("DELETE FROM run_state WHERE key=?", (aid,)) write("DELETE FROM run_log WHERE key=?", (aid,)) # ------------------------------------------------------------------ 登录记录 def login_log_add(username, ok, code, elapsed, stage="", error="", ts=None): write("INSERT INTO login_log(ts, username, ok, code, elapsed, stage, error) VALUES(?,?,?,?,?,?,?)", (ts or time.strftime("%Y-%m-%d %H:%M:%S"), username, 1 if ok else 0, str(code or ""), float(elapsed or 0), str(stage or ""), str(error or "").replace("\n", " ")[:500])) def login_log_tail(n=40): """返回旧版文本行格式（ts\\t用户名\\tOK\\tcode\\t耗时\\tstage\\t错误），前端不用改。""" n = max(1, min(int(n or 40), 500)) rows = query("SELECT * FROM login_log ORDER BY id DESC LIMIT ?", (n,)) out = [] for r in reversed(rows): out.append("%s\t%s\t%s\t%s\t%.1fs\t%s\t%s" % ( r["ts"], r["username"], "OK" if r["ok"] else "FAIL", r["code"], r["elapsed"] or 0, r["stage"] or "", r["error"] or "")) return out # ------------------------------------------------------------------ 运行日志 def run_log_add(key, line, ts=None): key = str(key or "") write("INSERT INTO run_log(key, ts, line) VALUES(?,?,?)", (key, ts or time.time(), line)) n = _write_count.get("run_log", 0) + 1 _write_count["run_log"] = n if n % 200 == 0: # 每 200 行清一次，避免每行都跑一遍 DELETE write("""DELETE FROM run_log WHERE key=? AND id NOT IN (SELECT id FROM run_log WHERE key=? ORDER BY id DESC LIMIT ?)""", (key, key, RUN_LOG_KEEP)) def run_log_tail(key, n=200): n = max(1, min(int(n or 200), RUN_LOG_KEEP)) rows = query("SELECT line FROM run_log WHERE key=? ORDER BY id DESC LIMIT ?", (str(key), n)) return [r["line"] for r in reversed(rows)] def run_log_keys(): return [{"key": r["key"], "lines": r["n"], "updated": r["ts"]} for r in query( "SELECT key, COUNT(*) AS n, MAX(ts) AS ts FROM run_log GROUP BY key ORDER BY ts DESC")] def run_log_clear(key): write("DELETE FROM run_log WHERE key=?", (str(key),)) # ------------------------------------------------------------------ 运行状态 def run_state_set(key, state): st = dict(state or {}) st.pop("log", None) # 日志单独进 run_log 表，不在这里重复一份 write("INSERT INTO run_state(key, state, updated) VALUES(?,?,?) " "ON CONFLICT(key) DO UPDATE SET state=excluded.state, updated=excluded.updated", (str(key), dumps(st), time.time())) def run_state_all(): out = {} for r in query("SELECT key, state FROM run_state"): st = loads(r["state"], None) if isinstance(st, dict): out[r["key"]] = st return out # ------------------------------------------------------------------ 页面快照 def snapshot_add(plan_id, project_id, url, html): cur = write("INSERT INTO snapshots(plan_id, project_id, url, html, created) VALUES(?,?,?,?,?)", (str(plan_id or ""), str(project_id or ""), url or "", html or "", time.time())) sid = cur.lastrowid write("""DELETE FROM snapshots WHERE id NOT IN (SELECT id FROM snapshots ORDER BY id DESC LIMIT ?)""", (SNAPSHOT_KEEP,)) return sid def snapshot_get(sid): r = query_one("SELECT * FROM snapshots WHERE id=?", (int(sid),)) return dict(r) if r else None def snapshot_list(limit=50): limit = max(1, min(int(limit or 50), 500)) rows = query("SELECT id, plan_id, project_id, url, created, LENGTH(html) AS size " "FROM snapshots ORDER BY id DESC LIMIT ?", (limit,)) return [dict(r) for r in rows]