如何通过邀请链接获取Telegram频道实体

from telethon import functions
from telethon.tl.types import Channel, Chat

async def get_channel_entity_by_invite_link(client, invite_hash: str):
    """
    通过邀请链接（invite hash）获取Telegram频道实体。
    该函数能够处理用户已加入和未加入频道两种情况。

    Args:
        client: 已经认证并连接的Telethon客户端实例。
        invite_hash: 邀请链接的哈希部分，例如 'XXXXXXX' (不包含 'https://t.me/joinchat/' 或 '+' )。

    Returns:
        telethon.tl.types.Channel 或 telethon.tl.types.Chat 对象，
        代表目标频道或群组的实体。
        如果无法获取实体，则返回 None。

    Raises:
        Exception: 如果在处理过程中遇到无法识别的错误。
    """
    entity = None
    full_invite_url = f'https://t.me/joinchat/{invite_hash}'

    try:
        # 尝试直接获取实体，适用于用户已是成员的情况
        # Telethon的get_entity可以识别完整的joinchat链接
        entity = await client.get_entity(full_invite_url)
        print(f"用户已是成员，直接获取到实体: {entity.title}")

    except Exception as ex:
        # 捕获异常，判断是否是“用户不是成员”的错误
        if 'you are not part of' in str(ex).lower():
            print(f"用户不是频道成员，尝试通过邀请链接加入...")
            try:
                # 尝试加入频道并获取实体
                updates = await client(functions.messages.ImportChatInviteRequest(invite_hash))

                # 从updates对象中提取频道实体
                # updates.chats 列表通常包含新加入的频道/群组实体
                if updates.chats:
                    entity = updates.chats[0]
                    print(f"成功加入频道并获取到实体: {entity.title}")
                else:
                    print("成功加入，但未在updates中找到频道实体。")

            except Exception as join_ex:
                print(f"加入频道时发生错误: {join_ex}")
                raise # 重新抛出加入时的异常
        else:
            # 捕获其他未知错误
            print(f"获取实体时发生未知错误: {ex}")
            raise # 重新抛出原始异常

    return entity

# 示例用法 (假设 client 已经初始化并连接)
async def main():
    # 替换为你的实际客户端初始化代码
    # from telethon.sync import TelegramClient
    # api_id = YOUR_API_ID
    # api_hash = 'YOUR_API_HASH'
    # client = TelegramClient('session_name', api_id, api_hash)
    # await client.start()

    # 假设 client 已经是一个可用的 TelegramClient 实例
    # 模拟一个 client 对象用于示例，实际应用中请使用真实的 TelegramClient
    class MockClient:
        async def get_entity(self, link):
            if 'joined_channel_hash' in link:
                class MockChannel:
                    title = "已加入的测试频道"
                return MockChannel()
            else:
                raise Exception("YOU ARE NOT PART OF THIS CHAT")
        async def __call__(self, request):
            if isinstance(request, functions.messages.ImportChatInviteRequest):
                class MockUpdates:
                    class MockNewChannel:
                        title = "新加入的测试频道"
                    chats = [MockNewChannel()]
                return MockUpdates()
            return None

    client = MockClient() # 实际使用时请替换为你的 TelegramClient 实例

    # 情况1: 用户已是成员
    print("\n--- 场景1: 用户已是成员 ---")
    invite_hash_joined = 'joined_channel_hash' # 替换为实际的邀请哈希
    channel_entity_joined = await get_channel_entity_by_invite_link(client, invite_hash_joined)
    if channel_entity_joined:
        print(f"最终获取到的实体名称 (已加入): {channel_entity_joined.title}")
    else:
        print("未能获取到已加入的频道实体。")

    # 情况2: 用户不是成员
    print("\n--- 场景2: 用户不是成员 ---")
    invite_hash_not_joined = 'new_channel_hash' # 替换为实际的邀请哈希
    channel_entity_not_joined = await get_channel_entity_by_invite_link(client, invite_hash_not_joined)
    if channel_entity_not_joined:
        print(f"最终获取到的实体名称 (未加入): {channel_entity_not_joined.title}")
    else:
        print("未能获取到未加入的频道实体。")

# 运行示例
# import asyncio
# asyncio.run(main())