Handle login email requirement
This commit is contained in:
@@ -33,6 +33,7 @@ class ClientWrapper:
|
||||
from telethon import TelegramClient
|
||||
from telethon.errors import SessionPasswordNeededError
|
||||
from telethon.sessions import StringSession
|
||||
from telethon.tl import functions, types
|
||||
except ImportError as exc:
|
||||
raise RuntimeError("Telethon not installed") from exc
|
||||
|
||||
@@ -57,22 +58,14 @@ class ClientWrapper:
|
||||
except SessionPasswordNeededError:
|
||||
password = await self._prompt("Two-step verification enabled. Enter password: ")
|
||||
await self.client.sign_in(password=password)
|
||||
elif login_mode in {"sms", "code"}:
|
||||
elif login_mode in {"sms", "code", "phone"}:
|
||||
if not phone:
|
||||
raise RuntimeError("Phone number required for login")
|
||||
await self.client.connect()
|
||||
if await self.client.is_user_authorized():
|
||||
logger.info("Telethon client already authorized")
|
||||
return
|
||||
sent = await self.client.send_code_request(phone, force_sms=login_mode == "sms")
|
||||
code_type = getattr(sent, "type", None)
|
||||
logger.info("Login code type: %s", type(code_type).__name__ if code_type else "unknown")
|
||||
code = await self._prompt("Please enter the code you received: ")
|
||||
try:
|
||||
await self.client.sign_in(phone=phone, code=code, phone_code_hash=sent.phone_code_hash)
|
||||
except SessionPasswordNeededError:
|
||||
password = await self._prompt("Two-step verification enabled. Enter password: ")
|
||||
await self.client.sign_in(password=password)
|
||||
await self._login_by_code(phone, force_sms=login_mode == "sms", functions=functions, types=types)
|
||||
else:
|
||||
await self.client.start(phone=phone)
|
||||
logger.info("Telethon client connected")
|
||||
@@ -151,6 +144,29 @@ class ClientWrapper:
|
||||
if self.client:
|
||||
await self.client.disconnected
|
||||
|
||||
async def _login_by_code(self, phone: str, force_sms: bool, functions, types) -> None:
|
||||
sent = await self.client.send_code_request(phone, force_sms=force_sms)
|
||||
code_type = getattr(sent, "type", None)
|
||||
logger.info("Login code type: %s", type(code_type).__name__ if code_type else "unknown")
|
||||
if isinstance(code_type, types.auth.SentCodeTypeSetUpEmailRequired):
|
||||
logger.warning("Telegram requires login email verification; sending email code")
|
||||
sent = await self.client(functions.auth.ResetLoginEmailRequest(phone, sent.phone_code_hash))
|
||||
code_type = getattr(sent, "type", None)
|
||||
logger.info("Login code type: %s", type(code_type).__name__ if code_type else "unknown")
|
||||
if isinstance(code_type, types.auth.SentCodeTypeEmailCode):
|
||||
logger.warning("Check your email for the login code")
|
||||
code = await self._prompt("Please enter the code you received: ")
|
||||
try:
|
||||
await self.client.sign_in(phone=phone, code=code, phone_code_hash=sent.phone_code_hash)
|
||||
except Exception as exc:
|
||||
from telethon.errors import SessionPasswordNeededError
|
||||
|
||||
if isinstance(exc, SessionPasswordNeededError):
|
||||
password = await self._prompt("Two-step verification enabled. Enter password: ")
|
||||
await self.client.sign_in(password=password)
|
||||
return
|
||||
raise
|
||||
|
||||
def _print_qr(self, url: str, qr_path: Optional[str] = None, qr_open: bool = False) -> None:
|
||||
logger.info("QR login URL: %s", url)
|
||||
print("Telegram app -> Settings -> Devices -> Scan QR")
|
||||
|
||||
@@ -9,6 +9,7 @@ import os
|
||||
|
||||
from telethon import TelegramClient
|
||||
from telethon.errors import SessionPasswordNeededError
|
||||
from telethon.tl import functions, types
|
||||
from telethon.sessions import StringSession
|
||||
|
||||
|
||||
@@ -49,6 +50,13 @@ async def generate(args: argparse.Namespace) -> None:
|
||||
sent = await client.send_code_request(phone, force_sms=args.sms)
|
||||
code_type = getattr(sent, "type", None)
|
||||
print(f"Code delivery type: {type(code_type).__name__ if code_type else 'unknown'}")
|
||||
if isinstance(code_type, types.auth.SentCodeTypeSetUpEmailRequired):
|
||||
print("Telegram requires login email verification, sending email code...")
|
||||
sent = await client(functions.auth.ResetLoginEmailRequest(phone, sent.phone_code_hash))
|
||||
code_type = getattr(sent, "type", None)
|
||||
print(f"Code delivery type: {type(code_type).__name__ if code_type else 'unknown'}")
|
||||
if isinstance(code_type, types.auth.SentCodeTypeEmailCode):
|
||||
print("Check your email for the login code.")
|
||||
code = input("Enter the code you received: ").strip()
|
||||
try:
|
||||
await client.sign_in(phone=phone, code=code, phone_code_hash=sent.phone_code_hash)
|
||||
|
||||
Reference in New Issue
Block a user