D2
Администратор
- Регистрация
- 19 Фев 2025
- Сообщения
- 4,380
- Реакции
- 0
Авторство: hackeryaroslav
Источник: xss.is
Интро
Всем привет! Сегодня мы с вами погрузимся в python, а именно в разработку полезных telegram-ботов. Мы создадим бота, с помощью которого можно будет мониторить состояние сервера, собирать статистику, создавать резервные копии, генерировать отчеты, удалять, скачивать и просматривать файлы, а также получать фидбек от ИИ. Для практики добавим смену языков и реализуем защиту бота, используя 2FA для подтверждения личности, и разрешим доступ только выбранным пользователям. Давайте приступим.
Подготовка
Перед тем как начать работать с ботом, нужно выполнить несколько шагов, а именно получить несколько важных ключей и токенов, без которых бот не сможет функционировать.
Первым делом получаем ключ от ИИ — Gemini от Google. Сделать это очень просто:
Вторым шагом получаем токен бота. Заходим в Telegram и пишем в поиске BotFather. Далее создаем бота, вводим команду /newbot, задаем имя и юзернейм, а затем копируем полученный токен. Он будет выглядеть примерно так: 7………0:AAE……….Qats.
Последним шагом будет получение API-ключа от Google Drive. Здесь немного сложнее, потому что Google иногда усложняет процессы. Я использовал этот гайд: https://developers.google.com/drive/api/quickstart/python?hl=ru, он краткий и понятный. Когда получите файл credentials.json и создадите папку, скопировав её ID, можно переходить к постройке бота.
Пишем бота
Файлов у нас всего несколько - app.py (наш тестовый сервер), bot.py (весь бот в одном файле), report_template.html (расскажу об этом файле позже) и пару других мелких файлов.
Начнем с импортов:
Python: Скопировать в буфер обмена
Код:
import os
import psutil
import requests
import google.generativeai as genai
import datetime
import logging
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, ParseMode
from telegram.ext import (
Updater,
CommandHandler,
CallbackQueryHandler,
CallbackContext,
ConversationHandler,
MessageHandler,
Filters,
)
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from google.oauth2.service_account import Credentials
import random
import string
import traceback
import html
import json
from jinja2 import Environment, FileSystemLoader
import tempfileЧто это? Это набор импортов, из себя он представляет библиотеки для работы с операционной системой (os, psutil), HTTP-запросами (requests), ИИ (google.generativeai), а также для создание и управление тг-ботами (telegram, telegram.ext). Он также включает модули для работы с Google API (googleapiclient, google.oauth2), аутентификации, генерации случайных данных (random, string), обработки ошибок (traceback), работы с HTML (html), форматами данных (json), шаблонизации (jinja2) и создания временных файлов (tempfile). Отлично, идем дальше.
Python: Скопировать в буфер обмена
Код:
# === Configuration ===
BOT_TOKEN = "........"
LOG_FILE_PATH = "server.log"
BACKUP_FOLDER = "backup"
GDRIVE_FOLDER_ID = "......."
CREDENTIALS_FILE = "credentials.json"
ADMIN_USER_ID = 111111111
NGROK_URL_FILE = "ngrok_url.txt"
ALLOWED_DIRECTORY = "C:\\Users\\USer\\Server"
genai.configure(api_key=".......")
model = genai.GenerativeModel("gemini-1.5-flash")На этом этапе мы передаем боту все необходимые API-ключи, пути и другие данные для его работы. Это включает токен бота, полученный из BotFather, файл credentials.json, а также ID папки для бэкапов. ALLOWED_DIRECTORY — это директория на нашем сервере, которую мы указываем для возможности работы с файлами (скачивание, удаление, просмотр).
Python: Скопировать в буфер обмена
Код:
# === Logging ===
logging.basicConfig(
filename=LOG_FILE_PATH,
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
def restricted_command(func):
def wrapper(update: Update, context: CallbackContext):
user_id = update.effective_user.id
user_language = context.user_data.get("language", "en")
if user_id != ADMIN_USER_ID:
if update.callback_query:
update.callback_query.answer(
translations[user_language]["Unauthorized access."]
)
else:
update.message.reply_text(
translations[user_language][
"You are not authorized to use this command."
]
)
return
return func(update, context)
return wrapperТут настроили логирование с использованием модуля logging для записи сообщений в файл, определённый в переменной LOG_FILE_PATH. Помните переменную сверху ADMIN_USER_ID?
Так вот, это было для того чтобы запретить остальным пользователям выполнять другие любые команды, обертка restricted_command, которая ограничивает доступ к определённым командам для пользователей с конкретным ADMIN_USER_ID. Внутри этой функции проверяется, является ли пользователь администратором, и если нет, то ему отправляется сообщение о несанкционированном доступе на соответствующем языке (в зависимости от настроек пользователя, хранящихся в context.user_data). Если же пользователь — администратор, то вызывается исходная команда. Она нам понадобится перед показом главного меню, в остальных случаях мы будем использовать if сценарии.
Python: Скопировать в буфер обмена
Код:
# === Ngrok Functions ===
def get_ngrok_url():
try:
with open(NGROK_URL_FILE, "r") as f:
return f.read().strip()
except FileNotFoundError:
logging.error(
"NGROK_URL_FILE not found. Please run Ngrok and save the URL to ngrok_url.txt"
)
return None
def update_ngrok_url(url):
with open(NGROK_URL_FILE, "w") as f:
f.write(url)
NGROK_URL = get_ngrok_url()get_ngrok_url — пытается открыть файл, указанный в переменной NGROK_URL_FILE, и считывает из него URL, который должен быть сохранён в текстовом файле. Если файл не найден, генерируется ошибка в журнал (лог) с сообщением, что необходимо запустить Ngrok и сохранить URL в файл ngrok_url.txt. В случае ошибки функция возвращает None.
update_ngrok_url — принимает URL в качестве аргумента и записывает его в файл NGROK_URL_FILE, тем самым обновляя сохранённый URL. В файл txt нужно заранее вставить ссылку на ngrok адрес тестового сервера. Вернемся к этому в самом конце, когда закончим написание бота.
Python: Скопировать в буфер обмена
Код:
# === 2FA States ===
TYPING_CODE = 0
def start_2fa(update: Update, context: CallbackContext) -> int:
code = "".join(random.choices(string.digits, k=6))
context.user_data["2fa_code"] = code
update.message.reply_text(f"Please enter the 2FA code: {code}")
return TYPING_CODE
def check_2fa_code(update: Update, context: CallbackContext) -> int:
user_input = update.message.text
if user_input == context.user_data["2fa_code"]:
update.message.reply_text("2FA successful!")
start_cmd(update, context)
return ConversationHandler.END
else:
update.message.reply_text("Incorrect 2FA code. Please try again.")
return TYPING_CODE
def cancel_2fa(update: Update, context: CallbackContext) -> int:
update.message.reply_text("2FA Cancelled")
return ConversationHandler.ENDФункция start_2fa генерирует случайный 6-значный код с помощью random.choices(string.digits, k=6) и сохраняет его в context.user_data["2fa_code"]. Затем бот отправляет сообщение с этим кодом и переводит диалог в состояние TYPING_CODE.
Функция check_2fa_code сравнивает введённый пользователем код с сохранённым в context.user_data["2fa_code"]. Если коды совпадают, пользователю отправляется сообщение об успешном прохождении 2FA, вызывается команда start_cmd, и диалог завершается. В случае неправильного кода бот просит попробовать снова и остаётся в состоянии TYPING_CODE.
Функция cancel_2fa позволяет пользователю отменить процесс 2FA. Она отправляет сообщение об отмене и завершает диалог с помощью ConversationHandler.END.
Тут 2FA обеспечит нам дополнительную безопасность на случай кражи токена бота. Дальше, мы напишем функции выбора языка и старта нашего терминала:
Python: Скопировать в буфер обмена
Код:
# === Start Command (after successful 2FA) ===
def set_language(update: Update, context: CallbackContext):
if update.effective_user.id != ADMIN_USER_ID:
update.message.reply_text("Unauthorized access.")
return
if not context.args or len(context.args) != 1:
update.message.reply_text("Usage: /set_language <en|ru>")
return
language = context.args[0].lower()
if language not in translations:
update.message.reply_text("Supported languages: en, ru")
return
context.user_data["language"] = language
update.message.reply_text(translations[language]["Language set successfully!"])
@restricted_command
def start_cmd(update: Update, context: CallbackContext):
user_language = context.user_data.get("language", "en")
keyboard = [
[
InlineKeyboardButton(
translations[user_language]["System Info"], callback_data="system_menu"
),
],
[
InlineKeyboardButton(
translations[user_language]["Other Tools"], callback_data="other_menu"
)
],
[
InlineKeyboardButton(
translations[user_language]["Files"], callback_data="files_menu"
)
],
[InlineKeyboardButton("🌐 Change Language", callback_data="change_language")],
]
reply_markup = InlineKeyboardMarkup(keyboard)
update.message.reply_text(
translations[user_language]["Welcome to the Admin Panel!"],
reply_markup=reply_markup,
)Функция set_language позволяет нам изменить язык интерфейса бота. Она проверяет, имеет ли пользователь права администратора, и отказывает в доступе, если это не так. (так и будем продолжать в остальных функциях).
Если команда вызвана без аргументов или аргумент задан неправильно, бот отправляет сообщение с инструкцией по правильному использованию команды: /set_language <en|ru>.
Когда переданный язык поддерживается (есть в словаре translations), он сохраняется в context.user_data["language"], и пользователь получает подтверждение на выбранном языке.
Спойлер: Сам словарь
Python: Скопировать в буфер обмена
Код:
# === Load Translations from JSON ===
try:
with open("translations.json", "r", encoding="utf-8") as f:
translations = json.load(f)
except FileNotFoundError:
print("translations.json not found. Using default English translations.")
translations = {
"en": {
"Welcome to the Admin Panel!": "Welcome to the Admin Panel!",
"Language set successfully!": "Language set successfully!",
"You are not authorized to use this command.": "You are not authorized to use this command.",
"An error occurred. Please try again later.": "An error occurred. Please try again later.",
"Server is available at": "Server is available at",
"System Stats:": "System Stats:",
"CPU Usage:": "CPU Usage:",
"RAM Usage:": "RAM Usage:",
"Disk Usage:": "Disk Usage:",
"Used:": "Used:",
"Total:": "Total:",
"GB": "GB",
"Server Status:": "Server Status:",
"Server is unavailable:": "Server is unavailable:",
"Daily Report:": "Daily Report:",
"Log Analysis:": "Log Analysis:",
"Log file not found.": "Log file not found.",
"Error analyzing logs:": "Error analyzing logs:",
"Backup failed:": "Backup failed:",
"Credential file or Backup folder not found:": "Credential file or Backup folder not found:",
"No files found in the backup folder.": "No files found in the backup folder.",
"Error: Ngrok URL is unavailable.": "Error: Ngrok URL is unavailable.",
"Files in": "Files in",
"No files found.": "No files found.",
"Directory not found.": "Directory not found.",
"Error listing files.": "Error listing files.",
"Please provide a file path:": "Please provide a file path:",
"File not found.": "File not found.",
"Error downloading file:": "Error downloading file:",
"File deleted successfully.": "File deleted successfully.",
"Error deleting the file. Check permissions or if the file is in use.": "Error deleting the file. Check permissions or if the file is in use.",
"Please provide the Ngrok URL:": "Please provide the Ngrok URL:",
"Ngrok URL updated to:": "Ngrok URL updated to:",
"Backup scheduled successfully!": "Backup scheduled successfully!",
"Failed to schedule backup:": "Failed to schedule backup:",
"Please enter directory path:": "Please enter directory path:",
"Please enter file path to download:": "Please enter file path to download:",
"Please enter file path to delete:": "Please enter file path to delete:",
"Unauthorized access.": "Unauthorized access.",
"You are not authorized to use this command.": "You are not authorized to use this command.",
"Usage: /set_language <en|ru>": "Usage: /set_language <en|ru>",
"Analysis Result": "Analysis Result",
"Backup Result": "Backup Result",
"Choose an option:": "Choose an option:",
"System Info": "System Info",
"Other Tools": "Other Tools",
"Files": "Files",
"Stats": "Stats",
"Status": "Status",
"Back": "Back",
"Analyze Logs": "Analyze Logs",
"Backup Now": "Backup Now",
"List Files": "List Files",
"Download File": "Download File",
"Delete File": "Delete File",
"Schedule Backup": "Schedule Backup",
"Please enter the 2FA code:": "Please enter the 2FA code:",
"2FA successful!": "2FA successful!",
"Incorrect 2FA code. Please try again.": "Incorrect 2FA code. Please try again.",
"2FA Cancelled": "2FA Cancelled",
"File Management Options:": "File Management Options:",
"Are you sure you want to schedule daily backups at 2:00 AM?": "Are you sure you want to schedule daily backups at 2:00 AM?",
"Backup scheduling confirmed!": "Backup scheduling confirmed!",
},
"ru": {
"Welcome to the Admin Panel!": "Добро пожаловать в админ панель!",
"Language set successfully!": "Язык успешно установлен!",
"You are not authorized to use this command.": "Вы не авторизованы для использования этой команды.",
"An error occurred. Please try again later.": "Произошла ошибка. Пожалуйста, попробуйте позже.",
"Server is available at": "Сервер доступен по адресу",
"System Stats:": "Системная статистика:",
"CPU Usage:": "Использование процессора:",
"RAM Usage:": "Использование ОЗУ:",
"Disk Usage:": "Использование диска:",
"Used:": "Использовано:",
"Total:": "Всего:",
"GB": "ГБ",
"Server Status:": "Статус сервера:",
"Server is unavailable:": "Сервер недоступен:",
"Daily Report:": "Ежедневный отчет:",
"Log Analysis:": "Анализ логов:",
"Log file not found.": "Файл журнала не найден.",
"Error analyzing logs:": "Ошибка анализа журналов:",
"Backup failed:": "Ошибка резервного копирования:",
"Credential file or Backup folder not found:": "Файл учетных данных или папка резервного копирования не найдены:",
"No files found in the backup folder.": "В папке резервного копирования файлов не найдено.",
"Error: Ngrok URL is unavailable.": "Ошибка: URL-адрес Ngrok недоступен.",
"Files in": "Файлы в",
"No files found.": "Файлов не найдено.",
"Directory not found.": "Директория не найдена.",
"Error listing files.": "Ошибка при перечислении файлов.",
"Please provide a file path:": "Пожалуйста, укажите путь к файлу:",
"File not found.": "Файл не найден.",
"Error downloading file:": "Ошибка загрузки файла:",
"File deleted successfully.": "Файл успешно удален.",
"Error deleting the file. Check permissions or if the file is in use.": "Ошибка удаления файла. Проверьте разрешения или используется ли файл.",
"Please provide the Ngrok URL:": "Пожалуйста, укажите URL-адрес Ngrok:",
"Ngrok URL updated to:": "URL-адрес Ngrok обновлен до:",
"Backup scheduled successfully!": "Резервное копирование успешно запланировано!",
"Failed to schedule backup:": "Не удалось запланировать резервное копирование:",
"Please enter directory path:": "Пожалуйста, введите путь к каталогу:",
"Please enter file path to download:": "Пожалуйста, введите путь к файлу для загрузки:",
"Please enter file path to delete:": "Пожалуйста, введите путь к файлу для удаления:",
"Unauthorized access.": "Несанкционированный доступ.",
"You are not authorized to use this command.": "Вы не авторизованы использовать эту команду.",
"Usage: /set_language <en|ru>": "Использование: /set_language <en|ru>",
"Analysis Result": "Результат анализа",
"Backup Result": "Результат резервного копирования",
"Choose an option:": "Выберите опцию:",
"System Info": "Системная информация",
"Other Tools": "Другие инструменты",
"Files": "Файлы",
"Stats": "Статистика",
"Status": "Статус",
"Back": "Назад",
"Analyze Logs": "Анализировать журналы",
"Backup Now": "Резервное копирование сейчас",
"List Files": "Список файлов",
"Download File": "Скачать[HIDE][GROUPS=5,6,7,8,9]https://file.io",
https://lh7-rt.googleusercontent.com/docsz/AD_4nXfHpquObSI5t94ZKRwrTfbZSiZ5TUxp5jMC1cbCNPwc546PhsBuzKm-xHDb4FzXnD6boCVJMbzXxXuAnDu9YR_LiGDbn61nO03c-jkUQdFIPnw5fgnN0W3Wd4ojLrmA0sBu8YFfvA?key=LO0RxDP1UGSnskzDzHBM92N1
https://download.ngrok.com/windows
https://download.ngrok.com/windows[/GROUPS][/HIDE]