diff --git a/app/handlers/admin/bot_configuration.py b/app/handlers/admin/bot_configuration.py
index b129ea68..96e93be8 100644
--- a/app/handlers/admin/bot_configuration.py
+++ b/app/handlers/admin/bot_configuration.py
@@ -1,13 +1,21 @@
+import io
+import json
import math
+import re
import time
-from typing import Iterable, List, Tuple
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Dict, Iterable, List, Optional, Tuple
from aiogram import Dispatcher, F, types
from aiogram.filters import BaseFilter, StateFilter
from aiogram.fsm.context import FSMContext
+from aiogram.types import BufferedInputFile
+from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
-from app.database.models import User
+from app.database.models import SystemSetting, User
+from app.database.crud.system_setting import delete_system_setting, upsert_system_setting
from app.localization.texts import get_texts
from app.config import settings
from app.services.remnawave_service import RemnaWaveService
@@ -20,80 +28,230 @@ from app.utils.currency_converter import currency_converter
from app.external.telegram_stars import TelegramStarsService
-CATEGORY_PAGE_SIZE = 10
SETTINGS_PAGE_SIZE = 8
-CATEGORY_GROUP_DEFINITIONS: Tuple[Tuple[str, str, Tuple[str, ...]], ...] = (
- (
- "core",
- "⚙️ Основные настройки",
- ("SUPPORT", "LOCALIZATION", "MAINTENANCE"),
+@dataclass(frozen=True)
+class SpecCategory:
+ key: str
+ title: str
+ description: str
+ icon: str
+ category_keys: Tuple[str, ...]
+
+
+SPEC_CATEGORIES: Tuple[SpecCategory, ...] = (
+ SpecCategory(
+ key="core",
+ title="🤖 Основные",
+ description="Базовые настройки бота и поведение по умолчанию.",
+ icon="🤖",
+ category_keys=("CHANNEL",),
),
- (
- "channels_notifications",
- "📢 Каналы и уведомления",
- ("CHANNEL", "ADMIN_NOTIFICATIONS", "ADMIN_REPORTS"),
+ SpecCategory(
+ key="support",
+ title="💬 Поддержка",
+ description="Система тикетов, контакты и SLA.",
+ icon="💬",
+ category_keys=("SUPPORT",),
),
- (
- "subscriptions",
- "💎 Подписки и тарифы",
- ("TRIAL", "PAID_SUBSCRIPTION", "PERIODS", "SUBSCRIPTION_PRICES", "TRAFFIC", "TRAFFIC_PACKAGES", "DISCOUNTS"),
+ SpecCategory(
+ key="payments",
+ title="💳 Платежные системы",
+ description="Управление всеми способами оплаты и текстами чеков.",
+ icon="💳",
+ category_keys=("PAYMENT", "TELEGRAM", "CRYPTOBOT", "YOOKASSA", "TRIBUTE", "MULENPAY", "PAL24"),
),
- (
- "payments",
- "💳 Платежные системы",
- ("PAYMENT", "TELEGRAM", "CRYPTOBOT", "YOOKASSA", "TRIBUTE", "MULENPAY", "PAL24"),
+ SpecCategory(
+ key="subscriptions",
+ title="📅 Подписки и цены",
+ description="Периоды, тарифы, трафик и автопродление.",
+ icon="📅",
+ category_keys=(
+ "PAID_SUBSCRIPTION",
+ "PERIODS",
+ "SUBSCRIPTION_PRICES",
+ "TRAFFIC",
+ "TRAFFIC_PACKAGES",
+ "DISCOUNTS",
+ "AUTOPAY",
+ ),
),
- (
- "remnawave",
- "🔗 RemnaWave API",
- ("REMNAWAVE",),
+ SpecCategory(
+ key="trial",
+ title="🎁 Пробный период",
+ description="Настройки бесплатного доступа и ограничений.",
+ icon="🎁",
+ category_keys=("TRIAL",),
),
- (
- "referral",
- "🤝 Реферальная система",
- ("REFERRAL",),
+ SpecCategory(
+ key="referral",
+ title="👥 Реферальная программа",
+ description="Бонусы, комиссии и уведомления за приглашения.",
+ icon="👥",
+ category_keys=("REFERRAL",),
),
- (
- "autopay",
- "🔄 Автопродление",
- ("AUTOPAY",),
+ SpecCategory(
+ key="notifications",
+ title="🔔 Уведомления",
+ description="Админ-уведомления, отчеты и SLA.",
+ icon="🔔",
+ category_keys=("ADMIN_NOTIFICATIONS", "ADMIN_REPORTS", "NOTIFICATIONS"),
),
- (
- "interface",
- "🎨 Интерфейс и UX",
- ("INTERFACE_BRANDING", "INTERFACE_SUBSCRIPTION", "CONNECT_BUTTON", "HAPP", "SKIP", "ADDITIONAL"),
+ SpecCategory(
+ key="branding",
+ title="🎨 Интерфейс и брендинг",
+ description="Логотип, тексты, языки и Mini App.",
+ icon="🎨",
+ category_keys=(
+ "LOCALIZATION",
+ "INTERFACE_BRANDING",
+ "INTERFACE_SUBSCRIPTION",
+ "CONNECT_BUTTON",
+ "HAPP",
+ "SKIP",
+ "ADDITIONAL",
+ "MINIAPP",
+ ),
),
- (
- "database",
- "🗄️ База данных",
- ("DATABASE", "POSTGRES", "SQLITE", "REDIS"),
+ SpecCategory(
+ key="database",
+ title="💾 База данных",
+ description="Настройки PostgreSQL, SQLite и Redis.",
+ icon="💾",
+ category_keys=("DATABASE", "POSTGRES", "SQLITE", "REDIS"),
),
- (
- "monitoring",
- "📊 Мониторинг",
- ("MONITORING", "NOTIFICATIONS", "SERVER"),
+ SpecCategory(
+ key="remnawave",
+ title="🌐 RemnaWave API",
+ description="Интеграция с панелью RemnaWave и тест соединения.",
+ icon="🌐",
+ category_keys=("REMNAWAVE",),
),
- (
- "backup",
- "💾 Система бэкапов",
- ("BACKUP",),
+ SpecCategory(
+ key="servers",
+ title="📊 Статус серверов",
+ description="Мониторинг инфраструктуры и внешние метрики.",
+ icon="📊",
+ category_keys=("SERVER", "MONITORING"),
),
- (
- "updates",
- "🔄 Обновления",
- ("VERSION",),
+ SpecCategory(
+ key="maintenance",
+ title="🔧 Обслуживание",
+ description="Техработы, резервные копии и проверки обновлений.",
+ icon="🔧",
+ category_keys=("MAINTENANCE", "BACKUP", "VERSION"),
),
- (
- "development",
- "🔧 Разработка",
- ("LOG", "WEBHOOK", "WEB_API", "DEBUG"),
+ SpecCategory(
+ key="advanced",
+ title="⚡ Расширенные",
+ description="Webhook, Web API, логирование и режим разработки.",
+ icon="⚡",
+ category_keys=("WEBHOOK", "WEB_API", "LOG", "DEBUG"),
),
)
-CATEGORY_FALLBACK_KEY = "other"
-CATEGORY_FALLBACK_TITLE = "📦 Прочие настройки"
+SPEC_CATEGORY_MAP: Dict[str, SpecCategory] = {category.key: category for category in SPEC_CATEGORIES}
+CATEGORY_TO_SPEC: Dict[str, str] = {}
+for category in SPEC_CATEGORIES:
+ for cat_key in category.category_keys:
+ CATEGORY_TO_SPEC[cat_key] = category.key
+
+MAPPED_CATEGORY_KEYS = set(CATEGORY_TO_SPEC.keys())
+DEFAULT_SPEC_KEY = SPEC_CATEGORIES[0].key
+CUSTOM_PRESET_PREFIX = "botcfg_preset::"
+
+BUILTIN_PRESETS: Dict[str, Dict[str, Any]] = {
+ "recommended": {
+ "title": "Рекомендуемые настройки",
+ "description": "Сбалансированная конфигурация для стабильной работы.",
+ "changes": {
+ "SUPPORT_MENU_ENABLED": True,
+ "SUPPORT_SYSTEM_MODE": "both",
+ "SUPPORT_TICKET_SLA_ENABLED": True,
+ "ENABLE_NOTIFICATIONS": True,
+ "ADMIN_REPORTS_ENABLED": True,
+ "ADMIN_REPORTS_SEND_TIME": "09:00",
+ "REFERRAL_NOTIFICATIONS_ENABLED": True,
+ },
+ },
+ "minimal": {
+ "title": "Минимальная конфигурация",
+ "description": "Только основные функции без дополнительных уведомлений.",
+ "changes": {
+ "SUPPORT_MENU_ENABLED": False,
+ "ENABLE_NOTIFICATIONS": False,
+ "ADMIN_NOTIFICATIONS_ENABLED": False,
+ "ADMIN_REPORTS_ENABLED": False,
+ "TRIAL_DURATION_DAYS": 0,
+ },
+ },
+ "secure": {
+ "title": "Максимальная безопасность",
+ "description": "Ограничение внешнего доступа и усиленный контроль.",
+ "changes": {
+ "ENABLE_DEEP_LINKS": False,
+ "WEB_API_ENABLED": False,
+ "MAINTENANCE_AUTO_ENABLE": True,
+ "CONNECT_BUTTON_MODE": "guide",
+ "REFERRAL_NOTIFICATIONS_ENABLED": False,
+ },
+ },
+ "testing": {
+ "title": "Для тестирования",
+ "description": "Удобно для стендов и проверки интеграций.",
+ "changes": {
+ "DEBUG": True,
+ "ENABLE_NOTIFICATIONS": False,
+ "TELEGRAM_STARS_ENABLED": True,
+ "YOOKASSA_ENABLED": False,
+ "CRYPTOBOT_ENABLED": False,
+ "MAINTENANCE_MODE": False,
+ },
+ },
+}
+
+QUICK_ACTIONS: Dict[str, Dict[str, Any]] = {
+ "enable_notifications": {
+ "title": "🟢 Включить все уведомления",
+ "description": "Активирует пользовательские и админские уведомления.",
+ "changes": {
+ "ENABLE_NOTIFICATIONS": True,
+ "ADMIN_NOTIFICATIONS_ENABLED": True,
+ "ADMIN_REPORTS_ENABLED": True,
+ "REFERRAL_NOTIFICATIONS_ENABLED": True,
+ },
+ },
+ "disable_payments": {
+ "title": "⚪ Отключить все платежи",
+ "description": "Мгновенно выключает все интеграции оплаты.",
+ "changes": {
+ "YOOKASSA_ENABLED": False,
+ "CRYPTOBOT_ENABLED": False,
+ "MULENPAY_ENABLED": False,
+ "PAL24_ENABLED": False,
+ "TRIBUTE_ENABLED": False,
+ "TELEGRAM_STARS_ENABLED": False,
+ },
+ },
+ "enable_maintenance": {
+ "title": "🔧 Включить режим обслуживания",
+ "description": "Переводит бота в режим техработ с текущим сообщением.",
+ "changes": {
+ "MAINTENANCE_MODE": True,
+ },
+ },
+}
+
+
+def _format_actor(db_user: User) -> str:
+ try:
+ full_name = db_user.full_name # type: ignore[attr-defined]
+ except AttributeError:
+ full_name = None
+ if not full_name:
+ full_name = db_user.username or f"ID{db_user.telegram_id}"
+ return f"{full_name}#{db_user.telegram_id}"
async def _store_setting_context(
@@ -153,123 +311,312 @@ def _chunk(buttons: Iterable[types.InlineKeyboardButton], size: int) -> Iterable
yield buttons_list[index : index + size]
-def _parse_category_payload(payload: str) -> Tuple[str, str, int, int]:
- parts = payload.split(":")
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
- category_key = parts[2] if len(parts) > 2 else ""
+def _iter_all_definitions() -> Iterable[Any]:
+ for category_key, _, _ in bot_configuration_service.get_categories():
+ for definition in bot_configuration_service.get_settings_for_category(category_key):
+ yield definition
- def _safe_int(value: str, default: int = 1) -> int:
+
+def _preset_storage_key(slug: str) -> str:
+ return f"{CUSTOM_PRESET_PREFIX}{slug}"
+
+
+async def _load_custom_presets(db: AsyncSession) -> Dict[str, Dict[str, Any]]:
+ result = await db.execute(
+ select(SystemSetting).where(SystemSetting.key.like(f"{CUSTOM_PRESET_PREFIX}%"))
+ )
+ presets: Dict[str, Dict[str, Any]] = {}
+ for setting in result.scalars():
+ slug = setting.key[len(CUSTOM_PRESET_PREFIX) :]
try:
- return max(1, int(value))
- except (TypeError, ValueError):
- return default
-
- category_page = _safe_int(parts[3]) if len(parts) > 3 else 1
- settings_page = _safe_int(parts[4]) if len(parts) > 4 else 1
- return group_key, category_key, category_page, settings_page
+ payload = json.loads(setting.value or "{}")
+ except json.JSONDecodeError:
+ continue
+ if not isinstance(payload, dict):
+ continue
+ changes = payload.get("changes")
+ if not isinstance(changes, dict):
+ continue
+ presets[slug] = {
+ "title": payload.get("title") or f"Пользовательский ({slug})",
+ "description": payload.get("description") or "Сохраненный администратором пресет.",
+ "changes": changes,
+ }
+ return presets
-def _parse_group_payload(payload: str) -> Tuple[str, int]:
+def _parse_spec_payload(payload: str) -> Tuple[str, int]:
parts = payload.split(":")
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
+ spec_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
try:
page = max(1, int(parts[2]))
except (IndexError, ValueError):
page = 1
- return group_key, page
+ return spec_key, page
-def _get_grouped_categories() -> List[Tuple[str, str, List[Tuple[str, str, int]]]]:
- categories = bot_configuration_service.get_categories()
- categories_map = {key: (label, count) for key, label, count in categories}
- used: set[str] = set()
- grouped: List[Tuple[str, str, List[Tuple[str, str, int]]]] = []
+def _get_spec_settings(spec_key: str) -> List[Any]:
+ category = SPEC_CATEGORY_MAP.get(spec_key)
+ if not category:
+ return []
- for group_key, title, category_keys in CATEGORY_GROUP_DEFINITIONS:
- items: List[Tuple[str, str, int]] = []
- for category_key in category_keys:
- if category_key in categories_map:
- label, count = categories_map[category_key]
- items.append((category_key, label, count))
- used.add(category_key)
- if items:
- grouped.append((group_key, title, items))
+ definitions: List[Any] = []
+ seen: set[str] = set()
- remaining = [
- (key, label, count)
- for key, (label, count) in categories_map.items()
- if key not in used
+ for category_key in category.category_keys:
+ for definition in bot_configuration_service.get_settings_for_category(category_key):
+ if definition.key not in seen:
+ definitions.append(definition)
+ seen.add(definition.key)
+
+ if spec_key == "advanced":
+ for category_key, _, _ in bot_configuration_service.get_categories():
+ if category_key not in MAPPED_CATEGORY_KEYS:
+ for definition in bot_configuration_service.get_settings_for_category(category_key):
+ if definition.key not in seen:
+ definitions.append(definition)
+ seen.add(definition.key)
+
+ definitions.sort(key=lambda definition: definition.display_name.lower())
+ return definitions
+
+
+async def _apply_changeset(
+ db: AsyncSession,
+ changes: Dict[str, Any],
+ *,
+ db_user: User,
+ reason: str,
+) -> Tuple[List[Tuple[str, Any]], List[Tuple[str, str]]]:
+ actor = _format_actor(db_user)
+ applied: List[Tuple[str, Any]] = []
+ failed: List[Tuple[str, str]] = []
+
+ for key, value in changes.items():
+ try:
+ if value is None:
+ await bot_configuration_service.reset_value(
+ db, key, actor=actor, reason=reason
+ )
+ applied.append((key, value))
+ else:
+ prepared_value = value
+ if isinstance(value, str):
+ try:
+ prepared_value = bot_configuration_service.parse_user_value(
+ key, value
+ )
+ except ValueError:
+ prepared_value = value
+ await bot_configuration_service.set_value(
+ db, key, prepared_value, actor=actor, reason=reason
+ )
+ applied.append((key, prepared_value))
+ except Exception as error: # pragma: no cover - defensive
+ failed.append((key, str(error)))
+
+ if applied:
+ await db.commit()
+ else:
+ await db.rollback()
+
+ return applied, failed
+
+
+def _get_spec_key_for_category(category_key: str) -> str:
+ return CATEGORY_TO_SPEC.get(category_key, "advanced")
+
+
+def _get_spec_page_for_setting(spec_key: str, setting_key: str) -> int:
+ definitions = _get_spec_settings(spec_key)
+ for index, definition in enumerate(definitions):
+ if definition.key == setting_key:
+ return (index // SETTINGS_PAGE_SIZE) + 1
+ return 1
+
+
+def _compute_category_health(spec_key: str) -> Tuple[str, str]:
+ if spec_key == "core":
+ if settings.CHANNEL_IS_REQUIRED_SUB and not settings.CHANNEL_LINK:
+ return "🟡", "Добавьте ссылку на обязательный канал"
+ if settings.CHANNEL_IS_REQUIRED_SUB:
+ return "🟢", "Обязательная подписка активна"
+ return "⚪", "Канал не обязателен"
+
+ if spec_key == "support":
+ return (
+ "🟢" if settings.SUPPORT_MENU_ENABLED else "⚪",
+ "Меню поддержки включено" if settings.SUPPORT_MENU_ENABLED else "Меню поддержки скрыто",
+ )
+
+ if spec_key == "payments":
+ active = sum(
+ [
+ 1 if settings.is_yookassa_enabled() else 0,
+ 1 if settings.is_cryptobot_enabled() else 0,
+ 1 if settings.is_mulenpay_enabled() else 0,
+ 1 if settings.is_pal24_enabled() else 0,
+ 1 if settings.TRIBUTE_ENABLED else 0,
+ 1 if settings.TELEGRAM_STARS_ENABLED else 0,
+ ]
+ )
+ if active:
+ return "🟢", f"Активных методов: {active}"
+ return "🔴", "Нет настроенных платежных систем"
+
+ if spec_key == "subscriptions":
+ periods = [period.strip() for period in (settings.AVAILABLE_SUBSCRIPTION_PERIODS or "").split(",") if period.strip()]
+ base_price_ok = bool(settings.BASE_SUBSCRIPTION_PRICE)
+ if base_price_ok and periods:
+ return "🟢", f"Доступно периодов: {len(periods)}"
+ return "🟡", "Проверьте периоды и цены"
+
+ if spec_key == "trial":
+ if settings.TRIAL_DURATION_DAYS > 0:
+ return "🟢", f"{settings.TRIAL_DURATION_DAYS} дн. пробного периода"
+ return "⚪", "Триал отключен"
+
+ if spec_key == "referral":
+ if settings.REFERRAL_COMMISSION_PERCENT > 0:
+ return "🟢", f"Комиссия {settings.REFERRAL_COMMISSION_PERCENT}%"
+ return "⚪", "Комиссии не настроены"
+
+ if spec_key == "notifications":
+ if settings.ENABLE_NOTIFICATIONS:
+ return "🟢", "Пользовательские уведомления активны"
+ return "⚪", "Уведомления выключены"
+
+ if spec_key == "branding":
+ if settings.ENABLE_LOGO_MODE:
+ return "🟢", "Брендирование включено"
+ return "⚪", "Используется базовый интерфейс"
+
+ if spec_key == "database":
+ return "🟢", f"Режим БД: {settings.DATABASE_MODE}"
+
+ if spec_key == "remnawave":
+ if settings.REMNAWAVE_API_URL and settings.REMNAWAVE_API_KEY:
+ return "🟢", "API подключен"
+ return "🟡", "Проверьте URL и ключ"
+
+ if spec_key == "servers":
+ if settings.SERVER_STATUS_MODE != "disabled":
+ return "🟢", f"Режим: {settings.SERVER_STATUS_MODE}"
+ return "⚪", "Мониторинг выключен"
+
+ if spec_key == "maintenance":
+ if settings.MAINTENANCE_MODE:
+ return "🟡", "Техработы активны"
+ if settings.BACKUP_AUTO_ENABLED:
+ return "🟢", "Бэкапы включены"
+ return "⚪", "Автобэкапы отключены"
+
+ if spec_key == "advanced":
+ if settings.DEBUG or settings.WEB_API_ENABLED:
+ return "🟡", "Проверьте режимы разработки"
+ return "🟢", "Продакшен режим"
+
+ return "⚪", "Без статуса"
+
+
+def _render_spec_category_text(spec_key: str, page: int, language: str) -> str:
+ category = SPEC_CATEGORY_MAP.get(spec_key)
+ if not category:
+ return ""
+
+ definitions = _get_spec_settings(spec_key)
+ total_pages = max(1, math.ceil(len(definitions) / SETTINGS_PAGE_SIZE))
+ page = max(1, min(page, total_pages))
+
+ start = (page - 1) * SETTINGS_PAGE_SIZE
+ end = start + SETTINGS_PAGE_SIZE
+ sliced = definitions[start:end]
+
+ lines = [
+ f"🏠 Главная → {category.icon} {category.title}",
+ "",
+ category.description,
+ "",
+ f"Страница {page}/{total_pages}",
+ "",
]
- if remaining:
- remaining.sort(key=lambda item: item[1])
- grouped.append((CATEGORY_FALLBACK_KEY, CATEGORY_FALLBACK_TITLE, remaining))
+ for definition in sliced:
+ entry = bot_configuration_service.get_setting_dashboard_entry(definition.key)
+ metadata = bot_configuration_service.get_metadata(definition.key)
+ lines.append(f"{entry['state_icon']} {entry['icon']} {entry['name']}")
+ lines.append(f" {entry['value']}")
+ if entry["has_override"]:
+ lines.append(" ♻️ Переопределено в БД")
+ if metadata.recommended is not None:
+ recommended_display = bot_configuration_service.format_setting_value(
+ definition.key, metadata.recommended
+ )
+ if recommended_display != entry["value"]:
+ lines.append(f" ✨ Рекомендуемое: {recommended_display}")
- return grouped
+ lines.append("")
+ lines.append("ℹ️ Используйте кнопки ниже для редактирования и справки.")
+ return "\n".join(lines)
-def _build_groups_keyboard() -> types.InlineKeyboardMarkup:
- grouped = _get_grouped_categories()
+def _build_spec_category_keyboard(
+ spec_key: str,
+ page: int,
+ language: str,
+) -> types.InlineKeyboardMarkup:
+ definitions = _get_spec_settings(spec_key)
+ total_pages = max(1, math.ceil(len(definitions) / SETTINGS_PAGE_SIZE))
+ page = max(1, min(page, total_pages))
+
+ start = (page - 1) * SETTINGS_PAGE_SIZE
+ end = start + SETTINGS_PAGE_SIZE
+ sliced = definitions[start:end]
+
rows: list[list[types.InlineKeyboardButton]] = []
+ texts = get_texts(language)
- for group_key, title, items in grouped:
- total = sum(count for _, _, count in items)
+ if spec_key == "remnawave":
rows.append(
[
types.InlineKeyboardButton(
- text=f"{title} ({total})",
- callback_data=f"botcfg_group:{group_key}:1",
+ text="🔌 Проверить подключение",
+ callback_data=f"botcfg_test_remnawave:{spec_key}:{page}",
)
]
)
- rows.append(
- [
- types.InlineKeyboardButton(
- text="⬅️ Назад",
- callback_data="admin_submenu_settings",
+ if spec_key == "payments":
+ def _test_button(label: str, method: str) -> types.InlineKeyboardButton:
+ return types.InlineKeyboardButton(
+ text=label,
+ callback_data=f"botcfg_test_payment:{method}:{spec_key}:{page}",
)
- ]
- )
- return types.InlineKeyboardMarkup(inline_keyboard=rows)
-
-
-def _build_categories_keyboard(
- group_key: str,
- group_title: str,
- categories: List[Tuple[str, str, int]],
- page: int = 1,
-) -> types.InlineKeyboardMarkup:
- total_pages = max(1, math.ceil(len(categories) / CATEGORY_PAGE_SIZE))
- page = max(1, min(page, total_pages))
-
- start = (page - 1) * CATEGORY_PAGE_SIZE
- end = start + CATEGORY_PAGE_SIZE
- sliced = categories[start:end]
-
- rows: list[list[types.InlineKeyboardButton]] = []
- rows.append(
- [
- types.InlineKeyboardButton(
- text=f"— {group_title} —",
- callback_data="botcfg_group:noop",
- )
- ]
- )
-
- buttons: List[types.InlineKeyboardButton] = []
- for category_key, label, count in sliced:
- button_text = f"{label} ({count})"
- buttons.append(
- types.InlineKeyboardButton(
- text=button_text,
- callback_data=f"botcfg_cat:{group_key}:{category_key}:{page}:1",
- )
+ rows.extend(
+ [
+ [_test_button(texts.t("PAYMENT_CARD_YOOKASSA", "💳 YooKassa · тест"), "yookassa")],
+ [_test_button(texts.t("PAYMENT_CARD_TRIBUTE", "💳 Tribute · тест"), "tribute")],
+ [_test_button(texts.t("PAYMENT_CARD_MULENPAY", "💳 MulenPay · тест"), "mulenpay")],
+ [_test_button(texts.t("PAYMENT_CARD_PAL24", "💳 PayPalych · тест"), "pal24")],
+ [_test_button(texts.t("PAYMENT_TELEGRAM_STARS", "⭐ Telegram Stars · тест"), "stars")],
+ [_test_button(texts.t("PAYMENT_CRYPTOBOT", "🪙 CryptoBot · тест"), "cryptobot")],
+ ]
)
- for chunk in _chunk(buttons, 2):
- rows.append(list(chunk))
+ for definition in sliced:
+ entry = bot_configuration_service.get_setting_dashboard_entry(definition.key)
+ callback_token = bot_configuration_service.get_callback_token(definition.key)
+ name_prefix = "★ " if entry["has_override"] else ""
+ button_text = f"{entry['icon']} {name_prefix}{entry['name']}"
+ info_callback = f"botcfg_info:{spec_key}:{page}:1:{callback_token}"
+ edit_callback = f"botcfg_setting:{spec_key}:{page}:1:{callback_token}"
+ rows.append(
+ [
+ types.InlineKeyboardButton(text=button_text, callback_data=edit_callback),
+ types.InlineKeyboardButton(text="ℹ️", callback_data=info_callback),
+ ]
+ )
if total_pages > 1:
nav_row: list[types.InlineKeyboardButton] = []
@@ -277,7 +624,7 @@ def _build_categories_keyboard(
nav_row.append(
types.InlineKeyboardButton(
text="⬅️",
- callback_data=f"botcfg_group:{group_key}:{page - 1}",
+ callback_data=f"botcfg_group:{spec_key}:{page - 1}",
)
)
nav_row.append(
@@ -290,135 +637,57 @@ def _build_categories_keyboard(
nav_row.append(
types.InlineKeyboardButton(
text="➡️",
- callback_data=f"botcfg_group:{group_key}:{page + 1}",
+ callback_data=f"botcfg_group:{spec_key}:{page + 1}",
)
)
rows.append(nav_row)
rows.append(
[
- types.InlineKeyboardButton(
- text="⬅️ К разделам",
- callback_data="admin_bot_config",
- )
+ types.InlineKeyboardButton(text="🔍 Поиск", callback_data="botcfg_search:start"),
+ types.InlineKeyboardButton(text="⚡ Быстрые действия", callback_data="botcfg_quick_menu"),
+ ]
+ )
+
+ rows.append(
+ [
+ types.InlineKeyboardButton(text="⬅️ К разделам", callback_data="admin_bot_config"),
+ types.InlineKeyboardButton(text="🏠 Главное меню", callback_data="admin_panel"),
]
)
return types.InlineKeyboardMarkup(inline_keyboard=rows)
-def _build_settings_keyboard(
- category_key: str,
- group_key: str,
- category_page: int,
- language: str,
- page: int = 1,
-) -> types.InlineKeyboardMarkup:
- definitions = bot_configuration_service.get_settings_for_category(category_key)
- total_pages = max(1, math.ceil(len(definitions) / SETTINGS_PAGE_SIZE))
- page = max(1, min(page, total_pages))
-
- start = (page - 1) * SETTINGS_PAGE_SIZE
- end = start + SETTINGS_PAGE_SIZE
- sliced = definitions[start:end]
-
+def _build_main_keyboard() -> types.InlineKeyboardMarkup:
rows: list[list[types.InlineKeyboardButton]] = []
- texts = get_texts(language)
-
- if category_key == "REMNAWAVE":
- rows.append(
- [
- types.InlineKeyboardButton(
- text="🔌 Проверить подключение",
- callback_data=(
- f"botcfg_test_remnawave:{group_key}:{category_key}:{category_page}:{page}"
- ),
- )
- ]
- )
-
- test_payment_buttons: list[list[types.InlineKeyboardButton]] = []
-
- def _test_button(text: str, method: str) -> types.InlineKeyboardButton:
- return types.InlineKeyboardButton(
- text=text,
- callback_data=(
- f"botcfg_test_payment:{method}:{group_key}:{category_key}:{category_page}:{page}"
- ),
- )
-
- if category_key == "YOOKASSA":
- label = texts.t("PAYMENT_CARD_YOOKASSA", "💳 Банковская карта (YooKassa)")
- test_payment_buttons.append([_test_button(f"{label} · тест", "yookassa")])
- elif category_key == "TRIBUTE":
- label = texts.t("PAYMENT_CARD_TRIBUTE", "💳 Банковская карта (Tribute)")
- test_payment_buttons.append([_test_button(f"{label} · тест", "tribute")])
- elif category_key == "MULENPAY":
- label = texts.t("PAYMENT_CARD_MULENPAY", "💳 Банковская карта (Mulen Pay)")
- test_payment_buttons.append([_test_button(f"{label} · тест", "mulenpay")])
- elif category_key == "PAL24":
- label = texts.t("PAYMENT_CARD_PAL24", "💳 Банковская карта (PayPalych)")
- test_payment_buttons.append([_test_button(f"{label} · тест", "pal24")])
- elif category_key == "TELEGRAM":
- label = texts.t("PAYMENT_TELEGRAM_STARS", "⭐ Telegram Stars")
- test_payment_buttons.append([_test_button(f"{label} · тест", "stars")])
- elif category_key == "CRYPTOBOT":
- label = texts.t("PAYMENT_CRYPTOBOT", "🪙 Криптовалюта (CryptoBot)")
- test_payment_buttons.append([_test_button(f"{label} · тест", "cryptobot")])
-
- if test_payment_buttons:
- rows.extend(test_payment_buttons)
-
- for definition in sliced:
- value_preview = bot_configuration_service.format_value_for_list(definition.key)
- button_text = f"{definition.display_name} · {value_preview}"
- if len(button_text) > 64:
- button_text = button_text[:63] + "…"
- callback_token = bot_configuration_service.get_callback_token(definition.key)
- rows.append(
- [
- types.InlineKeyboardButton(
- text=button_text,
- callback_data=(
- f"botcfg_setting:{group_key}:{category_page}:{page}:{callback_token}"
- ),
- )
- ]
- )
-
- if total_pages > 1:
- nav_row: list[types.InlineKeyboardButton] = []
- if page > 1:
- nav_row.append(
- types.InlineKeyboardButton(
- text="⬅️",
- callback_data=(
- f"botcfg_cat:{group_key}:{category_key}:{category_page}:{page - 1}"
- ),
- )
- )
- nav_row.append(
- types.InlineKeyboardButton(
- text=f"{page}/{total_pages}", callback_data="botcfg_cat_page:noop"
- )
- )
- if page < total_pages:
- nav_row.append(
- types.InlineKeyboardButton(
- text="➡️",
- callback_data=(
- f"botcfg_cat:{group_key}:{category_key}:{category_page}:{page + 1}"
- ),
- )
- )
- rows.append(nav_row)
-
- rows.append([
+ category_buttons = [
types.InlineKeyboardButton(
- text="⬅️ К категориям",
- callback_data=f"botcfg_group:{group_key}:{category_page}",
+ text=f"{category.icon} {category.title}",
+ callback_data=f"botcfg_group:{category.key}:1",
)
- ])
+ for category in SPEC_CATEGORIES
+ ]
+
+ for chunk in _chunk(category_buttons, 2):
+ rows.append(list(chunk))
+
+ rows.append([types.InlineKeyboardButton(text="🔍 Найти настройку", callback_data="botcfg_search:start")])
+ rows.append([types.InlineKeyboardButton(text="🎛 Пресеты", callback_data="botcfg_presets")])
+ rows.append(
+ [
+ types.InlineKeyboardButton(text="📤 Экспорт .env", callback_data="botcfg_export"),
+ types.InlineKeyboardButton(text="📥 Импорт .env", callback_data="botcfg_import"),
+ ]
+ )
+ rows.append([types.InlineKeyboardButton(text="🕑 История изменений", callback_data="botcfg_history")])
+ rows.append([types.InlineKeyboardButton(text="⚡ Быстрые действия", callback_data="botcfg_quick_menu")])
+ rows.append(
+ [
+ types.InlineKeyboardButton(text="⬅️ Назад", callback_data="admin_submenu_settings"),
+ types.InlineKeyboardButton(text="🏠 В главное меню", callback_data="admin_panel"),
+ ]
+ )
return types.InlineKeyboardMarkup(inline_keyboard=rows)
@@ -430,12 +699,13 @@ def _build_setting_keyboard(
settings_page: int,
) -> types.InlineKeyboardMarkup:
definition = bot_configuration_service.get_definition(key)
+ metadata = bot_configuration_service.get_metadata(key)
rows: list[list[types.InlineKeyboardButton]] = []
callback_token = bot_configuration_service.get_callback_token(key)
+ current_value = bot_configuration_service.get_current_value(key)
choice_options = bot_configuration_service.get_choice_options(key)
if choice_options:
- current_value = bot_configuration_service.get_current_value(key)
choice_buttons: list[types.InlineKeyboardButton] = []
for option in choice_options:
choice_token = bot_configuration_service.get_choice_token(key, option.value)
@@ -457,60 +727,124 @@ def _build_setting_keyboard(
rows.append(list(chunk))
if definition.python_type is bool:
- rows.append([
+ toggle_text = "❌ Выключить" if bool(current_value) else "✅ Включить"
+ rows.append(
+ [
+ types.InlineKeyboardButton(
+ text=toggle_text,
+ callback_data=(
+ f"botcfg_toggle:{group_key}:{category_page}:{settings_page}:{callback_token}"
+ ),
+ )
+ ]
+ )
+
+ rows.append(
+ [
types.InlineKeyboardButton(
- text="🔁 Переключить",
+ text="✏️ Изменить",
callback_data=(
- f"botcfg_toggle:{group_key}:{category_page}:{settings_page}:{callback_token}"
+ f"botcfg_edit:{group_key}:{category_page}:{settings_page}:{callback_token}"
),
)
- ])
+ ]
+ )
- rows.append([
- types.InlineKeyboardButton(
- text="✏️ Изменить",
- callback_data=(
- f"botcfg_edit:{group_key}:{category_page}:{settings_page}:{callback_token}"
- ),
+ if metadata.recommended is not None:
+ rows.append(
+ [
+ types.InlineKeyboardButton(
+ text="✨ Применить рекомендуемое",
+ callback_data=(
+ f"botcfg_recommend:{group_key}:{category_page}:{settings_page}:{callback_token}"
+ ),
+ )
+ ]
)
- ])
+
+ rows.append(
+ [
+ types.InlineKeyboardButton(
+ text="ℹ️ Помощь",
+ callback_data=(
+ f"botcfg_info:{group_key}:{category_page}:{settings_page}:{callback_token}"
+ ),
+ )
+ ]
+ )
if bot_configuration_service.has_override(key):
- rows.append([
- types.InlineKeyboardButton(
- text="♻️ Сбросить",
- callback_data=(
- f"botcfg_reset:{group_key}:{category_page}:{settings_page}:{callback_token}"
- ),
- )
- ])
-
- rows.append([
- types.InlineKeyboardButton(
- text="⬅️ Назад",
- callback_data=(
- f"botcfg_cat:{group_key}:{definition.category_key}:{category_page}:{settings_page}"
- ),
+ rows.append(
+ [
+ types.InlineKeyboardButton(
+ text="♻️ Сбросить",
+ callback_data=(
+ f"botcfg_reset:{group_key}:{category_page}:{settings_page}:{callback_token}"
+ ),
+ )
+ ]
)
- ])
+
+ rows.append(
+ [
+ types.InlineKeyboardButton(
+ text="⬅️ К списку",
+ callback_data=f"botcfg_group:{group_key}:{category_page}",
+ ),
+ types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config"),
+ ]
+ )
+
+ rows.append(
+ [
+ types.InlineKeyboardButton(
+ text="⚙️ Настройки",
+ callback_data="admin_submenu_settings",
+ )
+ ]
+ )
return types.InlineKeyboardMarkup(inline_keyboard=rows)
-def _render_setting_text(key: str) -> str:
+def _render_setting_text(key: str, spec_category: SpecCategory) -> str:
summary = bot_configuration_service.get_setting_summary(key)
+ metadata = bot_configuration_service.get_metadata(key)
lines = [
- "🧩 Настройка",
- f"Название: {summary['name']}",
- f"Ключ: {summary['key']}",
- f"Категория: {summary['category_label']}",
- f"Тип: {summary['type']}",
- f"Текущее значение: {summary['current']}",
- f"Значение по умолчанию: {summary['original']}",
- f"Переопределено в БД: {'✅ Да' if summary['has_override'] else '❌ Нет'}",
+ f"🏠 Главная → {spec_category.icon} {spec_category.title} → ⚙️ {summary['name']}",
+ "",
]
+ if metadata.description:
+ lines.append(f"📝 {metadata.description}")
+ lines.append("")
+
+ lines.extend(
+ [
+ f"🔑 Ключ: {summary['key']}",
+ f"📂 Категория: {summary['category_label']}",
+ f"📦 Тип: {summary['type']}",
+ f"📘 Текущее значение: {summary['current']}",
+ f"📗 По умолчанию: {summary['original']}",
+ f"📥 Переопределено: {'✅ Да' if summary['has_override'] else '❌ Нет'}",
+ ]
+ )
+
+ if metadata.format_hint:
+ lines.append(f"📐 Формат: {metadata.format_hint}")
+ if metadata.example:
+ lines.append(f"💡 Пример: {metadata.example}")
+ if metadata.warning:
+ lines.append(f"⚠️ Важно: {metadata.warning}")
+ if metadata.dependencies:
+ lines.append(f"🔗 Связано: {metadata.dependencies}")
+ if metadata.recommended is not None:
+ recommended_display = bot_configuration_service.format_setting_value(
+ key, metadata.recommended
+ )
+ lines.append(f"✨ Рекомендуемое: {recommended_display}")
+
choices = bot_configuration_service.get_choice_options(key)
if choices:
current_raw = bot_configuration_service.get_current_value(key)
@@ -518,7 +852,9 @@ def _render_setting_text(key: str) -> str:
lines.append("Доступные значения:")
for option in choices:
marker = "✅" if current_raw == option.value else "•"
- value_display = bot_configuration_service.format_value(option.value)
+ value_display = bot_configuration_service.format_setting_value(
+ key, option.value
+ )
description = option.description or ""
if description:
lines.append(
@@ -536,13 +872,25 @@ async def show_bot_config_menu(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
+ acknowledge: bool = True,
):
- keyboard = _build_groups_keyboard()
+ lines = ["⚙️ Панель управления ботом", ""]
+
+ for category in SPEC_CATEGORIES:
+ status_icon, summary = _compute_category_health(category.key)
+ lines.append(f"{status_icon} {category.icon} {category.title} — {summary}")
+
+ lines.append("")
+ lines.append("Выберите раздел или воспользуйтесь поиском ниже.")
+
+ keyboard = _build_main_keyboard()
await callback.message.edit_text(
- "🧩 Конфигурация бота\n\nВыберите раздел настроек:",
+ "\n".join(lines),
reply_markup=keyboard,
+ parse_mode="HTML",
)
- await callback.answer()
+ if acknowledge:
+ await callback.answer()
@admin_required
@@ -552,50 +900,584 @@ async def show_bot_config_group(
db_user: User,
db: AsyncSession,
):
- group_key, page = _parse_group_payload(callback.data)
- grouped = _get_grouped_categories()
- group_lookup = {key: (title, items) for key, title, items in grouped}
-
- if group_key not in group_lookup:
- await callback.answer("Эта группа больше недоступна", show_alert=True)
+ spec_key, page = _parse_spec_payload(callback.data)
+ category = SPEC_CATEGORY_MAP.get(spec_key)
+ if not category:
+ await callback.answer("Раздел больше недоступен", show_alert=True)
return
- group_title, items = group_lookup[group_key]
- keyboard = _build_categories_keyboard(group_key, group_title, items, page)
+ text = _render_spec_category_text(spec_key, page, db_user.language)
+ keyboard = _build_spec_category_keyboard(spec_key, page, db_user.language)
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def start_search_workflow(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ await state.set_state(BotConfigStates.waiting_for_search_query)
+ await state.update_data(
+ botcfg_origin="bot_config_search",
+ search_return_payload=callback.data,
+ )
+
+ lines = [
+ "🔍 Поиск по настройкам",
+ "",
+ "Введите часть названия, ключа или описания настройки.",
+ "Можно вводить несколько слов для уточнения.",
+ "Напишите cancel, чтобы выйти из поиска.",
+ ]
+
+ keyboard = types.InlineKeyboardMarkup(
+ inline_keyboard=[
+ [
+ types.InlineKeyboardButton(
+ text="🏠 Панель", callback_data="admin_bot_config"
+ ),
+ types.InlineKeyboardButton(
+ text="❌ Отмена", callback_data="botcfg_search:cancel"
+ ),
+ ]
+ ]
+ )
+
+ await callback.message.answer("\n".join(lines), parse_mode="HTML", reply_markup=keyboard)
+ await callback.answer("Введите запрос для поиска", show_alert=True)
+
+
+@admin_required
+@error_handler
+async def cancel_search(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ await state.clear()
+ await callback.answer("Поиск отменен")
+ await show_bot_config_menu(callback, db_user, db, acknowledge=False)
+
+
+@admin_required
+@error_handler
+async def handle_search_query(
+ message: types.Message,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ query = (message.text or "").strip()
+ if not query:
+ await message.answer("Введите запрос для поиска настроек.")
+ return
+
+ if query.lower() in {"cancel", "отмена", "стоп"}:
+ await message.answer(
+ "Поиск отменен.",
+ reply_markup=types.InlineKeyboardMarkup(
+ inline_keyboard=[[types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config")]]
+ ),
+ )
+ await state.clear()
+ return
+
+ keys = bot_configuration_service.search_settings(query)
+ if not keys:
+ await message.answer(
+ "😕 Ничего не найдено. Попробуйте уточнить запрос или используйте другое ключевое слово.",
+ )
+ return
+
+ lines = ["🔎 Найдены настройки", ""]
+ keyboard_rows: List[List[types.InlineKeyboardButton]] = []
+
+ for key in keys:
+ definition = bot_configuration_service.get_definition(key)
+ metadata = bot_configuration_service.get_metadata(key)
+ spec_key = _get_spec_key_for_category(definition.category_key)
+ spec_category = SPEC_CATEGORY_MAP.get(spec_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
+ page = _get_spec_page_for_setting(spec_key, key)
+ callback_token = bot_configuration_service.get_callback_token(key)
+
+ summary = bot_configuration_service.get_setting_summary(key)
+ value_preview = summary["current"]
+
+ lines.append(
+ f"{spec_category.icon} {definition.display_name} — {key}\n"
+ f"Текущее значение: {value_preview}\n"
+ f"Категория: {summary['category_label']}"
+ )
+ if metadata.description:
+ lines.append(_shorten_text(metadata.description))
+ lines.append("")
+
+ keyboard_rows.append(
+ [
+ types.InlineKeyboardButton(
+ text=f"{spec_category.icon} {definition.display_name}",
+ callback_data=f"botcfg_setting:{spec_key}:{page}:1:{callback_token}",
+ )
+ ]
+ )
+
+ keyboard_rows.append([
+ types.InlineKeyboardButton(text="🔁 Новый поиск", callback_data="botcfg_search:start"),
+ types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config"),
+ ])
+
+ await message.answer(
+ "\n".join(lines).rstrip(),
+ reply_markup=types.InlineKeyboardMarkup(inline_keyboard=keyboard_rows),
+ parse_mode="HTML",
+ )
+ await state.update_data(botcfg_search_last_query=query)
+
+
+def _shorten_text(text: str, limit: int = 120) -> str:
+ text = text.strip()
+ if len(text) <= limit:
+ return text
+ return text[: limit - 1] + "…"
+
+
+@admin_required
+@error_handler
+async def show_quick_actions_menu(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ lines = ["⚡ Быстрые действия", ""]
+ buttons: List[types.InlineKeyboardButton] = []
+
+ for key, action in QUICK_ACTIONS.items():
+ lines.append(f"{action['title']}")
+ lines.append(f"— {action['description']}")
+ lines.append("")
+ buttons.append(
+ types.InlineKeyboardButton(text=action["title"], callback_data=f"botcfg_quick:{key}")
+ )
+
+ keyboard_rows = [list(chunk) for chunk in _chunk(buttons, 1)]
+ keyboard_rows.append(
+ [
+ types.InlineKeyboardButton(text="🔁 Обновить", callback_data="botcfg_quick_menu"),
+ types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config"),
+ ]
+ )
+
await callback.message.edit_text(
- f"🧩 {group_title}\n\nВыберите категорию настроек:",
- reply_markup=keyboard,
+ "\n".join(lines).rstrip(),
+ reply_markup=types.InlineKeyboardMarkup(inline_keyboard=keyboard_rows),
+ parse_mode="HTML",
)
await callback.answer()
@admin_required
@error_handler
-async def show_bot_config_category(
+async def apply_quick_action(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ action_key = callback.data.split(":", 1)[1] if ":" in callback.data else ""
+ action = QUICK_ACTIONS.get(action_key)
+ if not action:
+ await callback.answer("Действие недоступно", show_alert=True)
+ return
+
+ applied, failed = await _apply_changeset(
+ db,
+ action["changes"],
+ db_user=db_user,
+ reason=f"quick:{action_key}",
+ )
+
+ lines = [f"{action['title']}", ""]
+ if applied:
+ lines.append(f"✅ Применено: {len(applied)} настроек")
+ if failed:
+ lines.append(f"⚠️ Ошибок: {len(failed)}")
+ for key, error in failed[:5]:
+ lines.append(f"• {key}: {error}")
+ lines.append("")
+ lines.append("Вы можете открыть панель и проверить результат.")
+
+ keyboard = types.InlineKeyboardMarkup(
+ inline_keyboard=[
+ [types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config")],
+ [types.InlineKeyboardButton(text="⚡ К действиям", callback_data="botcfg_quick_menu")],
+ ]
+ )
+
+ await callback.message.edit_text(
+ "\n".join(lines).rstrip(),
+ reply_markup=keyboard,
+ parse_mode="HTML",
+ )
+ await callback.answer("Готово")
+
+
+@admin_required
+@error_handler
+async def show_presets_menu(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ acknowledge: bool = True,
+):
+ custom_presets = await _load_custom_presets(db)
+ lines = ["🎛 Пресеты настроек", ""]
+
+ if BUILTIN_PRESETS:
+ lines.append("Встроенные пресеты:")
+ for slug, preset in BUILTIN_PRESETS.items():
+ lines.append(f"• {preset['title']} — {preset['description']}")
+ lines.append("")
+
+ if custom_presets:
+ lines.append("Пользовательские пресеты:")
+ for slug, preset in custom_presets.items():
+ lines.append(f"• {preset['title']} — {preset['description']}")
+ lines.append("")
+ else:
+ lines.append("Пользовательские пресеты пока не сохранены.")
+ lines.append("")
+
+ keyboard_rows: List[List[types.InlineKeyboardButton]] = []
+ for slug, preset in BUILTIN_PRESETS.items():
+ keyboard_rows.append(
+ [
+ types.InlineKeyboardButton(
+ text=preset["title"],
+ callback_data=f"botcfg_preset_apply:{slug}",
+ )
+ ]
+ )
+
+ for slug, preset in custom_presets.items():
+ keyboard_rows.append(
+ [
+ types.InlineKeyboardButton(
+ text=preset["title"],
+ callback_data=f"botcfg_preset_apply:{slug}",
+ ),
+ types.InlineKeyboardButton(
+ text="🗑️", callback_data=f"botcfg_preset_delete:{slug}"
+ ),
+ ]
+ )
+
+ keyboard_rows.append(
+ [
+ types.InlineKeyboardButton(
+ text="💾 Сохранить текущие настройки", callback_data="botcfg_preset_save"
+ )
+ ]
+ )
+ keyboard_rows.append(
+ [
+ types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config"),
+ types.InlineKeyboardButton(text="⚡ Быстрые", callback_data="botcfg_quick_menu"),
+ ]
+ )
+
+ await callback.message.edit_text(
+ "\n".join(lines).rstrip(),
+ reply_markup=types.InlineKeyboardMarkup(inline_keyboard=keyboard_rows),
+ parse_mode="HTML",
+ )
+ if acknowledge:
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def apply_preset(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ slug = callback.data.split(":", 1)[1] if ":" in callback.data else ""
+ preset = BUILTIN_PRESETS.get(slug)
+ if not preset:
+ custom_presets = await _load_custom_presets(db)
+ preset = custom_presets.get(slug)
+ if not preset:
+ await callback.answer("Пресет не найден", show_alert=True)
+ return
+
+ applied, failed = await _apply_changeset(
+ db,
+ preset["changes"],
+ db_user=db_user,
+ reason=f"preset:{slug}",
+ )
+
+ lines = [f"{preset['title']}", ""]
+ if applied:
+ lines.append(f"✅ Применено настроек: {len(applied)}")
+ if failed:
+ lines.append(f"⚠️ Ошибок: {len(failed)}")
+ for key, error in failed[:5]:
+ lines.append(f"• {key}: {error}")
+ if not failed:
+ lines.append("Все изменения успешно применены.")
+
+ keyboard = types.InlineKeyboardMarkup(
+ inline_keyboard=[
+ [types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config")],
+ [types.InlineKeyboardButton(text="🎛 К пресетам", callback_data="botcfg_presets")],
+ ]
+ )
+
+ await callback.message.edit_text(
+ "\n".join(lines).rstrip(),
+ reply_markup=keyboard,
+ parse_mode="HTML",
+ )
+ await callback.answer("Пресет применен")
+
+
+@admin_required
+@error_handler
+async def start_save_preset(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ await state.set_state(BotConfigStates.waiting_for_preset_name)
+ await state.update_data(botcfg_origin="bot_config_preset")
+
+ await callback.message.answer(
+ "💾 Сохранение пресета\n\nВведите имя пресета. Используйте короткое и понятное название.",
+ parse_mode="HTML",
+ )
+ await callback.answer("Введите название пресета")
+
+
+@admin_required
+@error_handler
+async def handle_preset_name(
+ message: types.Message,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ name = (message.text or "").strip()
+ if not name:
+ await message.answer("Название не может быть пустым.")
+ return
+
+ slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
+ if not slug:
+ await message.answer("Используйте латинские буквы и цифры в названии.")
+ return
+
+ if slug in BUILTIN_PRESETS:
+ await message.answer("Это имя зарезервировано. Выберите другое название.")
+ return
+
+ custom_presets = await _load_custom_presets(db)
+ if slug in custom_presets:
+ await message.answer("Пресет с таким именем уже существует. Укажите другое имя.")
+ return
+
+ snapshot: Dict[str, Any] = {}
+ for definition in _iter_all_definitions():
+ snapshot[definition.key] = bot_configuration_service.get_current_value(
+ definition.key
+ )
+
+ payload = {
+ "title": name,
+ "description": f"Сохранено администратором {db_user.username or db_user.telegram_id}",
+ "changes": snapshot,
+ }
+
+ await upsert_system_setting(
+ db,
+ _preset_storage_key(slug),
+ json.dumps(payload, ensure_ascii=False, default=str),
+ )
+ await db.commit()
+
+ await message.answer(
+ f"✅ Пресет {name} сохранен.",
+ parse_mode="HTML",
+ reply_markup=types.InlineKeyboardMarkup(
+ inline_keyboard=[[types.InlineKeyboardButton(text="🎛 К пресетам", callback_data="botcfg_presets")]]
+ ),
+ )
+ await state.clear()
+
+
+@admin_required
+@error_handler
+async def delete_preset(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ slug = callback.data.split(":", 1)[1] if ":" in callback.data else ""
+ key = _preset_storage_key(slug)
+ await delete_system_setting(db, key)
+ await db.commit()
+
+ await callback.answer("Пресет удален")
+ await show_presets_menu(callback, db_user, db, acknowledge=False)
+
+
+@admin_required
+@error_handler
+async def export_settings_env(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
):
- group_key, category_key, category_page, settings_page = _parse_category_payload(
- callback.data
+ dump = bot_configuration_service.generate_env_dump(include_secrets=False)
+ file = BufferedInputFile(dump.encode("utf-8"), filename="bot-settings.env")
+ caption = (
+ "📤 Экспорт настроек\n"
+ "Секретные значения скрыты. Отправьте файл разработчику или сохраните как бэкап."
)
- definitions = bot_configuration_service.get_settings_for_category(category_key)
+ await callback.message.answer_document(file, caption=caption, parse_mode="HTML")
+ await callback.answer("Файл сформирован")
- if not definitions:
- await callback.answer("В этой категории пока нет настроек", show_alert=True)
+
+@admin_required
+@error_handler
+async def start_import_settings(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ await state.set_state(BotConfigStates.waiting_for_import_payload)
+ await state.update_data(botcfg_origin="bot_config_import")
+
+ lines = [
+ "📥 Импорт настроек",
+ "",
+ "Отправьте содержимое .env файла текстом или прикрепите файл.",
+ "Секретные значения будут применены сразу после проверки.",
+ "Напишите cancel для отмены.",
+ ]
+
+ await callback.message.answer("\n".join(lines), parse_mode="HTML")
+ await callback.answer("Пришлите .env файл")
+
+
+@admin_required
+@error_handler
+async def handle_import_payload(
+ message: types.Message,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ if message.text and message.text.strip().lower() in {"cancel", "отмена"}:
+ await message.answer("Импорт отменен.")
+ await state.clear()
return
- category_label = definitions[0].category_label
- keyboard = _build_settings_keyboard(
- category_key,
- group_key,
- category_page,
- db_user.language,
- settings_page,
+ content: Optional[str] = None
+
+ if message.document:
+ buffer = io.BytesIO()
+ await message.document.download(destination=buffer)
+ try:
+ content = buffer.getvalue().decode("utf-8")
+ except UnicodeDecodeError:
+ await message.answer("Не удалось прочитать файл. Убедитесь, что он в UTF-8.")
+ return
+ elif message.text:
+ content = message.text
+
+ if not content:
+ await message.answer("Пришлите текст или файл с настройками.")
+ return
+
+ try:
+ parsed = bot_configuration_service.parse_env_dump(content)
+ except ValueError as error:
+ await message.answer(f"⚠️ Ошибка: {error}")
+ return
+
+ if not parsed:
+ await message.answer("Файл не содержит подходящих настроек.")
+ return
+
+ applied, failed = await _apply_changeset(
+ db,
+ parsed,
+ db_user=db_user,
+ reason="import",
)
+
+ lines = ["📥 Импорт завершен", ""]
+ if applied:
+ lines.append(f"✅ Обновлено настроек: {len(applied)}")
+ if failed:
+ lines.append(f"⚠️ Ошибок: {len(failed)}")
+ for key, error in failed[:5]:
+ lines.append(f"• {key}: {error}")
+ lines.append("Перейдите в панель, чтобы проверить изменения.")
+
+ await message.answer(
+ "\n".join(lines),
+ parse_mode="HTML",
+ reply_markup=types.InlineKeyboardMarkup(
+ inline_keyboard=[[types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config")]]
+ ),
+ )
+ await state.clear()
+
+
+@admin_required
+@error_handler
+async def show_history(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ history = bot_configuration_service.get_history()
+ lines = ["🕑 История изменений", ""]
+
+ if not history:
+ lines.append("История пока пуста. Измените любую настройку, чтобы увидеть журнал.")
+ else:
+ for entry in history:
+ timestamp: datetime = entry["timestamp"]
+ formatted = timestamp.strftime("%d.%m.%Y %H:%M:%S")
+ actor = entry.get("actor") or "system"
+ reason = entry.get("reason") or "manual"
+ lines.append(
+ f"• {entry['name']} ({entry['key']})\n"
+ f" {formatted} — {actor}\n"
+ f" {entry['old']} → {entry['new']} ({reason})"
+ )
+ lines.append("")
+
+ keyboard = types.InlineKeyboardMarkup(
+ inline_keyboard=[[types.InlineKeyboardButton(text="🏠 Панель", callback_data="admin_bot_config")]]
+ )
+
await callback.message.edit_text(
- f"🧩 {category_label}\n\nВыберите настройку для просмотра:",
+ "\n".join(lines).rstrip(),
reply_markup=keyboard,
+ parse_mode="HTML",
)
await callback.answer()
@@ -607,19 +1489,16 @@ async def test_remnawave_connection(
db_user: User,
db: AsyncSession,
):
- parts = callback.data.split(":", 5)
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
- category_key = parts[2] if len(parts) > 2 else "REMNAWAVE"
-
+ parts = callback.data.split(":", 3)
+ spec_key = parts[1] if len(parts) > 1 else "remnawave"
try:
- category_page = max(1, int(parts[3])) if len(parts) > 3 else 1
+ page = max(1, int(parts[2])) if len(parts) > 2 else 1
except ValueError:
- category_page = 1
+ page = 1
- try:
- settings_page = max(1, int(parts[4])) if len(parts) > 4 else 1
- except ValueError:
- settings_page = 1
+ if spec_key not in SPEC_CATEGORY_MAP:
+ await callback.answer("Раздел недоступен", show_alert=True)
+ return
service = RemnaWaveService()
result = await service.test_api_connection()
@@ -639,20 +1518,11 @@ async def test_remnawave_connection(
else:
message = f"❌ {base_message}"
- definitions = bot_configuration_service.get_settings_for_category(category_key)
- if definitions:
- keyboard = _build_settings_keyboard(
- category_key,
- group_key,
- category_page,
- db_user.language,
- settings_page,
- )
- try:
- await callback.message.edit_reply_markup(reply_markup=keyboard)
- except Exception:
- # ignore inability to refresh markup, main result shown in alert
- pass
+ try:
+ keyboard = _build_spec_category_keyboard(spec_key, page, db_user.language)
+ await callback.message.edit_reply_markup(reply_markup=keyboard)
+ except Exception:
+ pass
await callback.answer(message, show_alert=True)
@@ -664,20 +1534,18 @@ async def test_payment_provider(
db_user: User,
db: AsyncSession,
):
- parts = callback.data.split(":", 6)
+ parts = callback.data.split(":", 4)
method = parts[1] if len(parts) > 1 else ""
- group_key = parts[2] if len(parts) > 2 else CATEGORY_FALLBACK_KEY
- category_key = parts[3] if len(parts) > 3 else "PAYMENT"
+ spec_key = parts[2] if len(parts) > 2 else "payments"
try:
- category_page = max(1, int(parts[4])) if len(parts) > 4 else 1
+ page = max(1, int(parts[3])) if len(parts) > 3 else 1
except ValueError:
- category_page = 1
+ page = 1
- try:
- settings_page = max(1, int(parts[5])) if len(parts) > 5 else 1
- except ValueError:
- settings_page = 1
+ if spec_key not in SPEC_CATEGORY_MAP:
+ await callback.answer("Раздел недоступен", show_alert=True)
+ return
language = db_user.language
texts = get_texts(language)
@@ -686,19 +1554,11 @@ async def test_payment_provider(
message_text: str
async def _refresh_markup() -> None:
- definitions = bot_configuration_service.get_settings_for_category(category_key)
- if definitions:
- keyboard = _build_settings_keyboard(
- category_key,
- group_key,
- category_page,
- language,
- settings_page,
- )
- try:
- await callback.message.edit_reply_markup(reply_markup=keyboard)
- except Exception:
- pass
+ keyboard = _build_spec_category_keyboard(spec_key, page, language)
+ try:
+ await callback.message.edit_reply_markup(reply_markup=keyboard)
+ except Exception:
+ pass
if method == "yookassa":
if not settings.is_yookassa_enabled():
@@ -1053,7 +1913,7 @@ async def show_bot_config_setting(
state: FSMContext,
):
parts = callback.data.split(":", 4)
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
+ spec_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
try:
category_page = max(1, int(parts[2])) if len(parts) > 2 else 1
except ValueError:
@@ -1068,19 +1928,103 @@ async def show_bot_config_setting(
except KeyError:
await callback.answer("Эта настройка больше недоступна", show_alert=True)
return
- text = _render_setting_text(key)
- keyboard = _build_setting_keyboard(key, group_key, category_page, settings_page)
- await callback.message.edit_text(text, reply_markup=keyboard)
+ category = SPEC_CATEGORY_MAP.get(spec_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
+ text = _render_setting_text(key, category)
+ keyboard = _build_setting_keyboard(key, spec_key, category_page, settings_page)
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await _store_setting_context(
state,
key=key,
- group_key=group_key,
+ group_key=spec_key,
category_page=category_page,
settings_page=settings_page,
)
await callback.answer()
+@admin_required
+@error_handler
+async def show_setting_info(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ parts = callback.data.split(":", 4)
+ spec_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
+ try:
+ category_page = max(1, int(parts[2])) if len(parts) > 2 else 1
+ except ValueError:
+ category_page = 1
+ try:
+ settings_page = max(1, int(parts[3])) if len(parts) > 3 else 1
+ except ValueError:
+ settings_page = 1
+ token = parts[4] if len(parts) > 4 else ""
+ try:
+ key = bot_configuration_service.resolve_callback_token(token)
+ except KeyError:
+ await callback.answer("Настройка больше недоступна", show_alert=True)
+ return
+
+ definition = bot_configuration_service.get_definition(key)
+ metadata = bot_configuration_service.get_metadata(key)
+ summary = bot_configuration_service.get_setting_summary(key)
+ spec_category = SPEC_CATEGORY_MAP.get(spec_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
+
+ lines = [
+ f"ℹ️ Подробности: {definition.display_name}",
+ f"{spec_category.icon} Раздел: {spec_category.title}",
+ f"🔑 Ключ: {key}",
+ f"📦 Тип: {summary['type']}",
+ f"📘 Текущее: {summary['current']}",
+ f"📗 По умолчанию: {summary['original']}",
+ ]
+
+ if metadata.description:
+ lines.append("")
+ lines.append(f"📝 {metadata.description}")
+ if metadata.format_hint:
+ lines.append(f"📐 Формат: {metadata.format_hint}")
+ if metadata.example:
+ lines.append(f"💡 Пример: {metadata.example}")
+ if metadata.warning:
+ lines.append(f"⚠️ Важно: {metadata.warning}")
+ if metadata.dependencies:
+ lines.append(f"🔗 Связано: {metadata.dependencies}")
+ if metadata.recommended is not None:
+ recommended = bot_configuration_service.format_setting_value(key, metadata.recommended)
+ lines.append(f"✨ Рекомендуемое: {recommended}")
+
+ keyboard = types.InlineKeyboardMarkup(
+ inline_keyboard=[
+ [
+ types.InlineKeyboardButton(
+ text="⚙️ Открыть настройку",
+ callback_data=f"botcfg_setting:{spec_key}:{category_page}:{settings_page}:{token}",
+ )
+ ],
+ [
+ types.InlineKeyboardButton(
+ text="⬅️ К списку",
+ callback_data=f"botcfg_group:{spec_key}:{category_page}",
+ ),
+ types.InlineKeyboardButton(
+ text="🏠 Панель",
+ callback_data="admin_bot_config",
+ ),
+ ],
+ ]
+ )
+
+ await callback.message.edit_text(
+ "\n".join(lines),
+ reply_markup=keyboard,
+ parse_mode="HTML",
+ )
+ await callback.answer()
+
+
@admin_required
@error_handler
async def start_edit_setting(
@@ -1090,7 +2034,7 @@ async def start_edit_setting(
state: FSMContext,
):
parts = callback.data.split(":", 4)
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
+ spec_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
try:
category_page = max(1, int(parts[2])) if len(parts) > 2 else 1
except ValueError:
@@ -1109,19 +2053,29 @@ async def start_edit_setting(
summary = bot_configuration_service.get_setting_summary(key)
texts = get_texts(db_user.language)
+ metadata = bot_configuration_service.get_metadata(key)
+ spec_category = SPEC_CATEGORY_MAP.get(spec_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
instructions = [
- "✏️ Редактирование настройки",
- f"Название: {summary['name']}",
+ f"✏️ Редактирование: {summary['name']}",
+ f"Раздел: {spec_category.icon} {spec_category.title}",
f"Ключ: {summary['key']}",
f"Тип: {summary['type']}",
f"Текущее значение: {summary['current']}",
- "\nОтправьте новое значение сообщением.",
+ "",
+ "Отправьте новое значение сообщением.",
]
if definition.is_optional:
instructions.append("Отправьте 'none' или оставьте пустым для сброса на значение по умолчанию.")
+ if metadata.format_hint:
+ instructions.append(f"Формат: {metadata.format_hint}")
+ if metadata.example:
+ instructions.append(f"Пример: {metadata.example}")
+ if metadata.warning:
+ instructions.append(f"Важно: {metadata.warning}")
+
instructions.append("Для отмены отправьте 'cancel'.")
await callback.message.edit_text(
@@ -1132,18 +2086,19 @@ async def start_edit_setting(
types.InlineKeyboardButton(
text=texts.BACK,
callback_data=(
- f"botcfg_setting:{group_key}:{category_page}:{settings_page}:{token}"
+ f"botcfg_setting:{spec_key}:{category_page}:{settings_page}:{token}"
),
)
]
]
),
+ parse_mode="HTML",
)
await _store_setting_context(
state,
key=key,
- group_key=group_key,
+ group_key=spec_key,
category_page=category_page,
settings_page=settings_page,
)
@@ -1161,7 +2116,8 @@ async def handle_edit_setting(
):
data = await state.get_data()
key = data.get("setting_key")
- group_key = data.get("setting_group_key", CATEGORY_FALLBACK_KEY)
+ group_key = data.get("setting_group_key", DEFAULT_SPEC_KEY)
+ spec_category = SPEC_CATEGORY_MAP.get(group_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
category_page = data.get("setting_category_page", 1)
settings_page = data.get("setting_settings_page", 1)
@@ -1179,10 +2135,10 @@ async def handle_edit_setting(
await bot_configuration_service.set_value(db, key, value)
await db.commit()
- text = _render_setting_text(key)
+ text = _render_setting_text(key, spec_category)
keyboard = _build_setting_keyboard(key, group_key, category_page, settings_page)
await message.answer("✅ Настройка обновлена")
- await message.answer(text, reply_markup=keyboard)
+ await message.answer(text, reply_markup=keyboard, parse_mode="HTML")
await state.clear()
await _store_setting_context(
state,
@@ -1204,7 +2160,8 @@ async def handle_direct_setting_input(
data = await state.get_data()
key = data.get("setting_key")
- group_key = data.get("setting_group_key", CATEGORY_FALLBACK_KEY)
+ group_key = data.get("setting_group_key", DEFAULT_SPEC_KEY)
+ spec_category = SPEC_CATEGORY_MAP.get(group_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
category_page = int(data.get("setting_category_page", 1) or 1)
settings_page = int(data.get("setting_settings_page", 1) or 1)
@@ -1220,10 +2177,10 @@ async def handle_direct_setting_input(
await bot_configuration_service.set_value(db, key, value)
await db.commit()
- text = _render_setting_text(key)
+ text = _render_setting_text(key, spec_category)
keyboard = _build_setting_keyboard(key, group_key, category_page, settings_page)
await message.answer("✅ Настройка обновлена")
- await message.answer(text, reply_markup=keyboard)
+ await message.answer(text, reply_markup=keyboard, parse_mode="HTML")
await state.clear()
await _store_setting_context(
@@ -1244,7 +2201,8 @@ async def reset_setting(
state: FSMContext,
):
parts = callback.data.split(":", 4)
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
+ group_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
+ spec_category = SPEC_CATEGORY_MAP.get(group_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
try:
category_page = max(1, int(parts[2])) if len(parts) > 2 else 1
except ValueError:
@@ -1262,9 +2220,9 @@ async def reset_setting(
await bot_configuration_service.reset_value(db, key)
await db.commit()
- text = _render_setting_text(key)
+ text = _render_setting_text(key, spec_category)
keyboard = _build_setting_keyboard(key, group_key, category_page, settings_page)
- await callback.message.edit_text(text, reply_markup=keyboard)
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await _store_setting_context(
state,
key=key,
@@ -1284,7 +2242,8 @@ async def toggle_setting(
state: FSMContext,
):
parts = callback.data.split(":", 4)
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
+ group_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
+ spec_category = SPEC_CATEGORY_MAP.get(group_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
try:
category_page = max(1, int(parts[2])) if len(parts) > 2 else 1
except ValueError:
@@ -1304,9 +2263,9 @@ async def toggle_setting(
await bot_configuration_service.set_value(db, key, new_value)
await db.commit()
- text = _render_setting_text(key)
+ text = _render_setting_text(key, spec_category)
keyboard = _build_setting_keyboard(key, group_key, category_page, settings_page)
- await callback.message.edit_text(text, reply_markup=keyboard)
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await _store_setting_context(
state,
key=key,
@@ -1326,7 +2285,8 @@ async def apply_setting_choice(
state: FSMContext,
):
parts = callback.data.split(":", 5)
- group_key = parts[1] if len(parts) > 1 else CATEGORY_FALLBACK_KEY
+ group_key = parts[1] if len(parts) > 1 else DEFAULT_SPEC_KEY
+ spec_category = SPEC_CATEGORY_MAP.get(group_key, SPEC_CATEGORY_MAP[DEFAULT_SPEC_KEY])
try:
category_page = max(1, int(parts[2])) if len(parts) > 2 else 1
except ValueError:
@@ -1353,9 +2313,9 @@ async def apply_setting_choice(
await bot_configuration_service.set_value(db, key, value)
await db.commit()
- text = _render_setting_text(key)
+ text = _render_setting_text(key, spec_category)
keyboard = _build_setting_keyboard(key, group_key, category_page, settings_page)
- await callback.message.edit_text(text, reply_markup=keyboard)
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await _store_setting_context(
state,
key=key,
@@ -1376,8 +2336,12 @@ def register_handlers(dp: Dispatcher) -> None:
F.data.startswith("botcfg_group:") & (~F.data.endswith(":noop")),
)
dp.callback_query.register(
- show_bot_config_category,
- F.data.startswith("botcfg_cat:"),
+ start_search_workflow,
+ F.data == "botcfg_search:start",
+ )
+ dp.callback_query.register(
+ cancel_search,
+ F.data == "botcfg_search:cancel",
)
dp.callback_query.register(
test_remnawave_connection,
@@ -1391,6 +2355,10 @@ def register_handlers(dp: Dispatcher) -> None:
show_bot_config_setting,
F.data.startswith("botcfg_setting:"),
)
+ dp.callback_query.register(
+ show_setting_info,
+ F.data.startswith("botcfg_info:"),
+ )
dp.callback_query.register(
start_edit_setting,
F.data.startswith("botcfg_edit:"),
@@ -1407,6 +2375,54 @@ def register_handlers(dp: Dispatcher) -> None:
apply_setting_choice,
F.data.startswith("botcfg_choice:"),
)
+ dp.callback_query.register(
+ show_quick_actions_menu,
+ F.data == "botcfg_quick_menu",
+ )
+ dp.callback_query.register(
+ apply_quick_action,
+ F.data.startswith("botcfg_quick:"),
+ )
+ dp.callback_query.register(
+ show_presets_menu,
+ F.data == "botcfg_presets",
+ )
+ dp.callback_query.register(
+ apply_preset,
+ F.data.startswith("botcfg_preset_apply:"),
+ )
+ dp.callback_query.register(
+ start_save_preset,
+ F.data == "botcfg_preset_save",
+ )
+ dp.callback_query.register(
+ delete_preset,
+ F.data.startswith("botcfg_preset_delete:"),
+ )
+ dp.callback_query.register(
+ export_settings_env,
+ F.data == "botcfg_export",
+ )
+ dp.callback_query.register(
+ start_import_settings,
+ F.data == "botcfg_import",
+ )
+ dp.callback_query.register(
+ show_history,
+ F.data == "botcfg_history",
+ )
+ dp.message.register(
+ handle_search_query,
+ BotConfigStates.waiting_for_search_query,
+ )
+ dp.message.register(
+ handle_import_payload,
+ BotConfigStates.waiting_for_import_payload,
+ )
+ dp.message.register(
+ handle_preset_name,
+ BotConfigStates.waiting_for_preset_name,
+ )
dp.message.register(
handle_direct_setting_input,
StateFilter(None),
diff --git a/app/services/system_settings_service.py b/app/services/system_settings_service.py
index 1ce9432a..2348c41a 100644
--- a/app/services/system_settings_service.py
+++ b/app/services/system_settings_service.py
@@ -1,8 +1,12 @@
import hashlib
import json
import logging
-from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Tuple, Type, Union, get_args, get_origin
+import re
+from collections import deque
+from dataclasses import dataclass, field
+from datetime import datetime
+from decimal import Decimal, InvalidOperation
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union, get_args, get_origin
from app.database.universal_migration import ensure_default_web_api_token
@@ -35,6 +39,23 @@ def _truncate(value: str, max_len: int = 60) -> str:
return value[: max_len - 1] + "…"
+@dataclass(slots=True)
+class SettingMetadata:
+ display_name: Optional[str] = None
+ description: Optional[str] = None
+ format_hint: Optional[str] = None
+ example: Optional[str] = None
+ warning: Optional[str] = None
+ dependencies: Optional[str] = None
+ icon: Optional[str] = None
+ input_type: Optional[str] = None
+ unit: Optional[str] = None
+ recommended: Optional[Any] = None
+ tags: Tuple[str, ...] = field(default_factory=tuple)
+ secret: Optional[bool] = None
+ category_description: Optional[str] = None
+
+
@dataclass(slots=True)
class SettingDefinition:
key: str
@@ -43,10 +64,16 @@ class SettingDefinition:
python_type: Type[Any]
type_label: str
is_optional: bool
+ display_name_override: Optional[str] = None
+ icon_override: Optional[str] = None
@property
def display_name(self) -> str:
- return _title_from_key(self.key)
+ return self.display_name_override or _title_from_key(self.key)
+
+ @property
+ def icon(self) -> str:
+ return self.icon_override or "⚙️"
@dataclass(slots=True)
@@ -203,6 +230,318 @@ class BotConfigurationService:
"DEBUG": "DEBUG",
}
+ CATEGORY_DESCRIPTIONS: Dict[str, str] = {
+ "SUPPORT": "Настройки поддержки, тикетов и SLA.",
+ "LOCALIZATION": "Языки и тексты интерфейса.",
+ "MAINTENANCE": "Режим технических работ и сообщение для пользователей.",
+ "CHANNEL": "Обязательная подписка на канал и переходы.",
+ "ADMIN_NOTIFICATIONS": "Мгновенные уведомления администраторам.",
+ "ADMIN_REPORTS": "Регулярные отчеты и сводки.",
+ "TRIAL": "Параметры пробного периода.",
+ "PAID_SUBSCRIPTION": "Базовые лимиты платных подписок.",
+ "PERIODS": "Доступные периоды подписки.",
+ "SUBSCRIPTION_PRICES": "Цены на подписки по периодам.",
+ "TRAFFIC": "Лимиты и сброс трафика.",
+ "TRAFFIC_PACKAGES": "Пакеты дополнительного трафика и их стоимость.",
+ "DISCOUNTS": "Скидки и промо-настройки.",
+ "PAYMENT": "Общие платежные параметры.",
+ "TELEGRAM": "Telegram Stars и покупки в приложении.",
+ "CRYPTOBOT": "Оплата через CryptoBot.",
+ "YOOKASSA": "Интеграция YooKassa.",
+ "TRIBUTE": "Интеграция Tribute.",
+ "MULENPAY": "Интеграция MulenPay.",
+ "PAL24": "Интеграция PayPalych (PAL24).",
+ "REMNAWAVE": "Соединение с RemnaWave API.",
+ "REFERRAL": "Реферальная программа и бонусы.",
+ "AUTOPAY": "Автопродление подписок.",
+ "INTERFACE_BRANDING": "Логотип и брендовые элементы.",
+ "INTERFACE_SUBSCRIPTION": "Ссылка на подписку в интерфейсе.",
+ "CONNECT_BUTTON": "Действие кнопки «Подключиться».",
+ "HAPP": "Интеграция Happ и CryptoLink.",
+ "SKIP": "Опции быстрого старта и пропуска шагов.",
+ "ADDITIONAL": "Дополнительные конфигурации приложения.",
+ "MINIAPP": "Настройки мини-приложения Telegram.",
+ "DATABASE": "Выбор и подключение базы данных.",
+ "POSTGRES": "Параметры PostgreSQL.",
+ "SQLITE": "Путь и параметры SQLite.",
+ "REDIS": "Настройки кеша Redis.",
+ "MONITORING": "Мониторинг и хранение логов.",
+ "NOTIFICATIONS": "Пользовательские уведомления и SLA.",
+ "SERVER": "Проверка статуса серверов.",
+ "BACKUP": "Резервное копирование.",
+ "VERSION": "Проверка обновлений.",
+ "LOG": "Логирование и файлы.",
+ "WEBHOOK": "Webhook Telegram.",
+ "WEB_API": "Встроенный Web API.",
+ "DEBUG": "Режимы разработки и отладки.",
+ }
+
+ METADATA_KEY_OVERRIDES: Dict[str, SettingMetadata] = {
+ "MAINTENANCE_MODE": SettingMetadata(
+ display_name="Режим обслуживания",
+ description="Переводит бота в режим технических работ и скрывает основные функции.",
+ format_hint="Используйте кнопки включения/выключения или ответьте 'вкл/выкл'.",
+ example="вкл",
+ warning="Пользователи не смогут использовать бота, пока режим активен.",
+ dependencies="MAINTENANCE_MESSAGE",
+ icon="🔧",
+ input_type="toggle",
+ recommended=False,
+ ),
+ "MAINTENANCE_MESSAGE": SettingMetadata(
+ description="Текст, который увидит пользователь во время технических работ.",
+ format_hint="Обычный текст, поддерживается базовое форматирование Telegram.",
+ example="🔧 Ведутся технические работы...",
+ dependencies="MAINTENANCE_MODE",
+ icon="💬",
+ ),
+ "DEBUG": SettingMetadata(
+ display_name="Режим отладки",
+ description="Включает подробные логи для разработчиков.",
+ warning="Не держите включенным в продакшене — возможна утечка чувствительных данных.",
+ icon="🐞",
+ input_type="toggle",
+ recommended=False,
+ ),
+ "ENABLE_NOTIFICATIONS": SettingMetadata(
+ description="Включает отправку уведомлений пользователям о подписках, триалах и лимитах.",
+ dependencies="NOTIFICATION_RETRY_ATTEMPTS, NOTIFICATION_CACHE_HOURS",
+ icon="🔔",
+ input_type="toggle",
+ recommended=True,
+ ),
+ "ADMIN_NOTIFICATIONS_ENABLED": SettingMetadata(
+ description="Рассылка мгновенных уведомлений администраторам о важных событиях.",
+ dependencies="ADMIN_NOTIFICATIONS_CHAT_ID",
+ icon="📣",
+ input_type="toggle",
+ ),
+ "ADMIN_REPORTS_SEND_TIME": SettingMetadata(
+ description="Время отправки ежедневных отчетов администраторам.",
+ format_hint="Введите время в формате ЧЧ:ММ.",
+ example="09:00",
+ input_type="time",
+ icon="🕒",
+ ),
+ "AVAILABLE_SUBSCRIPTION_PERIODS": SettingMetadata(
+ description="Список доступных периодов подписки в днях.",
+ format_hint="Перечислите значения через запятую.",
+ example="30,90,180",
+ input_type="list",
+ icon="📅",
+ ),
+ "BASE_SUBSCRIPTION_PRICE": SettingMetadata(
+ description="Базовая стоимость подписки в копейках.",
+ format_hint="Введите цену в рублях, например 99 000.",
+ example="99 000",
+ input_type="price",
+ unit="₽",
+ icon="💰",
+ ),
+ "TRIAL_DURATION_DAYS": SettingMetadata(
+ description="Количество дней пробного периода.",
+ example="3",
+ unit="дней",
+ icon="🎁",
+ recommended=3,
+ ),
+ "YOOKASSA_ENABLED": SettingMetadata(
+ description="Включает прием платежей через YooKassa.",
+ warning="Не активируйте без действующих ключей магазина.",
+ dependencies="YOOKASSA_SHOP_ID, YOOKASSA_SECRET_KEY",
+ icon="💸",
+ input_type="toggle",
+ ),
+ "CRYPTOBOT_ENABLED": SettingMetadata(
+ description="Разрешает оплату через CryptoBot.",
+ dependencies="CRYPTOBOT_API_TOKEN",
+ icon="🪙",
+ input_type="toggle",
+ ),
+ "REMNAWAVE_API_URL": SettingMetadata(
+ description="Базовый URL панели RemnaWave.",
+ format_hint="Полный адрес, например https://panel.remnawave.com",
+ example="https://panel.remnawave.com",
+ icon="🌐",
+ ),
+ "DATABASE_MODE": SettingMetadata(
+ description="Выбор источника данных: auto, sqlite или postgresql.",
+ format_hint="Введите одно из значений auto/sqlite/postgresql.",
+ example="postgresql",
+ icon="💾",
+ ),
+ "REFERRAL_COMMISSION_PERCENT": SettingMetadata(
+ description="Процент комиссии для пригласившего пользователя.",
+ unit="%",
+ example="25",
+ icon="👥",
+ ),
+ "BACKUP_TIME": SettingMetadata(
+ description="Время запуска автоматического бэкапа.",
+ format_hint="ЧЧ:ММ, 24-часовой формат.",
+ example="03:00",
+ input_type="time",
+ icon="💾",
+ ),
+ "WEB_API_ENABLED": SettingMetadata(
+ description="Включает встроенный Web API для интеграций.",
+ warning="Убедитесь, что токены доступа настроены и хранятся безопасно.",
+ icon="🌐",
+ input_type="toggle",
+ ),
+ "ENABLE_DEEP_LINKS": SettingMetadata(
+ description="Позволяет открывать бота через глубокие ссылки.",
+ warning="Отключение сделает недоступными промо-ссылки и мини-приложение.",
+ icon="🔗",
+ input_type="toggle",
+ ),
+ }
+
+ METADATA_PREFIX_HINTS: Tuple[Tuple[str, SettingMetadata], ...] = (
+ (
+ "PRICE_",
+ SettingMetadata(
+ icon="💰",
+ input_type="price",
+ format_hint="Введите цену в рублях, разделяя тысячи пробелами.",
+ example="9 990",
+ unit="₽",
+ ),
+ ),
+ (
+ "YOOKASSA_",
+ SettingMetadata(
+ icon="💸",
+ category_description="Настройка платежей через YooKassa.",
+ ),
+ ),
+ (
+ "CRYPTOBOT_",
+ SettingMetadata(icon="🪙", category_description="Интеграция с CryptoBot."),
+ ),
+ (
+ "MULENPAY_",
+ SettingMetadata(icon="💳", category_description="Интеграция MulenPay."),
+ ),
+ (
+ "PAL24_",
+ SettingMetadata(icon="🏦", category_description="Интеграция PayPalych (PAL24)."),
+ ),
+ (
+ "TRIBUTE_",
+ SettingMetadata(icon="🎁", category_description="Настройки Tribute."),
+ ),
+ (
+ "TELEGRAM_STARS",
+ SettingMetadata(icon="⭐", category_description="Платежи через Telegram Stars."),
+ ),
+ (
+ "TRIAL_",
+ SettingMetadata(icon="🎁", category_description="Параметры пробного периода."),
+ ),
+ (
+ "REFERRAL_",
+ SettingMetadata(icon="👥", category_description="Реферальная программа."),
+ ),
+ (
+ "BACKUP_",
+ SettingMetadata(icon="💾", category_description="Автоматические резервные копии."),
+ ),
+ )
+
+ METADATA_SUFFIX_HINTS: Tuple[Tuple[str, SettingMetadata], ...] = (
+ (
+ "_ENABLED",
+ SettingMetadata(
+ input_type="toggle",
+ format_hint="Используйте кнопки или отправьте 'вкл'/'выкл'.",
+ example="вкл",
+ ),
+ ),
+ (
+ "_IDS",
+ SettingMetadata(
+ input_type="list",
+ format_hint="Перечислите значения через запятую.",
+ example="123456789,987654321",
+ ),
+ ),
+ (
+ "_PERCENT",
+ SettingMetadata(
+ input_type="number",
+ unit="%",
+ format_hint="Введите целое число от 0 до 100.",
+ example="25",
+ ),
+ ),
+ (
+ "_KOPEKS",
+ SettingMetadata(
+ input_type="price",
+ unit="₽",
+ format_hint="Введите цену в рублях — бот сконвертирует в копейки.",
+ example="500",
+ ),
+ ),
+ (
+ "_HOURS",
+ SettingMetadata(
+ unit="часов",
+ input_type="number",
+ example="24",
+ ),
+ ),
+ (
+ "_MINUTES",
+ SettingMetadata(
+ unit="минут",
+ input_type="number",
+ example="15",
+ ),
+ ),
+ (
+ "_SECONDS",
+ SettingMetadata(
+ unit="секунд",
+ input_type="number",
+ example="60",
+ ),
+ ),
+ (
+ "_DAYS",
+ SettingMetadata(
+ unit="дней",
+ input_type="number",
+ example="30",
+ ),
+ ),
+ (
+ "_TIME",
+ SettingMetadata(
+ input_type="time",
+ format_hint="Введите время в формате ЧЧ:ММ.",
+ example="12:30",
+ ),
+ ),
+ (
+ "_URL",
+ SettingMetadata(
+ input_type="text",
+ format_hint="Полный URL, начинающийся с http или https.",
+ example="https://example.com",
+ ),
+ ),
+ )
+
+ SECRET_KEY_PATTERNS: Tuple[str, ...] = (
+ "SECRET",
+ "TOKEN",
+ "PASSWORD",
+ "API_KEY",
+ "PRIVATE_KEY",
+ )
+
CHOICES: Dict[str, List[ChoiceOption]] = {
"DATABASE_MODE": [
ChoiceOption("auto", "🤖 Авто"),
@@ -297,6 +636,8 @@ class BotConfigurationService:
_token_to_key: Dict[str, str] = {}
_choice_tokens: Dict[str, Dict[Any, str]] = {}
_choice_token_lookup: Dict[str, Dict[str, Any]] = {}
+ _metadata_cache: Dict[str, SettingMetadata] = {}
+ _history: deque[Dict[str, Any]] = deque(maxlen=10)
@classmethod
def initialize_definitions(cls) -> None:
@@ -317,7 +658,7 @@ class BotConfigurationService:
category_key.capitalize() if category_key else "Прочее",
)
- cls._definitions[key] = SettingDefinition(
+ definition = SettingDefinition(
key=key,
category_key=category_key or "other",
category_label=category_label,
@@ -326,12 +667,349 @@ class BotConfigurationService:
is_optional=is_optional,
)
+ metadata = cls._build_metadata(definition)
+ if metadata.display_name:
+ definition.display_name_override = metadata.display_name
+ if metadata.icon:
+ definition.icon_override = metadata.icon
+
+ cls._definitions[key] = definition
+ cls._metadata_cache[key] = metadata
+
cls._register_callback_token(key)
if key in cls.CHOICES:
cls._ensure_choice_tokens(key)
@classmethod
+ def _build_metadata(cls, definition: SettingDefinition) -> SettingMetadata:
+ key = definition.key
+ base_metadata = SettingMetadata(
+ icon=cls._extract_category_icon(definition.category_label),
+ category_description=cls.CATEGORY_DESCRIPTIONS.get(definition.category_key),
+ )
+
+ metadata = cls._merge_metadata(base_metadata, cls._metadata_for_python_type(definition))
+
+ for prefix, hint in cls.METADATA_PREFIX_HINTS:
+ if key.startswith(prefix):
+ metadata = cls._merge_metadata(metadata, hint)
+
+ for suffix, hint in cls.METADATA_SUFFIX_HINTS:
+ if key.endswith(suffix):
+ metadata = cls._merge_metadata(metadata, hint)
+
+ key_override = cls.METADATA_KEY_OVERRIDES.get(key)
+ if key_override:
+ metadata = cls._merge_metadata(metadata, key_override)
+
+ if metadata.display_name is None:
+ metadata.display_name = cls._guess_display_name(key)
+
+ if metadata.description is None:
+ metadata.description = cls._default_description(definition)
+
+ if metadata.input_type is None:
+ metadata.input_type = cls._default_input_type(definition)
+
+ if metadata.format_hint is None:
+ metadata.format_hint = cls._default_format_hint(metadata)
+
+ if metadata.example is None:
+ metadata.example = cls._default_example(metadata)
+
+ if metadata.secret is None and cls._is_secret_key(key):
+ metadata.secret = True
+
+ return metadata
+
+ @classmethod
+ def _metadata_for_python_type(cls, definition: SettingDefinition) -> SettingMetadata:
+ python_type = definition.python_type
+ if python_type is bool:
+ return SettingMetadata(
+ input_type="toggle",
+ format_hint="Используйте кнопки включения/выключения или ответьте 'вкл'/'выкл'.",
+ example="вкл",
+ )
+ if python_type is int:
+ return SettingMetadata(
+ input_type="number",
+ format_hint="Введите целое число.",
+ example="10",
+ )
+ if python_type is float:
+ return SettingMetadata(
+ input_type="number",
+ format_hint="Введите число, можно использовать запятую.",
+ example="1,5",
+ )
+ return SettingMetadata(
+ input_type="text",
+ format_hint="Введите текстовое значение.",
+ example="Пример",
+ )
+
+ @staticmethod
+ def _merge_metadata(base: SettingMetadata, override: SettingMetadata) -> SettingMetadata:
+ if override is base:
+ return base
+
+ merged = SettingMetadata(
+ display_name=override.display_name or base.display_name,
+ description=override.description or base.description,
+ format_hint=override.format_hint or base.format_hint,
+ example=override.example or base.example,
+ warning=override.warning or base.warning,
+ dependencies=override.dependencies or base.dependencies,
+ icon=override.icon or base.icon,
+ input_type=override.input_type or base.input_type,
+ unit=override.unit or base.unit,
+ recommended=override.recommended if override.recommended is not None else base.recommended,
+ secret=override.secret if override.secret is not None else base.secret,
+ category_description=override.category_description or base.category_description,
+ )
+
+ if base.tags or override.tags:
+ tags: List[str] = list(base.tags)
+ for tag in override.tags:
+ if tag not in tags:
+ tags.append(tag)
+ merged.tags = tuple(tags)
+
+ return merged
+
+ @staticmethod
+ def _extract_category_icon(category_label: str) -> Optional[str]:
+ if not category_label:
+ return None
+ stripped = category_label.strip()
+ if not stripped:
+ return None
+ first_char = stripped[0]
+ if first_char.isascii():
+ return None
+ return first_char
+
+ @staticmethod
+ def _guess_display_name(key: str) -> Optional[str]:
+ if key.endswith("_ENABLED"):
+ base = key[:-8]
+ return _title_from_key(base)
+ if key.endswith("_URL"):
+ base = key[:-4]
+ return f"{_title_from_key(base)} URL"
+ if key.endswith("_ID"):
+ base = key[:-3]
+ return f"{_title_from_key(base)} ID"
+ if key.endswith("_TIME"):
+ base = key[:-5]
+ return f"{_title_from_key(base)} Время"
+ return _title_from_key(key)
+
+ @staticmethod
+ def _default_description(definition: SettingDefinition) -> str:
+ return (
+ f"Настройка «{definition.display_name}» в категории "
+ f"{definition.category_label}."
+ )
+
+ @staticmethod
+ def _default_input_type(definition: SettingDefinition) -> str:
+ if definition.python_type is bool:
+ return "toggle"
+ if definition.python_type in {int, float}:
+ return "number"
+ return "text"
+
+ @staticmethod
+ def _default_format_hint(metadata: SettingMetadata) -> str:
+ mapping = {
+ "toggle": "Используйте кнопки включения/выключения.",
+ "number": "Введите числовое значение.",
+ "price": "Введите сумму в рублях.",
+ "time": "Введите время в формате ЧЧ:ММ.",
+ "list": "Перечислите значения через запятую.",
+ "text": "Введите текстовое значение.",
+ }
+ return mapping.get(metadata.input_type or "text", "Введите значение и отправьте сообщением.")
+
+ @staticmethod
+ def _default_example(metadata: SettingMetadata) -> str:
+ mapping = {
+ "toggle": "вкл",
+ "number": "10",
+ "price": "9 990",
+ "time": "12:00",
+ "list": "значение1, значение2",
+ "text": "пример",
+ }
+ return mapping.get(metadata.input_type or "text", "пример")
+
+ @classmethod
+ def get_metadata(cls, key: str) -> SettingMetadata:
+ cls.initialize_definitions()
+ metadata = cls._metadata_cache.get(key)
+ if metadata is None:
+ definition = cls._definitions[key]
+ metadata = cls._build_metadata(definition)
+ cls._metadata_cache[key] = metadata
+ return metadata
+
+ @classmethod
+ def _is_secret_key(cls, key: str) -> bool:
+ upper = key.upper()
+ return any(pattern in upper for pattern in cls.SECRET_KEY_PATTERNS)
+
+ @staticmethod
+ def _mask_secret(value: Any) -> str:
+ text = str(value or "")
+ if not text:
+ return "—"
+ if len(text) <= 4:
+ return "••••"
+ return "••••••••" + text[-4:]
+
+ @staticmethod
+ def _format_rubles(raw_value: Any) -> str:
+ try:
+ dec_value = Decimal(str(raw_value))
+ except InvalidOperation:
+ return str(raw_value)
+
+ rubles = dec_value / Decimal(100)
+ quantized = rubles.quantize(Decimal("0.01"))
+
+ if quantized == quantized.to_integral_value():
+ integer = int(quantized)
+ formatted = f"{integer:,}".replace(",", " ")
+ else:
+ formatted = f"{quantized:.2f}".replace(",", " ")
+
+ return f"{formatted} ₽"
+
+ @staticmethod
+ def _parse_time(text: str) -> str:
+ if not re.fullmatch(r"\d{1,2}:\d{2}", text):
+ raise ValueError("Используйте формат ЧЧ:ММ")
+ hours_str, minutes_str = text.split(":", 1)
+ hours = int(hours_str)
+ minutes = int(minutes_str)
+ if hours < 0 or hours > 23 or minutes < 0 or minutes > 59:
+ raise ValueError("Часы должны быть 0-23, минуты 0-59")
+ return f"{hours:02d}:{minutes:02d}"
+
+ @staticmethod
+ def _parse_price(text: str) -> int:
+ normalized = text.replace(" ", "").replace("₽", "").replace(",", ".")
+ if not normalized:
+ raise ValueError("Введите сумму в рублях")
+ try:
+ value = Decimal(normalized)
+ except InvalidOperation as error:
+ raise ValueError("Некорректное значение цены") from error
+ if value < 0:
+ raise ValueError("Цена не может быть отрицательной")
+ kopeks = (value * 100).quantize(Decimal("1"))
+ return int(kopeks)
+
+ @staticmethod
+ def _parse_list(text: str) -> str:
+ if not text:
+ return ""
+ normalized = text.replace("\n", ",")
+ items = [item.strip() for item in normalized.split(",") if item.strip()]
+ return ",".join(items)
+
+ @classmethod
+ def format_setting_value(
+ cls,
+ key: str,
+ value: Any,
+ *,
+ include_unit: bool = True,
+ mask_secrets: bool = True,
+ ) -> str:
+ metadata = cls.get_metadata(key)
+ definition = cls.get_definition(key)
+
+ if value is None or value == "":
+ return "—"
+
+ if mask_secrets and (metadata.secret or cls._is_secret_key(key)):
+ return cls._mask_secret(value)
+
+ input_type = metadata.input_type or cls._default_input_type(definition)
+ unit = metadata.unit if include_unit else None
+
+ if input_type == "toggle":
+ return "ВКЛЮЧЕНО" if bool(value) else "ВЫКЛЮЧЕНО"
+
+ if input_type == "price":
+ return cls._format_rubles(value)
+
+ if input_type == "list":
+ if isinstance(value, str):
+ items = [item.strip() for item in value.split(",") if item.strip()]
+ elif isinstance(value, Iterable) and not isinstance(value, (str, bytes)):
+ items = [str(item).strip() for item in value]
+ else:
+ items = [str(value)]
+ if not items:
+ return "—"
+ return " • ".join(items)
+
+ if input_type == "time":
+ return str(value)
+
+ if input_type == "number":
+ try:
+ number = Decimal(str(value))
+ if number == number.to_integral_value():
+ rendered = f"{int(number)}"
+ else:
+ rendered = str(number).replace(".", ",")
+ except InvalidOperation:
+ rendered = str(value)
+ if unit:
+ return f"{rendered} {unit}"
+ return rendered
+
+ if unit:
+ return f"{value} {unit}"
+
+ return str(value)
+
+ @classmethod
+ def get_state_icon(cls, key: str, value: Any) -> str:
+ metadata = cls.get_metadata(key)
+ definition = cls.get_definition(key)
+ input_type = metadata.input_type or cls._default_input_type(definition)
+
+ if input_type == "toggle":
+ return "✅" if bool(value) else "❌"
+ if value in (None, "", [], {}):
+ return "⚪"
+ return "🟢"
+
+ @classmethod
+ def get_setting_dashboard_entry(cls, key: str) -> Dict[str, Any]:
+ definition = cls.get_definition(key)
+ metadata = cls.get_metadata(key)
+ current = cls.get_current_value(key)
+ return {
+ "key": key,
+ "name": definition.display_name,
+ "icon": metadata.icon or definition.icon,
+ "state_icon": cls.get_state_icon(key, current),
+ "value": cls.format_setting_value(key, current),
+ "has_override": cls.has_override(key),
+ "description": metadata.description or cls._default_description(definition),
+ "recommended": metadata.recommended,
+ "unit": metadata.unit,
+ "category_description": metadata.category_description,
+ }
+
def _resolve_category_key(cls, key: str) -> str:
override = cls.CATEGORY_KEY_OVERRIDES.get(key)
if override:
@@ -443,7 +1121,7 @@ class BotConfigurationService:
@classmethod
def format_value_for_list(cls, key: str) -> str:
value = cls.get_current_value(key)
- formatted = cls.format_value(value)
+ formatted = cls.format_setting_value(key, value)
if formatted == "—":
return formatted
return _truncate(formatted)
@@ -604,9 +1282,12 @@ class BotConfigurationService:
if definition.is_optional and text.lower() in {"none", "null", "пусто", ""}:
return None
+ metadata = cls.get_metadata(key)
+ input_type = metadata.input_type or cls._default_input_type(definition)
+
python_type = definition.python_type
- if python_type is bool:
+ if input_type == "toggle" or python_type is bool:
lowered = text.lower()
if lowered in {"1", "true", "on", "yes", "да", "вкл", "enable", "enabled"}:
return True
@@ -614,8 +1295,14 @@ class BotConfigurationService:
return False
raise ValueError("Введите 'true' или 'false' (или 'да'/'нет')")
- if python_type is int:
- parsed_value: Any = int(text)
+ if input_type == "price":
+ parsed_value = cls._parse_price(text)
+ elif input_type == "time":
+ parsed_value = cls._parse_time(text)
+ elif input_type == "list":
+ parsed_value = cls._parse_list(text)
+ elif python_type is int:
+ parsed_value = int(text)
elif python_type is float:
parsed_value = float(text.replace(",", "."))
else:
@@ -645,22 +1332,57 @@ class BotConfigurationService:
return parsed_value
@classmethod
- async def set_value(cls, db: AsyncSession, key: str, value: Any) -> None:
+ async def set_value(
+ cls,
+ db: AsyncSession,
+ key: str,
+ value: Any,
+ *,
+ actor: Optional[str] = None,
+ reason: str = "manual",
+ ) -> None:
+ old_value = cls.get_current_value(key)
raw_value = cls.serialize_value(key, value)
await upsert_system_setting(db, key, raw_value)
cls._overrides_raw[key] = raw_value
cls._apply_to_settings(key, value)
+ cls._record_history(key, old_value, value, actor=actor, reason=reason)
+ logger.info(
+ "Настройка %s обновлена: %s → %s (%s)",
+ key,
+ cls.format_setting_value(key, old_value),
+ cls.format_setting_value(key, value),
+ actor or "system",
+ )
+
if key in {"WEB_API_DEFAULT_TOKEN", "WEB_API_DEFAULT_TOKEN_NAME"}:
await cls._sync_default_web_api_token()
@classmethod
- async def reset_value(cls, db: AsyncSession, key: str) -> None:
+ async def reset_value(
+ cls,
+ db: AsyncSession,
+ key: str,
+ *,
+ actor: Optional[str] = None,
+ reason: str = "reset",
+ ) -> None:
+ old_value = cls.get_current_value(key)
await delete_system_setting(db, key)
cls._overrides_raw.pop(key, None)
original = cls.get_original_value(key)
cls._apply_to_settings(key, original)
+ cls._record_history(key, old_value, original, actor=actor, reason=reason)
+ logger.info(
+ "Настройка %s сброшена: %s → %s (%s)",
+ key,
+ cls.format_setting_value(key, old_value),
+ cls.format_setting_value(key, original),
+ actor or "system",
+ )
+
if key in {"WEB_API_DEFAULT_TOKEN", "WEB_API_DEFAULT_TOKEN_NAME"}:
await cls._sync_default_web_api_token()
@@ -693,14 +1415,122 @@ class BotConfigurationService:
return {
"key": key,
"name": definition.display_name,
- "current": cls.format_value(current),
- "original": cls.format_value(original),
+ "current": cls.format_setting_value(key, current),
+ "original": cls.format_setting_value(key, original),
"type": definition.type_label,
"category_key": definition.category_key,
"category_label": definition.category_label,
"has_override": has_override,
}
+ @classmethod
+ def _record_history(
+ cls,
+ key: str,
+ old_value: Any,
+ new_value: Any,
+ *,
+ actor: Optional[str],
+ reason: str,
+ ) -> None:
+ definition = cls.get_definition(key)
+ entry = {
+ "timestamp": datetime.utcnow(),
+ "key": key,
+ "name": definition.display_name,
+ "old": cls.format_setting_value(key, old_value),
+ "new": cls.format_setting_value(key, new_value),
+ "actor": actor,
+ "reason": reason,
+ }
+ cls._history.appendleft(entry)
+
+ @classmethod
+ def get_history(cls) -> List[Dict[str, Any]]:
+ return list(cls._history)
+
+ @classmethod
+ def generate_env_dump(cls, *, include_secrets: bool = True) -> str:
+ cls.initialize_definitions()
+ lines: List[str] = []
+ for key in sorted(cls._definitions.keys()):
+ value = cls.get_current_value(key)
+ raw = cls.serialize_value(key, value)
+ if raw is None:
+ continue
+ if not include_secrets and cls._is_secret_key(key):
+ lines.append(f"{key}=")
+ else:
+ escaped = raw.replace("\\", "\\\\").replace("\n", "\\n")
+ lines.append(f"{key}={escaped}")
+ return "\n".join(lines) + "\n"
+
+ @classmethod
+ def parse_env_dump(cls, content: str) -> Dict[str, Any]:
+ cls.initialize_definitions()
+ result: Dict[str, Any] = {}
+ for raw_line in content.splitlines():
+ line = raw_line.strip()
+ if not line or line.startswith("#"):
+ continue
+ if "=" not in line:
+ continue
+ key, raw_value = line.split("=", 1)
+ key = key.strip()
+ if key not in cls._definitions:
+ continue
+ value_text = raw_value.strip().strip('"').strip("'")
+ value_text = value_text.replace("\\n", "\n")
+ try:
+ parsed_value = cls.parse_user_value(key, value_text)
+ except ValueError as error:
+ raise ValueError(f"{key}: {error}") from error
+ result[key] = parsed_value
+ return result
+
+ @classmethod
+ def search_settings(cls, query: str, limit: int = 12) -> List[str]:
+ cls.initialize_definitions()
+ normalized = (query or "").strip().lower()
+ if not normalized:
+ return []
+
+ tokens = [token for token in re.split(r"\s+", normalized) if token]
+ if not tokens:
+ return []
+
+ scored: List[Tuple[float, str]] = []
+ for key, definition in cls._definitions.items():
+ if key in cls.EXCLUDED_KEYS:
+ continue
+
+ metadata = cls.get_metadata(key)
+ haystacks: List[str] = [
+ definition.display_name.lower(),
+ definition.category_label.lower(),
+ key.lower(),
+ ]
+ if metadata.description:
+ haystacks.append(metadata.description.lower())
+ if metadata.tags:
+ haystacks.extend(tag.lower() for tag in metadata.tags)
+
+ score = 0.0
+ for token in tokens:
+ for haystack in haystacks:
+ if token == haystack:
+ score += 5.0
+ elif token in haystack:
+ score += 1.0 + (len(token) / max(len(haystack), 1))
+
+ if score > 0:
+ if definition.category_key.startswith("PAYMENT"):
+ score += 0.1
+ scored.append((score, key))
+
+ scored.sort(key=lambda item: (-item[0], cls._definitions[item[1]].display_name.lower()))
+ return [key for _, key in scored[:limit]]
+
bot_configuration_service = BotConfigurationService
diff --git a/app/states.py b/app/states.py
index 43655b35..96198dc7 100644
--- a/app/states.py
+++ b/app/states.py
@@ -132,6 +132,9 @@ class SupportSettingsStates(StatesGroup):
class BotConfigStates(StatesGroup):
waiting_for_value = State()
+ waiting_for_search_query = State()
+ waiting_for_import_payload = State()
+ waiting_for_preset_name = State()
class AutoPayStates(StatesGroup):
setting_autopay_days = State()