Niken
121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
from datetime import datetime, timedelta
|
|
from typing import Optional, Tuple
|
|
from playwright.async_api import async_playwright, ViewportSize, FloatRect
|
|
import logging
|
|
import aiohttp
|
|
from bs4 import BeautifulSoup
|
|
import ssl
|
|
import certifi
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ScheduleService:
|
|
def __init__(self):
|
|
self.base_url = "https://college.by/accounts/raspis/{mouth:02d}/{day:02d}-PODNAM.htm"
|
|
|
|
def _make_url(self, day: int = 0) -> Tuple[str, int, int]:
|
|
"""Генерация URL для расписания"""
|
|
d = datetime.now()
|
|
if day == 0:
|
|
if d.hour >= 12:
|
|
d += timedelta(days=1)
|
|
if d.weekday() == 6:
|
|
d += timedelta(days=1)
|
|
return self.base_url.format(day=d.day, mouth=d.month), d.day, d.month
|
|
else:
|
|
return self.base_url.format(day=int(day), mouth=d.month), int(day), int(d.month)
|
|
|
|
async def get_schedule(
|
|
self, group: str, day_offset: int = 0
|
|
) -> Tuple[str, str, int, int]:
|
|
"""Получение текста расписания (аналог Rust parse_schedule)"""
|
|
url, day, month = self._make_url(day_offset)
|
|
|
|
ssl_context = ssl.create_default_context(cafile=certifi.where())
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
|
|
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36'
|
|
}
|
|
|
|
# тут можно использовать aiohttp + chardet/charset_normalizer
|
|
async with aiohttp.ClientSession(connector=connector, headers=headers) as session:
|
|
async with session.get(url) as resp:
|
|
raw_bytes = await resp.read()
|
|
|
|
decoded = raw_bytes.decode("cp1251", errors="ignore")
|
|
document = BeautifulSoup(decoded, "html.parser")
|
|
|
|
# ищем <p class="MsoPlainText"><b>...</b>
|
|
elements = document.select("p.MsoPlainText b")
|
|
|
|
found_group = False
|
|
schedule_lines = []
|
|
for el in elements:
|
|
text = el.get_text(strip=True)
|
|
if not found_group:
|
|
if group in text:
|
|
found_group = True
|
|
schedule_lines.append(text)
|
|
else:
|
|
if "-----" in text or "+----" in text:
|
|
break
|
|
schedule_lines.append(text)
|
|
|
|
if not schedule_lines:
|
|
result = f"Расписание для группы {group} на {day} число не найдено"
|
|
else:
|
|
result = f"📅 Расписание для {day} числа:\n```\n"
|
|
for line in schedule_lines:
|
|
formatted = (
|
|
line.replace("¦", "│")
|
|
.replace(" ", " ")
|
|
.strip()
|
|
)
|
|
if formatted:
|
|
result += f"{formatted}\n"
|
|
result += "```"
|
|
|
|
return result, url, day, month
|
|
|
|
async def get_pschedule(self, group: str, day_offset: int = 0) -> Tuple[Optional[bytes], str, int, int]:
|
|
"""Получение скриншота расписания"""
|
|
url, day, month = self._make_url(day_offset)
|
|
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch(headless=True)
|
|
context = await browser.new_context(viewport=ViewportSize(width=400, height=3000))
|
|
page = await context.new_page()
|
|
|
|
try:
|
|
response = await page.goto(url, wait_until="networkidle", timeout=30000)
|
|
|
|
if not response or response.status != 200:
|
|
logger.warning(f"Ошибка загрузки страницы: {url}")
|
|
return None, url, day, month
|
|
|
|
locator = page.locator("p", has_text=group).first
|
|
if await locator.count() > 0:
|
|
await locator.scroll_into_view_if_needed()
|
|
box = await locator.bounding_box()
|
|
|
|
if box:
|
|
clip_rect = FloatRect(
|
|
x=float(max(box["x"] - 0, 0)),
|
|
y=float(max(box["y"] - 0, 0)),
|
|
width=float(box["width"] + 150),
|
|
height=float(box["height"] + 100)
|
|
)
|
|
return await page.screenshot(clip=clip_rect), url, day, month
|
|
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при получении расписания: {e}")
|
|
finally:
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
return None, url, day, month |