258 lines
8.9 KiB
Python
258 lines
8.9 KiB
Python
import logging
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from typing import Optional, Dict, List
|
|
import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_phpsessid() -> Optional[str]:
|
|
"""
|
|
Получает валидный PHPSESSID через phpsessid-bot
|
|
|
|
Returns:
|
|
str: PHPSESSID или None в случае ошибки
|
|
"""
|
|
try:
|
|
url = f"{config.PHPSESSID_BOT_URL}/get-session"
|
|
logger.info(f"Requesting PHPSESSID from {url}")
|
|
|
|
response = requests.post(url, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('success'):
|
|
phpsessid = data.get('phpsessid')
|
|
logger.info(f"Got PHPSESSID: {phpsessid[:10]}...")
|
|
return phpsessid
|
|
else:
|
|
logger.error(f"Failed to get PHPSESSID: {data.get('error')}")
|
|
return None
|
|
else:
|
|
logger.error(f"PHPSESSID bot returned status {response.status_code}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting PHPSESSID: {e}", exc_info=True)
|
|
return None
|
|
|
|
|
|
def parse_webinar_table(html_content: str) -> List[Dict[str, str]]:
|
|
"""
|
|
Парсит таблицу с вебинарами
|
|
|
|
Args:
|
|
html_content: HTML контент страницы
|
|
|
|
Returns:
|
|
List[Dict]: Список вебинаров или пустой список
|
|
"""
|
|
try:
|
|
soup = BeautifulSoup(html_content, 'html.parser')
|
|
|
|
# Находим таблицу с вебинарами
|
|
meetings_div = soup.find('div', {'id': 'meetings'})
|
|
if not meetings_div:
|
|
logger.warning("meetings div not found")
|
|
return []
|
|
|
|
table = meetings_div.find('table', {'class': 'table table-zebra'})
|
|
if not table:
|
|
logger.warning("table not found")
|
|
return []
|
|
|
|
tbody = table.find('tbody')
|
|
if not tbody:
|
|
logger.warning("tbody not found")
|
|
return []
|
|
|
|
rows = tbody.find_all('tr')
|
|
if not rows:
|
|
return []
|
|
|
|
# Проверяем на сообщение "Жодного онлайн уроку зараз"
|
|
first_row = rows[0]
|
|
td = first_row.find('td')
|
|
if td and 'Жодного онлайн уроку зараз' in td.get_text(strip=True):
|
|
logger.info("No webinars available")
|
|
return []
|
|
|
|
# Парсим активные вебинары
|
|
webinars = []
|
|
for row in rows:
|
|
tds = row.find_all('td')
|
|
if len(tds) >= 4:
|
|
webinar = {
|
|
'topic': tds[0].get_text(strip=True),
|
|
'course': tds[1].get_text(strip=True),
|
|
'teacher': tds[2].get_text(strip=True),
|
|
'join_link': tds[3].find('a')['href'] if tds[3].find('a') else ''
|
|
}
|
|
webinars.append(webinar)
|
|
|
|
logger.info(f"Parsed {len(webinars)} webinars")
|
|
return webinars
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error parsing webinar table: {e}", exc_info=True)
|
|
return []
|
|
|
|
|
|
def fetch_webinars_with_playwright() -> Optional[List[Dict[str, str]]]:
|
|
"""
|
|
Получает список активных вебинаров используя Playwright для динамического контента
|
|
|
|
Returns:
|
|
List[Dict]: Список вебинаров или None в случае ошибки
|
|
"""
|
|
# Получаем PHPSESSID
|
|
phpsessid = get_phpsessid()
|
|
if not phpsessid:
|
|
logger.error("Failed to get PHPSESSID")
|
|
return None
|
|
|
|
try:
|
|
# Подготавливаем cookies для Playwright
|
|
cookies = [
|
|
{
|
|
'name': 'PHPSESSID',
|
|
'value': phpsessid,
|
|
'domain': config.EDU_HOST,
|
|
'path': '/'
|
|
}
|
|
]
|
|
|
|
# Запрос к Playwright серверу
|
|
playwright_request = {
|
|
'url': config.EDU_WEBINAR_URL,
|
|
'cookies': cookies,
|
|
'wait_until': 'networkidle', # Ждем пока сеть успокоится
|
|
'wait_time': config.WEBINAR_WAIT_TIME * 1000, # Дополнительное ожидание в миллисекундах
|
|
'user_agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36'
|
|
}
|
|
|
|
headers = {
|
|
'Authorization': f'Bearer {config.PLAYWRIGHT_API_KEY}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
logger.info(f"Fetching webinars via Playwright from {config.EDU_WEBINAR_URL}")
|
|
logger.info(f"Will wait {config.WEBINAR_WAIT_TIME} seconds for dynamic content")
|
|
|
|
response = requests.post(
|
|
f"{config.PLAYWRIGHT_SERVER}/render",
|
|
json=playwright_request,
|
|
headers=headers,
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
logger.error(f"Playwright server returned status {response.status_code}")
|
|
logger.error(f"Response: {response.text}")
|
|
return None
|
|
|
|
result = response.json()
|
|
html_content = result.get('html', '')
|
|
|
|
if not html_content:
|
|
logger.error("No HTML content in Playwright response")
|
|
return None
|
|
|
|
# Парсим таблицу
|
|
webinars = parse_webinar_table(html_content)
|
|
return webinars
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching webinars via Playwright: {e}", exc_info=True)
|
|
return None
|
|
|
|
|
|
def fetch_webinars() -> Optional[List[Dict[str, str]]]:
|
|
"""
|
|
Получает список активных вебинаров
|
|
Сначала пробует через Playwright (для динамического контента),
|
|
при неудаче - через обычный requests
|
|
|
|
Returns:
|
|
List[Dict]: Список вебинаров или None в случае ошибки
|
|
"""
|
|
# Пробуем через Playwright
|
|
logger.info("Attempting to fetch via Playwright for dynamic content")
|
|
webinars = fetch_webinars_with_playwright()
|
|
|
|
if webinars is not None:
|
|
return webinars
|
|
|
|
# Fallback на обычный requests
|
|
logger.warning("Playwright fetch failed, falling back to simple requests")
|
|
|
|
# Получаем PHPSESSID
|
|
phpsessid = get_phpsessid()
|
|
if not phpsessid:
|
|
logger.error("Failed to get PHPSESSID")
|
|
return None
|
|
|
|
# Запрашиваем страницу с вебинарами
|
|
try:
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36',
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
|
'Accept-Language': 'ru-RU,ru;q=0.9,uk;q=0.8',
|
|
'Referer': f'https://{config.EDU_HOST}/'
|
|
}
|
|
|
|
cookies = {
|
|
'PHPSESSID': phpsessid
|
|
}
|
|
|
|
logger.info(f"Fetching webinars from {config.EDU_WEBINAR_URL}")
|
|
response = requests.get(
|
|
config.EDU_WEBINAR_URL,
|
|
headers=headers,
|
|
cookies=cookies,
|
|
timeout=15
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
logger.error(f"Failed to fetch webinars page: {response.status_code}")
|
|
return None
|
|
|
|
# Парсим таблицу
|
|
webinars = parse_webinar_table(response.text)
|
|
return webinars
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching webinars: {e}", exc_info=True)
|
|
return None
|
|
|
|
|
|
def format_webinar_message(webinars: List[Dict[str, str]]) -> str:
|
|
"""
|
|
Форматирует список вебинаров для отправки в Telegram
|
|
|
|
Args:
|
|
webinars: Список вебинаров
|
|
|
|
Returns:
|
|
str: Отформатированное сообщение
|
|
"""
|
|
if not webinars:
|
|
return "📭 Жодного онлайн уроку зараз"
|
|
|
|
message = "🎓 <b>Активні онлайн уроки:</b>\n\n"
|
|
|
|
for i, webinar in enumerate(webinars, 1):
|
|
message += f"<b>{i}. {webinar['topic']}</b>\n"
|
|
message += f"📚 Курс: {webinar['course']}\n"
|
|
message += f"👨🏫 Вчитель: {webinar['teacher']}\n"
|
|
|
|
if webinar['join_link']:
|
|
full_link = webinar['join_link']
|
|
if not full_link.startswith('http'):
|
|
full_link = f"https://{config.EDU_HOST}{webinar['join_link']}"
|
|
message += f"🔗 <a href='{full_link}'>Увійти до уроку</a>\n"
|
|
|
|
message += "\n"
|
|
|
|
return message |