"""This class represents a Client, the main mean for interacting with Telegram.

def __init__(

self,

session_name: str,

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,

phone_number: str = None,

phone_code: Union[str, callable] = None,

password: str = None,

recovery_code: callable = None,

force_sms: bool = False,

bot_token: str = None,

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

self.phone_code = phone_code

self.password = password

self.recovery_code = recovery_code

self.force_sms = force_sms

self.bot_token = bot_token

self.first_name = first_name

self.last_name = last_name

self.workers = workers

self.workdir = workdir

self.config_file = config_file

self.plugins = plugins

self.no_updates = no_updates

self.takeout = takeout

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

"""Use this method to start the Client after creating it.

if self.is_started:

raise ConnectionError("Client has already been started")

if self.BOT_TOKEN_RE.match(self.session_name):

self.is_bot = True

self.bot_token = self.session_name

self.session_name = self.session_name.split(":")[0]

warnings.warn(

"\nYou are using a bot token as session name.\n"

"It will be deprecated in next update, please use session file name to load "

"existing sessions and bot_token argument to create new sessions.",

DeprecationWarning,

stacklevel=2,

)

self.load_config()

self.load_session()

self.load_plugins()

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

self.session.start()

self.is_started = True

try:

if self.user_id is None:

if self.bot_token is None:

self.is_bot = False

self.authorize_user()

else:

self.is_bot = True

self.authorize_bot()

self.save_session()

if not self.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.date) > Client.OFFLINE_SLEEP:

self.peers_by_username = {}

self.peers_by_phone = {}

self.get_initial_dialogs()

self.get_contacts()

else:

self.send(functions.messages.GetPinnedDialogs())

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

"""Use this method to manually 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):

"""Use this method to restart the Client.

self.stop()

self.start()

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

"""Blocks the program execution until one of the signals are received,

def signal_handler(*args):

self.is_idle = False

for s in stop_signals:

signal(s, signal_handler)

self.is_idle = True

while self.is_idle:

time.sleep(1)

self.stop()

def run(self):

"""Use this method to automatically start and idle a Client.

self.start()

self.idle()

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

"""Use this method to 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):

"""Removes a previously-added update handler.

if isinstance(handler, DisconnectHandler):

self.disconnect_handler = None

else:

self.dispatcher.remove_handler(handler, group)

def stop_transmission(self):

"""Use this method to stop downloading or uploading a file.

raise Client.StopTransmission

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.dc_id = e.x

self.auth_key = Auth(self.dc_id, self.test_mode, self.ipv6, self._proxy).create()

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

self.session.start()

self.authorize_bot()

else:

self.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(self.phone_number, self.api_id, self.api_hash))

except (PhoneMigrate, NetworkMigrate) as e:

self.session.stop()

self.dc_id = e.x

self.auth_key = Auth(self.dc_id, self.test_mode, self.ipv6, self._proxy).create()

self.session = Session(self, self.dc_id, self.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:

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

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(self.phone_number, phone_code_hash, self.phone_code))

except PhoneNumberUnoccupied:

log.warning("Phone number unregistered")

phone_registered = False

continue

else:

try:

r = self.send(

functions.auth.SignUp(

self.phone_number, phone_code_hash, self.phone_code, self.first_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(terms_of_service.id))

self.password = None

self.user_id = r.user.id

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

def fetch_peers(

self, entities: List[Union[types.User, types.Chat, types.ChatForbidden, types.Channel, types.ChannelForbidden]]

):

for entity in entities:

if isinstance(entity, types.User):

user_id = entity.id

access_hash = entity.access_hash

if access_hash is None:

continue

username = entity.username

phone = entity.phone

input_peer = types.InputPeerUser(user_id=user_id, access_hash=access_hash)

self.peers_by_id[user_id] = input_peer

if username is not None:

self.peers_by_username[username.lower()] = input_peer

if phone is not None:

self.peers_by_phone[phone] = input_peer

if isinstance(entity, (types.Chat, types.ChatForbidden)):

chat_id = entity.id

peer_id = -chat_id

input_peer = types.InputPeerChat(chat_id=chat_id)

self.peers_by_id[peer_id] = input_peer

if isinstance(entity, (types.Channel, types.ChannelForbidden)):

channel_id = entity.id

peer_id = int("-100" + str(channel_id))

access_hash = entity.access_hash

if access_hash is None:

continue

username = getattr(entity, "username", None)

input_peer = types.InputPeerChannel(channel_id=channel_id, access_hash=access_hash)

self.peers_by_id[peer_id] = input_peer

if username is not None:

self.peers_by_username[username.lower()] = input_peer

def download_worker(self):

name = threading.current_thread().name

log.debug("{} started".format(name))

while True:

media = self.download_queue.get()

if media is None:

break

temp_file_path = ""

final_file_path = ""

try:

media, file_name, done, progress, progress_args, path = media

file_id = media.file_id

size = media.file_size

directory, file_name = os.path.split(file_name)

directory = directory or "downloads"

try:

decoded = utils.decode(file_id)

fmt = "<iiqqqqi" if len(decoded) > 24 else "<iiqq"

unpacked = struct.unpack(fmt, decoded)

except (AssertionError, binascii.Error, struct.error):

raise FileIdInvalid from None

else:

media_type = unpacked[0]

dc_id = unpacked[1]

id = unpacked[2]

access_hash = unpacked[3]

volume_id = None

secret = None

local_id = None

if len(decoded) > 24:

volume_id = unpacked[4]

secret = unpacked[5]

local_id = unpacked[6]

media_type_str = Client.MEDIA_TYPE_ID.get(media_type, None)

if media_type_str is None:

raise FileIdInvalid("Unknown media type: {}".format(unpacked[0]))

file_name = file_name or getattr(media, "file_name", None)

if not file_name:

if media_type == 3:

extension = ".ogg"

elif media_type in (4, 10, 13):

extension = mimetypes.guess_extension(media.mime_type) or ".mp4"

elif media_type == 5:

extension = mimetypes.guess_extension(media.mime_type) or ".unknown"

elif media_type == 8:

extension = ".webp"

elif media_type == 9:

extension = mimetypes.guess_extension(media.mime_type) or ".mp3"

elif media_type in (0, 1, 2):

extension = ".jpg"

else:

continue

file_name = "{}_{}_{}{}".format(

media_type_str,

datetime.fromtimestamp(getattr(media, "date", None) or time.time()).strftime(

"%Y-%m-%d_%H-%M-%S"

),

self.rnd_id(),

extension,

)

temp_file_path = self.get_file(

dc_id=dc_id,

id=id,

access_hash=access_hash,

volume_id=volume_id,

local_id=local_id,

secret=secret,

size=size,

progress=progress,

progress_args=progress_args,

)

if temp_file_path:

final_file_path = os.path.abspath(re.sub("\\\\", "/", os.path.join(directory, file_name)))

os.makedirs(directory, exist_ok=True)

shutil.move(temp_file_path, final_file_path)

except Exception as e:

log.error(e, exc_info=True)

try:

os.remove(temp_file_path)

except OSError:

pass

else:

# TODO: "" or None for faulty download, which is better?

# os.path methods return "" in case something does not exist, I prefer this.

# For now let's keep None

path[0] = final_file_path or None

finally:

done.set()

log.debug("{} stopped".format(name))

def updates_worker(self):

name = threading.current_thread().name

log.debug("{} started".format(name))

while True:

updates = self.updates_queue.get()

if updates is None:

break

try:

if isinstance(updates, (types.Update, types.UpdatesCombined)):

self.fetch_peers(updates.users)

self.fetch_peers(updates.chats)

for update in updates.updates:

channel_id = getattr(

getattr(getattr(update, "message", None), "to_id", None), "channel_id", None

) or getattr(update, "channel_id", None)

pts = getattr(update, "pts", None)

pts_count = getattr(update, "pts_count", None)

if isinstance(update, types.UpdateChannelTooLong):

log.warning(update)

if isinstance(update, types.UpdateNewChannelMessage):

message = update.message

if not isinstance(message, types.MessageEmpty):

try:

diff = self.send(

functions.updates.GetChannelDifference(

channel=self.resolve_peer(int("-100" + str(channel_id))),

filter=types.ChannelMessagesFilter(

ranges=[

types.MessageRange(

min_id=update.message.id, max_id=update.message.id

)

]

),

pts=pts - pts_count,

limit=pts,

)

)

except ChannelPrivate:

pass

else:

if not isinstance(diff, types.updates.ChannelDifferenceEmpty):

updates.users += diff.users

updates.chats += diff.chats

if channel_id and pts:

if channel_id not in self.channels_pts:

self.channels_pts[channel_id] = []

if pts in self.channels_pts[channel_id]:

continue

self.channels_pts[channel_id].append(pts)

if len(self.channels_pts[channel_id]) > 50:

self.channels_pts[channel_id] = self.channels_pts[channel_id][25:]

self.dispatcher.updates_queue.put((update, updates.users, updates.chats))

elif isinstance(updates, (types.UpdateShortMessage, types.UpdateShortChatMessage)):

diff = self.send(

functions.updates.GetDifference(pts=updates.pts - updates.pts_count, date=updates.date, qts=-1)

)

if diff.new_messages:

self.dispatcher.updates_queue.put(

(

types.UpdateNewMessage(

message=diff.new_messages[0], pts=updates.pts, pts_count=updates.pts_count

),

diff.users,

diff.chats,

)

)

else:

self.dispatcher.updates_queue.put((diff.other_updates[0], [], []))

elif isinstance(updates, types.UpdateShort):

self.dispatcher.updates_queue.put((updates.update, [], []))

elif isinstance(updates, types.UpdatesTooLong):

log.warning(updates)

except Exception as e:

log.error(e, exc_info=True)

log.debug("{} stopped".format(name))

def send(self, data: Object, retries: int = Session.MAX_RETRIES, timeout: float = Session.WAIT_TIMEOUT):

"""Use this method to send Raw Function queries.

if not self.is_started:

raise ConnectionError("Client has not been started")

if self.no_updates:

data = functions.InvokeWithoutUpdates(data)

if self.takeout_id:

data = functions.InvokeWithTakeout(self.takeout_id, data)

r = self.session.send(data, retries, timeout)