videoDownloadTGbot/admin_bot.py

"""
Admin Telegram Bot
Админский бот для получения статистики и всех скачанных видео
"""
import os
import logging
import sqlite3
from pathlib import Path
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
from telegram.request import HTTPXRequest

# Настройка логирования
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

# Токен админ бота из переменных окружения
ADMIN_BOT_TOKEN = os.getenv('ADMIN_BOT_TOKEN')

# Базовая директория проекта
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / 'data'
DB_FILE = DATA_DIR / 'bot.db'
ADMIN_CHAT_ID_FILE = DATA_DIR / 'admin_chat_id.txt'


def get_total_downloads() -> int:
    """Возвращает общее количество скачанных видео"""
    try:
        conn = sqlite3.connect(str(DB_FILE))
        cursor = conn.cursor()
        cursor.execute('SELECT total_downloads FROM stats WHERE id = 1')
        result = cursor.fetchone()
        conn.close()
        return result[0] if result else 0
    except Exception as e:
        logger.error(f"Ошибка при получении количества скачанных видео: {e}")
        return 0


def get_total_users() -> int:
    """Возвращает общее количество уникальных пользователей"""
    try:
        conn = sqlite3.connect(str(DB_FILE))
        cursor = conn.cursor()
        cursor.execute('SELECT COUNT(*) FROM users')
        result = cursor.fetchone()
        conn.close()
        return result[0] if result else 0
    except Exception as e:
        logger.error(f"Ошибка при получении количества пользователей: {e}")
        return 0


def get_error_stats() -> dict[str, int]:
    """Возвращает статистику ошибок по сервисам"""
    try:
        conn = sqlite3.connect(str(DB_FILE))
        cursor = conn.cursor()
        cursor.execute('SELECT service, error_count FROM error_stats ORDER BY service')
        results = cursor.fetchall()
        conn.close()
        return {service: count for service, count in results}
    except Exception as e:
        logger.error(f"Ошибка при получении статистики ошибок: {e}")
        return {}


def save_admin_chat_id(chat_id: int):
    """Сохраняет chat_id админа в файл"""
    try:
        DATA_DIR.mkdir(parents=True, exist_ok=True)
        with open(ADMIN_CHAT_ID_FILE, 'w') as f:
            f.write(str(chat_id))
        logger.info(f"Сохранен chat_id админа: {chat_id}")
    except Exception as e:
        logger.error(f"Ошибка при сохранении chat_id админа: {e}")


def get_admin_chat_id() -> int | None:
    """Получает сохраненный chat_id админа"""
    try:
        if ADMIN_CHAT_ID_FILE.exists():
            with open(ADMIN_CHAT_ID_FILE, 'r') as f:
                chat_id = f.read().strip()
                if chat_id:
                    return int(chat_id)
    except Exception as e:
        logger.error(f"Ошибка при чтении chat_id админа: {e}")
    return None


async def stat_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обрабатывает команду /stat"""
    # Сохраняем chat_id админа при первом использовании
    chat_id = update.message.chat_id
    saved_chat_id = get_admin_chat_id()
    if saved_chat_id != chat_id:
        save_admin_chat_id(chat_id)
        if saved_chat_id is None:
            await update.message.reply_text("✅ Админ бот активирован! Теперь вы будете получать все скачанные видео.")
    
    total_downloads = get_total_downloads()
    total_users = get_total_users()
    error_stats = get_error_stats()
    
    # Форматируем статистику ошибок
    error_stats_text = ""
    service_names = {
        'youtube': 'YouTube',
        'instagram': 'Instagram',
        'tiktok': 'TikTok',
        'vk': 'VK',
        'yapfiles': 'Yapfiles',
        'unknown': 'Unknown'
    }
    
    for service, count in sorted(error_stats.items()):
        if count > 0:
            service_name = service_names.get(service, service)
            error_stats_text += f"  • {service_name}: {count}\n"
    
    if not error_stats_text:
        error_stats_text = "  Нет ошибок"
    
    stat_message = (
        f"📊 Статистика бота:\n\n"
        f"👥 Всего пользователей: {total_users}\n"
        f"📹 Всего скачано видео: {total_downloads}\n\n"
        f"❌ Ошибки по сервисам:\n{error_stats_text.strip()}"
    )
    
    await update.message.reply_text(stat_message)


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обрабатывает все сообщения (на случай если админ отправит что-то)"""
    # Сохраняем chat_id админа при первом сообщении
    chat_id = update.message.chat_id
    saved_chat_id = get_admin_chat_id()
    if saved_chat_id != chat_id:
        save_admin_chat_id(chat_id)
        if saved_chat_id is None:
            await update.message.reply_text("✅ Админ бот активирован! Теперь вы будете получать все скачанные видео.\n\nДоступные команды: /stat")
        else:
            await update.message.reply_text("Это админский бот. Доступные команды: /stat")
    else:
        if update.message and update.message.text:
            await update.message.reply_text("Это админский бот. Доступные команды: /stat")


def main():
    """Главная функция для запуска админ бота"""
    if not ADMIN_BOT_TOKEN:
        logger.error("ADMIN_BOT_TOKEN не установлен!")
        return
    
    # Создаем приложение
    request = HTTPXRequest(
        read_timeout=120,
        connect_timeout=60
    )
    application = (
        Application.builder()
        .token(ADMIN_BOT_TOKEN)
        .request(request)
        .get_updates_request(HTTPXRequest(read_timeout=120, connect_timeout=60))
        .build()
    )
    
    # Регистрируем обработчики
    application.add_handler(CommandHandler("stat", stat_command))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    # Запускаем бота
    logger.info("Админ бот запущен")
    application.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == '__main__':
    main()