Revert "Add admin-managed bot configuration settings"

This commit is contained in:
Egor
2025-09-25 16:57:55 +03:00
committed by GitHub
parent 1be83b7b48
commit 198ce23625
11 changed files with 12 additions and 950 deletions

View File

@@ -36,7 +36,6 @@ from app.handlers.admin import (
user_messages as admin_user_messages,
updates as admin_updates,
backup as admin_backup,
bot_config as admin_bot_config,
welcome_text as admin_welcome_text,
tickets as admin_tickets,
reports as admin_reports,
@@ -137,7 +136,6 @@ async def setup_bot() -> tuple[Bot, Dispatcher]:
admin_campaigns.register_handlers(dp)
admin_maintenance.register_handlers(dp)
admin_user_messages.register_handlers(dp)
admin_bot_config.register_handlers(dp)
admin_updates.register_handlers(dp)
admin_backup.register_handlers(dp)
admin_welcome_text.register_welcome_text_handlers(dp)

View File

@@ -974,22 +974,14 @@ class Settings(BaseSettings):
settings = Settings()
PERIOD_PRICES: Dict[int, int] = {}
def refresh_period_prices() -> None:
PERIOD_PRICES.clear()
PERIOD_PRICES.update({
14: settings.PRICE_14_DAYS,
30: settings.PRICE_30_DAYS,
60: settings.PRICE_60_DAYS,
90: settings.PRICE_90_DAYS,
180: settings.PRICE_180_DAYS,
360: settings.PRICE_360_DAYS,
})
refresh_period_prices()
PERIOD_PRICES = {
14: settings.PRICE_14_DAYS,
30: settings.PRICE_30_DAYS,
60: settings.PRICE_60_DAYS,
90: settings.PRICE_90_DAYS,
180: settings.PRICE_180_DAYS,
360: settings.PRICE_360_DAYS,
}
def get_traffic_prices() -> Dict[int, int]:
packages = settings.get_traffic_packages()

View File

@@ -25,17 +25,6 @@ from sqlalchemy.sql import func
Base = declarative_base()
class BotConfig(Base):
__tablename__ = "bot_config"
key = Column(String(120), primary_key=True)
value = Column(Text, nullable=True)
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
def __repr__(self) -> str: # pragma: no cover - debug helper
return f"<BotConfig(key={self.key!r}, value={self.value!r})>"
server_squad_promo_groups = Table(
"server_squad_promo_groups",
Base.metadata,

View File

@@ -1,13 +1,10 @@
import logging
from sqlalchemy import text, inspect
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings, Settings
from app.database.database import engine
logger = logging.getLogger(__name__)
BOT_CONFIG_EXCLUDED_KEYS = {"BOT_TOKEN", "ADMIN_IDS"}
async def get_database_type():
return engine.dialect.name
@@ -43,94 +40,6 @@ async def check_table_exists(table_name: str) -> bool:
logger.error(f"Ошибка проверки существования таблицы {table_name}: {e}")
return False
def _serialize_config_value(value):
if value is None:
return None
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
async def create_bot_config_table() -> bool:
try:
async with engine.begin() as conn:
db_type = await get_database_type()
if db_type == "sqlite":
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS bot_config (
key TEXT PRIMARY KEY,
value TEXT NULL,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""))
elif db_type == "postgresql":
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS bot_config (
key VARCHAR(120) PRIMARY KEY,
value TEXT NULL,
updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT CURRENT_TIMESTAMP
)
"""))
elif db_type == "mysql":
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS bot_config (
`key` VARCHAR(120) PRIMARY KEY,
`value` TEXT NULL,
`updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
)
"""))
else:
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS bot_config (
key VARCHAR(120) PRIMARY KEY,
value TEXT NULL,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""))
return True
except Exception as e:
logger.error(f"Ошибка создания таблицы bot_config: {e}")
return False
async def sync_bot_config_defaults() -> bool:
try:
all_keys = [
key for key in Settings.model_fields.keys()
if key not in BOT_CONFIG_EXCLUDED_KEYS
]
async with engine.begin() as conn:
existing = await conn.execute(text("SELECT key FROM bot_config"))
existing_keys = {row[0] for row in existing.fetchall()}
inserted = 0
for key in all_keys:
if key in existing_keys:
continue
value = getattr(settings, key, None)
serialized = _serialize_config_value(value)
await conn.execute(
text("INSERT INTO bot_config (key, value) VALUES (:key, :value)"),
{"key": key, "value": serialized},
)
inserted += 1
if inserted:
logger.info(f"Добавлено {inserted} новых параметров в bot_config")
else:
logger.info("Конфигурация bot_config уже актуальна")
return True
except Exception as e:
logger.error(f"Ошибка синхронизации значений bot_config: {e}")
return False
async def check_column_exists(table_name: str, column_name: str) -> bool:
try:
async with engine.begin() as conn:
@@ -1934,19 +1843,7 @@ async def run_universal_migration():
referral_migration_success = await add_referral_system_columns()
if not referral_migration_success:
logger.warning("⚠️ Проблемы с миграцией реферальной системы")
logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ BOT_CONFIG ===")
bot_config_created = await create_bot_config_table()
if bot_config_created:
logger.info("✅ Таблица bot_config готова")
bot_config_synced = await sync_bot_config_defaults()
if bot_config_synced:
logger.info("✅ Конфигурация bot_config синхронизирована с текущими настройками")
else:
logger.warning("⚠️ Не удалось синхронизировать значения bot_config")
else:
logger.warning("⚠️ Проблемы с созданием таблицы bot_config")
logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ CRYPTOBOT ===")
cryptobot_created = await create_cryptobot_payments_table()
if cryptobot_created:

View File

@@ -1,368 +0,0 @@
import html
import logging
from typing import Optional
from aiogram import F, Router, types
from aiogram.exceptions import TelegramBadRequest
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.configuration_service import (
ConfigurationValidationError,
configuration_service,
)
from app.states import AdminStates
from app.utils.decorators import admin_required, error_handler
logger = logging.getLogger(__name__)
router = Router()
PER_PAGE = 6
async def _build_config_page(language: str, page: int) -> tuple[str, types.InlineKeyboardMarkup, int]:
data = await configuration_service.get_paginated_items(page, PER_PAGE)
items = data["items"]
total = data["total"]
total_pages = data["total_pages"]
current_page = data["page"]
texts = get_texts(language)
header = texts.t("ADMIN_BOT_CONFIG_TITLE", "🧩 <b>Конфигурация бота</b>")
hint = texts.t(
"ADMIN_BOT_CONFIG_HINT",
"Выберите параметр для изменения. Текущие значения указаны ниже.",
)
summary_template = texts.t("ADMIN_BOT_CONFIG_SUMMARY", "Всего параметров: {total}")
page_template = texts.t("ADMIN_BOT_CONFIG_PAGE_INFO", "Страница {current}/{total}")
empty_text = texts.t("ADMIN_BOT_CONFIG_EMPTY", "Доступных настроек не найдено.")
lines = [header, ""]
if total:
lines.append(summary_template.format(total=total))
lines.append(page_template.format(current=current_page, total=total_pages))
lines.append("")
lines.append(hint)
lines.append("")
else:
lines.append(empty_text)
keyboard_rows: list[list[types.InlineKeyboardButton]] = []
for item in items:
key = item["key"]
value_display = html.escape(item["display"])
type_display = html.escape(item["type"])
lines.append(f"• <code>{html.escape(key)}</code> = <b>{value_display}</b> <i>({type_display})</i>")
short_value = item["short_display"]
button_text = key
if short_value:
button_text = f"{key}{short_value}"
if len(button_text) > 64:
button_text = button_text[:61] + ""
keyboard_rows.append([
types.InlineKeyboardButton(
text=button_text,
callback_data=f"admin_bot_config_edit|{current_page}|{key}",
)
])
if total and total_pages > 1:
nav_row: list[types.InlineKeyboardButton] = []
if current_page > 1:
nav_row.append(
types.InlineKeyboardButton(
text="⬅️", callback_data=f"admin_bot_config_page_{current_page - 1}"
)
)
nav_row.append(
types.InlineKeyboardButton(
text=f"{current_page}/{total_pages}", callback_data="admin_bot_config_page_current"
)
)
if current_page < total_pages:
nav_row.append(
types.InlineKeyboardButton(
text="➡️", callback_data=f"admin_bot_config_page_{current_page + 1}"
)
)
keyboard_rows.append(nav_row)
keyboard_rows.append([
types.InlineKeyboardButton(
text=texts.BACK, callback_data="admin_submenu_settings"
)
])
keyboard = types.InlineKeyboardMarkup(inline_keyboard=keyboard_rows)
text = "\n".join(lines)
return text, keyboard, current_page
async def _edit_config_message(
bot,
chat_id: int,
message_id: int,
language: str,
page: int,
*,
business_connection_id: Optional[str] = None,
) -> None:
text, keyboard, _ = await _build_config_page(language, page)
edit_kwargs = {
"chat_id": chat_id,
"message_id": message_id,
"text": text,
"reply_markup": keyboard,
"parse_mode": "HTML",
}
if business_connection_id:
edit_kwargs["business_connection_id"] = business_connection_id
try:
await bot.edit_message_text(**edit_kwargs)
except TelegramBadRequest as exc:
if "message is not modified" in str(exc).lower():
return
logger.error("Не удалось обновить сообщение конфигурации: %s", exc)
def _map_validation_error(error: ConfigurationValidationError, language: str) -> str:
texts = get_texts(language)
if error.code == "invalid_bool":
return texts.t(
"ADMIN_BOT_CONFIG_ERROR_BOOL",
"❌ Значение должно быть true/false (доступно: true, false, 1, 0, yes, no).",
)
if error.code == "invalid_int":
return texts.t("ADMIN_BOT_CONFIG_ERROR_INT", "❌ Значение должно быть целым числом.")
if error.code == "invalid_float":
return texts.t(
"ADMIN_BOT_CONFIG_ERROR_FLOAT",
"❌ Значение должно быть числом. Используйте точку в качестве разделителя.",
)
if error.code == "invalid_value" or error.code == "not_optional":
return texts.t(
"ADMIN_BOT_CONFIG_ERROR_REQUIRED",
"❌ Это обязательный параметр, пустое значение недопустимо.",
)
if error.code == "unknown_setting":
return texts.t(
"ADMIN_BOT_CONFIG_ERROR_UNKNOWN",
"❌ Параметр не найден или недоступен.",
)
if error.code == "db_error":
return texts.t(
"ADMIN_BOT_CONFIG_ERROR_GENERIC",
"Не удалось обновить параметр. Попробуйте ещё раз.",
)
return texts.ERROR
@router.callback_query(F.data == "admin_bot_config")
@admin_required
@error_handler
async def show_bot_config(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
state: FSMContext,
):
await state.clear()
text, keyboard, _ = await _build_config_page(db_user.language, 1)
await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await callback.answer()
@router.callback_query(F.data.startswith("admin_bot_config_page_"))
@admin_required
@error_handler
async def paginate_bot_config(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
state: FSMContext,
):
data = callback.data.split("_")
try:
page = int(data[-1])
except ValueError:
page = 1
await state.clear()
text, keyboard, _ = await _build_config_page(db_user.language, page)
await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await callback.answer()
@router.callback_query(F.data == "admin_bot_config_page_current")
@admin_required
@error_handler
async def current_page_callback(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
):
await callback.answer()
@router.callback_query(F.data.startswith("admin_bot_config_edit|"))
@admin_required
@error_handler
async def start_edit_config(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
state: FSMContext,
):
_, page_str, key = callback.data.split("|", 2)
try:
page = int(page_str)
except ValueError:
page = 1
try:
item = await configuration_service.get_item(key)
except ConfigurationValidationError as error:
await callback.answer(
_map_validation_error(error, db_user.language),
show_alert=True,
)
return
texts = get_texts(db_user.language)
header = texts.t("ADMIN_BOT_CONFIG_EDIT_TITLE", "✏️ <b>Изменение параметра</b>")
key_label = texts.t("ADMIN_BOT_CONFIG_KEY_LABEL", "Ключ")
type_label = texts.t("ADMIN_BOT_CONFIG_TYPE_LABEL", "Тип")
current_label = texts.t("ADMIN_BOT_CONFIG_CURRENT_LABEL", "Текущее значение")
instructions = texts.t(
"ADMIN_BOT_CONFIG_EDIT_INSTRUCTIONS",
"Отправьте новое значение одним сообщением.",
)
optional_note = (
texts.t(
"ADMIN_BOT_CONFIG_OPTIONAL_RESET",
"Чтобы очистить значение, отправьте <code>null</code>.",
)
if item["optional"]
else texts.t(
"ADMIN_BOT_CONFIG_REQUIRED_NOTE",
"Этот параметр обязательный — пустое значение недоступно.",
)
)
message_lines = [
header,
"",
f"{key_label}: <code>{html.escape(key)}</code>",
f"{type_label}: <code>{html.escape(item['type'])}</code>",
f"{current_label}: <code>{html.escape(item['display'])}</code>",
"",
instructions,
optional_note,
]
back_keyboard = types.InlineKeyboardMarkup(
inline_keyboard=[
[
types.InlineKeyboardButton(
text=texts.BACK, callback_data=f"admin_bot_config_page_{page}"
)
]
]
)
await callback.message.edit_text(
"\n".join(message_lines),
reply_markup=back_keyboard,
parse_mode="HTML",
)
await state.update_data(
bot_config_key=key,
bot_config_page=page,
bot_config_message_chat=callback.message.chat.id,
bot_config_message_id=callback.message.message_id,
bot_config_language=db_user.language,
bot_config_business_connection_id=(
str(getattr(callback.message, "business_connection_id", None))
if getattr(callback.message, "business_connection_id", None) is not None
else None
),
)
await state.set_state(AdminStates.editing_bot_config_value)
await callback.answer()
@router.message(AdminStates.editing_bot_config_value)
@admin_required
@error_handler
async def handle_config_update(
message: types.Message,
db_user: User,
db: AsyncSession,
state: FSMContext,
):
if not message.text:
error_text = get_texts(db_user.language).t(
"ADMIN_BOT_CONFIG_ERROR_TEXT_REQUIRED",
"❌ Отправьте текстовое значение.",
)
await message.answer(error_text)
return
data = await state.get_data()
key = data.get("bot_config_key")
if not key:
await state.clear()
await message.answer(
get_texts(db_user.language).t(
"ADMIN_BOT_CONFIG_ERROR_UNKNOWN",
"❌ Параметр не найден или недоступен.",
)
)
return
try:
_, display = await configuration_service.update_setting(key, message.text)
except ConfigurationValidationError as error:
await message.answer(_map_validation_error(error, db_user.language))
return
success_text = get_texts(db_user.language).t(
"ADMIN_BOT_CONFIG_UPDATED",
"✅ Параметр <code>{key}</code> обновлён: <b>{value}</b>",
).format(key=html.escape(key), value=html.escape(display))
await message.answer(success_text, parse_mode="HTML")
chat_id = data.get("bot_config_message_chat")
message_id = data.get("bot_config_message_id")
language = data.get("bot_config_language", db_user.language)
page = data.get("bot_config_page", 1)
business_connection_id = data.get("bot_config_business_connection_id")
if chat_id and message_id:
await _edit_config_message(
message.bot,
chat_id,
message_id,
language,
page,
business_connection_id=business_connection_id,
)
await state.clear()
def register_handlers(dp):
dp.include_router(router)

View File

@@ -105,12 +105,6 @@ def get_admin_settings_submenu_keyboard(language: str = "ru") -> InlineKeyboardM
callback_data="admin_mon_settings"
)
],
[
InlineKeyboardButton(
text=texts.t("ADMIN_BOT_CONFIG", "🧩 Конфигурация бота"),
callback_data="admin_bot_config"
)
],
[
InlineKeyboardButton(text=texts.ADMIN_RULES, callback_data="admin_rules"),
InlineKeyboardButton(text="🔧 Техработы", callback_data="maintenance_panel")

View File

@@ -1,389 +0,0 @@
import asyncio
import logging
import math
from typing import Any, Dict, List, Optional, Tuple, Annotated, Union, get_args, get_origin
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from app.config import (
Settings,
refresh_period_prices,
refresh_traffic_prices,
settings,
)
from app.database.database import AsyncSessionLocal
from app.database.models import BotConfig
logger = logging.getLogger(__name__)
class ConfigurationValidationError(Exception):
def __init__(self, code: str, key: str):
super().__init__(code)
self.code = code
self.key = key
class ConfigurationService:
EXCLUDED_KEYS = {"BOT_TOKEN", "ADMIN_IDS"}
PERIOD_PRICE_KEYS = {
"PRICE_14_DAYS",
"PRICE_30_DAYS",
"PRICE_60_DAYS",
"PRICE_90_DAYS",
"PRICE_180_DAYS",
"PRICE_360_DAYS",
}
TRAFFIC_PRICE_KEYS = {
"TRAFFIC_PACKAGES_CONFIG",
"PRICE_TRAFFIC_5GB",
"PRICE_TRAFFIC_10GB",
"PRICE_TRAFFIC_25GB",
"PRICE_TRAFFIC_50GB",
"PRICE_TRAFFIC_100GB",
"PRICE_TRAFFIC_250GB",
"PRICE_TRAFFIC_500GB",
"PRICE_TRAFFIC_1000GB",
"PRICE_TRAFFIC_UNLIMITED",
}
DATABASE_KEYS = {
"DATABASE_URL",
"DATABASE_MODE",
"POSTGRES_HOST",
"POSTGRES_PORT",
"POSTGRES_DB",
"POSTGRES_USER",
"POSTGRES_PASSWORD",
"SQLITE_PATH",
}
BOOL_TRUE_VALUES = {"1", "true", "yes", "on", "enable", "enabled", "y", "да", "истина"}
BOOL_FALSE_VALUES = {"0", "false", "no", "off", "disable", "disabled", "n", "нет", "ложь"}
def __init__(self) -> None:
self._cache: Dict[str, Optional[str]] = {}
self._loaded = False
self._lock = asyncio.Lock()
async def ensure_loaded(self) -> None:
if self._loaded:
return
await self.load_and_apply()
async def load_and_apply(self) -> None:
async with self._lock:
try:
await self._load_from_db()
self._loaded = True
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Не удалось загрузить конфигурацию бота: %s", exc)
self._loaded = False
raise
async def _load_from_db(self) -> None:
async with AsyncSessionLocal() as session:
result = await session.execute(select(BotConfig))
records = result.scalars().all()
self._cache = {record.key: record.value for record in records}
await self._ensure_missing_keys()
overrides: Dict[str, Any] = {}
for key, stored in self._cache.items():
if not self.is_configurable(key):
continue
overrides[key] = self._deserialize_value(key, stored)
self._apply_overrides(overrides)
async def _ensure_missing_keys(self) -> None:
missing = [key for key in self._ordered_keys() if key not in self._cache]
if not missing:
return
async with AsyncSessionLocal() as session:
try:
for key in missing:
value = getattr(settings, key, None)
serialized = self._serialize_for_storage(key, value)
session.add(BotConfig(key=key, value=serialized))
self._cache[key] = serialized
await session.commit()
except Exception as exc: # pragma: no cover - defensive logging
await session.rollback()
logger.error("Не удалось сохранить значения по умолчанию bot_config: %s", exc)
raise
def _ordered_keys(self) -> List[str]:
return [key for key in Settings.model_fields.keys() if self.is_configurable(key)]
def is_configurable(self, key: str) -> bool:
return key not in self.EXCLUDED_KEYS and key in Settings.model_fields
async def get_paginated_items(self, page: int, per_page: int) -> Dict[str, Any]:
await self.ensure_loaded()
keys = self._ordered_keys()
total = len(keys)
total_pages = max(1, math.ceil(total / per_page)) if total else 1
page = max(1, min(page, total_pages))
start = (page - 1) * per_page
end = start + per_page
items: List[Dict[str, Any]] = []
for key in keys[start:end]:
value = getattr(settings, key, None)
display = self.format_value_for_display(key, value)
items.append(
{
"key": key,
"value": value,
"display": display,
"short_display": self._shorten_display(display),
"type": self.get_type_name(key),
"optional": self._is_optional(key),
}
)
return {
"items": items,
"total": total,
"total_pages": total_pages,
"page": page,
}
async def get_item(self, key: str) -> Dict[str, Any]:
await self.ensure_loaded()
if not self.is_configurable(key):
raise ConfigurationValidationError("unknown_setting", key)
value = getattr(settings, key, None)
return {
"key": key,
"value": value,
"display": self.format_value_for_display(key, value),
"type": self.get_type_name(key),
"optional": self._is_optional(key),
}
async def update_setting(self, key: str, raw_value: str) -> Tuple[Any, str]:
await self.ensure_loaded()
if not self.is_configurable(key):
raise ConfigurationValidationError("unknown_setting", key)
python_value = self._parse_user_input(key, raw_value)
serialized = self._serialize_for_storage(key, python_value)
async with AsyncSessionLocal() as session:
try:
record = await session.get(BotConfig, key)
if record is None:
record = BotConfig(key=key, value=serialized)
session.add(record)
else:
record.value = serialized
await session.commit()
except SQLAlchemyError as exc:
await session.rollback()
logger.error("Ошибка обновления bot_config %s: %s", key, exc)
raise ConfigurationValidationError("db_error", key) from exc
self._cache[key] = serialized
self._apply_overrides({key: python_value})
display = self.format_value_for_display(key, python_value)
return python_value, display
def format_value_for_display(self, key: str, value: Any) -> str:
if value is None:
return ""
if isinstance(value, bool):
return "✅ True" if value else "❌ False"
return str(value)
def get_type_name(self, key: str) -> str:
base_type, optional = self._resolve_type(key)
type_name = getattr(base_type, "__name__", str(base_type))
if optional:
return f"Optional[{type_name}]"
return type_name
def _shorten_display(self, text: str, limit: int = 24) -> str:
clean = " ".join(text.split())
if len(clean) > limit:
return clean[: limit - 1] + ""
return clean
def _parse_user_input(self, key: str, raw_value: Optional[str]) -> Any:
base_type, optional = self._resolve_type(key)
text = (raw_value or "").strip()
if optional and text.lower() in {"", "null", "none"}:
return None
if base_type is bool:
lowered = text.lower()
if lowered in self.BOOL_TRUE_VALUES:
return True
if lowered in self.BOOL_FALSE_VALUES:
return False
raise ConfigurationValidationError("invalid_bool", key)
if base_type is int:
try:
return int(text)
except ValueError as exc:
raise ConfigurationValidationError("invalid_int", key) from exc
if base_type is float:
try:
normalized = text.replace(",", ".")
return float(normalized)
except ValueError as exc:
raise ConfigurationValidationError("invalid_float", key) from exc
if not text and not optional and base_type is not str:
raise ConfigurationValidationError("invalid_value", key)
return raw_value or ""
def _serialize_for_storage(self, key: str, value: Any) -> Optional[str]:
if value is None:
return None
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def _deserialize_value(self, key: str, stored: Optional[str]) -> Any:
if stored is None:
return None if self._is_optional(key) else getattr(settings, key, None)
base_type, _ = self._resolve_type(key)
text = str(stored).strip()
if base_type is bool:
lowered = text.lower()
if lowered in self.BOOL_TRUE_VALUES:
return True
if lowered in self.BOOL_FALSE_VALUES:
return False
logger.warning("Неверное булево значение для %s: %s", key, stored)
return getattr(settings, key, None)
if base_type is int:
try:
return int(text)
except ValueError:
logger.warning("Неверное числовое значение для %s: %s", key, stored)
return getattr(settings, key, None)
if base_type is float:
try:
return float(text)
except ValueError:
logger.warning("Неверное значение с плавающей точкой для %s: %s", key, stored)
return getattr(settings, key, None)
return stored
def _apply_overrides(self, overrides: Dict[str, Any]) -> None:
if not overrides:
return
changed_keys = []
for key, value in overrides.items():
if not self.is_configurable(key):
continue
try:
setattr(settings, key, value)
changed_keys.append(key)
except Exception as exc: # pragma: no cover - defensive logging
logger.error("Не удалось применить параметр %s: %s", key, exc)
if not changed_keys:
return
if any(key in self.PERIOD_PRICE_KEYS for key in changed_keys):
refresh_period_prices()
if any(key in self.TRAFFIC_PRICE_KEYS for key in changed_keys):
refresh_traffic_prices()
if any(key in self.DATABASE_KEYS for key in changed_keys):
try:
settings.DATABASE_URL = settings.get_database_url()
except Exception as exc: # pragma: no cover - defensive logging
logger.warning("Не удалось обновить DATABASE_URL после изменения настроек: %s", exc)
def _resolve_type(self, key: str) -> Tuple[type, bool]:
field = Settings.model_fields.get(key)
optional = False
if field is None:
return str, optional
annotation = field.annotation
annotation, optional = self._unwrap_annotation(annotation)
if isinstance(annotation, type):
if issubclass(annotation, bool):
return bool, optional
if issubclass(annotation, int):
return int, optional
if issubclass(annotation, float):
return float, optional
if issubclass(annotation, str):
return str, optional
default_value = getattr(settings, key, None)
if isinstance(default_value, bool):
return bool, optional
if isinstance(default_value, int):
return int, optional
if isinstance(default_value, float):
return float, optional
if isinstance(default_value, str):
return str, optional
return str, optional
def _unwrap_annotation(self, annotation: Any) -> Tuple[Any, bool]:
origin = get_origin(annotation)
if origin is Annotated:
args = get_args(annotation)
if args:
return self._unwrap_annotation(args[0])
if origin is Union:
args = get_args(annotation)
non_none = [arg for arg in args if arg is not type(None)]
optional = len(non_none) != len(args)
if not non_none:
return str, True
if len(non_none) == 1:
inner, inner_optional = self._unwrap_annotation(non_none[0])
return inner, optional or inner_optional
return str, optional
if origin in {list, tuple, dict, set}:
return str, False
return annotation, False
def _is_optional(self, key: str) -> bool:
_, optional = self._resolve_type(key)
return optional
configuration_service = ConfigurationService()

View File

@@ -87,7 +87,6 @@ class AdminStates(StatesGroup):
editing_rules_page = State()
editing_notification_value = State()
editing_bot_config_value = State()
confirming_sync = State()

View File

@@ -141,27 +141,6 @@
"ADMIN_MESSAGES": "📨 Broadcasts",
"ADMIN_MONITORING": "🔍 Monitoring",
"ADMIN_MONITORING_SETTINGS": "⚙️ Monitoring settings",
"ADMIN_BOT_CONFIG": "🧩 Bot configuration",
"ADMIN_BOT_CONFIG_TITLE": "🧩 <b>Bot configuration</b>",
"ADMIN_BOT_CONFIG_HINT": "Choose a setting to edit. Current values are listed below.",
"ADMIN_BOT_CONFIG_SUMMARY": "Total settings: {total}",
"ADMIN_BOT_CONFIG_PAGE_INFO": "Page {current}/{total}",
"ADMIN_BOT_CONFIG_EMPTY": "No configurable settings found.",
"ADMIN_BOT_CONFIG_EDIT_TITLE": "✏️ <b>Edit setting</b>",
"ADMIN_BOT_CONFIG_KEY_LABEL": "Key",
"ADMIN_BOT_CONFIG_TYPE_LABEL": "Type",
"ADMIN_BOT_CONFIG_CURRENT_LABEL": "Current value",
"ADMIN_BOT_CONFIG_EDIT_INSTRUCTIONS": "Send the new value as a single message.",
"ADMIN_BOT_CONFIG_OPTIONAL_RESET": "Send <code>null</code> to clear the value.",
"ADMIN_BOT_CONFIG_REQUIRED_NOTE": "This setting is required — empty value is not allowed.",
"ADMIN_BOT_CONFIG_UPDATED": "✅ Setting <code>{key}</code> updated: <b>{value}</b>",
"ADMIN_BOT_CONFIG_ERROR_TEXT_REQUIRED": "❌ Please send a text value.",
"ADMIN_BOT_CONFIG_ERROR_UNKNOWN": "❌ Setting not found or unavailable.",
"ADMIN_BOT_CONFIG_ERROR_BOOL": "❌ The value must be true/false (accepted: true, false, 1, 0, yes, no).",
"ADMIN_BOT_CONFIG_ERROR_INT": "❌ The value must be an integer.",
"ADMIN_BOT_CONFIG_ERROR_FLOAT": "❌ The value must be a number. Use a dot as the decimal separator.",
"ADMIN_BOT_CONFIG_ERROR_REQUIRED": "❌ This setting is required and cannot be empty.",
"ADMIN_BOT_CONFIG_ERROR_GENERIC": "❌ Failed to update the setting. Please try again.",
"ADMIN_PANEL": "\n⚙ <b>Administration panel</b>\n\nSelect a section to manage:\n",
"ADMIN_PROMOCODES": "🎫 Promo codes",
"ADMIN_REFERRALS": "🤝 Referral program",

View File

@@ -6,27 +6,6 @@
"ADMIN_MESSAGES": "📨 Рассылки",
"ADMIN_MONITORING": "🔍 Мониторинг",
"ADMIN_MONITORING_SETTINGS": "⚙️ Настройки мониторинга",
"ADMIN_BOT_CONFIG": "🧩 Конфигурация бота",
"ADMIN_BOT_CONFIG_TITLE": "🧩 <b>Конфигурация бота</b>",
"ADMIN_BOT_CONFIG_HINT": "Выберите параметр для изменения. Текущие значения указаны ниже.",
"ADMIN_BOT_CONFIG_SUMMARY": "Всего параметров: {total}",
"ADMIN_BOT_CONFIG_PAGE_INFO": "Страница {current}/{total}",
"ADMIN_BOT_CONFIG_EMPTY": "Доступных настроек не найдено.",
"ADMIN_BOT_CONFIG_EDIT_TITLE": "✏️ <b>Изменение параметра</b>",
"ADMIN_BOT_CONFIG_KEY_LABEL": "Ключ",
"ADMIN_BOT_CONFIG_TYPE_LABEL": "Тип",
"ADMIN_BOT_CONFIG_CURRENT_LABEL": "Текущее значение",
"ADMIN_BOT_CONFIG_EDIT_INSTRUCTIONS": "Отправьте новое значение одним сообщением.",
"ADMIN_BOT_CONFIG_OPTIONAL_RESET": "Чтобы очистить значение, отправьте <code>null</code>.",
"ADMIN_BOT_CONFIG_REQUIRED_NOTE": "Этот параметр обязательный — пустое значение недоступно.",
"ADMIN_BOT_CONFIG_UPDATED": "✅ Параметр <code>{key}</code> обновлён: <b>{value}</b>",
"ADMIN_BOT_CONFIG_ERROR_TEXT_REQUIRED": "❌ Отправьте текстовое значение.",
"ADMIN_BOT_CONFIG_ERROR_UNKNOWN": "❌ Параметр не найден или недоступен.",
"ADMIN_BOT_CONFIG_ERROR_BOOL": "❌ Значение должно быть true/false (доступно: true, false, 1, 0, yes, no).",
"ADMIN_BOT_CONFIG_ERROR_INT": "❌ Значение должно быть целым числом.",
"ADMIN_BOT_CONFIG_ERROR_FLOAT": "❌ Значение должно быть числом. Используйте точку в качестве разделителя.",
"ADMIN_BOT_CONFIG_ERROR_REQUIRED": "❌ Это обязательный параметр, пустое значение недопустимо.",
"ADMIN_BOT_CONFIG_ERROR_GENERIC": "❌ Не удалось обновить параметр. Попробуйте ещё раз.",
"ADMIN_REPORTS": "📊 Отчеты",
"ADMIN_PANEL": "\n⚙ <b>Административная панель</b>\n\nВыберите раздел для управления:\n",
"ADMIN_PROMOCODES": "🎫 Промокоды",

14
main.py
View File

@@ -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.configuration_service import configuration_service
class GracefulExit:
@@ -74,25 +73,18 @@ async def main():
logger.info("🔧 Выполняем проверку и миграцию базы данных...")
try:
migration_success = await run_universal_migration()
if migration_success:
logger.info("✅ Миграция базы данных завершена успешно")
else:
logger.warning("⚠️ Миграция завершилась с предупреждениями, но продолжаем запуск")
except Exception as migration_error:
logger.error(f"❌ Ошибка выполнения миграции: {migration_error}")
logger.warning("⚠️ Продолжаем запуск без миграции")
else:
logger.info(" Миграция пропущена (SKIP_MIGRATION=true)")
logger.info("🧩 Загрузка конфигурации бота из базы...")
try:
await configuration_service.load_and_apply()
logger.info("✅ Конфигурация бота применена")
except Exception as config_error:
logger.error(f"Не удалось загрузить конфигурацию бота: {config_error}")
logger.info("🤖 Настройка бота...")
bot, dp = await setup_bot()