forust/homelab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Wiki Activity

feat: telegram webianr checker bot

Browse Source
This commit is contained in:
mr-forust
2025-11-26 22:12:03 +01:00
parent cb92bff0c5
commit d4e0e7f37a
3 changed files with 217 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
edu_master/.env.example
+1 -1
View File
@@ -10,4 +10,4 @@ REDIS_HOST=redis
REDIS_PORT=6379
PLAYWRIGHT_WS=ws://playwright-service:3000/ws
WEBINAR_TELEGRAM_TOKEN=your_telegram_bot_token_here
WEBINAR_TELEGRAM_CHAT_ID=your_telegram_chat_id_here
WEBINAR_ADMIN_ID=123456789
edu_master/webinar-checker/Dockerfile
+1 -1
View File
@@ -3,7 +3,7 @@ FROM python:3.11-slim
WORKDIR /app
# Install dependencies
RUN pip install --upgrade pip && pip install playwright==1.56.0 redis requests
RUN pip install --upgrade pip && pip install playwright==1.56.0 redis requests "python-telegram-bot[job-queue]"
COPY checker.py .
edu_master/webinar-checker/checker.py
+222 -114
View File
@@ -1,9 +1,10 @@
import os
import time
import logging
import redis
import requests
from playwright.sync_api import sync_playwright
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(
@@ -20,128 +21,235 @@ 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')
WEBINAR_TELEGRAM_CHAT_ID = os.getenv('WEBINAR_TELEGRAM_CHAT_ID')
ADMIN_ID = int(os.getenv('WEBINAR_ADMIN_ID', '0'))
def send_telegram_message(message):
"""Send message to Telegram."""
if not WEBINAR_TELEGRAM_TOKEN or not WEBINAR_TELEGRAM_CHAT_ID:
logger.warning("Telegram credentials not configured. Skipping message.")
return
# Redis Keys
KEY_WHITELIST = "bot:whitelist"
KEY_WHITELIST_ENABLED = "bot:whitelist_enabled"
KEY_SUBSCRIBERS = "bot:subscribers"
KEY_PHPSESSID = "EDU_PHPSESSID"
try:
url = f"https://api.telegram.org/bot{WEBINAR_TELEGRAM_TOKEN}/sendMessage"
payload = {
'chat_id': WEBINAR_TELEGRAM_CHAT_ID,
'text': message,
'parse_mode': 'HTML'
}
response = requests.post(url, json=payload)
if response.status_code == 200:
logger.info("Telegram message sent successfully.")
else:
logger.error(f"Failed to send Telegram message: {response.text}")
except Exception as e:
logger.error(f"Error sending Telegram message: {e}")
def check_webinars(redis_client):
"""Check for active webinars and notify via Telegram."""
# Get PHPSESSID from Redis
phpsessid = redis_client.get('EDU_PHPSESSID')
if not phpsessid:
logger.warning("PHPSESSID not found in Redis. Waiting for session-keeper to provide it.")
return
logger.info(f"Using PHPSESSID from Redis: {phpsessid}")
with sync_playwright() as p:
try:
# Connect to remote Playwright server
browser = p.chromium.connect(PLAYWRIGHT_WS)
logger.info(f"Connected to Playwright server at {PLAYWRIGHT_WS}")
context = browser.new_context(
user_agent=USER_AGENT,
viewport={'width': 1280, 'height': 720}
)
# Set PHPSESSID cookie
context.add_cookies([{
'name': 'PHPSESSID',
'value': phpsessid,
'domain': 'edu.edu.vn.ua',
'path': '/'
}])
page = context.new_page()
try:
logger.info(f"Navigating to {WEBINAR_URL}...")
page.goto(WEBINAR_URL, wait_until='domcontentloaded')
# Wait for the table to load (dynamic content)
try:
page.wait_for_selector('#meetings table', timeout=10000)
except Exception:
logger.error("Timeout waiting for #meetings table. Page might not have loaded correctly.")
return
content = page.content()
if "Жодного онлайн уроку зараз" in content:
logger.info("Webinars check: No active webinars.")
else:
# Webinars found!
logger.info("!!! WEBINAR FOUND !!!")
# Extract details
rows = page.locator('#meetings table tbody tr')
count = rows.count()
webinar_details = []
if count > 0:
for i in range(count):
text = rows.nth(i).inner_text()
# Skip the "no webinars" row
if "Жодного онлайн уроку зараз" not in text:
webinar_details.append(text.strip())
# Send Telegram notification (Russian)
if webinar_details:
message = "🎓 <b>Найден вебинар!</b>\n\n"
for detail in webinar_details:
message += f"📌 {detail}\n\n"
send_telegram_message(message)
else:
send_telegram_message("🎓 <b>Найден вебинар!</b>\n\nДетали не удалось извлечь.")
except Exception as e:
logger.error(f"An error occurred during check: {e}")
finally:
browser.close()
except Exception as e:
logger.error(f"Failed to connect to Playwright server: {e}")
def main():
logger.info("Starting Webinar Checker Bot")
# Connect to Redis
# 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
# Send startup message (Russian)
send_telegram_message("🤖 <b>Бот запущен</b>\n\nМониторинг вебинаров активирован...")
# Add to subscribers
redis_client.sadd(KEY_SUBSCRIBERS, user.id)
while True:
check_webinars(redis_client)
logger.info(f"Sleeping for {WEBINAR_CHECK_INTERVAL} seconds...")
time.sleep(WEBINAR_CHECK_INTERVAL)
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>\nUse 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')
# --- 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...")
await check_webinars_job(context)
# --- Webinar Checking Job ---
async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE):
"""Background job to check for webinars using Async Playwright."""
logger.info("Running webinar check...")
phpsessid = redis_client.get(KEY_PHPSESSID)
if not phpsessid:
logger.warning("PHPSESSID missing. Skipping check.")
return
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}")
finally:
await page.close()
await context_browser.close()
finally:
await browser.close()
except Exception as e:
logger.error(f"Playwright error: {e}")
return
# 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}")
# --- 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(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()