本文介绍如何使用 aiogram 构建一个轻量级双用户配对系统：用户 a 启动会话后获得唯一配对码，用户 b 输入该码即完成绑定，双方随即建立单次通信通道，并支持状态持久化与自动清理。
本文介绍如何使用 aiogram 构建一个轻量级双用户配对系统：用户 a 启动会话后获得唯一配对码，用户 b 输入该码即完成绑定，双方随即建立单次通信通道，并支持状态持久化与自动清理。
在构建互动型 Telegram Bot（如双人问答、配对游戏或协作工具）时，常需让两个独立用户临时“连接”——例如用户 A 发起请求后，系统生成唯一配对码；用户 B 输入该码即触发事件，使 A 即时收到通知。这本质上是一个带状态的跨会话消息等待机制，不能依赖内存变量（易丢失），而应结合数据库实现可靠协调。
以下以 SQLite 为例，展示完整、健壮的实现方案（兼容 Aiogram 3.x）：
✅ 核心设计思路
- 使用一张全局配对表 pairing_requests，存储待匹配的请求；
- 每条记录包含：id（自增主键）、initiator_id（用户 A 的 user.id）、code（随机生成的唯一字符串）、status（'pending' / 'matched'）、created_at；
- 不为每对用户动态建表（原文中 CREATE TABLE {user1_id + user2_id} 方案存在 SQL 注入风险、表名非法字符问题且难以维护），而是统一管理、索引优化。
? 示例代码（Aiogram 3 + SQLite）
import sqlite3
import random
from aiogram import Router, F
from aiogram.types import Message
from aiogram.filters import Command
router = Router()
# 初始化数据库（建议在 bot 启动时执行一次）
def init_db():
conn = sqlite3.connect("bot.db")
cur = conn.cursor()
cur.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
)
""")
conn.commit()
conn.close()
# 预定义词库（确保无重复、无敏感字符）
CODE_WORDS = ["loremipsum", "randomtext", "quantumflux", "nebulaforge", "stardustkey"]
@router.message(Command("start"))
async def cmd_start(message: Message):
user_id = message.from_user.id
code = random.choice(CODE_WORDS)
# 存储配对请求
conn = sqlite3.connect("bot.db")
cur = conn.cursor()
cur.execute(
"INSERT INTO pairing_requests (initiator_id, code) VALUES (?, ?)",
(user_id, code)
)
conn.commit()
conn.close()
await message.answer(
f"✅ 已创建配对请求！\n请让另一位用户向本机器人发送以下指令：\n\n`/{code}`\n\n（注意：斜杠不可省略）",
parse_mode="Markdown"
)
@router.message(F.text.regexp(r"^/([a-zA-Z0-9]+)$"))
async def handle_code_command(message: Message):
code = message.text[1:] # 去掉开头的 '/'
conn = sqlite3.connect("bot.db")
cur = conn.cursor()
cur.execute(
"SELECT initiator_id FROM pairing_requests "
"WHERE code = ? AND status = 'pending'",
(code,)
)
row = cur.fetchone()
if row:
initiator_id = row[0]
# 标记为已匹配（防止重复触发）
cur.execute(
"UPDATE pairing_requests SET status = 'matched' WHERE code = ?",
(code,)
)
conn.commit()
# 通知发起者
await message.bot.send_message(
chat_id=initiator_id,
text=f"? 用户 `{message.from_user.id}` 已成功输入配对码 `{code}`！"
)
await message.answer("✅ 配对成功！对方已收到通知。")
else:
await message.answer("❌ 无效或已使用的配对码，请检查拼写或联系发起者。")
conn.close()⚠️ 关键注意事项
- 安全性：避免直接拼接用户输入进 SQL（如原文 f'''CREATE TABLE {user1_id + user2_id}...）；始终使用参数化查询。
- 幂等性：通过 status 字段确保同一配对码仅触发一次响应，防止刷屏或重复通知。
- 可扩展性：如需支持多轮配对或超时自动清理，可在表中增加 expires_at 字段，并配合后台任务定期清理过期记录。
- 错误处理：生产环境应补充异常捕获（如数据库连接失败）、日志记录及用户友好提示。
- 并发安全：SQLite 在多数 Bot 场景下足够，若并发极高，建议切换至 PostgreSQL 并加行级锁（SELECT ... FOR UPDATE）。
✅ 总结
该方案摒弃了动态建表等高风险做法，采用标准化关系模型，兼顾简洁性、可维护性与可靠性。通过 code 作为轻量级“握手凭证”，实现了跨用户、跨会话的状态同步——这是构建多人协作类 Bot 的基础能力。后续可基于此扩展为邀请链接、临时房间、双人投票等更复杂场景。
