使用Telethon获取Telegram频道消息作者数据报错排查

我正在从一个以用户名义而非频道名义发布内容的Telegram频道中获取最新5条帖子，目前仅能获取到作者的ID。尝试通过该ID调用client.get_entity获取用户名等其余用户数据时触发错误。

脚本代码 #

from telethon.sync import TelegramClient
from telethon.tl.functions.contacts import SearchRequest
from telethon.tl.types import PeerUser, PeerChat, PeerChannel
from telethon.tl.types import InputPeerChat
from telethon import functions

import asyncio
from asyncio import new_event_loop, set_event_loop
import time

from db.users_db import getAllUsersForSend
from db.channels_db import getAllTrChannels

api_id = 'api_id'
api_hash = 'api_hash'

async def bot_auth(client, tg_id):
    # Bot authorization
    if not await client.is_user_authorized():
        await client.start(bot_token='my_bot_token')
    dialogs = await client.get_dialogs()
    for dialog in dialogs:
        if isinstance(dialog.message.peer_id, PeerChannel):
            tr_channels = getAllTrChannels()
            for channel in tr_channels:
                if channel['channel_id'] == dialog.message.peer_id.channel_id:
                    print(dialog.name)
                    print(dialog.message.peer_id.channel_id)
                    messages = client.iter_messages(dialog, limit=5)
                    async for message in messages:
                        print(message.from_id.user_id)
                        from_user = await client.get_entity(message.from_id.user_id)
                        print(from_user)
                        await asyncio.sleep(10)

    # Session termination
    await client.disconnect()

def lets_go():
    users_for_send = getAllUsersForSend()

    for user in users_for_send:
        print(user['tg_id'])

        loop = new_event_loop()
        set_event_loop(loop)

        # Creating a Telegram client
        client = TelegramClient('session_name_testbot', api_id, api_hash)

        # Starting an asynchronous function
        with client:
            client.loop.run_until_complete(bot_auth(client, user['tg_id']))

报错信息 #

Job "lets_go (trigger: interval[0:01:00], next run at: 2024-01-24 14:55:32 +06)" raised an exception
Traceback (most recent call last):
File "E:\work\2024\get_vacancy_bot\venv\Lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
retval = job.func(*job.args, **job.kwargs)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\work\2024\get_vacancy_bot\handlers\get_chat_messages.py", line 67, in lets_go
client.loop.run_until_complete(bot_auth(client, user['tg_id']))
File "C:\Program Files\Python311\Lib\asyncio\base_events.py", line 653, in run_until_complete
return future.result()
       ^^^^^^^^^^^^^^^
File "E:\work\2024\get_vacancy_bot\handlers\get_chat_messages.py", line 37, in bot_auth
from_user = await client.get_entity(message.from_id.user_id)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\work\2024\get_vacancy_bot\venv\Lib\site-packages\telethon\client\users.py", line 292, in get_entity
  inputs.append(await self.get_input_entity(x))
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\work\2024\get_vacancy_bot\venv\Lib\site-packages\telethon\client\users.py", line 468, in get_input_entity
  raise ValueError(
ValueError: Could not find the input entity for PeerUser(user_id=5919601160) (PeerUser). Please read https://docs.telethon.dev/en/stable/concepts/entities.html to find out more details.

根据文档说明，client.get_entity应支持通过ID获取数据，请问问题出在哪里？

问题根源 #

你用的是Bot账号调用get_entity，而Telegram Bot的权限和普通用户账号有本质区别：

1. Bot无法直接获取未主动发起过对话的用户信息，哪怕知道ID也不行
2. 频道里的陌生用户如果没和Bot互动过，Bot的会话中没有该用户的实体缓存，Telegram服务器也不会返回该用户的详细信息

解决步骤 #

方案1：改用普通用户账号认证

把Bot token换成普通用户的session认证，普通用户账号只要能看到频道内容，就能获取频道内发帖用户的信息：
修改bot_auth函数中的认证部分：

async def bot_auth(client, tg_id):
    # 普通用户账号认证（替换Bot token方式）
    if not await client.is_user_authorized():
        # 第一次运行会提示输入手机号和验证码
        await client.start(phone='你的手机号')
    # ... 后续代码不变

方案2：如果必须用Bot账号，先让用户和Bot互动

如果业务要求必须用Bot，需要让目标用户先给Bot发一条消息（随便内容即可），这样Bot的会话中就会缓存该用户的实体，之后就能通过get_entity获取信息。但这个方案对频道里的陌生用户不现实，没法强制他们和Bot互动。

方案3：直接从消息对象提取可用信息

如果只是需要用户名等基础信息，可以先检查message.sender字段，部分情况下Telethon会直接返回用户的基本信息，不需要额外调用get_entity：

async for message in messages:
    print(message.from_id.user_id)
    # 先尝试直接获取sender信息
    if message.sender:
        print(f"用户名: {message.sender.username}, 昵称: {message.sender.first_name}")
    else:
        # 再尝试调用get_entity，同时捕获异常
        try:
            from_user = await client.get_entity(message.from_id.user_id)
            print(from_user)
        except ValueError:
            print(f"无法获取用户{message.from_id.user_id}的信息，该用户未与Bot互动")
    await asyncio.sleep(10)

额外优化点 #

• 当前循环嵌套过多（dialogs → tr_channels），可以先把目标频道ID存成集合，减少循环次数提升效率
• 每次lets_go都创建新的client和事件循环，资源浪费严重，建议复用client实例

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