feat: add backup for athena

athena/main.py

@@ -4,15 +4,16 @@
 from time import sleep
 
 import shared.logger
-from modules.admin import error_handler, help_command, users
+from modules.admin import backup, error_handler, help_command, users
 from modules.chat import chat_member_update, my_chat_update
 from shared.database import channel_remove, channel_set_limit, channel_toggle
 from shared.database import check_user, get_keyboard_chats, get_users
 from shared.database import is_forwards_enable, setup_databases
 from shared.database import toggle_forwards, user_remove
 from shared.db import DbDict
 from shared.dependencies import require_admin, require_joined
-from shared.settings import CONF, DATA_DIR, FORWARD_DELAY
+from shared.settings import BLOCKED_CHANNELS_PATH, BLOCKED_USERS_PATH, CONF
+from shared.settings import DATA_DIR, FORWARD_DELAY
 from telegram import Update
 from telegram.error import Forbidden, NetworkError, RetryAfter, TelegramError
 from telegram.ext import Application, CallbackQueryHandler, ChatMemberHandler
@@ -26,8 +27,8 @@
 }
 
 
-blocked_users = DbDict(path=DATA_DIR / 'blocked_users.json')
-blocked_channels = DbDict(path=DATA_DIR / 'blocked_channels.json')
+blocked_users = DbDict(path=BLOCKED_USERS_PATH)
+blocked_channels = DbDict(path=BLOCKED_CHANNELS_PATH)
 
 
 @require_joined
@@ -332,6 +333,7 @@ def main(args: list[str]):
     application.add_handler(CommandHandler('start', start))
     application.add_handler(CommandHandler('help', help_command))
     application.add_handler(CommandHandler('users', users))
+    application.add_handler(CommandHandler('backup', backup))
     application.add_handler(CommandHandler('send_all', send_all))
     application.add_handler(CommandHandler('block', block))
     application.add_handler(CommandHandler('block_channel', block_channel))

athena/modules/admin.py

@@ -7,7 +7,9 @@
 
 from shared.database import get_users
 from shared.dependencies import require_admin
-from shared.settings import CONF
+from shared.settings import BLOCKED_CHANNELS_PATH, BLOCKED_USERS_PATH
+from shared.settings import CHANNEL_DB_PATH, CONF, GENERAL_DB_PATH
+from shared.settings import USER_DB_PATH
 from telegram import Update
 from telegram.constants import ParseMode
 from telegram.error import Forbidden, NetworkError
@@ -17,6 +19,7 @@
 @require_admin
 async def help_command(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     await update.message.reply_text((
+        '/backup -> get backup\n\n'
         '/users -> get all the usernames\n'
         '/start -> get list of channels\n'
         '/send_all -> send a message to all users\n'
@@ -28,6 +31,25 @@ async def help_command(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     ))
 
 
+MAX_FILE_SIZE = 50 * 1024 * 1024
+
+
+@require_admin
+async def backup(update: Update, ctx: Ctx):
+    msg = update.effective_message
+
+    for p in [
+        USER_DB_PATH, CHANNEL_DB_PATH, GENERAL_DB_PATH,
+        BLOCKED_USERS_PATH, BLOCKED_CHANNELS_PATH
+    ]:
+        if p.stat().st_size >= MAX_FILE_SIZE:
+            await msg.reply_text(f'{p.name} is too big ❌')
+            continue
+
+        await msg.reply_document(p, caption=p.name)
+        logging.info(f'a backup for {p.name} was made')
+
+
 @require_admin
 async def users(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     msg = update.message

athena/shared/settings.py

@@ -12,6 +12,8 @@
 USER_DB_PATH = DATA_DIR / 'users.json'
 CHANNEL_DB_PATH = DATA_DIR / 'channels.json'
 GENERAL_DB_PATH = DATA_DIR / 'generals.json'
+BLOCKED_USERS_PATH = DATA_DIR / 'blocked_users.json'
+BLOCKED_CHANNELS_PATH = DATA_DIR / 'blocked_channels.json'
 
 
 EXPIRE_TIME = 15 * 60