houseassassin/remnawave-bedolaga-telegram-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #51 from Fr1ngg/dev

Browse Source 
Бекапы/Восстановление
This commit is contained in:
Egor
2025-09-11 03:32:44 +03:00
committed by GitHub
parent 02106bd64b 548dd2fdc6
commit fdfd0e2355
9 changed files with 1284 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+20 -10
View File
@@ -221,6 +221,26 @@ MAINTENANCE_MESSAGE=Ведутся технические работы. Серв
DEFAULT_LANGUAGE=ru
AVAILABLE_LANGUAGES=ru,en
# ===== ДОПОЛНИТЕЛЬНЫЕ НАСТРОЙКИ =====
# Конфигурация приложений для гайда подключения
APP_CONFIG_PATH=app-config.json
ENABLE_DEEP_LINKS=true
APP_CONFIG_CACHE_TTL=3600
# ===== СИСТЕМА БЕКАПОВ =====
BACKUP_AUTO_ENABLED=true
BACKUP_INTERVAL_HOURS=24
BACKUP_TIME=03:00
BACKUP_MAX_KEEP=7
BACKUP_COMPRESSION=true
BACKUP_INCLUDE_LOGS=false
BACKUP_LOCATION=/app/data/backups
# ===== ПРОВЕРКА ОБНОВЛЕНИЙ БОТА =====
VERSION_CHECK_ENABLED=true
VERSION_CHECK_REPO=fr1ngg/remnawave-bedolaga-telegram-bot
VERSION_CHECK_INTERVAL_HOURS=1
# ===== ЛОГИРОВАНИЕ =====
LOG_LEVEL=INFO
LOG_FILE=logs/bot.log
@@ -229,13 +249,3 @@ LOG_FILE=logs/bot.log
DEBUG=false
WEBHOOK_URL=
WEBHOOK_PATH=/webhook
# ===== ДОПОЛНИТЕЛЬНЫЕ НАСТРОЙКИ =====
# Конфигурация приложений для гайда подключения
APP_CONFIG_PATH=app-config.json
ENABLE_DEEP_LINKS=true
APP_CONFIG_CACHE_TTL=3600
VERSION_CHECK_ENABLED=true
VERSION_CHECK_REPO=fr1ngg/remnawave-bedolaga-telegram-bot
VERSION_CHECK_INTERVAL_HOURS=1
README.md
+3 -1
View File
@@ -70,7 +70,9 @@ cp .env.example .env
nano .env # Заполни токены и настройки
# 3. Создай необходимые директории
mkdir -p logs data
mkdir -p ./logs ./data ./data/backups ./data/referral_qr
chmod -R 755 ./logs ./data
sudo chown -R 1000:1000 ./logs ./data
# 4. Запусти всё разом
docker compose up -d
app/bot.py
+3 -1
View File
@@ -25,10 +25,11 @@ from app.handlers.admin import (
statistics as admin_statistics, servers as admin_servers,
maintenance as admin_maintenance,
user_messages as admin_user_messages,
updates as admin_updates
updates as admin_updates, backup as admin_backup
)
from app.handlers.stars_payments import register_stars_handlers
logger = logging.getLogger(__name__)
@@ -104,6 +105,7 @@ async def setup_bot() -> tuple[Bot, Dispatcher]:
admin_maintenance.register_handlers(dp)
admin_user_messages.register_handlers(dp)
admin_updates.register_handlers(dp)
admin_backup.register_handlers(dp)
common.register_handlers(dp)
app/config.py
+8
View File
@@ -156,6 +156,14 @@ class Settings(BaseSettings):
VERSION_CHECK_ENABLED: bool = True
VERSION_CHECK_REPO: str = "fr1ngg/remnawave-bedolaga-telegram-bot"
VERSION_CHECK_INTERVAL_HOURS: int = 1
BACKUP_AUTO_ENABLED: bool = True
BACKUP_INTERVAL_HOURS: int = 24
BACKUP_TIME: str = "03:00"
BACKUP_MAX_KEEP: int = 7
BACKUP_COMPRESSION: bool = True
BACKUP_INCLUDE_LOGS: bool = False
BACKUP_LOCATION: str = "/app/data/backups"
@field_validator('LOG_FILE', mode='before')
@classmethod
app/handlers/admin/backup.py
+647
View File
@@ -0,0 +1,647 @@
import logging
import os
from datetime import datetime
from pathlib import Path
from aiogram import Dispatcher, types, F
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import User
from app.services.backup_service import backup_service
from app.utils.decorators import admin_required, error_handler
logger = logging.getLogger(__name__)
class BackupStates(StatesGroup):
waiting_backup_file = State()
waiting_settings_update = State()
def get_backup_main_keyboard(language: str = "ru"):
return InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="🚀 Создать бекап", callback_data="backup_create"),
InlineKeyboardButton(text="📥 Восстановить", callback_data="backup_restore")
],
[
InlineKeyboardButton(text="📋 Список бекапов", callback_data="backup_list"),
InlineKeyboardButton(text="⚙️ Настройки", callback_data="backup_settings")
],
[
InlineKeyboardButton(text="◀️ Назад", callback_data="admin_panel")
]
])
def get_backup_list_keyboard(backups: list, page: int = 1, per_page: int = 5):
keyboard = []
start_idx = (page - 1) * per_page
end_idx = start_idx + per_page
page_backups = backups[start_idx:end_idx]
for backup in page_backups:
try:
if backup.get("timestamp"):
dt = datetime.fromisoformat(backup["timestamp"].replace('Z', '+00:00'))
date_str = dt.strftime("%d.%m %H:%M")
else:
date_str = "?"
except:
date_str = "?"
size_str = f"{backup.get('file_size_mb', 0):.1f}MB"
records_str = backup.get('total_records', '?')
button_text = f"📦 {date_str} • {size_str} • {records_str} записей"
callback_data = f"backup_manage_{backup['filename']}"
keyboard.append([InlineKeyboardButton(text=button_text, callback_data=callback_data)])
if len(backups) > per_page:
total_pages = (len(backups) + per_page - 1) // per_page
nav_row = []
if page > 1:
nav_row.append(InlineKeyboardButton(text="⬅️", callback_data=f"backup_list_page_{page-1}"))
nav_row.append(InlineKeyboardButton(text=f"{page}/{total_pages}", callback_data="noop"))
if page < total_pages:
nav_row.append(InlineKeyboardButton(text="➡️", callback_data=f"backup_list_page_{page+1}"))
keyboard.append(nav_row)
keyboard.extend([
[InlineKeyboardButton(text="◀️ Назад", callback_data="backup_panel")]
])
return InlineKeyboardMarkup(inline_keyboard=keyboard)
def get_backup_manage_keyboard(backup_filename: str):
return InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="📥 Восстановить", callback_data=f"backup_restore_file_{backup_filename}")
],
[
InlineKeyboardButton(text="🗑️ Удалить", callback_data=f"backup_delete_{backup_filename}")
],
[
InlineKeyboardButton(text="◀️ К списку", callback_data="backup_list")
]
])
def get_backup_settings_keyboard(settings_obj):
auto_status = "✅ Включены" if settings_obj.auto_backup_enabled else "❌ Отключены"
compression_status = "✅ Включено" if settings_obj.compression_enabled else "❌ Отключено"
logs_status = "✅ Включены" if settings_obj.include_logs else "❌ Отключены"
return InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(
text=f"🔄 Автобекапы: {auto_status}",
callback_data="backup_toggle_auto"
)
],
[
InlineKeyboardButton(
text=f"🗜️ Сжатие: {compression_status}",
callback_data="backup_toggle_compression"
)
],
[
InlineKeyboardButton(
text=f"📋 Логи в бекапе: {logs_status}",
callback_data="backup_toggle_logs"
)
],
[
InlineKeyboardButton(text="◀️ Назад", callback_data="backup_panel")
]
])
@admin_required
@error_handler
async def show_backup_panel(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
settings_obj = await backup_service.get_backup_settings()
status_auto = "✅ Включены" if settings_obj.auto_backup_enabled else "❌ Отключены"
text = f"""🗄️ <b>СИСТЕМА БЕКАПОВ</b>
📊 <b>Статус:</b>
• Автобекапы: {status_auto}
• Интервал: {settings_obj.backup_interval_hours} часов
• Хранить: {settings_obj.max_backups_keep} файлов
• Сжатие: {'Да' if settings_obj.compression_enabled else 'Нет'}
📁 <b>Расположение:</b> <code>/app/data/backups</code>
⚡ <b>Доступные операции:</b>
• Создание полного бекапа всех данных
• Восстановление из файла бекапа
• Управление автоматическими бекапами
"""
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=get_backup_main_keyboard(db_user.language)
)
await callback.answer()
@admin_required
@error_handler
async def create_backup_handler(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
await callback.answer("🔄 Создание бекапа запущено...")
progress_msg = await callback.message.edit_text(
"🔄 <b>Создание бекапа...</b>\n\n"
"⏳ Экспортируем данные из базы...\n"
"Это может занять несколько минут.",
parse_mode="HTML"
)
# Создаем бекап
success, message, file_path = await backup_service.create_backup(
created_by=db_user.telegram_id,
compress=True
)
if success:
await progress_msg.edit_text(
f"✅ <b>Бекап создан успешно!</b>\n\n{message}",
parse_mode="HTML",
reply_markup=get_backup_main_keyboard(db_user.language)
)
else:
await progress_msg.edit_text(
f"❌ <b>Ошибка создания бекапа</b>\n\n{message}",
parse_mode="HTML",
reply_markup=get_backup_main_keyboard(db_user.language)
)
@admin_required
@error_handler
async def show_backup_list(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
page = 1
if callback.data.startswith("backup_list_page_"):
try:
page = int(callback.data.split("_")[-1])
except:
page = 1
backups = await backup_service.get_backup_list()
if not backups:
text = "📦 <b>Список бекапов пуст</b>\n\nБекапы еще не создавались."
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="🚀 Создать первый бекап", callback_data="backup_create")],
[InlineKeyboardButton(text="◀️ Назад", callback_data="backup_panel")]
])
else:
text = f"📦 <b>Список бекапов</b> (всего: {len(backups)})\n\n"
text += "Выберите бекап для управления:"
keyboard = get_backup_list_keyboard(backups, page)
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=keyboard
)
await callback.answer()
@admin_required
@error_handler
async def manage_backup_file(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
filename = callback.data.replace("backup_manage_", "")
backups = await backup_service.get_backup_list()
backup_info = None
for backup in backups:
if backup["filename"] == filename:
backup_info = backup
break
if not backup_info:
await callback.answer("❌ Файл бекапа не найден", show_alert=True)
return
try:
if backup_info.get("timestamp"):
dt = datetime.fromisoformat(backup_info["timestamp"].replace('Z', '+00:00'))
date_str = dt.strftime("%d.%m.%Y %H:%M:%S")
else:
date_str = "Неизвестно"
except:
date_str = "Ошибка формата даты"
text = f"""📦 <b>Информация о бекапе</b>
📄 <b>Файл:</b> <code>{filename}</code>
📅 <b>Создан:</b> {date_str}
💾 <b>Размер:</b> {backup_info.get('file_size_mb', 0):.2f} MB
📊 <b>Таблиц:</b> {backup_info.get('tables_count', '?')}
📈 <b>Записей:</b> {backup_info.get('total_records', '?'):,}
🗜️ <b>Сжатие:</b> {'Да' if backup_info.get('compressed') else 'Нет'}
🗄️ <b>БД:</b> {backup_info.get('database_type', 'unknown')}
"""
if backup_info.get("error"):
text += f"\n⚠️ <b>Ошибка:</b> {backup_info['error']}"
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=get_backup_manage_keyboard(filename)
)
await callback.answer()
@admin_required
@error_handler
async def delete_backup_confirm(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
filename = callback.data.replace("backup_delete_", "")
text = f"🗑️ <b>Удаление бекапа</b>\n\n"
text += f"Вы уверены, что хотите удалить бекап?\n\n"
text += f"📄 <code>{filename}</code>\n\n"
text += "⚠️ <b>Это действие нельзя отменить!</b>"
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="✅ Да, удалить", callback_data=f"backup_delete_confirm_{filename}"),
InlineKeyboardButton(text="❌ Отмена", callback_data=f"backup_manage_{filename}")
]
])
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=keyboard
)
await callback.answer()
@admin_required
@error_handler
async def delete_backup_execute(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
filename = callback.data.replace("backup_delete_confirm_", "")
success, message = await backup_service.delete_backup(filename)
if success:
await callback.message.edit_text(
f"✅ <b>Бекап удален</b>\n\n{message}",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="📋 К списку бекапов", callback_data="backup_list")]
])
)
else:
await callback.message.edit_text(
f"❌ <b>Ошибка удаления</b>\n\n{message}",
parse_mode="HTML",
reply_markup=get_backup_manage_keyboard(filename)
)
await callback.answer()
@admin_required
@error_handler
async def restore_backup_start(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession,
state: FSMContext
):
if callback.data.startswith("backup_restore_file_"):
# Восстановление из конкретного файла
filename = callback.data.replace("backup_restore_file_", "")
text = f"📥 <b>Восстановление из бекапа</b>\n\n"
text += f"📄 <b>Файл:</b> <code>{filename}</code>\n\n"
text += "⚠️ <b>ВНИМАНИЕ!</b>\n"
text += "• Процесс может занять несколько минут\n"
text += "• Рекомендуется создать бекап перед восстановлением\n"
text += "• Существующие данные будут дополнены\n\n"
text += "Продолжить восстановление?"
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="✅ Да, восстановить", callback_data=f"backup_restore_execute_{filename}"),
InlineKeyboardButton(text="🗑️ Очистить и восстановить", callback_data=f"backup_restore_clear_{filename}")
],
[
InlineKeyboardButton(text="❌ Отмена", callback_data=f"backup_manage_{filename}")
]
])
else:
text = """📥 <b>Восстановление из бекапа</b>
📎 Отправьте файл бекапа (.json или .json.gz)
⚠️ <b>ВАЖНО:</b>
• Файл должен быть создан этой системой бекапов
• Процесс может занять несколько минут
• Рекомендуется создать бекап перед восстановлением
💡 Или выберите из существующих бекапов ниже."""
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="📋 Выбрать из списка", callback_data="backup_list")],
[InlineKeyboardButton(text="❌ Отмена", callback_data="backup_panel")]
])
await state.set_state(BackupStates.waiting_backup_file)
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=keyboard
)
await callback.answer()
@admin_required
@error_handler
async def restore_backup_execute(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
if callback.data.startswith("backup_restore_execute_"):
filename = callback.data.replace("backup_restore_execute_", "")
clear_existing = False
elif callback.data.startswith("backup_restore_clear_"):
filename = callback.data.replace("backup_restore_clear_", "")
clear_existing = True
else:
await callback.answer("❌ Неверный формат команды", show_alert=True)
return
await callback.answer("🔄 Восстановление запущено...")
# Показываем прогресс
action_text = "очисткой и восстановлением" if clear_existing else "восстановлением"
progress_msg = await callback.message.edit_text(
f"📥 <b>Восстановление из бекапа...</b>\n\n"
f"⏳ Работаем с {action_text} данных...\n"
f"📄 Файл: <code>{filename}</code>\n\n"
f"Это может занять несколько минут.",
parse_mode="HTML"
)
backup_path = backup_service.backup_dir / filename
success, message = await backup_service.restore_backup(
str(backup_path),
clear_existing=clear_existing
)
if success:
await progress_msg.edit_text(
f"✅ <b>Восстановление завершено!</b>\n\n{message}",
parse_mode="HTML",
reply_markup=get_backup_main_keyboard(db_user.language)
)
else:
await progress_msg.edit_text(
f"❌ <b>Ошибка восстановления</b>\n\n{message}",
parse_mode="HTML",
reply_markup=get_backup_manage_keyboard(filename)
)
@admin_required
@error_handler
async def handle_backup_file_upload(
message: types.Message,
db_user: User,
db: AsyncSession,
state: FSMContext
):
if not message.document:
await message.answer(
"❌ Пожалуйста, отправьте файл бекапа (.json или .json.gz)",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="◀️ Отмена", callback_data="backup_panel")]
])
)
return
document = message.document
if not (document.file_name.endswith('.json') or document.file_name.endswith('.json.gz')):
await message.answer(
"❌ Неподдерживаемый формат файла. Загрузите .json или .json.gz файл",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="◀️ Отмена", callback_data="backup_panel")]
])
)
return
if document.file_size > 50 * 1024 * 1024:
await message.answer(
"❌ Файл слишком большой (максимум 50MB)",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="◀️ Отмена", callback_data="backup_panel")]
])
)
return
try:
file = await message.bot.get_file(document.file_id)
temp_path = backup_service.backup_dir / f"uploaded_{document.file_name}"
await message.bot.download_file(file.file_path, temp_path)
text = f"""📥 <b>Файл загружен</b>
📄 <b>Имя:</b> <code>{document.file_name}</code>
💾 <b>Размер:</b> {document.file_size / 1024 / 1024:.2f} MB
⚠️ <b>ВНИМАНИЕ!</b>
Процесс восстановления изменит данные в базе.
Рекомендуется создать бекап перед восстановлением.
Продолжить?"""
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="✅ Восстановить", callback_data=f"backup_restore_uploaded_{temp_path.name}"),
InlineKeyboardButton(text="🗑️ Очистить и восстановить", callback_data=f"backup_restore_uploaded_clear_{temp_path.name}")
],
[
InlineKeyboardButton(text="❌ Отмена", callback_data="backup_panel")
]
])
await message.answer(text, parse_mode="HTML", reply_markup=keyboard)
await state.clear()
except Exception as e:
logger.error(f"Ошибка загрузки файла бекапа: {e}")
await message.answer(
f"❌ Ошибка загрузки файла: {str(e)}",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="◀️ Отмена", callback_data="backup_panel")]
])
)
@admin_required
@error_handler
async def show_backup_settings(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
settings_obj = await backup_service.get_backup_settings()
text = f"""⚙️ <b>Настройки системы бекапов</b>
🔄 <b>Автоматические бекапы:</b>
• Статус: {'✅ Включены' if settings_obj.auto_backup_enabled else '❌ Отключены'}
• Интервал: {settings_obj.backup_interval_hours} часов
• Время запуска: {settings_obj.backup_time}
📦 <b>Хранение:</b>
• Максимум файлов: {settings_obj.max_backups_keep}
• Сжатие: {'✅ Включено' if settings_obj.compression_enabled else '❌ Отключено'}
• Включать логи: {'✅ Да' if settings_obj.include_logs else '❌ Нет'}
📁 <b>Расположение:</b> <code>{settings_obj.backup_location}</code>
"""
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=get_backup_settings_keyboard(settings_obj)
)
await callback.answer()
@admin_required
@error_handler
async def toggle_backup_setting(
callback: types.CallbackQuery,
db_user: User,
db: AsyncSession
):
settings_obj = await backup_service.get_backup_settings()
if callback.data == "backup_toggle_auto":
new_value = not settings_obj.auto_backup_enabled
await backup_service.update_backup_settings(auto_backup_enabled=new_value)
status = "включены" if new_value else "отключены"
await callback.answer(f"Автобекапы {status}")
elif callback.data == "backup_toggle_compression":
new_value = not settings_obj.compression_enabled
await backup_service.update_backup_settings(compression_enabled=new_value)
status = "включено" if new_value else "отключено"
await callback.answer(f"Сжатие {status}")
elif callback.data == "backup_toggle_logs":
new_value = not settings_obj.include_logs
await backup_service.update_backup_settings(include_logs=new_value)
status = "включены" if new_value else "отключены"
await callback.answer(f"Логи в бекапе {status}")
await show_backup_settings(callback, db_user, db)
def register_handlers(dp: Dispatcher):
dp.callback_query.register(
show_backup_panel,
F.data == "backup_panel"
)
dp.callback_query.register(
create_backup_handler,
F.data == "backup_create"
)
dp.callback_query.register(
show_backup_list,
F.data.startswith("backup_list")
)
dp.callback_query.register(
manage_backup_file,
F.data.startswith("backup_manage_")
)
dp.callback_query.register(
delete_backup_confirm,
F.data.startswith("backup_delete_") & ~F.data.startswith("backup_delete_confirm_")
)
dp.callback_query.register(
delete_backup_execute,
F.data.startswith("backup_delete_confirm_")
)
dp.callback_query.register(
restore_backup_start,
F.data.in_(["backup_restore"]) | F.data.startswith("backup_restore_file_")
)
dp.callback_query.register(
restore_backup_execute,
F.data.startswith("backup_restore_execute_") | F.data.startswith("backup_restore_clear_")
)
dp.callback_query.register(
show_backup_settings,
F.data == "backup_settings"
)
dp.callback_query.register(
toggle_backup_setting,
F.data.in_(["backup_toggle_auto", "backup_toggle_compression", "backup_toggle_logs"])
)
dp.message.register(
handle_backup_file_upload,
BackupStates.waiting_backup_file
)
app/keyboards/admin.py
+6
View File
@@ -32,6 +32,9 @@ def get_admin_main_keyboard(language: str = "ru") -> InlineKeyboardMarkup:
InlineKeyboardButton(text="📢 Сообщения в меню", callback_data="user_messages_panel"),
InlineKeyboardButton(text="🔄 Обновления", callback_data="admin_updates")
],
[
InlineKeyboardButton(text="🗄️ Бекапы", callback_data="backup_panel")
],
[
InlineKeyboardButton(text=texts.BACK, callback_data="back_to_menu")
]
@@ -660,3 +663,6 @@ def get_sync_simplified_keyboard(language: str = "ru") -> InlineKeyboardMarkup:
]
return InlineKeyboardMarkup(inline_keyboard=keyboard)
app/services/backup_service.py
+574
View File
@@ -0,0 +1,574 @@
import asyncio
import json as json_lib
import logging
import gzip
import os
import tempfile
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, Any, Optional, List, Tuple
from dataclasses import dataclass, asdict
import aiofiles
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, text, inspect
from sqlalchemy.orm import selectinload
from app.config import settings
from app.database.database import get_db, engine
from app.database.models import (
User, Subscription, Transaction, PromoCode, PromoCodeUse,
ReferralEarning, Squad, ServiceRule, SystemSetting, MonitoringLog,
SubscriptionConversion, SentNotification, BroadcastHistory,
ServerSquad, SubscriptionServer, UserMessage, YooKassaPayment,
CryptoBotPayment, Base
)
logger = logging.getLogger(__name__)
@dataclass
class BackupMetadata:
timestamp: str
version: str = "1.0"
database_type: str = "postgresql"
backup_type: str = "full"
tables_count: int = 0
total_records: int = 0
compressed: bool = True
file_size_bytes: int = 0
created_by: Optional[int] = None
@dataclass
class BackupSettings:
auto_backup_enabled: bool = True
backup_interval_hours: int = 24
backup_time: str = "03:00"
max_backups_keep: int = 7
compression_enabled: bool = True
include_logs: bool = False
backup_location: str = "/app/data/backups"
class BackupService:
def __init__(self, bot=None):
self.bot = bot
self.backup_dir = Path(settings.SQLITE_PATH).parent / "backups"
self.backup_dir.mkdir(exist_ok=True)
self._auto_backup_task = None
self._settings = self._load_settings()
self.backup_models = [
User, Subscription, Transaction, PromoCode, PromoCodeUse,
ReferralEarning, ServiceRule, SystemSetting,
SubscriptionConversion, SentNotification, BroadcastHistory,
ServerSquad, SubscriptionServer, UserMessage,
YooKassaPayment, CryptoBotPayment
]
if self._settings.include_logs:
self.backup_models.append(MonitoringLog)
def _load_settings(self) -> BackupSettings:
return BackupSettings(
auto_backup_enabled=os.getenv("BACKUP_AUTO_ENABLED", "true").lower() == "true",
backup_interval_hours=int(os.getenv("BACKUP_INTERVAL_HOURS", "24")),
backup_time=os.getenv("BACKUP_TIME", "03:00"),
max_backups_keep=int(os.getenv("BACKUP_MAX_KEEP", "7")),
compression_enabled=os.getenv("BACKUP_COMPRESSION", "true").lower() == "true",
include_logs=os.getenv("BACKUP_INCLUDE_LOGS", "false").lower() == "true",
backup_location=os.getenv("BACKUP_LOCATION", "/app/data/backups")
)
async def create_backup(
self,
created_by: Optional[int] = None,
compress: bool = True,
include_logs: bool = None
) -> Tuple[bool, str, Optional[str]]:
try:
logger.info("🔄 Начинаем создание бекапа...")
if include_logs is None:
include_logs = self._settings.include_logs
models_to_backup = self.backup_models.copy()
if not include_logs and MonitoringLog in models_to_backup:
models_to_backup.remove(MonitoringLog)
elif include_logs and MonitoringLog not in models_to_backup:
models_to_backup.append(MonitoringLog)
backup_data = {}
total_records = 0
async for db in get_db():
try:
for model in models_to_backup:
table_name = model.__tablename__
logger.info(f"📊 Экспортируем таблицу: {table_name}")
result = await db.execute(select(model))
records = result.scalars().all()
table_data = []
for record in records:
record_dict = {}
for column in model.__table__.columns:
value = getattr(record, column.name)
if isinstance(value, datetime):
record_dict[column.name] = value.isoformat()
elif hasattr(value, '__dict__'):
record_dict[column.name] = str(value)
else:
record_dict[column.name] = value
table_data.append(record_dict)
backup_data[table_name] = table_data
total_records += len(table_data)
logger.info(f"✅ Экспортировано {len(table_data)} записей из {table_name}")
break
except Exception as e:
logger.error(f"Ошибка при экспорте данных: {e}")
raise e
finally:
await db.close()
metadata = BackupMetadata(
timestamp=datetime.utcnow().isoformat(),
database_type="postgresql" if settings.is_postgresql() else "sqlite",
backup_type="full",
tables_count=len(models_to_backup),
total_records=total_records,
compressed=compress,
created_by=created_by,
file_size_bytes=0
)
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
filename = f"backup_{timestamp}.json"
if compress:
filename += ".gz"
backup_path = self.backup_dir / filename
backup_structure = {
"metadata": asdict(metadata),
"data": backup_data
}
if compress:
backup_json_str = json_lib.dumps(backup_structure, ensure_ascii=False, indent=2)
async with aiofiles.open(backup_path, 'wb') as f:
compressed_data = gzip.compress(backup_json_str.encode('utf-8'))
await f.write(compressed_data)
else:
async with aiofiles.open(backup_path, 'w', encoding='utf-8') as f:
await f.write(json_lib.dumps(backup_structure, ensure_ascii=False, indent=2))
file_size = backup_path.stat().st_size
backup_structure["metadata"]["file_size_bytes"] = file_size
if compress:
backup_json_str = json_lib.dumps(backup_structure, ensure_ascii=False, indent=2)
async with aiofiles.open(backup_path, 'wb') as f:
compressed_data = gzip.compress(backup_json_str.encode('utf-8'))
await f.write(compressed_data)
else:
async with aiofiles.open(backup_path, 'w', encoding='utf-8') as f:
await f.write(json_lib.dumps(backup_structure, ensure_ascii=False, indent=2))
await self._cleanup_old_backups()
size_mb = file_size / 1024 / 1024
message = (f"✅ Бекап успешно создан!\n"
f"📁 Файл: {filename}\n"
f"📊 Таблиц: {len(models_to_backup)}\n"
f"📈 Записей: {total_records:,}\n"
f"💾 Размер: {size_mb:.2f} MB")
logger.info(message)
if self.bot:
await self._send_backup_notification(
"success", message, str(backup_path)
)
return True, message, str(backup_path)
except Exception as e:
error_msg = f"❌ Ошибка создания бекапа: {str(e)}"
logger.error(error_msg, exc_info=True)
if self.bot:
await self._send_backup_notification("error", error_msg)
return False, error_msg, None
async def restore_backup(
self,
backup_file_path: str,
clear_existing: bool = False
) -> Tuple[bool, str]:
try:
logger.info(f"🔄 Начинаем восстановление из {backup_file_path}")
backup_path = Path(backup_file_path)
if not backup_path.exists():
return False, f"❌ Файл бекапа не найден: {backup_file_path}"
if backup_path.suffix == '.gz':
async with aiofiles.open(backup_path, 'rb') as f:
compressed_data = await f.read()
uncompressed_data = gzip.decompress(compressed_data).decode('utf-8')
backup_structure = json_lib.loads(uncompressed_data)
else:
async with aiofiles.open(backup_path, 'r', encoding='utf-8') as f:
file_content = await f.read()
backup_structure = json_lib.loads(file_content)
metadata = backup_structure.get("metadata", {})
backup_data = backup_structure.get("data", {})
if not backup_data:
return False, "❌ Файл бекапа не содержит данных"
logger.info(f"📊 Загружен бекап от {metadata.get('timestamp')}")
logger.info(f"📈 Содержит {metadata.get('total_records', 0)} записей")
restored_records = 0
restored_tables = 0
async for db in get_db():
try:
if clear_existing:
logger.warning("🗑️ Очищаем существующие данные...")
await self._clear_database_tables(db)
for table_name, records in backup_data.items():
if not records:
continue
model = None
for m in self.backup_models:
if m.__tablename__ == table_name:
model = m
break
if not model:
logger.warning(f"⚠️ Модель для таблицы {table_name} не найдена, пропускаем")
continue
logger.info(f"📥 Восстанавливаем таблицу {table_name} ({len(records)} записей)")
for record_data in records:
try:
processed_data = {}
for key, value in record_data.items():
if value is None:
processed_data[key] = None
continue
column = getattr(model.__table__.columns, key, None)
if column is None:
continue
column_type_str = str(column.type).upper()
if ('DATETIME' in column_type_str or 'TIMESTAMP' in column_type_str) and isinstance(value, str):
try:
if 'T' in value:
processed_data[key] = datetime.fromisoformat(value.replace('Z', '+00:00'))
else:
processed_data[key] = datetime.strptime(value, '%Y-%m-%d %H:%M:%S')
except (ValueError, TypeError) as e:
logger.warning(f"Не удалось парсить дату {value} для поля {key}: {e}")
processed_data[key] = datetime.utcnow()
elif ('BOOLEAN' in column_type_str or 'BOOL' in column_type_str) and isinstance(value, str):
processed_data[key] = value.lower() in ('true', '1', 'yes', 'on')
elif ('INTEGER' in column_type_str or 'INT' in column_type_str) and isinstance(value, str):
try:
processed_data[key] = int(value)
except ValueError:
processed_data[key] = 0
elif ('FLOAT' in column_type_str or 'REAL' in column_type_str or 'NUMERIC' in column_type_str) and isinstance(value, str):
try:
processed_data[key] = float(value)
except ValueError:
processed_data[key] = 0.0
elif 'JSON' in column_type_str and isinstance(value, str):
try:
processed_data[key] = json_lib.loads(value)
except (ValueError, TypeError):
processed_data[key] = value
else:
processed_data[key] = value
# Проверяем существует ли запись с таким ID
primary_key_col = None
for col in model.__table__.columns:
if col.primary_key:
primary_key_col = col.name
break
if primary_key_col and primary_key_col in processed_data:
# Проверяем существование записи
existing_record = await db.execute(
select(model).where(
getattr(model, primary_key_col) == processed_data[primary_key_col]
)
)
existing = existing_record.scalar_one_or_none()
if existing:
# Обновляем существующую запись
for key, value in processed_data.items():
if key != primary_key_col: # Не обновляем primary key
setattr(existing, key, value)
logger.debug(f"Обновлена существующая запись {primary_key_col}={processed_data[primary_key_col]} в {table_name}")
else:
# Создаем новую запись
instance = model(**processed_data)
db.add(instance)
else:
# Если нет primary key или он не в данных, просто добавляем
instance = model(**processed_data)
db.add(instance)
restored_records += 1
except Exception as e:
logger.error(f"Ошибка восстановления записи в {table_name}: {e}")
continue
restored_tables += 1
logger.info(f"✅ Таблица {table_name} восстановлена")
await db.commit()
break
except Exception as e:
await db.rollback()
logger.error(f"Ошибка при восстановлении: {e}")
raise e
finally:
await db.close()
message = (f"✅ Восстановление завершено!\n"
f"📊 Таблиц: {restored_tables}\n"
f"📈 Записей: {restored_records:,}\n"
f"📅 Дата бекапа: {metadata.get('timestamp', 'неизвестно')}")
logger.info(message)
if self.bot:
await self._send_backup_notification("restore_success", message)
return True, message
except Exception as e:
error_msg = f"❌ Ошибка восстановления: {str(e)}"
logger.error(error_msg, exc_info=True)
if self.bot:
await self._send_backup_notification("restore_error", error_msg)
return False, error_msg
async def _clear_database_tables(self, db: AsyncSession):
tables_order = [
"subscription_servers", "sent_notifications", "broadcast_history",
"subscription_conversions", "referral_earnings", "promocode_uses",
"transactions", "yookassa_payments", "cryptobot_payments",
"subscriptions", "users", "promocodes", "server_squads",
"service_rules", "system_settings", "monitoring_logs", "user_messages"
]
for table_name in tables_order:
try:
await db.execute(text(f"DELETE FROM {table_name}"))
logger.info(f"🗑️ Очищена таблица {table_name}")
except Exception as e:
logger.warning(f"⚠️ Не удалось очистить таблицу {table_name}: {e}")
async def get_backup_list(self) -> List[Dict[str, Any]]:
backups = []
try:
for backup_file in sorted(self.backup_dir.glob("backup_*.json*"), reverse=True):
try:
if backup_file.suffix == '.gz':
with gzip.open(backup_file, 'rt', encoding='utf-8') as f:
backup_structure = json_lib.load(f)
else:
with open(backup_file, 'r', encoding='utf-8') as f:
backup_structure = json_lib.load(f)
metadata = backup_structure.get("metadata", {})
file_stats = backup_file.stat()
backup_info = {
"filename": backup_file.name,
"filepath": str(backup_file),
"timestamp": metadata.get("timestamp"),
"tables_count": metadata.get("tables_count", 0),
"total_records": metadata.get("total_records", 0),
"compressed": metadata.get("compressed", False),
"file_size_bytes": file_stats.st_size,
"file_size_mb": round(file_stats.st_size / 1024 / 1024, 2),
"created_by": metadata.get("created_by"),
"database_type": metadata.get("database_type", "unknown")
}
backups.append(backup_info)
except Exception as e:
logger.error(f"Ошибка чтения метаданных {backup_file}: {e}")
file_stats = backup_file.stat()
backups.append({
"filename": backup_file.name,
"filepath": str(backup_file),
"timestamp": datetime.fromtimestamp(file_stats.st_mtime).isoformat(),
"tables_count": "?",
"total_records": "?",
"compressed": backup_file.suffix == '.gz',
"file_size_bytes": file_stats.st_size,
"file_size_mb": round(file_stats.st_size / 1024 / 1024, 2),
"created_by": None,
"database_type": "unknown",
"error": f"Ошибка чтения: {str(e)}"
})
except Exception as e:
logger.error(f"Ошибка получения списка бекапов: {e}")
return backups
async def delete_backup(self, backup_filename: str) -> Tuple[bool, str]:
try:
backup_path = self.backup_dir / backup_filename
if not backup_path.exists():
return False, f"❌ Файл бекапа не найден: {backup_filename}"
backup_path.unlink()
message = f"✅ Бекап {backup_filename} удален"
logger.info(message)
return True, message
except Exception as e:
error_msg = f"❌ Ошибка удаления бекапа: {str(e)}"
logger.error(error_msg)
return False, error_msg
async def _cleanup_old_backups(self):
try:
backups = await self.get_backup_list()
if len(backups) > self._settings.max_backups_keep:
backups.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
for backup in backups[self._settings.max_backups_keep:]:
try:
await self.delete_backup(backup["filename"])
logger.info(f"🗑️ Удален старый бекап: {backup['filename']}")
except Exception as e:
logger.error(f"Ошибка удаления старого бекапа {backup['filename']}: {e}")
except Exception as e:
logger.error(f"Ошибка очистки старых бекапов: {e}")
async def get_backup_settings(self) -> BackupSettings:
return self._settings
async def update_backup_settings(self, **kwargs) -> bool:
try:
for key, value in kwargs.items():
if hasattr(self._settings, key):
setattr(self._settings, key, value)
if self._settings.auto_backup_enabled:
await self.start_auto_backup()
else:
await self.stop_auto_backup()
return True
except Exception as e:
logger.error(f"Ошибка обновления настроек бекапов: {e}")
return False
async def start_auto_backup(self):
if self._auto_backup_task and not self._auto_backup_task.done():
self._auto_backup_task.cancel()
if self._settings.auto_backup_enabled:
self._auto_backup_task = asyncio.create_task(self._auto_backup_loop())
logger.info(f"🔄 Автобекапы включены, интервал: {self._settings.backup_interval_hours}ч")
async def stop_auto_backup(self):
if self._auto_backup_task and not self._auto_backup_task.done():
self._auto_backup_task.cancel()
logger.info("⏹️ Автобекапы остановлены")
async def _auto_backup_loop(self):
while True:
try:
await asyncio.sleep(self._settings.backup_interval_hours * 3600)
logger.info("🔄 Запуск автоматического бекапа...")
success, message, _ = await self.create_backup()
if success:
logger.info(f"✅ Автобекап завершен: {message}")
else:
logger.error(f"❌ Ошибка автобекапа: {message}")
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Ошибка в цикле автобекапов: {e}")
await asyncio.sleep(3600)
async def _send_backup_notification(
self,
event_type: str,
message: str,
file_path: str = None
):
try:
if not settings.is_admin_notifications_enabled():
return
icons = {
"success": "✅",
"error": "❌",
"restore_success": "📥",
"restore_error": "❌"
}
icon = icons.get(event_type, "ℹ️")
notification_text = f"{icon} <b>СИСТЕМА БЕКАПОВ</b>\n\n{message}"
if file_path:
notification_text += f"\n📁 <code>{Path(file_path).name}</code>"
notification_text += f"\n\n⏰ <i>{datetime.now().strftime('%d.%m.%Y %H:%M:%S')}</i>"
try:
from app.services.admin_notification_service import AdminNotificationService
admin_service = AdminNotificationService(self.bot)
await admin_service._send_message(notification_text)
except Exception as e:
logger.error(f"Ошибка отправки уведомления через AdminNotificationService: {e}")
except Exception as e:
logger.error(f"Ошибка отправки уведомления о бекапе: {e}")
backup_service = BackupService()
main.py
+21
View File
@@ -17,6 +17,7 @@ from app.services.version_service import version_service
from app.external.webhook_server import WebhookServer
from app.external.yookassa_webhook import start_yookassa_webhook_server
from app.database.universal_migration import run_universal_migration
from app.services.backup_service import backup_service
class GracefulExit:
@@ -89,6 +90,20 @@ async def main():
logger.info(f"📦 Текущая версия: {version_service.current_version}")
logger.info("🔗 Бот подключен к сервисам мониторинга и техработ")
logger.info("🗄️ Инициализация сервиса бекапов...")
try:
backup_service.bot = bot
# Запускаем автобекапы если они включены
settings_obj = await backup_service.get_backup_settings()
if settings_obj.auto_backup_enabled:
await backup_service.start_auto_backup()
logger.info("✅ Автобекапы запущены")
logger.info("✅ Сервис бекапов инициализирован")
except Exception as e:
logger.error(f"❌ Ошибка инициализации сервиса бекапов: {e}")
payment_service = PaymentService(bot)
@@ -221,6 +236,12 @@ async def main():
await version_check_task
except asyncio.CancelledError:
pass
logger.info("ℹ️ Остановка сервиса бекапов...")
try:
await backup_service.stop_auto_backup()
except Exception as e:
logger.error(f"Ошибка остановки сервиса бекапов: {e}")
if polling_task and not polling_task.done():
logger.info("ℹ️ Остановка polling...")
requirements.txt
+2
View File
@@ -29,3 +29,5 @@ qrcode[pil]==7.4.2
# Для работы с версиями
packaging==23.2
aiofiles==23.2.1