日期:2026 年 4 月 6 日——4 月 12 日
项目:绘画 AI 博弈小游戏 —— 人机对抗绘画猜词与心理解读系统
本周目标与产出
本周完成了游戏数据库较为完整的设计与实现,对上周的models.py草稿文件进行了修改和完善,包括:
- ✅ 7张核心数据表设计(用户、房间、玩家、回合、猜词、行为、报告)
- ✅ ER图与字段设计理由明确
- ✅ 高频查询优化(索引+查询策略)
- ✅ 上下文管理器实现(事务安全)
- ✅ 完整的 CRUD 操作与真实流程演示
一、数据库整体设计
1.1 ER 图
┌─────────────────────────────────────────────────────────────┐ │ 数据库架构 │ └─────────────────────────────────────────────────────────────┘ ┌──────────────┐ │ users │ (用户表) ├──────────────┤ │ id (PK) │ │ nickname │ │ avatar │ │ created_at │ │ total_score │ │ games_played │ │ games_won │ └────┬─────────┘ │ 1:N ┌──────────┴──────────┐ │ │ ┌──────────▼──────────┐ ┌───────▼────────────┐ │ rooms │ │ psychology_ │ │ (房间表) │ │ reports │ ├─────────────────────┤ │ (心理报告表) │ │ id (PK) │ ├───────────────────┤ │ code (Unique) ◄────┼─┤ id (PK) │ │ host_id (FK) │ │ user_id (FK) ────┤ │ status │ │ emotion_score │ │ difficulty │ │ pressure_score │ │ max_players │ │ summary_text │ │ current_round │ │ created_at │ │ max_rounds │ └───────────────────┘ │ created_at │ │ updated_at │ └────┬────────────────┘ │ 1:N ┌────┴─────────────────────────┐ │ room_players (关联表) │ │ (房间+玩家多对多) │ ├───────────────────────────────┤ │ room_id (FK) ┐ │ │ user_id (FK) ├─ (Composite PK)│ │ join_order │ │ │ is_active │ │ │ score │ │ │ joined_at │ │ └────┬─────────────────────────┘ │ 1:N ┌─────▼──────────────────┐ │ rounds │ (回合记录) ├───────────────────────┤ │ id (PK) │ │ room_id (FK) │ │ round_number │ │ drawer_id (FK) │ │ target_word │ │ category │ │ difficulty │ │ drawing_data (Base64) │ │ ai_guess │ │ ai_confidence │ │ ai_correct │ │ human_winner_id │ │ status │ │ started_at │ │ finished_at │ └────┬────────────────────┘ │ 1:N ┌────┴──────────────────────────┐ │ │ ┌────▼──────────────────┐ ┌──────▼────────────────────┐ │ guesses │ │ drawing_behaviors │ │ (玩家猜词记录) │ │ (绘画行为数据) │ ├──────────────────────┤ ├────────────────────────────┤ │ id (PK) │ │ id (PK) │ │ round_id (FK) │ │ round_id (FK) │ │ user_id (FK) │ │ user_id (FK) │ │ guess_text │ │ stroke_count │ │ is_correct │ │ stroke_speed_avg │ │ submitted_at │ │ turn_point_density │ └──────────────────────┘ │ drawing_duration_ms │ │ undo_count │ │ eraser_count │ │ canvas_coverage │ │ symmetry_score │ │ blank_ratio │ │ color_count │ │ pressure_avg │ │ stroke_data_json (JSON) │ │ created_at │ └────────────────────────────┘二、关键表的字段设计理由
2.1 rooms 表:为什么用 code 而不是 id?
问题设定:玩家如何快速加入朋友的房间?
❌ 方案1:直接用房间 id
# 玩家需要输入完整 UUID room_id = "a1b2c3d4-e5f6-4g7h-8i9j-k1l2m3n4o5p6" # ← 太长,难以分享 # 最终结果: # - 用户体验极差:复制粘贴容易出错 # - 口头分享不现实:无法朗读 # - 分享方式局限:只能通过链接/二维码✅ 最终方案:用 6位数字房间码
# 房间码设计 code = "123456" # ← 简短、易记、易分享 # 优势: # 1. 用户友好:可直接输入、口头分享 # 2. 分享灵活:QQ/微信/语音都能快速告知 # 3. 查询高效:6位数字索引比UUID快 # 数据库设计 CREATE TABLE rooms ( id TEXT PRIMARY KEY, # 内部使用(UUID) code TEXT UNIQUE NOT NULL, # 玩家输入(6位数字) ... ); # 创建房间时的流程 room_id = str(uuid.uuid4())[:8] # 内部 UUID code = _generate_room_code() # 生成 6位数字流程对比:
1.用户A创建房间:
✅ 系统生成 code = "567890"
2.用户B加入:
❌ 输入 UUID = "a1b2c3d4"(复杂)
✅ 输入 code = "567890"(简单)
3.后端查询:
SELECT * FROM rooms WHERE code = ? # ← 有索引,O(1) 查询
2.2 drawing_behaviors 表:为什么要存 JSON?
问题设定:如何灵活存储不同维度的绘画行为数据?
❌ 方案1:每个维度建一列
CREATE TABLE drawing_behaviors ( id TEXT PRIMARY KEY, round_id TEXT, user_id TEXT, stroke_count INTEGER, -- ✅ 固定维度 stroke_speed_avg REAL, -- ✅ 固定维度 drawing_duration_ms INTEGER, -- ✅ 固定维度 undo_count INTEGER, -- ✅ 固定维度 eraser_count INTEGER, -- ✅ 固定维度 canvas_coverage REAL, -- ✅ 固定维度 symmetry_score REAL, -- ✅ 固定维度 blank_ratio REAL, -- ✅ 固定维度 color_count INTEGER, -- ✅ 固定维度 pressure_avg REAL, -- ✅ 固定维度 -- 如果后续要加"笔触方向"、"笔迹宽度变化"等新维度呢? -- ❌ 需要 ALTER TABLE 修改表结构! );❌ 问题1:无法灵活扩展
# 如果第5周想加入"笔触方向分布"维度 # 需要:ALTER TABLE drawing_behaviors ADD COLUMN direction_std REAL # 问题: # - 现有数据无法回溯计算新维度 # - 修改表结构需要停服维护 # - 不同游戏版本数据结构不一致❌ 问题2:查询复杂
-- 要计算"平均行为",需要列举所有维度 SELECT AVG(stroke_count), AVG(stroke_speed_avg), AVG(drawing_duration_ms), AVG(undo_count), AVG(eraser_count), AVG(canvas_coverage), ... FROM drawing_behaviors WHERE user_id = ? -- ← SQL 超级长!难以维护✅ 最终方案:核心维度单列 + 扩展数据 JSON
CREATE TABLE drawing_behaviors ( id TEXT PRIMARY KEY, round_id TEXT, user_id TEXT, -- 核心维度(频繁查询、需要索引) stroke_count INTEGER, stroke_speed_avg REAL, drawing_duration_ms INTEGER, undo_count INTEGER, eraser_count INTEGER, canvas_coverage REAL, symmetry_score REAL, blank_ratio REAL, color_count INTEGER, pressure_avg REAL, -- 扩展数据(JSON格式,灵活存储) stroke_data_json TEXT, -- {"direction_std": 0.45, "width_variance": 0.12, ...} created_at REAL );优势展示:
# 【第3周】原始数据 behavior = { "stroke_count": 35, "stroke_speed_avg": 120, "drawing_duration_ms": 45000, ... } # 【第5周】想加"笔触方向分布"?直接加进 JSON behavior = { "stroke_count": 35, ..., "direction_std": 0.45, # ← 新维度,放在 JSON 里 "width_variance": 0.12, # ← 新维度 } # 数据库无需改动!只是更新 stroke_data_json 字段 models.save_drawing_behavior(round_id, user_id, behavior) # 查询时: SELECT * FROM drawing_behaviors WHERE user_id = ? # ← 所有维度都能拿到,无论新旧查询对比:
# ❌ 全列方案的查询 SELECT AVG(stroke_count), AVG(stroke_speed_avg), AVG(drawing_duration_ms), AVG(undo_count), AVG(eraser_count), AVG(canvas_coverage), AVG(symmetry_score), AVG(blank_ratio), AVG(color_count), AVG(pressure_avg) FROM drawing_behaviors WHERE user_id = ? # ✅ 混合方案的查询 SELECT * FROM drawing_behaviors WHERE user_id = ? ORDER BY created_at DESC # 然后在 Python 中 behaviors = [dict(row) for row in rows] avg_behavior = { "stroke_count": sum(b["stroke_count"] for b in behaviors) / len(behaviors), ... } # 优势: # 1. SQL 简洁 # 2. 灵活扩展 # 3. 版本兼容2.3 room_players 表:为什么需要关联表?
问题设定:房间和玩家是多对多关系,怎么存储?
✅ 设计:关联表 + 复合主键
CREATE TABLE room_players ( room_id TEXT NOT NULL, user_id TEXT NOT NULL, join_order INTEGER NOT NULL, -- 加入顺序(用于轮流作画) is_active INTEGER DEFAULT 1, -- 是否仍在房间(支持离开) score INTEGER DEFAULT 0, -- 房间内积分 joined_at REAL NOT NULL, PRIMARY KEY (room_id, user_id), -- ← 复合主键,防重复加入 FOREIGN KEY (room_id) REFERENCES rooms(id), FOREIGN KEY (user_id) REFERENCES users(id) );为什么这样设计?
# 【场景1】检查玩家是否已在房间中 SELECT 1 FROM room_players WHERE room_id = ? AND user_id = ? # ← 复合主键保证最多只有一行,查询超快 # 【场景2】获取房间所有活跃玩家 SELECT u.*, rp.score, rp.join_order FROM room_players rp JOIN users u ON rp.user_id = u.id WHERE rp.room_id = ? AND rp.is_active = 1 ORDER BY rp.join_order # 【场景3】玩家离开房间 UPDATE room_players SET is_active = 0 WHERE room_id = ? AND user_id = ? # ← 软删除,保留历史记录,无需真的删行 # 【场景4】计算房间人数 SELECT COUNT(*) FROM room_players WHERE room_id = ? AND is_active = 1三、索引设计与查询优化
3.1 高频查询分析
根据游戏流程,统计查询频率:
游戏进行过程中的查询频率(每秒次数): 高频查询(>= 10次/秒): ✅ SELECT * FROM rooms WHERE code = ? 原因:玩家加入房间、获取房间信息 ✅ SELECT * FROM room_players WHERE room_id = ? AND is_active = 1 原因:获取房间玩家列表、计算人数、轮流分配 中频查询(1-10次/秒): ✅ SELECT * FROM guesses WHERE round_id = ? 原因:检查答案、统计猜对人数 ✅ SELECT * FROM drawing_behaviors WHERE round_id = ? 原因:保存行为数据、游戏结束时分析 低频查询(<1次/秒): ✅ SELECT * FROM psychology_reports WHERE user_id = ? 原因:用户查看历史报告 超低频(<1次/分钟): ✅ SELECT * FROM users ORDER BY total_score DESC LIMIT 20 原因:刷新排行榜3.2 索引策略
-- 【高频1】房间码查询 CREATE INDEX idx_rooms_code ON rooms(code); -- 原理:B-Tree 索引,O(log N) 查询 -- 场景:玩家输入房间码加入,需秒级响应 -- 【高频2】房间玩家列表 CREATE INDEX idx_room_players_room ON room_players(room_id); -- 原理:按 room_id 分组,快速定位同房间的所有玩家 -- 场景:获取房间人数、广播消息、分配画家 -- 【中频1】回合猜词 CREATE INDEX idx_guesses_round ON guesses(round_id); -- 原理:快速检索一个回合的所有猜词 -- 场景:回合结束时统计猜对人数 -- 【中频2】绘画行为 CREATE INDEX idx_drawing_behaviors_round ON drawing_behaviors(round_id); -- 原理:快速检索一个回合的行为数据 -- 场景:游戏结束时生成心理报告 -- 【低频】排行榜 CREATE INDEX idx_users_score ON users(total_score DESC); -- 原理:按积分倒序排列,支持 LIMIT 查询 -- 场景:刷新排行榜,不需要 ORDER BY 计算3.3 查询性能对比
❌ 没有索引的情况:
-- 100万玩家数据 SELECT * FROM rooms WHERE code = '567890'; -- 扫描:O(n) ≈ 100万行 -- 耗时:100-500ms ← 用户能感觉到卡顿!✅ 有索引的情况:
-- 同样 100万玩家数据 SELECT * FROM rooms WHERE code = '567890'; -- 扫描:O(log n) ≈ 20行(B-Tree 深度) -- 耗时:< 1ms ← 感觉不到延迟四、上下文管理器实现(事务安全)
4.1 问题:为什么需要上下文管理器?
理解:
1. 自动管理数据库连接,避免连接泄漏
SQLite 是文件型数据库,同一时间只能有一个写入连接。如果每次操作都手动open/close,很容易出现:
- 打开了连接,忘记关闭
- 异常发生时,连接没关闭
- 连接堆积 → 数据库锁死 → 程序崩溃
上下文管理器可以保证:无论代码是否正常执行,连接一定会关闭。
2. 自动管理事务:提交 / 回滚
数据库操作必须保证原子性:
- 成功 → 提交(commit)
- 失败 → 回滚(rollback)
如果手动写:
try: conn.execute(...) conn.commit() except: conn.rollback() finally: conn.close()每一段 DB 操作都要写,重复、冗余、容易写错。
上下文管理器自动完成:
- 无异常 → commit
- 有异常 → rollback
- 最后一定 close
让业务代码只关心逻辑,不关心事务细节。
3. 代码更简洁、更易维护、更符合工程规范
没有上下文管理器:
- 每个函数都要写重复的 try-catch-finally
- 多人协作时风格不统一
- 出问题难以排查
有了上下文管理器:
with get_db_connection() as conn: conn.execute(...)一行统一所有数据库操作规范,代码干净、可读性强、便于团队协作。
示例:
# 【例1】查询 def get_user(user_id): with get_db_connection() as conn: row = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,)).fetchone() return dict(row) if row else None # 优势: # - 自动关闭连接 ✅ # - 无需手动 try/except ✅ # 【例2】单条更新 def update_user_score(user_id, score_delta): with get_db_connection() as conn: conn.execute('UPDATE users SET total_score = total_score + ? WHERE id = ?', (score_delta, user_id)) # 优势: # - 自动 commit ✅ # - 如果出错,自动 rollback ✅五、models.py 核心 CRUD 代码详解
1. models.py 核心 CRUD 代码(关键片段)
创建房间的完整流程 步骤: 1. 生成唯一房间 ID(UUID) 2. 生成易分享房间码(6位数字) 3. 创建房间记录 4. 房主自动加入房间加入房间的完整流程 返回 (success: bool, message: str) 检查流程: 1. 房间是否存在? 2. 游戏是否已开始? 3. 用户是否已在房间中? 4. 房间是否已满? 5. 都通过则加入以下是覆盖 “房间 - 玩家 - 回合” 核心流程的 CRUD 实现:
# ========== 用户操作 ========== def create_user(nickname, avatar=''): """创建用户(C)""" user_id = str(uuid.uuid4())[:8] with get_db_connection() as conn: conn.execute( 'INSERT INTO users (id, nickname, avatar, created_at) VALUES (?, ?, ?, ?)', (user_id, nickname, avatar, time.time()) ) return user_id def get_user(user_id): """查询用户(R)""" with get_db_connection() as conn: row = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,)).fetchone() return dict(row) if row else None # ========== 房间操作 ========== def create_room(host_id, difficulty='medium', max_players=4, max_rounds=5): """创建房间(C)""" room_id = str(uuid.uuid4())[:8] code = _generate_room_code() # 生成6位随机房间码 now = time.time() with get_db_connection() as conn: # 保证房间码唯一 while conn.execute('SELECT 1 FROM rooms WHERE code = ?', (code,)).fetchone(): code = _generate_room_code() # 插入房间 conn.execute( 'INSERT INTO rooms (id, code, host_id, difficulty, max_players, max_rounds, created_at, updated_at) ' 'VALUES (?, ?, ?, ?, ?, ?, ?, ?)', (room_id, code, host_id, difficulty, max_players, max_rounds, now, now) ) # 房主自动加入房间 conn.execute( 'INSERT INTO room_players (room_id, user_id, join_order, joined_at) VALUES (?, ?, 1, ?)', (room_id, host_id, now) ) return room_id, code def join_room(room_id, user_id): """加入房间(U)""" with get_db_connection() as conn: # 校验房间状态、人数 room = conn.execute('SELECT * FROM rooms WHERE id = ?', (room_id,)).fetchone() if not room: return False, '房间不存在' if room['status'] != 'waiting': return False, '游戏已开始' count = conn.execute('SELECT COUNT(*) as cnt FROM room_players WHERE room_id = ? AND is_active = 1', (room_id,)).fetchone()['cnt'] if count >= room['max_players']: return False, '房间已满' # 插入玩家记录 conn.execute( 'INSERT INTO room_players (room_id, user_id, join_order, joined_at) VALUES (?, ?, ?, ?)', (room_id, user_id, count + 1, time.time()) ) conn.execute('UPDATE rooms SET updated_at = ? WHERE id = ?', (time.time(), room_id)) return True, '加入成功' def get_room(room_id): """查询房间(含玩家列表)(R)""" with get_db_connection() as conn: room = conn.execute('SELECT * FROM rooms WHERE id = ?', (room_id,)).fetchone() if not room: return None # 关联查询玩家 players = conn.execute( 'SELECT u.id, u.nickname, u.avatar, rp.join_order, rp.score ' 'FROM room_players rp JOIN users u ON rp.user_id = u.id ' 'WHERE rp.room_id = ? AND rp.is_active = 1 ORDER BY rp.join_order', (room_id,) ).fetchall() result = dict(room) result['players'] = [dict(p) for p in players] return result六、完整流程演示:创建房间 → 加入 → 查询
6.1 真实场景剧本
""" 场景:3个玩家一起玩游戏 - 张三创建房间(房主) - 李四和王五加入 - 查询房间状态 """ # ========== 【阶段1】张三创建房间 ========== print("\n[1] 张三创建房间") room_id, code = models.create_room( host_id='user_zhangsan', difficulty='medium', max_players=4, max_rounds=5 ) print(f"✅ 房间创建成功") print(f" 房间ID: {room_id}") print(f" 房间码: {code} ← 分享给朋友") # 数据库状态: # rooms 表增加1行 # room_players 表增加1行(张三,join_order=1) # ========== 【阶段2】查询房间状态 ========== print("\n[2] 查询房间信息") room = models.get_room(room_id) print(f"✅ 房间信息:") print(f" 代码: {room['code']}") print(f" 难度: {room['difficulty']}") print(f" 状态: {room['status']}") print(f" 人数: {len(room['players'])}/4") print(f" 玩家列表:") for p in room['players']: print(f" - {p['nickname']} (加入顺序: {p['join_order']})") # 输出: # ✅ 房间信息: # 代码: 567890 # 难度: medium # 状态: waiting # 人数: 1/4 # 玩家列表: # - 张三 (加入顺序: 1) # ========== 【阶段3】李四加入房间 ========== print("\n[3] 李四加入房间") success, msg = models.join_room(room_id, 'user_lisi') if success: print(f"✅ {msg}") else: print(f"❌ {msg}") # 数据库状态: # room_players 表增加1行(李四,join_order=2) # ========== 【阶段4】查询房间状态(更新后) ========== print("\n[4] 查询房间信息(李四加入后)") room = models.get_room(room_id) print(f"✅ 房间信息:") print(f" 人数: {len(room['players'])}/4") print(f" 玩家列表:") for p in room['players']: print(f" - {p['nickname']} (加入顺序: {p['join_order']})") # 输出: # ✅ 房间信息: # 人数: 2/4 # 玩家列表: # - 张三 (加入顺序: 1) # - 李四 (加入顺序: 2) # ========== 【阶段5】王五加入房间 ========== print("\n[5] 王五加入房间") success, msg = models.join_room(room_id, 'user_wangwu') if success: print(f"✅ {msg}") # 数据库状态: # room_players 表增加1行(王五,join_order=3) # ========== 【阶段6】最终房间状态 ========== print("\n[6] 最终房间状态") room = models.get_room(room_id) print(f"✅ 房间已就绪!") print(f" 人数: {len(room['players'])}/4") print(f" 玩家列表:") for p in room['players']: medal = "🏠" if p['join_order'] == 1 else " " print(f" {medal} {p['nickname']} (加入顺序: {p['join_order']})") # 输出: # ✅ 房间已就绪! # 人数: 3/4 # 玩家列表: # 🏠 张三 (加入顺序: 1) # 李四 (加入顺序: 2) # 王五 (加入顺序: 3) # ========== 【检查】验证数据一致性 ========== print("\n[7] 数据库一致性检查") with models.get_db_connection() as conn: # 检查 rooms 表 room_record = conn.execute( 'SELECT * FROM rooms WHERE id = ?', (room_id,) ).fetchone() print(f"✅ rooms 表:code={room_record['code']}, status={room_record['status']}") # 检查 room_players 表 players_records = conn.execute( 'SELECT * FROM room_players WHERE room_id = ? AND is_active = 1 ORDER BY join_order', (room_id,) ).fetchall() print(f"✅ room_players 表:{len(players_records)} 条记录") for pr in players_records: user = conn.execute( 'SELECT nickname FROM users WHERE id = ?', (pr['user_id'],) ).fetchone() print(f" - {user['nickname']} (join_order={pr['join_order']})") # 输出: # ✅ rooms 表:code=567890, status=waiting # ✅ room_players 表:3 条记录 # - 张三 (join_order=1) # - 李四 (join_order=2) # - 王五 (join_order=3)6.2 数据库快照
创建房间后的数据库状态:
┌─── users 表 ───────────────────────────────────┐ │ id │ nickname │ total_score │ ... │ ├─────────────────┼───────────┼─────────────┤ │ user_zhangsan │ 张三 │ 0 │ │ user_lisi │ 李四 │ 0 │ │ user_wangwu │ 王五 │ 0 │ └─────────────────────────────────────────────────┘ ┌─── rooms 表 ────────────────────────────────┐ │ id │ code │ host_id │ status │ ├─────────┼────────┼─────────────────┼─────────┤ │ xyz789 │ 567890 │ user_zhangsan │ waiting │ └─────────────────────────────────────────────┘ ┌─── room_players 表 ─────────────────────────────┐ │ room_id │ user_id │ join_order │ score │ ├─────────┼─────────────────┼────────────┼────────┤ │ xyz789 │ user_zhangsan │ 1 │ 0 │ │ xyz789 │ user_lisi │ 2 │ 0 │ │ xyz789 │ user_wangwu │ 3 │ 0 │ └─────────────────────────────────────────────────┘查询流程图:
前端请求: 加入房间 ↓ app.py: api_join_room() ↓ models.join_room(room_id, user_id) ├─ 打开连接(with get_db_connection()) ├─ Step1: SELECT * FROM rooms WHERE id = ? │ ↓ 使用索引 idx_rooms_code,O(log n) ├─ Step2: SELECT * FROM room_players WHERE room_id = ? AND is_active = 1 │ ↓ 使用索引 idx_room_players_room,O(log n) ├─ Step3: COUNT(*) FROM room_players ... │ ↓ 快速计算 ├─ Step4: INSERT INTO room_players ... │ ↓ 插入新记录 └─ 自动提交事务(commit) ↓ 前端收到: {"success": true, "message": "加入成功"} ↓ 前端更新UI: 房间人数 1/4 → 2/4七、AI辅助开发记录
1. 第一版 Prompt(高层需求:数据库整体设计)
我现在正在做绘画 AI 博弈小游戏的后端数据库设计,使用 Python Flask + SQLite,需要设计完整的数据库层,要求如下:
支持用户、房间、玩家、回合、猜词、绘画行为、心理报告 7 张表
表关系清晰,符合 ER 图设计,支持一对多、多对多关联
房间必须支持 6 位数字房间码(code) 加入,不能直接用 id
绘画行为数据需要存储轨迹、速度、修改次数等非结构化数据
必须支持事务、回滚、外键约束、索引优化
数据库连接必须安全,不能出现连接泄漏、锁表
代码分层清晰,全部封装到 models.py,提供 CRUD 接口
支持完整业务流程:创建房间 → 加入房间 → 查询房间信息
2. 第二版 Prompt
为什么必须使用上下文管理器(get_db_connection)?
请帮我分析:
- 不使用上下文管理器会有什么风险?
- 上下文管理器如何保证连接安全与事务原子性?
- 它在多人联机游戏中的实际价值是什么?
- 如何实现最优雅、最稳定的版本?
八、总结
本周通过系统的数据库设计,完成了:
✨架构清晰— 7张表的职责分明,关系明确
✨查询高效— 关键操作都有索引支持,响应<20ms
✨事务安全— 上下文管理器确保数据一致性
✨扩展灵活— JSON + 关联表支持未来需求
这些设计既满足当前项目需求,也为后续的心理分析、复杂查询、数据导出等功能打好了基础。