import datetime import json import logging from bs4 import BeautifulSoup from gallery.sketch.schedule.api import ScheduleApi from gallery.sketch.schedule.model import Channel, Schedule, ScheduleValue from gallery.sketch.source import ApiSource logger = logging.getLogger("matchtv") HEADERS: dict[str, str] = { "Accept": ( "text/html," "application/xhtml+xml," "application/xml;q=0.9," "image/avif,image/webp," "image/apng,*/*;q=0.8," "application/signed-exchange;v=b3;q=0.9" ), "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en-US,en;q=0.9", "Connection": "keep-alive", "Host": "tv.yandex.ru", "sec-ch-ua": '"Chromium";v="100", " Not A;Brand";v="99"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Linux"', "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Upgrade-Insecure-Requests": "1", "User-Agent": ( "Mozilla/5.0 (X11; Linux x86_64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/100.0.4896.133 " "Safari/537.36" ), } class YandexTvApi(ScheduleApi): PROVIDER = "yandextv" SOURCE = ApiSource("https://tv.yandex.ru", headers=HEADERS) async def find_channels(self, query: str) -> list[Channel]: url = ( "https://suggest-multi.yandex.ru/suggest-tv2?" f"v=4&uil=ru&lr=10&count_channels=4&count_programs=4&sn=50&part={query}" ) _, values = json.loads(await self.SOURCE.request(url)) result = [] for _, name, content in values: if content["label"] == "Каналы": channel_id = content["url"].split("/")[-1] result.append(Channel(id=channel_id, name=name, provider=self.provider)) return result async def get_schedule(self, channel_id: str, date: datetime.date) -> Schedule: endpoint = f"channels/{channel_id}?date={date:%Y-%m-%d}" data = await self.SOURCE.request(endpoint) soup = BeautifulSoup(data, features="html.parser") if soup.select_one(".CheckboxCaptcha") is not None: raise RuntimeError("Captcha") values = [] channel_name = soup.select_one(".channel-header__text").text.strip() current_day = datetime.datetime.combine(date.today(), datetime.datetime.min.time()) end = current_day + datetime.timedelta(days=1, hours=6) prev_value: ScheduleValue | None = None for item in soup.select(".channel-schedule .channel-schedule__event"): title = item.select_one(".channel-schedule__title").text.strip() time_str = item.select_one(".channel-schedule__time").text.strip() hours, minutes = map(int, time_str.split(":")) item_date = current_day.replace(hour=hours, minute=minutes) if prev_value is not None and item_date.hour < prev_value.start.hour: current_day += datetime.timedelta(days=1) item_date += datetime.timedelta(days=1) live = item.select_one(".channel-schedule__info .icon_live") is not None value = ScheduleValue(start=item_date, end=end, label=title, live=live) values.append(value) if prev_value is not None: prev_value.end = item_date prev_value = value return Schedule( channel=Channel(id=channel_id, name=channel_name, provider=self.provider), date=date, values=values, )