Refactored

alive.py

@@ -11,11 +11,7 @@
     BASE_URL = None
 
 IS_VPS = os.environ.get('IS_VPS', 'False')
-if IS_VPS.lower() == 'true':
-    IS_VPS = True
-else:
-    IS_VPS = False
-
+IS_VPS = IS_VPS.lower() == 'true'
 if not IS_VPS and BASE_URL is not None:
     while True:
         time.sleep(600)

bot/__init__.py

@@ -274,50 +274,32 @@ def get_client() -> qba.TorrentsAPIMixIn:
     BUTTON_SIX_URL = None
 try:
     STOP_DUPLICATE = getConfig('STOP_DUPLICATE')
-    if STOP_DUPLICATE.lower() == 'true':
-        STOP_DUPLICATE = True
-    else:
-        STOP_DUPLICATE = False
+    STOP_DUPLICATE = STOP_DUPLICATE.lower() == 'true'
 except KeyError:
     STOP_DUPLICATE = False
 try:
     VIEW_LINK = getConfig('VIEW_LINK')
-    if VIEW_LINK.lower() == 'true':
-        VIEW_LINK = True
-    else:
-        VIEW_LINK = False
+    VIEW_LINK = VIEW_LINK.lower() == 'true'
 except KeyError:
     VIEW_LINK = False
 try:
     IS_TEAM_DRIVE = getConfig('IS_TEAM_DRIVE')
-    if IS_TEAM_DRIVE.lower() == 'true':
-        IS_TEAM_DRIVE = True
-    else:
-        IS_TEAM_DRIVE = False
+    IS_TEAM_DRIVE = IS_TEAM_DRIVE.lower() == 'true'
 except KeyError:
     IS_TEAM_DRIVE = False
 try:
     USE_SERVICE_ACCOUNTS = getConfig('USE_SERVICE_ACCOUNTS')
-    if USE_SERVICE_ACCOUNTS.lower() == 'true':
-        USE_SERVICE_ACCOUNTS = True
-    else:
-        USE_SERVICE_ACCOUNTS = False
+    USE_SERVICE_ACCOUNTS = USE_SERVICE_ACCOUNTS.lower() == 'true'
 except KeyError:
     USE_SERVICE_ACCOUNTS = False
 try:
     BLOCK_MEGA_FOLDER = getConfig('BLOCK_MEGA_FOLDER')
-    if BLOCK_MEGA_FOLDER.lower() == 'true':
-        BLOCK_MEGA_FOLDER = True
-    else:
-        BLOCK_MEGA_FOLDER = False
+    BLOCK_MEGA_FOLDER = BLOCK_MEGA_FOLDER.lower() == 'true'
 except KeyError:
     BLOCK_MEGA_FOLDER = False
 try:
     BLOCK_MEGA_LINKS = getConfig('BLOCK_MEGA_LINKS')
-    if BLOCK_MEGA_LINKS.lower() == 'true':
-        BLOCK_MEGA_LINKS = True
-    else:
-        BLOCK_MEGA_LINKS = False
+    BLOCK_MEGA_LINKS = BLOCK_MEGA_LINKS.lower() == 'true'
 except KeyError:
     BLOCK_MEGA_LINKS = False
 try:
@@ -330,10 +312,7 @@ def get_client() -> qba.TorrentsAPIMixIn:
     SHORTENER_API = None
 try:
     IGNORE_PENDING_REQUESTS = getConfig("IGNORE_PENDING_REQUESTS")
-    if IGNORE_PENDING_REQUESTS.lower() == 'true':
-        IGNORE_PENDING_REQUESTS = True
-    else:
-        IGNORE_PENDING_REQUESTS = False
+    IGNORE_PENDING_REQUESTS = IGNORE_PENDING_REQUESTS.lower() == 'true'
 except KeyError:
     IGNORE_PENDING_REQUESTS = False
 try:
@@ -345,10 +324,7 @@ def get_client() -> qba.TorrentsAPIMixIn:
     BASE_URL = None
 try:
     IS_VPS = getConfig('IS_VPS')
-    if IS_VPS.lower() == 'true':
-        IS_VPS = True
-    else:
-        IS_VPS = False
+    IS_VPS = IS_VPS.lower() == 'true'
 except KeyError:
     IS_VPS = False
 try:

bot/__main__.py

@@ -44,23 +44,23 @@ def stats(update, context):
 
 
 def start(update, context):
-    start_string = f'''
-This bot can mirror all your links to Google Drive!
-Type /{BotCommands.HelpCommand} to get a list of available commands
-'''
     buttons = button_build.ButtonMaker()
     buttons.buildbutton("Channel", "https://t.me/Namexian")
     buttons.buildbutton("Group", "https://t.me/EunhaMirror")
     reply_markup = InlineKeyboardMarkup(buttons.build_menu(2))
     LOGGER.info('UID: {} - UN: {} - MSG: {}'.format(update.message.chat.id, update.message.chat.username, update.message.text))
     uptime = get_readable_time((time.time() - botStartTime))
     if CustomFilters.authorized_user(update) or CustomFilters.authorized_chat(update):
-        if update.message.chat.type == "private" :
+        if update.message.chat.type == "private":
             sendMessage(f"Hey I'm Alive 🙂\nSince: <code>{uptime}</code>", context.bot, update)
-        else :
+        else:
+            start_string = f'''
+This bot can mirror all your links to Google Drive!
+Type /{BotCommands.HelpCommand} to get a list of available commands
+'''
             sendMarkup(start_string, context.bot, update, reply_markup)
-    else :
-        sendMessage(f"Oops! not a Authorized user.", context.bot, update)
+    else:
+        sendMessage('Oops! not a Authorized user.', context.bot, update)
 
 
 def restart(update, context):

bot/helper/__init__.py

@@ -4,12 +4,7 @@
 from bot import HEROKU_API_KEY, HEROKU_APP_NAME
 
 
-# Preparing For Setting Config
-# Implement by https://github.com/jusidama18 and Based on this https://github.com/DevsExpo/FridayUserbot/blob/master/plugins/heroku_helpers.py
-
-heroku_client = None
-if HEROKU_API_KEY:
-    heroku_client = heroku3.from_key(HEROKU_API_KEY)
+heroku_client = heroku3.from_key(HEROKU_API_KEY) if HEROKU_API_KEY else None
 
 def check_heroku(func):
     @wraps(func)

bot/helper/ext_utils/bot_utils.py

@@ -72,32 +72,40 @@ def getDownloadByGid(gid):
     with download_dict_lock:
         for dl in download_dict.values():
             status = dl.status()
-            if status != MirrorStatus.STATUS_ARCHIVING and status != MirrorStatus.STATUS_EXTRACTING:
-                if dl.gid() == gid:
-                    return dl
+            if (
+                status
+                not in [
+                    MirrorStatus.STATUS_ARCHIVING,
+                    MirrorStatus.STATUS_EXTRACTING,
+                ]
+                and dl.gid() == gid
+            ):
+                return dl
     return None
 
 
 def getAllDownload():
     with download_dict_lock:
         for dlDetails in download_dict.values():
             status = dlDetails.status()
-            if status != MirrorStatus.STATUS_ARCHIVING and \
-               status != MirrorStatus.STATUS_EXTRACTING and \
-               status != MirrorStatus.STATUS_CLONING and \
-               status != MirrorStatus.STATUS_UPLOADING:
-                if dlDetails:
-                    return dlDetails
+            if (
+                status
+                not in [
+                    MirrorStatus.STATUS_ARCHIVING,
+                    MirrorStatus.STATUS_EXTRACTING,
+                    MirrorStatus.STATUS_CLONING,
+                    MirrorStatus.STATUS_UPLOADING,
+                ]
+                and dlDetails
+            ):
+                return dlDetails
     return None
 
 
 def get_progress_bar_string(status):
     completed = status.processed_bytes() / 8
     total = status.size_raw() / 8
-    if total == 0:
-        p = 0
-    else:
-        p = round(completed * 100 / total)
+    p = 0 if total == 0 else round(completed * 100 / total)
     p = min(max(p, 0), 100)
     cFull = p // 8
     cPart = p % 8 - 1
@@ -125,7 +133,10 @@ def get_readable_message():
             if INDEX > COUNT:
                 msg += f"<b>Filename:</b> <code>{download.name()}</code>"
                 msg += f"\n<b>Status:</b> <i>{download.status()}</i>"
-                if download.status() != MirrorStatus.STATUS_ARCHIVING and download.status() != MirrorStatus.STATUS_EXTRACTING:
+                if download.status() not in [
+                    MirrorStatus.STATUS_ARCHIVING,
+                    MirrorStatus.STATUS_EXTRACTING,
+                ]:
                     msg += f"\n<code>{get_progress_bar_string(download)} {download.progress()}</code>"
                     if download.status() == MirrorStatus.STATUS_CLONING:
                         msg += f"\n<b>Cloned:</b> <code>{get_readable_file_size(download.processed_bytes())}</code> of <code>{download.size()}</code>"
@@ -148,9 +159,8 @@ def get_readable_message():
                         pass
                     msg += f"\n<b>To Stop:</b> <code>/{BotCommands.CancelMirror} {download.gid()}</code>"
                 msg += "\n\n"
-                if STATUS_LIMIT is not None:
-                    if INDEX >= COUNT + STATUS_LIMIT:
-                        break
+                if STATUS_LIMIT is not None and INDEX >= COUNT + STATUS_LIMIT:
+                    break
         if STATUS_LIMIT is not None:
             if INDEX > COUNT + STATUS_LIMIT:
                 return None, None
@@ -186,7 +196,7 @@ def flip(update, context):
 
 
 def check_limit(size, limit, tar_unzip_limit=None, is_tar_ext=False):
-    LOGGER.info(f"Checking File/Folder Size...")
+    LOGGER.info('Checking File/Folder Size...')
     if is_tar_ext and tar_unzip_limit is not None:
         limit = tar_unzip_limit
     if limit is not None:
@@ -220,9 +230,7 @@ def get_readable_time(seconds: int) -> str:
 
 def is_url(url: str):
     url = re.findall(URL_REGEX, url)
-    if url:
-        return True
-    return False
+    return bool(url)
 
 
 def is_gdrive_link(url: str):
@@ -245,9 +253,7 @@ def get_mega_link_type(url: str):
 
 def is_magnet(url: str):
     magnet = re.findall(MAGNET_REGEX, url)
-    if magnet:
-        return True
-    return False
+    return bool(magnet)
 
 
 def new_thread(fn):

bot/helper/ext_utils/db_handler.py

@@ -20,56 +20,52 @@ def disconnect(self):
 
     def db_auth(self,chat_id: int):
         self.connect()
-        if self.err :
+        if self.err:
             return "There's some error check log for details"
-        else:
-            sql = 'INSERT INTO users VALUES ({});'.format(chat_id)
-            self.cur.execute(sql)
-            self.conn.commit()
-            self.disconnect()
-            AUTHORIZED_CHATS.add(chat_id)
-            return 'Authorized successfully'
+        sql = 'INSERT INTO users VALUES ({});'.format(chat_id)
+        self.cur.execute(sql)
+        self.conn.commit()
+        self.disconnect()
+        AUTHORIZED_CHATS.add(chat_id)
+        return 'Authorized successfully'
 
     def db_unauth(self,chat_id: int):
         self.connect()
-        if self.err :
+        if self.err:
             return "There's some error check log for details"
-        else:
-            sql = 'DELETE from users where uid = {};'.format(chat_id)
-            self.cur.execute(sql)
-            self.conn.commit()
-            self.disconnect()
-            AUTHORIZED_CHATS.remove(chat_id)
-            return 'Unauthorized successfully'
+        sql = 'DELETE from users where uid = {};'.format(chat_id)
+        self.cur.execute(sql)
+        self.conn.commit()
+        self.disconnect()
+        AUTHORIZED_CHATS.remove(chat_id)
+        return 'Unauthorized successfully'
 
     def db_addsudo(self,chat_id: int):
         self.connect()
-        if self.err :
+        if self.err:
             return "There's some error check log for details"
+        if chat_id in AUTHORIZED_CHATS:
+            sql = 'UPDATE users SET sudo = TRUE where uid = {};'.format(chat_id)
+            self.cur.execute(sql)
+            self.conn.commit()
+            self.disconnect()
+            SUDO_USERS.add(chat_id)
+            return 'Successfully promoted as Sudo'
         else:
-            if chat_id in AUTHORIZED_CHATS:
-                sql = 'UPDATE users SET sudo = TRUE where uid = {};'.format(chat_id)
-                self.cur.execute(sql)
-                self.conn.commit()
-                self.disconnect()
-                SUDO_USERS.add(chat_id)
-                return 'Successfully promoted as Sudo'
-            else:
-                sql = 'INSERT INTO users VALUES ({},TRUE);'.format(chat_id)
-                self.cur.execute(sql)
-                self.conn.commit()
-                self.disconnect()
-                SUDO_USERS.add(chat_id)
-                return 'Successfully Authorized and promoted as Sudo'
+            sql = 'INSERT INTO users VALUES ({},TRUE);'.format(chat_id)
+            self.cur.execute(sql)
+            self.conn.commit()
+            self.disconnect()
+            SUDO_USERS.add(chat_id)
+            return 'Successfully Authorized and promoted as Sudo'
 
     def db_rmsudo(self,chat_id: int):
         self.connect()
-        if self.err :
+        if self.err:
             return "There's some error check log for details"
-        else:
-            sql = 'UPDATE users SET sudo = FALSE where uid = {};'.format(chat_id)
-            self.cur.execute(sql)
-            self.conn.commit()
-            self.disconnect()
-            SUDO_USERS.remove(chat_id)
-            return 'Successfully removed from Sudo'
+        sql = 'UPDATE users SET sudo = FALSE where uid = {};'.format(chat_id)
+        self.cur.execute(sql)
+        self.conn.commit()
+        self.disconnect()
+        SUDO_USERS.remove(chat_id)
+        return 'Successfully removed from Sudo'

bot/helper/ext_utils/fs_utils.py

@@ -153,5 +153,5 @@ def get_base_name(orig_path: str):
 def get_mime_type(file_path):
     mime = magic.Magic(mime=True)
     mime_type = mime.from_file(file_path)
-    mime_type = mime_type if mime_type else "text/plain"
+    mime_type = mime_type or "text/plain"
     return mime_type

bot/helper/ext_utils/shortenurl.py

@@ -23,7 +23,6 @@ def short_url(longurl):
         return random.choice(linkvertise)
     elif "bitly.com" in SHORTENER:
         s = pyshorteners.Shortener(api_key=SHORTENER_API)
-        bitly = s.bitly.short(longurl)
-        return bitly
+        return s.bitly.short(longurl)
     else:
         return requests.get(f'https://{SHORTENER}/api?api={SHORTENER_API}&url={longurl}&format=text').text