获取Telegram频道实体：邀请链接的鲁棒处理方法

import asyncio
from telethon import TelegramClient, functions, types

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

async def get_channel_entity_if_joined(client: TelegramClient, invite_link_hash: str):
    """
    尝试在用户已是成员的情况下，通过邀请链接获取频道实体。
    :param client: TelegramClient 实例
    :param invite_link_hash: 不带 '+' 的邀请链接哈希，例如 'XXXXXX'
    :return: 频道实体 (types.Chat 或 types.Channel)
    """
    full_invite_url = 'https://t.me/joinchat/' + invite_link_hash
    try:
        entity = await client.get_entity(full_invite_url)
        print(f"用户已是成员，成功获取实体: {entity.title} (ID: {entity.id})")
        return entity
    except Exception as e:
        print(f"尝试直接获取实体失败: {e}")
        return None

# 示例调用 (假设 invite_link_hash 是一个有效的邀请哈希)
# entity = await get_channel_entity_if_joined(client, 'your_invite_hash_here')

import asyncio
from telethon import TelegramClient, functions, types

async def join_channel_and_get_entity(client: TelegramClient, invite_link_hash: str):
    """
    通过邀请链接加入频道并获取其实体。
    :param client: TelegramClient 实例
    :param invite_link_hash: 不带 '+' 的邀请链接哈希，例如 'XXXXXX'
    :return: 频道实体 (types.Chat 或 types.Channel)
    """
    try:
        updates = await client(functions.messages.ImportChatInviteRequest(invite_link_hash))
        if updates and updates.chats:
            entity = updates.chats[0]
            print(f"成功加入频道并获取实体: {entity.title} (ID: {entity.id})")
            return entity
        else:
            print("加入频道成功，但未在 updates 中找到频道实体。")
            return None
    except Exception as e:
        print(f"加入频道失败: {e}")
        return None

# 示例调用
# entity = await join_channel_and_get_entity(client, 'your_invite_hash_here')

import asyncio
from telethon import TelegramClient, functions, types

async def get_channel_entity_robustly(client: TelegramClient, invite_link_hash: str):
    """
    通过邀请链接鲁棒地获取频道实体，处理用户已加入和未加入两种情况。
    :param client: TelegramClient 实例
    :param invite_link_hash: 不带 '+' 的邀请链接哈希，例如 'XXXXXX'
    :return: 频道实体 (types.Chat 或 types.Channel)，如果失败则返回 None
    """
    entity = None
    try:
        # 尝试直接获取实体 (适用于用户已是成员的情况)
        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 Exception as ex:
        # 如果直接获取失败，检查是否是“你不是成员”的错误
        if 'you are not part of' in str(ex).lower():
            print(f"用户未加入频道，尝试通过邀请链接加入: {invite_link_hash}")
            try:
                # 尝试加入频道并从 updates 中获取实体
                res = await client(functions.messages.ImportChatInviteRequest(invite_link_hash))
                if res and res.chats:
                    entity = res.chats[0]
                    print(f"成功加入频道并获取实体: {entity.title} (ID: {entity.id})")
                else:
                    print("加入频道成功，但未在 updates 中找到频道实体。")
            except Exception as join_ex:
                print(f"加入频道失败: {join_ex}")
        else:
            # 处理其他未知错误
            print(f"获取频道实体时发生未知错误: {ex}")

    return entity

# 示例：如何使用这个函数
async def main():
    api_id = 1234567 # 替换为你的 API ID
    api_hash = 'your_api_hash_here' # 替换为你的 API Hash
    client = TelegramClient('my_session', api_id, api_hash)

    await client.start()

    # 替换为你的邀请链接哈希，例如 'B_aBcDeF1gH2iJkL'
    test_invite_hash = 'your_invite_link_hash_without_plus' 

    channel_entity = await get_channel_entity_robustly(client, test_invite_hash)

    if channel_entity:
        print(f"\n最终获取到的频道实体: {channel_entity.title} (ID: {channel_entity.id}, 类型: {type(channel_entity)})")
    else:
        print("\n未能获取到频道实体。")

    await client.run_until_disconnected()

if __name__ == '__main__':
    asyncio.run(main())