# ©️ Dan Gazizullin, 2021-2023
# This file is a part of Hikka Userbot
# 🌐 https://github.com/hikariatama/Hikka
# You can redistribute it and/or modify it under the terms of the GNU AGPLv3
# 🔑 https://www.gnu.org/licenses/agpl-3.0.html
import asyncio
import base64
import logging
import random
import re
import typing
import requests
import rsa
from hikkatl.tl.types import Message
from hikkatl.utils import resolve_inline_message_id
from .. import loader, translations, utils
from ..types import InlineCall
logger = logging.getLogger(__name__)
pubkey = rsa.PublicKey(
7110455561671499155469672749235101198284219627796886527432331759773809536504953770286294224729310191037878347906574131955439231159825047868272932664151403,
65537,
)
REGEXES = [
re.compile(
r"https:\/\/github\.com\/([^\/]+?)\/([^\/]+?)\/raw\/(?:main|master)\/([^\/]+\.py)"
),
re.compile(
r"https:\/\/raw\.githubusercontent\.com\/([^\/]+?)\/([^\/]+?)\/(?:main|master)\/([^\/]+\.py)"
),
]
@loader.tds
class UnitHeta(loader.Module):
"""Gives @hikkamods_bot a right to download modules from official modules aggregator and autoupdate them"""
e = "❌"
strings = {
"name": "UnitHeta",
"no_query": f"{e} You must specify query",
"no_results": f"{e} No results",
"api_error": f"{e} API is having issues",
"result": (
"🥰 Results for {query}:\n\n🧳 {name}"
" by {dev}\n👨🏫 {cls_doc}\n\n📚"
" Commands:\n{commands}\n\n🔗 Install: {prefix}dlm"
" {link}"
),
"install": "🪄 Install",
"loaded": "✅ Sucessfully installed",
"not_loaded": "❌ Installation failed",
"language": "en",
}
strings_ru = {
"no_query": f"{e} Вы должны указать запрос",
"no_results": f"{e} Нет результатов",
"api_error": f"{e} С API случилась беда",
"result": (
"🥰 Результаты для {query}:\n\n🧳"
" {name} от {dev}\n👨🏫"
" {cls_doc}\n\n📚 Команды:\n{commands}\n\n🔗 Установить:"
" {prefix}dlm {link}"
),
"install": "🪄 Установить",
"loaded": "✅ Успешно установлено",
"not_loaded": "❌ Установка не удалась",
"language": "ru",
}
strings_es = {
"no_query": f"{e} Debes especificar una consulta",
"no_results": f"{e} No hay resultados",
"api_error": f"{e} Hay problemas con la API",
"result": (
"🥰 Resultados para {query}:\n\n🧳"
" {name} por {dev}\n👨🏫"
" {cls_doc}\n\n📚 Comandos:\n{commands}\n\n🔗 Instalar:"
" {prefix}dlm {link}"
),
"install": "🪄 Instalar",
"loaded": "✅ Instalado con éxito",
"not_loaded": "❌ La instalación falló",
"language": "es",
}
strings_de = {
"no_query": f"{e} Du musst eine Abfrage angeben",
"no_results": f"{e} Keine Ergebnisse",
"api_error": f"{e} Es gibt Probleme mit der API",
"result": (
"🥰 Ergebnisse für {query}:\n\n🧳"
" {name} von {dev}\n👨🏫"
" {cls_doc}\n\n📚 Befehle:\n{commands}\n\n🔗"
" Installieren: {prefix}dlm {link}"
),
"install": "🪄 Installieren",
"loaded": "✅ Erfolgreich installiert",
"not_loaded": "❌ Die Installation ist fehlgeschlagen",
"language": "de",
}
strings_fr = {
"no_query": f"{e} Vous devez spécifier une requête",
"no_results": f"{e} Aucun résultat",
"api_error": f"{e} Quelque chose s'est mal passé avec l'API",
"result": (
"🥰 Résultats pour {query}:\n\n🧳"
" {name} par {dev}\n👨🏫"
" {cls_doc}\n\n📚 Commandes:\n{commands}\n\n🔗"
" Installer: {prefix}dlm {link}"
),
"install": "🪄 Installer",
"loaded": "✅ Installation réussie",
"not_loaded": "❌ Installation échouée",
"language": "fr",
}
strings_uz = {
"no_query": f"{e} Siz so'rovni belgilamadingiz",
"no_results": f"{e} Natija topilmadi",
"api_error": f"{e} API bilan muammo yuz berdi",
"result": (
"🥰 Ushbu {query}uchun natijalar:\n\n🧳"
" {name} to'g'risida {dev}\n👨🏫"
" {cls_doc}\n\n📚 Komandalar:\n{commands}\n\n🔗"
" O'rnatish: {prefix}dlm {link}"
),
"install": "🪄 O'rnatish",
"loaded": "✅ Muvaffaqiyatli o'rnatildi",
"not_loaded": "❌ O'rnatish muvaffaqiyatsiz bo'ldi",
"language": "uz",
}
strings_tr = {
"no_query": f"{e} Bir sorgu belirtmelisiniz",
"no_results": f"{e} Sonuç yok",
"api_error": f"{e} API ile ilgili bir sorun oluştu",
"result": (
"🥰 Sonuçlar için {query}:\n\n🧳"
" {name} geliştirici {dev}\n👨🏫"
" {cls_doc}\n\n📚 Komutlar:\n{commands}\n\n🔗 Yükle:"
" {prefix}dlm {link}"
),
"install": "🪄 Yükle",
"loaded": "✅ Başarıyla yüklendi",
"not_loaded": "❌ Yükleme başarısız oldu",
"language": "tr",
}
strings_it = {
"no_query": f"{e} Devi specificare una query",
"no_results": f"{e} Nessun risultato",
"api_error": f"{e} Si è verificato un'errore con l'API",
"result": (
"🥰 Risultati per {query}:\n\n🧳"
" {name} da {dev}\n👨🏫"
" {cls_doc}\n\n📚 Comandi:\n{commands}\n\n🔗 Installare:"
" {prefix}dlm {link}"
),
"install": "🪄 Installare",
"loaded": "✅ Installazione riuscita",
"not_loaded": "❌ Installazione non riuscita",
"language": "it",
}
strings_kk = {
"no_query": f"{e} Сұранымды көрсетуіңіз керек",
"no_results": f"{e} Нәтижелер жоқ",
"api_error": f"{e} API-ға қате кетті",
"result": (
"🥰 Сұранымдың нәтижелері {query}:\n\n🧳"
" {name} төлесін {dev}\n👨🏫"
" {cls_doc}\n\n📚 Командалар:\n{commands}\n\n🔗 Орнату:"
" {prefix}dlm {link}"
),
"install": "🪄 Орнату",
"loaded": "✅ Орнату сәтті аяқталды",
"not_loaded": "❌ Орнату сәтсіз аяқталды",
"language": "kk",
}
strings_tt = {
"no_query": f"{e} Зиндергә мәгълүматләр кертмәгәнсез",
"no_results": f"{e} Нәтиҗәләр табылмады",
"api_error": f"{e} API-сәхифәсе белән хата",
"result": (
"🥰 Зиндергә нәтиҗәләр {query}:\n\n🧳"
" {name} төзәтелгән {dev}\n👨🏫"
" {cls_doc}\n\n📚 Командалар:\n{commands}\n\n🔗"
" Установить: {prefix}dlm {link}"
),
"install": "🪄 Установить",
"loaded": "✅ Установка уңышлы тамамланды",
"not_loaded": "❌ Установка үтәлмәде",
"language": "tt",
}
def __init__(self):
self.config = loader.ModuleConfig(
loader.ConfigValue(
"autoupdate",
False,
(
"Do you want to autoupdate modules? (Join @heta_updates in order"
" for this option to take effect) ⚠️ Use at your own risk!"
),
validator=loader.validators.Boolean(),
),
loader.ConfigValue(
"translate",
True,
(
"Do you want to translate module descriptions and command docs to"
" the language, specified in Hikka? (This option is experimental,"
" and might not work properly)"
),
validator=loader.validators.Boolean(),
),
)
async def client_ready(self):
if self.config["autoupdate"]:
await self.request_join(
"@heta_updates",
"This channel is the source of update notifications",
)
if self.get("nomute"):
return
await utils.dnd(self._client, "@hikkamods_bot", archive=False)
self.set("nomute", True)
async def _install(self, call: InlineCall, url: str, text: str):
await call.edit(
text,
reply_markup={
"text": (
self.strings("loaded")
if await self._load_module(url)
else self.strings("not_loaded")
),
"data": "empty",
},
)
@loader.command(
ru_doc="<запрос> - Ищет модули в репозитории Heta",
de_doc=" - Sucht Module im Heta-Repository",
uz_doc=" - Heta ombori uchun modullarni qidiradi",
tr_doc=" - Heta deposunda modülleri arar",
it_doc=" - Cerca moduli nel repository Heta",
fr_doc=" - Recherche des modules dans le référentiel Heta",
kk_doc="<сұраным> - Heta орталығында модульларды іздейді",
tt_doc="<зиндергә> - Heta депозиториясендә модульләрне таба",
es_doc=" - Busca módulos en el repositorio Heta",
)
async def heta(self, message: Message):
""" - Searches Heta repository for modules"""
if not (query := utils.get_args_raw(message)):
await utils.answer(message, self.strings("no_query"))
return
if not (
response := await utils.run_sync(
requests.get,
"https://heta.hikariatama.ru/search",
params={"q": query, "limit": 1},
)
):
await utils.answer(message, self.strings("no_results"))
return
try:
response.raise_for_status()
except requests.exceptions.HTTPError:
await utils.answer(message, self.strings("api_error"))
return
if not (result := response.json()):
await utils.answer(message, self.strings("no_results"))
return
result = result[0]
commands = "\n".join(
[
f"▫️ {self.get_prefix()}{cmd}:"
f" {utils.escape_html(cmd_doc)}"
for cmd, cmd_doc in result["module"]["commands"].items()
]
)
kwargs = {
"name": utils.escape_html(result["module"]["name"]),
"dev": utils.escape_html(result["module"]["dev"]),
"commands": commands,
"cls_doc": utils.escape_html(result["module"]["cls_doc"]),
"link": result["module"]["link"],
"query": utils.escape_html(query),
"prefix": self.get_prefix(),
}
strings = (
self.strings._base_strings["result"]
if self.config["translate"]
else self.strings("result")
)
text = strings.format(**kwargs)
if len(text) > 2048:
kwargs["commands"] = "..."
text = strings.format(**kwargs)
mark = lambda text: {
"text": self.strings("install"),
"callback": self._install,
"args": (result["module"]["link"], text),
}
form = await self.inline.form(
message=message,
text=text,
**(
{"photo": result["module"]["banner"]}
if result["module"].get("banner")
else {}
),
reply_markup=mark(text),
)
if not self.config["translate"]:
return
message_id, peer, _, _ = resolve_inline_message_id(form.inline_message_id)
try:
text = await self._client.translate(
peer,
message_id,
self.strings("language"),
)
await form.edit(text=text, reply_markup=mark(text))
except Exception:
text = self.strings("result").format(**kwargs)
await form.edit(text=text, reply_markup=mark(text))
async def _load_module(
self,
url: str,
message: typing.Optional[Message] = None,
) -> bool:
loader_m = self.lookup("loader")
await loader_m.download_and_install(url, None)
if getattr(loader_m, "fully_loaded", False):
loader_m.update_modules_in_db()
loaded = any(
link == url for link in loader_m.get("loaded_modules", {}).values()
)
if message:
if loaded:
await self._client.inline_query(
"@hikkamods_bot",
f"#confirm_load {message.raw_text.splitlines()[2].strip()}",
)
else:
await self._client.inline_query(
"@hikkamods_bot",
f"#confirm_fload {message.raw_text.splitlines()[2].strip()}",
)
return loaded
@loader.watcher("in", "only_messages", chat_id=1688624566, contains="Heta url: ")
async def update_watcher(self, message: Message):
url = message.raw_text.split("Heta url: ")[1].strip()
dev, repo, mod = url.lower().split("hikariatama.ru/")[1].split("/")
if dev == "hikariatama" and repo == "ftg":
urls = [f"https://mods.hikariatama.ru/{mod}", url]
if any(
getattr(module, "__origin__", None).lower().strip("/") in urls
for module in self.allmodules.modules
):
await self._load_module(urls[0])
await asyncio.sleep(random.randint(1, 10))
await self._client.inline_query(
"@hikkamods_bot",
f"#confirm_update_noheta {url.split('hikariatama.ru/')[1]}",
)
return
if any(
getattr(module, "__origin__", "").lower().strip("/")
== url.lower().strip("/")
for module in self.allmodules.modules
):
await self._load_module(url)
await asyncio.sleep(random.randint(1, 10))
await self._client.inline_query(
"@hikkamods_bot",
f"#confirm_update {url.split('hikariatama.ru/')[1]}",
)
return
for module in self.allmodules.modules:
link = getattr(module, "__origin__", "").lower().strip("/")
for regex in REGEXES:
if regex.search(link):
dev, repo, mod = regex.search(link).groups()
if dev == dev and repo == repo and mod == mod:
await self._load_module(link)
await asyncio.sleep(random.randint(1, 10))
await self._client.inline_query(
"@hikkamods_bot",
f"#confirm_update_noheta {url.split('hikariatama.ru/')[1]}",
)
return
@loader.watcher("in", "only_messages", from_user=5519484330)
async def watcher(self, message: Message):
if message.raw_text.startswith("#install"):
await message.delete()
fileref = (
message.raw_text.split("#install:")[1].strip().splitlines()[0].strip()
)
sig = base64.b64decode(message.raw_text.splitlines()[1].strip().encode())
try:
rsa.verify(
rsa.compute_hash(fileref.encode("utf-8"), "SHA-1"), sig, pubkey
)
except rsa.pkcs1.VerificationError:
logger.error("Got message with non-verified signature %s", fileref)
return
await self._load_module(f"https://heta.hikariatama.ru/{fileref}", message)
elif message.raw_text.startswith("#setlang"):
lang = message.raw_text.split()[1]
self._db.set(translations.__name__, "lang", lang)
await self.allmodules.reload_translations()
await self._client.inline_query("@hikkamods_bot", "#confirm_setlang")