本文介绍使用 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"✅ 已创建配对请求！\n\n请让另一位用户向本 Bot 发送命令：\n<code>/{code}</code>\n\n（他/她发送后，你将立即收到通知）",
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 多实例部署与长期运行。只要合理设计验证码生命周期与错误边界，即可稳定支撑邀请制、协作任务、实时匹配等多样化业务场景。
