"""
_
__ _____ ___ ___ ___ __| | ___ _ __
\ \ / / __|/ _ \/ __/ _ \ / _` |/ _ \ '__|
\ 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": (
"👤 Info for {id}\n\n"
"📶 Status: {status}\n"
"⚙️ Server: {server}\n"
"❤️ The subscription expires after {days_end} days\n"
"💾 Used now: {cpu_percent}% CPU, {ram_usage}MB RAM\n\n"
"{warns}"
),
"logs": "📄 All docker container logs from the userbot\n\nIn t.me/hikkahost_bot/hhapp logs more readable",
"loading_info": "⌛️ Loading...",
"no_apikey": "🚫 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": "🚫 There are less than 5 days left until the end of the subscription\n",
"statuses": {
"running": "🟢",
"stopped": "🔴",
},
"server": "{flag} {name}",
"not_hh": "Your userbot is not running on hikkahost, please, go to @hikkahost_bot",
"restart": "🌘 Your bot user goes to reboot",
}
strings_ru = {
"name": "HH",
# "default_info": (
# "👤 Информация о {id}\n\n"
# "📶 Статус: {status}\n"
# "⚙️ Сервер: {server}\n"
# "❤️ Подписка истечёт через {days_end} дней\n"
# "💾 Используется: {cpu_percent}% CPU, {ram_usage}MB RAM\n\n"
# "{warns}"
# ),
"logs": "📄 Все логи docker контейнера от hikka\n\nВ t.me/hikkahost_bot/hhapp логи более читабельны",
"loading_info": "⌛️ Загрузка...",
"no_apikey": "🚫 Не задан ключ API, нужно получить токен:\n\n1. Зайдите в @hikkahost_bot\n2. Отправьте /token\n3. Запишите токен в .config HH",
"warn_sub_left": "🚫 Менее чем через 5 дней подписка истечёт\n",
"statuses": {
"running": "🟢",
"stopped": "🔴",
},
"server": "{flag} {name}",
"not_hh": "Ваш юзербот запущен не через hikkahost, пожалуйста, зайдите в @hikkahost_bot",
"restart": "🌘 Ваш юзербот отправлен в перезагрузку",
}
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