#!/usr/bin/env python
import logging
import os
import json
from datetime import datetime
from typing import Dict
from dotenv import load_dotenv
from telegram import ReplyKeyboardMarkup, KeyboardButton, Update
from telegram.constants import ParseMode
from telegram.ext import (
Application,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
CallbackContext,
)
# Загружаем переменные окружения
load_dotenv()
TOKEN = os.getenv("TOKEN")
ADMIN_ID = int(os.getenv("ADMIN_ID", "0"))
PASSWORD = os.getenv("PASSWORD", "supersecret")
USERS_FILE = "users.json" # Файл для хранения пользователей
# Логирование
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
def load_users() -> Dict[str, Dict]:
"""Загружает список авторизованных пользователей из JSON."""
if os.path.exists(USERS_FILE):
try:
with open(USERS_FILE, "r", encoding="utf-8") as file:
return json.load(file)
except Exception as e:
logger.error(f"Ошибка загрузки users.json: {e}")
return {}
def save_users(users: Dict[str, Dict]) -> None:
"""Сохраняет список авторизованных пользователей в JSON."""
try:
with open(USERS_FILE, "w", encoding="utf-8") as file:
json.dump(users, file, indent=4, ensure_ascii=False)
except Exception as e:
logger.error(f"Ошибка сохранения users.json: {e}")
def get_main_menu() -> ReplyKeyboardMarkup:
"""Создаёт Reply-клавиатуру для авторизованных пользователей."""
keyboard = [
[KeyboardButton("⚙️ Частота вращения"), KeyboardButton("📩 Тест")],
]
return ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
async def start(update: Update, context: CallbackContext) -> None:
"""Команда /start — Проверка JSON перед авторизацией."""
user_id = str(update.effective_user.id)
users = load_users()
if user_id in users:
await update.message.reply_text(
"✅ Вы уже авторизованы! Выберите действие:",
reply_markup=get_main_menu(),
)
else:
await update.message.reply_text(
"🚫 Вы не авторизованы. Используйте /password <пароль>."
)
async def handle_buttons(update: Update, context: CallbackContext) -> None:
"""Обработчик текстовых кнопок."""
user_id = str(update.effective_user.id)
users = load_users()
if user_id not in users:
await update.message.reply_text("🚫 Вы не авторизованы.")
return
text = update.message.text
if text == "⚙️ Частота вращения":
await update.message.reply_text("⚙️ Частота вращения: 1500 об/мин")
elif text == "📩 Тест":
await update.message.reply_text("✅ Это тестовое сообщение!")
async def password(update: Update, context: CallbackContext) -> None:
"""Команда /password <пароль> — Авторизация пользователя."""
user_id = str(update.effective_user.id)
users = load_users()
if user_id in users:
await update.message.reply_text("✅ Вы уже авторизованы!")
return
if len(context.args) != 1:
await update.message.reply_text("❌ Используйте: /password <пароль>")
return
if context.args[0] == PASSWORD:
first_name = update.effective_user.first_name
last_name = update.effective_user.last_name or "—"
registered_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
users[user_id] = {
"id": user_id,
"first_name": first_name,
"last_name": last_name,
"registered_at": registered_at,
}
save_users(users) # Сохраняем в файл
await update.message.reply_text(
f"✅ {first_name}, вы авторизованы!", reply_markup=get_main_menu()
)
# Уведомляем администратора о новом пользователе
if ADMIN_ID:
try:
await context.bot.send_message(
ADMIN_ID,
f"🔔 Новый пользователь авторизовался:\n"
f"👤 {first_name} {last_name}\n"
f"🆔 ID: {user_id}\n"
f"📅 Дата регистрации: {registered_at}"
)
except Exception as e:
logger.error(f"Ошибка отправки админу: {e}")
else:
await update.message.reply_text("❌ Неверный пароль.")
async def print_users(update: Update, context: CallbackContext) -> None:
"""Команда /print_users — Показывает список авторизованных пользователей."""
user_id = str(update.effective_user.id)
users = load_users()
if user_id != str(ADMIN_ID):
await update.message.reply_text("⛔ У вас нет доступа к этой команде.")
return
if not users:
await update.message.reply_text("Нет авторизованных пользователей.")
return
users_text = "📜 Список авторизованных пользователей:\n\n"
for user in users.values():
users_text += (
f"👤 {user['first_name']} {user['last_name']}\n"
f"🆔 ID: {user['id']}\n"
f"📅 Дата регистрации: {user['registered_at']}\n"
f"——————————\n"
)
await update.message.reply_text(users_text, parse_mode=ParseMode.HTML)
async def help_command(update: Update, context: CallbackContext) -> None:
"""Команда /help — выводит список команд."""
help_text = (
"🛠 Доступные команды:\n\n"
"/start - Начать работу с ботом\n"
"/password <пароль> - Авторизоваться\n"
"/print_users - Список пользователей (только для админа)\n"
"/help - Справка по командам\n\n"
"⚙️ Также доступны кнопки с действиями!"
)
await update.message.reply_text(help_text, parse_mode=ParseMode.HTML)
def main() -> None:
"""Запуск бота."""
if not TOKEN:
logger.error("❌ Не задан токен!")
return
application = Application.builder().token(TOKEN).build()
# Добавление команд
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("password", password))
application.add_handler(CommandHandler("print_users", print_users))
application.add_handler(CommandHandler("help", help_command))
# Обработка нажатий кнопок
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_buttons))
logger.info("🤖 Бот запущен...")
application.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()