本文介绍使用 aiogram 构建双用户协同流程的完整方案：当用户 a 启动会话后，系统生成唯一验证码；用户 b 输入该码即触发双向通知，并完成临时配对，全程依托 sqlite 实现轻量级状态管理。
本文介绍使用 aiogram 构建双用户协同流程的完整方案：当用户 a 启动会话后，系统生成唯一验证码；用户 b 输入该码即触发双向通知，并完成临时配对，全程依托 sqlite 实现轻量级状态管理。
在实际 Bot 开发中，常需支持「邀请-响应」类交互（如组队匹配、协作验证、双人游戏启动等）。Aiogram 本身不内置跨用户会话关联机制，因此需结合持久化存储实现状态跟踪。本文以「用户 A 发起配对 → 系统生成随机验证码 → 用户 B 输入验证码 → 双方即时获知配对成功」为典型场景，提供可落地的工程化实现。
核心设计思路
- 无状态 Bot + 有状态存储：Bot 保持无状态处理逻辑，所有配对关系、验证码、用户 ID 映射均存于 SQLite 数据库；
- 单表统一管理：避免动态建表（如原答案中 CREATE TABLE {user1_id + user2_id}），因其易引发 SQL 注入、权限问题及维护困难；推荐使用一张固定结构的 pairing_requests 表；
- 验证码防重 & 限时有效（可选增强）：虽基础需求未要求时效性，但生产环境建议添加 created_at 和 expires_at 字段，配合定时清理任务。
数据库表结构（推荐）
CREATE TABLE IF NOT EXISTS pairing_requests (
id INTEGER PRIMARY KEY AUTOINCREMENT,
initiator_id INTEGER NOT NULL, -- 用户 A 的 Telegram ID
code TEXT UNIQUE NOT NULL, -- 如 'loremipsum'
status TEXT DEFAULT 'pending', -- 'pending' / 'matched' / 'expired'
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
matched_at TIMESTAMP NULL
);完整 Aiogram 3.x 实现示例（Python）
import random
from aiogram import Router, F
from aiogram.types import Message
from aiogram.filters import Command
import sqlite3
router = Router()
DB_PATH = "pairing.db"
# 初始化数据库
def init_db():
with sqlite3.connect(DB_PATH) as conn:
conn.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,
matched_at TIMESTAMP NULL
)
""")
init_db()
# 预定义验证码词库（建议至少 100+ 项以降低碰撞概率）
CODE_WORDS = ["loremipsum", "randomtext", "bluewhale", "quantumflux", "nebulaedge", "zeroday"]
@router.message(Command("start"))
async def cmd_start(message: Message):
user_id = message.from_user.id
code = random.choice(CODE_WORDS)
# 记录配对请求
with sqlite3.connect(DB_PATH) as conn:
conn.execute(
"INSERT INTO pairing_requests (initiator_id, code) VALUES (?, ?)",
(user_id, code)
)
conn.commit()
await message.answer(
f"✅ 已创建配对请求！\n\n请让另一位用户向本 Bot 发送以下验证码：\n<code>{code}</code>\n\n（发送后你将立即收到通知）",
parse_mode="HTML"
)
@router.message(F.text.in_(CODE_WORDS)) # 简单匹配（生产环境建议用 WHERE code=? 查询）
async def handle_code(message: Message):
code = message.text.strip()
user_b_id = message.from_user.id
with sqlite3.connect(DB_PATH) as conn:
# 查找待匹配的请求（且未被使用）
cursor = conn.execute(
"SELECT initiator_id FROM pairing_requests "
"WHERE code = ? AND status = 'pending'",
(code,)
)
row = cursor.fetchone()
if not row:
await message.answer("❌ 无效或已被使用的验证码。")
return
user_a_id = row[0]
# 更新状态并记录匹配时间
conn.execute(
"UPDATE pairing_requests SET status = 'matched', matched_at = CURRENT_TIMESTAMP "
"WHERE code = ?",
(code,)
)
conn.commit()
# 通知双方
await message.bot.send_message(
user_a_id,
f"? 成功配对！用户 <code>{user_b_id}</code> 已输入验证码 <code>{code}</code>。",
parse_mode="HTML"
)
await message.answer(
f"✅ 验证码已确认！已通知用户 <code>{user_a_id}</code>。",
parse_mode="HTML"
)关键注意事项
- ✅ 安全性：F.text.in_(CODE_WORDS) 仅适用于小规模词库且无用户自定义输入场景；若词库动态扩展或需更高鲁棒性，应改用参数化查询 WHERE code = ?；
- ✅ 并发安全：SQLite 在多数 Bot 场景下足够，但高并发时建议加 conn.execute("BEGIN IMMEDIATE") 避免竞态；
- ⚠️ 错误处理：示例省略了异常捕获，生产环境务必包裹 try/except sqlite3.Error；
- ? 避免硬编码：CODE_WORDS 应从配置文件或环境变量加载，便于热更新与多语言支持；
- ? 清理策略：可定期执行 DELETE FROM pairing_requests WHERE status = 'pending' AND created_at
通过以上结构化设计，你不仅能稳定支撑双用户配对逻辑，还可快速扩展为多用户邀请链、带身份校验的协作流程等更复杂场景。核心在于——用数据库做“记忆”，让 Bot 始终专注“响应”。
