vincent/Lottery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(BIZ-129): 方案 C 用户系统 DB 层 + 业务层（BIZ-127 #1/#2/#5/#6/#8 对齐）

Browse Source 
- auth_db.py: users/sessions/password_reset_tokens/operation_logs 4 表 + migrate_legacy_schema 增量升级
- auth.py: hash_password(rounds=12) + create_session(token_hash=sha256) + authenticate(锁定→清零) + cleanup_op_logs(90d)
- 业务层已就绪，未完成：auth_api 蓝图 / require_auth 装饰器 / 限流 / pytest / 前端 3 页

Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
costcodev
2026-09-05 01:31:34 +08:00
parent dcc7339384
commit ab9b62e9b3
2 changed files with 723 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
auth.py
+480
View File
@@ -0,0 +1,480 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
用户系统业务逻辑 — BIZ-129 方案 C
封装：
- 密码哈希（bcrypt）
- 用户 CRUD（创建/查询/校验）
- Session 创建/校验/撤销
- 密码重置 token 生成/校验
- 操作日志写入
"""
import os
import json
import secrets
import sqlite3
import hashlib
import bcrypt
from datetime import datetime, timedelta
from auth_db import get_conn, init_db, now_iso
# ============================================================
# 常量（BIZ-127 8 条约束对齐）
# ============================================================
# #1 bcrypt cost 显式锁 12，防止未来库升级改了默认值
BCRYPT_ROUNDS = 12
# #6 登录失败锁定策略
LOGIN_MAX_FAILED = 5 # 连续 5 次失败锁
LOGIN_LOCKOUT_MINUTES = 15 # 锁 15 分钟
# #8 op_logs 保留天数（cleanup 默认值）
OP_LOG_RETENTION_DAYS = 90
# ============================================================
# 常量
# ============================================================
SESSION_COOKIE_NAME = 'lotto_session'
SESSION_LIFETIME_DAYS = 30
RESET_TOKEN_LIFETIME_HOURS = 2
RESET_TOKEN_BYTES = 32
SESSION_ID_BYTES = 32
def _sha256(s: str) -> str:
"""#2 session 存 hash 入库：sha256(cookie_value).hexdigest()。"""
return hashlib.sha256(s.encode('utf-8')).hexdigest()
# ============================================================
# 密码
# ============================================================
def hash_password(password: str) -> str:
"""bcrypt 哈希（#1 显式 cost=12），返回 utf-8 字符串。"""
return bcrypt.hashpw(
password.encode('utf-8'),
bcrypt.gensalt(rounds=BCRYPT_ROUNDS),
).decode('utf-8')
def verify_password(password: str, password_hash: str) -> bool:
try:
return bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8'))
except (ValueError, TypeError):
return False
# ============================================================
# 用户
# ============================================================
def create_user(username: str, password: str, role: str = 'user') -> int:
"""创建用户，返回 user_id。username 必须唯一。"""
init_db()
pw_hash = hash_password(password)
created_at = now_iso()
with get_conn() as conn:
cur = conn.cursor()
cur.execute(
"INSERT INTO users (username, password_hash, role, is_active, created_at) "
"VALUES (?, ?, ?, 1, ?)",
(username, pw_hash, role, created_at),
)
return cur.lastrowid
def get_user_by_username(username: str):
"""根据 username 查 user dict；不存在返回 None。"""
with get_conn() as conn:
row = conn.execute(
"SELECT * FROM users WHERE username = ?", (username,)
).fetchone()
return dict(row) if row else None
def get_user_by_id(user_id: int):
with get_conn() as conn:
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
return dict(row) if row else None
def count_users() -> int:
with get_conn() as conn:
row = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()
return row['c'] if row else 0
def update_password(user_id: int, new_password: str):
pw_hash = hash_password(new_password)
with get_conn() as conn:
conn.execute(
"UPDATE users SET password_hash = ? WHERE id = ?", (pw_hash, user_id)
)
def touch_last_login(user_id: int):
with get_conn() as conn:
conn.execute(
"UPDATE users SET last_login_at = ? WHERE id = ?", (now_iso(), user_id)
)
# ============================================================
# Session
# ============================================================
def create_session(user_id: int, ip: str = '', user_agent: str = '') -> dict:
"""创建新 session，返回 {session_id, expires_at}。
#2 BIZ-127：DB 存 token_hash（sha256(cookie_value)），cookie 只下发随机串。
session_id 字段名保留作为外部 cookie 值，token_hash 是入库列。
"""
init_db()
session_id = secrets.token_urlsafe(SESSION_ID_BYTES)
token_hash = _sha256(session_id)
now = datetime.utcnow()
expires = now + timedelta(days=SESSION_LIFETIME_DAYS)
with get_conn() as conn:
conn.execute(
"INSERT INTO sessions (token_hash, user_id, created_at, expires_at, ip, user_agent) "
"VALUES (?, ?, ?, ?, ?, ?)",
(
token_hash, user_id,
now.isoformat(), expires.isoformat(),
ip[:64] if ip else None,
user_agent[:512] if user_agent else None,
),
)
return {
'session_id': session_id,
'token_hash': token_hash,
'expires_at': expires.strftime('%Y-%m-%dT%H:%M:%S'),
}
def lookup_session(session_id: str):
"""校验 session_id，返回 user dict（含 session 元数据）或 None。
#2 BIZ-127：按 sha256(session_id) 即 token_hash 入库对比，
外部传入明文 cookie，内部永远不存明文。
"""
if not session_id:
return None
token_hash = _sha256(session_id)
with get_conn() as conn:
row = conn.execute(
"""
SELECT s.token_hash, s.user_id, s.expires_at, s.ip,
u.username, u.role, u.is_active
FROM sessions s
JOIN users u ON u.id = s.user_id
WHERE s.token_hash = ?
AND s.revoked = 0
AND u.is_active = 1
""",
(token_hash,),
).fetchone()
if not row:
return None
try:
expires = datetime.fromisoformat(row['expires_at'])
except (ValueError, TypeError):
return None
if expires < datetime.utcnow():
# 过期 — 标记撤销
conn.execute(
"UPDATE sessions SET revoked = 1 WHERE token_hash = ?", (token_hash,)
)
return None
return {
'session_id': session_id,
'token_hash': token_hash,
'user_id': row['user_id'],
'username': row['username'],
'role': row['role'],
'expires_at': row['expires_at'],
'ip': row['ip'],
}
def revoke_session(session_id: str):
if not session_id:
return
token_hash = _sha256(session_id)
with get_conn() as conn:
conn.execute(
"UPDATE sessions SET revoked = 1 WHERE token_hash = ?", (token_hash,)
)
def cleanup_expired_sessions():
with get_conn() as conn:
conn.execute(
"UPDATE sessions SET revoked = 1 WHERE revoked = 0 AND expires_at < ?",
(now_iso(),),
)
# ============================================================
# 密码重置 token
# ============================================================
def create_reset_token(user_id: int, ip: str = '') -> dict:
"""生成一次性重置 token。"""
token = secrets.token_urlsafe(RESET_TOKEN_BYTES)
now = datetime.utcnow()
expires = now + timedelta(hours=RESET_TOKEN_LIFETIME_HOURS)
with get_conn() as conn:
conn.execute(
"INSERT INTO password_reset_tokens (user_id, token, created_at, expires_at, ip) "
"VALUES (?, ?, ?, ?, ?)",
(user_id, token, now.isoformat(), expires.isoformat(), ip[:64] if ip else None),
)
return {
'token': token,
'expires_at': expires.strftime('%Y-%m-%dT%H:%M:%S'),
'lifetime_hours': RESET_TOKEN_LIFETIME_HOURS,
}
def consume_reset_token(token: str):
"""校验并消费重置 token。返回 user dict 或 None。
校验规则：存在、未过期、未使用。
消费规则：原子地将 used_at 写入；如果已被使用则返回 None。
"""
if not token:
return None
with get_conn() as conn:
row = conn.execute(
"""
SELECT t.id, t.user_id, t.expires_at, t.used_at,
u.username, u.is_active
FROM password_reset_tokens t
JOIN users u ON u.id = t.user_id
WHERE t.token = ?
""",
(token,),
).fetchone()
if not row:
return None
if row['used_at']:
return None
try:
expires = datetime.fromisoformat(row['expires_at'])
except (ValueError, TypeError):
return None
if expires < datetime.utcnow():
return None
if not row['is_active']:
return None
# 原子消费
cur = conn.execute(
"UPDATE password_reset_tokens SET used_at = ? "
"WHERE id = ? AND used_at IS NULL",
(now_iso(), row['id']),
)
if cur.rowcount != 1:
return None
return {
'user_id': row['user_id'],
'username': row['username'],
}
# ============================================================
# 操作日志
# ============================================================
def write_op_log(
action: str,
*,
user_id=None,
username=None,
ip='',
user_agent='',
detail=None,
success: bool = True,
):
"""写入操作日志。detail 可以是 dict（会被 JSON 化）。"""
detail_str = json.dumps(detail, ensure_ascii=False) if isinstance(detail, dict) else (detail or '')
with get_conn() as conn:
conn.execute(
"""
INSERT INTO operation_logs
(user_id, username, action, ip, user_agent, detail, success, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id, username, action,
ip[:64] if ip else None,
user_agent[:512] if user_agent else None,
detail_str[:2000] if detail_str else None,
1 if success else 0,
now_iso(),
),
)
def list_recent_op_logs(limit: int = 50, action: str = None):
"""查询最近操作日志（管理员用）。"""
limit = max(1, min(int(limit), 200))
with get_conn() as conn:
if action:
rows = conn.execute(
"SELECT * FROM operation_logs WHERE action = ? "
"ORDER BY id DESC LIMIT ?",
(action, limit),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM operation_logs ORDER BY id DESC LIMIT ?",
(limit,),
).fetchall()
return [dict(r) for r in rows]
# ============================================================
# 登录业务层（#6 失败锁定 BIZ-127）
# ============================================================
def is_account_locked(user_row) -> bool:
"""检查账号是否在锁定期内。返回 True 表示被锁。"""
if not user_row:
return False
locked_until = user_row.get('locked_until')
if not locked_until:
return False
try:
until = datetime.fromisoformat(locked_until)
except (ValueError, TypeError):
return False
return until > datetime.utcnow()
def _lock_account(user_id: int):
"""锁定账号 15 分钟，重置失败计数。"""
until = datetime.utcnow() + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)
with get_conn() as conn:
conn.execute(
"UPDATE users SET failed_attempts = 0, locked_until = ? WHERE id = ?",
(until.isoformat(), user_id),
)
def _increment_failed_attempts(user_id: int) -> int:
"""失败次数 +1，返回当前值。"""
with get_conn() as conn:
conn.execute(
"UPDATE users SET failed_attempts = failed_attempts + 1 WHERE id = ?",
(user_id,),
)
row = conn.execute(
"SELECT failed_attempts FROM users WHERE id = ?", (user_id,)
).fetchone()
return int(row['failed_attempts']) if row else 0
def _clear_failed_attempts(user_id: int):
with get_conn() as conn:
conn.execute(
"UPDATE users SET failed_attempts = 0, locked_until = NULL WHERE id = ?",
(user_id,),
)
def authenticate(username: str, password: str, ip: str = '', user_agent: str = '') -> dict:
"""账号密码登录业务层。
严格顺序（#6 BIZ-127）：
1. 查 user → 不存在返 None（记 login_fail 日志）
2. 检查 locked_until → 已锁返 'locked'（记 login_fail 日志）
3. 检查 is_active → 禁用返 None（记 login_fail 日志）
4. bcrypt verify → 失败计数 +1，达 5 次锁 15 分钟（记 login_fail 日志）
5. 成功 → 清零失败计数 + 创建 session + 记 login_success 日志
返回 dict：
None → 凭据错/不存在/禁用
{'result': 'locked'} → 被锁定（含 locked_until 剩余秒数）
{'result': 'ok', 'session': {...}, 'user': {...}}
"""
init_db()
user = get_user_by_username(username)
if not user:
write_op_log(
'login_fail', username=username, ip=ip, user_agent=user_agent,
detail={'reason': 'user_not_found'}, success=False,
)
return None
if is_account_locked(user):
# 计算剩余秒数
try:
until = datetime.fromisoformat(user['locked_until'])
remaining = int((until - datetime.utcnow()).total_seconds())
except Exception:
remaining = LOGIN_LOCKOUT_MINUTES * 60
write_op_log(
'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
detail={'reason': 'account_locked', 'remaining_seconds': remaining},
success=False,
)
return {'result': 'locked', 'locked_until': user['locked_until'], 'remaining_seconds': remaining}
if not user.get('is_active'):
write_op_log(
'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
detail={'reason': 'inactive'}, success=False,
)
return None
if not verify_password(password, user['password_hash']):
cnt = _increment_failed_attempts(user['id'])
if cnt >= LOGIN_MAX_FAILED:
_lock_account(user['id'])
write_op_log(
'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
detail={'reason': 'wrong_password', 'locked': True, 'failed_attempts': cnt},
success=False,
)
return {'result': 'locked', 'reason': 'too_many_failures',
'locked_until': (datetime.utcnow() + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)).isoformat()}
write_op_log(
'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
detail={'reason': 'wrong_password', 'failed_attempts': cnt},
success=False,
)
return None
# 成功
_clear_failed_attempts(user['id'])
touch_last_login(user['id'])
session = create_session(user['id'], ip=ip, user_agent=user_agent)
write_op_log(
'login_success', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
detail={'session_token_hash_prefix': session['token_hash'][:12]},
success=True,
)
return {
'result': 'ok',
'session': session,
'user': {
'id': user['id'],
'username': user['username'],
'role': user['role'],
'tenant_id': user.get('tenant_id'),
},
}
# ============================================================
# 操作日志清理（#8 BIZ-127，90 天保留）
# ============================================================
def cleanup_op_logs(days: int = OP_LOG_RETENTION_DAYS) -> int:
"""删除早于 days 天的操作日志，返回删除行数。
应由 cron 入口每日调用（例如每天凌晨 3:00）。
"""
cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
with get_conn() as conn:
cur = conn.execute("DELETE FROM operation_logs WHERE created_at < ?", (cutoff,))
return cur.rowcount
auth_db.py
+243
View File
@@ -0,0 +1,243 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
用户系统数据库模块 — BIZ-129 方案 C
表：
- users 用户账号（含多用户预留字段 / 登录失败锁定字段）
- sessions 服务端 session（存 token_hash，不存明文 cookie 值）
- password_reset_tokens 找回密码一次性 token
- operation_logs 操作日志（90 天保留）
约束对齐 — BIZ-127 架构侧清单：
#2 sessions.token_hash = sha256(cookie_value).hexdigest()
#5 users.tenant_id / users.status（数据层预留，UI 不写）
#6 users.failed_attempts / users.locked_until（登录失败锁定）
升级路径：旧装已存在的 `users` / `sessions` 表会被 `migrate_legacy_schema()` 增量升级，
不会丢数据。新装直接走 CREATE TABLE。
"""
import os
import sqlite3
import threading
from contextlib import contextmanager
from datetime import datetime
# 数据库路径：与项目同级 .data/lottery_auth.db
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, '.data')
DB_PATH = os.path.join(DATA_DIR, 'lottery_auth.db')
_db_lock = threading.Lock()
def _ensure_data_dir():
os.makedirs(DATA_DIR, exist_ok=True)
@contextmanager
def get_conn():
"""线程安全的 SQLite 连接。"""
_ensure_data_dir()
conn = sqlite3.connect(DB_PATH, timeout=10, isolation_level=None)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA journal_mode = WAL")
try:
yield conn
finally:
conn.close()
# ============================================================
# 表结构（BIZ-127 8 条约束对齐版）
# ============================================================
SCHEMA = [
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'user', -- 'admin' / 'user'
is_active INTEGER NOT NULL DEFAULT 1,
tenant_id TEXT, -- #5 多用户预留
status TEXT NOT NULL DEFAULT 'active',-- #5 active/disabled
failed_attempts INTEGER NOT NULL DEFAULT 0, -- #6 失败计数
locked_until TEXT, -- #6 锁定到期时间
created_at TEXT NOT NULL,
last_login_at TEXT
)
""",
"""
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
token_hash TEXT NOT NULL UNIQUE, -- #2 sha256(cookie_value)
user_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
expires_at TEXT NOT NULL, -- 默认 30 天
ip TEXT,
user_agent TEXT,
revoked INTEGER NOT NULL DEFAULT 0,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
""",
"""
CREATE TABLE IF NOT EXISTS password_reset_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
token TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
expires_at TEXT NOT NULL, -- 2 小时有效
used_at TEXT, -- 使用时间（一次性）
ip TEXT,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
""",
"""
CREATE TABLE IF NOT EXISTS operation_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER, -- 可为空（如登录失败）
username TEXT, -- 冗余存储便于查询
action TEXT NOT NULL, -- login_success / login_fail / ...
ip TEXT,
user_agent TEXT,
detail TEXT, -- JSON 字符串
success INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL
)
""",
"CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id)",
"CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at)",
"CREATE INDEX IF NOT EXISTS idx_sessions_token_hash ON sessions(token_hash)",
"CREATE INDEX IF NOT EXISTS idx_reset_tokens_token ON password_reset_tokens(token)",
"CREATE INDEX IF NOT EXISTS idx_op_logs_user ON operation_logs(user_id)",
"CREATE INDEX IF NOT EXISTS idx_op_logs_action ON operation_logs(action)",
"CREATE INDEX IF NOT EXISTS idx_op_logs_created ON operation_logs(created_at)",
]
# ============================================================
# 数据迁移 SQL — 旧装兼容
# ============================================================
# 旧版 users 缺 tenant_id / status / failed_attempts / locked_until
# 旧版 sessions 用 session_id 明文，需 rename → token_hash 并回填 sha256
#
# 注意：SQLite ALTER TABLE 不能改 UNIQUE 约束列名，所以采用
# 1. add column token_hash_temp（无 UNIQUE）
# 2. UPDATE token_hash_temp = sha256(session_id)
# 3. drop 原 session_id 列（不支持 DROP COLUMN 的旧版 SQLite 走 fallback）
# 4. rename token_hash_temp → token_hash
# 的方式。新版 SQLite（≥3.35）支持 DROP COLUMN，本脚本两种路径都跑。
def _table_has_column(conn, table: str, column: str) -> bool:
rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
return any(r['name'] == column for r in rows)
def _sqlite_supports_drop_column(conn) -> bool:
row = conn.execute("SELECT sqlite_version() AS v").fetchone()
try:
# SQLite 3.35.0 起支持 DROP COLUMN
major, minor, _ = (int(x) for x in row['v'].split('.'))
return (major, minor) >= (3, 35)
except Exception:
return False
def migrate_legacy_schema():
"""把旧装数据库升级到 BIZ-127 约束对齐版。幂等。
返回 dict 报告做了哪些迁移（便于测试断言）。
"""
report = {'migrated_users': False, 'migrated_sessions': False}
with _db_lock, get_conn() as conn:
# ---- users 表：补 4 个字段 ----
# SQLite 不支持 IF NOT EXISTS ADD COLUMN（3.35 之前），用 PRAGMA 查
if not _table_has_column(conn, 'users', 'tenant_id'):
conn.execute("ALTER TABLE users ADD COLUMN tenant_id TEXT")
report['migrated_users'] = True
if not _table_has_column(conn, 'users', 'status'):
conn.execute("ALTER TABLE users ADD COLUMN status TEXT NOT NULL DEFAULT 'active'")
report['migrated_users'] = True
if not _table_has_column(conn, 'users', 'failed_attempts'):
conn.execute("ALTER TABLE users ADD COLUMN failed_attempts INTEGER NOT NULL DEFAULT 0")
report['migrated_users'] = True
if not _table_has_column(conn, 'users', 'locked_until'):
conn.execute("ALTER TABLE users ADD COLUMN locked_until TEXT")
report['migrated_users'] = True
# ---- sessions 表：session_id → token_hash（sha256 化）----
if _table_has_column(conn, 'sessions', 'session_id') \
and not _table_has_column(conn, 'sessions', 'token_hash'):
# 用 temp 列 + 改名方式，不依赖 DROP COLUMN
conn.execute("ALTER TABLE sessions ADD COLUMN token_hash_temp TEXT")
# 回填 sha256(session_id)
rows = conn.execute(
"SELECT id, session_id FROM sessions WHERE session_id IS NOT NULL"
).fetchall()
for r in rows:
# Python 端算 sha256（避免 SQLite 函数兼容性问题）
import hashlib
digest = hashlib.sha256(r['session_id'].encode('utf-8')).hexdigest()
conn.execute(
"UPDATE sessions SET token_hash_temp = ? WHERE id = ?",
(digest, r['id']),
)
if _sqlite_supports_drop_column(conn):
# 新版 SQLite：直接 drop 旧列 + rename temp
conn.execute("ALTER TABLE sessions DROP COLUMN session_id")
conn.execute("ALTER TABLE sessions RENAME COLUMN token_hash_temp TO token_hash")
else:
# 旧版 SQLite：建新表，复制数据，drop 旧表，rename
conn.execute("""
CREATE TABLE sessions_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
token_hash TEXT NOT NULL UNIQUE,
user_id INTEGER NOT NULL,
created_at TEXT NOT NULL,
expires_at TEXT NOT NULL,
ip TEXT,
user_agent TEXT,
revoked INTEGER NOT NULL DEFAULT 0,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
""")
conn.execute("""
INSERT INTO sessions_new
(id, token_hash, user_id, created_at, expires_at, ip, user_agent, revoked)
SELECT id, token_hash_temp, user_id, created_at, expires_at, ip, user_agent, revoked
FROM sessions
""")
conn.execute("DROP TABLE sessions")
conn.execute("ALTER TABLE sessions_new RENAME TO sessions")
report['migrated_sessions'] = True
return report
# ============================================================
# 初始化
# ============================================================
def init_db():
"""初始化所有表（幂等），并对旧装做 schema 迁移。
新装：CREATE TABLE 走完即可。
旧装：CREATE TABLE IF NOT EXISTS 跳过已存在表，迁移 SQL 补齐字段。
"""
with _db_lock, get_conn() as conn:
cur = conn.cursor()
for stmt in SCHEMA:
cur.execute(stmt)
# 迁移在 init 完成后单独跑（需要先有 users/sessions 表存在）
migrate_legacy_schema()
def now_iso():
return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S')
if __name__ == '__main__':
init_db()
print(f"✅ 初始化完成：{DB_PATH}")