Telegram telethon: 鲁棒地通过邀请链接获取频道实体的高效策略

from telethon import TelegramClient, functions, types

# 假设 client 已经初始化并连接
# client = TelegramClient('session_name', api_id, api_hash)
# await client.start()

invite_link_hash = 'XXXXXXX' # 原始邀请链接哈希，不包含 '+'

try:
    # 尝试使用完整的邀请链接URL获取实体
    # 这适用于用户已经加入频道的情况
    entity = await client.get_entity('https://t.me/joinchat/' + invite_link_hash)
    print(f"已加入频道实体获取成功: {entity.title} (ID: {entity.id})")
except Exception as ex:
    # 如果用户未加入，get_entity 会抛出异常
    # 我们需要捕获这个异常并进行下一步处理
    print(f"尝试获取已加入频道实体失败: {ex}")
    # 这里我们捕获异常，后续在完整的代码块中处理

# 承接上一步的异常处理
if 'you are not part of' in str(ex): # 检查是否为未加入频道的错误
    print(f"用户未加入频道，尝试通过邀请链接加入...")
    res = await client(functions.messages.ImportChatInviteRequest(invite_link_hash))
    if isinstance(res, types.Updates):
        # 成功加入，频道实体在 updates.chats[0]
        entity = res.chats[0]
        print(f"成功加入频道并获取实体: {entity.title} (ID: {entity.id})")
    else:
        print("加入频道成功，但未在 updates 对象中找到频道实体。")
else:
    # 处理其他类型的异常，例如邀请链接无效等
    print(f"发生未知错误: {ex}")
    entity = None

from telethon import TelegramClient, functions, types
from telethon.errors import ChatInviteInvalidError, UserAlreadyParticipantError, RPCError

async def get_channel_entity_by_invite_link(client: TelegramClient, invite_link_hash: str):
    """
    通过邀请链接哈希值可靠地获取频道实体。
    该函数会尝试获取已加入频道的实体，如果失败则尝试加入频道并获取实体。

    Args:
        client (TelegramClient): 已认证的 Telegram 客户端实例。
        invite_link_hash (str): 频道邀请链接的哈希部分（例如 'XXXXXXX'）。

    Returns:
        types.Channel | None: 频道实体对象，如果失败则返回 None。
    """
    entity = None
    try:
        # 尝试通过完整的邀请链接URL获取实体
        # 这适用于用户已是频道成员的情况
        full_invite_url = 'https://t.me/joinchat/' + invite_link_hash
        entity = await client.get_entity(full_invite_url)
        print(f"成功获取已加入频道实体: {entity.title} (ID: {entity.id})")

    except UserAlreadyParticipantError:
        # 理论上，如果 get_entity 成功，不会抛出此错误。
        # 但作为防御性编程，如果其他操作导致此错误，可以处理。
        print(f"用户已是该频道成员，尝试重新获取实体...")
        # 重新尝试 get_entity，因为UserAlreadyParticipantError可能意味着之前尝试加入失败
        # 或其他逻辑导致此错误，此时应该能直接获取实体。
        entity = await client.get_entity(full_invite_url)
        print(f"重新获取已加入频道实体成功: {entity.title} (ID: {entity.id})")

    except RPCError as e:
        # 捕获 Telethon 的 RPC 错误
        error_message = str(e).lower()
        if 'you are not part of' in error_message or 'not a participant' in error_message:
            # 用户不是频道成员，尝试加入频道
            print(f"用户未加入频道 (错误: {e})，尝试通过邀请链接加入...")
            try:
                updates = await client(functions.messages.ImportChatInviteRequest(invite_link_hash))
                if isinstance(updates, types.Updates) and updates.chats:
                    entity = updates.chats[0]
                    print(f"成功加入频道并获取实体: {entity.title} (ID: {entity.id})")
                else:
                    print(f"加入频道成功，但未在 updates 对象中找到频道实体。")
            except ChatInviteInvalidError:
                print(f"邀请链接 '{invite_link_hash}' 无效或已过期。")
            except RPCError as join_error:
                print(f"加入频道时发生RPC错误: {join_error}")
            except Exception as join_ex:
                print(f"加入频道时发生未知错误: {join_ex}")
        elif 'channel invalid' in error_message or 'chat invalid' in error_message:
            print(f"邀请链接 '{invite_link_hash}' 对应的频道无效或不存在。")
        else:
            print(f"获取频道实体时发生RPC错误: {e}")
    except Exception as ex:
        # 捕获其他所有未知异常
        print(f"获取频道实体时发生未知错误: {ex}")

    return entity

# 示例用法 (请替换为您的实际 api_id, api_hash 和 invite_link_hash)
async def main():
    api_id = 1234567 # 替换为您的 API ID
    api_hash = 'your_api_hash' # 替换为您的 API Hash
    client = TelegramClient('my_session', api_id, api_hash)

    await client.start()

    # 替换为您的实际邀请链接哈希
    test_invite_link_hash = 'YOUR_INVITE_LINK_HASH_HERE'

    channel_entity = await get_channel_entity_by_invite_link(client, test_invite_link_hash)

    if channel_entity:
        print(f"\n最终获取到的频道信息:")
        print(f"标题: {channel_entity.title}")
        print(f"ID: {channel_entity.id}")
        print(f"类型: {'公开' if channel_entity.username else '私有'}")
    else:
        print("\n未能获取到频道实体。")

    await client.run_until_disconnected()

# 如果在Jupyter或asyncio环境中运行，可能需要不同的启动方式
# import asyncio
# asyncio.run(main())