319 lines
12 KiB
Python
319 lines
12 KiB
Python
import os
|
|
import logging
|
|
import redis
|
|
|
|
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
|
|
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes, JobQueue
|
|
from playwright.async_api import async_playwright
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Load configuration
|
|
WEBINAR_URL = os.getenv('WEBINAR_URL', 'https://edu.edu.vn.ua/webinar/useractive')
|
|
WEBINAR_CHECK_INTERVAL = int(os.getenv('WEBINAR_CHECK_INTERVAL', 60))
|
|
REDIS_HOST = os.getenv('REDIS_HOST', 'redis')
|
|
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
|
|
PLAYWRIGHT_WS = os.getenv('PLAYWRIGHT_WS', 'ws://playwright-service:3000/ws')
|
|
USER_AGENT = os.getenv('USER_AGENT', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36')
|
|
WEBINAR_TELEGRAM_TOKEN = os.getenv('WEBINAR_TELEGRAM_TOKEN')
|
|
ADMIN_ID = int(os.getenv('WEBINAR_ADMIN_ID', '0'))
|
|
|
|
# Redis Keys
|
|
KEY_WHITELIST = "bot:whitelist"
|
|
KEY_WHITELIST_ENABLED = "bot:whitelist_enabled"
|
|
KEY_SUBSCRIBERS = "bot:subscribers"
|
|
KEY_PHPSESSID = "EDU_PHPSESSID"
|
|
|
|
# Initialize Redis
|
|
try:
|
|
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
|
|
redis_client.ping()
|
|
logger.info(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to connect to Redis: {e}")
|
|
exit(1)
|
|
|
|
# --- Helper Functions ---
|
|
|
|
def is_whitelisted(user_id: int) -> bool:
|
|
"""Check if user is allowed to use the bot."""
|
|
if user_id == ADMIN_ID:
|
|
return True
|
|
|
|
enabled = redis_client.get(KEY_WHITELIST_ENABLED)
|
|
if enabled == "0": # Whitelist disabled
|
|
return True
|
|
|
|
return redis_client.sismember(KEY_WHITELIST, str(user_id))
|
|
|
|
def get_admin_keyboard():
|
|
"""Generate admin panel keyboard."""
|
|
whitelist_enabled = redis_client.get(KEY_WHITELIST_ENABLED) != "0"
|
|
toggle_text = "🔒 Disable Whitelist" if whitelist_enabled else "🔓 Enable Whitelist"
|
|
|
|
keyboard = [
|
|
[InlineKeyboardButton(toggle_text, callback_data="toggle_whitelist")],
|
|
[InlineKeyboardButton("📋 View Whitelist", callback_data="view_whitelist")],
|
|
[InlineKeyboardButton("👥 View Subscribers", callback_data="view_subscribers")],
|
|
[InlineKeyboardButton("🔄 Force Check", callback_data="force_check")]
|
|
]
|
|
return InlineKeyboardMarkup(keyboard)
|
|
|
|
# --- Command Handlers ---
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Handle /start command."""
|
|
user = update.effective_user
|
|
logger.info(f"User {user.id} ({user.username}) started the bot.")
|
|
|
|
if not is_whitelisted(user.id):
|
|
await update.message.reply_text("⛔ Access denied. You are not on the whitelist.")
|
|
return
|
|
|
|
# Add to subscribers
|
|
redis_client.sadd(KEY_SUBSCRIBERS, user.id)
|
|
|
|
msg = (
|
|
f"👋 Hello {user.first_name}!\n\n"
|
|
"I am the Webinar Checker Bot. I will notify you when a new webinar appears.\n"
|
|
"You have been subscribed to notifications."
|
|
)
|
|
|
|
if user.id == ADMIN_ID:
|
|
msg += "\n\n👑 <b>Admin Mode Active</b>"
|
|
await update.message.reply_text(msg, parse_mode='HTML', reply_markup=get_admin_keyboard())
|
|
else:
|
|
await update.message.reply_text(msg, parse_mode='HTML')
|
|
|
|
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Handle /help command."""
|
|
msg = (
|
|
"🤖 <b>Bot Help</b>\n\n"
|
|
"/start - Subscribe to notifications\n"
|
|
"/help - Show this message\n"
|
|
)
|
|
if update.effective_user.id == ADMIN_ID:
|
|
msg += (
|
|
"\n<b>Admin Commands:</b>\n"
|
|
"/adduser [user_id] - Add user to whitelist\n"
|
|
"/removeuser [user_id] - Remove user from whitelist\n"
|
|
"Or use the panel below to manage settings."
|
|
)
|
|
await update.message.reply_text(msg, parse_mode='HTML', reply_markup=get_admin_keyboard())
|
|
else:
|
|
await update.message.reply_text(msg, parse_mode='HTML')
|
|
|
|
async def add_user(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Add user to whitelist (admin only)."""
|
|
if update.effective_user.id != ADMIN_ID:
|
|
await update.message.reply_text("⛔ Admin only!")
|
|
return
|
|
|
|
if not context.args:
|
|
await update.message.reply_text("Usage: /adduser [user_id]")
|
|
return
|
|
|
|
try:
|
|
user_id = int(context.args[0])
|
|
redis_client.sadd(KEY_WHITELIST, str(user_id))
|
|
await update.message.reply_text(f"✅ User {user_id} added to whitelist")
|
|
logger.info(f"Admin added user {user_id} to whitelist")
|
|
except ValueError:
|
|
await update.message.reply_text("❌ Invalid user ID. Must be a number.")
|
|
|
|
async def remove_user(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Remove user from whitelist (admin only)."""
|
|
if update.effective_user.id != ADMIN_ID:
|
|
await update.message.reply_text("⛔ Admin only!")
|
|
return
|
|
|
|
if not context.args:
|
|
await update.message.reply_text("Usage: /removeuser [user_id]")
|
|
return
|
|
|
|
try:
|
|
user_id = int(context.args[0])
|
|
if str(user_id) == str(ADMIN_ID):
|
|
await update.message.reply_text("❌ Cannot remove admin from whitelist")
|
|
return
|
|
|
|
removed = redis_client.srem(KEY_WHITELIST, str(user_id))
|
|
if removed:
|
|
await update.message.reply_text(f"✅ User {user_id} removed from whitelist")
|
|
logger.info(f"Admin removed user {user_id} from whitelist")
|
|
else:
|
|
await update.message.reply_text(f"⚠️ User {user_id} was not in whitelist")
|
|
except ValueError:
|
|
await update.message.reply_text("❌ Invalid user ID. Must be a number.")
|
|
|
|
# --- Admin Callbacks ---
|
|
|
|
async def admin_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Handle admin panel button clicks."""
|
|
query = update.callback_query
|
|
user_id = query.from_user.id
|
|
|
|
if user_id != ADMIN_ID:
|
|
await query.answer("⛔ Admin only!", show_alert=True)
|
|
return
|
|
|
|
await query.answer()
|
|
data = query.data
|
|
|
|
if data == "toggle_whitelist":
|
|
current = redis_client.get(KEY_WHITELIST_ENABLED)
|
|
new_state = "0" if current != "0" else "1"
|
|
redis_client.set(KEY_WHITELIST_ENABLED, new_state)
|
|
state_text = "Disabled" if new_state == "0" else "Enabled"
|
|
await query.edit_message_reply_markup(reply_markup=get_admin_keyboard())
|
|
await query.message.reply_text(f"✅ Whitelist {state_text}")
|
|
|
|
elif data == "view_whitelist":
|
|
members = redis_client.smembers(KEY_WHITELIST)
|
|
msg = "📋 <b>Whitelist:</b>\n" + ("\n".join(members) if members else "Empty")
|
|
await query.message.reply_text(msg, parse_mode='HTML')
|
|
|
|
elif data == "view_subscribers":
|
|
subs = redis_client.smembers(KEY_SUBSCRIBERS)
|
|
msg = "👥 <b>Subscribers:</b>\n" + ("\n".join(subs) if subs else "Empty")
|
|
await query.message.reply_text(msg, parse_mode='HTML')
|
|
|
|
elif data == "force_check":
|
|
await query.message.reply_text("🔄 Running immediate check...")
|
|
result = await check_webinars_job(context)
|
|
|
|
if result is None:
|
|
await query.message.reply_text("❌ Check failed. See logs for details.")
|
|
elif result == 0:
|
|
await query.message.reply_text("✅ Check completed. No webinars found.")
|
|
else:
|
|
await query.message.reply_text(f"✅ Check completed. Found {result} webinar(s)!")
|
|
|
|
# --- Webinar Checking Job ---
|
|
|
|
async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE):
|
|
"""Background job to check for webinars using Async Playwright.
|
|
|
|
Returns:
|
|
int: Number of webinars found, or None if check failed
|
|
"""
|
|
logger.info("Running webinar check...")
|
|
|
|
phpsessid = redis_client.get(KEY_PHPSESSID)
|
|
if not phpsessid:
|
|
logger.warning("PHPSESSID missing. Skipping check.")
|
|
return None
|
|
|
|
webinar_details = []
|
|
|
|
try:
|
|
async with async_playwright() as p:
|
|
# Connect to remote Playwright service
|
|
browser = await p.chromium.connect(PLAYWRIGHT_WS)
|
|
|
|
try:
|
|
# Create browser context with user agent
|
|
context_browser = await browser.new_context(user_agent=USER_AGENT)
|
|
|
|
# Add PHPSESSID cookie
|
|
await context_browser.add_cookies([{
|
|
'name': 'PHPSESSID',
|
|
'value': phpsessid,
|
|
'domain': 'edu.edu.vn.ua',
|
|
'path': '/'
|
|
}])
|
|
|
|
# Create new page
|
|
page = await context_browser.new_page()
|
|
|
|
try:
|
|
# Navigate to webinar page
|
|
await page.goto(WEBINAR_URL, wait_until='domcontentloaded')
|
|
|
|
# Wait for the table to load
|
|
await page.wait_for_selector('#meetings table', timeout=10000)
|
|
|
|
# Get page content
|
|
content = await page.content()
|
|
|
|
# Check if "no webinar" message is present
|
|
if "Жодного онлайн уроку зараз" not in content:
|
|
logger.info("!!! WEBINAR FOUND !!!")
|
|
|
|
# Extract webinar details from table rows
|
|
rows = page.locator('#meetings table tbody tr')
|
|
count = await rows.count()
|
|
|
|
for i in range(count):
|
|
text = await rows.nth(i).inner_text()
|
|
if "Жодного онлайн уроку зараз" not in text:
|
|
webinar_details.append(text.strip())
|
|
else:
|
|
logger.info("No webinars found (expected message present)")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking page: {e}")
|
|
return None
|
|
finally:
|
|
await page.close()
|
|
await context_browser.close()
|
|
|
|
finally:
|
|
await browser.close()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Playwright error: {e}")
|
|
return None
|
|
|
|
# Notify subscribers if webinars found
|
|
if webinar_details:
|
|
message = "🎓 <b>Найден вебинар!</b>\n\n" + "\n\n".join([f"📌 {d}" for d in webinar_details])
|
|
subscribers = redis_client.smembers(KEY_SUBSCRIBERS)
|
|
|
|
for sub_id in subscribers:
|
|
try:
|
|
await context.bot.send_message(chat_id=sub_id, text=message, parse_mode='HTML')
|
|
logger.info(f"Notification sent to {sub_id}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to send to {sub_id}: {e}")
|
|
|
|
return len(webinar_details)
|
|
|
|
# --- Main ---
|
|
|
|
def main():
|
|
if not WEBINAR_TELEGRAM_TOKEN:
|
|
logger.error("WEBINAR_TELEGRAM_TOKEN is missing!")
|
|
return
|
|
|
|
# Set default whitelist state if not set
|
|
if not redis_client.exists(KEY_WHITELIST_ENABLED):
|
|
redis_client.set(KEY_WHITELIST_ENABLED, "1") # Enabled by default
|
|
|
|
# Add admin to whitelist
|
|
if ADMIN_ID:
|
|
redis_client.sadd(KEY_WHITELIST, str(ADMIN_ID))
|
|
|
|
app = Application.builder().token(WEBINAR_TELEGRAM_TOKEN).build()
|
|
|
|
# Handlers
|
|
app.add_handler(CommandHandler("start", start))
|
|
app.add_handler(CommandHandler("help", help_command))
|
|
app.add_handler(CommandHandler("adduser", add_user))
|
|
app.add_handler(CommandHandler("removeuser", remove_user))
|
|
app.add_handler(CallbackQueryHandler(admin_callback))
|
|
|
|
# Job Queue
|
|
job_queue = app.job_queue
|
|
job_queue.run_repeating(check_webinars_job, interval=WEBINAR_CHECK_INTERVAL, first=10)
|
|
|
|
logger.info("Bot started polling...")
|
|
app.run_polling()
|
|
|
|
if __name__ == "__main__":
|
|
main() |