'Refactored by Sourcery' (#44)

Co-authored-by: Sourcery AI <>

userge/core/methods/decorators/raw_decorator.py

@@ -197,10 +197,8 @@ async def _both_have_perm(flt: Union['types.raw.Command', 'types.raw.Filter'],
     if flt.check_invite_perm and not (
             (user.can_all or user.can_invite_users) and bot.can_invite_users):
         return False
-    if flt.check_pin_perm and not (
-            (user.can_all or user.can_pin_messages) and bot.can_pin_messages):
-        return False
-    return True
+    return bool(not flt.check_pin_perm or (
+            (user.can_all or user.can_pin_messages) and bot.can_pin_messages))
 
 
 class RawDecorator(RawClient):

userge/core/methods/messages/edit_message_text.py

@@ -90,10 +90,7 @@ async def edit_message_text(self,  # pylint: disable=arguments-differ
                                               reply_markup=reply_markup)
         module = inspect.currentframe().f_back.f_globals['__name__']
         if log:
-            if isinstance(log, bool):
-                args = (msg, module)
-            else:
-                args = (msg, log)
+            args = (msg, module) if isinstance(log, bool) else (msg, log)
             await self._channel.fwd_msg(*args)
         del_in = del_in or Config.MSG_DELETE_TIMEOUT
         if del_in > 0:

userge/core/methods/messages/send_as_file.py

@@ -80,9 +80,6 @@ async def send_as_file(self,
         os.remove(filename)
         module = inspect.currentframe().f_back.f_globals['__name__']
         if log:
-            if isinstance(log, bool):
-                args = (msg, module)
-            else:
-                args = (msg, log)
+            args = (msg, module) if isinstance(log, bool) else (msg, log)
             await self._channel.fwd_msg(*args)
         return types.bound.Message.parse(self, msg, module=module)

userge/core/methods/messages/send_message.py

@@ -101,10 +101,7 @@ async def send_message(self,  # pylint: disable=arguments-differ
                                          reply_markup=reply_markup)
         module = inspect.currentframe().f_back.f_globals['__name__']
         if log:
-            if isinstance(log, bool):
-                args = (msg, module)
-            else:
-                args = (msg, log)
+            args = (msg, module) if isinstance(log, bool) else (msg, log)
             await self._channel.fwd_msg(*args)
         del_in = del_in or Config.MSG_DELETE_TIMEOUT
         if del_in > 0:

userge/core/types/new/channel_logger.py

@@ -201,9 +201,10 @@ async def forward_stored(self,
         if caption:
             u_dict = await client.get_user_dict(user_id)
             chat = await client.get_chat(chat_id)
-            u_dict.update({
-                'chat': chat.title if chat.title else "this group",
-                'count': chat.members_count})
+            u_dict.update(
+                {'chat': chat.title or "this group", 'count': chat.members_count}
+            )
+
             caption = caption.format_map(SafeDict(**u_dict))
         file_id, file_ref = get_file_id_and_ref(message)
         caption, buttons = parse_buttons(caption)

userge/core/types/raw/command.py

@@ -50,24 +50,36 @@ def parse(cls, command: str,  # pylint: disable=arguments-differ
         filters_ = filters.regex(pattern=pattern)
         if filter_me:
             outgoing_flt = filters.create(
-                lambda _, __, m:
-                m.via_bot is None
-                and not (m.forward_sender_name or m.forward_from)
+                lambda _, __, m: m.via_bot is None
+                and not m.forward_sender_name
+                and not m.forward_from
                 and not (m.from_user and m.from_user.is_bot)
-                and (m.outgoing or (m.from_user and m.from_user.is_self))
-                and not (m.chat and m.chat.type == "channel" and m.edit_date)
-                and (m.text and m.text.startswith(trigger) if trigger else True))
+                and ((m.outgoing or (m.from_user and m.from_user.is_self)))
+                and (not m.chat or m.chat.type != "channel" or not m.edit_date)
+                and (m.text and m.text.startswith(trigger) if trigger else True)
+            )
+
             incoming_flt = filters.create(
-                lambda _, __, m:
-                m.via_bot is None
-                and not (m.forward_sender_name or m.forward_from)
+                lambda _, __, m: m.via_bot is None
+                and not m.forward_sender_name
+                and not m.forward_from
                 and not m.outgoing
                 and trigger
-                and m.from_user and m.text
-                and ((m.from_user.id == Config.OWNER_ID)
-                     or (Config.SUDO_ENABLED and (m.from_user.id in Config.SUDO_USERS)
-                         and (cname.lstrip(trigger) in Config.ALLOWED_COMMANDS)))
-                and m.text.startswith(Config.SUDO_TRIGGER))
+                and m.from_user
+                and m.text
+                and (
+                    (
+                        (m.from_user.id == Config.OWNER_ID)
+                        or (
+                            Config.SUDO_ENABLED
+                            and (m.from_user.id in Config.SUDO_USERS)
+                            and (cname.lstrip(trigger) in Config.ALLOWED_COMMANDS)
+                        )
+                    )
+                )
+                and m.text.startswith(Config.SUDO_TRIGGER)
+            )
+
             filters_ = filters_ & (outgoing_flt | incoming_flt)
         return cls(_format_about(about), trigger, pattern, filters=filters_, name=cname, **kwargs)
 

userge/core/types/raw/plugin.py

@@ -97,10 +97,7 @@ async def init(self) -> None:
     def add(self, obj: Union['command.Command', '_filter.Filter']) -> None:
         """ add command or filter to plugin """
         obj.plugin_name = self.name
-        if isinstance(obj, command.Command):
-            type_ = self.commands
-        else:
-            type_ = self.filters
+        type_ = self.commands if isinstance(obj, command.Command) else self.filters
         for flt in type_:
             if flt.name == obj.name:
                 type_.remove(flt)

userge/plugins/bot/utube_inline.py

@@ -62,7 +62,7 @@ def date_formatter(date_):
     async def ytdl_callback(_, c_q: CallbackQuery):
         startTime = time()
         u_id = c_q.from_user.id
-        if not (u_id == Config.OWNER_ID or u_id in Config.SUDO_USERS):
+        if u_id != Config.OWNER_ID and u_id not in Config.SUDO_USERS:
             return await c_q.answer(
                 "𝘿𝙚𝙥𝙡𝙤𝙮 𝙮𝙤𝙪𝙧 𝙤𝙬𝙣 𝙐𝙎𝙀𝙍𝙂𝙀-𝙓",
                 show_alert=True

userge/plugins/fun/kang.py

@@ -82,10 +82,7 @@ async def kang_(message: Message):
             emoji_ = "🤔"
 
         u_name = user.username
-        if u_name:
-            u_name = "@" + u_name
-        else:
-            u_name = user.first_name or user.id
+        u_name = "@" + u_name if u_name else user.first_name or user.id
         packname = f"a{user.id}_by_userge_{pack}"
         custom_packnick = Config.CUSTOM_PACK_NAME or f"{u_name}'s kang pack"
         packnick = f"{custom_packnick} Vol.{pack}"

userge/plugins/fun/spotify_autobio.py

@@ -370,31 +370,28 @@ async def now_playing_(message: Message):
 		for for_rec in get_rec:
 			track = for_rec['track']
 			get_name = track['name']
-			sf = open("status_recent_played_song.txt", "a")
-			sf.write('• __' + get_name +'__' + "\n")
-			sf.close()
+			with open("status_recent_played_song.txt", "a") as sf:
+				sf.write('• __' + get_name +'__' + "\n")
 		f = open("status_recent_played_song.txt", "r+")
 		recent_p = f.read()
 		f.truncate(0)
-		
+
 		device_info = device.json()
 		g_dlist = device_info["devices"][0]
 		device_name = g_dlist['name']
 		device_type = g_dlist['type']
 		device_vol = g_dlist['volume_percent']
-
 	#==================PLAYING_SONGS_INFO=======================================#
 		currently_playing_song = f"{spotify_biox.title} - {spotify_biox.interpret}"
 		currently_playing_song_dur = f"{spotify_biox.progress}/{spotify_biox.duration}"
-
 	#==================ASSINGING_VAR_VLAUE=======================================#
 		status_pn = f"""
 	**Device name:** {device_name} ({device_type}) 
 	**Device volume:** {device_vol}% 
 	**Currently playing song:** {currently_playing_song} 
 	**Duration:** {currently_playing_song_dur} 
 	**Recently played songs:** \n{recent_p}"""
-				
+
 	await message.edit(status_pn)
 
 
@@ -430,11 +427,10 @@ async def sp_recents_(message: Message):
 		get_name = track['name']
 		ex_link = track['external_urls']
 		get_link = ex_link['spotify']
-		f = open("userge/xcache/recent_played_song.txt", "a")
-		f.write('• [' + get_name + ']'+ '(' + get_link + ')'+ "\n")
-		f.close()
+		with open("userge/xcache/recent_played_song.txt", "a") as f:
+			f.write('• [' + get_name + ']'+ '(' + get_link + ')'+ "\n")
 	await message.edit("`Getting recent played songs...`")
 	f = open("userge/xcache/recent_played_song.txt", "r+")
 	recent = f.read()
-	f.truncate(0) 
+	f.truncate(0)
 	await message.edit("🎵 **Recently played songs:**\n" + recent, disable_web_page_preview=True)

userge/plugins/fun/stylish.py

@@ -74,7 +74,7 @@ async def _style_text(message: Message):
     await message.edit('🧙‍♂️ `Doing some magik ...`')
     if message.flags:
         flag_choice = list(message.flags.keys())[0]
-        input_str = message.filtered_input_str if message.filtered_input_str else reply.text
+        input_str = message.filtered_input_str or reply.text
         if flag_choice not in FONT_FLAGS:
             await message.err('Flag is Invalid', del_in=5)
             return

userge/plugins/fun/x.py

@@ -20,13 +20,13 @@ async def usx_(message: Message):
     replied = message.reply_to_message
     await message.edit("𝐗")
     a = []
-    for i in range(2):
+    for _ in range(2):
         r = random.randint(0,255)
         g = random.randint(0,255)
         b = random.randint(0,255)
         rgb = (r,g,b)
         a.append(rgb)
-        
+
     im = Image.open(path)
     gradient = Image.new('RGBA', im.size, color=0)
     draw = ImageDraw.Draw(gradient)
@@ -37,7 +37,7 @@ async def usx_(message: Message):
     for i, color in enumerate(interpolate(f_co, t_co, im.width * 2)):
         draw.line([(i, 0), (0, i)], tuple(color), width=1)
 
-    gradient = gradient.resize(im.size) 
+    gradient = gradient.resize(im.size)
     im_composite = Image.alpha_composite(gradient, im)
     image_name = "grad_x.webp"
     webp_file = os.path.join(Config.DOWN_PATH, image_name)

userge/plugins/misc/download.py

@@ -59,7 +59,7 @@ async def down_load_media(message: Message):
                 if message.process_is_canceled:
                     downloader.stop()
                     raise Exception('Process Canceled!')
-                total_length = downloader.filesize if downloader.filesize else 0
+                total_length = downloader.filesize or 0
                 downloaded = downloader.get_dl_size()
                 percentage = downloader.get_progress() * 100
                 speed = downloader.get_speed(human=True)

userge/plugins/misc/gdrive.py

@@ -764,7 +764,7 @@ async def upload(self) -> None:
                     if self._message.process_is_canceled:
                         downloader.stop()
                         raise Exception('Process Canceled!')
-                    total_length = downloader.filesize if downloader.filesize else 0
+                    total_length = downloader.filesize or 0
                     downloaded = downloader.get_dl_size()
                     percentage = downloader.get_progress() * 100
                     speed = downloader.get_speed(human=True)
@@ -801,7 +801,7 @@ async def upload(self) -> None:
             except Exception as d_e:
                 await self._message.err(d_e)
                 return
-        file_path = dl_loc if dl_loc else self._message.input_str
+        file_path = dl_loc or self._message.input_str
         if not os.path.exists(file_path):
             await self._message.err("invalid file path provided?")
             return

userge/plugins/misc/upload.py

@@ -119,7 +119,7 @@ async def uploadtotg(message: Message):
                 if message.process_is_canceled:
                     downloader.stop()
                     raise Exception('Process Canceled!')
-                total_length = downloader.filesize if downloader.filesize else 0
+                total_length = downloader.filesize or 0
                 downloaded = downloader.get_dl_size()
                 percentage = downloader.get_progress() * 100
                 speed = downloader.get_speed(human=True)

userge/plugins/utils/ocr.py

@@ -68,7 +68,7 @@ async def ocr_gen(message: Message):
 
     if message.reply_to_message:
 
-        lang_code = message.input_str if message.input_str else "eng"
+        lang_code = message.input_str or "eng"
         await message.edit(r"`Trying to Read.. 📖")
         downloaded_file_name = await message.client.download_media(message.reply_to_message)
         test_file = await ocr_space_file(downloaded_file_name, lang_code)

userge/plugins/utils/pmpermit.py

@@ -209,7 +209,7 @@ async def view_current_blockPM_msg(message: Message):
 async def uninvitedPmHandler(message: Message):
     """ pm message handler """
     user_dict = await userge.get_user_dict(message.from_user.id)
-    user_dict.update({'chat': message.chat.title if message.chat.title else "this group"})
+    user_dict.update({'chat': message.chat.title or "this group"})
     if message.from_user.is_verified:
         return
     if message.from_user.id in pmCounter:

userge/utils/functions.py

@@ -110,5 +110,4 @@ def rand_array(array):
 async def download_link(url):
     dest = Config.DOWN_PATH
     obj = SmartDL(url, dest)
-    path = obj.get_dest()
-    return path
+    return obj.get_dest()

userge/utils/progress.py

@@ -60,15 +60,25 @@ async def progress(current: int,
         progress_str = progress_str.format(
             ud_type,
             file_name,
-            ''.join((userge.Config.FINISHED_PROGRESS_STR
-                     for i in range(floor(percentage / 5)))),
-            ''.join((userge.Config.UNFINISHED_PROGRESS_STR
-                     for i in range(20 - floor(percentage / 5)))),
+            ''.join(
+                (
+                    userge.Config.FINISHED_PROGRESS_STR
+                    for i in range(floor(percentage / 5))
+                )
+            ),
+            ''.join(
+                (
+                    userge.Config.UNFINISHED_PROGRESS_STR
+                    for i in range(20 - floor(percentage / 5))
+                )
+            ),
             round(percentage, 2),
             humanbytes(current),
             humanbytes(total),
             humanbytes(speed),
-            time_to_completion if time_to_completion else "0 s")
+            time_to_completion or "0 s",
+        )
+
         try:
             await message.try_to_edit(progress_str)
         except FloodWait as f_e: