__version__ = (0, 0, 36)
# ▄▀█ █▄ █ █▀█ █▄ █ █▀█ ▀▀█ █▀█ █ █ █▀
# █▀█ █ ▀█ █▄█ █ ▀█ ▀▀█ █ ▀▀█ ▀▀█ ▄█
#
# © Copyright 2024
#
# developed by @anon97945
#
# https://t.me/apodiktum_modules
# https://github.com/anon97945
#
# 🔒 Licensed under the GNU GPLv3
# 🌐 https://www.gnu.org/licenses/gpl-3.0.html
# meta developer: @apodiktum_modules
# meta banner: https://t.me/apodiktum_dumpster/11
# meta pic: https://t.me/apodiktum_dumpster/13
# scope: hikka_only
# scope: hikka_min 1.3.3
import asyncio
import logging
import math
import os
import heroku3
import requests
from telethon.tl.types import Message
from .. import heroku, loader, main, utils
logger = logging.getLogger(__name__)
@loader.tds
class ApodiktumHerokuManagerMod(loader.Module):
"""
Show Remaining Dyno Usage And Manage The Settings Of Your 🦸🏼♂️ Hero!ku Hikka Instance.
"""
strings = {
"name": "Apo-HerokuManager",
"developer": "@anon97945",
"args_error": "[🦸🏼♂️ Hero!ku] Too many args are given.",
"dyno_usage": (
"Dyno Usage:\n"
"\nDyno usage for Hikka Userbot:\n"
" • {}h {}m | [{}%]\n"
"Dyno hours quota remaining this month:\n"
" • {}h {}m | [{}%]"
),
"get_usage": "[🦸🏼♂️ Hero!ku] Getting Dyno usage...",
"get_var": "[🦸🏼♂️ Hero!ku] Getting variable...",
"no_args": "[🦸🏼♂️ Hero!ku] No args are given...",
"no_force": (
"[🦸🏼♂️ Hero!ku] You must use '--force' but this will leak"
" credentials!"
),
"restarted": "[🦸🏼♂️ Hero!ku] Restart finished.",
"set_var": "[🦸🏼♂️ Hero!ku] Setting variable...",
"usage_error": "Error: An error occured.\n{}",
"var_added": (
"[🦸🏼♂️ Hero!ku] Variable successfully added:\n"
"{} = {}\n\n"
"The Heroku Dyno will now be restarted."
),
"var_changed": (
"[🦸🏼♂️ Hero!ku] Variable successfully changed to:\n"
"{} = {}\n\n"
"The Heroku Dyno will now be restarted."
),
"var_deleted": (
"[🦸🏼♂️ Hero!ku] Variable successfully deleted:\n"
"{}\n\n"
"The Heroku Dyno will now be restarted."
),
"var_not_exists": (
"[🦸🏼♂️ Hero!ku] Variable does not exist:\n{}"
),
"var_settings": (
"[🦸🏼♂️ Hero!ku] Current Config:\n{} = {}"
),
"wrong_platform": (
"[🦸🏼♂️ Hero!ku] This module only works on Heroku. {} is not supported."
),
"_cfg_cst_auto_migrate": "Wheather to auto migrate defined changes on startup.",
}
strings_en = {}
strings_de = {}
strings_ru = {
"_cls_doc": (
"Показать оставшееся использование Dyno и управлять настройками"
" вашего экземпляра 🦸🏼♂️ Hero!ku Hikka."
),
"_cmd_doc_herodel": (
" \n "
" Удалить переменную настроек"
" Heroku.\n "
" - Example: .herodel "
),
"_cmd_doc_heroget": (
" \n "
" Получить переменную настроек"
" Heroku.\n "
" - Example: .heroget "
),
"_cmd_doc_herogetall": (
" \n "
" Получить все переменные настроек"
" Heroku. Это может привести к утечке API!\n "
" - Example:"
" .herogetall --force"
),
"_cmd_doc_heroset": (
" \n "
" Установить переменную настроек"
" Heroku.\n "
" - Example: .heroset "
),
"_cmd_doc_herousage": (
" \n "
" Получить использование Heroku"
" Dyno."
),
"args_error": "[🦸🏼♂️ Hero!ku] Задано слишком много аргументов.",
"dyno_usage": (
"Dyno Usage:\n"
"\nИспользование Дино для Hikka Userbot:\n"
" • {}h {}m | [{}%]\n"
"Осталось часов Дино по квоте в месяц:\n"
" • {}h {}m | [{}%]"
),
"get_usage": "[🦸🏼♂️ Hero!ku] Получение использования Dyno...",
"get_var": "[🦸🏼♂️ Hero!ku] Получение переменной...",
"no_args": "[🦸🏼♂️ Hero!ku] Аргументы не указаны...",
"no_force": (
"[🦸🏼♂️ Hero!ku] Вы должны использовать '--force', но это"
" приведет к утечке учетных данных!"
),
"restarted": "[🦸🏼♂️ Hero!ku] Перезагрузка завершена.",
"set_var": "[🦸🏼♂️ Hero!ku] Настройка переменной...",
"usage_error": "Error: Произошла ошибка.\n{}",
"var_added": (
"[🦸🏼♂️ Hero!ku] Переменная успешно добавлена:\n"
"{} = {}\n\n"
"Теперь Heroku Dyno будет перезапущен."
),
"var_changed": (
"[🦸🏼♂️ Hero!ku] Переменная успешно изменена на:\n"
"{} = {}\n\n"
"Теперь Heroku Dyno будет перезапущен."
),
"var_deleted": (
"[🦸🏼♂️ Hero!ku] Переменная успешно удалена:\n"
"{}\n\n"
"Теперь Heroku Dyno будет перезапущен."
),
"var_not_exists": (
"[🦸🏼♂️ Hero!ku] Переменная не существует:\n{}"
),
"var_settings": (
"[🦸🏼♂️ Hero!ku] Текущая конфигурацияТекущая конфигурация:\n"
"{} = {}"
),
"wrong_platform": (
"[🦸🏼♂️ Hero!ku] Этот модуль работает только на Heroku. {} не"
" поддерживается."
),
}
all_strings = {
"strings": strings,
"strings_en": strings,
"strings_de": strings_de,
"strings_ru": strings_ru,
}
def __init__(self):
self._ratelimit = []
self.config = loader.ModuleConfig(
loader.ConfigValue(
"auto_migrate",
True,
doc=lambda: self.strings("_cfg_cst_auto_migrate"),
validator=loader.validators.Boolean(),
), # for MigratorClass
)
async def client_ready(self):
platform = utils.get_named_platform()
if "DYNO" not in os.environ:
raise loader.LoadError(self.strings("wrong_platform").format(platform))
self.apo_lib = await self.import_lib(
"https://raw.githubusercontent.com/anon97945/hikka-libs/master/apodiktum_library.py",
suspend_on_error=True,
)
self._init_heroku_vars()
def _init_heroku_vars(self):
self._happ, self._hconfig = heroku.get_app(api_token=main.hikka.api_token)
self._heroku_api = "https://api.heroku.com"
self._heroku_app_name = self._happ.name
self._heroku_app_id = self._happ.id
self._heroku_api_key = os.environ["heroku_api_token"]
self._heroku = heroku3.from_key(self._heroku_api_key)
self._heroku_app = self._heroku.app(self._heroku_app_name)
self._platform = utils.get_named_platform()
self._herokuid = self._heroku.account().id
async def herousagecmd(self, message: Message):
"""Get Heroku Dyno Usage."""
msg = await utils.answer(
message,
self.apo_lib.utils.get_str("get_usage", self.all_strings, message),
)
useragent = (
"Mozilla/5.0 (Linux; Android 10; SM-G975F)"
"AppleWebKit/537.36 (KHTML, like Gecko)"
"Chrome/80.0.3987.149 Mobile Safari/537.36"
)
headers = {
"User-Agent": useragent,
"Authorization": f"Bearer {self._heroku_api_key}",
"Accept": "application/vnd.heroku+json; version=3.account-quotas",
}
path = f"/accounts/{self._herokuid}/actions/get-quota"
r = requests.get(self._heroku_api + path, headers=headers)
if r.status_code != 200:
return await utils.answer(
message,
self.apo_lib.utils.get_str(
"usage_error", self.all_strings, message
).format(r.reason),
)
result = r.json()
quota = result["account_quota"]
quota_used = result["quota_used"]
# Used
remaining_quota = quota - quota_used
app_quota_used = 0
percentage = math.floor(remaining_quota / quota * 100)
minutes_remaining = remaining_quota / 60
hours = math.floor(minutes_remaining / 60)
minutes = math.floor(minutes_remaining % 60)
# Current
App = result["apps"]
try:
for app in App:
if app["app_uuid"] == self._heroku_app_id:
app_quota_used = app["quota_used"]
break
except IndexError:
AppQuotaUsed = 0
AppPercentage = 0
else:
AppQuotaUsed = app_quota_used / 60
AppPercentage = math.floor(app_quota_used * 100 / quota)
AppHours = math.floor(AppQuotaUsed / 60)
AppMinutes = math.floor(AppQuotaUsed % 60)
# AppName = self._heroku_app_name
await asyncio.sleep(1.5)
return await utils.answer(
msg,
self.apo_lib.utils.get_str("dyno_usage", self.all_strings, message).format(
AppHours, AppMinutes, AppPercentage, hours, minutes, percentage
),
)
@loader.owner
async def herosetcmd(self, message: Message):
"""
Set Heroku Settings Variable.
- Example: .heroset
"""
args = utils.get_args_raw(message)
if args := str(args).split():
heroku_var = self._heroku_app.config()
msg = await utils.answer(
message,
self.apo_lib.utils.get_str("set_var", self.all_strings, message),
)
await asyncio.sleep(1.5)
if args[0] in heroku_var:
msg = await utils.answer(
msg,
self.apo_lib.utils.get_str(
"var_changed", self.all_strings, message
).format(args[0], " ".join(args[1:])),
)
else:
msg = await utils.answer(
msg,
self.apo_lib.utils.get_str(
"var_added", self.all_strings, message
).format(args[0], " ".join(args[1:])),
)
heroku_var[args[0]] = " ".join(args[1:])
return
await utils.answer(
message,
self.apo_lib.utils.get_str("no_var", self.all_strings, message),
)
@loader.owner
async def herogetcmd(self, message: Message):
"""
Get Heroku Settings Variable.
- Example: .heroget
"""
args = utils.get_args_raw(message)
if args := str(args).split():
if len(args) > 1:
return await utils.answer(
message,
self.apo_lib.utils.get_str("args_error", self.all_strings, message),
)
heroku_var = self._heroku_app.config()
msg = await utils.answer(
message,
self.apo_lib.utils.get_str("get_var", self.all_strings, message),
)
await asyncio.sleep(1.5)
if args[0] in heroku_var:
return await utils.answer(
msg,
self.apo_lib.utils.get_str(
"var_settings", self.all_strings, message
).format(args[0], heroku_var[args[0]]),
)
return await utils.answer(
msg,
self.apo_lib.utils.get_str(
"var_not_exists", self.all_strings, message
).format(args[0]),
)
return await utils.answer(
message,
self.apo_lib.utils.get_str("no_var", self.all_strings, message),
)
@loader.owner
async def herogetallcmd(self, message: Message):
"""
Get All Heroku Settings Variable. This may leak API!
- Example: .herogetall --force
"""
args = utils.get_args_raw(message)
args = str(args).split()
if args and args[0] == "--force":
if len(args) > 1:
return await utils.answer(
message,
self.apo_lib.utils.get_str("args_error", self.all_strings, message),
)
heroku_var = self._heroku_app.config()
msg = await utils.answer(
message,
self.apo_lib.utils.get_str("get_var", self.all_strings, message),
)
await asyncio.sleep(1.5)
cmpl_cnfg = ""
for x in heroku_var.to_dict():
cmpl_cnfg = (
f"{cmpl_cnfg}{x}"
+ ":\n"
+ heroku_var.to_dict()[x]
+ "\n\n"
)
return await utils.answer(msg, cmpl_cnfg)
return await utils.answer(
message,
self.apo_lib.utils.get_str("no_force", self.all_strings, message),
)
@loader.owner
async def herodelcmd(self, message: Message):
"""
Delete Heroku Settings Variable.
- Example: .herodel
"""
args = utils.get_args_raw(message)
if not (args := str(args).split()):
await utils.answer(
message,
self.apo_lib.utils.get_str("no_var", self.all_strings, message),
)
return
if len(args) > 1:
return await utils.answer(
message,
self.apo_lib.utils.get_str("args_error", self.all_strings, message),
)
heroku_var = self._heroku_app.config()
msg = await utils.answer(
message,
self.apo_lib.utils.get_str("get_var", self.all_strings, message),
)
await asyncio.sleep(1.5)
if args[0] in heroku_var:
msg = await utils.answer(
msg,
self.apo_lib.utils.get_str(
"var_deleted", self.all_strings, message
).format(args[0]),
)
del heroku_var[args[0]]
return
await utils.answer(
message,
self.apo_lib.utils.get_str(
"var_not_exists", self.all_strings, message
).format(args[0]),
)