Telethon脚本运行出现AuthKeyUnregisteredError问题求助

问题核心原因 #

AuthKeyUnregisteredError本质是Telegram服务器端注销了你的会话密钥，通常由以下情况导致：

• 自定义循环未维护与服务器的心跳连接，超时后被踢下线
• 会话持久化处理不当，重复生成或丢失有效会话文件
• 多设备登录冲突（同一账号在其他设备登录导致当前会话失效）
• 自定义循环未正确处理重连逻辑，断开后无法自动恢复连接

具体解决方案 #

1. 改用带信号处理的官方run_until_disconnected（推荐）

你之前认为run_until_disconnected无法处理终止时机，其实可以通过捕获系统信号实现清理逻辑，官方循环比自定义循环更稳定，能自动维护心跳和重连：

import signal
from telethon import TelegramClient, events

# 初始化客户端（确保会话文件路径固定，复用同一个会话）
client = TelegramClient('tg_session', API_ID, API_HASH)
LOG_CHANNEL_ID = 123456789  # 替换为你的日志频道ID
TARGET_CHANNEL_ID = 987654321  # 替换为目标监听频道ID

# 终止处理函数
def handle_exit(signal_num, frame):
    print("开始清理会话...")
    # 异步发送终止日志到TG频道
    client.loop.create_task(client.send_message(LOG_CHANNEL_ID, "脚本已正常终止"))
    # 关闭客户端
    client.loop.create_task(client.disconnect())
    client.loop.stop()

# 注册信号处理（Ctrl+C对应SIGINT，系统终止指令对应SIGTERM）
signal.signal(signal.SIGINT, handle_exit)
signal.signal(signal.SIGTERM, handle_exit)

async def forward_to_discord(message):
    # 这里写你的Discord Webhook转发逻辑
    pass

async def main():
    await client.start("你的账号手机号或Bot Token")
    # 注册消息监听回调
    @client.on(events.NewMessage(chats=TARGET_CHANNEL_ID))
    async def message_handler(event):
        await forward_to_discord(event.message)
    
    print("脚本启动，开始监听消息...")
    await client.run_until_disconnected()

if __name__ == "__main__":
    client.loop.run_until_complete(main())

2. 自定义循环必须加入心跳与重连逻辑

如果坚持使用自定义循环，必须定期发送心跳请求，并在捕获AuthKeyUnregisteredError时触发重新登录：

import asyncio
from telethon import TelegramClient, events, errors

client = TelegramClient('tg_session', API_ID, API_HASH)
LOG_CHANNEL_ID = 123456789
TARGET_CHANNEL_ID = 987654321
is_running = True

async def heartbeat():
    """定期发送心跳维持连接，每30秒一次"""
    while is_running:
        try:
            if client.is_connected():
                await client.get_me()  # 用get_me作为心跳请求
            await asyncio.sleep(30)
        except errors.AuthKeyUnregisteredError:
            print("会话密钥失效，尝试重新登录...")
            await client.start("你的账号手机号或Bot Token")
        except Exception as e:
            print(f"心跳请求出错: {str(e)}")
            await asyncio.sleep(10)

async def forward_to_discord(message):
    # 你的Discord转发逻辑
    pass

async def main():
    await client.start("你的账号手机号或Bot Token")
    # 启动心跳任务
    asyncio.create_task(heartbeat())
    
    @client.on(events.NewMessage(chats=TARGET_CHANNEL_ID))
    async def message_handler(event):
        await forward_to_discord(event.message)
    
    print("脚本启动，开始监听消息...")
    while is_running:
        try:
            await client.updates.poll()  # 手动轮询更新
            await asyncio.sleep(1)
        except errors.AuthKeyUnregisteredError:
            print("轮询更新时密钥失效，重新登录...")
            await client.start("你的账号手机号或Bot Token")
        except KeyboardInterrupt:
            is_running = False
            await client.send_message(LOG_CHANNEL_ID, "脚本已正常终止")
            await client.disconnect()

if __name__ == "__main__":
    client.loop.run_until_complete(main())

3. 检查会话与账号状态

• 确保会话文件（如tg_session.session）未被删除或覆盖，每次启动都复用同一个会话文件
• 检查账号多设备登录设置：如果手机端登录时开启了"终止其他会话"，会导致脚本会话失效，建议关闭该选项或使用独立会话
• 避免频繁重启脚本，多次重复登录可能被Telegram判定为异常，触发会话注销

4. 优化错误捕获逻辑

在所有客户端操作中主动捕获AuthKeyUnregisteredError，触发重新登录后重试操作：

try:
    await client.send_message(LOG_CHANNEL_ID, "测试消息")
except errors.AuthKeyUnregisteredError:
    await client.start("你的账号手机号或Bot Token")
    await client.send_message(LOG_CHANNEL_ID, "测试消息")

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