File size: 3,296 Bytes
db78256 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
from speedtest import Speedtest, ConfigRetrievalError
from pyrogram import Client
from pyrogram.filters import command
from pyrogram.handlers import MessageHandler
from pyrogram.types import Message
from bot.core.plugin_manager import PluginBase, PluginInfo
from bot.helper.ext_utils.bot_utils import new_task, sync_to_async
from bot.helper.ext_utils.status_utils import get_readable_file_size
from bot.helper.telegram_helper.filters import CustomFilters
from bot.helper.telegram_helper.message_utils import (
send_message,
edit_message,
delete_message,
)
class SpeedtestPlugin(PluginBase):
PLUGIN_INFO = PluginInfo(
name="speedtest_plugin",
version="1.0.0",
author="WZML-X",
description="Speedtest plugin for testing internet speed",
enabled=True,
handlers=[],
commands=["speedtest"],
dependencies=[],
)
async def on_load(self) -> bool:
from bot import LOGGER
LOGGER.info("Speedtest plugin loaded")
return True
async def on_unload(self) -> bool:
from bot import LOGGER
LOGGER.info("Speedtest plugin unloaded")
return True
async def on_enable(self) -> bool:
from bot import LOGGER
LOGGER.info("Speedtest plugin enabled")
return True
async def on_disable(self) -> bool:
from bot import LOGGER
LOGGER.info("Speedtest plugin disabled")
return True
@new_task
async def speedtest_command(client: Client, message: Message):
speed = await send_message(message, "<i>Initiating Speedtest...</i>")
try:
speed_results = await sync_to_async(Speedtest)
await sync_to_async(speed_results.get_best_server)
await sync_to_async(speed_results.download)
await sync_to_async(speed_results.upload)
except ConfigRetrievalError:
await edit_message(
speed,
"<b>ERROR:</b> <i>Can't connect to Server at the Moment, Try Again Later !</i>",
)
return
speed_results.results.share()
result = speed_results.results.dict()
string_speed = f"""
➲ <b><i>SPEEDTEST INFO</i></b>
┠ <b>Upload:</b> <code>{get_readable_file_size(result['upload'] / 8)}/s</code>
┠ <b>Download:</b> <code>{get_readable_file_size(result['download'] / 8)}/s</code>
┠ <b>Ping:</b> <code>{result['ping']} ms</code>
┠ <b>Time:</b> <code>{result['timestamp']}</code>
┠ <b>Data Sent:</b> <code>{get_readable_file_size(int(result['bytes_sent']))}</code>
┖ <b>Data Received:</b> <code>{get_readable_file_size(int(result['bytes_received']))}</code>
➲ <b><i>SPEEDTEST SERVER</i></b>
┠ <b>Name:</b> <code>{result['server']['name']}</code>
┠ <b>Country:</b> <code>{result['server']['country']}, {result['server']['cc']}</code>
┠ <b>Sponsor:</b> <code>{result['server']['sponsor']}</code>
┠ <b>Latency:</b> <code>{result['server']['latency']}</code>
┠ <b>Latitude:</b> <code>{result['server']['lat']}</code>
┖ <b>Longitude:</b> <code>{result['server']['lon']}</code>
"""
try:
await send_message(message, string_speed, photo=result["share"])
await delete_message(speed)
except Exception as e:
from bot import LOGGER
LOGGER.error(str(e))
await edit_message(speed, string_speed)
|