some ruff fix

bot/helper/ext_utils/status_utils.py

@@ -1,3 +1,4 @@
+import contextlib
 from asyncio import iscoroutinefunction
 from html import escape
 from time import time
@@ -269,11 +270,11 @@ def source(self):
     )
 
 
-async def get_readable_message(sid, is_user, page_no=1, status="All", page_step=1):
+async def get_readable_message(sid, is_user, page_no=1, status="All"):
     msg = ""
     button = None
 
-    tasks = await sync_to_async(getSpecificTasks, status, sid if is_user else None)
+    tasks = await sync_to_async(get_specific_tasks, status, sid if is_user else None)
 
     STATUS_LIMIT = 4
     tasks_no = len(tasks)
@@ -310,7 +311,7 @@ async def get_readable_message(sid, is_user, page_no=1, status="All", page_step=
             msg += f"\n{task.processed_bytes()} of {task.size()}"
             msg += f"\n<b>Speed:</b> {task.speed()}\n<b>Estimated:</b> {task.eta()}"
             if hasattr(task, "seeders_num"):
-                with suppress(Exception):
+                with contextlib.suppress(Exception):
                     msg += f"\n<b>Seeders:</b> {task.seeders_num()} <b>Leechers:</b> {task.leechers_num()}"
         elif tstatus == MirrorStatus.STATUS_SEEDING:
             msg += f"\n<b>Size: </b>{task.size()}"

bot/helper/listeners/qbit_listener.py

@@ -58,14 +58,15 @@ async def _on_seed_finish(tor):
 
 @new_task
 async def _stop_duplicate(tor):
-    if task := await get_task_by_gid(tor.hash[:12]) and task.listener.stop_duplicate:
-        task.listener.name = tor.content_path.rsplit("/", 1)[-1].rsplit(
-            ".!qB",
-            1,
-        )[0]
-        msg, button = await stop_duplicate_check(task.listener)
-        if msg:
-            _on_download_error(msg, tor, button)
+    if task := await get_task_by_gid(tor.hash[:12]):
+        if task.listener.stop_duplicate:
+            task.listener.name = tor.content_path.rsplit("/", 1)[-1].rsplit(
+                ".!qB",
+                1,
+            )[0]
+            msg, button = await stop_duplicate_check(task.listener)
+            if msg:
+                _on_download_error(msg, tor, button)
 
 
 @new_task

bot/helper/mirror_leech_utils/download_utils/yt_dlp_download.py

@@ -1,3 +1,4 @@
+# ruff: noqa: ARG005
 import contextlib
 from logging import getLogger
 from os import listdir

bot/helper/mirror_leech_utils/gdrive_utils/list.py

@@ -239,9 +239,9 @@ async def get_items_buttons(self):
 
     async def get_items(self, itype=""):
         if itype:
-            self.item_type == itype
+            self.item_type = itype
         elif self.list_status == "gdu":
-            self.item_type == "folders"
+            self.item_type = "folders"
         try:
             files = self.get_files_by_folder_id(self.id, self.item_type)
             if self.listener.is_cancelled:

bot/helper/mirror_leech_utils/rclone_utils/list.py

@@ -272,9 +272,9 @@ async def get_path_buttons(self):
 
     async def get_path(self, itype=""):
         if itype:
-            self.item_type == itype
+            self.item_type = itype
         elif self.list_status == "rcu":
-            self.item_type == "--dirs-only"
+            self.item_type = "--dirs-only"
         cmd = [
             "xone",
             "lsjson",

bot/helper/telegram_helper/message_utils.py

@@ -277,13 +277,11 @@ async def update_status_message(sid, force=False):
         page_no = status_dict[sid]["page_no"]
         status = status_dict[sid]["status"]
         is_user = status_dict[sid]["is_user"]
-        page_step = status_dict[sid]["page_step"]
         text, buttons = await get_readable_message(
             sid,
             is_user,
             page_no,
             status,
-            page_step,
         )
         if text is None:
             del status_dict[sid]
@@ -322,13 +320,11 @@ async def send_status_message(msg, user_id=0):
         if sid in list(status_dict.keys()):
             page_no = status_dict[sid]["page_no"]
             status = status_dict[sid]["status"]
-            page_step = status_dict[sid]["page_step"]
             text, buttons = await get_readable_message(
                 sid,
                 is_user,
                 page_no,
                 status,
-                page_step,
             )
             if text is None:
                 del status_dict[sid]

ruff.toml

@@ -4,7 +4,7 @@ line-length = 85
 target-version = "py310"
 
 [lint.mccabe]
-max-complexity = 50
+max-complexity = 100
 
 [lint]
 select = [