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

def __init__(self,

session_name: str,

api_id: int or 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: str or callable = None,

password: str = 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_dir: str or None = BaseClient.PLUGINS_DIR):

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

self.first_name = first_name

self.last_name = last_name

self.workers = workers

self.workdir = workdir

self.config_file = config_file

self.plugins_dir = plugins_dir

self.dispatcher = Dispatcher(self, workers)

def __enter__(self):

self.start()

return self

def __exit__(self, exc_type, exc_val, exc_tb):

self.stop()

@property

def proxy(self):

return self._proxy

@proxy.setter

def proxy(self, value):

self._proxy["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.bot_token = self.session_name

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

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

else:

self.authorize_bot()

self.save_session()

if self.bot_token is None:

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)

def stop(self):

"""Use this method to manually stop the Client.

if not self.is_started:

raise ConnectionError("Client is already stopped")

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

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

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_hash_invalid_raises = self.password is not None

first_name_invalid_raises = self.first_name is not None

while True:

if self.phone_number is None:

self.phone_number = input("Enter phone number: ")

while True:

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

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

break

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

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

r = self.send(

functions.auth.SendCode(

self.phone_number,

self.api_id,

self.api_hash

)

)

break

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)

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:

self.phone_code = (

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

else self.phone_code if type(self.phone_code) is str

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

)

try:

if phone_registered:

r = self.send(

functions.auth.SignIn(

self.phone_number,

phone_code_hash,

self.phone_code

)

)

else:

try:

self.send(

functions.auth.SignIn(

self.phone_number,

phone_code_hash,

self.phone_code

)

)

except PhoneNumberUnoccupied:

pass

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

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

r = self.send(

functions.auth.SignUp(

self.phone_number,

phone_code_hash,

self.phone_code,

self.first_name,

self.last_name

)

)

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)

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

while True:

try:

if self.password is None:

print("Hint: {}".format(r.hint))

self.password = getpass.getpass("Enter password: ")

if type(self.password) is str:

self.password = r.current_salt + self.password.encode() + r.current_salt

password_hash = sha256(self.password).digest()

r = self.send(functions.auth.CheckPassword(password_hash))

except PasswordHashInvalid as e:

if password_hash_invalid_raises:

raise

else:

print(e.MESSAGE)

self.password = None

except FloodWait as e:

if password_hash_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)

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)

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("Login successful")

def fetch_peers(self, entities: list):

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.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.put((

types.UpdateNewMessage(

message=diff.new_messages[0],

pts=updates.pts,

pts_count=updates.pts_count

),

diff.users,

diff.chats

))

else:

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

elif isinstance(updates, types.UpdateShort):

self.dispatcher.updates.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")

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

self.fetch_peers(getattr(r, "users", []))

self.fetch_peers(getattr(r, "chats", []))

return r

def load_config(self):

parser = ConfigParser()

parser.read(self.config_file)

if self.api_id and self.api_hash:

pass

else:

if parser.has_section("pyrogram"):

self.api_id = parser.getint("pyrogram", "api_id")

self.api_hash = parser.get("pyrogram", "api_hash")

else:

raise AttributeError(

"No API Key found. "

"More info: https://docs.pyrogram.ml/start/ProjectSetup#configuration"

)

for option in ["app_version", "device_model", "system_version", "lang_code"]:

if getattr(self, option):

pass

else:

if parser.has_section("pyrogram"):

setattr(self, option, parser.get(

"pyrogram",

option,

fallback=getattr(Client, option.upper())

))

else:

setattr(self, option, getattr(Client, option.upper()))

if self._proxy:

self._proxy["enabled"] = True

else:

self._proxy = {}

if parser.has_section("proxy"):

self._proxy["enabled"] = parser.getboolean("proxy", "enabled")

self._proxy["hostname"] = parser.get("proxy", "hostname")

self._proxy["port"] = parser.getint("proxy", "port")

self._proxy["username"] = parser.get("proxy", "username", fallback=None) or None

self._proxy["password"] = parser.get("proxy", "password", fallback=None) or None

def load_session(self):

try:

with open(os.path.join(self.workdir, "{}.session".format(self.session_name)), encoding="utf-8") as f:

s = json.load(f)

except FileNotFoundError:

self.dc_id = 1

self.date = 0

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

else:

self.dc_id = s["dc_id"]

self.test_mode = s["test_mode"]

self.auth_key = base64.b64decode("".join(s["auth_key"]))

self.user_id = s["user_id"]

self.date = s.get("date", 0)

for k, v in s.get("peers_by_id", {}).items():

self.peers_by_id[int(k)] = utils.get_input_peer(int(k), v)

for k, v in s.get("peers_by_username", {}).items():

peer = self.peers_by_id.get(v, None)

if peer:

self.peers_by_username[k] = peer

for k, v in s.get("peers_by_phone", {}).items():

peer = self.peers_by_id.get(v, None)

if peer:

self.peers_by_phone[k] = peer

def load_plugins(self):

if self.plugins_dir is not None:

try:

dirs = os.listdir(self.plugins_dir)

except FileNotFoundError:

if self.plugins_dir == Client.PLUGINS_DIR:

log.info("No plugin loaded: default directory is missing")

else:

log.warning('No plugin loaded: "{}" directory is missing'.format(self.plugins_dir))

else:

plugins_dir = self.plugins_dir.lstrip("./").replace("/", ".")

plugins_count = 0

for i in dirs:

module = import_module("{}.{}".format(plugins_dir, i.split(".")[0]))

for j in dir(module):

# noinspection PyBroadException

try:

handler, group = getattr(module, j)