如何捕获Telethon事件协程错误并识别无效Telegram频道?
Telethon监听Telegram频道无效链接识别与崩溃修复方案
问题根因
你当前的代码直接将未校验的频道链接传入events.NewMessage监听器,一旦某个链接对应的频道被关闭、不存在、或你无访问权限,就会触发ValueError: No user has "xxx" as username错误,直接导致整个程序终止运行。
现有方案的优化方向
你使用的get_entity校验逻辑是可行的,在此基础上可以做两层优化,覆盖启动前预校验和运行时容错两个场景:
方案1:启动前预校验所有频道(推荐,适合固定监听场景)
启动程序时先批量校验所有待监听频道的有效性,过滤无效链接后再注册监听器,从根源避免报错。
完整实现代码如下:
import configparser from telethon.errors import SessionPasswordNeededError, RPCError from telethon import TelegramClient, events api_id = 'xxxxxx' api_hash = 'xxxxxxxxxxxxxxxxxxxxxxx' # 待监听的目标频道列表 input_channels = ['https://t.me/xxxxxx','https://t.me/xxxx','https://t.me/xxxxxxx'] # 存储校验通过的有效频道 valid_channels = [] client = TelegramClient('anon', api_id, api_hash) async def validate_channels(): global valid_channels for channel in input_channels: try: entity = await client.get_entity(channel) valid_channels.append(entity) print(f"频道校验通过:{channel}") except ValueError: print(f"无效频道,已过滤:{channel}") except RPCError as e: print(f"频道{channel}校验失败,错误信息:{e},已过滤") async def newMessageListener(event): newMessage = event.message.message print(newMessage) async def main(): await client.start() # 启动前先完成所有频道校验 await validate_channels() # 仅为有效频道注册消息监听器 client.add_event_handler(newMessageListener, events.NewMessage(chats=valid_channels)) await client.run_until_disconnected() if __name__ == "__main__": with client: client.loop.run_until_complete(main())
该方案的优势:
- 逻辑简单,启动阶段就完成所有风险排查,不会出现运行时突然崩溃的问题
- 直接存储校验后的
entity对象,比每次解析字符串链接的执行效率更高 - 针对不同错误类型做了分类捕获,方便你后续根据日志处理无效频道
方案2:运行时动态容错(适合频道列表动态更新的场景)
如果你需要在程序运行过程中动态添加监听频道,除了新增前调用get_entity校验外,还可以给全局加异常捕获逻辑,避免单频道失效影响整体运行:
# 给消息处理器加外层异常捕获 @client.on(events.NewMessage(chats=dynamic_channels)) async def newMessageListener(event): try: newMessage = event.message.message print(newMessage) except Exception as e: print(f"处理来自频道{event.chat.id}的消息出错:{e},已自动跳过")
内容的提问来源于stack exchange,提问作者Stamatis Tiniakos
相关产品推荐
相关产品推荐

