使用Pyrogram开发Telegram API Bot的会话与验证码问题

核心问题分析 #

你当前的代码每次调用create_and_initialize_client和continue_session都会新建Client实例，而Pyrogram的会话状态存储在以会话名命名的.session文件中，每次新建实例相当于从头启动流程，必然会触发新的验证码请求。

解决方案步骤 #

1. 复用同一Client实例 #

要维持会话状态，必须在整个登录流程中复用同一个Client对象，不能每次创建新实例。可以将实例保存为全局变量、类属性，或者通过业务上下文传递，避免重复初始化。

2. 用start()的回调处理验证码 #

Pyrogram的Client.start()方法支持传入phone_code_handler回调函数，专门用于接收验证码输入，这样就能在同一会话中完成验证码传递，无需重新创建Client。

修改后的示例代码：

from pyrogram import Client
from pyrogram.errors import SessionPasswordNeeded

# 全局保存Client实例（可根据架构调整为类属性/上下文管理）
client_instance = None

async def init_client(api_id, api_hash, phone_number):
    global client_instance
    client_instance = Client(
        session_name=phone_number,  # 确保会话名唯一，对应生成的.session文件
        api_id=api_id,
        api_hash=api_hash,
        phone_number=phone_number
    )
    
    # 定义验证码处理回调
    async def phone_code_handler():
        # 实际Bot场景中，这里替换为等待用户在对话中发送验证码的逻辑
        code = input("请输入收到的验证码：")
        return code
    
    try:
        await client_instance.start(phone_code_handler=phone_code_handler)
        print("会话初始化完成")
    except SessionPasswordNeeded:
        # 处理两步验证场景
        password = input("请输入两步验证密码：")
        await client_instance.check_password(password)

async def use_existing_session():
    # 直接复用已初始化的实例
    if client_instance and client_instance.is_connected:
        me = await client_instance.get_me()
        print(f"当前登录账号：{me.phone_number}")

3. 关键细节说明 #

• 会话文件：Pyrogram会自动在当前目录生成{session_name}.session文件，保存会话状态。只要复用同一会话名的Client实例，就能加载未完成的会话，不会重新发送验证码。
• 验证码回调：phone_code_handler是start()的可选参数，当Telegram要求输入验证码时，Pyrogram会调用该函数获取验证码，从而在同一会话中完成登录。
• 两步验证处理：若账号开启两步验证，start()会抛出SessionPasswordNeeded异常，需调用check_password()传入密码完成验证。

4. Bot场景适配调整 #

实际Bot开发中不能用input()获取验证码，需通过监听用户消息实现：

import asyncio
from pyrogram import Client, filters

client_instance = None
code_event = None
user_code = None

@app.on_message(filters.command("start_login"))
async def start_login(client, message):
    global client_instance, code_event
    api_id = YOUR_API_ID
    api_hash = YOUR_API_HASH
    phone_number = message.text.split()[1]  # 假设用户发送 /start_login +86xxxxxx
    
    code_event = asyncio.Event()
    
    async def phone_code_handler():
        await message.reply("请发送你收到的验证码")
        await code_event.wait()
        return user_code
    
    client_instance = Client(phone_number, api_id, api_hash, phone_number=phone_number)
    try:
        await client_instance.start(phone_code_handler=phone_code_handler)
        await message.reply("登录成功！")
    except SessionPasswordNeeded:
        await message.reply("请发送两步验证密码")
        # 类似验证码逻辑处理密码输入

@app.on_message(filters.text & filters.private)
async def get_code(client, message):
    global user_code, code_event
    if code_event and not code_event.is_set():
        user_code = message.text.strip()
        code_event.set()
        await message.reply("验证码已接收，正在登录...")

内容的提问来源于stack exchange，提问作者Babak Majd