"""Pyrogram Client, the main means for interacting with Telegram.

def __init__(

self,

session_name: Union[str, Storage],

api_id: Union[int, str] = None,

api_hash: str = None,

app_version: str = None,

device_model: str = None,

system_version: str = None,

lang_code: str = None,

ipv6: bool = False,

proxy: dict = None,

test_mode: bool = False,

bot_token: str = None,

phone_number: str = None,

phone_code: str = None,

password: str = None,

force_sms: bool = False,

workers: int = BaseClient.WORKERS,

workdir: str = BaseClient.WORKDIR,

config_file: str = BaseClient.CONFIG_FILE,

plugins: dict = None,

parse_mode: str = BaseClient.PARSE_MODES[0],

no_updates: bool = None,

takeout: bool = None,

sleep_threshold: int = Session.SLEEP_THRESHOLD

):

super().__init__()

self.session_name = session_name

self.api_id = int(api_id) if api_id else None

self.api_hash = api_hash

self.app_version = app_version

self.device_model = device_model

self.system_version = system_version

self.lang_code = lang_code

self.ipv6 = ipv6

# TODO: Make code consistent, use underscore for private/protected fields

self._proxy = proxy

self.test_mode = test_mode

self.bot_token = bot_token

self.phone_number = phone_number

self.phone_code = phone_code

self.password = password

self.force_sms = force_sms

self.workers = workers

self.workdir = Path(workdir)

self.config_file = Path(config_file)

self.plugins = plugins

self.parse_mode = parse_mode

self.no_updates = no_updates

self.takeout = takeout

self.sleep_threshold = sleep_threshold

if isinstance(session_name, str):

if session_name == ":memory:" or len(session_name) >= MemoryStorage.SESSION_STRING_SIZE:

session_name = re.sub(r"[\n\s]+", "", session_name)

self.storage = MemoryStorage(session_name)

else:

self.storage = FileStorage(session_name, self.workdir)

elif isinstance(session_name, Storage):

self.storage = session_name

else:

raise ValueError("Unknown storage engine")

self.dispatcher = Dispatcher(self, 0 if no_updates else workers)

def __enter__(self):

return self.start()

def __exit__(self, *args):

try:

self.stop()

except ConnectionError:

pass

@property

def proxy(self):

return self._proxy

@proxy.setter

def proxy(self, value):

if value is None:

self._proxy = None

return

if self._proxy is None:

self._proxy = {}

self._proxy["enabled"] = bool(value.get("enabled", True))

self._proxy.update(value)

def connect(self) -> bool:

"""

if self.is_connected:

raise ConnectionError("Client is already connected")

self.load_config()

self.load_session()

self.session = Session(self, self.storage.dc_id(), self.storage.auth_key())

self.session.start()

self.is_connected = True

return bool(self.storage.user_id())

def disconnect(self):

"""Disconnect the client from Telegram servers.

if not self.is_connected:

raise ConnectionError("Client is already disconnected")

if self.is_initialized:

raise ConnectionError("Can't disconnect an initialized client")

self.session.stop()

self.storage.close()

self.is_connected = False

def initialize(self):

"""Initialize the client by starting up workers.

if not self.is_connected:

raise ConnectionError("Can't initialize a disconnected client")

if self.is_initialized:

raise ConnectionError("Client is already initialized")

self.load_plugins()

if not self.no_updates:

for i in range(self.UPDATES_WORKERS):

self.updates_workers_list.append(

Thread(

target=self.updates_worker,

name="UpdatesWorker#{}".format(i + 1)

)

)

self.updates_workers_list[-1].start()

for i in range(self.DOWNLOAD_WORKERS):

self.download_workers_list.append(

Thread(

target=self.download_worker,

name="DownloadWorker#{}".format(i + 1)

)

)

self.download_workers_list[-1].start()

self.dispatcher.start()

Syncer.add(self)

self.is_initialized = True

def terminate(self):

"""Terminate the client by shutting down workers.

if not self.is_initialized:

raise ConnectionError("Client is already terminated")

if self.takeout_id:

self.send(functions.account.FinishTakeoutSession())

log.warning("Takeout session {} finished".format(self.takeout_id))

Syncer.remove(self)

self.dispatcher.stop()

for _ in range(self.DOWNLOAD_WORKERS):

self.download_queue.put(None)

for i in self.download_workers_list:

i.join()

self.download_workers_list.clear()

if not self.no_updates:

for _ in range(self.UPDATES_WORKERS):

self.updates_queue.put(None)

for i in self.updates_workers_list:

i.join()

self.updates_workers_list.clear()

for i in self.media_sessions.values():

i.stop()

self.media_sessions.clear()

self.is_initialized = False

def send_code(self, phone_number: str) -> SentCode:

"""Send the confirmation code to the given phone number.

phone_number = phone_number.strip(" +")

while True:

try:

r = self.send(

functions.auth.SendCode(

phone_number=phone_number,

api_id=self.api_id,

api_hash=self.api_hash,

settings=types.CodeSettings()

)

)

except (PhoneMigrate, NetworkMigrate) as e:

self.session.stop()

self.storage.dc_id(e.x)

self.storage.auth_key(Auth(self, self.storage.dc_id()).create())

self.session = Session(self, self.storage.dc_id(), self.storage.auth_key())

self.session.start()

else:

return SentCode._parse(r)

def resend_code(self, phone_number: str, phone_code_hash: str) -> SentCode:

"""Re-send the confirmation code using a different type.

phone_number = phone_number.strip(" +")

r = self.send(

functions.auth.ResendCode(

phone_number=phone_number,

phone_code_hash=phone_code_hash

)

)

return SentCode._parse(r)

def sign_in(self, phone_number: str, phone_code_hash: str, phone_code: str) -> Union[User, TermsOfService, bool]:

"""Authorize a user in Telegram with a valid confirmation code.

phone_number = phone_number.strip(" +")

r = self.send(

functions.auth.SignIn(

phone_number=phone_number,

phone_code_hash=phone_code_hash,

phone_code=phone_code

)

)

if isinstance(r, types.auth.AuthorizationSignUpRequired):

if r.terms_of_service:

return TermsOfService._parse(terms_of_service=r.terms_of_service)

return False

else:

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def sign_up(self, phone_number: str, phone_code_hash: str, first_name: str, last_name: str = "") -> User:

"""Register a new user in Telegram.

phone_number = phone_number.strip(" +")

r = self.send(

functions.auth.SignUp(

phone_number=phone_number,

first_name=first_name,

last_name=last_name,

phone_code_hash=phone_code_hash

)

)

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def sign_in_bot(self, bot_token: str) -> User:

"""Authorize a bot using its bot token generated by BotFather.

while True:

try:

r = self.send(

functions.auth.ImportBotAuthorization(

flags=0,

api_id=self.api_id,

api_hash=self.api_hash,

bot_auth_token=bot_token

)

)

except UserMigrate as e:

self.session.stop()

self.storage.dc_id(e.x)

self.storage.auth_key(Auth(self, self.storage.dc_id()).create())

self.session = Session(self, self.storage.dc_id(), self.storage.auth_key())

self.session.start()

else:

self.storage.user_id(r.user.id)

self.storage.is_bot(True)

return User._parse(self, r.user)

def get_password_hint(self) -> str:

"""Get your Two-Step Verification password hint.

return self.send(functions.account.GetPassword()).hint

def check_password(self, password: str) -> User:

"""Check your Two-Step Verification password and log in.

r = self.send(

functions.auth.CheckPassword(

password=compute_check(

self.send(functions.account.GetPassword()),

password

)

)

)

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def send_recovery_code(self) -> str:

"""Send a code to your email to recover your password.

return self.send(

functions.auth.RequestPasswordRecovery()

).email_pattern

def recover_password(self, recovery_code: str) -> User:

"""Recover your password with a recovery code and log in.

r = self.send(

functions.auth.RecoverPassword(

code=recovery_code

)

)

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def accept_terms_of_service(self, terms_of_service_id: str) -> bool:

"""Accept the given terms of service.

r = self.send(

functions.help.AcceptTermsOfService(

id=types.DataJSON(

data=terms_of_service_id

)

)

)

assert r

return True

def authorize(self) -> User:

if self.bot_token:

return self.sign_in_bot(self.bot_token)

while True:

try:

if not self.phone_number:

while True:

value = input("Enter phone number or bot token: ")

if not value:

continue

confirm = input("Is \"{}\" correct? (y/N): ".format(value)).lower()

if confirm == "y":

break

if ":" in value:

self.bot_token = value

return self.sign_in_bot(value)

else:

self.phone_number = value

sent_code = self.send_code(self.phone_number)

except BadRequest as e:

print(e.MESSAGE)

self.phone_number = None

self.bot_token = None

else:

break

if self.force_sms:

sent_code = self.resend_code(self.phone_number, sent_code.phone_code_hash)

print("The confirmation code has been sent via {}".format(

{

"app": "Telegram app",

"sms": "SMS",

"call": "phone call",

"flash_call": "phone flash call"

}[sent_code.type]

))

while True:

if not self.phone_code:

self.phone_code = input("Enter confirmation code: ")

try:

signed_in = self.sign_in(self.phone_number, sent_code.phone_code_hash, self.phone_code)

except BadRequest as e:

print(e.MESSAGE)

self.phone_code = None

except SessionPasswordNeeded as e:

print(e.MESSAGE)

while True:

print("Password hint: {}".format(self.get_password_hint()))

if not self.password:

self.password = input("Enter password (empty to recover): ")

try:

if not self.password:

confirm = input("Confirm password recovery (y/n): ")

if confirm == "y":

email_pattern = self.send_recovery_code()

print("The recovery code has been sent to {}".format(email_pattern))

while True:

recovery_code = input("Enter recovery code: ")

try:

return self.recover_password(recovery_code)

except BadRequest as e:

print(e.MESSAGE)

except Exception as e:

log.error(e, exc_info=True)

raise

else:

self.password = None

else:

return self.check_password(self.password)

except BadRequest as e:

print(e.MESSAGE)

self.password = None

else:

break

if isinstance(signed_in, User):

return signed_in

while True:

first_name = input("Enter first name: ")

last_name = input("Enter last name (empty to skip): ")

try:

signed_up = self.sign_up(

self.phone_number,

sent_code.phone_code_hash,

first_name,

last_name

)

except BadRequest as e:

print(e.MESSAGE)

else:

break

if isinstance(signed_in, TermsOfService):

print("\n" + signed_in.text + "\n")

self.accept_terms_of_service(signed_in.id)

return signed_up

def log_out(self):

"""Log out from Telegram and delete the *\\*.session* file.

self.send(functions.auth.LogOut())

self.stop()

self.storage.delete()

return True

def start(self):

"""Start the client.

is_authorized = self.connect()

try:

if not is_authorized:

self.authorize()

if not self.storage.is_bot() and self.takeout:

self.takeout_id = self.send(functions.account.InitTakeoutSession()).id

log.warning("Takeout session {} initiated".format(self.takeout_id))

self.send(functions.updates.GetState())

except (Exception, KeyboardInterrupt):

self.disconnect()

raise

else:

self.initialize()

return self

def stop(self, block: bool = True):

"""Stop the Client.

def do_it():

self.terminate()

self.disconnect()

if block:

do_it()

else:

Thread(target=do_it).start()

return self

def restart(self, block: bool = True):

"""Restart the Client.

def do_it():

self.stop()

self.start()

if block:

do_it()

else:

Thread(target=do_it).start()

return self

@staticmethod

def idle(stop_signals: tuple = (SIGINT, SIGTERM, SIGABRT)):

"""Block the main script execution until a signal is received.

def signal_handler(_, __):

Client.is_idling = False

for s in stop_signals:

signal(s, signal_handler)

Client.is_idling = True

while Client.is_idling:

time.sleep(1)

def run(self):

"""Start the client, idle the main script and finally stop the client.