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

terms_of_service_displayed = False

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: Union[str, callable] = None,

password: str = None,

recovery_code: callable = None,

force_sms: bool = False,

first_name: str = None,

last_name: str = None,

workers: int = BaseClient.WORKERS,

workdir: str = BaseClient.WORKDIR,

config_file: str = BaseClient.CONFIG_FILE,

plugins: dict = None,

no_updates: bool = None,

takeout: bool = None

):

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.recovery_code = recovery_code

self.force_sms = force_sms

self.first_name = first_name

self.last_name = last_name

self.workers = workers

self.workdir = Path(workdir)

self.config_file = Path(config_file)

self.plugins = plugins

self.no_updates = no_updates

self.takeout = takeout

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, workers)

def __enter__(self):

return self.start()

def __exit__(self, *args):

self.stop()

@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 start(self):

"""Start the client.

if self.is_started:

raise ConnectionError("Client has already been started")

self.load_config()

self.load_session()

self.load_plugins()

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

self.session.start()

self.is_started = True

try:

if self.storage.user_id is None:

if self.bot_token is None:

self.storage.is_bot = False

self.authorize_user()

else:

self.storage.is_bot = True

self.authorize_bot()

if not self.storage.is_bot:

if self.takeout:

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

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

now = time.time()

if abs(now - self.storage.date) > Client.OFFLINE_SLEEP:

self.get_initial_dialogs()

self.get_contacts()

else:

self.send(functions.messages.GetPinnedDialogs(folder_id=0))

self.get_initial_dialogs_chunk()

else:

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

except Exception as e:

self.is_started = False

self.session.stop()

raise e

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()

mimetypes.init()

Syncer.add(self)

return self

def stop(self):

"""Stop the Client.

if not self.is_started:

raise ConnectionError("Client is already stopped")

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()

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_started = False

self.session.stop()

return self

def restart(self):

"""Restart the Client.

self.stop()

self.start()

@staticmethod

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

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

def signal_handler(*args):

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.

self.start()

Client.idle()

self.stop()

def add_handler(self, handler: Handler, group: int = 0):

"""Register an update handler.

if isinstance(handler, DisconnectHandler):

self.disconnect_handler = handler.callback

else:

self.dispatcher.add_handler(handler, group)

return handler, group

def remove_handler(self, handler: Handler, group: int = 0):

"""Remove a previously-registered update handler.

if isinstance(handler, DisconnectHandler):

self.disconnect_handler = None

else:

self.dispatcher.remove_handler(handler, group)

def stop_transmission(self):

"""Stop downloading or uploading a file.

raise Client.StopTransmission

def export_session_string(self):

"""Export the current authorized session as a serialized string.

return self.storage.export_session_string()

def set_parse_mode(self, parse_mode: Union[str, None] = "combined"):

"""Set the parse mode to be used globally by the client.

if parse_mode not in self.PARSE_MODES:

raise ValueError('parse_mode must be one of {} or None. Not "{}"'.format(

", ".join('"{}"'.format(m) for m in self.PARSE_MODES[:-1]),

parse_mode

))

self.parse_mode = parse_mode

def authorize_bot(self):

try:

r = self.send(

functions.auth.ImportBotAuthorization(

flags=0,

api_id=self.api_id,

api_hash=self.api_hash,

bot_auth_token=self.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()

self.authorize_bot()

else:

self.storage.user_id = r.user.id

print("Logged in successfully as @{}".format(r.user.username))

def authorize_user(self):

phone_number_invalid_raises = self.phone_number is not None

phone_code_invalid_raises = self.phone_code is not None

password_invalid_raises = self.password is not None

first_name_invalid_raises = self.first_name is not None

def default_phone_number_callback():

while True:

phone_number = input("Enter phone number: ")

confirm = input("Is \"{}\" correct? (y/n): ".format(phone_number))

if confirm in ("y", "1"):

return phone_number

elif confirm in ("n", "2"):

continue

while True:

self.phone_number = (

default_phone_number_callback() if self.phone_number is None

else str(self.phone_number()) if callable(self.phone_number)

else str(self.phone_number)

)

self.phone_number = self.phone_number.strip("+")

try:

r = self.send(

functions.auth.SendCode(

phone_number=self.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()

except (PhoneNumberInvalid, PhoneNumberBanned) as e:

if phone_number_invalid_raises:

raise

else:

print(e.MESSAGE)

self.phone_number = None

except FloodWait as e:

if phone_number_invalid_raises:

raise

else:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

except Exception as e:

log.error(e, exc_info=True)

raise

else:

break

phone_registered = r.phone_registered

phone_code_hash = r.phone_code_hash

terms_of_service = r.terms_of_service

if terms_of_service and not Client.terms_of_service_displayed:

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

Client.terms_of_service_displayed = True

if self.force_sms:

self.send(

functions.auth.ResendCode(

phone_number=self.phone_number,

phone_code_hash=phone_code_hash

)

)

while True:

if not phone_registered:

self.first_name = (

input("First name: ") if self.first_name is None

else str(self.first_name()) if callable(self.first_name)

else str(self.first_name)

)

self.last_name = (

input("Last name: ") if self.last_name is None

else str(self.last_name()) if callable(self.last_name)

else str(self.last_name)

)

self.phone_code = (

input("Enter phone code: ") if self.phone_code is None

else str(self.phone_code(self.phone_number)) if callable(self.phone_code)

else str(self.phone_code)

)

try:

if phone_registered:

try:

r = self.send(

functions.auth.SignIn(

phone_number=self.phone_number,

phone_code_hash=phone_code_hash,

phone_code=self.phone_code

)

)

except PhoneNumberUnoccupied:

log.warning("Phone number unregistered")

phone_registered = False

continue

else:

try:

r = self.send(

functions.auth.SignUp(

phone_number=self.phone_number,

phone_code_hash=phone_code_hash,

phone_code=self.phone_code,

first_name=self.first_name,

last_name=self.last_name

)

)

except PhoneNumberOccupied:

log.warning("Phone number already registered")

phone_registered = True

continue

except (PhoneCodeInvalid, PhoneCodeEmpty, PhoneCodeExpired, PhoneCodeHashEmpty) as e:

if phone_code_invalid_raises:

raise

else:

print(e.MESSAGE)

self.phone_code = None

except FirstnameInvalid as e:

if first_name_invalid_raises:

raise

else:

print(e.MESSAGE)

self.first_name = None

except SessionPasswordNeeded as e:

print(e.MESSAGE)

def default_password_callback(password_hint: str) -> str:

print("Hint: {}".format(password_hint))

return input("Enter password (empty to recover): ")

def default_recovery_callback(email_pattern: str) -> str:

print("An e-mail containing the recovery code has been sent to {}".format(email_pattern))

return input("Enter password recovery code: ")

while True:

try:

r = self.send(functions.account.GetPassword())

self.password = (

default_password_callback(r.hint) if self.password is None

else str(self.password(r.hint) or "") if callable(self.password)

else str(self.password)

)

if self.password == "":

r = self.send(functions.auth.RequestPasswordRecovery())

self.recovery_code = (

default_recovery_callback(r.email_pattern) if self.recovery_code is None

else str(self.recovery_code(r.email_pattern)) if callable(self.recovery_code)

else str(self.recovery_code)

)

r = self.send(

functions.auth.RecoverPassword(

code=self.recovery_code

)

)

else:

r = self.send(

functions.auth.CheckPassword(

password=compute_check(r, self.password)

)

)

except (PasswordEmpty, PasswordRecoveryNa, PasswordHashInvalid) as e:

if password_invalid_raises:

raise

else:

print(e.MESSAGE)

self.password = None

self.recovery_code = None

except FloodWait as e:

if password_invalid_raises:

raise

else:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

self.password = None

self.recovery_code = None

except Exception as e:

log.error(e, exc_info=True)

raise

else:

break

break

except FloodWait as e:

if phone_code_invalid_raises or first_name_invalid_raises:

raise

else:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

except Exception as e:

log.error(e, exc_info=True)

raise

else:

break

if terms_of_service:

assert self.send(

functions.help.AcceptTermsOfService(

id=terms_of_service.id

)

)

self.password = None

self.storage.user_id = r.user.id

print("Logged in successfully as {}".format(r.user.first_name))

def fetch_peers(

self,

peers: List[

Union[

types.User,

types.Chat, types.ChatForbidden,

types.Channel, types.ChannelForbidden

]

]

) -> bool:

is_min = False

parsed_peers = []

for peer in peers:

username = None

phone_number = None

if isinstance(peer, types.User):

peer_id = peer.id

access_hash = peer.access_hash

username = peer.username