Pyrogram用户账户登录与会话管理深度解析

from pyrogram import Client
from pyrogram.errors import PhoneNumberInvalid, AuthBytesInvalid

async def initiate_login(api_id: int, api_hash: str, phone_number: str):
    """
    初始化 Pyrogram 客户端并发送登录验证码。
    :param api_id: 你的 Telegram API ID。
    :param api_hash: 你的 Telegram API Hash。
    :param phone_number: 用户的手机号码。
    :return: Pyrogram Client 实例和发送验证码的信息。
    """
    # 使用手机号作为会话名，确保会话持久化
    # 如果是临时会话，可以使用 ":memory:"
    client = Client(f"my_account_session_{phone_number}", api_id, api_hash)

    await client.connect() # 连接到 Telegram 服务器

    try:
        # 发送登录验证码
        sent_code_info = await client.send_code(phone_number)
        print(f"验证码已发送到 {phone_number}。")
        return client, sent_code_info
    except PhoneNumberInvalid:
        print("错误：手机号码无效。")
        await client.disconnect()
        return None, None
    except Exception as e:
        print(f"发送验证码时发生错误: {e}")
        await client.disconnect()
        return None, None

from pyrogram import Client
from pyrogram.errors import PhoneCodeInvalid, PhoneCodeEmpty, SessionPasswordNeeded, AuthBytesInvalid

async def complete_login(client: Client, sent_code_info, phone_code: str):
    """
    使用用户提供的验证码完成登录。
    :param client: 之前用于发送验证码的 Pyrogram Client 实例。
    :param sent_code_info: send_code 返回的 SentCode 对象。
    :param phone_code: 用户输入的验证码。
    :return: 登录成功返回 True，否则返回 False。
    """
    try:
        await client.sign_in(
            phone_number=sent_code_info.phone_number,
            phone_code_hash=sent_code_info.phone_code_hash,
            phone_code=phone_code
        )
        print("登录成功！")
        return True
    except PhoneCodeInvalid:
        print("错误：验证码无效。")
    except PhoneCodeEmpty:
        print("错误：验证码不能为空。")
    except SessionPasswordNeeded:
        print("错误：需要两步验证密码。")
        # 如果需要两步验证，这里需要进一步处理
        # 例如，await client.check_password("your_2fa_password")
    except AuthBytesInvalid:
        print("错误：认证信息无效，请重新尝试登录流程。")
    except Exception as e:
        print(f"登录时发生错误: {e}")
    finally:
        # 无论登录成功与否，如果不再需要客户端，可以断开连接
        # 但通常登录成功后会保持连接以便后续操作
        if not client.is_connected: # 避免重复断开
             await client.disconnect()
    return False

import asyncio
from pyrogram import Client

# 替换为你的 API ID 和 API Hash
API_ID = 1234567 # 从 my.telegram.org 获取
API_HASH = "your_api_hash_here" # 从 my.telegram.org 获取

async def main():
    phone_number = input("请输入你的手机号码 (例如: +8612345678900): ")

    client, sent_code_info = await initiate_login(API_ID, API_HASH, phone_number)

    if client and sent_code_info:
        phone_code = input("请输入你收到的验证码: ")
        success = await complete_login(client, sent_code_info, phone_code)

        if success:
            print("你现在已经登录，可以执行其他 Pyrogram 操作了。")
            # 示例：获取自己的信息
            me = await client.get_me()
            print(f"当前登录用户: {me.first_name} (@{me.username})")
            # 保持连接，直到程序结束或手动断开
            # await client.disconnect() # 如果不再需要，可以断开
        else:
            print("登录失败，请检查输入或重试。")
            await client.disconnect() # 登录失败，断开连接

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