admin panel
This commit is contained in:
vrubelroman
parent
3362bf89e2
commit
23de80f94d
6 changed files with 424 additions and 20 deletions
223
LichessClientTG_bot/admin_bot.py
Normal file
223
LichessClientTG_bot/admin_bot.py
Normal file
|
|
@ -0,0 +1,223 @@
|
|||
"""
|
||||
Admin Panel Telegram Bot
|
||||
Sends notifications about new users and new tracked players
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from telegram import Update
|
||||
from telegram.ext import Application, CommandHandler, ContextTypes
|
||||
|
||||
from config import ADMINPANEL_TELEGRAM_BOT_TOKEN, DATABASE_PATH
|
||||
from database import Database
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
level=logging.INFO
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AdminBot:
|
||||
def __init__(self):
|
||||
self.db = Database()
|
||||
self.application = None
|
||||
|
||||
async def send_notification(self, message: str):
|
||||
"""Send notification to admin chat"""
|
||||
admin_chat_id = self.get_admin_chat_id()
|
||||
if not admin_chat_id:
|
||||
logger.warning("Admin chat ID not set, cannot send notification")
|
||||
return
|
||||
|
||||
# Try to use application if available (when running as separate service)
|
||||
if self.application:
|
||||
try:
|
||||
await self.application.bot.send_message(
|
||||
chat_id=admin_chat_id,
|
||||
text=message,
|
||||
parse_mode='HTML'
|
||||
)
|
||||
logger.debug(f"Notification sent via application to chat {admin_chat_id}")
|
||||
return
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to send via application: {e}, trying direct API call")
|
||||
|
||||
# Fallback: use direct Telegram API call
|
||||
try:
|
||||
import aiohttp
|
||||
url = f"https://api.telegram.org/bot{ADMINPANEL_TELEGRAM_BOT_TOKEN}/sendMessage"
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, json={
|
||||
"chat_id": admin_chat_id,
|
||||
"text": message,
|
||||
"parse_mode": "HTML"
|
||||
}) as response:
|
||||
if response.status == 200:
|
||||
logger.debug(f"Notification sent via direct API to chat {admin_chat_id}")
|
||||
else:
|
||||
error_text = await response.text()
|
||||
logger.error(f"Failed to send notification: {response.status} - {error_text}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send admin notification via API: {e}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
|
||||
def get_admin_chat_id(self) -> Optional[int]:
|
||||
"""Get admin chat ID from database"""
|
||||
return self.db.get_admin_chat_id()
|
||||
|
||||
def set_admin_chat_id(self, chat_id: int):
|
||||
"""Set admin chat ID in database"""
|
||||
self.db.set_admin_chat_id(chat_id)
|
||||
|
||||
async def notify_new_user(self, user_id: int, username: Optional[str], first_name: Optional[str]):
|
||||
"""Send notification about new Telegram user"""
|
||||
username_text = f"@{username}" if username else "без username"
|
||||
name_text = first_name if first_name else "без имени"
|
||||
|
||||
message = (
|
||||
f"🆕 <b>Новый пользователь Telegram</b>\n\n"
|
||||
f"ID: {user_id}\n"
|
||||
f"Username: {username_text}\n"
|
||||
f"Имя: {name_text}"
|
||||
)
|
||||
await self.send_notification(message)
|
||||
|
||||
async def notify_new_player(self, player_username: str, added_by_user_id: int, added_by_username: Optional[str]):
|
||||
"""Send notification about new tracked player"""
|
||||
added_by_text = f"@{added_by_username}" if added_by_username else f"ID: {added_by_user_id}"
|
||||
lichess_url = f"https://lichess.org/@/{player_username}"
|
||||
|
||||
message = (
|
||||
f"🎮 <b>Добавлен новый игрок для отслеживания</b>\n\n"
|
||||
f"Игрок: <a href=\"{lichess_url}\">{player_username}</a>\n"
|
||||
f"Добавил: {added_by_text}"
|
||||
)
|
||||
await self.send_notification(message)
|
||||
|
||||
async def status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Status command - show statistics"""
|
||||
try:
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(self.db.db_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Count users
|
||||
cursor.execute("SELECT COUNT(*) FROM telegram_users")
|
||||
users_count = cursor.fetchone()[0]
|
||||
|
||||
# Count unique gamers
|
||||
cursor.execute("SELECT COUNT(DISTINCT username) FROM gamers")
|
||||
gamers_count = cursor.fetchone()[0]
|
||||
|
||||
conn.close()
|
||||
|
||||
message = (
|
||||
f"📊 <b>Статистика базы данных</b>\n\n"
|
||||
f"👥 Пользователей Telegram: {users_count}\n"
|
||||
f"🎮 Отслеживаемых игроков: {gamers_count}"
|
||||
)
|
||||
await update.message.reply_text(message, parse_mode='HTML')
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting status: {e}")
|
||||
await update.message.reply_text(f"❌ Ошибка получения статистики: {e}")
|
||||
|
||||
async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Start command - register admin chat"""
|
||||
try:
|
||||
chat_id = update.effective_chat.id
|
||||
self.set_admin_chat_id(chat_id)
|
||||
|
||||
await update.message.reply_text(
|
||||
"✅ Админ-панель активирована!\n\n"
|
||||
"Доступные команды:\n"
|
||||
"/status - Показать статистику"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in start command: {e}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
try:
|
||||
await update.message.reply_text(f"Ошибка: {e}")
|
||||
except:
|
||||
pass
|
||||
|
||||
def setup_handlers(self, application: Application):
|
||||
"""Setup all handlers"""
|
||||
self.application = application # Store application reference
|
||||
|
||||
# Add start command handler
|
||||
application.add_handler(CommandHandler("start", self.start))
|
||||
|
||||
# Add status command handler
|
||||
application.add_handler(CommandHandler("status", self.status))
|
||||
|
||||
|
||||
# Global admin bot instance
|
||||
_admin_bot_instance: Optional[AdminBot] = None
|
||||
|
||||
|
||||
def get_admin_bot() -> Optional[AdminBot]:
|
||||
"""Get global admin bot instance"""
|
||||
return _admin_bot_instance
|
||||
|
||||
|
||||
def init_admin_bot():
|
||||
"""Initialize admin bot"""
|
||||
global _admin_bot_instance
|
||||
if not _admin_bot_instance:
|
||||
_admin_bot_instance = AdminBot()
|
||||
return _admin_bot_instance
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function"""
|
||||
admin_bot = init_admin_bot()
|
||||
|
||||
# Create application with Long Polling configuration
|
||||
application = Application.builder().token(ADMINPANEL_TELEGRAM_BOT_TOKEN).build()
|
||||
|
||||
# Setup handlers
|
||||
admin_bot.setup_handlers(application)
|
||||
|
||||
# Set application reference
|
||||
admin_bot.application = application
|
||||
|
||||
# Add error handler
|
||||
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
"""Log the error and send a telegram message to notify the developer."""
|
||||
logger.error(f"Exception while handling an update: {context.error}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
|
||||
application.add_error_handler(error_handler)
|
||||
|
||||
# Add post_init hook to log bot info
|
||||
async def post_init(app: Application) -> None:
|
||||
bot_info = await app.bot.get_me()
|
||||
logger.info(f"Admin bot initialized: @{bot_info.username} (ID: {bot_info.id})")
|
||||
|
||||
application.post_init = post_init
|
||||
|
||||
# Start the bot with Long Polling
|
||||
logger.info("Starting Admin Panel Bot with Long Polling...")
|
||||
try:
|
||||
application.run_polling(
|
||||
poll_interval=1.0,
|
||||
timeout=30,
|
||||
drop_pending_updates=True,
|
||||
allowed_updates=["message", "callback_query"]
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Fatal error in run_polling: {e}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue