# █ █ ▀ █▄▀ ▄▀█ █▀█ ▀
# █▀█ █ █ █ █▀█ █▀▄ █
# © Copyright 2022
# https://t.me/hikariatama
#
# 🔒 Licensed under the GNU AGPLv3
# 🌐 https://www.gnu.org/licenses/agpl-3.0.html
import asyncio
import contextlib
import logging
import os
import signal
import subprocess
import sys
import time
import typing
import git
from git import GitCommandError, Repo
from telethon.extensions.html import CUSTOM_EMOJIS
from telethon.tl.functions.messages import (
GetDialogFiltersRequest,
UpdateDialogFilterRequest,
)
from telethon.tl.types import DialogFilter, Message
from .. import loader, main, utils, version
from .._internal import get_startup_callback
from ..inline.types import InlineCall
logger = logging.getLogger(__name__)
@loader.tds
class UpdaterMod(loader.Module):
"""Updates itself"""
strings = {
"name": "Updater",
"source": (
"📖 Read the source code"
" from here"
),
"restarting_caption": (
"🕗 Your {} is"
" restarting..."
),
"downloading": (
"🕗 Downloading"
" updates..."
),
"installing": (
"🕗 Installing"
" updates..."
),
"success": (
"⏱ Restart successful!"
" {}\nBut still loading modules...\nRestart took {}s"
),
"origin_cfg_doc": "Git origin URL, for where to update from",
"btn_restart": "🔄 Restart",
"btn_update": "🧭 Update",
"restart_confirm": "❓ Are you sure you want to restart?",
"secure_boot_confirm": (
"❓ Are you sure you want to restart in secure boot mode?"
),
"update_confirm": (
"❓ Are you sure you"
" want to update?\n\n{} ⤑ {}'
),
"no_update": "🚸 You are on the latest version, pull updates anyway?",
"cancel": "🚫 Cancel",
"lavhost_restart": (
"✌️ Your {} is"
" restarting..."
),
"lavhost_update": (
"✌️ Your {} is"
" updating..."
),
"full_success": (
"👍 Userbot is fully"
" loaded! {}\nFull restart took {}s"
),
"secure_boot_complete": (
"🔒 Secure boot completed! {}\nRestart took {}s"
),
}
strings_ru = {
"source": (
"📖 Исходный код можно"
" прочитать здесь"
),
"restarting_caption": (
"🕗 Твоя {}"
" перезагружается..."
),
"downloading": (
"🕗 Скачивание"
" обновлений..."
),
"installing": (
"🕗 Установка"
" обновлений..."
),
"success": (
"⏱ Перезагрузка"
" успешна! {}\nНо модули еще загружаются...\nПерезагрузка"
" заняла {} сек"
),
"full_success": (
"👍 Юзербот полностью"
" загружен! {}\nПолная перезагрузка заняла {} сек"
),
"secure_boot_complete": (
"🔒 Безопасная загрузка завершена! {}\nПерезагрузка заняла {}"
" сек"
),
"origin_cfg_doc": "Ссылка, из которой будут загружаться обновления",
"btn_restart": "🔄 Перезагрузиться",
"btn_update": "🧭 Обновиться",
"restart_confirm": "❓ Ты уверен, что хочешь перезагрузиться?",
"secure_boot_confirm": (
"❓ Ты уверен, что"
" хочешь перезагрузиться в режиме безопасной загрузки?"
),
"update_confirm": (
"❓ Ты уверен, что"
" хочешь обновиться??\n\n{} ⤑ {}'
),
"no_update": "🚸 У тебя последняя версия. Обновиться принудительно?",
"cancel": "🚫 Отмена",
"_cls_doc": "Обновляет юзербот",
"lavhost_restart": (
"✌️ Твой {}"
" перезагружается..."
),
"lavhost_update": (
"✌️ Твой {}"
" обновляется..."
),
}
strings_de = {
"source": (
"📖 Der Quellcode kann"
" hier gelesen werden"
),
"restarting_caption": (
"🕗 Dein {}"
" wird neugestartet..."
),
"downloading": (
"🕗 Updates"
" werden heruntergeladen..."
),
"installing": (
"🕗 Updates"
" werden installiert..."
),
"success": (
"⏱ Neustart erfolgreich!"
" {}\nAber Module werden noch geladen...\nNeustart dauerte {}"
" Sekunden"
),
"full_success": (
"👍 Dein Userbot ist"
" vollständig geladen! {}\nVollständiger Neustart dauerte {}"
" Sekunden"
),
"secure_boot_complete": (
"🔒 Sicherer Bootvorgang abgeschlossen! {}\nNeustart dauerte"
" {} Sekunden"
),
"origin_cfg_doc": "Link, von dem Updates heruntergeladen werden",
"btn_restart": "🔄 Neustart",
"btn_update": "🧭 Update",
"restart_confirm": "❓ Bist du sicher, dass du neustarten willst?",
"secure_boot_confirm": (
"❓ Bist du sicher, dass du in den sicheren Modus neustarten willst?"
),
"update_confirm": (
"❓ Bist du sicher, dass"
" du updaten willst??\n\n{} ⤑ {}'
),
"no_update": (
"🚸 Du hast die neueste Version. Willst du trotzdem updaten?"
),
"cancel": "🚫 Abbrechen",
"_cls_doc": "Aktualisiert den Userbot",
"lavhost_restart": (
"✌️ Dein {}"
" wird neugestartet..."
),
"lavhost_update": (
"✌️ Dein {}"
" wird aktualisiert..."
),
}
strings_tr = {
"source": (
"📖 Manba kodini shu yerdan oʻqing"
),
"restarting": (
"🕗 {}"
" yeniden başlatılıyor..."
),
"restarting_caption": (
"🕗 {}"
" yeniden başlatılıyor..."
),
"downloading": (
"🕗 Güncelleme"
" indiriliyor..."
),
"installing": (
"🕗 Güncelleme"
" yükleniyor..."
),
"success": (
"⏱ Yeniden başlatma"
" başarılı! {}\nModüller yükleniyor...\nYeniden başlatma {}"
" saniye sürdü"
),
"full_success": (
"👍 Botunuz tamamen"
" yüklendi! {}\nToplam yeniden başlatma {} saniye sürdü"
),
"secure_boot_complete": (
"🔒 Güvenli mod başarıyla tamamlandı! {}\nYeniden başlatma {}"
" saniye sürdü"
),
"origin_cfg_doc": "dan güncelleme indirilecek",
"btn_restart": "🔄 Yeniden başlat",
"btn_update": "🧭 Güncelle",
"restart_confirm": "❓ Gerçekten yeniden başlatmak istiyor musunuz?",
"secure_boot_confirm": (
"❓ Gerçekten güvenli modda yeniden başlatmak istiyor musunuz?"
),
"update_confirm": (
"❓ Gerçekten güncellemek istiyor musunuz??\n\n{} ⤑ {}'
),
"no_update": "🚸 Zaten son sürümünüz. Güncelleme yapmak ister misiniz?",
"cancel": "🚫 İptal",
"_cls_doc": "Kullanıcı botunu günceller",
"lavhost_restart": (
"✌️ {}"
" yeniden başlatılıyor..."
),
"lavhost_update": (
"✌️ {}"
" güncelleniyor..."
),
}
strings_uz = {
"restarting": (
"🕗 {}"
" qayta ishga tushirilmoqda..."
),
"restarting_caption": (
"🕗 {}"
" qayta ishga tushirilmoqda..."
),
"downloading": (
"🕗 Yangilanish"
" yuklanmoqda..."
),
"installing": (
"🕗 Yangilanish"
" o'rnatilmoqda..."
),
"success": (
"⏱ Qayta ishga tushirish"
" muvaffaqiyatli yakunlandi! {}\nModullar"
" yuklanmoqda...\nQayta ishga tushirish {} soniya davom etdi"
),
"full_success": (
"👍 Sizning botingiz"
" to'liq yuklandi! {}\nJami qayta ishga tushirish {} soniya davom"
" etdi"
),
"secure_boot_complete": (
"🔒 Xavfsiz rejim muvaffaqiyatli yakunlandi! {}\nQayta ishga"
" tushirish {} soniya davom etdi"
),
"origin_cfg_doc": "dan yangilanish yuklanadi",
"btn_restart": "🔄 Qayta ishga tushirish",
"btn_update": "🧭 Yangilash",
"restart_confirm": "❓ Haqiqatan ham qayta ishga tushirmoqchimisiz?",
"secure_boot_confirm": (
"❓ Haqiqatan ham xavfsiz rejimda qayta ishga tushirmoqchimisiz?"
),
"update_confirm": (
"❓ Haqiqatan ham yangilamoqchimisiz??\n\n{} ⤑ {}'
),
"no_update": (
"🚸 Siz allaqachon eng so'nggi versiyasiz. Yangilamoqchimisiz?"
),
"cancel": "🚫 Bekor qilish",
"_cls_doc": "Foydalanuvchi botini yangilaydi",
"lavhost_restart": (
"✌️ {}"
" qayta ishga tushirilmoqda..."
),
"lavhost_update": (
"✌️ {}"
" yangilanmoqda..."
),
}
strings_es = {
"restarting": (
"🕗 {} Reiniciando..."
),
"restarting_caption": (
"🕗 {} Reiniciando..."
),
"downloading": (
"🕗 Descargando la"
" actualización..."
),
"installing": (
"🕗 Instalando la"
" actualización..."
),
"success": (
"⏱ Reiniciado con éxito!"
" {}\nDescargandomódulos...\nReiniciado en {} segundos"
),
"full_success": (
"👍 ¡Bot actualizado con"
" éxito! {}\nReiniciado en {} segundos"
),
"secure_boot_complete": (
"🔒 ¡Modo de arranque seguro activado! {}\nReiniciado en {}"
" segundos"
),
"origin_cfg_doc": "Descargar actualización desde",
"btn_restart": "🔄 Reiniciar",
"btn_update": "🧭 Actualizar",
"restart_confirm": "❓ ¿Quieres reiniciar?",
"secure_boot_confirm": (
"❓ ¿Quieres reiniciar en modo de arranque seguro?"
),
"update_confirm": (
"❓ ¿Quieres actualizar?\n\n{} ⤑ {}'
),
"no_update": "🚸 Esta es la última versión. ¿Quieres actualizar?",
"cancel": "🚫 Cancelar",
"_cls_doc": "El usuario reinicia el bot",
"lavhost_restart": (
"✌️ {} Reiniciando..."
),
"lavhost_update": (
"✌️ {}"
" Actualizando..."
),
}
strings_kk = {
"source": (
"📖 Бастапқы коды бұл жерде қарауға болады'
),
"restarting_caption": (
"🕗 Твой {}"
" перезагружается..."
),
"downloading": (
"🕗 Жаңартуларды"
" жүктеу..."
),
"installing": (
"🕗 Жаңартуларды"
" орнату..."
),
"success": (
"⏱ Жаңарту сәтті"
" аяқталды! {}\nБірақ модульдер әлі жүктелуде...\nЖаңарту"
" {} секундқа аяқталды"
),
"full_success": (
"👍 Юзербот толық"
" жүктелді! {}\nТолық жаңарту {} секундқа аяқталды"
),
"secure_boot_complete": (
"🔒 Безпеке режимі аяқталды! {}\nЖаңарту {} секундқа аяқталды"
),
"origin_cfg_doc": "Жаңартуларды жүктеу үшін сілтеме",
"btn_restart": "🔄 Жаңарту",
"btn_update": "🧭 Жаңарту",
"restart_confirm": "❓ Сен жаңартуға сенімдісін бе?",
"secure_boot_confirm": (
"❓ Сен бұл бетті безпеке режимінде жаңартуға сенімдісін бе?"
),
"update_confirm": (
"❓ Сен жаңартуға сенімдісін бе??\n\n{} ⤑ {}'
),
"no_update": (
"🚸 Сіздің соңғы нұсқаныңыз бар. Сіз жаңартуға мүмкіндік береді ме?"
),
"cancel": "🚫 Бас тарту",
"_cls_doc": "Юзерботты жаңарту",
"lavhost_restart": (
"✌️ Сіздің {}"
" жаңартуға басталды..."
),
"lavhost_update": (
"✌️ Сіздің {}"
" жаңартуға басталды..."
),
}
strings_tt = {
"source": (
"📖 Чыганак кодын монда укып була"
),
"restarting_caption": (
"🕗 Сезнең {} яңадан"
" башлана..."
),
"downloading": (
"🕗 Яңартуларны"
" йөкләү..."
),
"installing": (
"🕗 Яңартулар"
" урнаштыру..."
),
"success": (
"⏱ Яңарту бетте! {}\n"
"Ләкин модульләр әле йөкләнә...\nЯңарту {} сек дәвам итте"
),
"full_success": (
"👍 Юзербот тулысынча"
" йөкләнгән! {}\nТулы яңадан башлау {} сек дәвам итте"
),
"secure_boot_complete": (
"🔒 Куркынычсыз йөкләү тәмамланды! {}\n"
"Яңарту {} сек дәвам итте"
),
"origin_cfg_doc": "Яңартулар йөкләнәчәк сылтама",
"btn_restart": "🔄 Кабызу",
"btn_update": "🧭 Яңару",
"restart_confirm": "❓ Ты уверен, что хочешь перезагрузиться?",
"secure_boot_confirm": "❓ Сез яңадан башларга телисезме?",
"update_confirm": (
"❓ Сез яңартырга телисезме?\n\n{} ⤑ {}'
),
"no_update": "🚸 Сезнең соңгы версиягез бар. Яңарту мәҗбүриме?",
"cancel": "🚫 Бетерү",
"_cls_doc": "Юзерботны яңарта",
"lavhost_restart": (
"✌️ Сезнең {} яңадан"
" башлана..."
),
"lavhost_update": (
"✌️ Сезнең {}"
" яңартыла..."
),
}
def __init__(self):
self.config = loader.ModuleConfig(
loader.ConfigValue(
"GIT_ORIGIN_URL",
"https://github.com/hikariatama/Hikka",
lambda: self.strings("origin_cfg_doc"),
validator=loader.validators.Link(),
)
)
@loader.owner
@loader.command(
ru_doc="Перезагружает юзербот",
de_doc="Startet den Userbot neu",
tr_doc="Kullanıcı botunu yeniden başlatır",
uz_doc="Foydalanuvchi botini qayta ishga tushiradi",
es_doc="Reinicia el bot",
kk_doc="Жүктеген ботты қайта жүктейді",
)
async def restart(self, message: Message):
"""Restarts the userbot"""
secure_boot = "--secure-boot" in utils.get_args_raw(message)
try:
if (
"--force" in (utils.get_args_raw(message) or "")
or "-f" in (utils.get_args_raw(message) or "")
or not self.inline.init_complete
or not await self.inline.form(
message=message,
text=self.strings(
"secure_boot_confirm" if secure_boot else "restart_confirm"
),
reply_markup=[
{
"text": self.strings("btn_restart"),
"callback": self.inline_restart,
"args": (secure_boot,),
},
{"text": self.strings("cancel"), "action": "close"},
],
)
):
raise
except Exception:
await self.restart_common(message, secure_boot)
async def inline_restart(self, call: InlineCall, secure_boot: bool = False):
await self.restart_common(call, secure_boot=secure_boot)
async def process_restart_message(self, msg_obj: typing.Union[InlineCall, Message]):
self.set(
"selfupdatemsg",
msg_obj.inline_message_id
if hasattr(msg_obj, "inline_message_id")
else f"{utils.get_chat_id(msg_obj)}:{msg_obj.id}",
)
async def restart_common(
self,
msg_obj: typing.Union[InlineCall, Message],
secure_boot: bool = False,
):
if (
hasattr(msg_obj, "form")
and isinstance(msg_obj.form, dict)
and "uid" in msg_obj.form
and msg_obj.form["uid"] in self.inline._units
and "message" in self.inline._units[msg_obj.form["uid"]]
):
message = self.inline._units[msg_obj.form["uid"]]["message"]
else:
message = msg_obj
if secure_boot:
self._db.set(loader.__name__, "secure_boot", True)
msg_obj = await utils.answer(
msg_obj,
self.strings("restarting_caption").format(
utils.get_platform_emoji(self._client)
if self._client.hikka_me.premium
and CUSTOM_EMOJIS
and isinstance(msg_obj, Message)
else "Hikka"
)
if "LAVHOST" not in os.environ
else self.strings("lavhost_restart").format(
'✌️✌️✌️✌️'
if self._client.hikka_me.premium
and CUSTOM_EMOJIS
and isinstance(msg_obj, Message)
else "lavHost"
),
)
await self.process_restart_message(msg_obj)
self.set("restart_ts", time.time())
await self._db.remote_force_save()
if "LAVHOST" in os.environ:
os.system("lavhost restart")
return
with contextlib.suppress(Exception):
await main.hikka.web.stop()
handler = logging.getLogger().handlers[0]
handler.setLevel(logging.CRITICAL)
for client in self.allclients:
# Terminate main loop of all running clients
# Won't work if not all clients are ready
if client is not message.client:
await client.disconnect()
await message.client.disconnect()
utils.atexit(get_startup_callback(), signal.SIGTERM)
os.kill(os.getpid(), signal.SIGTERM)
async def download_common(self):
try:
repo = Repo(os.path.dirname(utils.get_base_dir()))
origin = repo.remote("origin")
r = origin.pull()
new_commit = repo.head.commit
for info in r:
if info.old_commit:
for d in new_commit.diff(info.old_commit):
if d.b_path == "requirements.txt":
return True
return False
except git.exc.InvalidGitRepositoryError:
repo = Repo.init(os.path.dirname(utils.get_base_dir()))
origin = repo.create_remote("origin", self.config["GIT_ORIGIN_URL"])
origin.fetch()
repo.create_head("master", origin.refs.master)
repo.heads.master.set_tracking_branch(origin.refs.master)
repo.heads.master.checkout(True)
return False
@staticmethod
def req_common():
# Now we have downloaded new code, install requirements
logger.debug("Installing new requirements...")
try:
subprocess.run(
[
sys.executable,
"-m",
"pip",
"install",
"-r",
os.path.join(
os.path.dirname(utils.get_base_dir()),
"requirements.txt",
),
"--user",
],
check=True,
)
except subprocess.CalledProcessError:
logger.exception("Req install failed")
@loader.owner
@loader.command(
ru_doc="Скачивает обновления юзербота",
de_doc="Lädt Updates für den Userbot herunter",
tr_doc="Userbot güncellemelerini indirir",
uz_doc="Userbot yangilanishlarini yuklaydi",
es_doc="Descarga las actualizaciones del bot",
kk_doc="Жүйе жаңартуларын жүктейді",
)
async def update(self, message: Message):
"""Downloads userbot updates"""
try:
current = utils.get_git_hash()
upcoming = next(
git.Repo().iter_commits(f"origin/{version.branch}", max_count=1)
).hexsha
if (
"--force" in (utils.get_args_raw(message) or "")
or "-f" in (utils.get_args_raw(message) or "")
or not self.inline.init_complete
or not await self.inline.form(
message=message,
text=self.strings("update_confirm").format(
current, current[:8], upcoming, upcoming[:8]
)
if upcoming != current
else self.strings("no_update"),
reply_markup=[
{
"text": self.strings("btn_update"),
"callback": self.inline_update,
},
{"text": self.strings("cancel"), "action": "close"},
],
)
):
raise
except Exception:
await self.inline_update(message)
async def inline_update(
self,
msg_obj: typing.Union[InlineCall, Message],
hard: bool = False,
):
# We don't really care about asyncio at this point, as we are shutting down
if hard:
os.system(f"cd {utils.get_base_dir()} && cd .. && git reset --hard HEAD")
try:
if "LAVHOST" in os.environ:
msg_obj = await utils.answer(
msg_obj,
self.strings("lavhost_update").format(
"✌️✌️✌️✌️"
if self._client.hikka_me.premium
and CUSTOM_EMOJIS
and isinstance(msg_obj, Message)
else "lavHost"
),
)
await self.process_restart_message(msg_obj)
os.system("lavhost update")
return
with contextlib.suppress(Exception):
msg_obj = await utils.answer(msg_obj, self.strings("downloading"))
req_update = await self.download_common()
with contextlib.suppress(Exception):
msg_obj = await utils.answer(msg_obj, self.strings("installing"))
if req_update:
self.req_common()
await self.restart_common(msg_obj)
except GitCommandError:
if not hard:
await self.inline_update(msg_obj, True)
return
logger.critical("Got update loop. Update manually via .terminal")
@loader.unrestricted
@loader.command(
ru_doc="Показать ссылку на исходный код проекта",
de_doc="Zeigt den Link zum Quellcode des Projekts an",
tr_doc="Proje kaynak kodu bağlantısını gösterir",
uz_doc="Loyihaning manba kodiga havola ko'rsatadi",
es_doc="Muestra el enlace al código fuente del proyecto",
kk_doc="Жобаның қайнар кодына сілтеме көрсетеді",
)
async def source(self, message: Message):
"""Links the source code of this project"""
await utils.answer(
message,
self.strings("source").format(self.config["GIT_ORIGIN_URL"]),
)
async def client_ready(self):
if self.get("selfupdatemsg") is not None:
try:
await self.update_complete()
except Exception:
logger.exception("Failed to complete update!")
if self.get("do_not_create", False):
return
try:
await self._add_folder()
except Exception:
logger.exception("Failed to add folder!")
self.set("do_not_create", True)
async def _add_folder(self):
folders = await self._client(GetDialogFiltersRequest())
if any(getattr(folder, "title", None) == "hikka" for folder in folders):
return
try:
folder_id = (
max(
folders,
key=lambda x: x.id,
).id
+ 1
)
except ValueError:
folder_id = 2
try:
await self._client(
UpdateDialogFilterRequest(
folder_id,
DialogFilter(
folder_id,
title="hikka",
pinned_peers=(
[
await self._client.get_input_entity(
self._client.loader.inline.bot_id
)
]
if self._client.loader.inline.init_complete
else []
),
include_peers=[
await self._client.get_input_entity(dialog.entity)
async for dialog in self._client.iter_dialogs(
None,
ignore_migrated=True,
)
if dialog.name
in {
"hikka-logs",
"hikka-onload",
"hikka-assets",
"hikka-backups",
"hikka-acc-switcher",
"silent-tags",
}
and dialog.is_channel
and (
dialog.entity.participants_count == 1
or dialog.entity.participants_count == 2
and dialog.name in {"hikka-logs", "silent-tags"}
)
or (
self._client.loader.inline.init_complete
and dialog.entity.id
== self._client.loader.inline.bot_id
)
or dialog.entity.id
in [
1554874075,
1697279580,
1679998924,
] # official hikka chats
],
emoticon="🐱",
exclude_peers=[],
contacts=False,
non_contacts=False,
groups=False,
broadcasts=False,
bots=False,
exclude_muted=False,
exclude_read=False,
exclude_archived=False,
),
)
)
except Exception:
logger.critical(
"Can't create Hikka folder. Possible reasons are:\n"
"- User reached the limit of folders in Telegram\n"
"- User got floodwait\n"
"Ignoring error and adding folder addition to ignore list"
)
async def update_complete(self):
logger.debug("Self update successful! Edit message")
start = self.get("restart_ts")
try:
took = round(time.time() - start)
except Exception:
took = "n/a"
msg = self.strings("success").format(utils.ascii_face(), took)
ms = self.get("selfupdatemsg")
if ":" in str(ms):
chat_id, message_id = ms.split(":")
chat_id, message_id = int(chat_id), int(message_id)
await self._client.edit_message(chat_id, message_id, msg)
return
await self.inline.bot.edit_message_text(
inline_message_id=ms,
text=self.inline.sanitise_text(msg),
)
async def full_restart_complete(self, secure_boot: bool = False):
start = self.get("restart_ts")
try:
took = round(time.time() - start)
except Exception:
took = "n/a"
self.set("restart_ts", None)
ms = self.get("selfupdatemsg")
msg = self.strings(
"secure_boot_complete" if secure_boot else "full_success"
).format(utils.ascii_face(), took)
if ms is None:
return
self.set("selfupdatemsg", None)
if ":" in str(ms):
chat_id, message_id = ms.split(":")
chat_id, message_id = int(chat_id), int(message_id)
await self._client.edit_message(chat_id, message_id, msg)
await asyncio.sleep(60)
await self._client.delete_messages(chat_id, message_id)
return
await self.inline.bot.edit_message_text(
inline_message_id=ms,
text=self.inline.sanitise_text(msg),
)