update

bot/__init__.py

@@ -44,7 +44,7 @@
 getLogger("httpx").setLevel(ERROR)
 getLogger("pymongo").setLevel(ERROR)
 
-botStartTime = time()
+bot_start_time = time()
 
 bot_loop = new_event_loop()
 set_event_loop(bot_loop)

bot/__main__.py

@@ -31,7 +31,7 @@
     scheduler,
     user_data,
     config_dict,
-    botStartTime,
+    bot_start_time,
 )
 
 from .modules import (
@@ -89,7 +89,7 @@ async def stats(_, message):
     memory = virtual_memory()
     stats = (
         f"<b>Commit Date:</b> {last_commit}\n\n"
-        f"<b>Bot Uptime:</b> {get_readable_time(time() - botStartTime)}\n"
+        f"<b>Bot Uptime:</b> {get_readable_time(time() - bot_start_time)}\n"
         f"<b>OS Uptime:</b> {get_readable_time(time() - boot_time())}\n\n"
         f"<b>Total Disk Space:</b> {get_readable_file_size(total)}\n"
         f"<b>Used:</b> {get_readable_file_size(used)} | <b>Free:</b> {get_readable_file_size(free)}\n\n"

bot/helper/ext_utils/db_handler.py

@@ -183,19 +183,19 @@ async def trunc_table(self, name):
     async def get_pm_uids(self):
         if self._return:
             return None
-        return [doc["_id"] async for doc in self._db.pm_users[bot_id].find({})]
+        return [doc["_id"] async for doc in self._db.pm_users[BOT_ID].find({})]
 
     async def update_pm_users(self, user_id):
         if self._return:
             return
-        if not bool(await self._db.pm_users[bot_id].find_one({"_id": user_id})):
-            await self._db.pm_users[bot_id].insert_one({"_id": user_id})
+        if not bool(await self._db.pm_users[BOT_ID].find_one({"_id": user_id})):
+            await self._db.pm_users[BOT_ID].insert_one({"_id": user_id})
             LOGGER.info(f"New PM User Added : {user_id}")
 
     async def rm_pm_user(self, user_id):
         if self._return:
             return
-        await self._db.pm_users[bot_id].delete_one({"_id": user_id})
+        await self._db.pm_users[BOT_ID].delete_one({"_id": user_id})
 
     async def update_user_tdata(self, user_id, token, time):
         if self._return:

bot/helper/ext_utils/files_utils.py

@@ -108,7 +108,7 @@ def clean_all():
     makedirs(DOWNLOAD_DIR, exist_ok=True)
 
 
-def exit_clean_up(signal, frame):
+def exit_clean_up(_, __):
     try:
         LOGGER.info("Please wait, while we clean up and stop the running downloads")
         clean_all()

bot/helper/ext_utils/help_messages.py

@@ -207,8 +207,8 @@
 rclone_cl = """<b>Rclone</b>: path
 If DEFAULT_UPLOAD is `gd` then you can pass up: `rc` to upload to RCLONE_PATH.
 /cmd rcl/rclone_path -up rcl/rclone_path/rc -rcf flagkey:flagvalue|flagkey|flagkey:flagvalue
-/cmd rcl or rclonePath -up rclonePath or rc or rcl
-/cmd mrcc:rclonePath -up rcl or rc(if you have add rclone path from usetting) (to use user config)"""
+/cmd rcl or rclone_path -up rclone_path or rc or rcl
+/cmd mrcc:rclone_path -up rcl or rc(if you have add rclone path from usetting) (to use user config)"""
 
 name_sub = r"""<b>Name Substitution</b>: -ns
 /cmd link -ns script/code/s | mirror/leech | tea/ /s | clone | cpu/ | \[hello\]/hello | \\text\\/text/s

bot/helper/ext_utils/media_utils.py

@@ -1,3 +1,4 @@
+import suppress
 from os import path as ospath
 from os import cpu_count
 from re import escape

bot/helper/ext_utils/status_utils.py

@@ -10,7 +10,7 @@
     task_dict,
     config_dict,
     status_dict,
-    botStartTime,
+    bot_start_time,
     task_dict_lock,
 )
 from bot.helper.telegram_helper.button_build import ButtonMaker
@@ -241,5 +241,5 @@ async def get_readable_message(sid, is_user, page_no=1, status="All", page_step=
     buttons.data_button("♻️", f"status {sid} ref", position="header")
     button = buttons.build_menu(8)
     msg += f"<b>CPU:</b> {cpu_percent()}% | <b>FREE:</b> {get_readable_file_size(disk_usage(DOWNLOAD_DIR).free)}"
-    msg += f"\n<b>RAM:</b> {virtual_memory().percent}% | <b>UPTIME:</b> {get_readable_time(time() - botStartTime)}"
+    msg += f"\n<b>RAM:</b> {virtual_memory().percent}% | <b>UPTIME:</b> {get_readable_time(time() - bot_start_time)}"
     return msg, button

bot/helper/listeners/qbit_listener.py

@@ -45,7 +45,7 @@ async def _on_download_error(err, tor, button=None):
 
 
 @new_task
-async def _onSeedFinish(tor):
+async def _on_seed_finish(tor):
     ext_hash = tor.hash
     LOGGER.info(f"Cancelling Seed: {tor.name}")
     if task := await get_task_by_gid(ext_hash[:12]):
@@ -191,7 +191,7 @@ async def _qb_listener():
                         and qb_torrents[tag]["seeding"]
                     ):
                         qb_torrents[tag]["seeding"] = False
-                        await _onSeedFinish(tor_info)
+                        await _on_seed_finish(tor_info)
                         await sleep(0.5)
             except Exception as e:
                 LOGGER.error(str(e))

bot/helper/listeners/task_listener.py

@@ -281,7 +281,7 @@ async def on_download_complete(self):
             )
 
     async def on_upload_complete(
-        self, link, files, folders, mime_type, rclonePath="", dir_id=""
+        self, link, files, folders, mime_type, rclone_path="", dir_id=""
     ):
         msg = f"<b>Name: </b><code>{escape(self.name)}</code>\n\n<b>Size: </b>{get_readable_file_size(self.size)}"
         LOGGER.info(f"Task Done: {self.name}")
@@ -307,16 +307,16 @@ async def on_upload_complete(
             if mime_type == "Folder":
                 msg += f"\n<b>SubFolders: </b>{folders}"
                 msg += f"\n<b>Files: </b>{files}"
-            if link or (rclonePath and not self.private_link):
+            if link or (rclone_path and not self.private_link):
                 buttons = ButtonMaker()
                 if link:
                     buttons.url_button("☁️ Cloud Link", link)
                 else:
-                    msg += f"\n\nPath: <code>{rclonePath}</code>"
-                if rclonePath and not self.private_link:
-                    remote, path = rclonePath.split(":", 1)
+                    msg += f"\n\nPath: <code>{rclone_path}</code>"
+                if rclone_path and not self.private_link:
+                    remote, path = rclone_path.split(":", 1)
                     rutils.quote(f"{path}")
-                if not rclonePath and dir_id:
+                if not rclone_path and dir_id:
                     INDEX_URL = ""
                     if self.private_link:
                         INDEX_URL = self.user_dict.get("index_url", "") or ""
@@ -330,7 +330,7 @@ async def on_upload_complete(
                             buttons.url_button("🌐 View Link", share_urls)
                 button = buttons.build_menu(2)
             else:
-                msg += f"\n\nPath: <code>{rclonePath}</code>"
+                msg += f"\n\nPath: <code>{rclone_path}</code>"
                 button = None
             msg += f"\n\n<b>cc: </b>{self.tag}"
             await send_message(self.message, msg, button)

bot/helper/mirror_leech_utils/download_utils/direct_link_generator.py

@@ -168,7 +168,7 @@ def direct_link_generator(link):
             "telbx.net",
         ]
     ):
-        return linkBox(link)
+        return linkbox(link)
     if is_share_link(link):
         if "gdtot" in domain:
             return gdtot(link)
@@ -865,7 +865,7 @@ def shrdsk(url):
     raise DirectDownloadLinkException("ERROR: cannot find direct link in headers")
 
 
-def linkBox(url: str):
+def linkbox(url: str):
     parsed_url = urlparse(url)
     try:
         shareToken = parsed_url.path.split("/")[-1]
@@ -874,7 +874,7 @@ def linkBox(url: str):
 
     details = {"contents": [], "title": "", "total_size": 0}
 
-    def __singleItem(session, itemId):
+    def __single_item(session, itemId):
         try:
             _json = session.get(
                 "https://www.linkbox.to/api/file/detail",
@@ -932,7 +932,7 @@ def __fetch_links(session, _id=0, folderPath=""):
             raise DirectDownloadLinkException("ERROR: data not found")
         try:
             if data["shareType"] == "singleItem":
-                return __singleItem(session, data["itemId"])
+                return __single_item(session, data["itemId"])
         except Exception:
             pass
         if not details["title"]:

bot/helper/mirror_leech_utils/download_utils/telegram_download.py

@@ -56,7 +56,7 @@ async def _on_download_start(self, file_id, from_queue):
                 f"Start Queued Download from Telegram: {self._listener.name}"
             )
 
-    async def _onDownloadProgress(self, current, total):
+    async def _on_download_progress(self, current, _):
         if self._listener.is_cancelled:
             if self.session == "user":
                 user.stop_transmission()
@@ -78,7 +78,7 @@ async def _on_download_complete(self):
     async def _download(self, message, path):
         try:
             download = await message.download(
-                file_name=path, progress=self._onDownloadProgress
+                file_name=path, progress=self._on_download_progress
             )
             if self._listener.is_cancelled:
                 await self._on_download_error("Cancelled by user!")

bot/helper/mirror_leech_utils/gdrive_utils/clone.py

@@ -62,7 +62,7 @@ def clone(self):
                 dir_id = self.create_directory(
                     meta.get("name"), self.listener.up_dest
                 )
-                self._cloneFolder(meta.get("name"), meta.get("id"), dir_id)
+                self._clone_folder(meta.get("name"), meta.get("id"), dir_id)
                 durl = self.G_DRIVE_DIR_BASE_DOWNLOAD_URL.format(dir_id)
                 if self.listener.is_cancelled:
                     LOGGER.info("Deleting cloned data from Drive...")
@@ -73,7 +73,7 @@ def clone(self):
                 mime_type = "Folder"
                 self.listener.size = self.proc_bytes
             else:
-                file = self._copyFile(meta.get("id"), self.listener.up_dest)
+                file = self._copy_file(meta.get("id"), self.listener.up_dest)
                 msg += f'<b>Name: </b><code>{file.get("name")}</code>'
                 durl = self.G_DRIVE_BASE_DOWNLOAD_URL.format(file.get("id"))
                 if mime_type is None:
@@ -105,7 +105,7 @@ def clone(self):
             async_to_sync(self.listener.on_upload_error, msg)
             return None, None, None, None, None
 
-    def _cloneFolder(self, folder_name, folder_id, dest_id):
+    def _clone_folder(self, folder_name, folder_id, dest_id):
         LOGGER.info(f"Syncing: {folder_name}")
         files = self.get_files_by_folder_id(folder_id)
         if len(files) == 0:
@@ -115,14 +115,14 @@ def _cloneFolder(self, folder_name, folder_id, dest_id):
                 self.total_folders += 1
                 file_path = ospath.join(folder_name, file.get("name"))
                 current_dir_id = self.create_directory(file.get("name"), dest_id)
-                self._cloneFolder(file_path, file.get("id"), current_dir_id)
+                self._clone_folder(file_path, file.get("id"), current_dir_id)
             elif (
                 not file.get("name")
                 .lower()
                 .endswith(tuple(self.listener.extension_filter))
             ):
                 self.total_files += 1
-                self._copyFile(file.get("id"), dest_id)
+                self._copy_file(file.get("id"), dest_id)
                 self.proc_bytes += int(file.get("size", 0))
                 self.total_time = int(time() - self._start_time)
             if self.listener.is_cancelled:
@@ -134,7 +134,7 @@ def _cloneFolder(self, folder_name, folder_id, dest_id):
         stop=stop_after_attempt(3),
         retry=retry_if_exception_type(Exception),
     )
-    def _copyFile(self, file_id, dest_id):
+    def _copy_file(self, file_id, dest_id):
         body = {"parents": [dest_id]}
         try:
             return (
@@ -164,7 +164,7 @@ def _copyFile(self, file_id, dest_id):
                     if self.listener.is_cancelled:
                         return None
                     self.switch_service_account()
-                    return self._copyFile(file_id, dest_id)
+                    return self._copy_file(file_id, dest_id)
                 else:
                     LOGGER.error(f"Got: {reason}")
                     raise err

bot/helper/mirror_leech_utils/gdrive_utils/count.py

@@ -48,20 +48,20 @@ def _proceed_count(self, file_id):
         LOGGER.info(f"Counting: {name}")
         mime_type = meta.get("mimeType")
         if mime_type == self.G_DRIVE_DIR_MIME_TYPE:
-            self._gDrive_directory(meta)
+            self._gdrive_directory(meta)
             mime_type = "Folder"
         else:
             if mime_type is None:
                 mime_type = "File"
             self.total_files += 1
-            self._gDrive_file(meta)
+            self._gdrive_file(meta)
         return name, mime_type, self.proc_bytes, self.total_files, self.total_folders
 
-    def _gDrive_file(self, filee):
+    def _gdrive_file(self, filee):
         size = int(filee.get("size", 0))
         self.proc_bytes += size
 
-    def _gDrive_directory(self, drive_folder):
+    def _gdrive_directory(self, drive_folder):
         files = self.get_files_by_folder_id(drive_folder["id"])
         if len(files) == 0:
             return
@@ -75,7 +75,7 @@ def _gDrive_directory(self, drive_folder):
                 mime_type = filee.get("mimeType")
             if mime_type == self.G_DRIVE_DIR_MIME_TYPE:
                 self.total_folders += 1
-                self._gDrive_directory(filee)
+                self._gdrive_directory(filee)
             else:
                 self.total_files += 1
-                self._gDrive_file(filee)
+                self._gdrive_file(filee)

bot/helper/mirror_leech_utils/gdrive_utils/search.py

@@ -17,7 +17,7 @@ def __init__(
         self._is_recursive = is_recursive
         self._item_type = item_type
 
-    def _drive_query(self, dirId, file_name, is_recursive):
+    def _drive_query(self, dir_id, file_name, is_recursive):
         try:
             if is_recursive:
                 if self._stop_dup:
@@ -34,7 +34,7 @@ def _drive_query(self, dirId, file_name, is_recursive):
                     elif self._item_type == "folders":
                         query += f"mimeType = '{self.G_DRIVE_DIR_MIME_TYPE}' and "
                 query += "trashed = false"
-                if dirId == "root":
+                if dir_id == "root":
                     return (
                         self.service.files()
                         .list(
@@ -51,7 +51,7 @@ def _drive_query(self, dirId, file_name, is_recursive):
                     .list(
                         supportsAllDrives=True,
                         includeItemsFromAllDrives=True,
-                        driveId=dirId,
+                        driveId=dir_id,
                         q=query,
                         spaces="drive",
                         pageSize=150,
@@ -62,9 +62,9 @@ def _drive_query(self, dirId, file_name, is_recursive):
                     .execute()
                 )
             if self._stop_dup:
-                query = f"'{dirId}' in parents and name = '{file_name}' and "
+                query = f"'{dir_id}' in parents and name = '{file_name}' and "
             else:
-                query = f"'{dirId}' in parents and "
+                query = f"'{dir_id}' in parents and "
                 file_name = file_name.split()
                 for name in file_name:
                     if name != "":