format: auto-format code by ruff.

Signed-off-by: 5hojib <yesiamshojib@gmail.com>

Author: 5hojib

bot/helper/common.py

@@ -230,7 +230,9 @@ async def before_start(self):
             else ["aria2", "!qB"]
         )
         self.included_extensions = self.user_dict.get("INCLUDED_EXTENSIONS") or (
-            included_extensions if "INCLUDED_EXTENSIONS" not in self.user_dict else []
+            included_extensions
+            if "INCLUDED_EXTENSIONS" not in self.user_dict
+            else []
         )
         if not self.rc_flags:
             if self.user_dict.get("RCLONE_FLAGS"):
@@ -274,11 +276,14 @@ async def before_start(self):
             if self.up_dest in self.user_dict["UPLOAD_PATHS"]:
                 self.up_dest = self.user_dict["UPLOAD_PATHS"][self.up_dest]
         elif (
-            "UPLOAD_PATHS" not in self.user_dict
-            or not self.user_dict["UPLOAD_PATHS"]
-        ) and Config.UPLOAD_PATHS:
-            if self.up_dest in Config.UPLOAD_PATHS:
-                self.up_dest = Config.UPLOAD_PATHS[self.up_dest]
+            (
+                "UPLOAD_PATHS" not in self.user_dict
+                or not self.user_dict["UPLOAD_PATHS"]
+            )
+            and Config.UPLOAD_PATHS
+            and self.up_dest in Config.UPLOAD_PATHS
+        ):
+            self.up_dest = Config.UPLOAD_PATHS[self.up_dest]
 
         if self.ffmpeg_cmds:
             if self.user_dict.get("FFMPEG_CMDS", None):

bot/helper/ext_utils/files_utils.py

@@ -243,6 +243,7 @@ async def remove_excluded_files(fpath, ee):
             if f.strip().lower().endswith(tuple(ee)):
                 await remove(ospath.join(root, f))
 
+
 async def remove_non_included_files(fpath, ie):
     for root, _, files in await sync_to_async(walk, fpath):
         if root.strip().endswith("/yt-dlp-thumb"):
@@ -252,6 +253,7 @@ async def remove_non_included_files(fpath, ie):
                 continue
             await remove(ospath.join(root, f))
 
+
 async def join_files(opath):
     files = await listdir(opath)
     results = []

bot/helper/ext_utils/media_utils.py

@@ -1,5 +1,5 @@
 import contextlib
-from asyncio import create_subprocess_exec, gather, sleep, wait_for
+from asyncio import create_subprocess_exec, gather, wait_for
 from asyncio.subprocess import PIPE
 from os import path as ospath
 from re import escape
@@ -11,7 +11,7 @@
 from aioshutil import rmtree
 from PIL import Image
 
-from bot import DOWNLOAD_DIR, LOGGER, cpu_no, threads, cores
+from bot import DOWNLOAD_DIR, LOGGER, cores, threads
 
 from .bot_utils import cmd_exec, sync_to_async
 from .files_utils import get_mime_type, is_archive, is_archive_split
@@ -131,8 +131,8 @@ async def take_ss(video_file, ss_nb) -> bool:
             output = f"{dirpath}/SS.{name}_{i:02}.png"
             cmd = [
                 "taskset",
-                    "-c",
-                    f"{cores}",
+                "-c",
+                f"{cores}",
                 "xtra",
                 "-hide_banner",
                 "-loglevel",
@@ -176,8 +176,8 @@ async def get_audio_thumbnail(audio_file):
     output = ospath.join(output_dir, f"{time()}.jpg")
     cmd = [
         "taskset",
-                    "-c",
-                    f"{cores}",
+        "-c",
+        f"{cores}",
         "xtra",
         "-hide_banner",
         "-loglevel",
@@ -218,8 +218,8 @@ async def get_video_thumbnail(video_file, duration):
     duration = duration // 2
     cmd = [
         "taskset",
-                    "-c",
-                    f"{cores}",
+        "-c",
+        f"{cores}",
         "xtra",
         "-hide_banner",
         "-loglevel",
@@ -264,8 +264,8 @@ async def get_multiple_frames_thumbnail(video_file, layout, keep_screenshots):
     output = ospath.join(output_dir, f"{time()}.jpg")
     cmd = [
         "taskset",
-                    "-c",
-                    f"{cores}",
+        "-c",
+        f"{cores}",
         "xtra",
         "-hide_banner",
         "-loglevel",
@@ -388,7 +388,6 @@ async def _ffmpeg_progress(self):
                         except Exception:
                             self._progress_raw = 0
                             self._eta_raw = 0
-        
 
     async def ffmpeg_cmds(self, ffmpeg, f_path):
         self.clear()
@@ -484,8 +483,8 @@ async def convert_video(self, video_file, ext, retry=False):
         if retry:
             cmd = [
                 "taskset",
-                    "-c",
-                    f"{cores}",
+                "-c",
+                f"{cores}",
                 "xtra",
                 "-hide_banner",
                 "-loglevel",
@@ -513,8 +512,8 @@ async def convert_video(self, video_file, ext, retry=False):
         else:
             cmd = [
                 "taskset",
-                    "-c",
-                    f"{cores}",
+                "-c",
+                f"{cores}",
                 "xtra",
                 "-hide_banner",
                 "-loglevel",
@@ -568,8 +567,8 @@ async def convert_audio(self, audio_file, ext):
         output = f"{base_name}.{ext}"
         cmd = [
             "taskset",
-                    "-c",
-                    f"{cores}",
+            "-c",
+            f"{cores}",
             "xtra",
             "-hide_banner",
             "-loglevel",
@@ -642,8 +641,8 @@ async def sample_video(self, video_file, sample_duration, part_duration):
 
         cmd = [
             "taskset",
-                    "-c",
-                    f"{cores}",
+            "-c",
+            f"{cores}",
             "xtra",
             "-hide_banner",
             "-loglevel",
@@ -707,8 +706,8 @@ async def split(self, f_path, file_, parts, split_size):
             out_path = f_path.replace(file_, f"{base_name}.part{i:03}{extension}")
             cmd = [
                 "taskset",
-                    "-c",
-                    f"{cores}",
+                "-c",
+                f"{cores}",
                 "xtra",
                 "-hide_banner",
                 "-loglevel",

bot/helper/listeners/task_listener.py

@@ -32,7 +32,7 @@
     get_path_size,
     join_files,
     remove_excluded_files,
-    remove_non_included_files
+    remove_non_included_files,
 )
 from bot.helper.ext_utils.links_utils import is_gdrive_id
 from bot.helper.ext_utils.status_utils import get_readable_file_size

bot/helper/mirror_leech_utils/download_utils/yt_dlp_download.py

@@ -1,16 +1,21 @@
+import builtins
+import contextlib
 from logging import getLogger
-from os import path as ospath, listdir
+from os import listdir
+from os import path as ospath
 from re import search as re_search
-from secrets import token_urlsafe
-from yt_dlp import YoutubeDL, DownloadError
 
+from yt_dlp import DownloadError, YoutubeDL
 
-from .... import task_dict_lock, task_dict
-from ...ext_utils.bot_utils import sync_to_async, async_to_sync
-from ...ext_utils.task_manager import check_running_tasks, stop_duplicate_check
-from ...mirror_leech_utils.status_utils.queue_status import QueueStatus
-from ...telegram_helper.message_utils import send_status_message
-from ..status_utils.yt_dlp_status import YtDlpStatus
+from bot import task_dict, task_dict_lock
+from bot.helper.ext_utils.bot_utils import async_to_sync, sync_to_async
+from bot.helper.ext_utils.task_manager import (
+    check_running_tasks,
+    stop_duplicate_check,
+)
+from bot.helper.mirror_leech_utils.status_utils.queue_status import QueueStatus
+from bot.helper.mirror_leech_utils.status_utils.yt_dlp_status import YtDlpStatus
+from bot.helper.telegram_helper.message_utils import send_status_message
 
 LOGGER = getLogger(__name__)
 
@@ -22,14 +27,14 @@ def __init__(self, obj, listener):
 
     def debug(self, msg):
         # Hack to fix changing extension
-        if not self._obj.is_playlist:
-            if match := re_search(
-                r".Merger..Merging formats into..(.*?).$", msg
-            ) or re_search(r".ExtractAudio..Destination..(.*?)$", msg):
-                LOGGER.info(msg)
-                newname = match.group(1)
-                newname = newname.rsplit("/", 1)[-1]
-                self._listener.name = newname
+        if not self._obj.is_playlist and (
+            match := re_search(r".Merger..Merging formats into..(.*?).$", msg)
+            or re_search(r".ExtractAudio..Destination..(.*?)$", msg)
+        ):
+            LOGGER.info(msg)
+            newname = match.group(1)
+            newname = newname.rsplit("/", 1)[-1]
+            self._listener.name = newname
 
     @staticmethod
     def warning(msg):
@@ -116,14 +121,14 @@ def _on_download_progress(self, d):
                     self._listener.size = d["total_bytes_estimate"] or 0
                 self._downloaded_bytes = d["downloaded_bytes"] or 0
                 self._eta = d.get("eta", "-") or "-"
-            try:
+            with contextlib.suppress(builtins.BaseException):
                 self._progress = (self._downloaded_bytes / self._listener.size) * 100
-            except:
-                pass
 
     async def _on_download_start(self, from_queue=False):
         async with task_dict_lock:
-            task_dict[self._listener.mid] = YtDlpStatus(self._listener, self, self._gid)
+            task_dict[self._listener.mid] = YtDlpStatus(
+                self._listener, self, self._gid
+            )
         if not from_queue:
             await self._listener.on_download_start()
             if self._listener.multi <= 1:
@@ -147,7 +152,7 @@ def _extract_meta_data(self):
                 for entry in result["entries"]:
                     if not entry:
                         continue
-                    elif "filesize_approx" in entry:
+                    if "filesize_approx" in entry:
                         self._listener.size += entry.get("filesize_approx", 0) or 0
                     elif "filesize" in entry:
                         self._listener.size += entry.get("filesize", 0) or 0
@@ -163,10 +168,13 @@ def _extract_meta_data(self):
                 realName = ydl.prepare_filename(result, outtmpl=outtmpl_)
                 ext = ospath.splitext(realName)[-1]
                 self._listener.name = (
-                    f"{self._listener.name}{ext}" if self._listener.name else realName
+                    f"{self._listener.name}{ext}"
+                    if self._listener.name
+                    else realName
                 )
                 if not self._ext:
                     self._ext = ext
+        return None
 
     def _download(self, path):
         try:
@@ -356,4 +364,4 @@ def _set_options(self, options):
             else:
                 if key == "writethumbnail" and value is True:
                     self.keep_thumb = True
-                self.opts[key] = value
+                self.opts[key] = value

bot/helper/mirror_leech_utils/gdrive_utils/clone.py

@@ -120,13 +120,19 @@ def _clone_folder(self, folder_name, folder_id, dest_id):
                 file_path = ospath.join(folder_name, file.get("name"))
                 current_dir_id = self.create_directory(file.get("name"), dest_id)
                 self._clone_folder(file_path, file.get("id"), current_dir_id)
-            elif self.listener.included_extensions and not file.get(
-                "name"
-            ).strip().lower().endswith(tuple(self.listener.included_extensions)):
-                continue
-            elif not self.listener.included_extensions and file.get(
-                "name"
-            ).strip().lower().endswith(tuple(self.listener.excluded_extensions)):
+            elif (
+                self.listener.included_extensions
+                and not file.get("name")
+                .strip()
+                .lower()
+                .endswith(tuple(self.listener.included_extensions))
+            ) or (
+                not self.listener.included_extensions
+                and file.get("name")
+                .strip()
+                .lower()
+                .endswith(tuple(self.listener.excluded_extensions))
+            ):
                 continue
             else:
                 self.total_files += 1

bot/helper/mirror_leech_utils/gdrive_utils/download.py

@@ -86,20 +86,20 @@ def _download_folder(self, folder_id, path, folder_name):
                 mime_type = item.get("mimeType")
             if mime_type == self.G_DRIVE_DIR_MIME_TYPE:
                 self._download_folder(file_id, path, filename)
-            elif ospath.isfile(f"{path}{filename}"):
-                continue
-            elif (
-                self.listener.included_extensions
-                and not filename.strip()
-                .lower()
-                .endswith(tuple(self.listener.included_extensions))
-            ):
-                continue
             elif (
-                not self.listener.included_extensions
-                and filename.strip()
-                .lower()
-                .endswith(tuple(self.listener.excluded_extensions))
+                ospath.isfile(f"{path}{filename}")
+                or (
+                    self.listener.included_extensions
+                    and not filename.strip()
+                    .lower()
+                    .endswith(tuple(self.listener.included_extensions))
+                )
+                or (
+                    not self.listener.included_extensions
+                    and filename.strip()
+                    .lower()
+                    .endswith(tuple(self.listener.excluded_extensions))
+                )
             ):
                 continue
             else:

bot/helper/mirror_leech_utils/rclone_utils/transfer.py

@@ -33,7 +33,6 @@ def __init__(self, listener):
         self._sa_index = 0
         self._sa_number = 0
         self._use_service_accounts = Config.USE_SERVICE_ACCOUNTS
-        
 
     @property
     def transferred_size(self):

bot/modules/stats.py

@@ -43,7 +43,9 @@ async def bot_stats(_, message):
     swap = swap_memory()
     memory = virtual_memory()
     per_cpu = cpu_percent(interval=1, percpu=True)
-    per_cpu_str = " | ".join([f"CPU{i+1}: {round(p)}%" for i, p in enumerate(per_cpu)])
+    per_cpu_str = " | ".join(
+        [f"CPU{i + 1}: {round(p)}%" for i, p in enumerate(per_cpu)]
+    )
     stats = f"""
 <b>Commit Date:</b> {commands["commit"]}
 

bot/modules/users_settings.py

@@ -289,7 +289,7 @@ async def get_user_settings(from_user, stype="main"):
             and Config.UPLOAD_PATHS
         ):
             upload_paths = Config.UPLOAD_PATHS
-        if not upload_paths: 
+        if not upload_paths:
             upload_paths = "None"
 
         buttons.data_button("Upload Paths", f"userset {user_id} menu UPLOAD_PATHS")