本文介绍使用 aiogram 构建双用户协同流程的核心方法：通过唯一随机验证码关联两名用户，借助 sqlite 数据库存储配对状态，并在验证成功后完成消息中转与资源清理。
本文介绍使用 aiogram 构建双用户协同流程的核心方法：通过唯一随机验证码关联两名用户，借助 sqlite 数据库存储配对状态，并在验证成功后完成消息中转与资源清理。
在实际 Bot 开发中，常需设计需要多方参与的交互逻辑（如“邀请配对”、“组队验证”或“双向确认”场景）。Aiogram 本身不提供原生的跨用户会话绑定机制，因此需结合外部状态管理（如数据库）来实现可靠、可扩展的双用户协作流程。以下是一个典型且健壮的实现方案。
✅ 核心设计思路
- 用户 A（发起者）：发送 /start 后，Bot 生成一个唯一、易记的验证码（如 loremipsum），并记录其 ID 与待匹配状态；
- 用户 B（响应者）：向 Bot 发送该验证码（如 /loremipsum），Bot 验证通过后，向用户 A 发送通知（如 “用户 123456789 已输入此验证码”），并立即清理临时数据；
- 状态持久化：使用轻量级 SQLite 存储配对关系（user_a_id, code, created_at），避免内存泄漏与并发冲突。
?️ 实现步骤与代码示例（Aiogram 3.x）
import sqlite3 import random from aiogram import Router, F from aiogram.types import Message, CallbackQuery from aiogram.filters import Command router = Router() # 验证码词库（建议使用更长、更唯一的单词组合，或加入时间戳哈希增强唯一性） CODE_WORDS = ["loremipsum", "randomtext", "botmatch", "aiogrampair", "verifynow"] # 初始化数据库表 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, 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(CODE_WORDS) # 存入待配对表 conn = sqlite3.connect("pairing.db") cur = conn.cursor() try: cur.execute( "INSERT INTO pending_pairs (user_a_id, code) VALUES (?, ?)", (user_a_id, code) ) conn.commit() await message.answer( f"✅ 已创建配对请求！nn请让另一位用户向本 Bot 发送命令：n<code>/{code}</code>nn（他/她发送后，你将立即收到通知）", parse_mode="HTML" ) except sqlite3.IntegrityError: await message.answer("⚠️ 当前已有未完成的配对请求，请先完成或重试。") finally: conn.close() # 匹配处理器：响应 /{code} 命令 @router.message(F.text.startswith("/")) async def handle_code_command(message: Message): full_text = message.text.strip() if not full_text.startswith("/") or len(full_text) <= 1: return input_code = full_text[1:] # 去掉 '/' conn = sqlite3.connect("pairing.db") cur = conn.cursor() cur.execute( "SELECT user_a_id FROM pending_pairs WHERE code = ?", (input_code,) ) row = cur.fetchone() if row: user_a_id = row[0] # 向发起者发送通知 try: await message.bot.send_message( chat_id=user_a_id, text=f"? 用户 <code>{message.from_user.id}</code> 已成功输入验证码 <code>{input_code}</code>。", parse_mode="HTML" ) except Exception as e: # 若用户已屏蔽 Bot 或非私聊，忽略错误（可选记录日志） pass # 清理已匹配记录 cur.execute("DELETE FROM pending_pairs WHERE code = ?", (input_code,)) conn.commit() await message.answer("✅ 配对成功！对方已收到通知。") else: await message.answer("❌ 无效的验证码。请确认拼写是否正确，或联系发起者获取最新代码。") conn.close()⚠️ 关键注意事项
-
安全性增强建议：
- 避免纯静态词库（易被枚举），可改为 code = random.choice(CODE_WORDS) + str(time.time_ns())[-4:]；
- 添加过期时间字段（如 expires_at），配合定时任务或首次查询时校验；
- 使用 code 字段加索引提升查询性能：CREATE INDEX idx_code ON pending_pairs(code);
-
并发与幂等性：
SQLite 的 INSERT … ON CONFLICT IGNORE 或事务封装可防止重复插入；当前示例已通过 UNIQUE 约束+异常捕获保障基础幂等性。 -
用户体验优化：
- 可为用户 A 提供 /cancel 命令主动终止待配对；
- 向用户 B 返回更友好的提示（如“正在为您连接…”）；
- 支持多语言时，将提示文本抽象为 gettext 或配置字典。
✅ 总结
该方案以最小依赖（仅标准库 + aiogram）实现了清晰、可维护的双用户联动逻辑。它不依赖全局变量或复杂状态机，而是通过数据库作为“共享上下文”，天然支持 Bot 多实例部署与长期运行。只要合理设计验证码生命周期与错误边界，即可稳定支撑邀请制、协作任务、实时匹配等多样化业务场景。