d5201df802
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 2m11s
chat_id is shared by every member of a group chat, so all group messages incremented one shared row and username kept getting overwritten by whoever sent last. Track link counts in a new user_links table keyed by the real from_user.id instead.
342 lines
15 KiB
Python
342 lines
15 KiB
Python
"""
|
||
Admin Telegram Bot
|
||
Админский бот для получения статистики и всех скачанных видео
|
||
"""
|
||
import os
|
||
import asyncio
|
||
import logging
|
||
import sqlite3
|
||
from pathlib import Path
|
||
from telegram import Bot, 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')
|
||
|
||
# Токен клиентского бота — рассылка должна идти от его имени, иначе сообщения
|
||
# уходят в чат с админ-ботом, а не в чат с клиентским ботом, который видит пользователь
|
||
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_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_users_sent_link_count() -> int:
|
||
"""Возвращает количество пользователей, отправивших хотя бы одну ссылку.
|
||
|
||
Ключ — реальный Telegram user_id (таблица user_links), а не chat_id: в
|
||
групповых чатах chat_id общий для всех участников и не годится для учёта
|
||
по отдельным людям.
|
||
"""
|
||
try:
|
||
conn = sqlite3.connect(str(DB_FILE))
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT COUNT(*) FROM user_links WHERE link_count > 0')
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
return result[0] if result else 0
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении количества активных пользователей: {e}")
|
||
return 0
|
||
|
||
|
||
def get_top_active_users(limit: int = 5) -> list[tuple[int, str, str, int]]:
|
||
"""Возвращает топ пользователей по количеству отправленных ссылок (по user_id)"""
|
||
try:
|
||
conn = sqlite3.connect(str(DB_FILE))
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
'SELECT user_id, username, first_name, link_count FROM user_links '
|
||
'WHERE link_count > 0 ORDER BY link_count DESC LIMIT ?',
|
||
(limit,)
|
||
)
|
||
results = cursor.fetchall()
|
||
conn.close()
|
||
return results
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении топа активных пользователей: {e}")
|
||
return []
|
||
|
||
|
||
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 get_all_users() -> list[int]:
|
||
"""Возвращает chat_id всех пользователей бота"""
|
||
try:
|
||
conn = sqlite3.connect(str(DB_FILE))
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT chat_id FROM users')
|
||
users = [row[0] for row in cursor.fetchall()]
|
||
conn.close()
|
||
return users
|
||
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 do_broadcast(text: str) -> tuple[int, int, int]:
|
||
"""Рассылает текстовое сообщение всем пользователям клиентского бота.
|
||
|
||
Отправлять нужно от имени клиентского бота (TELEGRAM_BOT_TOKEN), а не
|
||
админского — пользователи переписываются именно с клиентским ботом, и
|
||
письмо от админ-бота уйдёт в отдельный, незнакомый им чат.
|
||
"""
|
||
users = get_all_users()
|
||
success = 0
|
||
failed = 0
|
||
|
||
async with Bot(token=TELEGRAM_BOT_TOKEN) as client_bot:
|
||
for chat_id in users:
|
||
try:
|
||
await client_bot.send_message(chat_id=chat_id, text=text)
|
||
success += 1
|
||
except Exception as e:
|
||
error_str = str(e).lower()
|
||
if 'blocked' not in error_str and 'chat not found' not in error_str:
|
||
logger.warning(f"Не удалось отправить сообщение {chat_id}: {e}")
|
||
failed += 1
|
||
|
||
await asyncio.sleep(0.05)
|
||
|
||
return success, failed, len(users)
|
||
|
||
|
||
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Обрабатывает команду /start"""
|
||
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"
|
||
"Доступные команды:\n"
|
||
"/stat — статистика бота\n"
|
||
"/send_messages — рассылка сообщения всем пользователям"
|
||
)
|
||
else:
|
||
await update.message.reply_text(
|
||
"Это админский бот.\n\n"
|
||
"Доступные команды:\n"
|
||
"/stat — статистика бота\n"
|
||
"/send_messages — рассылка сообщения всем пользователям"
|
||
)
|
||
else:
|
||
await update.message.reply_text(
|
||
"Доступные команды:\n"
|
||
"/stat — статистика бота\n"
|
||
"/send_messages — рассылка сообщения всем пользователям"
|
||
)
|
||
|
||
|
||
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()
|
||
users_sent_link = get_users_sent_link_count()
|
||
top_users = get_top_active_users(10)
|
||
error_stats = get_error_stats()
|
||
|
||
# Форматируем статистику ошибок
|
||
service_names = {
|
||
'youtube': 'YouTube',
|
||
'instagram': 'Instagram',
|
||
'tiktok': 'TikTok',
|
||
'vk': 'VK',
|
||
'yapfiles': 'Yapfiles',
|
||
'unknown': 'Unknown'
|
||
}
|
||
|
||
error_lines = [
|
||
f"• {service_names.get(service, service)}: {count}"
|
||
for service, count in sorted(error_stats.items())
|
||
if count > 0
|
||
]
|
||
error_stats_text = "\n".join(error_lines) if error_lines else "Нет ошибок"
|
||
|
||
# Форматируем топ активных пользователей.
|
||
# (LRM) перед каждой строкой не даёт Telegram переразвернуть строку
|
||
# справа налево из-за RTL-символов в имени (арабский и т.п.) — иначе номер
|
||
# и счётчик "уезжают" вправо и ломают выравнивание списка.
|
||
top_lines = [
|
||
f"{i}. {('@' + username) if username else (first_name or str(uid))} — {link_count}"
|
||
for i, (uid, username, first_name, link_count) in enumerate(top_users, start=1)
|
||
]
|
||
top_users_text = "\n".join(top_lines) if top_lines else "Пока никто не отправлял ссылки"
|
||
|
||
stat_message = (
|
||
f"📊 Статистика бота:\n\n"
|
||
f"👥 Всего пользователей: {total_users}\n"
|
||
f"🔗 Отправляли ссылки: {users_sent_link}\n"
|
||
f"📹 Всего скачано видео: {total_downloads}\n\n"
|
||
f"🏆 Топ-10 активных пользователей:\n{top_users_text}\n\n"
|
||
f"❌ Ошибки по сервисам:\n{error_stats_text}"
|
||
)
|
||
|
||
await update.message.reply_text(stat_message)
|
||
|
||
|
||
async def send_messages_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Обрабатывает команду /send_messages — запрашивает текст для рассылки"""
|
||
if not TELEGRAM_BOT_TOKEN:
|
||
await update.message.reply_text("❌ TELEGRAM_BOT_TOKEN не задан, рассылка недоступна")
|
||
return
|
||
context.user_data['awaiting_broadcast'] = True
|
||
await update.message.reply_text(
|
||
"Введите текст сообщения для рассылки всем пользователям.\n"
|
||
"Чтобы отменить — /cancel"
|
||
)
|
||
|
||
|
||
async def cancel_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Обрабатывает команду /cancel — отменяет ожидание текста рассылки"""
|
||
if context.user_data.pop('awaiting_broadcast', None):
|
||
await update.message.reply_text("❌ Рассылка отменена")
|
||
else:
|
||
await update.message.reply_text("Нечего отменять")
|
||
|
||
|
||
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Обрабатывает все сообщения (на случай если админ отправит что-то)"""
|
||
if context.user_data.pop('awaiting_broadcast', False):
|
||
text = f"📢 Service message\n\n{update.message.text}"
|
||
success, failed, total = await do_broadcast(text)
|
||
await update.message.reply_text(
|
||
f"📤 Рассылка завершена\n\n"
|
||
f"✅ Отправлено: {success}\n"
|
||
f"❌ Не доставлено: {failed}\n"
|
||
f"👥 Всего: {total}"
|
||
)
|
||
return
|
||
|
||
# Сохраняем 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, /send_messages")
|
||
else:
|
||
await update.message.reply_text("Это админский бот. Доступные команды: /stat, /send_messages")
|
||
else:
|
||
if update.message and update.message.text:
|
||
await update.message.reply_text("Это админский бот. Доступные команды: /stat, /send_messages")
|
||
|
||
|
||
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("start", start_command))
|
||
application.add_handler(CommandHandler("stat", stat_command))
|
||
application.add_handler(CommandHandler("send_messages", send_messages_command))
|
||
application.add_handler(CommandHandler("cancel", cancel_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()
|
||
|