TelegramBotAdmin/wbb/modules/sudoers.py

import os
import re
import requests
import time
import wget
from io import BytesIO
from pyrogram import filters, types
from pyrogram.types import Message
import speedtest
import psutil
from wbb.utils import cust_filter, nekobin, formatter
from wbb import app, OWNER_ID, SUDO_USER_ID, bot_start_time, NEOFETCH, JSMAPI

__MODULE__ = "Sudoers"
__HELP__ = '''/log - To Get Logs From Last Run.
/speedtest - To Perform A Speedtest.
/stats - To Check System Status.
/song - To Download Songs From JioSaavn
/upload - To Upload Files Via Direct Link'''


SUDOERS = [OWNER_ID, SUDO_USER_ID]


# Logs Module


@app.on_message(filters.user(SUDOERS) & cust_filter.command("log"))
async def logs_chat(_, message: Message):
    keyb = types.InlineKeyboardMarkup(
        [
            [
                types.InlineKeyboardButton(
                    "Paste on Nekobin", callback_data="paste_log_nekobin"
                )
            ]
        ]
    )
    await message.reply_document(
        "error.log", reply_markup=keyb
    )


def logs_callback(_, __, query):
    if re.match("paste_log_nekobin", query.data):
        return True


logs_create = filters.create(logs_callback)


@app.on_callback_query(logs_create)
async def paste_log_neko(client, query):
    if query.from_user.id == OWNER_ID or SUDO_USER_ID:
        j = open("error.log", "r")
        data = await nekobin.neko(j.read())
        keyb = types.InlineKeyboardMarkup(
            [[types.InlineKeyboardButton("Pasted!", url=f"{data}")]]
        )
        await query.message.edit_caption("Successfully Nekofied",
                                         reply_markup=keyb)
    else:
        await client.answer_callback_query(
            query.id, "'Blue Button Must Press', huh?", show_alert=True
        )


# SpeedTest Module


def speed_convert(size):
    power = 2 ** 10
    zero = 0
    units = {0: "", 1: "Kb/s", 2: "Mb/s", 3: "Gb/s", 4: "Tb/s"}
    while size > power:
        size /= power
        zero += 1
    return f"{round(size, 2)} {units[zero]}"


@app.on_message(
    filters.user(SUDOERS) & cust_filter.command(commands=("speedtest"))
)
async def get_speedtest_result(_, message: Message):
    app.set_parse_mode("markdown")
    m = await message.reply_text("```Performing A Speedtest!```")
    speed = speedtest.Speedtest()
    i = speed.get_best_server()
    j = speed.download()
    k = speed.upload()
    await m.edit(f'''
```Download - {speed_convert(j)}
Upload   - {speed_convert(k)}
Latency  - {round((i["latency"]))} ms
```''')

# Stats Module


@app.on_message(
    filters.user(SUDOERS) & cust_filter.command(commands=("stats"))
)
async def get_stats(_, message: Message):
    bot_uptime = int(time.time() - bot_start_time)
    cpu = psutil.cpu_percent(interval=0.5)
    mem = psutil.virtual_memory().percent
    disk = psutil.disk_usage("/").percent
    if NEOFETCH == "True":
        os.system("neofetch --stdout > neofetch.txt")
        i = open("neofetch.txt", "r")
        read_file = i.read()
        neofetch = (f'''
----------[Neofetch]----------

{read_file}
''')
        i.close()
    else:
        neofetch = "NeoFetch Is Disabled!"
    stats = (f'''
```
----------[Stats]----------

      Uptime: {formatter.get_readable_time((bot_uptime))}
      CPU: {cpu}%
      RAM: {mem}%
      Disk: {disk}%

{neofetch}
```''')
    await message.reply_text(stats)

# Song


@app.on_message(
    filters.user(SUDOERS) & cust_filter.command(commands=("song"))
)
async def song(_, message: Message):
    text = message.text.replace("/song", "")
    query = text.replace(" ", "")
    if text != "":
        m = await message.reply_text("Searching...")
        r = requests.get(f"{JSMAPI}{query}")

        i = 0
        while i < 2:
            sname = r.json()[i]['song']
            slink = r.json()[i]['media_url']
            ssingers = r.json()[i]['singers']
            file = wget.download(slink)
            ffile = file.replace("mp4", "m4a")
            os.rename(file, ffile)
            await message.reply_audio(audio=ffile, title=sname,
                                      performer=ssingers)
            i += 1
            os.remove(ffile)
        await m.delete()
    else:
        await message.reply_text("/song requires an argument.")

# Upload


@app.on_message(
    filters.user(SUDOERS) & cust_filter.command(commands=("upload"))
)
async def upload(_, message: Message):
    if len(message.command) != 2:
        await message.reply_text("/upload requires one argument")
        return

    url = message.command[1]
    data = BytesIO(requests.get(url).content)
    data.name = url.split("/")[-1]
    m = await message.reply_text("Uploading.....")
    await message.reply_document(data)
    await m.delete()