import asyncio import contextlib import json import logging import os import re import tempfile import threading import time from datetime import datetime, timedelta from html import escape from http.server import BaseHTTPRequestHandler, HTTPServer 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')) METRICS_PORT = int(_env('METRICS_PORT', '8000')) # --- Prometheus metrics (stdlib only, no extra deps) --- # Scraped by prometheus-stack via ServiceMonitor (edu_master/k8s/servicemonitor.yaml). # Critical alerts in edu_master/k8s/alerts.yaml fire to Telegram via Alertmanager. _METRICS_LOCK = threading.Lock() _METRICS = { 'last_run': 0.0, # Unix ts of last check start 'last_success': 0.0, # Unix ts of last successful check 'last_duration': 0.0, # Duration of last check in seconds 'success_total': 0, 'failure_total': 0, 'consecutive_failures': 0, 'phpsessid_present': 1, # 1 if EDU_PHPSESSID found in redis, else 0 } def _metric_check_start(): with _METRICS_LOCK: _METRICS['last_run'] = time.time() def _metric_check_ok(duration: float): now = time.time() with _METRICS_LOCK: _METRICS['last_success'] = now _METRICS['last_duration'] = duration _METRICS['success_total'] += 1 _METRICS['consecutive_failures'] = 0 _METRICS['phpsessid_present'] = 1 def _metric_check_fail(duration: float, phpsessid_missing: bool = False): with _METRICS_LOCK: _METRICS['last_duration'] = duration _METRICS['failure_total'] += 1 _METRICS['consecutive_failures'] += 1 _METRICS['phpsessid_present'] = 0 if phpsessid_missing else 1 def _metrics_render() -> bytes: with _METRICS_LOCK: m = dict(_METRICS) lines = [ '# HELP webinar_check_last_run_timestamp_seconds Unix timestamp of last webinar check start.', '# TYPE webinar_check_last_run_timestamp_seconds gauge', f'webinar_check_last_run_timestamp_seconds {m["last_run"]}', '# HELP webinar_check_last_success_timestamp_seconds Unix timestamp of last successful webinar check.', '# TYPE webinar_check_last_success_timestamp_seconds gauge', f'webinar_check_last_success_timestamp_seconds {m["last_success"]}', '# HELP webinar_check_last_duration_seconds Duration of last webinar check in seconds.', '# TYPE webinar_check_last_duration_seconds gauge', f'webinar_check_last_duration_seconds {m["last_duration"]}', '# HELP webinar_check_success_total Total successful webinar checks.', '# TYPE webinar_check_success_total counter', f'webinar_check_success_total {m["success_total"]}', '# HELP webinar_check_failure_total Total failed webinar checks (timeout, playwright error, page error).', '# TYPE webinar_check_failure_total counter', f'webinar_check_failure_total {m["failure_total"]}', '# HELP webinar_check_consecutive_failures Consecutive failed webinar checks (reset on success).', '# TYPE webinar_check_consecutive_failures gauge', f'webinar_check_consecutive_failures {m["consecutive_failures"]}', '# HELP edu_phpsessid_present 1 if EDU_PHPSESSID exists in redis, 0 otherwise.', '# TYPE edu_phpsessid_present gauge', f'edu_phpsessid_present {m["phpsessid_present"]}', ] return ('\n'.join(lines) + '\n').encode() class _MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path == '/metrics': body = _metrics_render() self.send_response(200) self.send_header('Content-Type', 'text/plain; version=0.0.4') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) elif self.path in ('/healthz', '/health'): body = b'ok\n' self.send_response(200) self.send_header('Content-Type', 'text/plain') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) else: self.send_response(404) self.end_headers() def log_message(self, *args): pass # keep bot logs clean def start_metrics_server(port: int = METRICS_PORT): server = HTTPServer(('0.0.0.0', port), _MetricsHandler) # noqa: S104 - k8s ServiceMonitor scrapes pod IP thread = threading.Thread(target=server.serve_forever, name='metrics-server', daemon=True) thread.start() logger.info(f'Metrics server listening on :{port}/metrics') return server # 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👑 Режим администратора активен', 'access_denied': '⛔ Доступ запрещен. Вас нет в белом списке.', 'help_title': '🤖 Помощь по боту\n\n', 'help_commands': '/start - Подписаться на уведомления\n/stop - Отписаться от уведомлений\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': '✅ Язык изменен на {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': '📅 Дневник — выберите период:', 'diary_week_title': '📅 Неделя {start} – {end}', 'schedule_title': '📅 {weekday} — {class_num} класс', 'loading_diary': '🔄 Загружаю дневник...', 'diary_load_failed': '❌ Не удалось загрузить дневник.', 'diary_no_session': '❌ Не удалось загрузить дневник. Нет сессии или ошибка.', 'diary_next_month': '❌ Данные за следующий месяц недоступны. Перейдите на сайт.', 'schedule_load_failed': '❌ Не удалось загрузить расписание.', 'class_not_set': '❌ Класс не настроен. Используйте /setclass.', 'class_not_found': '❌ Классы не найдены в расписании.', 'select_class': '🎒 Выберите класс — сохранится и больше не спросится:', 'admin_class_not_set': '⚠️ Администратор еще не настроил класс для этой группы. Используйте /setclass 11', 'loading_schedule': '🔄 Загружаю расписание...', 'invalid_class_num': '❌ Неверный номер класса.', 'class_saved': '✅ Класс {class_num} сохранен.', '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👑 Режим адміністратора активний', 'access_denied': '⛔ Доступ заборонено. Вас немає в білому списку.', 'help_title': '🤖 Довідка по боту\n\n', 'help_commands': '/start - Підписатися на сповіщення\n/stop - Відписатися від сповіщень\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': '✅ Мову змінено на {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': '📅 Щоденник — виберіть період:', 'diary_week_title': '📅 Тиждень {start} – {end}', 'schedule_title': '📅 {weekday} — {class_num} клас', 'loading_diary': '🔄 Завантажую щоденник...', 'diary_load_failed': '❌ Не вдалося завантажити щоденник.', 'diary_no_session': '❌ Не вдалося завантажити щоденник. Немає сесії або помилка.', 'diary_next_month': '❌ Дані за наступний місяць недоступні. Перейдіть на сайт.', 'schedule_load_failed': '❌ Не вдалося завантажити розклад.', 'class_not_set': '❌ Клас не налаштовано. Використайте /setclass.', 'class_not_found': '❌ Класи не знайдені в розкладі.', 'select_class': '🎒 Оберіть клас — збережеться і більше не питатиметься:', 'admin_class_not_set': '⚠️ Адміністратор ще не налаштував клас для цієї групи. Використайте /setclass 11', 'loading_schedule': '🔄 Завантажую розклад...', 'invalid_class_num': '❌ Невірний номер класу.', 'class_saved': '✅ Клас {class_num} збережено.', '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👑 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/stop - Unsubscribe from 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 {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': '📅 Diary — choose a period:', 'diary_week_title': '📅 Week {start} – {end}', 'schedule_title': '📅 {weekday} — {class_num} class', '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': '🎒 Choose a class — 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 {class_num} 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']*>(.*?)', table_html, re.DOTALL) month_text = '' for r_idx, row in enumerate(rows): cells = re.findall(r']*>(.*?)', 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 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']*>(.*?)', 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: """Read event times straight from the rendered calendar DOM. The diary page embeds `div.event-full-info[data-event-full-info-id]` containing `span.data` (e.g. "2026-09-02 16:30:00") for every event, so no AJAX popup clicks are needed. Returns {event_id: 'HH:MM'} for events that have a date; events without one are simply skipped. """ times_by_id: dict[str, str] = {} try: raw_times = await page.evaluate( """() => { const out = {}; for (const div of document.querySelectorAll( 'div.event-full-info[data-event-full-info-id]' )) { const id = div.getAttribute('data-event-full-info-id'); const date_span = div.querySelector('p.date span.data'); if (id && date_span) { out[id] = date_span.textContent.trim(); } } return out; }""" ) except Exception as e: logger.warning(f'Failed to read diary event times from DOM: {e}') return times_by_id for event_id, time_text in (raw_times or {}).items(): if not time_text: continue m = re.search(r'(\d{1,2}:\d{2})', time_text) if m: times_by_id[event_id] = m.group(1) 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 asyncio.timeout(60): async with async_playwright() as p: browser = await asyncio.wait_for(p.chromium.connect(PLAYWRIGHT_WS), timeout=15) 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 asyncio.wait_for(page.goto(DIARY_URL, wait_until='domcontentloaded'), timeout=30) 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: with contextlib.suppress(Exception): await asyncio.wait_for(page.close(), timeout=5) with contextlib.suppress(Exception): await asyncio.wait_for(context_browser.close(), timeout=5) finally: with contextlib.suppress(Exception): await asyncio.wait_for(browser.close(), timeout=5) except TimeoutError: logger.error('Diary fetch timed out (60s)') return None 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'📅 {day_num} {month_str}', '─' * 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'─ {short[i]} {d.day}.{d.month} ─') 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'📅 {month_str}\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'─ {weekday} {day_num} ─') 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'', cell_html, flags=re.IGNORECASE) for part in parts: part = re.sub(r'', '', part, flags=re.DOTALL) part = re.sub(r'