'Refactored by Sourcery'

Author: Sourcery AI

dbplugins/blacklist.py

@@ -75,8 +75,9 @@ async def on_delete_blacklist(event):
     to_unblacklist = list(
         {trigger.strip() for trigger in text.split("\n") if trigger.strip()}
     )
-    successful = 0
-    for trigger in to_unblacklist:
-        if sql.rm_from_blacklist(event.chat_id, trigger.lower()):
-            successful += 1
+    successful = sum(
+        bool(sql.rm_from_blacklist(event.chat_id, trigger.lower()))
+        for trigger in to_unblacklist
+    )
+
     await event.edit(f"Removed {successful} / {len(to_unblacklist)} from the blacklist")

dbplugins/filters.py

@@ -33,18 +33,15 @@ async def on_snip(event):
         # avoid userbot spam
         # "I demand rights for us bots, we are equal to you humans." -Henri Koivuneva (t.me/UserbotTesting/2698)
         return False
-    snips = get_all_filters(event.chat_id)
-    if snips:
+    if snips := get_all_filters(event.chat_id):
         for snip in snips:
             pattern = r"( |^|[^\w])" + re.escape(snip.keyword) + r"( |$|[^\w])"
             if re.search(pattern, name, flags=re.IGNORECASE):
                 msg_o = await event.client.get_messages(
                     entity=Config.PRIVATE_CHANNEL_BOT_API_ID,
                     ids=int(snip.f_mesg_id)
                 )
-                message_id = event.message.id
-                if event.reply_to_msg_id:
-                    message_id = event.reply_to_msg_id
+                message_id = event.reply_to_msg_id or event.message.id
                 await event.client.send_message(
                     event.chat_id,
                     msg_o.message,
@@ -110,4 +107,4 @@ async def on_snip_delete(event):
 @borg.on(slitu.admin_cmd(pattern="clearallfilters"))
 async def on_all_snip_delete(event):
     remove_all_filters(event.chat_id)
-    await event.edit(f"filters **in current chat** deleted successfully")
+    await event.edit('filters **in current chat** deleted successfully')

dbplugins/locks.py

@@ -120,16 +120,15 @@ async def _(event):
         logger.info(str(e))
         return False
     res = ""
-    current_db_locks = get_locks(event.chat_id)
-    if not current_db_locks:
-        res = "There are no DataBase locks in this chat"
-    else:
+    if current_db_locks := get_locks(event.chat_id):
         res = "Following are the DataBase locks in this chat: \n"
         res += "👉 `bots`: `{}`\n".format(current_db_locks.bots)
         res += "👉 `commands`: `{}`\n".format(current_db_locks.commands)
         res += "👉 `email`: `{}`\n".format(current_db_locks.email)
         res += "👉 `forward`: `{}`\n".format(current_db_locks.forward)
         res += "👉 `url`: `{}`\n".format(current_db_locks.url)
+    else:
+        res = "There are no DataBase locks in this chat"
     current_chat = await event.get_chat()
     try:
         current_api_locks = current_chat.default_banned_rights
@@ -163,9 +162,8 @@ async def check_incoming_messages(event):
         return
     peer_id = event.chat_id
     if is_locked(peer_id, "commands"):
-        entities = event.message.entities
         is_command = False
-        if entities:
+        if entities := event.message.entities:
             for entity in entities:
                 if isinstance(entity, types.MessageEntityBotCommand):
                     is_command = True
@@ -186,9 +184,8 @@ async def check_incoming_messages(event):
             )
             update_lock(peer_id, "forward", False)
     if is_locked(peer_id, "email"):
-        entities = event.message.entities
         is_email = False
-        if entities:
+        if entities := event.message.entities:
             for entity in entities:
                 if isinstance(entity, types.MessageEntityEmail):
                     is_email = True
@@ -201,9 +198,8 @@ async def check_incoming_messages(event):
                 )
                 update_lock(peer_id, "email", False)
     if is_locked(peer_id, "url"):
-        entities = event.message.entities
         is_url = False
-        if entities:
+        if entities := event.message.entities:
             for entity in entities:
                 if isinstance(entity, (types.MessageEntityTextUrl, types.MessageEntityUrl)):
                     is_url = True
@@ -227,34 +223,31 @@ async def _(event):
         return False
     if await slitu.is_admin(event.client, event.chat_id, event.action_message.sender_id):
         return
-    if is_locked(event.chat_id, "bots"):
-        # bots are limited Telegram accounts,
-        # and cannot join by themselves
-        if event.user_added:
-            users_added_by = event.action_message.sender_id
-            is_ban_able = False
-            rights = types.ChatBannedRights(
-                until_date=None,
-                view_messages=True
+    if is_locked(event.chat_id, "bots") and event.user_added:
+        users_added_by = event.action_message.sender_id
+        is_ban_able = False
+        rights = types.ChatBannedRights(
+            until_date=None,
+            view_messages=True
+        )
+        added_users = event.action_message.action.users
+        for user_id in added_users:
+            user_obj = await event.client.get_entity(user_id)
+            if user_obj.bot:
+                is_ban_able = True
+                try:
+                    await event.client(functions.channels.EditBannedRequest(
+                        event.chat_id,
+                        user_obj,
+                        rights
+                    ))
+                except Exception as e:
+                    await event.reply(
+                        "I don't seem to have ADMIN permission here. \n`{}`".format(str(e))
+                    )
+                    update_lock(event.chat_id, "bots", False)
+                    break
+        if Config.G_BAN_LOGGER_GROUP is not None and is_ban_able:
+            ban_reason_msg = await event.reply(
+                "!warn [user](tg://user?id={}) Please Do Not Add BOTs to this chat.".format(users_added_by)
             )
-            added_users = event.action_message.action.users
-            for user_id in added_users:
-                user_obj = await event.client.get_entity(user_id)
-                if user_obj.bot:
-                    is_ban_able = True
-                    try:
-                        await event.client(functions.channels.EditBannedRequest(
-                            event.chat_id,
-                            user_obj,
-                            rights
-                        ))
-                    except Exception as e:
-                        await event.reply(
-                            "I don't seem to have ADMIN permission here. \n`{}`".format(str(e))
-                        )
-                        update_lock(event.chat_id, "bots", False)
-                        break
-            if Config.G_BAN_LOGGER_GROUP is not None and is_ban_able:
-                ban_reason_msg = await event.reply(
-                    "!warn [user](tg://user?id={}) Please Do Not Add BOTs to this chat.".format(users_added_by)
-                )

dbplugins/snip.py

@@ -14,15 +14,12 @@
 @borg.on(slitu.admin_cmd(pattern=r'\#(\S+)', outgoing=True))
 async def on_snip(event):
     name = event.pattern_match.group(1)
-    snip = get_snips(name)
-    if snip:
+    if snip := get_snips(name):
         msg_o = await event.client.get_messages(
             entity=Config.PRIVATE_CHANNEL_BOT_API_ID,
             ids=int(snip.f_mesg_id)
         )
-        message_id = event.message.id
-        if event.reply_to_msg_id:
-            message_id = event.reply_to_msg_id
+        message_id = event.reply_to_msg_id or event.message.id
         media_message = msg_o.media
         if isinstance(media_message, types.MessageMediaWebPage):
             media_message = None

dbplugins/welcome.py

@@ -10,8 +10,7 @@
 
 @borg.on(events.ChatAction())  # pylint:disable=E0602
 async def _(event):
-    cws = get_current_welcome_settings(event.chat_id)
-    if cws:
+    if cws := get_current_welcome_settings(event.chat_id):
         # logger.info(event.stringify())
         """user_added=False,
         user_joined=True,

sql_helpers/__init__.py

@@ -21,5 +21,5 @@ def start() -> scoped_session:
 except AttributeError as e:
     # this is a dirty way for the work-around required for #23
     print("DB_URI is not configured. Features depending on the database might have issues.")
-    print(str(e))
+    print(e)
 

sql_helpers/antiflood_sql.py

@@ -74,8 +74,7 @@ def get_flood_limit(chat_id):
 
 def migrate_chat(old_chat_id, new_chat_id):
     with INSERTION_LOCK:
-        flood = SESSION.query(FloodControl).get(str(old_chat_id))
-        if flood:
+        if flood := SESSION.query(FloodControl).get(str(old_chat_id)):
             CHAT_FLOOD[str(new_chat_id)] = CHAT_FLOOD.get(str(old_chat_id), DEF_OBJ)
             flood.chat_id = str(new_chat_id)
             SESSION.commit()

stdplugins/_help.py

@@ -40,13 +40,11 @@ async def _(event):
     help_string += f"<b>Total Disk Space</b>: <code>{total}</code>\n"
     help_string += f"<b>Used Disk Space</b>: <code>{used}</code>\n"
     help_string += f"<b>Free Disk Space</b>: <code>{free}</code>\n\n"
-    help_string += f"UserBot Forked from https://github.com/udf/uniborg"
-    borg._iiqsixfourstore[str(event.chat_id)] = {}
-    borg._iiqsixfourstore[
-        str(event.chat_id)
-    ][
-        str(event.id)
-    ] = help_string + "\n\n" + s_help_string
+    help_string += 'UserBot Forked from https://github.com/udf/uniborg'
+    borg._iiqsixfourstore[str(event.chat_id)] = {
+        str(event.id): help_string + "\n\n" + s_help_string
+    }
+
     if borg.tgbot:
         tgbot_username = await tgbot.get_me()
         results = await borg.inline_query(  # pylint:disable=E0602
@@ -138,7 +136,7 @@ def check_data_base_heal_th():
         # to check database we will execute raw query
         SESSION.execute("SELECT 1")
     except Exception as e:
-        output = f"❌ {str(e)}"
+        output = f'❌ {e}'
         is_database_working = False
     else:
         output = "✅"

stdplugins/barcode.py

@@ -33,9 +33,7 @@ async def _(event):
             m_list = None
             with open(downloaded_file_name, "rb") as fd:
                 m_list = fd.readlines()
-            message = ""
-            for m in m_list:
-                message += m.decode("UTF-8") + "\r\n"
+            message = "".join(m.decode("UTF-8") + "\r\n" for m in m_list)
             os.remove(downloaded_file_name)
         else:
             message = previous_message.message

stdplugins/chat_file_s_count_er.py

@@ -7,8 +7,7 @@ async def _(event):
     if event.fwd_from:
         return
     entity = event.chat_id
-    input_str = event.pattern_match.group(1)
-    if input_str:
+    if input_str := event.pattern_match.group(1):
         entity = input_str
     status_message = await event.reply(
         "... this might take some time "
@@ -25,9 +24,7 @@ async def _(event):
             if message.file.mime_type not in hmm:
                 hmm[message.file.mime_type] = 0
             hmm[message.file.mime_type] += message.file.size
-    hnm = {}
-    for key in hmm:
-        hnm[key] = slitu.humanbytes(hmm[key])
+    hnm = {key: slitu.humanbytes(hmm[key]) for key in hmm}
     await status_message.edit(
         slitu.yaml_format(hnm),
         parse_mode=slitu.parse_pre

stdplugins/checkrestrictionsbot.py

@@ -34,7 +34,7 @@ def get_restriction_string(a) -> str:
     b = ""
     c = ""
     if isinstance(a, Channel):
-        c = f"[{a.title}](https://t.me/c/{a.id}/{2})"
+        c = f'[{a.title}](https://t.me/c/{a.id}/2)'
     elif isinstance(a, User):
         c = f"[{a.first_name}](tg://user?id={a.id})"
     elif isinstance(a, Chat):

stdplugins/colors.py

@@ -9,9 +9,7 @@ async def _(event):
     if event.fwd_from:
         return
     input_str = event.pattern_match.group(1)
-    message_id = event.message.id
-    if event.reply_to_msg_id:
-        message_id = event.reply_to_msg_id
+    message_id = event.reply_to_msg_id or event.message.id
     if input_str.startswith("#"):
         try:
             usercolor = ImageColor.getrgb(input_str)

stdplugins/count.py

@@ -10,7 +10,7 @@
 """Type `.count` and see Magic."""
 
 @borg.on(slitu.admin_cmd(pattern='count'))
-async def stats(event: NewMessage.Event) -> None:  # pylint: disable = R0912, R0914, R0915
+async def stats(event: NewMessage.Event) -> None:    # pylint: disable = R0912, R0914, R0915
     """Command to get stats about the account"""
     waiting_message = await event.edit('`Collecting stats, Wait Nibba`')
     start_time = time.time()
@@ -30,38 +30,35 @@ async def stats(event: NewMessage.Event) -> None:  # pylint: disable = R0912, R0
     async for dialog in event.client.iter_dialogs():
         entity = dialog.entity
 
-        if isinstance(entity, Channel):
-            # participants_count = (await event.get_participants(dialog, limit=0)).total
-            if entity.broadcast:
-                broadcast_channels += 1
-                if entity.creator or entity.admin_rights:
-                    admin_in_broadcast_channels += 1
-                if entity.creator:
-                    creator_in_channels += 1
-
-            elif entity.megagroup:
-                groups += 1
-                # if participants_count > largest_group_member_count:
-                #     largest_group_member_count = participants_count
-                if entity.creator or entity.admin_rights:
-                    # if participants_count > largest_group_with_admin:
-                    #     largest_group_with_admin = participants_count
-                    admin_in_groups += 1
-                if entity.creator:
-                    creator_in_groups += 1
-
-        elif isinstance(entity, User):
-            private_chats += 1
-            if entity.bot:
-                bots += 1
-
-        elif isinstance(entity, Chat):
+        if isinstance(entity, Channel) and entity.broadcast:
+            broadcast_channels += 1
+            if entity.creator or entity.admin_rights:
+                admin_in_broadcast_channels += 1
+            if entity.creator:
+                creator_in_channels += 1
+
+        elif (
+            isinstance(entity, Channel)
+            and entity.megagroup
+            or not isinstance(entity, Channel)
+            and not isinstance(entity, User)
+            and isinstance(entity, Chat)
+        ):
             groups += 1
+            # if participants_count > largest_group_member_count:
+            #     largest_group_member_count = participants_count
             if entity.creator or entity.admin_rights:
+                # if participants_count > largest_group_with_admin:
+                #     largest_group_with_admin = participants_count
                 admin_in_groups += 1
             if entity.creator:
                 creator_in_groups += 1
 
+        elif not isinstance(entity, Channel) and isinstance(entity, User):
+            private_chats += 1
+            if entity.bot:
+                bots += 1
+
         unread_mentions += dialog.unread_mentions_count
         unread += dialog.unread_count
     stop_time = time.time() - start_time
@@ -87,10 +84,7 @@ async def stats(event: NewMessage.Event) -> None:  # pylint: disable = R0912, R0
 
 
 def make_mention(user):
-    if user.username:
-        return f"@{user.username}"
-    else:
-        return inline_mention(user)
+    return f"@{user.username}" if user.username else inline_mention(user)
 
 
 def inline_mention(user):

stdplugins/dagd.py

@@ -16,8 +16,7 @@ async def _(event):
         return
     input_str = event.pattern_match.group(1)
     sample_url = "https://da.gd/dns/{}".format(input_str)
-    response_api = requests.get(sample_url).text
-    if response_api:
+    if response_api := requests.get(sample_url).text:
         await event.edit("DNS records of {} are \n{}".format(input_str, response_api))
     else:
         await event.edit("i can't seem to find {} on the internet".format(input_str))
@@ -29,8 +28,7 @@ async def _(event):
         return
     input_str = event.pattern_match.group(1)
     sample_url = "https://da.gd/s?url={}".format(input_str)
-    response_api = requests.get(sample_url).text
-    if response_api:
+    if response_api := requests.get(sample_url).text:
         await event.edit("Generated {} for {}.".format(response_api, input_str))
     else:
         await event.edit("something is wrong. please try again later.")

stdplugins/decide.py

@@ -7,9 +7,7 @@
 async def _(event):
     if event.fwd_from:
         return
-    message_id = event.message.id
-    if event.reply_to_msg_id:
-        message_id = event.reply_to_msg_id
+    message_id = event.reply_to_msg_id or event.message.id
     r = requests.get("https://yesno.wtf/api").json()
     await borg.send_message(
         event.chat_id,