generate_session.py

#
# PaperplaneRemix - A modular Telegram selfbot script
# Copyright (C) 2022, Avinash Reddy and the PaperplaneRemix contributors
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
import asyncio
import configparser
import os
import sys

ERROR = (
    "Something is wrong with your API {}, " "please double check and re-enter the same."
)

if sys.version_info < (3, 9, 1):
    print("Please run this script with Python 3.9 or above." "\nExiting the script.")
    sys.exit(1)

if sys.platform.startswith("win"):
    loop = asyncio.ProactorEventLoop()
else:
    loop = asyncio.get_event_loop()


async def install_pip_package(package: str) -> bool:
    args = ["-m", "pip", "install", "--upgrade", "--user", package]
    process = await asyncio.create_subprocess_exec(
        sys.executable.replace(" ", "\\ "),
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    await process.communicate()
    return process.returncode == 0


try:
    import telethon
    from telethon.errors import AuthKeyError, InvalidBufferError
except (ModuleNotFoundError, ImportError):
    print("Installing Telethon...")
    pip = loop.run_until_complete(install_pip_package("telethon"))
    if pip:
        print("Successfully installed Telethon. Run the script again.")
    else:
        print(
            "Failed to install Telethon. "
            "Run the script again after installing it manually using:"
        )
        print('"python3 -m pip install -U telethon --user"')
    sys.exit()

api_id = os.environ.get("api_id", None)
api_hash = os.environ.get("api_hash", None)
endpoint = os.environ.get("redis_endpoint", None)
password = os.environ.get("redis_password", None)
redis = False

if all((api_id, api_hash)):
    redis_ip = input("Would you like to generate a Redis session? (yes/no): ")
    if redis_ip.lower() in ("y", "yes"):
        if ":" not in endpoint:
            print("Invalid Redis endpoint! Try again.")
            sys.exit(1)
        redis = True
elif os.path.exists("./config.ini"):
    config = configparser.ConfigParser()
    config.read("./config.ini")
    api_id = config["telethon"].getint("api_id", False)
    api_hash = config["telethon"].get("api_hash", False)
    if not (api_id or api_hash):
        print("Invalid config file! Fix it before generating a session.")
        sys.exit(1)
    redis_ip = input("Would you like to generate a Redis session? (yes/no): ")
    if redis_ip.lower() in ("y", "yes"):
        endpoint = config["database"].get("redis_endpoint", False)
        password = config["database"].get("redis_password", False)
        if not (endpoint or password):
            print(
                "Make sure you have redis_endpoint and redis_password "
                "in your config.ini"
            )
            sys.exit(1)
        elif ":" not in endpoint:
            print("Invalid Redis endpoint.")
            sys.exit(1)
        redis = True
else:
    while True:
        api_id = input("Enter your API ID: ")
        if len(api_id) >= 2 and api_id.isnumeric():
            break
        else:
            print(ERROR.format("ID"))
    while True:
        api_hash = input("Enter your API hash: ")
        if len(api_hash) == 32 and api_hash.isalnum():
            break
        else:
            print(ERROR.format("hash"))
    redis_ip = input("Would you like to generate a Redis session? (yes/no): ")
    if redis_ip.lower() in ("y", "yes"):
        while True:
            endpoint = input("Enter your Redis endpoint: ")
            if ":" in endpoint:
                break
            else:
                print("Invalid Redis endpoint! Try again.")
        password = input("Enter your Redis password: ")
        redis = True

if redis:
    try:
        import redis
    except (ModuleNotFoundError, ImportError):
        print("Installing Redis...")
        pip = loop.run_until_complete(install_pip_package("redis"))
        if pip:
            print("Successfully installed Redis. Run the script again.")
        else:
            print(
                "Failed to install Redis. "
                "Run the script again after installing it manually using:"
            )
            print('"python3 -m pip install -U redis --user"')
        sys.exit()

    from sessions.redis import RedisSession

    redis_connection = redis.Redis(
        host=endpoint.split(":")[0],
        port=endpoint.split(":")[1],
        password=password.strip(),
    )
    try:
        redis_connection.ping()
    except Exception:
        print("Invalid Redis credentials! Exiting the script")
        sys.exit(1)
    session = RedisSession("userbot", redis_connection)
else:
    session = "userbot"

client = telethon.TelegramClient(session, api_id, api_hash)

try:
    with client:
        me = client.loop.run_until_complete(client.get_me())
        name = telethon.utils.get_display_name(me)
    print(f"Successfully generated a session for {name}")
except (AuthKeyError, InvalidBufferError):
    client.session.delete()
    print(
        "Your old session was invalid and has been automatically deleted! "
        "Run the script again to generate a new session."
    )
    sys.exit(1)