videoDownloadTGbot/broadcast.py

#!/usr/bin/env python3
"""
Скрипт для рассылки сообщений всем пользователям бота.

Использование:
    python broadcast.py "Текст сообщения"
    python broadcast.py --html "<b>Жирный</b> текст"
    python broadcast.py --file message.txt
"""
import os
import sys
import sqlite3
import argparse
from pathlib import Path

# Загружаем переменные окружения из .env если есть
def load_env_file():
    env_file = Path(__file__).resolve().parent / '.env'
    if env_file.exists():
        with open(env_file, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#') and '=' in line:
                    key, value = line.split('=', 1)
                    os.environ.setdefault(key.strip(), value.strip())

load_env_file()

import requests
import time

# Настройки
BASE_DIR = Path(__file__).resolve().parent
DB_FILE = BASE_DIR / 'data' / 'bot.db'
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')


def get_telegram_api_url():
    return f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}"


def get_all_users() -> list[tuple[int, str, str]]:
    """Получает всех пользователей из базы данных"""
    if not DB_FILE.exists():
        print(f"❌ База данных не найдена: {DB_FILE}")
        sys.exit(1)
    
    conn = sqlite3.connect(str(DB_FILE))
    cursor = conn.cursor()
    cursor.execute('SELECT chat_id, username, locale FROM users')
    users = cursor.fetchall()
    conn.close()
    return users


def send_message(chat_id: int, text: str, parse_mode: str = None) -> bool:
    """Отправляет сообщение пользователю"""
    payload = {
        "chat_id": chat_id,
        "text": text,
    }
    if parse_mode:
        payload["parse_mode"] = parse_mode
    
    try:
        response = requests.post(
            f"{get_telegram_api_url()}/sendMessage",
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            return True
        else:
            result = response.json()
            error = result.get('description', 'Unknown error')
            # Пользователь заблокировал бота или удалил чат
            if 'blocked' in error.lower() or 'chat not found' in error.lower():
                return False
            print(f"  ⚠️  {chat_id}: {error}")
            return False
    except Exception as e:
        print(f"  ❌ {chat_id}: {e}")
        return False


def broadcast(text: str, parse_mode: str = None):
    """Рассылает сообщение всем пользователям"""
    users = get_all_users()
    
    if not users:
        print("❌ Нет пользователей в базе данных")
        return
    
    print(f"📤 Начинаю рассылку для {len(users)} пользователей...")
    print(f"📝 Сообщение:\n{text[:100]}{'...' if len(text) > 100 else ''}\n")
    
    success = 0
    failed = 0
    
    for chat_id, username, locale in users:
        result = send_message(chat_id, text, parse_mode)
        if result:
            success += 1
            print(f"  ✅ {chat_id} (@{username or 'no_username'})")
        else:
            failed += 1
        
        # Небольшая задержка чтобы не превысить лимиты Telegram
        time.sleep(0.05)
    
    print(f"\n📊 Результат:")
    print(f"  ✅ Успешно: {success}")
    print(f"  ❌ Ошибок: {failed}")
    print(f"  📨 Всего: {len(users)}")


def main():
    parser = argparse.ArgumentParser(
        description='Рассылка сообщений всем пользователям бота',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Примеры:
  python broadcast.py "Привет! Это важное сообщение."
  python broadcast.py --html "<b>Важно!</b> Новая функция добавлена."
  python broadcast.py --file announcement.txt
  python broadcast.py --file announcement.txt --html
        """
    )
    
    parser.add_argument(
        'message',
        nargs='?',
        help='Текст сообщения для рассылки'
    )
    parser.add_argument(
        '--html',
        action='store_true',
        help='Использовать HTML-разметку в сообщении'
    )
    parser.add_argument(
        '--file', '-f',
        type=str,
        help='Файл с текстом сообщения'
    )
    parser.add_argument(
        '--list', '-l',
        action='store_true',
        help='Показать список пользователей без отправки'
    )
    parser.add_argument(
        '--yes', '-y',
        action='store_true',
        help='Пропустить подтверждение'
    )
    
    args = parser.parse_args()
    
    # Проверяем токен
    if not TELEGRAM_BOT_TOKEN:
        print("❌ TELEGRAM_BOT_TOKEN не установлен!")
        print("   Установите переменную окружения или создайте .env файл")
        sys.exit(1)
    
    # Показать список пользователей
    if args.list:
        users = get_all_users()
        print(f"👥 Всего пользователей: {len(users)}\n")
        for chat_id, username, locale in users:
            print(f"  {chat_id:>12} | @{username or 'no_username':<20} | {locale}")
        sys.exit(0)
    
    # Получаем текст сообщения
    if args.file:
        file_path = Path(args.file)
        if not file_path.exists():
            print(f"❌ Файл не найден: {args.file}")
            sys.exit(1)
        text = file_path.read_text(encoding='utf-8')
    elif args.message:
        text = args.message
    else:
        parser.print_help()
        sys.exit(1)
    
    if not text.strip():
        print("❌ Сообщение не может быть пустым")
        sys.exit(1)
    
    # Подтверждение
    users = get_all_users()
    print(f"⚠️  Вы собираетесь отправить сообщение {len(users)} пользователям.")
    
    if not args.yes:
        try:
            confirm = input("Продолжить? (y/N): ").strip().lower()
        except (UnicodeDecodeError, EOFError):
            confirm = ''
        if confirm not in ('y', 'yes', 'д', 'да'):
            print("❌ Отменено")
            sys.exit(0)
    
    # Отправляем
    parse_mode = 'HTML' if args.html else None
    broadcast(text, parse_mode)


if __name__ == '__main__':
    main()