Add interactive_login

This commit is contained in:
Lonami Exo 2023-09-12 20:54:22 +02:00
parent aa83e7b043
commit 03a85dfd1a
3 changed files with 66 additions and 3 deletions

View File

@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
import getpass
import re
from typing import TYPE_CHECKING, Optional, Union from typing import TYPE_CHECKING, Optional, Union
from ...mtproto import RpcError from ...mtproto import RpcError
@ -125,6 +127,67 @@ async def sign_in(
return await complete_login(self, result) return await complete_login(self, result)
async def interactive_login(self: Client) -> User:
try:
return await self.get_me()
except RpcError as e:
if e.code != 401:
raise
phone_or_token = ""
while not re.match(r"\+?[\s()]*\d", phone_or_token):
print("Please enter your phone (+1 23...) or bot token (12:abcd...)")
phone_or_token = input(": ").strip()
# Bot flow
if re.match(r"\d+:", phone_or_token):
user = await self.bot_sign_in(phone_or_token)
try:
print("Signed in as bot:", user.full_name)
except UnicodeEncodeError:
print("Signed in as bot")
return user
# User flow
login_token = await self.request_login_code(phone_or_token)
while True:
code = input("Please enter the code you received: ")
try:
user_or_token = await self.sign_in(login_token, code)
except RpcError as e:
if e.name.startswith("PHONE_CODE"):
print("Invalid code:", e)
else:
raise
else:
break
if isinstance(user_or_token, PasswordToken):
while True:
print("Please enter your password (prompt is hidden; type and press enter)")
password = getpass.getpass(": ")
try:
user = await self.check_password(user_or_token, password)
except RpcError as e:
if e.name.startswith("PASSWORD"):
print("Invalid password:", e)
else:
raise
else:
user = user_or_token
try:
print("Signed in as user:", user.full_name)
except UnicodeEncodeError:
print("Signed in as user")
print("Remember to not break the ToS or you will risk an account ban!")
print("https://telegram.org/tos & https://core.telegram.org/api/terms")
return user
async def get_password_information(client: Client) -> PasswordToken: async def get_password_information(client: Client) -> PasswordToken:
result = client(functions.account.get_password()) result = client(functions.account.get_password())
assert isinstance(result, types.account.Password) assert isinstance(result, types.account.Password)

View File

@ -204,8 +204,8 @@ class Client:
) -> Optional[Filter]: ) -> Optional[Filter]:
return get_handler_filter(self, handler) return get_handler_filter(self, handler)
async def get_me(self) -> None: async def get_me(self) -> User:
await get_me(self) return await get_me(self)
def get_messages( def get_messages(
self, self,

View File

@ -10,7 +10,7 @@ if TYPE_CHECKING:
from .client import Client from .client import Client
async def get_me(self: Client) -> None: async def get_me(self: Client) -> User:
self self
raise NotImplementedError raise NotImplementedError