Sections

wbb/core/sections.py

@@ -0,0 +1,12 @@
+def section(title: str, body: dict, indent: int = 2, underline: bool = True) -> str:
+    text = (
+        f"**--{title}**--:\n"
+        if underline
+        else f"**{title}**:\n"
+    )
+    for key, value in body.items():
+        if not isinstance(value, list):
+            text += f"{' ' * indent}**{key}:** `{value}`\n"
+        else:
+            text += f"{' ' * indent}**{key}:** {value[0]}\n"
+    return text

wbb/modules/anime.py

@@ -1,345 +0,0 @@
-# Part of Pull Req #2 by @MaskedVirus | github.com/swatv3nub
-
-import time
-from datetime import timedelta
-
-import requests
-from pyrogram import filters
-from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup
-
-from wbb import app
-from wbb.core.decorators.errors import capture_err
-
-__MODULE__ = "Anime"
-__HELP__ = """
-/anime - search anime on AniList
-/manga - search manga on Anilist
-/char - search character on Anilist
-"""
-
-
-def shorten(description, info="anilist.co"):
-    ms_g = ""
-    if len(description) > 700:
-        description = description[0:500] + "...."
-        ms_g += (
-            f"\n**Description**: __{description}__[More here]({info})"
-        )
-    else:
-        ms_g += f"\n**Description**: __{description}__"
-    return (
-        ms_g.replace("<br>", "")
-        .replace("</br>", "")
-        .replace("<i>", "")
-        .replace("</i>", "")
-    )
-
-
-def t(milliseconds: int) -> str:
-    """Inputs time in milliseconds, to get beautified time,
-    as string"""
-    seconds, milliseconds = divmod(int(milliseconds), 1000)
-    minutes, seconds = divmod(seconds, 60)
-    hours, minutes = divmod(minutes, 60)
-    days, hours = divmod(hours, 24)
-    tmp = (
-        ((str(days) + " Days, ") if days else "")
-        + ((str(hours) + " Hours, ") if hours else "")
-        + ((str(minutes) + " Minutes, ") if minutes else "")
-        + ((str(seconds) + " Seconds, ") if seconds else "")
-        + ((str(milliseconds) + " ms, ") if milliseconds else "")
-    )
-    return tmp[:-2]
-
-
-airing_query = """
-    query ($id: Int,$search: String) {
-      Media (id: $id, type: ANIME,search: $search) {
-        id
-        episodes
-        title {
-          romaji
-          english
-          native
-        }
-        siteUrl
-        nextAiringEpisode {
-           airingAt
-           timeUntilAiring
-           episode
-        }
-      }
-    }
-    """
-
-fav_query = """
-query ($id: Int) {
-      Media (id: $id, type: ANIME) {
-        id
-        title {
-          romaji
-          english
-          native
-        }
-     }
-}
-"""
-
-anime_query = """
-   query ($id: Int,$search: String) {
-      Media (id: $id, type: ANIME,search: $search) {
-        id
-        idMal
-        title {
-          romaji
-          english
-          native
-        }
-        description (asHtml: false)
-        startDate{
-            year
-          }
-          episodes
-          season
-          type
-          format
-          status
-          duration
-          siteUrl
-          studios{
-              nodes{
-                   name
-              }
-          }
-          trailer{
-               id
-               site
-               thumbnail
-          }
-          averageScore
-          genres
-          bannerImage
-      }
-    }
-"""
-character_query = """
-    query ($query: String) {
-        Character (search: $query) {
-               id
-               name {
-                     first
-                     last
-                     full
-               }
-               siteUrl
-               favourites
-               image {
-                        large
-               }
-               description
-        }
-    }
-"""
-
-manga_query = """
-query ($id: Int,$search: String) {
-      Media (id: $id, type: MANGA,search: $search) {
-        id
-        title {
-          romaji
-          english
-          native
-        }
-        description (asHtml: false)
-        startDate{
-            year
-          }
-          type
-          format
-          status
-          siteUrl
-          averageScore
-          genres
-          bannerImage
-      }
-    }
-"""
-
-
-def format_bytes(size):
-    size = int(size)
-    power = 1024
-    n = 0
-    power_labels = {0: "", 1: "K", 2: "M", 3: "G", 4: "T"}
-    while size > power:
-        size /= power
-        n += 1
-    return f"{size:.2f} {power_labels[n]+'B'}"
-
-
-def return_progress_string(current, total):
-    filled_length = int(30 * current // total)
-    return (
-        "[" + "=" * filled_length + " " * (30 - filled_length) + "]"
-    )
-
-
-def calculate_eta(current, total, start_time):
-    if not current:
-        return "00:00:00"
-    end_time = time.time()
-    elapsed_time = end_time - start_time
-    seconds = (elapsed_time * (total / current)) - elapsed_time
-    thing = "".join(
-        str(timedelta(seconds=seconds)).split(".")[:-1]
-    ).split(", ")
-    thing[-1] = thing[-1].rjust(8, "0")
-    return ", ".join(thing)
-
-
-url = "https://graphql.anilist.co"
-
-
-@app.on_message(filters.command("anime"))
-@capture_err
-async def anime_search(_, message):
-    if len(message.command) < 2:
-        await message.delete()
-        return
-    search = message.text.split(None, 1)[1]
-    variables = {"search": search}
-    json = (
-        requests.post(
-            url, json={"query": anime_query, "variables": variables}
-        )
-        .json()["data"]
-        .get("Media", None)
-    )
-    if json:
-        msg = f"**{json['title']['romaji']}**(`{json['title']['native']}`)\n**Type**: {json['format']}\n**Status**: {json['status']}\n**Episodes**: {json.get('episodes', 'N/A')}\n**Duration**: {json.get('duration', 'N/A')} Per Ep.\n**Score**: {json['averageScore']}\n**Genres**: `"
-        for x in json["genres"]:
-            msg += f"{x}, "
-        msg = msg[:-2] + "`\n"
-        msg += "**Studios**: `"
-        for x in json["studios"]["nodes"]:
-            msg += f"{x['name']}, "
-        msg = msg[:-2] + "`\n"
-        info = json.get("siteUrl")
-        trailer = json.get("trailer", None)
-        if trailer:
-            trailer_id = trailer.get("id", None)
-            site = trailer.get("site", None)
-            if site == "youtube":
-                trailer = "https://youtu.be/" + trailer_id
-        description = (
-            json.get("description", "N/A")
-            .replace("<i>", "")
-            .replace("</i>", "")
-            .replace("<br>", "")
-        )
-        msg += shorten(description, info)
-        image = info.replace(
-            "anilist.co/anime/", "img.anili.st/media/"
-        )
-        if trailer:
-            buttons = [
-                [
-                    InlineKeyboardButton("More Info", url=info),
-                    InlineKeyboardButton("Trailer", url=trailer),
-                ]
-            ]
-        else:
-            buttons = [[InlineKeyboardButton("More Info", url=info)]]
-        if image:
-            try:
-                await message.reply_photo(
-                    image,
-                    caption=msg,
-                    reply_markup=InlineKeyboardMarkup(buttons),
-                )
-            except Exception:
-                msg += f" [✔️️]({image})"
-                await message.edit(msg)
-        else:
-            await message.edit(msg)
-
-
-@app.on_message(filters.command("manga"))
-@capture_err
-async def manga_search(_, message):
-    if len(message.command) < 2:
-        await message.delete()
-        return
-    search = message.text.split(None, 1)[1]
-    variables = {"search": search}
-    json = (
-        requests.post(
-            url, json={"query": manga_query, "variables": variables}
-        )
-        .json()["data"]
-        .get("Media", None)
-    )
-    ms_g = ""
-    if json:
-        title, title_native = json["title"].get(
-            "romaji", False
-        ), json["title"].get("native", False)
-        start_date, status, score = (
-            json["startDate"].get("year", False),
-            json.get("status", False),
-            json.get("averageScore", False),
-        )
-        if title:
-            ms_g += f"**{title}**"
-            if title_native:
-                ms_g += f"(`{title_native}`)"
-        if start_date:
-            ms_g += f"\n**Start Date** - `{start_date}`"
-        if status:
-            ms_g += f"\n**Status** - `{status}`"
-        if score:
-            ms_g += f"\n**Score** - `{score}`"
-        ms_g += "\n**Genres** - "
-        for x in json.get("genres", []):
-            ms_g += f"{x}, "
-        ms_g = ms_g[:-2]
-
-        image = json.get("bannerImage", False)
-        ms_g += f"_{json.get('description', None)}_"
-        if image:
-            try:
-                await message.reply_photo(image, caption=ms_g)
-            except Exception:
-                ms_g += f" [✔️️]({image})"
-                await message.reply(ms_g)
-        else:
-            await message.reply(ms_g)
-
-
-@app.on_message(filters.command("char"))
-@capture_err
-async def character_search(_, message):
-    if len(message.command) < 2:
-        await message.delete()
-        return
-    search = message.text.split(None, 1)[1]
-    variables = {"query": search}
-    json = (
-        requests.post(
-            url,
-            json={"query": character_query, "variables": variables},
-        )
-        .json()["data"]
-        .get("Character", None)
-    )
-    if json:
-        ms_g = f"**{json.get('name').get('full')}**(`{json.get('name').get('native')}`)\n"
-        description = f"{json['description']}"
-        site_url = json.get("siteUrl")
-        ms_g += shorten(description, site_url)
-        image = json.get("image", None)
-        if image:
-            image = image.get("large")
-            await message.reply_photo(image, caption=ms_g)
-        else:
-            await message.reply(ms_g)

wbb/modules/arq.py

@@ -24,38 +24,23 @@ SOFTWARE.
 from pyrogram import filters
 
 from wbb import ARQ_API_URL, app, arq
-
+from wbb.core.sections import section
 
 @app.on_message(filters.command("arq"))
 async def arq_stats(_, message):
     data = await arq.stats()
     if not data.ok:
         return await message.reply_text(data.result)
-    data = data.result
-    uptime = data.uptime
-    requests = data.requests
-    cpu = data.cpu
-    server_mem = data.memory.server
-    api_mem = data.memory.api
-    disk = data.disk
-    platform = data.platform
-    python_version = data.python
-    users = data.users
-    bot = data.bot
-    statistics = f"""
-**Uptime:** `{uptime}`
-**Requests Since Uptime:** `{requests}`
-**CPU:** `{cpu}`
-**Memory:**
-    **Total Used:** `{server_mem}`
-    **API:** `{api_mem}`
-**Disk:** `{disk}`
-**Platform:** `{platform}`
-**Python:** `{python_version}`
-**Users:** `{users}`
-**Bot:** {bot}
-**Address:** {ARQ_API_URL}
-"""
-    await message.reply_text(
-        statistics, disable_web_page_preview=True
-    )
+    server = data.result
+    body = {
+            "Uptime": server.uptime,
+            "Requests Since Uptime": server.requests,
+            "CPU": server.cpu,
+            "Memory": server.memory.server,
+            "Platform": server.platform,
+            "Python": server.python,
+            "Users": server.users,
+            "Bot": [server.bot],
+            }
+    text = section("A.R.Q", body)
+    await message.reply_text(text)

wbb/modules/info.py

@@ -28,6 +28,7 @@ from pyrogram.types import Message
 
 from wbb import SUDOERS, app
 from wbb.core.decorators.errors import capture_err
+from wbb.core.sections import section
 from wbb.utils.dbfunctions import is_gbanned_user, user_global_karma
 
 __MODULE__ = "Info"
@@ -50,16 +51,17 @@ async def get_user_info(user):
     is_gbanned = await is_gbanned_user(user_id)
     is_sudo = user_id in SUDOERS
     karma = await user_global_karma(user_id)
-    caption = f"""
-**ID:** `{user_id}`
-**DC:** {dc_id}
-**Name:** {first_name}
-**Username:** {("@" + username) if username else None}
-**Mention:** {mention}
-**Sudo:** {is_sudo}
-**Karma:** {karma}
-**Gbanned:** {is_gbanned}
-"""
+    body = {
+            "ID": user_id,
+            "DC": dc_id,
+            "Name": [first_name],
+            "Username": [("@" + username) if username else None],
+            "Mention": [mention],
+            "Sudo": is_sudo,
+            "Karma": karma,
+            "Gbanned": is_gbanned
+            }
+    caption = section("User info", body)
     return [caption, photo_id]
 
 
@@ -76,18 +78,19 @@ async def get_chat_info(chat):
     link = f"[Link](t.me/{username})" if username else None
     dc_id = chat.dc_id
     photo_id = chat.photo.big_file_id if chat.photo else None
-    caption = f"""
-**ID:** `{chat_id}`
-**DC:** {dc_id}
-**Type:** {type}
-**Name:** {title}
-**Username:** {("@" + username) if username else None}
-**Mention:** {link}
-**Members:** {members}
-**Scam:** {is_scam}
-**Restricted:** {is_restricted}
-**Description:** {description}
-"""
+    body = {
+            "ID": chat_id,
+            "DC": dc_id,
+            "Type": type,
+            "Name": [title],
+            "Username": [("@" + username) if username else None],
+            "Mention": [link],
+            "Members": members,
+            "Scam": is_scam,
+            "Restricted": is_restricted,
+            "Description": [description],
+            }
+    caption = section("Chat info", body)
     return [caption, photo_id]
 
 

wbb/modules/stickers.py

@@ -28,7 +28,7 @@ from traceback import format_exc
 from pyrogram import filters
 from pyrogram.errors.exceptions.bad_request_400 import (
     PeerIdInvalid, ShortnameOccupyFailed, StickerPngDimensions,
-    StickerPngNopng, UserIsBlocked)
+    StickerPngNopng, UserIsBlocked, StickerEmojiInvalid)
 from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup
 
 from wbb import BOT_USERNAME, app
@@ -162,7 +162,10 @@ async def kang(client, message):
                 )
                 continue
             else:
-                await add_sticker_to_set(client, stickerset, sticker)
+                try:
+                    await add_sticker_to_set(client, stickerset, sticker)
+                except StickerEmojiInvalid:
+                    return await msg.edit("[ERROR]: INVALID_EMOJI_IN_ARGUMENT")
             break
 
         await msg.edit(