mirror of
https://github.com/MuRuLOSE/limoka.git
synced 2026-06-16 22:34:19 +02:00
350 lines
13 KiB
Python
350 lines
13 KiB
Python
"""
|
||
_
|
||
__ _____ ___ ___ ___ __| | ___ _ __
|
||
\ \ / / __|/ _ \/ __/ _ \ / _` |/ _ \ '__|
|
||
\ V /\__ \ __/ (_| (_) | (_| | __/ |
|
||
\_/ |___/\___|\___\___/ \__,_|\___|_|
|
||
|
||
Copyleft 2022 t.me/vsecoder
|
||
This program is free software; you can redistribute it and/or modify
|
||
|
||
"""
|
||
|
||
# meta developer: @vsecoder_m
|
||
# meta pic: https://avatars.githubusercontent.com/u/128410002
|
||
# meta banner: https://chojuu.vercel.app/api/banner?img=https://avatars.githubusercontent.com/u/128410002&title=HH&description=Hikkahost%20userbot%20manager%20module
|
||
|
||
import os
|
||
import enum
|
||
import aiohttp
|
||
from aiohttp import ClientConnectorError
|
||
from datetime import datetime, timezone
|
||
from typing import Union, Optional, Tuple, List, Dict
|
||
|
||
from .. import loader, utils
|
||
|
||
__version__ = (2, 0, 0)
|
||
|
||
|
||
class Error(enum.Enum):
|
||
critical = 500
|
||
not_found = 404
|
||
unauthorized = 403
|
||
unknown = 0
|
||
|
||
|
||
class Host:
|
||
def __init__(
|
||
self,
|
||
id: int,
|
||
name: str,
|
||
server_id: int,
|
||
port: int,
|
||
start_date: str,
|
||
end_date: str,
|
||
password_hash: str,
|
||
rate: float,
|
||
userbot: str,
|
||
):
|
||
self.id = id
|
||
self.name = name
|
||
self.server_id = server_id
|
||
self.port = port
|
||
self.start_date = datetime.strptime(start_date, "%Y-%m-%dT%H:%M:%S.%f%z")
|
||
self.end_date = datetime.strptime(end_date, "%Y-%m-%dT%H:%M:%S.%f%z")
|
||
self.userbot = userbot
|
||
self.rate = rate
|
||
|
||
|
||
class API:
|
||
async def _request(
|
||
self,
|
||
url: str,
|
||
method: str = "GET",
|
||
params: Optional[Dict] = None,
|
||
data: Optional[Dict] = None,
|
||
headers: Optional[Dict] = None,
|
||
) -> Union[Dict, List[Union[Dict, int]]]:
|
||
async with aiohttp.ClientSession() as session:
|
||
try:
|
||
async with session.request(
|
||
method, url, params=params, data=data, headers=headers
|
||
) as response:
|
||
if response.status == 200:
|
||
answer = await response.json()
|
||
|
||
if "status_code" in answer:
|
||
return [{"detail": answer["detail"]}, answer["status_code"]]
|
||
|
||
return answer if isinstance(answer, dict) else {"data": answer}
|
||
|
||
return [{"detail": await response.text()}, response.status]
|
||
|
||
except ClientConnectorError:
|
||
return [{"detail": "Connection error"}, 500]
|
||
|
||
except Exception as e:
|
||
return [{"detail": f"Unknown error: {e}"}, 500]
|
||
|
||
|
||
class HostAPI(API):
|
||
def __init__(self, url: str, token: str):
|
||
self.auth_header = {"token": token}
|
||
self._url = f"{url}/api/host"
|
||
|
||
async def check_answer(
|
||
self, res: Union[Dict, List]
|
||
) -> Tuple[bool, Union["Error", Dict]]:
|
||
if isinstance(res, list):
|
||
for error in Error:
|
||
if error.value == res[1]:
|
||
return False, error
|
||
|
||
return False, Error.unknown
|
||
|
||
return True, res
|
||
|
||
async def get_host(self, user_id: Union[str, int]) -> Union[Host, "Error"]:
|
||
route = f"{self._url}/{user_id}"
|
||
res = await self._request(route, method="GET", headers=self.auth_header)
|
||
|
||
answer = await self.check_answer(res)
|
||
if not answer[0]:
|
||
return answer[1]
|
||
|
||
host = res["host"]
|
||
return Host(**host)
|
||
|
||
async def action(self, user_id, action):
|
||
route = f"{self._url}/{user_id}"
|
||
payload = {"action": action}
|
||
|
||
await self._request(
|
||
route,
|
||
method="PUT",
|
||
params=payload,
|
||
headers=self.auth_header,
|
||
)
|
||
|
||
async def get_stats(self, user_id) -> Dict:
|
||
return await self._request(
|
||
f"{self._url}/{user_id}/stats", headers=self.auth_header
|
||
)
|
||
|
||
async def get_status(self, user_id) -> Dict:
|
||
return await self._request(
|
||
f"{self._url}/{user_id}/status", headers=self.auth_header
|
||
)
|
||
|
||
async def get_servers(self) -> List:
|
||
return await self._request(
|
||
"https://api.hikka.host/api/server/get/all-open"
|
||
)
|
||
|
||
async def get_logs(
|
||
self, tg_id: Union[str, int], lines: Union[str, int] = "all"
|
||
) -> Dict:
|
||
route = f"{self._url}/{tg_id}/logs/{lines}"
|
||
return await self._request(route, method="GET", headers=self.auth_header)
|
||
|
||
def get_flag(country_code: str = None) -> str:
|
||
return ''.join( chr(ord(c.upper()) + 127397) for c in country_code)
|
||
|
||
@loader.tds
|
||
class HHMod(loader.Module):
|
||
"""@hikkahost userbot manager module"""
|
||
|
||
strings = {
|
||
"name": "HH",
|
||
|
||
"_cfg_doc_hinfo_message": "Custom message text in hinfo. May contain keywords: {id}, {status}, {server}, {days_end}, {cpu_percent}, {ram_usage}, {warns}.",
|
||
"_cfg_doc_hinfo_banner_url": "Link to banner image or None.",
|
||
|
||
"default_info": (
|
||
"<emoji document_id=5413334818047940135>👤</emoji> <b>Info for</b> <code>{id}</code>\n\n"
|
||
"<emoji document_id=5418136591484865679>📶</emoji> <b>Status:</b> {status}\n"
|
||
"<emoji document_id=5415992848753379520>⚙️</emoji> <b>Server:</b> {server}\n"
|
||
"<emoji document_id=5416042764863293485>❤️</emoji> <b>The subscription expires after</b> <code>{days_end} days</code>\n"
|
||
"<emoji document_id=5413394354884596702>💾</emoji> <b>Used now:</b> <code>{cpu_percent}%</code> CPU, <code>{ram_usage}MB</code> RAM\n\n"
|
||
"{warns}"
|
||
),
|
||
"logs": "<emoji document_id=5411608069396254249>📄</emoji> All docker container logs from the userbot\n\n<i>In t.me/hikkahost_bot/hhapp logs more readable</i>",
|
||
"loading_info": "<emoji document_id=5416094132672156295>⌛️</emoji> Loading...",
|
||
"no_apikey": "<emoji document_id=5411402525146370107>🚫</emoji> Not specified API Key, need get token:\n\n1. Go to the @hikkahost_bot\n2. Send /token\n3. Paste token to .config HH",
|
||
"warn_sub_left": "<emoji document_id=5411402525146370107>🚫</emoji> <i>There are less than 5 days left until the end of the subscription</i>\n",
|
||
"statuses": {
|
||
"running": "🟢",
|
||
"stopped": "🔴",
|
||
},
|
||
"server": "{flag} {name}",
|
||
"not_hh": "Your userbot is not running on hikkahost, please, go to @hikkahost_bot",
|
||
"restart": "<emoji document_id=5418136591484865679>🌘</emoji> Your bot user goes to reboot",
|
||
}
|
||
|
||
strings_ru = {
|
||
"name": "HH",
|
||
# "default_info": (
|
||
# "<emoji document_id=5413334818047940135>👤</emoji> <b>Информация о</b> <code>{id}</code>\n\n"
|
||
# "<emoji document_id=5418136591484865679>📶</emoji> <b>Статус:</b> {status}\n"
|
||
# "<emoji document_id=5415992848753379520>⚙️</emoji> <b>Сервер:</b> {server}\n"
|
||
# "<emoji document_id=5416042764863293485>❤️</emoji> <b>Подписка истечёт через</b> <code>{days_end} дней</code>\n"
|
||
# "<emoji document_id=5413394354884596702>💾</emoji> <b>Используется:</b> <code>{cpu_percent}%</code> CPU, <code>{ram_usage}MB</code> RAM\n\n"
|
||
# "{warns}"
|
||
# ),
|
||
"logs": "<emoji document_id=5411608069396254249>📄</emoji> Все логи docker контейнера от hikka\n\n<i>В t.me/hikkahost_bot/hhapp логи более читабельны</i>",
|
||
"loading_info": "<emoji document_id=5416094132672156295>⌛️</emoji> Загрузка...",
|
||
"no_apikey": "<emoji document_id=5411402525146370107>🚫</emoji> Не задан ключ API, нужно получить токен:\n\n1. Зайдите в @hikkahost_bot\n2. Отправьте /token\n3. Запишите токен в .config HH",
|
||
"warn_sub_left": "<emoji document_id=5411402525146370107>🚫</emoji> <i>Менее чем через 5 дней подписка истечёт</i>\n",
|
||
"statuses": {
|
||
"running": "🟢",
|
||
"stopped": "🔴",
|
||
},
|
||
"server": "{flag} {name}",
|
||
"not_hh": "Ваш юзербот запущен не через hikkahost, пожалуйста, зайдите в @hikkahost_bot",
|
||
"restart": "<emoji document_id=5418136591484865679>🌘</emoji> Ваш юзербот отправлен в перезагрузку",
|
||
}
|
||
|
||
def __init__(self):
|
||
self.name = self.strings["name"]
|
||
self.config = loader.ModuleConfig(
|
||
loader.ConfigValue(
|
||
"token",
|
||
None,
|
||
validator=loader.validators.Hidden(),
|
||
),
|
||
loader.ConfigValue(
|
||
"hinfo_message",
|
||
self.strings["default_info"],
|
||
self.strings['_cfg_doc_hinfo_message']
|
||
),
|
||
loader.ConfigValue(
|
||
"hinfo_banner_url",
|
||
"https://github.com/hikkahost/.github/blob/main/banners/main.jpg?raw=true",
|
||
self.strings['_cfg_doc_hinfo_banner_url']
|
||
),
|
||
)
|
||
|
||
async def client_ready(self, client, db):
|
||
self.host = True
|
||
self.url = "https://api.hikka.host"
|
||
|
||
if "HIKKAHOST" not in os.environ:
|
||
self.host = False
|
||
await self.inline.bot.send_message(
|
||
self._tg_id, self.strings("not_hh")
|
||
)
|
||
|
||
self._client = client
|
||
self._db = db
|
||
self.me = await client.get_me()
|
||
self.bot = "@hikkahost_bot"
|
||
|
||
@loader.command(
|
||
en_doc=" - ub status",
|
||
)
|
||
async def hinfocmd(self, message):
|
||
""" - статус юзербота"""
|
||
message = await utils.answer(message, self.strings("loading_info"))
|
||
|
||
if self.config["token"] is None:
|
||
await utils.answer(message, self.strings("no_apikey"))
|
||
return
|
||
|
||
token = self.config["token"]
|
||
user_id = token.split(":")[0]
|
||
api = HostAPI(self.url, token)
|
||
|
||
host = await api.get_host(user_id)
|
||
|
||
if isinstance(host, Error):
|
||
await utils.answer(message, str(host))
|
||
return
|
||
|
||
status = await api.get_status(user_id)
|
||
stats = (await api.get_stats(user_id))["stats"]
|
||
working = True if status["status"] == "running" else False
|
||
|
||
load = {}
|
||
|
||
if working:
|
||
cpu_stats = stats["cpu_stats"]
|
||
cpu_total_usage = cpu_stats['cpu_usage']['total_usage']
|
||
system_cpu_usage = cpu_stats['system_cpu_usage']
|
||
|
||
load = {
|
||
"ram_usage": round(stats["memory_stats"]["usage"] / (1024 * 1024), 2),
|
||
"cpu_percent": round((cpu_total_usage / system_cpu_usage) * 100.0, 2)
|
||
}
|
||
|
||
end_date = host.end_date.replace(tzinfo=timezone.utc)
|
||
warns = ""
|
||
days_end = (end_date - datetime.now(timezone.utc)).days
|
||
if days_end < 5:
|
||
warns += self.strings["warn_sub_left"]
|
||
|
||
servers = (await api.get_servers())["data"]
|
||
servers_dict = {s["id"]: s for s in servers}
|
||
|
||
server = servers_dict.get(host.server_id)
|
||
server = self.strings["server"].format(
|
||
flag=get_flag(server["country_code"]),
|
||
name=server["name"],
|
||
)
|
||
|
||
await utils.answer(
|
||
message,
|
||
self.config["hinfo_message"].format(
|
||
id=user_id,
|
||
warns=warns,
|
||
server=server,
|
||
days_end=days_end,
|
||
status=self.strings["statuses"][status["status"]],
|
||
ram_usage=load.get("ram_usage", "0.00"),
|
||
cpu_percent=load.get("cpu_percent", "0.00")
|
||
),
|
||
file=self.config['hinfo_banner_url']
|
||
)
|
||
|
||
@loader.command(
|
||
en_doc=" - ub logs",
|
||
)
|
||
async def hlogscmd(self, message):
|
||
""" - логи юзербота"""
|
||
if self.config["token"] is None:
|
||
await utils.answer(message, self.strings("no_apikey"))
|
||
return
|
||
|
||
token = self.config["token"]
|
||
user_id = token.split(":")[0]
|
||
api = HostAPI(self.url, token)
|
||
data = await api.get_logs(user_id)
|
||
|
||
files_log = data["logs"].split("\\r\\n")
|
||
|
||
with open("logs.txt", "w") as log_file:
|
||
for log in files_log:
|
||
log_file.write(log + "\n")
|
||
|
||
await utils.answer_file(message, "logs.txt", self.strings("logs"))
|
||
|
||
@loader.command(
|
||
en_doc=" - ub restart",
|
||
)
|
||
async def hrestartcmd(self, message):
|
||
""" - перезагрузить юзербота"""
|
||
await utils.answer(message, self.strings("restart"))
|
||
|
||
if self.config["token"] is None:
|
||
await utils.answer(message, self.strings("no_apikey"))
|
||
return
|
||
|
||
token = self.config["token"]
|
||
user_id = token.split(":")[0]
|
||
api = HostAPI(self.url, token)
|
||
|
||
data = await api.action(user_id, "restart")
|
||
|
||
if isinstance(data, Error):
|
||
await utils.answer(message, str(data))
|
||
return
|