# ruff: noqa: F821, I001 from __future__ import annotations from functools import wraps from math import isfinite from time import monotonic from typing import TYPE_CHECKING if TYPE_CHECKING: from collections.abc import Awaitable, Callable from router import Event def guarded(func: Callable[..., Awaitable[None]]) -> Callable[..., Awaitable[None]]: @wraps(func) async def call(self, ctx: Event) -> None: try: if func.__name__ != "input" or (ctx.chat_id, ctx.sender_id) in self.pending: await self.localize(ctx) return await func(self, ctx) except RuntimeError: if ctx.group == "callbacks": await ctx.answer(self.t("Settings are unavailable. Reopen cfg and check the module permissions."), alert=True) else: await ctx.reply(self.t("Settings are unavailable. Reopen cfg and check the module permissions.")) return call class Module: def __init__(self) -> None: self.pending, self.language = {}, "ru" translations = { "Settings are unavailable. Reopen cfg and check the module permissions.": "Настройки недоступны. Открой cfg заново и проверь права модуля.", "Previous": "Назад", "Next": "Далее", "Main menu": "Главное меню", "Close": "Закрыть", "Choose a module to configure.": "Выбери модуль.", "No configurable modules on this page.": "На этой странице нет модулей с настройками или командами.", "Open Saved Messages or your personal bot to configure your userbot.": "Открой «Избранное» или личного бота, чтобы изменить настройки юзербота.", "Userbot settings": "Настройки юзербота", "Module settings": "Настройки модулей", "Choose the settings to change.": "Что хочешь настроить?", "Configurator opened in your personal bot.": "Настройки открыты в личном боте.", "Open your personal bot and send /cfg. Check that the bot is connected and the module has configuration permission.": "Открой личного бота и отправь /cfg. Если меню не появляется, проверь подключение бота и права Core Utils.", "Command names": "Названия команд", "Back": "Вернуться", "Modules": "Модули", "The field no longer exists.": "Этой настройки больше нет.", "Reset to default": "Сбросить значение", "The choice no longer exists.": "Этот вариант больше недоступен.", "This field has no default.": "Для этой настройки нет значения по умолчанию.", "The value was not saved. Check its type and limits, then reopen the field. The settings may have changed.": "Значение не сохранено. Проверь формат и ограничения, затем открой настройку заново. Возможно, её уже изменили.", "Cancelled.": "Ввод отменён.", "Enter a number.": "Введи число.", "Configurator closed.": "Настройки закрыты.", "Choose a field.": "Выбери настройку.", "Current": "Текущее значение", "Edit this field on the module's website page.": "Эта настройка меняется на странице модуля на сайте.", "Send the new value in this chat within 3 minutes. Send /cancel to cancel.": "Отправь новое значение в этот чат в течение трёх минут. Для отмены отправь /cancel.", "Enabled": "Включено", "Disabled": "Выключено", "Not set": "Не задано", "No time limit": "Без ограничения срока", } def t(self, value: str) -> str: return self.translations.get(value, value) if self.language == "ru" else value async def localize(self, ctx: Event) -> None: try: self.language = (await ctx.app.core.settings())["values"]["language"] except RuntimeError: pass def value(self, value: object, key: str) -> str: if value is None: return self.t("No time limit" if key == "log_retention" else "Not set") if type(value) is bool: return self.t("Enabled" if value else "Disabled") if key == "language" and value in ("ru", "en"): return "Русский" if value == "ru" else "English" return str(value) @staticmethod def button(text: str, callback: str, **values: object) -> dict: raw = text.encode() label = raw[:61].decode(errors="ignore") + "…" if len(raw) > 64 else text return {"text": label, "callback": callback, "values": values} def label(self, field: dict, key: str) -> str: value = field.get("label", key) return str(value.get(self.language, key) if isinstance(value, dict) else value) async def form(self, ctx: Event, text: str, buttons: list) -> None: await ctx.app.inline.form(ctx.chat_id, text, buttons, ttl=600, users=[ctx.sender_id]) if ctx.group == "callbacks": try: await ctx.app.bot.delete_message(ctx.chat_id, ctx.id) except RuntimeError: pass async def modules(self, ctx: Event, offset: int = 0) -> None: page = await ctx.app.core.modules(offset) buttons = [self.button(item["name"] + " · " + item["instance"].split("/")[1], "settings", instance=item["instance"]) for item in page["modules"]] rows = [buttons[index:index + 2] for index in range(0, len(buttons), 2)] navigation = [] if offset: navigation.append(self.button(self.t("Previous"), "menu", offset=max(0, offset - 16))) if page["next"] is not None: navigation.append(self.button(self.t("Next"), "menu", offset=page["next"])) if navigation: rows.append(navigation) rows.append([self.button(self.t("Main menu"), "home"), self.button(self.t("Close"), "close")]) await self.form(ctx, self.t("Choose a module to configure.") if buttons else self.t("No configurable modules on this page."), rows) async def home_page(self, ctx: Event) -> None: if ctx.chat_id != ctx.sender_id: await ctx.reply(self.t("Open Saved Messages or your personal bot to configure your userbot.")); return rows = [[self.button(self.t("Userbot settings"), "settings", instance="core/main")], [self.button(self.t("Module settings"), "menu")], [self.button(self.t("Close"), "close")]] await self.form(ctx, self.t("Choose the settings to change."), rows) @loader.callback(access="owner") @guarded async def home(self, ctx: Event) -> None: self.pending.pop((ctx.chat_id, ctx.sender_id), None) await self.home_page(ctx) async def configuration(self, ctx: Event, instance: str) -> dict: if instance == "core/main": return await ctx.app.core.settings() if instance.startswith("commands:"): return await ctx.app.core.commands(instance.removeprefix("commands:")) return await ctx.app.core.configuration(instance) @loader.command("cfg", access="owner", source="both") async def cfg(self, ctx: Event) -> None: self.pending.pop((ctx.chat_id, ctx.sender_id), None) try: await self.localize(ctx) await self.home_page(ctx) if ctx.chat_id != ctx.sender_id: return if ctx.source == "telegram": if ctx.data.get("out"): await ctx.edit(self.t("Configurator opened in your personal bot.")) else: await ctx.reply(self.t("Configurator opened in your personal bot.")) except RuntimeError: await ctx.reply(self.t("Open your personal bot and send /cfg. Check that the bot is connected and the module has configuration permission.")) @loader.callback(access="owner", fields={"offset": {"type": "integer", "default": 0, "min": 0}}) @guarded async def menu(self, ctx: Event) -> None: self.pending.pop((ctx.chat_id, ctx.sender_id), None) await self.modules(ctx, ctx.values["offset"]) @loader.callback(access="owner", fields={"instance": {"type": "string", "required": True}, "page": {"type": "integer", "default": 0, "min": 0}}) @guarded async def settings(self, ctx: Event) -> None: self.pending.pop((ctx.chat_id, ctx.sender_id), None) await self.settings_page(ctx, ctx.values["instance"], ctx.values["page"]) async def settings_page(self, ctx: Event, instance: str, page: int = 0) -> None: data = await self.configuration(ctx, instance) keys = list(data["fields"]) page = min(page, max(0, (len(keys) - 1) // 8)) rows = [[self.button(self.label(data["fields"][key], key), "field", instance=instance, field=key)] for key in keys[page * 8:page * 8 + 8]] navigation = [] if page: navigation.append(self.button(self.t("Previous"), "settings", instance=instance, page=page - 1)) if (page + 1) * 8 < len(keys): navigation.append(self.button(self.t("Next"), "settings", instance=instance, page=page + 1)) if navigation: rows.append(navigation) if data.get("commands"): rows.append([self.button(self.t("Command names"), "settings", instance="commands:" + instance)]) if instance == "core/main": title, back = self.t("Userbot settings"), self.button(self.t("Main menu"), "home") elif instance.startswith("commands:"): title, back = self.t("Command names"), self.button(self.t("Back"), "settings", instance=instance.removeprefix("commands:")) else: title, back = self.t("Module settings"), self.button(self.t("Modules"), "menu") rows.append([back, self.button(self.t("Close"), "close")]) await self.form(ctx, title + "\n" + self.t("Choose a field."), rows) @loader.callback(access="owner", fields={"instance": {"type": "string", "required": True}, "field": {"type": "string", "required": True}, "page": {"type": "integer", "default": 0, "min": 0}}) @guarded async def field(self, ctx: Event) -> None: instance, key, page = ctx.values["instance"], ctx.values["field"], ctx.values["page"] data = await self.configuration(ctx, instance) field = data["fields"].get(key) if not field: await ctx.answer(self.t("The field no longer exists."), alert=True); return value = data["values"][key] kind = field["type"] rows = [] caption = f"{self.label(field, key)}\n{self.t('Current')}: {self.value(value, key)[:400]}" self.pending.pop((ctx.chat_id, ctx.sender_id), None) if kind in ("secret", "file", "list", "json"): caption = self.label(field, key) + "\n" + self.t("Edit this field on the module's website page.") elif kind in ("boolean", "select"): choices = [True, False] if kind == "boolean" else field["choices"] for index in range(page * 8, min(len(choices), (page + 1) * 8)): rows.append([self.button(self.value(choices[index], key), "choose", instance=instance, field=key, index=index, revision=data["revision"])]) navigation = [] if page: navigation.append(self.button(self.t("Previous"), "field", instance=instance, field=key, page=page - 1)) if (page + 1) * 8 < len(choices): navigation.append(self.button(self.t("Next"), "field", instance=instance, field=key, page=page + 1)) if navigation: rows.append(navigation) else: self.pending[(ctx.chat_id, ctx.sender_id)] = {"instance": instance, "field": key, "type": kind, "revision": data["revision"], "prefix": data["prefix"], "expires": monotonic() + 180} caption += "\n" + self.t("Send the new value in this chat within 3 minutes. Send /cancel to cancel.") if "default" in field and kind not in ("secret", "file"): rows.append([self.button(self.t("Reset to default"), "reset", instance=instance, field=key, revision=data["revision"])]) rows.append([self.button(self.t("Back"), "settings", instance=instance), self.button(self.t("Close"), "close")]) await self.form(ctx, caption, rows) @loader.callback(access="owner", fields={"instance": {"type": "string", "required": True}, "field": {"type": "string", "required": True}, "index": {"type": "integer", "min": 0, "required": True}, "revision": {"type": "integer", "min": 0, "required": True}}) @guarded async def choose(self, ctx: Event) -> None: values = ctx.values data = await self.configuration(ctx, values["instance"]) field = data["fields"].get(values["field"], {}) choices = [True, False] if field.get("type") == "boolean" else field.get("choices", []) if values["index"] >= len(choices): await ctx.answer(self.t("The choice no longer exists."), alert=True); return await self.save(ctx, values, choices[values["index"]]) @loader.callback(access="owner", fields={"instance": {"type": "string", "required": True}, "field": {"type": "string", "required": True}, "revision": {"type": "integer", "min": 0, "required": True}}) @guarded async def reset(self, ctx: Event) -> None: values = ctx.values data = await self.configuration(ctx, values["instance"]) field = data["fields"].get(values["field"], {}) if "default" not in field: await ctx.answer(self.t("This field has no default."), alert=True); return await self.save(ctx, values, field["default"]) async def save(self, ctx: Event, values: dict, value: object) -> None: self.pending.pop((ctx.chat_id, ctx.sender_id), None) try: instance, changes, revision = values["instance"], {values["field"]: value}, values["revision"] if instance == "core/main": await ctx.app.core.preferences(changes, revision) elif instance.startswith("commands:"): await ctx.app.core.aliases(instance.removeprefix("commands:"), changes, revision) else: await ctx.app.core.configure(instance, changes, revision) except RuntimeError: await ctx.reply(self.t("The value was not saved. Check its type and limits, then reopen the field. The settings may have changed.")) return await self.localize(ctx) await self.settings_page(ctx, values["instance"]) @loader.handler(source="both", access="owner") @guarded async def input(self, ctx: Event) -> None: key = (ctx.chat_id, ctx.sender_id) pending = self.pending.get(key) if not pending: return if monotonic() >= pending["expires"]: self.pending.pop(key, None) return if ctx.text == "/cancel": self.pending.pop(key, None); await ctx.reply(self.t("Cancelled.")); return prefix_field = pending["instance"] == "core/main" and pending["field"] == "prefix" if not prefix_field and ctx.text.startswith((pending.get("prefix", "!"), "/")): return try: if pending["type"] in ("integer", "chat", "user"): value = int(ctx.text) elif pending["type"] == "number": value = float(ctx.text) if not isfinite(value): raise ValueError("A finite number is required") else: value = ctx.text except ValueError: await ctx.reply(self.t("Enter a number.")); return await self.save(ctx, pending, value) @loader.callback(access="owner") @guarded async def close(self, ctx: Event) -> None: self.pending.pop((ctx.chat_id, ctx.sender_id), None) await ctx.answer(self.t("Configurator closed.")) await ctx.app.bot.delete_message(ctx.chat_id, ctx.id)