forust
dad9cf2104
ci / lint-prettier (push) Successful in 9s
ci / lint-ruff (push) Successful in 4s
ci / lint-yaml (push) Successful in 6s
ci / lint-dockerfiles (push) Successful in 5s
ci / validate (push) Successful in 6s
ci / build (push) Has been skipped
ci / deploy-userbot-panel (push) Has been skipped
- fetch each event's time via Playwright (click event-link, read span.data, close fancybox) and render as 'title (HH:MM)' - '08:00' placeholder renders as localized 'unknown' (time_unknown key in ru/uk/en) - diary week view shows Mon-Fri only (title ends at Friday) - diary month view skips Sat/Sun by weekday_idx with name-based fallback - schedule keyboard drops Sat/Sun day buttons
1718 lines
73 KiB
Python
1718 lines
73 KiB
Python
import contextlib
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from html import escape
|
||
|
||
import redis
|
||
from playwright.async_api import async_playwright
|
||
from telegram import ChatMember, InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||
from telegram.constants import ChatType
|
||
from telegram.ext import Application, CallbackQueryHandler, CommandHandler, ContextTypes
|
||
|
||
# Logger
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Suppress HTTP request logs
|
||
logging.getLogger('urllib3').setLevel(logging.WARNING)
|
||
logging.getLogger('httpx').setLevel(logging.WARNING)
|
||
logging.getLogger('telegram.ext._application').setLevel(logging.WARNING)
|
||
|
||
|
||
# Load environment variables
|
||
def _env(key, default=None):
|
||
v = os.getenv(key, default)
|
||
if isinstance(v, str) and len(v) >= 2 and ((v[0] == '"' and v[-1] == '"') or (v[0] == "'" and v[-1] == "'")):
|
||
return v[1:-1]
|
||
return v
|
||
|
||
|
||
EDU_BASE = _env('EDU_URL_BASE', 'https://edu.edu.vn.ua')
|
||
EDU_WEBINAR_PATH = _env('EDU_URL_WEBINAR', '/webinar/useractive')
|
||
WEBINAR_URL = f'{EDU_BASE.rstrip("/")}/{EDU_WEBINAR_PATH.lstrip("/")}'
|
||
DIARY_URL = f'{EDU_BASE.rstrip("/")}/user/diary'
|
||
SCHEDULE_URL = f'{EDU_BASE.rstrip("/")}/lessons/table'
|
||
|
||
WEBINAR_CHECK_INTERVAL = int(_env('WEBINAR_CHECK_INTERVAL', 60))
|
||
REDIS_HOST = _env('REDIS_HOST', 'redis')
|
||
REDIS_PORT = int(_env('REDIS_PORT', 6379))
|
||
PLAYWRIGHT_WS = _env('PLAYWRIGHT_WS', 'ws://playwright-service:3000/ws')
|
||
USER_AGENT = _env(
|
||
'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 = _env('WEBINAR_TELEGRAM_TOKEN')
|
||
ADMIN_ID = int(_env('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 3 webinars
|
||
|
||
# Marker shown by the site when there are no active online lessons
|
||
NO_WEBINAR_MARKER = 'Жодного онлайн уроку зараз'
|
||
|
||
# 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👑 <b>Режим администратора активен</b>',
|
||
'access_denied': '⛔ Доступ запрещен. Вас нет в белом списке.',
|
||
'help_title': '🤖 <b>Помощь по боту</b>\n\n',
|
||
'help_commands': '/start - Подписаться на уведомления\n/stop - Отписаться от уведомлений\n/help - Показать это сообщение\n/language - Сменить язык',
|
||
'help_admin': '\n<b>Команды администратора:</b>\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': '📋 <b>Белый список:</b>\n',
|
||
'subscribers_title': '👥 <b>Подписчики:</b>\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': '🎓 <b>Новый вебинар!</b>\n\n',
|
||
'webinar_item': '📌 <b>{name}</b>\n🔗 https://edu.edu.vn.ua{url}',
|
||
'select_language': '🌐 <b>Выберите язык / Оберіть мову / Select language:</b>',
|
||
'language_changed': '✅ Язык изменен на {lang}',
|
||
'flag_ru': '🇷🇺 Русский',
|
||
'flag_uk': '🇺🇦 Українська',
|
||
'flag_en': '🇬🇧 English',
|
||
'history_cleared': '✅ История вебинаров очищена',
|
||
'history_clear_failed': '❌ Ошибка при очистке истории',
|
||
'today': '📌 Сегодня',
|
||
'tomorrow': '📌 Завтра',
|
||
'week': '📅 Эта неделя',
|
||
'month': '📅 Весь месяц',
|
||
'no_events': 'Нет событий',
|
||
'no_lessons': 'Нет уроков',
|
||
'time_unknown': 'неизвестно',
|
||
'free_period': 'Свободно',
|
||
'unsubscribed': '🔕 Вы отписались от уведомлений.',
|
||
'diary_title': '📅 <b>Дневник</b> — выберите период:',
|
||
'diary_week_title': '📅 <b>Неделя {start} – {end}</b>',
|
||
'schedule_title': '📅 <b>{weekday} — {class_num} класс</b>',
|
||
'loading_diary': '🔄 Загружаю дневник...',
|
||
'diary_load_failed': '❌ Не удалось загрузить дневник.',
|
||
'diary_no_session': '❌ Не удалось загрузить дневник. Нет сессии или ошибка.',
|
||
'diary_next_month': '❌ Данные за следующий месяц недоступны. Перейдите на сайт.',
|
||
'schedule_load_failed': '❌ Не удалось загрузить расписание.',
|
||
'class_not_set': '❌ Класс не настроен. Используйте /setclass.',
|
||
'class_not_found': '❌ Классы не найдены в расписании.',
|
||
'select_class': '🎒 <b>Выберите класс</b> — сохранится и больше не спросится:',
|
||
'admin_class_not_set': '⚠️ Администратор еще не настроил класс для этой группы. Используйте /setclass 11',
|
||
'loading_schedule': '🔄 Загружаю расписание...',
|
||
'invalid_class_num': '❌ Неверный номер класса.',
|
||
'class_saved': '✅ Класс <b>{class_num}</b> сохранен.',
|
||
'admin_only_msg': '⚠️ Только администратор может настроить класс для группы.',
|
||
'truncation': '\n\n✂️ ...(обрезано)',
|
||
'day_mon': 'Пн',
|
||
'day_tue': 'Вт',
|
||
'day_wed': 'Ср',
|
||
'day_thu': 'Чт',
|
||
'day_fri': 'Пт',
|
||
'day_sat': 'Сб',
|
||
'day_sun': 'Вс',
|
||
},
|
||
'uk': {
|
||
'welcome': "👋 Привіт, {name}!\n\nЯ бот-сповіщувач про вебінари. Я повідомлятиму вас, коли з'явиться новий вебінар.\nВи підписані на сповіщення.",
|
||
'welcome_admin': '\n\n👑 <b>Режим адміністратора активний</b>',
|
||
'access_denied': '⛔ Доступ заборонено. Вас немає в білому списку.',
|
||
'help_title': '🤖 <b>Довідка по боту</b>\n\n',
|
||
'help_commands': '/start - Підписатися на сповіщення\n/stop - Відписатися від сповіщень\n/help - Показати це повідомлення\n/language - Змінити мову',
|
||
'help_admin': '\n<b>Команди адміністратора:</b>\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': '📋 <b>Білий список:</b>\n',
|
||
'subscribers_title': '👥 <b>Підписники:</b>\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': '🎓 <b>Новий вебінар!</b>\n\n',
|
||
'webinar_item': '📌 <b>{name}</b>\n🔗 https://edu.edu.vn.ua{url}',
|
||
'select_language': '🌐 <b>Виберіть мову / Выберите язык / Select language:</b>',
|
||
'language_changed': '✅ Мову змінено на {lang}',
|
||
'flag_ru': '🇷🇺 Русский',
|
||
'flag_uk': '🇺🇦 Українська',
|
||
'flag_en': '🇬🇧 English',
|
||
'history_cleared': '✅ Історія вебінарів очищена',
|
||
'history_clear_failed': '❌ Помилка при очищенні історії',
|
||
'today': '📌 Сьогодні',
|
||
'tomorrow': '📌 Завтра',
|
||
'week': '📅 Цей тиждень',
|
||
'month': '📅 Весь місяць',
|
||
'no_events': 'Немає подій',
|
||
'no_lessons': 'Немає уроків',
|
||
'time_unknown': 'невідомо',
|
||
'free_period': 'Вільно',
|
||
'unsubscribed': '🔕 Ви відписалися від сповіщень.',
|
||
'diary_title': '📅 <b>Щоденник</b> — виберіть період:',
|
||
'diary_week_title': '📅 <b>Тиждень {start} – {end}</b>',
|
||
'schedule_title': '📅 <b>{weekday} — {class_num} клас</b>',
|
||
'loading_diary': '🔄 Завантажую щоденник...',
|
||
'diary_load_failed': '❌ Не вдалося завантажити щоденник.',
|
||
'diary_no_session': '❌ Не вдалося завантажити щоденник. Немає сесії або помилка.',
|
||
'diary_next_month': '❌ Дані за наступний місяць недоступні. Перейдіть на сайт.',
|
||
'schedule_load_failed': '❌ Не вдалося завантажити розклад.',
|
||
'class_not_set': '❌ Клас не налаштовано. Використайте /setclass.',
|
||
'class_not_found': '❌ Класи не знайдені в розкладі.',
|
||
'select_class': '🎒 <b>Оберіть клас</b> — збережеться і більше не питатиметься:',
|
||
'admin_class_not_set': '⚠️ Адміністратор ще не налаштував клас для цієї групи. Використайте /setclass 11',
|
||
'loading_schedule': '🔄 Завантажую розклад...',
|
||
'invalid_class_num': '❌ Невірний номер класу.',
|
||
'class_saved': '✅ Клас <b>{class_num}</b> збережено.',
|
||
'admin_only_msg': '⚠️ Тільки адміністратор може налаштувати клас для групи.',
|
||
'truncation': '\n\n✂️ ...(обрізано)',
|
||
'day_mon': 'Пн',
|
||
'day_tue': 'Вт',
|
||
'day_wed': 'Ср',
|
||
'day_thu': 'Чт',
|
||
'day_fri': 'Пт',
|
||
'day_sat': 'Сб',
|
||
'day_sun': 'Нд',
|
||
},
|
||
'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👑 <b>Admin Mode Active</b>',
|
||
'access_denied': '⛔ Access denied. You are not on the whitelist.',
|
||
'help_title': '🤖 <b>Bot Help</b>\n\n',
|
||
'help_commands': '/start - Subscribe to notifications\n/stop - Unsubscribe from notifications\n/help - Show this message\n/language - Change language',
|
||
'help_admin': '\n<b>Admin Commands:</b>\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': '📋 <b>Whitelist:</b>\n',
|
||
'subscribers_title': '👥 <b>Subscribers:</b>\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': '🎓 <b>New webinar found!</b>\n\n',
|
||
'webinar_item': '📌 <b>{name}</b>\n🔗 https://edu.edu.vn.ua{url}',
|
||
'select_language': '🌐 <b>Select language / Виберіть мову / Выберите язык:</b>',
|
||
'language_changed': '✅ Language changed to {lang}',
|
||
'flag_ru': '🇷🇺 Русский',
|
||
'flag_uk': '🇺🇦 Українська',
|
||
'flag_en': '🇬🇧 English',
|
||
'history_cleared': '✅ Webinar history cleared',
|
||
'history_clear_failed': '❌ Error clearing history',
|
||
'today': '📌 Today',
|
||
'tomorrow': '📌 Tomorrow',
|
||
'week': '📅 This week',
|
||
'month': '📅 Whole month',
|
||
'no_events': 'No events',
|
||
'no_lessons': 'No lessons',
|
||
'time_unknown': 'unknown',
|
||
'free_period': 'Free',
|
||
'unsubscribed': '🔕 You have unsubscribed from notifications.',
|
||
'diary_title': '📅 <b>Diary</b> — choose a period:',
|
||
'diary_week_title': '📅 <b>Week {start} – {end}</b>',
|
||
'schedule_title': '📅 <b>{weekday} — {class_num} class</b>',
|
||
'loading_diary': '🔄 Loading diary...',
|
||
'diary_load_failed': '❌ Failed to load diary.',
|
||
'diary_no_session': '❌ Failed to load diary. No session or error.',
|
||
'diary_next_month': '❌ Next month data is not available. Please visit the website.',
|
||
'schedule_load_failed': '❌ Failed to load schedule.',
|
||
'class_not_set': '❌ Class is not set. Use /setclass.',
|
||
'class_not_found': '❌ No classes found in the schedule.',
|
||
'select_class': '🎒 <b>Choose a class</b> — it will be saved and not asked again:',
|
||
'admin_class_not_set': '⚠️ Admin has not set a class for this group yet. Use /setclass 11',
|
||
'loading_schedule': '🔄 Loading schedule...',
|
||
'invalid_class_num': '❌ Invalid class number.',
|
||
'class_saved': '✅ Class <b>{class_num}</b> saved.',
|
||
'admin_only_msg': '⚠️ Only an admin can set the class for this group.',
|
||
'truncation': '\n\n✂️ ...(truncated)',
|
||
'day_mon': 'Mon',
|
||
'day_tue': 'Tue',
|
||
'day_wed': 'Wed',
|
||
'day_thu': 'Thu',
|
||
'day_fri': 'Fri',
|
||
'day_sat': 'Sat',
|
||
'day_sun': 'Sun',
|
||
},
|
||
}
|
||
|
||
# --- Language Helper Functions ---
|
||
|
||
|
||
LANG_NAME_MAP = {'ru': 'Русский', 'uk': 'Українська', 'en': 'English'}
|
||
|
||
|
||
def get_user_language(user_id: int) -> str:
|
||
"""Get user's preferred language from Redis. Default: English."""
|
||
lang = redis_client.get(f'user:{user_id}:language')
|
||
return lang if lang in ['ru', 'uk', 'en'] else 'en'
|
||
|
||
|
||
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.
|
||
|
||
Fallback chain: user lang → en → uk → ru, otherwise return the key.
|
||
"""
|
||
lang = get_user_language(user_id)
|
||
message = None
|
||
for fallback in (lang, 'en', 'uk', 'ru'):
|
||
message = TRANSLATIONS.get(fallback, {}).get(key)
|
||
if message is not None:
|
||
break
|
||
if message is None:
|
||
return key
|
||
if kwargs:
|
||
return message.format(**kwargs)
|
||
return message
|
||
|
||
|
||
def _tr(lang: str, key: str, **kwargs) -> str:
|
||
"""Translate by explicit lang code (for format_* helpers).
|
||
|
||
Fallback chain: lang → en → uk → ru, otherwise return the key.
|
||
"""
|
||
message = None
|
||
for fallback in (lang, 'en', 'uk', 'ru'):
|
||
message = TRANSLATIONS.get(fallback, {}).get(key)
|
||
if message is not None:
|
||
break
|
||
if message is None:
|
||
return key
|
||
if kwargs:
|
||
return message.format(**kwargs)
|
||
return message
|
||
|
||
|
||
def get_chat_language(chat_id: int) -> str:
|
||
"""Get group chat language from Redis. Default: Ukrainian."""
|
||
lang = redis_client.get(f'chat:{chat_id}:language')
|
||
return lang if lang in ['ru', 'uk', 'en'] else 'uk'
|
||
|
||
|
||
def set_chat_language(chat_id: int, lang: str):
|
||
"""Save group chat language preference to Redis."""
|
||
if lang in ['ru', 'uk', 'en']:
|
||
redis_client.set(f'chat:{chat_id}:language', lang)
|
||
logger.info(f'Chat {chat_id} language set to {lang}')
|
||
|
||
|
||
def resolve_lang(chat, user_id=None) -> str:
|
||
"""Resolve effective language: private → user lang, groups → chat lang (default 'uk')."""
|
||
chat_id = getattr(chat, 'id', chat)
|
||
chat_type = getattr(chat, 'type', None)
|
||
if chat_type is None:
|
||
try:
|
||
is_private = int(chat_id) > 0
|
||
except (TypeError, ValueError):
|
||
is_private = True
|
||
if is_private:
|
||
uid = user_id if user_id is not None else chat_id
|
||
return get_user_language(int(uid))
|
||
return get_chat_language(int(chat_id))
|
||
if chat_type in (ChatType.PRIVATE, 'private'):
|
||
uid = user_id if user_id is not None else chat_id
|
||
return get_user_language(int(uid))
|
||
return get_chat_language(chat_id)
|
||
|
||
|
||
def t_chat(chat, user_id, key: str, **kwargs) -> str:
|
||
"""Translate using chat-resolved language (private → user, groups → chat 'uk' default)."""
|
||
return _tr(resolve_lang(chat, user_id), key, **kwargs)
|
||
|
||
|
||
def get_language_keyboard(user_id: int | None = None):
|
||
"""Generate language selection keyboard."""
|
||
lang = get_user_language(user_id) if user_id is not None else 'en'
|
||
keyboard = [
|
||
[
|
||
InlineKeyboardButton(_tr(lang, 'flag_ru'), callback_data='lang_ru'),
|
||
InlineKeyboardButton(_tr(lang, 'flag_uk'), callback_data='lang_uk'),
|
||
],
|
||
[
|
||
InlineKeyboardButton(_tr(lang, 'flag_en'), 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))
|
||
|
||
|
||
async def is_group_admin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
|
||
"""Check if the user is an administrator in the group."""
|
||
user = update.effective_user
|
||
chat = update.effective_chat
|
||
|
||
if chat.type in [ChatType.PRIVATE, 'private']:
|
||
return True
|
||
|
||
try:
|
||
member = await context.bot.get_chat_member(chat.id, user.id)
|
||
return member.status in [ChatMember.OWNER, ChatMember.ADMINISTRATOR]
|
||
except Exception as e:
|
||
logger.error(f'Failed to check admin status: {e}')
|
||
return False
|
||
|
||
|
||
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)
|
||
|
||
|
||
# --- Diary Functions ---
|
||
|
||
SCHEDULE_WEEKDAYS_FULL = {
|
||
'uk': ['Понеділок', 'Вівторок', 'Середа', 'Четвер', "П'ятниця", 'Субота', 'Неділя'],
|
||
'ru': ['Понедельник', 'Вторник', 'Среда', 'Четверг', 'Пятница', 'Суббота', 'Воскресенье'],
|
||
'en': ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'],
|
||
}
|
||
# Canonical weekday names used in callback_data and schedule lookup (site language).
|
||
SCHEDULE_WEEKDAYS_CANONICAL = SCHEDULE_WEEKDAYS_FULL['uk']
|
||
SCHEDULE_CACHE_TTL = 18000 # 5 hours
|
||
|
||
|
||
def _norm_day(s):
|
||
"""Normalize weekday name for tolerant matching (case, spaces, apostrophes)."""
|
||
if not isinstance(s, str):
|
||
return ''
|
||
s = s.strip().casefold()
|
||
for _q in ('\u2019', '\u2018', '\u02bc', '`'):
|
||
s = s.replace(_q, "'")
|
||
s = re.sub(r'\s+', ' ', s)
|
||
return s.strip()
|
||
|
||
|
||
_SCHEDULE_WEEKDAYS_CANONICAL_NORM = [_norm_day(d) for d in SCHEDULE_WEEKDAYS_CANONICAL]
|
||
|
||
DIARY_WEEKDAYS_SHORT = {
|
||
'uk': ['Пн', 'Вт', 'Ср', 'Чт', 'Пт', 'Сб', 'Нд'],
|
||
'ru': ['Пн', 'Вт', 'Ср', 'Чт', 'Пт', 'Сб', 'Вс'],
|
||
'en': ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'],
|
||
}
|
||
|
||
|
||
def get_diary_keyboard(user_id: int | None = None):
|
||
today = datetime.now()
|
||
lang = get_user_language(user_id) if user_id is not None else 'en'
|
||
today_label = _tr(lang, 'today')
|
||
tomorrow_label = _tr(lang, 'tomorrow')
|
||
week_label = _tr(lang, 'week')
|
||
month_label = _tr(lang, 'month')
|
||
keyboard = [
|
||
[
|
||
InlineKeyboardButton(f'{today_label} ({today.day}.{today.month:02d})', callback_data='diary_today'),
|
||
InlineKeyboardButton(tomorrow_label, callback_data='diary_tomorrow'),
|
||
],
|
||
[
|
||
InlineKeyboardButton(week_label, callback_data='diary_week'),
|
||
InlineKeyboardButton(month_label, callback_data='diary_month'),
|
||
],
|
||
]
|
||
return InlineKeyboardMarkup(keyboard)
|
||
|
||
|
||
def _parse_calendar_html(table_html: str) -> tuple:
|
||
"""Parse calendar HTML table into (month_text, {day_num: {weekday, events}}).
|
||
|
||
Each event is a dict: {'title': str, 'id': site event id | None, 'time': 'HH:MM' | None}.
|
||
"""
|
||
days = {}
|
||
weekdays = []
|
||
rows = re.findall(r'<tr[^>]*>(.*?)</tr>', table_html, re.DOTALL)
|
||
|
||
month_text = ''
|
||
for r_idx, row in enumerate(rows):
|
||
cells = re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row, re.DOTALL)
|
||
|
||
if r_idx == 0:
|
||
# Month navigation row: extract "Травень 2026" from nav text
|
||
raw = re.sub(r'<[^>]+>', ' ', row).strip()
|
||
raw = re.sub(r'\s+', ' ', raw)
|
||
m = re.search(r'([А-Яа-яіїєґ\']+\s*:?\s*\d{4})', raw)
|
||
month_text = m.group(1).replace(' : ', ' ').strip() if m else raw
|
||
elif r_idx == 1:
|
||
# Day names row
|
||
for cell in cells:
|
||
name = re.sub(r'<[^>]+>', '', cell).strip()
|
||
if name:
|
||
weekdays.append(name)
|
||
else:
|
||
# Data rows: each cell = a day
|
||
for col_idx, cell in enumerate(cells):
|
||
# Extract day number — first number in the cell text
|
||
text = re.sub(r'<[^>]+>', ' ', cell).strip()
|
||
text = re.sub(r'\s+', ' ', text)
|
||
dm = re.match(r'(\d+)', text)
|
||
if not dm:
|
||
continue
|
||
day_num = dm.group(1)
|
||
|
||
# Extract events: title attribute (full name) of ALL <a> tags inside the cell.
|
||
# Each event keeps its site data-event-id and a 'time' slot (filled later
|
||
# from the AJAX popup by fetch_diary_data). 'time' is 'HH:MM' or None.
|
||
events = []
|
||
for a_match in re.finditer(r'<a[^>]*>(.*?)</a>', cell, re.DOTALL):
|
||
a_tag = a_match.group(0)
|
||
# Prefer the title attribute (contains full name, not truncated)
|
||
title_m = re.search(r'title\s*=\s*"([^"]*)"', a_tag)
|
||
et = title_m.group(1).strip() if title_m else re.sub(r'<[^>]+>', '', a_match.group(1)).strip()
|
||
if not et:
|
||
continue
|
||
id_m = re.search(r'data-event-id\s*=\s*"?(\d+)"?', a_tag)
|
||
events.append(
|
||
{
|
||
'title': et,
|
||
'id': id_m.group(1) if id_m else None,
|
||
'time': None,
|
||
}
|
||
)
|
||
|
||
weekday = weekdays[col_idx] if col_idx < len(weekdays) else ''
|
||
days[day_num] = {'weekday': weekday, 'weekday_idx': col_idx, 'events': events}
|
||
|
||
return month_text, days
|
||
|
||
|
||
async def _collect_event_times(page) -> dict:
|
||
"""Click every calendar `a.event-link` and read the time from the AJAX popup.
|
||
|
||
The site shows event details only after a click fills the fancybox container
|
||
(`div.event-full-info span.data`). Returns {event_id: 'HH:MM'} for events
|
||
that have a date; events without one are simply skipped. A failed click on a
|
||
single event never breaks the whole diary parse.
|
||
"""
|
||
times_by_id: dict[str, str] = {}
|
||
try:
|
||
links = await page.query_selector_all('a.event-link')
|
||
except Exception as e:
|
||
logger.warning(f'Failed to list diary event links: {e}')
|
||
return times_by_id
|
||
|
||
for link in links:
|
||
event_id = None
|
||
event_deadline = time.monotonic() + 10
|
||
try:
|
||
event_id = await link.get_attribute('data-event-id')
|
||
if not event_id or event_id in times_by_id:
|
||
continue
|
||
await link.evaluate('(el) => el.click()')
|
||
await page.wait_for_selector(
|
||
f'div.event-full-info[data-event-full-info-id="{event_id}"] span.data',
|
||
timeout=3000,
|
||
state='visible',
|
||
)
|
||
if time.monotonic() > event_deadline:
|
||
break
|
||
time_text = await page.evaluate(
|
||
"""(id) => {
|
||
const nodes = document.querySelectorAll(
|
||
'div.event-full-info[data-event-full-info-id="' + id + '"] span.data'
|
||
);
|
||
for (const el of nodes) {
|
||
if (el.getClientRects().length > 0 || el.offsetParent !== null) {
|
||
return el.textContent.trim();
|
||
}
|
||
}
|
||
return null;
|
||
}""",
|
||
event_id,
|
||
)
|
||
if time_text:
|
||
m = re.search(r'(\d{1,2}:\d{2})', time_text)
|
||
if m:
|
||
times_by_id[event_id] = m.group(1)
|
||
except Exception as e:
|
||
logger.debug(f'Failed to read time for diary event {event_id}: {e}')
|
||
if time.monotonic() > event_deadline:
|
||
break
|
||
finally:
|
||
# Close the fancybox popup (click the close button, else Escape).
|
||
closed = False
|
||
with contextlib.suppress(Exception):
|
||
close_btn = await page.query_selector('#fancybox-close')
|
||
if close_btn is not None:
|
||
await close_btn.click(timeout=2000)
|
||
closed = True
|
||
if not closed:
|
||
with contextlib.suppress(Exception):
|
||
await page.keyboard.press('Escape')
|
||
await page.wait_for_timeout(150)
|
||
|
||
logger.info(f'Diary event times collected for {len(times_by_id)} events')
|
||
return times_by_id
|
||
|
||
|
||
async def fetch_diary_data(phpsessid: str) -> dict | None:
|
||
logger.info('Fetching diary data via Playwright...')
|
||
try:
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.connect(PLAYWRIGHT_WS)
|
||
try:
|
||
context_browser = await browser.new_context(user_agent=USER_AGENT)
|
||
await context_browser.add_cookies(
|
||
[{'name': 'PHPSESSID', 'value': phpsessid, 'domain': 'edu.edu.vn.ua', 'path': '/'}]
|
||
)
|
||
page = await context_browser.new_page()
|
||
|
||
try:
|
||
await page.goto(DIARY_URL, wait_until='domcontentloaded')
|
||
await page.wait_for_selector('table.calendar', timeout=10000)
|
||
await page.wait_for_timeout(1500)
|
||
|
||
table_html = await page.evaluate("""
|
||
() => {
|
||
const t = document.querySelector('table.calendar');
|
||
return t ? t.outerHTML : null;
|
||
}
|
||
""")
|
||
if not table_html:
|
||
logger.error('table.calendar not found in DOM')
|
||
return None
|
||
|
||
# Debug: save HTML for troubleshooting
|
||
with contextlib.suppress(Exception), open('/tmp/diary_debug.html', 'w', encoding='utf-8') as f: # noqa: S108
|
||
f.write(table_html)
|
||
|
||
month_text, days = _parse_calendar_html(table_html)
|
||
|
||
# Read event times by opening each event's AJAX popup.
|
||
times_by_id = await _collect_event_times(page)
|
||
if times_by_id:
|
||
for day_data in days.values():
|
||
for ev in day_data.get('events', []):
|
||
eid = ev.get('id')
|
||
if eid and eid in times_by_id:
|
||
ev['time'] = times_by_id[eid]
|
||
|
||
logger.info(
|
||
f'Diary parsed: month={month_text!r}, days_with_events={sum(1 for d in days.values() if d["events"])}/{len(days)}'
|
||
)
|
||
|
||
return {'monthFullText': month_text, 'days': days}
|
||
|
||
except Exception as e:
|
||
logger.error(f'Error parsing diary: {e}')
|
||
return None
|
||
finally:
|
||
await page.close()
|
||
await context_browser.close()
|
||
finally:
|
||
await browser.close()
|
||
except Exception as e:
|
||
logger.error(f'Playwright error in diary fetch: {e}')
|
||
return None
|
||
|
||
|
||
def _parse_diary_month(text: str) -> str:
|
||
match = re.search(r'([А-Яа-яіїєґ\']+\s*:\s*\d{4})', text)
|
||
if match:
|
||
return match.group(1).replace(' : ', ' ').strip()
|
||
return text.strip()
|
||
|
||
|
||
def _format_event_line(event, lang: str) -> str:
|
||
"""Render one diary event line: 📌 title, with time in parens when known.
|
||
|
||
'08:00' is the site's placeholder for "no time specified" — show a localized
|
||
'time_unknown' mark instead. Unknown/absent time renders as before, no parens.
|
||
"""
|
||
if isinstance(event, dict):
|
||
title = escape(str(event.get('title') or ''))
|
||
tme = event.get('time')
|
||
else:
|
||
title = escape(str(event))
|
||
tme = None
|
||
if tme == '08:00':
|
||
return f'📌 {title} ({_tr(lang, "time_unknown")})'
|
||
if tme:
|
||
return f'📌 {title} ({escape(str(tme))})'
|
||
return f'📌 {title}'
|
||
|
||
|
||
def format_diary_day(data: dict, day_num: int, lang: str = 'en') -> str:
|
||
days = data.get('days', {})
|
||
month_str = _parse_diary_month(data.get('monthFullText', ''))
|
||
day_data = days.get(str(day_num))
|
||
lines = [f'📅 <b>{day_num} {month_str}</b>', '─' * 18]
|
||
if not day_data or not day_data.get('events'):
|
||
lines.append(_tr(lang, 'no_events'))
|
||
else:
|
||
for e in day_data['events']:
|
||
lines.append(_format_event_line(e, lang))
|
||
lines.append(f'\n🔗 {DIARY_URL}')
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def format_diary_week(data: dict, today: datetime, lang: str = 'en') -> str:
|
||
days = data.get('days', {})
|
||
_parse_diary_month(data.get('monthFullText', ''))
|
||
monday = today - timedelta(days=today.weekday())
|
||
friday = monday + timedelta(days=4)
|
||
start = f'{monday.day}.{monday.month}'
|
||
end = f'{friday.day}.{friday.month}'
|
||
short = DIARY_WEEKDAYS_SHORT.get(lang, DIARY_WEEKDAYS_SHORT['en'])
|
||
lines = [_tr(lang, 'diary_week_title', start=start, end=end) + '\n']
|
||
for i in range(5):
|
||
d = monday + timedelta(days=i)
|
||
day_data = days.get(str(d.day))
|
||
lines.append(f'─ <b>{short[i]} {d.day}.{d.month}</b> ─')
|
||
if not day_data or not day_data.get('events'):
|
||
lines.append(_tr(lang, 'no_events') + '\n')
|
||
else:
|
||
for e in day_data['events']:
|
||
lines.append(_format_event_line(e, lang))
|
||
lines.append('')
|
||
lines.append(f'🔗 {DIARY_URL}')
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def format_diary_month(data: dict, lang: str = 'en') -> str:
|
||
days = data.get('days', {})
|
||
month_str = _parse_diary_month(data.get('monthFullText', ''))
|
||
lines = [f'📅 <b>{month_str}</b>\n']
|
||
for day_num in sorted(days.keys(), key=int):
|
||
day_data = days[day_num]
|
||
if day_data.get('weekday_idx', 0) >= 5:
|
||
continue
|
||
if day_data.get('weekday', '').strip().lower() in (
|
||
'субота',
|
||
'суббота',
|
||
'saturday',
|
||
'неділя',
|
||
'воскресенье',
|
||
'sunday',
|
||
):
|
||
continue
|
||
events = day_data.get('events', [])
|
||
weekday = day_data.get('weekday', '')
|
||
lines.append(f'─ <b>{weekday} {day_num}</b> ─')
|
||
if not events:
|
||
lines.append(_tr(lang, 'no_events') + '\n')
|
||
else:
|
||
for e in events:
|
||
lines.append(_format_event_line(e, lang))
|
||
lines.append('')
|
||
lines.append(f'🔗 {DIARY_URL}')
|
||
return '\n'.join(lines)
|
||
|
||
|
||
async def _get_diary_data(context: ContextTypes.DEFAULT_TYPE) -> dict | None:
|
||
cached = context.user_data.get('diary_cache')
|
||
now_ts = time.time()
|
||
if cached and (now_ts - cached.get('timestamp', 0)) < 300:
|
||
return cached['data']
|
||
phpsessid = redis_client.get(KEY_PHPSESSID)
|
||
if not phpsessid:
|
||
return None
|
||
data = await fetch_diary_data(phpsessid)
|
||
if data:
|
||
context.user_data['diary_cache'] = {'data': data, 'timestamp': now_ts}
|
||
return data
|
||
|
||
|
||
# --- Schedule Functions ---
|
||
|
||
|
||
def get_schedule_day_keyboard(user_id: int | None = None):
|
||
lang = get_user_language(user_id) if user_id is not None else 'en'
|
||
full = SCHEDULE_WEEKDAYS_FULL.get(lang, SCHEDULE_WEEKDAYS_FULL['en'])
|
||
today = datetime.now()
|
||
tomorrow = today + timedelta(days=1)
|
||
today_label = _tr(lang, 'today')
|
||
tomorrow_label = _tr(lang, 'tomorrow')
|
||
short_keys = ['day_mon', 'day_tue', 'day_wed', 'day_thu', 'day_fri']
|
||
keyboard = [
|
||
[
|
||
InlineKeyboardButton(
|
||
f'{today_label} ({full[today.weekday()]})',
|
||
callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[today.weekday()]}',
|
||
),
|
||
InlineKeyboardButton(
|
||
f'{tomorrow_label} ({full[tomorrow.weekday()]})',
|
||
callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[tomorrow.weekday()]}',
|
||
),
|
||
],
|
||
[
|
||
InlineKeyboardButton(
|
||
_tr(lang, short_keys[0]), callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[0]}'
|
||
),
|
||
InlineKeyboardButton(
|
||
_tr(lang, short_keys[1]), callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[1]}'
|
||
),
|
||
InlineKeyboardButton(
|
||
_tr(lang, short_keys[2]), callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[2]}'
|
||
),
|
||
],
|
||
[
|
||
InlineKeyboardButton(
|
||
_tr(lang, short_keys[3]), callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[3]}'
|
||
),
|
||
InlineKeyboardButton(
|
||
_tr(lang, short_keys[4]), callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_CANONICAL[4]}'
|
||
),
|
||
],
|
||
]
|
||
return InlineKeyboardMarkup(keyboard)
|
||
|
||
|
||
def get_schedule_class_keyboard(classes: list):
|
||
rows = []
|
||
for i in range(0, len(classes), 4):
|
||
row = [InlineKeyboardButton(str(c), callback_data=f'schedule_class_{c}') for c in classes[i : i + 4]]
|
||
rows.append(row)
|
||
return InlineKeyboardMarkup(rows)
|
||
|
||
|
||
def _get_chat_class(chat, user_id) -> str | None:
|
||
"""Get stored schedule class for a private user or a group chat."""
|
||
if chat.type == 'private':
|
||
return redis_client.get(f'user:{user_id}:schedule_class')
|
||
return redis_client.get(f'chat:{chat.id}:schedule_class')
|
||
|
||
|
||
def _set_chat_class(chat, user_id, class_num) -> None:
|
||
"""Save schedule class for a private user or a group chat."""
|
||
if chat.type == 'private':
|
||
redis_client.set(f'user:{user_id}:schedule_class', str(class_num))
|
||
else:
|
||
redis_client.set(f'chat:{chat.id}:schedule_class', str(class_num))
|
||
|
||
|
||
def _parse_schedule_cell(cell_html: str) -> list:
|
||
"""Parse a single schedule cell, returning list of {subject, note, teacher}."""
|
||
lessons = []
|
||
parts = re.split(r'<hr\s*/?>', cell_html, flags=re.IGNORECASE)
|
||
for part in parts:
|
||
part = re.sub(r'<!--.*?-->', '', part, flags=re.DOTALL)
|
||
part = re.sub(r'<!--', '', part)
|
||
text = re.sub(r'<[^>]+>', '\n', part)
|
||
text = re.sub(r' ', ' ', text)
|
||
text = text.strip()
|
||
if not text:
|
||
continue
|
||
lines = [ln.strip() for ln in text.split('\n') if ln.strip()]
|
||
if not lines:
|
||
continue
|
||
subject = lines[0] if lines else ''
|
||
note = ''
|
||
teacher = ''
|
||
for line in lines[1:]:
|
||
if (
|
||
line in ('вебінар', 'онлайн', 'асинхрон', 'асинхронно')
|
||
or 'Вебінар' in line
|
||
or 'Асинхронн' in line
|
||
or 'вебінар' in line.lower()
|
||
or 'онлайн' in line.lower()
|
||
or 'асинхрон' in line.lower()
|
||
or 'А/С урок' in line
|
||
or 'Онлайн' in line
|
||
):
|
||
note = line
|
||
else:
|
||
teacher = line
|
||
lessons.append({'subject': subject, 'note': note, 'teacher': teacher})
|
||
return lessons
|
||
|
||
|
||
def _parse_schedule_html(table_html: str) -> dict:
|
||
"""Parse schedule HTML table into {classes: [...], weekdays: {day: [...]}}."""
|
||
result = {'classes': [], 'weekdays': {}, 'weekday_norm': {}}
|
||
row_matches = re.finditer(r'<tr([^>]*)>(.*?)</tr>', table_html, re.DOTALL)
|
||
|
||
current_day = None
|
||
classes_header = []
|
||
|
||
for match in row_matches:
|
||
tr_attrs, row = match.group(1), match.group(2)
|
||
|
||
if 'first-row' in tr_attrs:
|
||
cells = re.findall(r'<td[^>]*>(.*?)</td>', row, re.DOTALL)
|
||
if cells:
|
||
day_text = re.sub(r'<[^>]+>', '', cells[0]).strip()
|
||
current_day = day_text
|
||
result['weekdays'][current_day] = []
|
||
result['weekday_norm'][_norm_day(day_text)] = current_day
|
||
classes_header = []
|
||
for cell in cells[1:]:
|
||
cls = re.sub(r'<[^>]+>', '', cell).strip()
|
||
if cls:
|
||
try:
|
||
classes_header.append(int(cls))
|
||
except ValueError:
|
||
classes_header.append(cls)
|
||
result['classes'] = classes_header
|
||
continue
|
||
|
||
if 'second-row' in tr_attrs:
|
||
continue
|
||
|
||
if current_day is None:
|
||
continue
|
||
|
||
cells = re.findall(r'<td[^>]*>(.*?)</td>', row, re.DOTALL)
|
||
if len(cells) < 3:
|
||
continue
|
||
|
||
first_cell_text = re.sub(r'<[^>]+>', '', cells[0]).strip()
|
||
if not re.match(r'^\d+\.', first_cell_text) and (
|
||
'Позакласне' in first_cell_text or re.search(r'colspan\s*=\s*"?(\d+)"?', row)
|
||
):
|
||
continue
|
||
|
||
lesson_num_match = re.match(r'^(\d+)\.', first_cell_text)
|
||
lesson_num = lesson_num_match.group(1) if lesson_num_match else ''
|
||
time_text = re.sub(r'<[^>]+>', '', cells[1]).strip()
|
||
|
||
entry = {'lesson_num': lesson_num, 'time': time_text, 'classes': {}}
|
||
|
||
data_cells = cells[2:]
|
||
for idx, cell in enumerate(data_cells):
|
||
cls = classes_header[idx] if idx < len(classes_header) else idx
|
||
lessons = _parse_schedule_cell(cell)
|
||
entry['classes'][cls] = lessons
|
||
|
||
result['weekdays'][current_day].append(entry)
|
||
|
||
return result
|
||
|
||
|
||
async def fetch_schedule_data(phpsessid: str) -> dict | None:
|
||
logger.info('Fetching schedule data via Playwright...')
|
||
try:
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.connect(PLAYWRIGHT_WS)
|
||
try:
|
||
context_browser = await browser.new_context(user_agent=USER_AGENT)
|
||
await context_browser.add_cookies(
|
||
[{'name': 'PHPSESSID', 'value': phpsessid, 'domain': 'edu.edu.vn.ua', 'path': '/'}]
|
||
)
|
||
page = await context_browser.new_page()
|
||
|
||
try:
|
||
await page.goto(SCHEDULE_URL, wait_until='domcontentloaded')
|
||
await page.wait_for_selector('table.schedule-table', timeout=10000)
|
||
await page.wait_for_timeout(1500)
|
||
|
||
table_html = await page.evaluate("""
|
||
() => {
|
||
const t = document.querySelector('table.schedule-table');
|
||
return t ? t.outerHTML : null;
|
||
}
|
||
""")
|
||
if not table_html:
|
||
logger.error('table.schedule-table not found in DOM')
|
||
return None
|
||
|
||
debug_path = os.path.join(tempfile.gettempdir(), 'schedule_debug.html')
|
||
with contextlib.suppress(Exception), open(debug_path, 'w', encoding='utf-8') as f:
|
||
f.write(table_html)
|
||
|
||
data = _parse_schedule_html(table_html)
|
||
logger.info(list(data['weekdays'].keys()))
|
||
logger.info(f'Schedule parsed: {len(data["weekdays"])} days, classes={data["classes"]}')
|
||
|
||
return data
|
||
|
||
except Exception as e:
|
||
logger.error(f'Error parsing schedule: {e}')
|
||
return None
|
||
finally:
|
||
await page.close()
|
||
await context_browser.close()
|
||
finally:
|
||
await browser.close()
|
||
except Exception as e:
|
||
logger.error(f'Playwright error in schedule fetch: {e}')
|
||
return None
|
||
|
||
|
||
async def _get_schedule_data(context: ContextTypes.DEFAULT_TYPE) -> dict | None:
|
||
cached = context.user_data.get('schedule_cache')
|
||
now_ts = time.time()
|
||
if cached and (now_ts - cached.get('timestamp', 0)) < SCHEDULE_CACHE_TTL:
|
||
return cached['data']
|
||
phpsessid = redis_client.get(KEY_PHPSESSID)
|
||
if not phpsessid:
|
||
return None
|
||
data = await fetch_schedule_data(phpsessid)
|
||
if data:
|
||
context.user_data['schedule_cache'] = {'data': data, 'timestamp': now_ts}
|
||
return data
|
||
|
||
|
||
def format_schedule_day(data: dict, weekday: str, class_num, lang: str = 'en') -> str:
|
||
days = data.get('weekdays', {})
|
||
norm_map = data.get('weekday_norm') or {_norm_day(k): k for k in days}
|
||
norm = _norm_day(weekday)
|
||
day_key = norm_map.get(norm)
|
||
if day_key is None:
|
||
logger.warning(f'Schedule weekday not matched: {weekday!r}')
|
||
entries = days.get(day_key, []) if day_key is not None else []
|
||
try:
|
||
class_key = int(class_num)
|
||
except (TypeError, ValueError):
|
||
class_key = class_num
|
||
try:
|
||
idx = _SCHEDULE_WEEKDAYS_CANONICAL_NORM.index(norm)
|
||
display_weekday = SCHEDULE_WEEKDAYS_FULL.get(lang, SCHEDULE_WEEKDAYS_FULL['en'])[idx]
|
||
except ValueError:
|
||
logger.warning(f'Schedule weekday not in canonical list: {weekday!r}')
|
||
display_weekday = weekday
|
||
lines = [_tr(lang, 'schedule_title', weekday=display_weekday, class_num=class_num), '─' * 20]
|
||
|
||
if not entries:
|
||
lines.append(_tr(lang, 'no_lessons'))
|
||
else:
|
||
for entry in entries:
|
||
time_str = entry.get('time', '').replace(' ', '–')
|
||
lines.append(f'\n<b>{entry["lesson_num"]}. </b>({time_str})')
|
||
|
||
lessons = entry.get('classes', {}).get(class_key, [])
|
||
if not lessons:
|
||
lines.append(f' {_tr(lang, "free_period")}')
|
||
else:
|
||
for i, lesson in enumerate(lessons):
|
||
note_part = f' ({escape(lesson["note"])})' if lesson.get('note') else ''
|
||
lines.append(f' 📌 <b>{escape(lesson["subject"])}</b>{note_part}')
|
||
if lesson.get('teacher'):
|
||
lines.append(f' {escape(lesson["teacher"])}')
|
||
if i < len(lessons) - 1:
|
||
lines.append(' ──')
|
||
|
||
lines.append(f'\n🔗 {SCHEDULE_URL}')
|
||
return '\n'.join(lines)
|
||
|
||
|
||
# --- Command Handlers ---
|
||
|
||
|
||
async def start(update: Update, _context: ContextTypes.DEFAULT_TYPE):
|
||
"""Handle /start command."""
|
||
user = update.effective_user
|
||
chat = update.effective_chat
|
||
logger.info(f'User {user.id} ({user.username}) started the bot in chat {chat.id} ({chat.type}).')
|
||
|
||
# Check whitelist - MUST be the user executing the command
|
||
if not is_whitelisted(user.id):
|
||
await update.message.reply_text(t(user.id, 'access_denied'))
|
||
return
|
||
|
||
# Add to subscribers (Chat ID!)
|
||
redis_client.sadd(KEY_SUBSCRIBERS, chat.id)
|
||
|
||
msg = t_chat(chat, user.id, 'welcome', name=escape(user.first_name))
|
||
|
||
if user.id == ADMIN_ID and chat.type == 'private':
|
||
msg += t_chat(chat, 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
|
||
chat = update.effective_chat
|
||
msg = t_chat(chat, user_id, 'help_title') + t_chat(chat, user_id, 'help_commands')
|
||
|
||
if user_id == ADMIN_ID and update.effective_chat.type == 'private':
|
||
msg += t_chat(chat, 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 stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Handle /stop command (unsubscribe)."""
|
||
user = update.effective_user
|
||
chat = update.effective_chat
|
||
|
||
# Permission check: Whitelisted user OR Group Admin
|
||
lang = resolve_lang(chat, user.id)
|
||
if not (is_whitelisted(user.id) or await is_group_admin(update, context)):
|
||
await update.message.reply_text(_tr(lang, 'access_denied')) # Or specific "admin only" message
|
||
return
|
||
|
||
redis_client.srem(KEY_SUBSCRIBERS, chat.id)
|
||
await update.message.reply_text(_tr(lang, 'unsubscribed'))
|
||
|
||
|
||
async def language_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Handle /language command."""
|
||
user = update.effective_user
|
||
chat = update.effective_chat
|
||
|
||
# Permission check for groups
|
||
if not (is_whitelisted(user.id) or await is_group_admin(update, context)):
|
||
return
|
||
|
||
await update.message.reply_text(
|
||
t_chat(chat, 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'))
|
||
|
||
|
||
async def clear_history(update: Update, _context: ContextTypes.DEFAULT_TYPE):
|
||
"""Clear webinar history (admin only)."""
|
||
admin_id = update.effective_user.id
|
||
if admin_id != ADMIN_ID:
|
||
await update.message.reply_text(t(admin_id, 'admin_only'))
|
||
return
|
||
|
||
try:
|
||
redis_client.delete(KEY_WEBINAR_HISTORY)
|
||
await update.message.reply_text(t(admin_id, 'history_cleared'))
|
||
logger.info('Admin cleared webinar history')
|
||
except Exception as e:
|
||
logger.error(f'Failed to clear history: {e}')
|
||
await update.message.reply_text(t(admin_id, 'history_clear_failed'))
|
||
|
||
|
||
async def diary_command(update: Update, _context: ContextTypes.DEFAULT_TYPE):
|
||
user = update.effective_user
|
||
if not is_whitelisted(user.id):
|
||
await update.message.reply_text(t(user.id, 'access_denied'))
|
||
return
|
||
await update.message.reply_text(
|
||
t(user.id, 'diary_title'), parse_mode='HTML', reply_markup=get_diary_keyboard(user.id)
|
||
)
|
||
|
||
|
||
async def diary_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
query = update.callback_query
|
||
user_id = query.from_user.id
|
||
await query.answer()
|
||
|
||
if user_id != ADMIN_ID and not is_whitelisted(user_id):
|
||
await query.edit_message_text(t(user_id, 'access_denied'))
|
||
return
|
||
|
||
data = query.data
|
||
if data == 'diary_refresh':
|
||
context.user_data.pop('diary_cache', None)
|
||
await query.edit_message_text(t(user_id, 'loading_diary'))
|
||
diary_data = await _get_diary_data(context)
|
||
if not diary_data:
|
||
await query.edit_message_text(t(user_id, 'diary_no_session'))
|
||
return
|
||
await query.edit_message_text(
|
||
t(user_id, 'diary_title'), parse_mode='HTML', reply_markup=get_diary_keyboard(user_id)
|
||
)
|
||
return
|
||
|
||
await query.edit_message_text(t(user_id, 'loading_diary'))
|
||
diary_data = await _get_diary_data(context)
|
||
if not diary_data:
|
||
await query.edit_message_text(t(user_id, 'diary_load_failed'))
|
||
return
|
||
|
||
today = datetime.now()
|
||
lang = get_user_language(user_id)
|
||
if data == 'diary_today':
|
||
text = format_diary_day(diary_data, today.day, lang=lang)
|
||
elif data == 'diary_tomorrow':
|
||
tomorrow = today + timedelta(days=1)
|
||
if tomorrow.day < today.day:
|
||
text = t(user_id, 'diary_next_month')
|
||
else:
|
||
text = format_diary_day(diary_data, tomorrow.day, lang=lang)
|
||
elif data == 'diary_week':
|
||
text = format_diary_week(diary_data, today, lang=lang)
|
||
elif data == 'diary_month':
|
||
text = format_diary_month(diary_data, lang=lang)
|
||
else:
|
||
return
|
||
|
||
if len(text) > 4096:
|
||
text = text[:4090] + t(user_id, 'truncation')
|
||
|
||
await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_diary_keyboard(user_id))
|
||
|
||
|
||
# --- Schedule Command Handlers ---
|
||
|
||
|
||
async def _schedule_text_for(context, chat, user_id, weekday) -> tuple:
|
||
"""Load schedule data and render one weekday for the stored class."""
|
||
schedule_data = await _get_schedule_data(context)
|
||
if not schedule_data:
|
||
return None, t(user_id, 'schedule_load_failed')
|
||
class_num = _get_chat_class(chat, user_id)
|
||
if not class_num:
|
||
return None, t(user_id, 'class_not_set')
|
||
text = format_schedule_day(schedule_data, weekday, class_num, lang=get_user_language(user_id))
|
||
if len(text) > 4096:
|
||
text = text[:4090] + t(user_id, 'truncation')
|
||
return text, None
|
||
|
||
|
||
async def _show_class_picker(update, context) -> bool:
|
||
"""Fetch classes from schedule and render the class-picker keyboard."""
|
||
user_id = update.effective_user.id
|
||
schedule_data = await _get_schedule_data(context)
|
||
if not schedule_data:
|
||
await update.message.reply_text(t(user_id, 'schedule_load_failed'))
|
||
return False
|
||
classes = schedule_data.get('classes', [])
|
||
if not classes:
|
||
await update.message.reply_text(t(user_id, 'class_not_found'))
|
||
return False
|
||
await update.message.reply_text(
|
||
t(user_id, 'select_class'),
|
||
parse_mode='HTML',
|
||
reply_markup=get_schedule_class_keyboard(classes),
|
||
)
|
||
return True
|
||
|
||
|
||
async def schedule_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user = update.effective_user
|
||
if not is_whitelisted(user.id):
|
||
await update.message.reply_text(t(user.id, 'access_denied'))
|
||
return
|
||
|
||
chat = update.effective_chat
|
||
stored_class = _get_chat_class(chat, user.id)
|
||
|
||
if not stored_class:
|
||
if chat.type != 'private' and not await is_group_admin(update, context):
|
||
await update.message.reply_text(t(user.id, 'admin_class_not_set'))
|
||
return
|
||
await _show_class_picker(update, context)
|
||
return
|
||
|
||
await update.message.reply_text(t(user.id, 'loading_schedule'))
|
||
today = datetime.now()
|
||
weekday = SCHEDULE_WEEKDAYS_CANONICAL[today.weekday()]
|
||
text, err = await _schedule_text_for(context, chat, user.id, weekday)
|
||
if err:
|
||
await update.message.reply_text(err)
|
||
return
|
||
await update.message.reply_text(text, parse_mode='HTML', reply_markup=get_schedule_day_keyboard(user.id))
|
||
|
||
|
||
async def setclass_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user = update.effective_user
|
||
if not is_whitelisted(user.id):
|
||
await update.message.reply_text(t(user.id, 'access_denied'))
|
||
return
|
||
|
||
chat = update.effective_chat
|
||
if chat.type != 'private' and not await is_group_admin(update, context):
|
||
await update.message.reply_text(t(user.id, 'admin_only'))
|
||
return
|
||
|
||
if context.args:
|
||
try:
|
||
class_num = int(context.args[0])
|
||
except ValueError:
|
||
await update.message.reply_text(t(user.id, 'invalid_class_num'))
|
||
return
|
||
_set_chat_class(chat, user.id, class_num)
|
||
await update.message.reply_text(t(user.id, 'class_saved', class_num=class_num), parse_mode='HTML')
|
||
return
|
||
|
||
await _show_class_picker(update, context)
|
||
|
||
|
||
async def schedule_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
query = update.callback_query
|
||
user_id = query.from_user.id
|
||
await query.answer()
|
||
|
||
if user_id != ADMIN_ID and not is_whitelisted(user_id):
|
||
await query.edit_message_text(t(user_id, 'access_denied'))
|
||
return
|
||
|
||
data = query.data
|
||
|
||
if data.startswith('schedule_class_'):
|
||
class_num_str = data[len('schedule_class_') :]
|
||
try:
|
||
class_num = int(class_num_str)
|
||
except ValueError:
|
||
await query.edit_message_text(t(user_id, 'invalid_class_num'))
|
||
return
|
||
|
||
chat = update.effective_chat
|
||
if chat.type != 'private' and not await is_group_admin(update, context):
|
||
await query.edit_message_text(t(user_id, 'admin_only_msg'))
|
||
return
|
||
|
||
_set_chat_class(chat, user_id, class_num)
|
||
await query.edit_message_text(t(user_id, 'loading_schedule'))
|
||
|
||
today = datetime.now()
|
||
weekday = SCHEDULE_WEEKDAYS_CANONICAL[today.weekday()]
|
||
text, err = await _schedule_text_for(context, chat, user_id, weekday)
|
||
if err:
|
||
await query.edit_message_text(err)
|
||
return
|
||
await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_schedule_day_keyboard(user_id))
|
||
return
|
||
|
||
if data.startswith('schedule_day_'):
|
||
weekday = data[len('schedule_day_') :]
|
||
try:
|
||
_SCHEDULE_WEEKDAYS_CANONICAL_NORM.index(_norm_day(weekday))
|
||
except ValueError:
|
||
logger.warning(f'Schedule callback weekday not in canonical list: {weekday!r}')
|
||
await query.edit_message_text(t(user_id, 'loading_schedule'))
|
||
text, err = await _schedule_text_for(context, query.message.chat, user_id, weekday)
|
||
if err:
|
||
await query.edit_message_text(err)
|
||
return
|
||
await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_schedule_day_keyboard(user_id))
|
||
return
|
||
|
||
if data == 'schedule_refresh':
|
||
context.user_data.pop('schedule_cache', None)
|
||
await query.edit_message_text(t(user_id, 'loading_schedule'))
|
||
today = datetime.now()
|
||
weekday = SCHEDULE_WEEKDAYS_CANONICAL[today.weekday()]
|
||
text, err = await _schedule_text_for(context, query.message.chat, user_id, weekday)
|
||
if err:
|
||
await query.edit_message_text(err)
|
||
return
|
||
await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_schedule_day_keyboard(user_id))
|
||
return
|
||
|
||
|
||
# --- 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
|
||
chat = update.effective_chat
|
||
data = query.data
|
||
|
||
if data.startswith('lang_'):
|
||
lang = data.split('_')[1]
|
||
is_private = chat is None or getattr(chat, 'type', 'private') in (ChatType.PRIVATE, 'private')
|
||
if is_private:
|
||
set_user_language(user_id, lang)
|
||
await query.answer()
|
||
await query.edit_message_text(
|
||
t(user_id, 'language_changed', lang=LANG_NAME_MAP.get(lang, lang)), parse_mode='HTML'
|
||
)
|
||
else:
|
||
if not await is_group_admin(update, context):
|
||
await query.answer(t_chat(chat, user_id, 'admin_only'), show_alert=True)
|
||
return
|
||
set_chat_language(chat.id, lang)
|
||
await query.answer()
|
||
await query.edit_message_text(
|
||
_tr(lang, 'language_changed', lang=LANG_NAME_MAP.get(lang, lang)), parse_mode='HTML'
|
||
)
|
||
|
||
|
||
# --- Webinar Checking Job ---
|
||
|
||
|
||
def get_webinar_key(url: str) -> str:
|
||
"""Generate unique key for a webinar based on URL."""
|
||
return 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 3 most recent webinar keys in Redis."""
|
||
# Keep only last 3
|
||
webinar_keys = webinar_keys[-3:]
|
||
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 NO_WEBINAR_MARKER 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 NO_WEBINAR_MARKER 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
|
||
with contextlib.suppress(Exception):
|
||
if page and not content:
|
||
content = await page.content()
|
||
|
||
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['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
|
||
# sub_id comes from redis set as string, convert to int for translation lookup
|
||
webinar_items = '\n\n'.join(
|
||
[
|
||
t_chat(
|
||
int(sub_id), int(sub_id), 'webinar_item', name=escape(w['name']), url=escape(w['url'])
|
||
)
|
||
for w in new_webinars
|
||
]
|
||
)
|
||
message = t_chat(int(sub_id), 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('stop', stop_command))
|
||
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))
|
||
app.add_handler(CommandHandler('clearhistory', clear_history))
|
||
app.add_handler(CommandHandler('diary', diary_command))
|
||
app.add_handler(CommandHandler('schedule', schedule_command))
|
||
app.add_handler(CommandHandler('setclass', setclass_command))
|
||
|
||
# Callback handlers - diary/schedule first, then language selection, then admin panel
|
||
app.add_handler(CallbackQueryHandler(diary_callback, pattern='^diary_'))
|
||
app.add_handler(CallbackQueryHandler(schedule_callback, pattern='^schedule_'))
|
||
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()
|