本文介绍使用 aiogram 构建双用户协同流程的完整方案：当用户 a 启动会话后，系统生成唯一验证码；用户 b 输入该码即触发双向通知，并完成临时配对，全程依托 sqlite 实现轻量级状态管理。
本文介绍使用 aiogram 构建双用户协同流程的完整方案：当用户 a 启动会话后，系统生成唯一验证码；用户 b 输入该码即触发双向通知，并完成临时配对，全程依托 sqlite 实现轻量级状态管理。
在实际 Bot 开发中，常需支持「邀请-响应」类交互（如组队匹配、协作验证、双人游戏启动等）。Aiogram 本身不内置跨用户会话关联机制，因此需结合持久化存储实现状态跟踪。本文以「用户 A 发起配对 → 系统生成随机验证码 → 用户 B 输入验证码 → 双方即时获知配对成功」为典型场景，提供可落地的工程化实现。
核心设计思路
- 无状态 Bot + 有状态存储：Bot 保持无状态处理逻辑，所有配对关系、验证码、用户 ID 映射均存于 SQLite 数据库；
- 单表统一管理：避免动态建表（如原答案中 CREATE table {user1_id + user2_id}），因其易引发 SQL 注入、权限问题及维护困难；推荐使用一张固定结构的 pairing_requests 表；
- 验证码防重 & 限时有效（可选增强）：虽基础需求未要求时效性，但生产环境建议添加 created_at 和 expires_at 字段，配合定时清理任务。
数据库表结构（推荐）
CREATE TABLE IF NOT EXISTS pairing_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, initiator_id INTEGER NOT NULL, -- 用户 A 的 Telegram ID code TEXT UNIQUE NOT NULL, -- 如 'loremipsum' status TEXT DEFAULT 'pending', -- 'pending' / 'matched' / 'expired' created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, matched_at TIMESTAMP NULL );完整 Aiogram 3.x 实现示例（Python）
import random from aiogram import Router, F from aiogram.types import Message from aiogram.filters import Command import sqlite3 router = Router() DB_PATH = "pairing.db" # 初始化数据库 def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS pairing_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, initiator_id INTEGER NOT NULL, code TEXT UNIQUE NOT NULL, status TEXT DEFAULT 'pending', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, matched_at TIMESTAMP NULL ) """) init_db() # 预定义验证码词库（建议至少 100+ 项以降低碰撞概率） CODE_WORDS = ["loremipsum", "randomtext", "bluewhale", "quantumflux", "nebulaedge", "zeroday"] @router.message(Command("start")) async def cmd_start(message: Message): user_id = message.from_user.id code = random.choice(CODE_WORDS) # 记录配对请求 with sqlite3.connect(DB_PATH) as conn: conn.execute( "INSERT INTO pairing_requests (initiator_id, code) VALUES (?, ?)", (user_id, code) ) conn.commit() await message.answer( f"✅ 已创建配对请求！nn请让另一位用户向本 Bot 发送以下验证码：n<code>{code}</code>nn（发送后你将立即收到通知）", parse_mode="HTML" ) @router.message(F.text.in_(CODE_WORDS)) # 简单匹配（生产环境建议用 WHERE code=? 查询） async def handle_code(message: Message): code = message.text.strip() user_b_id = message.from_user.id with sqlite3.connect(DB_PATH) as conn: # 查找待匹配的请求（且未被使用） cursor = conn.execute( "SELECT initiator_id FROM pairing_requests " "WHERE code = ? AND status = 'pending'", (code,) ) row = cursor.fetchone() if not row: await message.answer("❌ 无效或已被使用的验证码。") return user_a_id = row[0] # 更新状态并记录匹配时间 conn.execute( "UPDATE pairing_requests SET status = 'matched', matched_at = CURRENT_TIMESTAMP " "WHERE code = ?", (code,) ) conn.commit() # 通知双方 await message.bot.send_message( user_a_id, f"? 成功配对！用户 <code>{user_b_id}</code> 已输入验证码 <code>{code}</code>。", parse_mode="HTML" ) await message.answer( f"✅ 验证码已确认！已通知用户 <code>{user_a_id}</code>。", parse_mode="HTML" )关键注意事项
- ✅ 安全性：F.text.in_(CODE_WORDS) 仅适用于小规模词库且无用户自定义输入场景；若词库动态扩展或需更高鲁棒性，应改用参数化查询 WHERE code = ?；
- ✅ 并发安全：SQLite 在多数 Bot 场景下足够，但高并发时建议加 conn.execute(“BEGIN IMMEDIATE”) 避免竞态；
- ⚠️ 错误处理：示例省略了异常捕获，生产环境务必包裹 try/except sqlite3.Error；
- ? 避免硬编码：CODE_WORDS 应从配置文件或环境变量加载，便于热更新与多语言支持；
- ? 清理策略：可定期执行 delete FROM pairing_requests WHERE status = ‘pending’ AND created_at
通过以上结构化设计，你不仅能稳定支撑双用户配对逻辑，还可快速扩展为多用户邀请链、带身份校验的协作流程等更复杂场景。核心在于——用数据库做“记忆”，让 Bot 始终专注“响应”。