bilal
12b9992079
Create a dedicated session generator script that prompts for IP mode first and update README with the new generation flow and behavior in non-interactive shells. Co-authored-by: Cursor <cursoragent@cursor.com>
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
import asyncio
|
|
import getpass
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from telethon import TelegramClient
|
|
from telethon.sessions import StringSession
|
|
|
|
|
|
API_ID = int(os.getenv("TG_API_ID", "0"))
|
|
API_HASH = os.getenv("TG_API_HASH", "")
|
|
SESSION_FILE_PATH = os.getenv("SESSION_FILE_PATH", "./session_data/session.session")
|
|
TG_USE_IPV6 = os.getenv("TG_USE_IPV6", "true").lower() in {"1", "true", "yes"}
|
|
TG_CONNECT_TIMEOUT = int(os.getenv("TG_CONNECT_TIMEOUT", "30"))
|
|
TG_REQUEST_RETRIES = int(os.getenv("TG_REQUEST_RETRIES", "10"))
|
|
TG_CONNECTION_RETRIES = int(os.getenv("TG_CONNECTION_RETRIES", "20"))
|
|
TG_RETRY_DELAY = int(os.getenv("TG_RETRY_DELAY", "5"))
|
|
|
|
|
|
def validate() -> None:
|
|
if API_ID <= 0:
|
|
raise ValueError("Set TG_API_ID in environment")
|
|
if not API_HASH:
|
|
raise ValueError("Set TG_API_HASH in environment")
|
|
|
|
|
|
def choose_ip_mode(default_ipv6: bool) -> bool:
|
|
default_label = "2" if default_ipv6 else "1"
|
|
print("Choose connection mode for Telegram session generation:")
|
|
print("1) IPv4")
|
|
print("2) IPv6")
|
|
|
|
if not sys.stdin.isatty():
|
|
print(f"Non-interactive shell detected, using default from env: {'IPv6' if default_ipv6 else 'IPv4'}")
|
|
return default_ipv6
|
|
|
|
choice = input(f"Select [1/2] (default {default_label}): ").strip().lower()
|
|
if choice in {"2", "ipv6", "v6"}:
|
|
return True
|
|
if choice in {"1", "ipv4", "v4"}:
|
|
return False
|
|
return default_ipv6
|
|
|
|
|
|
async def main() -> None:
|
|
validate()
|
|
selected_use_ipv6 = choose_ip_mode(TG_USE_IPV6)
|
|
|
|
print("Starting Telegram login for StringSession generation")
|
|
print(
|
|
"Connection settings:",
|
|
{
|
|
"use_ipv6": selected_use_ipv6,
|
|
"connect_timeout": TG_CONNECT_TIMEOUT,
|
|
"request_retries": TG_REQUEST_RETRIES,
|
|
"connection_retries": TG_CONNECTION_RETRIES,
|
|
"retry_delay": TG_RETRY_DELAY,
|
|
},
|
|
)
|
|
|
|
client = TelegramClient(
|
|
StringSession(),
|
|
API_ID,
|
|
API_HASH,
|
|
use_ipv6=selected_use_ipv6,
|
|
timeout=TG_CONNECT_TIMEOUT,
|
|
request_retries=TG_REQUEST_RETRIES,
|
|
connection_retries=TG_CONNECTION_RETRIES,
|
|
retry_delay=TG_RETRY_DELAY,
|
|
auto_reconnect=True,
|
|
)
|
|
|
|
async with client:
|
|
phone = input("Phone number (international format, e.g. +1234567890): ").strip()
|
|
await client.send_code_request(phone)
|
|
code = input("Telegram code: ").strip()
|
|
|
|
try:
|
|
await client.sign_in(phone=phone, code=code)
|
|
except Exception:
|
|
password = getpass.getpass("2FA password: ")
|
|
await client.sign_in(password=password)
|
|
|
|
session_string = client.session.save()
|
|
session_path = Path(SESSION_FILE_PATH)
|
|
session_path.parent.mkdir(parents=True, exist_ok=True)
|
|
session_path.write_text(session_string + "\n", encoding="utf-8")
|
|
print(f"Session saved to: {session_path.resolve()}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|