import os
import logging
import redis
import json
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes
from playwright.async_api import async_playwright
# Logger
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Load environment variables
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"
KEY_WEBINAR_HISTORY = "bot:webinar_history" # Stores last 5 webinars
# 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)
# --- Translations ---
TRANSLATIONS = {
'ru': {
'welcome': "👋 Привет, {name}!\n\nЯ бот-уведомитель о вебинарах. Я буду сообщать вам, когда появится новый вебинар.\nВы подписаны на уведомления.",
'welcome_admin': "\n\n👑 Режим администратора активен",
'access_denied': "⛔ Доступ запрещен. Вас нет в белом списке.",
'help_title': "🤖 Помощь по боту\n\n",
'help_commands': "/start - Подписаться на уведомления\n/help - Показать это сообщение\n/language - Сменить язык",
'help_admin': "\nКоманды администратора:\n/adduser [user_id] - Добавить пользователя в белый список\n/removeuser [user_id] - Удалить пользователя из белого списка\nИли используйте панель ниже для управления настройками.",
'admin_only': "⛔ Только для администратора!",
'user_added': "✅ Пользователь {user_id} добавлен в белый список",
'user_removed': "✅ Пользователь {user_id} удален из белого списка",
'user_not_in_whitelist': "⚠️ Пользователь {user_id} не был в белом списке",
'cannot_remove_admin': "❌ Невозможно удалить администратора из белого списка",
'invalid_user_id': "❌ Неверный ID пользователя. Должно быть число.",
'usage_adduser': "Использование: /adduser [user_id]",
'usage_removeuser': "Использование: /removeuser [user_id]",
'whitelist_enabled': "✅ Белый список включен",
'whitelist_disabled': "✅ Белый список отключен",
'whitelist_title': "📋 Белый список:\n",
'subscribers_title': "👥 Подписчики:\n",
'empty': "Пусто",
'force_check_running': "🔄 Запускаю проверку...",
'check_failed': "❌ Проверка не удалась. Смотрите логи.",
'check_completed_none': "✅ Проверка завершена. Вебинаров не найдено.",
'check_completed': "✅ Проверка завершена. Найдено {count} вебинар(ов)!",
'toggle_whitelist_disable': "🔒 Отключить белый список",
'toggle_whitelist_enable': "🔓 Включить белый список",
'view_whitelist': "📋 Посмотреть белый список",
'view_subscribers': "👥 Посмотреть подписчиков",
'force_check': "🔄 Принудительная проверка",
'webinar_found': "🎓 Новый вебинар!\n\n",
'webinar_item': "📌 {name}\n🔗 https://edu.edu.vn.ua{url}",
'select_language': "🌐 Выберите язык / Оберіть мову / Select language:",
'language_changed': "✅ Язык изменен на русский",
'flag_ru': "🇷🇺 Русский",
'flag_uk': "🇺🇦 Українська",
'flag_en': "🇬🇧 English",
},
'uk': {
'welcome': "👋 Привіт, {name}!\n\nЯ бот-сповіщувач про вебінари. Я повідомлятиму вас, коли з'явиться новий вебінар.\nВи підписані на сповіщення.",
'welcome_admin': "\n\n👑 Режим адміністратора активний",
'access_denied': "⛔ Доступ заборонено. Вас немає в білому списку.",
'help_title': "🤖 Довідка по боту\n\n",
'help_commands': "/start - Підписатися на сповіщення\n/help - Показати це повідомлення\n/language - Змінити мову",
'help_admin': "\nКоманди адміністратора:\n/adduser [user_id] - Додати користувача до білого списку\n/removeuser [user_id] - Видалити користувача з білого списку\nАбо використовуйте панель нижче для керування налаштуваннями.",
'admin_only': "⛔ Тільки для адміністратора!",
'user_added': "✅ Користувач {user_id} доданий до білого списку",
'user_removed': "✅ Користувач {user_id} видалений з білого списку",
'user_not_in_whitelist': "⚠️ Користувач {user_id} не був у білому списку",
'cannot_remove_admin': "❌ Неможливо видалити адміністратора з білого списку",
'invalid_user_id': "❌ Невірний ID користувача. Має бути число.",
'usage_adduser': "Використання: /adduser [user_id]",
'usage_removeuser': "Використання: /removeuser [user_id]",
'whitelist_enabled': "✅ Білий список увімкнено",
'whitelist_disabled': "✅ Білий список вимкнено",
'whitelist_title': "📋 Білий список:\n",
'subscribers_title': "👥 Підписники:\n",
'empty': "Порожньо",
'force_check_running': "🔄 Запускаю перевірку...",
'check_failed': "❌ Перевірка не вдалася. Дивіться логи.",
'check_completed_none': "✅ Перевірка завершена. Вебінарів не знайдено.",
'check_completed': "✅ Перевірка завершена. Знайдено {count} вебінар(ів)!",
'toggle_whitelist_disable': "🔒 Вимкнути білий список",
'toggle_whitelist_enable': "🔓 Увімкнути білий список",
'view_whitelist': "📋 Переглянути білий список",
'view_subscribers': "👥 Переглянути підписників",
'force_check': "🔄 Примусова перевірка",
'webinar_found': "🎓 Новий вебінар!\n\n",
'webinar_item': "📌 {name}\n🔗 https://edu.edu.vn.ua{url}",
'select_language': "🌐 Виберіть мову / Выберите язык / Select language:",
'language_changed': "✅ Мову змінено на українську",
'flag_ru': "🇷🇺 Русский",
'flag_uk': "🇺🇦 Українська",
'flag_en': "🇬🇧 English",
},
'en': {
'welcome': "👋 Hello, {name}!\n\nI am the Webinar Checker Bot. I will notify you when a new webinar appears.\nYou have been subscribed to notifications.",
'welcome_admin': "\n\n👑 Admin Mode Active",
'access_denied': "⛔ Access denied. You are not on the whitelist.",
'help_title': "🤖 Bot Help\n\n",
'help_commands': "/start - Subscribe to notifications\n/help - Show this message\n/language - Change language",
'help_admin': "\nAdmin Commands:\n/adduser [user_id] - Add user to whitelist\n/removeuser [user_id] - Remove user from whitelist\nOr use the panel below to manage settings.",
'admin_only': "⛔ Admin only!",
'user_added': "✅ User {user_id} added to whitelist",
'user_removed': "✅ User {user_id} removed from whitelist",
'user_not_in_whitelist': "⚠️ User {user_id} was not in whitelist",
'cannot_remove_admin': "❌ Cannot remove admin from whitelist",
'invalid_user_id': "❌ Invalid user ID. Must be a number.",
'usage_adduser': "Usage: /adduser [user_id]",
'usage_removeuser': "Usage: /removeuser [user_id]",
'whitelist_enabled': "✅ Whitelist Enabled",
'whitelist_disabled': "✅ Whitelist Disabled",
'whitelist_title': "📋 Whitelist:\n",
'subscribers_title': "👥 Subscribers:\n",
'empty': "Empty",
'force_check_running': "🔄 Running immediate check...",
'check_failed': "❌ Check failed. See logs for details.",
'check_completed_none': "✅ Check completed. No webinars found.",
'check_completed': "✅ Check completed. Found {count} webinar(s)!",
'toggle_whitelist_disable': "🔒 Disable Whitelist",
'toggle_whitelist_enable': "🔓 Enable Whitelist",
'view_whitelist': "📋 View Whitelist",
'view_subscribers': "👥 View Subscribers",
'force_check': "🔄 Force Check",
'webinar_found': "🎓 New webinar found!\n\n",
'webinar_item': "📌 {name}\n🔗 https://edu.edu.vn.ua{url}",
'select_language': "🌐 Select language / Виберіть мову / Выберите язык:",
'language_changed': "✅ Language changed to English",
'flag_ru': "🇷🇺 Русский",
'flag_uk': "🇺🇦 Українська",
'flag_en': "🇬🇧 English",
}
}
# --- Language Helper Functions ---
def get_user_language(user_id: int) -> str:
"""Get user's preferred language from Redis. Default: Ukrainian."""
lang = redis_client.get(f"user:{user_id}:language")
return lang if lang in ['ru', 'uk', 'en'] else 'uk'
def set_user_language(user_id: int, lang: str):
"""Save user's language preference to Redis."""
if lang in ['ru', 'uk', 'en']:
redis_client.set(f"user:{user_id}:language", lang)
logger.info(f"User {user_id} language set to {lang}")
def t(user_id: int, key: str, **kwargs) -> str:
"""Translate message for user with optional formatting."""
lang = get_user_language(user_id)
message = TRANSLATIONS.get(lang, TRANSLATIONS['uk']).get(key, key)
if kwargs:
return message.format(**kwargs)
return message
def get_language_keyboard():
"""Generate language selection keyboard."""
keyboard = [
[
InlineKeyboardButton("🇷🇺 Русский", callback_data="lang_ru"),
InlineKeyboardButton("🇺🇦 Українська", callback_data="lang_uk"),
],
[
InlineKeyboardButton("🇬🇧 English", callback_data="lang_en"),
]
]
return InlineKeyboardMarkup(keyboard)
# --- 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(user_id: int):
"""Generate admin panel keyboard."""
whitelist_enabled = redis_client.get(KEY_WHITELIST_ENABLED) != "0"
toggle_text = t(user_id, 'toggle_whitelist_disable') if whitelist_enabled else t(user_id, 'toggle_whitelist_enable')
keyboard = [
[InlineKeyboardButton(toggle_text, callback_data="toggle_whitelist")],
[InlineKeyboardButton(t(user_id, 'view_whitelist'), callback_data="view_whitelist")],
[InlineKeyboardButton(t(user_id, 'view_subscribers'), callback_data="view_subscribers")],
[InlineKeyboardButton(t(user_id, '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(t(user.id, 'access_denied'))
return
# Add to subscribers
redis_client.sadd(KEY_SUBSCRIBERS, user.id)
msg = t(user.id, 'welcome', name=user.first_name)
if user.id == ADMIN_ID:
msg += t(user.id, 'welcome_admin')
await update.message.reply_text(msg, parse_mode='HTML', reply_markup=get_admin_keyboard(user.id))
else:
await update.message.reply_text(msg, parse_mode='HTML')
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /help command."""
user_id = update.effective_user.id
msg = t(user_id, 'help_title') + t(user_id, 'help_commands')
if user_id == ADMIN_ID:
msg += t(user_id, 'help_admin')
await update.message.reply_text(msg, parse_mode='HTML', reply_markup=get_admin_keyboard(user_id))
else:
await update.message.reply_text(msg, parse_mode='HTML')
async def language_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /language command."""
user_id = update.effective_user.id
await update.message.reply_text(
t(user_id, 'select_language'),
parse_mode='HTML',
reply_markup=get_language_keyboard()
)
async def add_user(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Add user to whitelist (admin only)."""
admin_id = update.effective_user.id
if admin_id != ADMIN_ID:
await update.message.reply_text(t(admin_id, 'admin_only'))
return
if not context.args:
await update.message.reply_text(t(admin_id, 'usage_adduser'))
return
try:
user_id = int(context.args[0])
redis_client.sadd(KEY_WHITELIST, str(user_id))
await update.message.reply_text(t(admin_id, 'user_added', user_id=user_id))
logger.info(f"Admin added user {user_id} to whitelist")
except ValueError:
await update.message.reply_text(t(admin_id, 'invalid_user_id'))
async def remove_user(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Remove user from whitelist (admin only)."""
admin_id = update.effective_user.id
if admin_id != ADMIN_ID:
await update.message.reply_text(t(admin_id, 'admin_only'))
return
if not context.args:
await update.message.reply_text(t(admin_id, 'usage_removeuser'))
return
try:
user_id = int(context.args[0])
if str(user_id) == str(ADMIN_ID):
await update.message.reply_text(t(admin_id, 'cannot_remove_admin'))
return
removed = redis_client.srem(KEY_WHITELIST, str(user_id))
if removed:
await update.message.reply_text(t(admin_id, 'user_removed', user_id=user_id))
logger.info(f"Admin removed user {user_id} from whitelist")
else:
await update.message.reply_text(t(admin_id, 'user_not_in_whitelist', user_id=user_id))
except ValueError:
await update.message.reply_text(t(admin_id, 'invalid_user_id'))
# --- 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(t(user_id, '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 = t(user_id, 'whitelist_disabled') if new_state == "0" else t(user_id, 'whitelist_enabled')
await query.edit_message_reply_markup(reply_markup=get_admin_keyboard(user_id))
await query.message.reply_text(state_text)
elif data == "view_whitelist":
members = redis_client.smembers(KEY_WHITELIST)
msg = t(user_id, 'whitelist_title') + ("\n".join(members) if members else t(user_id, 'empty'))
await query.message.reply_text(msg, parse_mode='HTML')
elif data == "view_subscribers":
subs = redis_client.smembers(KEY_SUBSCRIBERS)
msg = t(user_id, 'subscribers_title') + ("\n".join(subs) if subs else t(user_id, 'empty'))
await query.message.reply_text(msg, parse_mode='HTML')
elif data == "force_check":
await query.message.reply_text(t(user_id, 'force_check_running'))
result = await check_webinars_job(context)
if result is None:
await query.message.reply_text(t(user_id, 'check_failed'))
elif result == 0:
await query.message.reply_text(t(user_id, 'check_completed_none'))
else:
await query.message.reply_text(t(user_id, 'check_completed', count=result))
async def language_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle language selection button clicks."""
query = update.callback_query
user_id = query.from_user.id
data = query.data
if data.startswith("lang_"):
lang = data.split("_")[1]
set_user_language(user_id, lang)
await query.answer()
await query.edit_message_text(
t(user_id, 'language_changed'),
parse_mode='HTML'
)
# --- Webinar Checking Job ---
def get_webinar_key(name: str, url: str) -> str:
"""Generate unique key for a webinar based on name and URL."""
return f"{name}|{url}"
def get_stored_webinars() -> list:
"""Get list of stored webinar keys from Redis."""
data = redis_client.get(KEY_WEBINAR_HISTORY)
if data:
try:
return json.loads(data)
except Exception as e:
logger.error(f"Failed to parse webinar history: {e}")
return []
def store_webinars(webinar_keys: list):
"""Store up to 5 most recent webinar keys in Redis."""
# Keep only last 5
webinar_keys = webinar_keys[-5:]
try:
redis_client.set(KEY_WEBINAR_HISTORY, json.dumps(webinar_keys))
logger.info(f"Stored {len(webinar_keys)} webinar(s) in history")
except Exception as e:
logger.error(f"Failed to store webinar history: {e}")
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.")
# --- DEBUG LOGGING ---
try:
with open('phpsessid_missing.log', 'a') as f:
f.write(f"[{os.getcwd()}] PHPSESSID missing at {context.job.last_run: %Y-%m-%d %H:%M:%S}\n")
except Exception as e:
logger.error(f"Failed to write PHPSESSID debug log: {e}")
# ---------------------
return None
current_webinars = [] # List of dicts with name, url, and formatted text
content = ""
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)
await page.wait_for_timeout(2000)
# 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):
row = rows.nth(i)
text = await row.inner_text()
if "Жодного онлайн уроку зараз" not in text:
# Extract name (topic) from first column
name_elem = row.locator('td').nth(0)
name = await name_elem.inner_text()
name = name.strip()
# Extract join URL from fourth column
url_elem = row.locator('td').nth(3).locator('a[href*="/webinar/join/"]').first
url = await url_elem.get_attribute('href')
if name and url:
current_webinars.append({
'name': name,
'url': url,
'text': text.strip()
})
logger.info(f"Found webinar: {name} -> {url}")
else:
logger.info("No webinars found (expected message present)")
except Exception as e:
logger.error(f"Error checking page: {e}. Saving content for debug.")
# If page content is available, save it on error
try:
if page and not content:
content = await page.content()
except Exception:
pass # Ignore error during content retrieval on check error
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
# --- DEBUG LOGGING (Saving last response content) ---
if not current_webinars and content: #if no webinars found, save the page content
try:
with open('response.html', 'w', encoding='utf-8') as f:
f.write(content)
logger.info("Saved page content to response.html for debug.")
except Exception as e:
logger.error(f"Failed to write debug HTML: {e}")
# -----------------------------------------------------
# Check for NEW webinars and notify
if current_webinars:
# Get stored webinar history
stored_keys = get_stored_webinars()
logger.info(f"Stored webinar keys: {stored_keys}")
# Find new webinars (not in history)
new_webinars = []
current_keys = []
for webinar in current_webinars:
key = get_webinar_key(webinar['name'], webinar['url'])
current_keys.append(key)
if key not in stored_keys:
new_webinars.append(webinar)
logger.info(f"NEW webinar detected: {webinar['name']}")
# Update stored history with current webinars
# Merge old and new, keeping only last 5
updated_keys = stored_keys + [k for k in current_keys if k not in stored_keys]
store_webinars(updated_keys)
# Notify subscribers ONLY about NEW webinars
if new_webinars:
subscribers = redis_client.smembers(KEY_SUBSCRIBERS)
logger.info(f"Sending notification about {len(new_webinars)} new webinar(s) to {len(subscribers)} subscriber(s)")
for sub_id in subscribers:
try:
# Build message in user's language
webinar_items = "\n\n".join([
t(int(sub_id), 'webinar_item', name=w['name'], url=w['url'])
for w in new_webinars
])
message = t(int(sub_id), 'webinar_found') + webinar_items
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}")
else:
logger.info(f"Found {len(current_webinars)} webinar(s), but all are already known")
return len(current_webinars)
# --- 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("language", language_command))
app.add_handler(CommandHandler("adduser", add_user))
app.add_handler(CommandHandler("removeuser", remove_user))
# Callback handlers - language selection first, then admin panel
app.add_handler(CallbackQueryHandler(language_callback, pattern="^lang_"))
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()