本文详解如何利用数据库临时表与随机验证码，在 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 架构中，稳健支撑跨用户、有时序依赖的协作流程——这是构建教育类配对练习、小团队协作工具或轻量级审核系统的理想基础。
