diff --git a/app/bot.py b/app/bot.py
index a13e2237..d2b7f56b 100644
--- a/app/bot.py
+++ b/app/bot.py
@@ -39,6 +39,7 @@ from app.handlers.admin import (
welcome_text as admin_welcome_text,
tickets as admin_tickets,
reports as admin_reports,
+ bot_configuration as admin_bot_configuration,
)
from app.handlers.stars_payments import register_stars_handlers
@@ -141,6 +142,7 @@ async def setup_bot() -> tuple[Bot, Dispatcher]:
admin_welcome_text.register_welcome_text_handlers(dp)
admin_tickets.register_handlers(dp)
admin_reports.register_handlers(dp)
+ admin_bot_configuration.register_handlers(dp)
common.register_handlers(dp)
register_stars_handlers(dp)
logger.info("⭐ Зарегистрированы обработчики Telegram Stars платежей")
diff --git a/app/database/crud/system_setting.py b/app/database/crud/system_setting.py
new file mode 100644
index 00000000..63aaf719
--- /dev/null
+++ b/app/database/crud/system_setting.py
@@ -0,0 +1,40 @@
+from typing import Optional
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.models import SystemSetting
+
+
+async def upsert_system_setting(
+ db: AsyncSession,
+ key: str,
+ value: Optional[str],
+ description: Optional[str] = None,
+) -> SystemSetting:
+ result = await db.execute(
+ select(SystemSetting).where(SystemSetting.key == key)
+ )
+ setting = result.scalar_one_or_none()
+
+ if setting is None:
+ setting = SystemSetting(key=key, value=value, description=description)
+ db.add(setting)
+ else:
+ setting.value = value
+ if description is not None:
+ setting.description = description
+
+ await db.flush()
+ return setting
+
+
+async def delete_system_setting(db: AsyncSession, key: str) -> None:
+ result = await db.execute(
+ select(SystemSetting).where(SystemSetting.key == key)
+ )
+ setting = result.scalar_one_or_none()
+ if setting is not None:
+ await db.delete(setting)
+ await db.flush()
+
diff --git a/app/database/universal_migration.py b/app/database/universal_migration.py
index b0edb1b6..ee1e5cee 100644
--- a/app/database/universal_migration.py
+++ b/app/database/universal_migration.py
@@ -1833,6 +1833,59 @@ async def ensure_server_promo_groups_setup() -> bool:
)
return False
+async def create_system_settings_table() -> bool:
+ table_exists = await check_table_exists("system_settings")
+ if table_exists:
+ logger.info("ℹ️ Таблица system_settings уже существует")
+ return True
+
+ try:
+ async with engine.begin() as conn:
+ db_type = await get_database_type()
+
+ if db_type == "sqlite":
+ create_sql = """
+ CREATE TABLE system_settings (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ key VARCHAR(255) NOT NULL UNIQUE,
+ value TEXT NULL,
+ description TEXT NULL,
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ );
+ """
+ elif db_type == "postgresql":
+ create_sql = """
+ CREATE TABLE system_settings (
+ id SERIAL PRIMARY KEY,
+ key VARCHAR(255) NOT NULL UNIQUE,
+ value TEXT NULL,
+ description TEXT NULL,
+ created_at TIMESTAMP DEFAULT NOW(),
+ updated_at TIMESTAMP DEFAULT NOW()
+ );
+ """
+ else:
+ create_sql = """
+ CREATE TABLE system_settings (
+ id INT AUTO_INCREMENT PRIMARY KEY,
+ key VARCHAR(255) NOT NULL UNIQUE,
+ value TEXT NULL,
+ description TEXT NULL,
+ created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+ updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+ );
+ """
+
+ await conn.execute(text(create_sql))
+ logger.info("✅ Таблица system_settings создана")
+ return True
+
+ except Exception as error:
+ logger.error(f"Ошибка создания таблицы system_settings: {error}")
+ return False
+
+
async def run_universal_migration():
logger.info("=== НАЧАЛО УНИВЕРСАЛЬНОЙ МИГРАЦИИ ===")
@@ -1844,6 +1897,13 @@ async def run_universal_migration():
if not referral_migration_success:
logger.warning("⚠️ Проблемы с миграцией реферальной системы")
+ logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ SYSTEM_SETTINGS ===")
+ system_settings_ready = await create_system_settings_table()
+ if system_settings_ready:
+ logger.info("✅ Таблица system_settings готова")
+ else:
+ logger.warning("⚠️ Проблемы с таблицей system_settings")
+
logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ CRYPTOBOT ===")
cryptobot_created = await create_cryptobot_payments_table()
if cryptobot_created:
diff --git a/app/handlers/admin/bot_configuration.py b/app/handlers/admin/bot_configuration.py
new file mode 100644
index 00000000..76120f61
--- /dev/null
+++ b/app/handlers/admin/bot_configuration.py
@@ -0,0 +1,410 @@
+import math
+from typing import Tuple
+
+from aiogram import Dispatcher, F, types
+from aiogram.fsm.context import FSMContext
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.models import User
+from app.localization.texts import get_texts
+from app.services.system_settings_service import bot_configuration_service
+from app.states import BotConfigStates
+from app.utils.decorators import admin_required, error_handler
+
+
+CATEGORY_PAGE_SIZE = 10
+SETTINGS_PAGE_SIZE = 8
+
+
+def _parse_category_payload(payload: str) -> Tuple[str, int]:
+ parts = payload.split(":")
+ if len(parts) == 3:
+ _, category_key, page_raw = parts
+ try:
+ return category_key, max(1, int(page_raw))
+ except ValueError:
+ return category_key, 1
+ if len(parts) == 2:
+ _, category_key = parts
+ return category_key, 1
+ return "", 1
+
+
+def _build_categories_keyboard(language: str, page: int = 1) -> types.InlineKeyboardMarkup:
+ categories = bot_configuration_service.get_categories()
+ 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]] = []
+ for category_key, label, count in sliced:
+ button_text = f"{label} ({count})"
+ rows.append([
+ types.InlineKeyboardButton(
+ text=button_text,
+ callback_data=f"botcfg_cat:{category_key}:1",
+ )
+ ])
+
+ if total_pages > 1:
+ nav_row: list[types.InlineKeyboardButton] = []
+ if page > 1:
+ nav_row.append(
+ types.InlineKeyboardButton(
+ text="⬅️", callback_data=f"botcfg_categories:{page - 1}"
+ )
+ )
+ nav_row.append(
+ types.InlineKeyboardButton(
+ text=f"{page}/{total_pages}", callback_data="botcfg_categories:noop"
+ )
+ )
+ if page < total_pages:
+ nav_row.append(
+ types.InlineKeyboardButton(
+ text="➡️", callback_data=f"botcfg_categories:{page + 1}"
+ )
+ )
+ rows.append(nav_row)
+
+ rows.append([
+ types.InlineKeyboardButton(text="⬅️ Назад", callback_data="admin_submenu_settings")
+ ])
+
+ return types.InlineKeyboardMarkup(inline_keyboard=rows)
+
+
+def _build_settings_keyboard(
+ category_key: str,
+ 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]
+
+ rows: list[list[types.InlineKeyboardButton]] = []
+
+ for definition in sliced:
+ value_preview = bot_configuration_service.format_value_for_list(definition.key)
+ button_text = f"{definition.key} = {value_preview}"
+ rows.append([
+ types.InlineKeyboardButton(
+ text=button_text,
+ callback_data=f"botcfg_setting:{definition.key}",
+ )
+ ])
+
+ if total_pages > 1:
+ nav_row: list[types.InlineKeyboardButton] = []
+ if page > 1:
+ nav_row.append(
+ types.InlineKeyboardButton(
+ text="⬅️",
+ callback_data=f"botcfg_cat:{category_key}:{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:{category_key}:{page + 1}",
+ )
+ )
+ rows.append(nav_row)
+
+ rows.append([
+ types.InlineKeyboardButton(
+ text="⬅️ К категориям",
+ callback_data="admin_bot_config",
+ )
+ ])
+
+ return types.InlineKeyboardMarkup(inline_keyboard=rows)
+
+
+def _build_setting_keyboard(key: str) -> types.InlineKeyboardMarkup:
+ definition = bot_configuration_service.get_definition(key)
+ rows: list[list[types.InlineKeyboardButton]] = []
+
+ if definition.python_type is bool:
+ rows.append([
+ types.InlineKeyboardButton(
+ text="🔁 Переключить",
+ callback_data=f"botcfg_toggle:{key}",
+ )
+ ])
+
+ rows.append([
+ types.InlineKeyboardButton(
+ text="✏️ Изменить",
+ callback_data=f"botcfg_edit:{key}",
+ )
+ ])
+
+ if bot_configuration_service.has_override(key):
+ rows.append([
+ types.InlineKeyboardButton(
+ text="♻️ Сбросить",
+ callback_data=f"botcfg_reset:{key}",
+ )
+ ])
+
+ rows.append([
+ types.InlineKeyboardButton(
+ text="⬅️ Назад",
+ callback_data=f"botcfg_cat:{definition.category_key}:1",
+ )
+ ])
+
+ return types.InlineKeyboardMarkup(inline_keyboard=rows)
+
+
+def _render_setting_text(key: str) -> str:
+ summary = bot_configuration_service.get_setting_summary(key)
+
+ lines = [
+ "🧩 Настройка",
+ f"Ключ: {summary['key']}",
+ f"Тип: {summary['type']}",
+ f"Текущее значение: {summary['current']}",
+ f"Значение по умолчанию: {summary['original']}",
+ f"Переопределено в БД: {'✅ Да' if summary['has_override'] else '❌ Нет'}",
+ ]
+
+ return "\n".join(lines)
+
+
+@admin_required
+@error_handler
+async def show_bot_config_menu(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ keyboard = _build_categories_keyboard(db_user.language)
+ await callback.message.edit_text(
+ "🧩 Конфигурация бота\n\nВыберите категорию настроек:",
+ reply_markup=keyboard,
+ )
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def show_bot_config_categories_page(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ parts = callback.data.split(":")
+ try:
+ page = int(parts[1])
+ except (IndexError, ValueError):
+ page = 1
+
+ keyboard = _build_categories_keyboard(db_user.language, page)
+ await callback.message.edit_text(
+ "🧩 Конфигурация бота\n\nВыберите категорию настроек:",
+ reply_markup=keyboard,
+ )
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def show_bot_config_category(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ category_key, page = _parse_category_payload(callback.data)
+ definitions = bot_configuration_service.get_settings_for_category(category_key)
+
+ if not definitions:
+ await callback.answer("В этой категории пока нет настроек", show_alert=True)
+ return
+
+ category_label = definitions[0].category_label
+ keyboard = _build_settings_keyboard(category_key, db_user.language, page)
+ await callback.message.edit_text(
+ f"🧩 {category_label}\n\nВыберите настройку для просмотра:",
+ reply_markup=keyboard,
+ )
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def show_bot_config_setting(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ key = callback.data.split(":", 1)[1]
+ text = _render_setting_text(key)
+ keyboard = _build_setting_keyboard(key)
+ await callback.message.edit_text(text, reply_markup=keyboard)
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def start_edit_setting(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ key = callback.data.split(":", 1)[1]
+ definition = bot_configuration_service.get_definition(key)
+
+ summary = bot_configuration_service.get_setting_summary(key)
+ texts = get_texts(db_user.language)
+
+ instructions = [
+ "✏️ Редактирование настройки",
+ f"Ключ: {summary['key']}",
+ f"Тип: {summary['type']}",
+ f"Текущее значение: {summary['current']}",
+ "\nОтправьте новое значение сообщением.",
+ ]
+
+ if definition.is_optional:
+ instructions.append("Отправьте 'none' или оставьте пустым для сброса на значение по умолчанию.")
+
+ instructions.append("Для отмены отправьте 'cancel'.")
+
+ await callback.message.edit_text(
+ "\n".join(instructions),
+ reply_markup=types.InlineKeyboardMarkup(
+ inline_keyboard=[
+ [
+ types.InlineKeyboardButton(
+ text=texts.BACK, callback_data=f"botcfg_setting:{key}"
+ )
+ ]
+ ]
+ ),
+ )
+
+ await state.update_data(setting_key=key)
+ await state.set_state(BotConfigStates.waiting_for_value)
+ await callback.answer()
+
+
+@admin_required
+@error_handler
+async def handle_edit_setting(
+ message: types.Message,
+ db_user: User,
+ db: AsyncSession,
+ state: FSMContext,
+):
+ data = await state.get_data()
+ key = data.get("setting_key")
+
+ if not key:
+ await message.answer("Не удалось определить редактируемую настройку. Попробуйте снова.")
+ await state.clear()
+ return
+
+ try:
+ value = bot_configuration_service.parse_user_value(key, message.text or "")
+ except ValueError as error:
+ await message.answer(f"⚠️ {error}")
+ return
+
+ await bot_configuration_service.set_value(db, key, value)
+ await db.commit()
+
+ text = _render_setting_text(key)
+ keyboard = _build_setting_keyboard(key)
+ await message.answer("✅ Настройка обновлена")
+ await message.answer(text, reply_markup=keyboard)
+ await state.clear()
+
+
+@admin_required
+@error_handler
+async def reset_setting(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ key = callback.data.split(":", 1)[1]
+ await bot_configuration_service.reset_value(db, key)
+ await db.commit()
+
+ text = _render_setting_text(key)
+ keyboard = _build_setting_keyboard(key)
+ await callback.message.edit_text(text, reply_markup=keyboard)
+ await callback.answer("Сброшено к значению по умолчанию")
+
+
+@admin_required
+@error_handler
+async def toggle_setting(
+ callback: types.CallbackQuery,
+ db_user: User,
+ db: AsyncSession,
+):
+ key = callback.data.split(":", 1)[1]
+ current = bot_configuration_service.get_current_value(key)
+ new_value = not bool(current)
+ await bot_configuration_service.set_value(db, key, new_value)
+ await db.commit()
+
+ text = _render_setting_text(key)
+ keyboard = _build_setting_keyboard(key)
+ await callback.message.edit_text(text, reply_markup=keyboard)
+ await callback.answer("Обновлено")
+
+
+def register_handlers(dp: Dispatcher) -> None:
+ dp.callback_query.register(
+ show_bot_config_menu,
+ F.data == "admin_bot_config",
+ )
+ dp.callback_query.register(
+ show_bot_config_categories_page,
+ F.data.startswith("botcfg_categories:")
+ & (~F.data.endswith(":noop")),
+ )
+ dp.callback_query.register(
+ show_bot_config_category,
+ F.data.startswith("botcfg_cat:"),
+ )
+ dp.callback_query.register(
+ show_bot_config_setting,
+ F.data.startswith("botcfg_setting:"),
+ )
+ dp.callback_query.register(
+ start_edit_setting,
+ F.data.startswith("botcfg_edit:"),
+ )
+ dp.callback_query.register(
+ reset_setting,
+ F.data.startswith("botcfg_reset:"),
+ )
+ dp.callback_query.register(
+ toggle_setting,
+ F.data.startswith("botcfg_toggle:"),
+ )
+ dp.message.register(
+ handle_edit_setting,
+ BotConfigStates.waiting_for_value,
+ )
+
diff --git a/app/keyboards/admin.py b/app/keyboards/admin.py
index 13c37204..f9701c6f 100644
--- a/app/keyboards/admin.py
+++ b/app/keyboards/admin.py
@@ -99,6 +99,9 @@ def get_admin_settings_submenu_keyboard(language: str = "ru") -> InlineKeyboardM
InlineKeyboardButton(text=texts.ADMIN_REMNAWAVE, callback_data="admin_remnawave"),
InlineKeyboardButton(text=texts.ADMIN_MONITORING, callback_data="admin_monitoring")
],
+ [
+ InlineKeyboardButton(text="🧩 Конфигурация бота", callback_data="admin_bot_config"),
+ ],
[
InlineKeyboardButton(
text=texts.t("ADMIN_MONITORING_SETTINGS", "⚙️ Настройки мониторинга"),
diff --git a/app/services/system_settings_service.py b/app/services/system_settings_service.py
new file mode 100644
index 00000000..70a9d890
--- /dev/null
+++ b/app/services/system_settings_service.py
@@ -0,0 +1,361 @@
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple, Type, Union, get_args, get_origin
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import Settings, settings
+from app.database.crud.system_setting import (
+ delete_system_setting,
+ upsert_system_setting,
+)
+from app.database.database import AsyncSessionLocal
+from app.database.models import SystemSetting
+
+
+logger = logging.getLogger(__name__)
+
+
+def _title_from_key(key: str) -> str:
+ parts = key.split("_")
+ if not parts:
+ return key
+ return " ".join(part.capitalize() for part in parts)
+
+
+def _truncate(value: str, max_len: int = 60) -> str:
+ value = value.strip()
+ if len(value) <= max_len:
+ return value
+ return value[: max_len - 1] + "…"
+
+
+@dataclass(slots=True)
+class SettingDefinition:
+ key: str
+ category_key: str
+ category_label: str
+ python_type: Type[Any]
+ type_label: str
+ is_optional: bool
+
+ @property
+ def display_name(self) -> str:
+ return _title_from_key(self.key)
+
+
+class BotConfigurationService:
+ EXCLUDED_KEYS: set[str] = {"BOT_TOKEN", "ADMIN_IDS"}
+
+ CATEGORY_TITLES: Dict[str, str] = {
+ "DATABASE": "База данных",
+ "POSTGRES": "PostgreSQL",
+ "SQLITE": "SQLite",
+ "REDIS": "Redis",
+ "REMNAWAVE": "Remnawave",
+ "SUPPORT": "Поддержка",
+ "ADMIN": "Администрирование",
+ "CHANNEL": "Каналы",
+ "TRIAL": "Триал",
+ "DEFAULT": "Значения по умолчанию",
+ "PRICE": "Цены",
+ "TRAFFIC": "Трафик",
+ "REFERRAL": "Реферальная программа",
+ "AUTOPAY": "Автопродление",
+ "MONITORING": "Мониторинг",
+ "SERVER": "Статус серверов",
+ "MAINTENANCE": "Техработы",
+ "PAYMENT": "Оплаты",
+ "YOOKASSA": "YooKassa",
+ "CRYPTOBOT": "CryptoBot",
+ "MULENPAY": "MulenPay",
+ "PAL24": "PayPalych",
+ "CONNECT": "Кнопка подключения",
+ "HAPP": "Happ",
+ "VERSION": "Версии",
+ "BACKUP": "Бекапы",
+ "WEBHOOK": "Вебхуки",
+ "LOG": "Логи",
+ "DEBUG": "Отладка",
+ "TRIBUTE": "Tribute",
+ "TELEGRAM": "Telegram Stars",
+ }
+
+ _definitions: Dict[str, SettingDefinition] = {}
+ _original_values: Dict[str, Any] = settings.model_dump()
+ _overrides_raw: Dict[str, Optional[str]] = {}
+
+ @classmethod
+ def initialize_definitions(cls) -> None:
+ if cls._definitions:
+ return
+
+ for key, field in Settings.model_fields.items():
+ if key in cls.EXCLUDED_KEYS:
+ continue
+
+ annotation = field.annotation
+ python_type, is_optional = cls._normalize_type(annotation)
+ type_label = cls._type_to_label(python_type, is_optional)
+
+ category_key = cls._resolve_category_key(key)
+ category_label = cls.CATEGORY_TITLES.get(
+ category_key,
+ category_key.capitalize() if category_key else "Прочее",
+ )
+
+ cls._definitions[key] = SettingDefinition(
+ key=key,
+ category_key=category_key or "other",
+ category_label=category_label,
+ python_type=python_type,
+ type_label=type_label,
+ is_optional=is_optional,
+ )
+
+ @classmethod
+ def _resolve_category_key(cls, key: str) -> str:
+ if "_" not in key:
+ return key.upper()
+ prefix = key.split("_", 1)[0]
+ return prefix.upper()
+
+ @classmethod
+ def _normalize_type(cls, annotation: Any) -> Tuple[Type[Any], bool]:
+ if annotation is None:
+ return str, True
+
+ origin = get_origin(annotation)
+ if origin is Union:
+ args = [arg for arg in get_args(annotation) if arg is not type(None)]
+ if len(args) == 1:
+ nested_type, nested_optional = cls._normalize_type(args[0])
+ return nested_type, True
+ return str, True
+
+ if annotation in {int, float, bool, str}:
+ return annotation, False
+
+ if annotation in {Optional[int], Optional[float], Optional[bool], Optional[str]}:
+ nested = get_args(annotation)[0]
+ return nested, True
+
+ # Paths, lists, dicts и прочее будем хранить как строки
+ return str, False
+
+ @classmethod
+ def _type_to_label(cls, python_type: Type[Any], is_optional: bool) -> str:
+ base = {
+ bool: "bool",
+ int: "int",
+ float: "float",
+ str: "str",
+ }.get(python_type, "str")
+ return f"optional[{base}]" if is_optional else base
+
+ @classmethod
+ def get_categories(cls) -> List[Tuple[str, str, int]]:
+ cls.initialize_definitions()
+ categories: Dict[str, List[SettingDefinition]] = {}
+
+ for definition in cls._definitions.values():
+ categories.setdefault(definition.category_key, []).append(definition)
+
+ result: List[Tuple[str, str, int]] = []
+ for category_key, items in categories.items():
+ label = items[0].category_label
+ result.append((category_key, label, len(items)))
+
+ result.sort(key=lambda item: item[1])
+ return result
+
+ @classmethod
+ def get_settings_for_category(cls, category_key: str) -> List[SettingDefinition]:
+ cls.initialize_definitions()
+ filtered = [
+ definition
+ for definition in cls._definitions.values()
+ if definition.category_key == category_key
+ ]
+ filtered.sort(key=lambda definition: definition.key)
+ return filtered
+
+ @classmethod
+ def get_definition(cls, key: str) -> SettingDefinition:
+ cls.initialize_definitions()
+ return cls._definitions[key]
+
+ @classmethod
+ def has_override(cls, key: str) -> bool:
+ return key in cls._overrides_raw
+
+ @classmethod
+ def get_current_value(cls, key: str) -> Any:
+ return getattr(settings, key)
+
+ @classmethod
+ def get_original_value(cls, key: str) -> Any:
+ return cls._original_values.get(key)
+
+ @classmethod
+ def format_value(cls, value: Any) -> str:
+ if value is None:
+ return "—"
+ if isinstance(value, bool):
+ return "✅ Да" if value else "❌ Нет"
+ if isinstance(value, (int, float)):
+ return str(value)
+ if isinstance(value, (list, dict, tuple, set)):
+ try:
+ return json.dumps(value, ensure_ascii=False)
+ except Exception:
+ return str(value)
+ return str(value)
+
+ @classmethod
+ def format_value_for_list(cls, key: str) -> str:
+ value = cls.get_current_value(key)
+ formatted = cls.format_value(value)
+ if formatted == "—":
+ return formatted
+ return _truncate(formatted)
+
+ @classmethod
+ async def initialize(cls) -> None:
+ cls.initialize_definitions()
+
+ async with AsyncSessionLocal() as session:
+ result = await session.execute(select(SystemSetting))
+ rows = result.scalars().all()
+
+ overrides: Dict[str, Optional[str]] = {}
+ for row in rows:
+ if row.key in cls._definitions:
+ overrides[row.key] = row.value
+
+ for key, raw_value in overrides.items():
+ try:
+ parsed_value = cls.deserialize_value(key, raw_value)
+ except Exception as error:
+ logger.error("Не удалось применить настройку %s: %s", key, error)
+ continue
+
+ cls._overrides_raw[key] = raw_value
+ cls._apply_to_settings(key, parsed_value)
+
+ @classmethod
+ async def reload(cls) -> None:
+ cls._overrides_raw.clear()
+ await cls.initialize()
+
+ @classmethod
+ def deserialize_value(cls, key: str, raw_value: Optional[str]) -> Any:
+ if raw_value is None:
+ return None
+
+ definition = cls.get_definition(key)
+ python_type = definition.python_type
+
+ if python_type is bool:
+ value_lower = raw_value.strip().lower()
+ if value_lower in {"1", "true", "on", "yes", "да"}:
+ return True
+ if value_lower in {"0", "false", "off", "no", "нет"}:
+ return False
+ raise ValueError(f"Неверное булево значение: {raw_value}")
+
+ if python_type is int:
+ return int(raw_value)
+
+ if python_type is float:
+ return float(raw_value)
+
+ return raw_value
+
+ @classmethod
+ def serialize_value(cls, key: str, value: Any) -> Optional[str]:
+ if value is None:
+ return None
+
+ definition = cls.get_definition(key)
+ python_type = definition.python_type
+
+ if python_type is bool:
+ return "true" if value else "false"
+ if python_type in {int, float}:
+ return str(value)
+ return str(value)
+
+ @classmethod
+ def parse_user_value(cls, key: str, user_input: str) -> Any:
+ definition = cls.get_definition(key)
+ text = (user_input or "").strip()
+
+ if text.lower() in {"отмена", "cancel"}:
+ raise ValueError("Ввод отменен пользователем")
+
+ if definition.is_optional and text.lower() in {"none", "null", "пусто", ""}:
+ return None
+
+ python_type = definition.python_type
+
+ if python_type is bool:
+ lowered = text.lower()
+ if lowered in {"1", "true", "on", "yes", "да", "вкл", "enable", "enabled"}:
+ return True
+ if lowered in {"0", "false", "off", "no", "нет", "выкл", "disable", "disabled"}:
+ return False
+ raise ValueError("Введите 'true' или 'false' (или 'да'/'нет')")
+
+ if python_type is int:
+ return int(text)
+
+ if python_type is float:
+ return float(text.replace(",", "."))
+
+ return text
+
+ @classmethod
+ async def set_value(cls, db: AsyncSession, key: str, value: Any) -> None:
+ 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)
+
+ @classmethod
+ async def reset_value(cls, db: AsyncSession, key: str) -> None:
+ await delete_system_setting(db, key)
+ cls._overrides_raw.pop(key, None)
+ original = cls.get_original_value(key)
+ cls._apply_to_settings(key, original)
+
+ @classmethod
+ def _apply_to_settings(cls, key: str, value: Any) -> None:
+ try:
+ setattr(settings, key, value)
+ except Exception as error:
+ logger.error("Не удалось применить значение %s=%s: %s", key, value, error)
+
+ @classmethod
+ def get_setting_summary(cls, key: str) -> Dict[str, Any]:
+ definition = cls.get_definition(key)
+ current = cls.get_current_value(key)
+ original = cls.get_original_value(key)
+ has_override = cls.has_override(key)
+
+ return {
+ "key": key,
+ "name": definition.display_name,
+ "current": cls.format_value(current),
+ "original": cls.format_value(original),
+ "type": definition.type_label,
+ "category_key": definition.category_key,
+ "category_label": definition.category_label,
+ "has_override": has_override,
+ }
+
+
+bot_configuration_service = BotConfigurationService
+
diff --git a/app/states.py b/app/states.py
index f824f9a5..2b93a56e 100644
--- a/app/states.py
+++ b/app/states.py
@@ -123,6 +123,10 @@ class AdminTicketStates(StatesGroup):
class SupportSettingsStates(StatesGroup):
waiting_for_desc = State()
+
+class BotConfigStates(StatesGroup):
+ waiting_for_value = State()
+
class AutoPayStates(StatesGroup):
setting_autopay_days = State()
confirming_autopay_toggle = State()
diff --git a/main.py b/main.py
index 254e5c5d..cf19d0b5 100644
--- a/main.py
+++ b/main.py
@@ -21,6 +21,7 @@ from app.database.universal_migration import run_universal_migration
from app.services.backup_service import backup_service
from app.services.reporting_service import reporting_service
from app.localization.loader import ensure_locale_templates
+from app.services.system_settings_service import bot_configuration_service
class GracefulExit:
@@ -85,6 +86,13 @@ async def main():
else:
logger.info("ℹ️ Миграция пропущена (SKIP_MIGRATION=true)")
+ logger.info("⚙️ Загрузка конфигурации из БД...")
+ try:
+ await bot_configuration_service.initialize()
+ logger.info("✅ Конфигурация загружена")
+ except Exception as error:
+ logger.error(f"❌ Не удалось загрузить конфигурацию: {error}")
+
logger.info("🤖 Настройка бота...")
bot, dp = await setup_bot()