feat: add schedule scraper for lessons table
ci / lint-prettier (push) Successful in 9s
ci / lint-ruff (push) Failing after 4s
ci / lint-yaml (push) Successful in 7s
ci / lint-dockerfiles (push) Successful in 6s
ci / validate (push) Successful in 7s
ci / build (push) Has been skipped

Add /schedule command to scrape edu.edu.vn.ua/lessons/table via Playwright.
Inline keyboard flow: pick weekday (with today/tomorrow shortcuts),
then pick class. Cache 5h per user. Parse subjects/notes/teachers,
multi-lesson cells (hr-separated).
This commit is contained in:
2026-09-06 15:46:41 +02:00
parent b752bf88bd
commit d2d4efb0d7
+299 -1
View File
@@ -34,6 +34,7 @@ EDU_BASE = _env('EDU_URL_BASE', 'https://edu.edu.vn.ua')
EDU_WEBINAR_PATH = _env('EDU_URL_WEBINAR', '/webinar/useractive') EDU_WEBINAR_PATH = _env('EDU_URL_WEBINAR', '/webinar/useractive')
WEBINAR_URL = f'{EDU_BASE.rstrip("/")}/{EDU_WEBINAR_PATH.lstrip("/")}' WEBINAR_URL = f'{EDU_BASE.rstrip("/")}/{EDU_WEBINAR_PATH.lstrip("/")}'
DIARY_URL = f'{EDU_BASE.rstrip("/")}/user/diary' 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)) WEBINAR_CHECK_INTERVAL = int(_env('WEBINAR_CHECK_INTERVAL', 60))
REDIS_HOST = _env('REDIS_HOST', 'redis') REDIS_HOST = _env('REDIS_HOST', 'redis')
@@ -270,6 +271,9 @@ def get_admin_keyboard(user_id: int):
# --- Diary Functions --- # --- Diary Functions ---
SCHEDULE_WEEKDAYS_FULL = ['Понеділок', 'Вівторок', 'Середа', 'Четвер', "П'ятниця", 'Субота', 'Неділя']
SCHEDULE_CACHE_TTL = 18000 # 5 hours
DIARY_MONTH_NAMES = [ DIARY_MONTH_NAMES = [
'', '',
'Січня', 'Січня',
@@ -478,6 +482,220 @@ async def _get_diary_data(context: ContextTypes.DEFAULT_TYPE) -> dict | None:
return data return data
# --- Schedule Functions ---
def get_schedule_day_keyboard():
today = datetime.now()
tomorrow = today + timedelta(days=1)
keyboard = [
[
InlineKeyboardButton(f'📌 Сьогодні ({SCHEDULE_WEEKDAYS_FULL[today.weekday()]})', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[today.weekday()]}'),
InlineKeyboardButton(f'📌 Завтра ({SCHEDULE_WEEKDAYS_FULL[tomorrow.weekday()]})', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[tomorrow.weekday()]}'),
],
[
InlineKeyboardButton('Пн', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[0]}'),
InlineKeyboardButton('Вт', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[1]}'),
InlineKeyboardButton('Ср', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[2]}'),
],
[
InlineKeyboardButton('Чт', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[3]}'),
InlineKeyboardButton('Пт', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[4]}'),
InlineKeyboardButton('Сб', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[5]}'),
],
[
InlineKeyboardButton('Нд', callback_data=f'schedule_day_{SCHEDULE_WEEKDAYS_FULL[6]}'),
],
]
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 _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:
text = re.sub(r'<[^>]+>', '\n', part)
text = re.sub(r'&nbsp;', ' ', text)
text = text.strip()
if not text:
continue
lines = [l.strip() for l in text.split('\n') if l.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': {}}
rows = re.findall(r'<tr[^>]*>(.*?)</tr>', table_html, re.DOTALL)
current_day = None
classes_header = []
for row in rows:
row_classes_match = re.search(r'class="([^"]*)"', row)
row_classes = row_classes_match.group(1) if row_classes_match else ''
if 'first-row' in row_classes:
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] = []
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 row_classes:
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):
if 'Позакласне' 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
with contextlib.suppress(Exception), open('/tmp/schedule_debug.html', 'w', encoding='utf-8') as f:
f.write(table_html)
data = _parse_schedule_html(table_html)
total_lessons = sum(len(entry['classes'].get(c, [])) for day in data['weekdays'].values() for entry in day for c in data['classes'])
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) -> str:
days = data.get('weekdays', {})
entries = days.get(weekday, [])
class_str = str(class_num)
lines = [f'📅 <b>{weekday} — Клас {class_num}</b>', '' * 20]
if not entries:
lines.append('Немає уроків')
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_num, [])
if not lessons:
lines.append(' Вільно')
else:
for i, lesson in enumerate(lessons):
note_part = f' ({lesson["note"]})' if lesson.get('note') else ''
lines.append(f' 📌 <b>{lesson["subject"]}</b>{note_part}')
if lesson.get('teacher'):
lines.append(f' 👨‍🏫 {lesson["teacher"]}')
if i < len(lessons) - 1:
lines.append(' ──')
lines.append(f'\n🔗 {SCHEDULE_URL}')
return '\n'.join(lines)
# --- Command Handlers --- # --- Command Handlers ---
@@ -675,6 +893,84 @@ async def diary_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_diary_keyboard()) await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_diary_keyboard())
# --- Schedule Command Handlers ---
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
await update.message.reply_text(
'🗓 <b>Розклад уроків</b> — виберіть день:', parse_mode='HTML', reply_markup=get_schedule_day_keyboard()
)
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('⛔ Доступ заборонено.')
return
data = query.data
if data.startswith('schedule_day_'):
weekday = data[len('schedule_day_'):]
context.user_data['schedule_weekday'] = weekday
await query.edit_message_text(f'🗓 <b>{weekday}</b> — виберіть клас:', parse_mode='HTML')
schedule_data = await _get_schedule_data(context)
if not schedule_data:
await query.edit_message_text('❌ Не вдалося завантажити розклад.')
return
classes = schedule_data.get('classes', [])
if not classes:
await query.edit_message_text('❌ Класи не знайдені в розкладі.')
return
await query.edit_message_reply_markup(reply_markup=get_schedule_class_keyboard(classes))
return
if data.startswith('schedule_class_'):
class_num_str = data[len('schedule_class_'):]
weekday = context.user_data.get('schedule_weekday')
if not weekday:
await query.edit_message_text('❌ Спочатку виберіть день.')
return
await query.edit_message_text('🔄 Завантажую розклад...')
schedule_data = await _get_schedule_data(context)
if not schedule_data:
await query.edit_message_text('❌ Не вдалося завантажити розклад.')
return
try:
class_num = int(class_num_str)
except ValueError:
class_num = class_num_str
text = format_schedule_day(schedule_data, weekday, class_num)
if len(text) > 4096:
text = text[:4090] + '\n\n✂️ ...(обрізано)'
await query.edit_message_text(text, parse_mode='HTML')
return
if data == 'schedule_refresh':
context.user_data.pop('schedule_cache', None)
await query.edit_message_text('🔄 Завантажую розклад...')
schedule_data = await _get_schedule_data(context)
if not schedule_data:
await query.edit_message_text('❌ Не вдалося завантажити розклад.')
return
await query.edit_message_text(
'🗓 <b>Розклад уроків</b> — виберіть день:', parse_mode='HTML', reply_markup=get_schedule_day_keyboard()
)
return
# --- Admin Callbacks --- # --- Admin Callbacks ---
@@ -947,9 +1243,11 @@ def main():
app.add_handler(CommandHandler('removeuser', remove_user)) app.add_handler(CommandHandler('removeuser', remove_user))
app.add_handler(CommandHandler('clearhistory', clear_history)) app.add_handler(CommandHandler('clearhistory', clear_history))
app.add_handler(CommandHandler('diary', diary_command)) app.add_handler(CommandHandler('diary', diary_command))
app.add_handler(CommandHandler('schedule', schedule_command))
# Callback handlers - diary first, then language selection, then admin panel # Callback handlers - diary/schedule first, then language selection, then admin panel
app.add_handler(CallbackQueryHandler(diary_callback, pattern='^diary_')) 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(language_callback, pattern='^lang_'))
app.add_handler(CallbackQueryHandler(admin_callback)) app.add_handler(CallbackQueryHandler(admin_callback))