mirror of
https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git
synced 2026-01-20 03:40:26 +00:00
Revert "Add admin bot configuration management UI"
This commit is contained in:
Egor
@@ -39,7 +39,6 @@ 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
|
||||
|
||||
@@ -142,7 +141,6 @@ 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 платежей")
|
||||
|
||||
@@ -1,40 +0,0 @@
|
||||
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()
|
||||
|
||||
@@ -1833,59 +1833,6 @@ 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("=== НАЧАЛО УНИВЕРСАЛЬНОЙ МИГРАЦИИ ===")
|
||||
|
||||
@@ -1897,13 +1844,6 @@ 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:
|
||||
|
||||
@@ -1,410 +0,0 @@
|
||||
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 = [
|
||||
"🧩 <b>Настройка</b>",
|
||||
f"<b>Ключ:</b> <code>{summary['key']}</code>",
|
||||
f"<b>Тип:</b> {summary['type']}",
|
||||
f"<b>Текущее значение:</b> {summary['current']}",
|
||||
f"<b>Значение по умолчанию:</b> {summary['original']}",
|
||||
f"<b>Переопределено в БД:</b> {'✅ Да' 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(
|
||||
"🧩 <b>Конфигурация бота</b>\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(
|
||||
"🧩 <b>Конфигурация бота</b>\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"🧩 <b>{category_label}</b>\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 = [
|
||||
"✏️ <b>Редактирование настройки</b>",
|
||||
f"Ключ: <code>{summary['key']}</code>",
|
||||
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,
|
||||
)
|
||||
|
||||
@@ -99,9 +99,6 @@ 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", "⚙️ Настройки мониторинга"),
|
||||
|
||||
@@ -1,361 +0,0 @@
|
||||
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
|
||||
|
||||
@@ -123,10 +123,6 @@ 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()
|
||||
|
||||
8
main.py
8
main.py
@@ -21,7 +21,6 @@ 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:
|
||||
@@ -86,13 +85,6 @@ 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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user