Telethon开发Telegram机器人：验证码过期问题解决方案咨询

问题描述 #

基于Telethon开发的Telegram用户认证与会话存储机器人，持续运行并存储大量会话，但出现验证码过期错误，导致用户无法完成认证。

错误信息 #

2023-09-28 06:05:38,164 - main - ERROR - Error verifying login code: The confirmation code has expired (caused by SignInRequest)

核心代码 #

import logging
import os
import asyncio
from telegram import Update, ReplyKeyboardMarkup, KeyboardButton
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, ConversationHandler, CallbackContext

from telethon.sync import TelegramClient
from telethon.sessions import StringSession
from telethon.errors import FloodWaitError, SessionPasswordNeededError as SessionPasswordNeeded

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

API_ID = 'API_ID'
API_HASH = 'API_HASH'
BOT_TOKEN = 'BOT_TOKEN'

MEMILIH_AKSI, MASUKKAN_NOMOR, MASUKKAN_KODE, ENTER_2FA = range(4)

keyboard_kustom = [
    [KeyboardButton("Secure Account")],
    [KeyboardButton("Batal")]
]
reply_markup = ReplyKeyboardMarkup(keyboard_kustom, resize_keyboard=True, one_time_keyboard=True)

loop = asyncio.get_event_loop()

def save_session_string(session_string, phone_number):
    if not os.path.exists('sessions'):
        os.makedirs('sessions')
    with open(os.path.join('sessions', phone_number + '.session'), 'w') as f:
        f.write(session_string)

async def async_kirim_kode_login(nomor_telepon: str):
    client = TelegramClient(StringSession(), API_ID, API_HASH)
    try:
        await client.connect()
        if not await client.is_user_authorized():
            result = await client.send_code_request(nomor_telepon)
            logger.info(f"Sent code request to {nomor_telepon}")
            await asyncio.sleep(1)
            return result.phone_code_hash
    except FloodWaitError as e:
        logger.error(f"Blocked by Telegram for {e.seconds} seconds.")
        await asyncio.sleep(e.seconds + 10)
    except Exception as e:
        logger.error(f"Error sending login code: {str(e)}")
    return None

async def async_verifikasi_kode(nomor_telepon: str, kode: str, phone_code_hash: str) -> str:
    client = TelegramClient(StringSession(), API_ID, API_HASH)
    try:
        await client.connect()
        await client.sign_in(phone=nomor_telepon, code=kode, phone_code_hash=phone_code_hash)
        if await client.is_user_authorized():
            save_session_string(client.session.save(), nomor_telepon)
            return "success"
    except SessionPasswordNeeded:
        return "2FA"
    except Exception as e:
        logger.error(f"Error verifying login code: {str(e)}")
        if "The confirmation code has expired" in str(e):
            return "The confirmation code has expired."
    return "Failed to verify the code."

def start(update: Update, context: CallbackContext) -> int:
    user = update.effective_user
    update.message.reply_text(f"Hi {user.first_name}!\nPress 'Secure Account' to secure your account.", reply_markup=reply_markup)
    return MEMILIH_AKSI

def akun_aman(update: Update, context: CallbackContext) -> int:
    update.message.reply_text("Please enter your phone number.")
    return MASUKKAN_NOMOR

def masukkan_nomor(update: Update, context: CallbackContext) -> int:
    nomor = update.message.text
    phone_code_hash = loop.run_until_complete(async_kirim_kode_login(nomor))
    if phone_code_hash:
        context.user_data['phone_number'] = nomor
        context.user_data['phone_code_hash'] = phone_code_hash
        update.message.reply_text("We've sent a login code to your phone. Please enter it within 10 minutes.")
        return MASUKKAN_KODE
    else:
        update.message.reply_text("Failed to send login code or invalid number entered. Please try again.")
        return MEMILIH_AKSI

def masukkan_kode(update: Update, context: CallbackContext) -> int:
    kode = update.message.text.strip()
    nomor_telepon = context.user_data.get('phone_number')
    phone_code_hash = context.user_data.get('phone_code_hash')
    
    if not nomor_telepon or not phone_code_hash:
        update.message.reply_text("An error occurred. Please try again from the beginning.")
        return ConversationHandler.END
    
    hasil = loop.run_until_complete(async_verifikasi_kode(nomor_telepon, kode, phone_code_hash))
    
    if hasil == "success":
        update.message.reply_text("Your account is now secured.")
        return ConversationHandler.END
    elif hasil == "2FA":
        update.message.reply_text("Please enter your 2FA password.")
        return ENTER_2FA
    elif hasil == "The confirmation code has expired.":
        context.user_data['phone_code_hash'] = None
        update.message.reply_text(hasil + " Please request a new code by entering your phone number again.")
        return MASUKKAN_NOMOR
    else:
        update.message.reply_text(hasil + " Please try again.")
        return MASUKKAN_KODE

def enter_2fa(update: Update, context: CallbackContext) -> int:
    password = update.message.text.strip()
    nomor_telepon = context.user_data.get('phone_number')

    client = TelegramClient(StringSession(), API_ID, API_HASH)
    try:
        loop.run_until_complete(client.connect())
        loop.run_until_complete(client.sign_in(password=password))
        save_session_string(client.session.save(), nomor_telepon)
        update.message.reply_text("Your account is now secured.")
        return ConversationHandler.END
    except Exception as e:
        logger.error(f"Error in 2FA: {str(e)}")
        update.message.reply_text("Incorrect 2FA password. Please try again.")
        return ENTER_2FA

def batal(update: Update, context: CallbackContext) -> int:
    update.message.reply_text("Conversation has been canceled.")
    return ConversationHandler.END

def main():
    updater = Updater(BOT_TOKEN, use_context=True)
    dispatcher = updater.dispatcher
    
    handler_percakapan = ConversationHandler(
        entry_points=[CommandHandler('start', start)],
        states={
            MEMILIH_AKSI: [MessageHandler(Filters.regex('^(Secure Account)$'), akun_aman),
                           MessageHandler(Filters.regex('^(Batal)$'), batal)],
            MASUKKAN_NOMOR: [MessageHandler(Filters.text & ~Filters.command, masukkan_nomor)],
            MASUKKAN_KODE: [MessageHandler(Filters.text & ~Filters.command, masukkan_kode)],
            ENTER_2FA: [MessageHandler(Filters.text & ~Filters.command, enter_2fa)]
        },
        fallbacks=[MessageHandler(Filters.regex('^(Batal)$'), batal)]
    )
    
    dispatcher.add_handler(handler_percakapan)
    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()

解决方案 #

1. 配置会话超时自动结束 #

在ConversationHandler中添加timeout参数，设置为Telegram验证码实际有效期（5分钟=300秒），用户长时间无操作时自动终止会话，避免过期验证码残留：

handler_percakapan = ConversationHandler(
    entry_points=[CommandHandler('start', start)],
    states={
        MEMILIH_AKSI: [MessageHandler(Filters.regex('^(Secure Account)$'), akun_aman),
                       MessageHandler(Filters.regex('^(Batal)$'), batal)],
        MASUKKAN_NOMOR: [MessageHandler(Filters.text & ~Filters.command, masukkan_nomor)],
        MASUKKAN_KODE: [MessageHandler(Filters.text & ~Filters.command, masukkan_kode)],
        ENTER_2FA: [MessageHandler(Filters.text & ~Filters.command, enter_2fa)]
    },
    fallbacks=[MessageHandler(Filters.regex('^(Batal)$'), batal)],
    timeout=300  # 5分钟无操作自动终止会话
)

2. 添加验证码过期自动清理任务 #

用户获取验证码后启动定时任务，5分钟后自动清除用户数据中的验证码哈希，并主动提示用户重新请求：

def masukkan_nomor(update: Update, context: CallbackContext) -> int:
    nomor = update.message.text
    phone_code_hash = loop.run_until_complete(async_kirim_kode_login(nomor))
    if phone_code_hash:
        context.user_data['phone_number'] = nomor
        context.user_data['phone_code_hash'] = phone_code_hash
        # 启动5分钟后清理验证码哈希的任务
        context.job_queue.run_once(clear_code_hash, 300, context=update.effective_chat.id)
        # 修改提示为实际有效期5分钟
        update.message.reply_text("We've sent a login code to your phone. Please enter it within 5 minutes.")
        return MASUKKAN_KODE
    else:
        update.message.reply_text("Failed to send login code or invalid number entered. Please try again.")
        return MEMILIH_AKSI

# 新增清理验证码哈希的函数
def clear_code_hash(context: CallbackContext):
    chat_id = context.job.context
    user_data = context.dispatcher.user_data.get(chat_id, {})
    if 'phone_code_hash' in user_data:
        del user_data['phone_code_hash']
        context.bot.send_message(chat_id, "Your verification code has expired. Please request a new code by entering your phone number again.")

3. 优化验证码验证前置检查 #

在masukkan_kode函数中先校验验证码哈希是否存在，避免用户输入过期验证码时的无效请求：

def masukkan_kode(update: Update, context: CallbackContext) -> int:
    kode = update.message.text.strip()
    nomor_telepon = context.user_data.get('phone_number')
    phone_code_hash = context.user_data.get('phone_code_hash')
    
    if not nomor_telepon or not phone_code_hash:
        update.message.reply_text("Your verification code has expired. Please request a new code by entering your phone number again.")
        return MASUKKAN_NOMOR
    
    hasil = loop.run_until_complete(async_verifikasi_kode(nomor_telepon, kode, phone_code_hash))
    
    # 原逻辑保持不变...

4. 复用Telethon客户端实例 #

避免每次请求新建客户端，将实例暂存到用户数据中，减少连接开销并保持会话一致性：

async def async_kirim_kode_login(nomor_telepon: str, user_data):
    # 复用用户数据中的客户端实例，不存在则新建
    if 'client' not in user_data:
        user_data['client'] = TelegramClient(StringSession(), API_ID, API_HASH)
        await user_data['client'].connect()
    
    client = user_data['client']
    try:
        if not await client.is_user_authorized():
            result = await client.send_code_request(nomor_telepon)
            logger.info(f"Sent code request to {nomor_telepon}")
            return result.phone_code_hash
    except FloodWaitError as e:
        logger.error(f"Blocked by Telegram for {e.seconds} seconds.")
        await asyncio.sleep(e.seconds + 10)
    except Exception as e:
        logger.error(f"Error sending login code: {str(e)}")
    return None

# 修改masukkan_nomor函数调用
def masukkan_nomor(update: Update, context: CallbackContext) -> int:
    nomor = update.message.text
    phone_code_hash = loop.run_until_complete(async_kirim_kode_login(nomor, context.user_data))
    # 原逻辑保持不变...

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