aiogram框架下双用户消息交互的两类技术实现问题

Hey there! Let's break down how to implement this user linking feature in aiogram. We'll cover two practical approaches—one quick for testing, and another production-ready with persistent storage.

方法一：基于内存字典的轻量实现（适合快速验证） #

这种方式不需要额外依赖，用一个全局字典临时存储验证码和发起用户的ID映射，完美用来快速验证功能逻辑。

实现逻辑： #

• 初始化一个全局字典，保存 验证码: 发起用户ID 的映射关系
• 当第一个用户发送 /start 时，随机生成验证码并存入字典，回复用户让第二个用户输入该码
• 监听所有文本消息，当用户输入的内容匹配字典中的验证码时，找到对应的发起用户ID发送通知，同时移除已使用的验证码避免重复触发

代码示例： #

import asyncio
import random
from aiogram import Bot, Dispatcher
from aiogram.filters import CommandStart
from aiogram.types import Message

# 替换成你的Bot Token
BOT_TOKEN = "your_bot_token_here"
# 存储验证码与发起用户ID的映射
code_to_user = {}
# 验证码候选词库
CODE_WORDS = ["loremipsum", "randomtext", "testcode", "linkmeup", "connectnow"]

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

@dp.message(CommandStart())
async def handle_start(message: Message):
    # 生成随机验证码
    code = random.choice(CODE_WORDS)
    # 绑定验证码与当前用户ID
    code_to_user[code] = message.from_user.id
    # 回复用户引导语
    await message.answer(f"请让第二个用户向Bot发送这个验证码：`{code}`", parse_mode="MarkdownV2")

@dp.message()
async def handle_code_input(message: Message):
    input_code = message.text.strip()
    # 检查输入是否为有效验证码
    if input_code in code_to_user:
        first_user_id = code_to_user[input_code]
        # 通知发起用户
        await bot.send_message(
            chat_id=first_user_id,
            text=f"用户 `{message.from_user.id}` 输入了验证码，已完成连接！",
            parse_mode="MarkdownV2"
        )
        # 移除已使用的验证码
        del code_to_user[input_code]
        # 回复输入验证码的用户
        await message.answer("已成功与发起用户建立连接！")
    # 可选：处理无效验证码的情况
    # else:
    #     await message.answer("无效的验证码，请检查后重试！")

async def main():
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

方法二：基于数据库的持久化实现（适合生产环境） #

内存字典在Bot重启后会丢失所有数据，所以生产环境建议用数据库存储验证码信息。这里以SQLite + SQLAlchemy为例，实现持久化存储，还能轻松扩展验证码过期等功能。

实现逻辑： #

• 配置SQLAlchemy连接，创建表存储验证码、发起用户ID、创建时间、使用状态等字段
• 用户发送 /start 时，生成验证码并插入数据库
• 监听文本消息，查询数据库匹配未使用的验证码，找到发起用户ID后发送通知，同时标记验证码为已使用

代码示例： #

import asyncio
import random
from datetime import datetime
from aiogram import Bot, Dispatcher
from aiogram.filters import CommandStart
from aiogram.types import Message
from sqlalchemy import create_engine, Column, String, Integer, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

BOT_TOKEN = "your_bot_token_here"
CODE_WORDS = ["loremipsum", "randomtext", "testcode", "linkmeup", "connectnow"]

# SQLite数据库配置
DATABASE_URL = "sqlite:///./user_links.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 定义数据库模型
class UserLink(Base):
    __tablename__ = "user_links"
    id = Column(Integer, primary_key=True, index=True)
    code = Column(String, unique=True, index=True)
    first_user_id = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)
    is_used = Column(Boolean, default=False)

Base.metadata.create_all(bind=engine)

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@dp.message(CommandStart())
async def handle_start(message: Message):
    db = next(get_db())
    # 生成唯一验证码（避免重复）
    code = random.choice(CODE_WORDS)
    while db.query(UserLink).filter(UserLink.code == code).first():
        code = random.choice(CODE_WORDS)
    # 插入数据库
    new_link = UserLink(code=code, first_user_id=message.from_user.id)
    db.add(new_link)
    db.commit()
    # 回复用户
    await message.answer(f"请让第二个用户向Bot发送这个验证码：`{code}`", parse_mode="MarkdownV2")

@dp.message()
async def handle_code_input(message: Message):
    input_code = message.text.strip()
    db = next(get_db())
    # 查询未使用的有效验证码
    link_record = db.query(UserLink).filter(
        UserLink.code == input_code,
        UserLink.is_used == False
    ).first()
    if link_record:
        # 标记为已使用
        link_record.is_used = True
        db.commit()
        # 通知发起用户
        await bot.send_message(
            chat_id=link_record.first_user_id,
            text=f"用户 `{message.from_user.id}` 输入了验证码，已完成连接！",
            parse_mode="MarkdownV2"
        )
        await message.answer("已成功与发起用户建立连接！")
    else:
        await message.answer("无效或已过期的验证码，请检查后重试！")

async def main():
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

额外优化建议 #

• 验证码唯一性：如果候选词列表较短，可以结合随机数字生成更唯一的验证码，比如 random.choice(CODE_WORDS) + str(random.randint(100,999))
• 过期时间：在数据库模型中添加expires_at字段，生成验证码时设置为当前时间+N分钟，查询时过滤过期记录
• 错误处理：添加异常捕获逻辑，比如处理发送消息失败（用户可能已拉黑Bot）的情况
• 并发安全：内存字典在多线程环境下可能出现问题，生产环境优先用数据库或Redis这类支持原子操作的存储

备注：内容来源于stack exchange，提问作者Zailox