本文详解如何利用数据库临时表与随机验证码，在 aiogram 机器人中构建可靠的双用户配对通信流程，确保用户 a 发起请求后，仅当用户 b 输入正确动态码时，a 才能收到确认通知。
本文详解如何利用数据库临时表与随机验证码，在 aiogram 机器人中构建可靠的双用户配对通信流程，确保用户 a 发起请求后，仅当用户 b 输入正确动态码时，a 才能收到确认通知。
在开发交互式 Telegram Bot（如配对问答、双人验证、协作者邀请等场景）时，常需实现「用户 A 触发流程 → 等待用户 B 响应 → 向 A 反馈 B 的行为」这一异步协同逻辑。Aiogram 本身不提供跨会话的原生等待机制，因此必须借助外部状态管理——轻量级 sqlite 数据库存储临时配对关系是最简洁、可靠且无需额外服务的方案。
核心设计思路
- 用户 A 发送 /start 后，Bot 生成唯一、易读的随机验证码（如 “sunrise”、”quantum”），并持久化记录：user_a_id、user_b_id（暂为空）、code、created_at；
- Bot 将该 code 返回给 A，并提示：“请让另一位用户向本机器人发送此代码”；
- 当任意用户发送该 code 时，Bot 查询数据库：若存在未匹配的记录，且 user_b_id 为空，则将当前发送者 ID 写入 user_b_id，并向 user_a_id 发送通知（如 “用户 123456789 已输入验证码”），随后立即删除该记录（或标记为已使用）。
✅ 实现示例（基于 Aiogram 3.x + SQLite）
import sqlite3 import random from aiogram import Router, F from aiogram.types import Message from aiogram.filters import Command router = Router() WORDS = ["sunrise", "quantum", "ember", "lunar", "nova", "pulse", "orbit"] # 初始化数据库 def init_db(): conn = sqlite3.connect("pairing.db") cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS pending_pairs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_a_id INTEGER NOT NULL, user_b_id INTEGER, code TEXT UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close() init_db() @router.message(Command("start")) async def cmd_start(message: Message): user_a_id = message.from_user.id code = random.choice(WORDS) conn = sqlite3.connect("pairing.db") cur = conn.cursor() cur.execute( "INSERT INTO pending_pairs (user_a_id, code) VALUES (?, ?)", (user_a_id, code) ) conn.commit() conn.close() await message.answer( f"✅ 已创建配对请求！n" f"请让另一位用户向本机器人发送以下代码：n" f"<code>{code}</code>n" f"（注意：区分大小写，无需斜杠）", parse_mode="HTML" ) @router.message(F.text) async def handle_code(message: Message): code_input = message.text.strip() # 查询待匹配的记录（且 user_b_id 为空） conn = sqlite3.connect("pairing.db") cur = conn.cursor() cur.execute( "SELECT user_a_id FROM pending_pairs WHERE code = ? AND user_b_id IS NULL", (code_input,) ) row = cur.fetchone() if row: user_a_id = row[0] user_b_id = message.from_user.id # 更新记录，标记 user_b_id 并通知 user_a cur.execute( "UPDATE pending_pairs SET user_b_id = ? WHERE code = ?", (user_b_id, code_input) ) conn.commit() # 向用户 A 发送确认 await message.bot.send_message( chat_id=user_a_id, text=f"? 用户 <code>{user_b_id}</code> 已成功输入验证码 <code>{code_input}</code>。", parse_mode="HTML" ) # 清理：删除已使用的记录（可选：也可保留日志，改用 DELETE） cur.execute("DELETE FROM pending_pairs WHERE code = ?", (code_input,)) conn.commit() await message.answer("✅ 配对已确认！对方已收到通知。") else: await message.answer("⚠️ 无效或已使用的验证码，请检查拼写或联系发起人。") conn.close()⚠️ 关键注意事项
- 安全性：本例使用明文词表，适用于低敏感场景；生产环境建议改用 secrets.token_urlsafe(4) 生成不可预测码，并增加过期时间（如 created_at
- 并发风险：SQLite 在高并发下可能产生竞态，若预期多用户高频触发，应加 BEGIN IMMEDIATE 事务或切换至 postgresql；
- 数据清理：务必在匹配成功后及时清除记录，避免 DB 膨胀；可增设定时任务清理超时未匹配条目；
- 用户体验：建议为用户 A 添加超时提醒（如 5 分钟未响应则自动失效），可通过 aiogram.scheduler 或外部 cron 实现。
通过该模式，你无需依赖全局状态变量或复杂事件总线，即可在无状态的 Bot 架构中，稳健支撑跨用户、有时序依赖的协作流程——这是构建教育类配对练习、小团队协作工具或轻量级审核系统的理想基础。