feat(yandextv): add yandextv schedule api

.editorconfig

@@ -6,6 +6,7 @@ indent_style = space
 indent_size = 2
 insert_final_newline = true
 trim_trailing_whitespace = true
+max_line_length = 120
 
 [*.md]
 max_line_length = off

gallery/easel/route/api/schedule.py

@@ -1,5 +1,20 @@
+import datetime
+
 from fastapi import FastAPI
 
+from gallery.easel.core import AppRequest
+from gallery.sketch.schedule.model import ChannelId, Schedule
+
 
 def mount(app: FastAPI):
-    pass
+    @app.get("/api/schedule/channels")
+    async def get_api_schedule_channels(request: AppRequest) -> list[ChannelId]:
+        schedule_api = request.app.state.api.schedule
+        return await schedule_api.get_channels()
+
+    @app.get("/api/schedule/{channel}/{date}")
+    async def get_api_schedule_channel_schedule(
+        request: AppRequest, channel: str, date: datetime.date
+    ) -> Schedule:
+        schedule_api = request.app.state.api.schedule
+        return await schedule_api.get_channel_schedule(ChannelId(channel), date)

gallery/easel/route/view/schedule/__init__.py

@@ -1,3 +1,4 @@
+import asyncio
 import datetime
 from pathlib import Path
 

gallery/main.py

@@ -7,12 +7,14 @@ from gallery.easel import build_app
 from gallery.painting.gismeteo.api import GismeteoApi
 from gallery.painting.matchtv.api import MatchTvApi
 from gallery.painting.openweather.api import OpenWeatherApi
+from gallery.painting.yandextv.api import YandexTvApi
 from gallery.sketch.bundle import ApiBundle
 from gallery.sketch.schedule.cached import CachedScheduleApi
 from gallery.sketch.weather.cached import CachedWeatherApi
 
 api = ApiBundle(
     [
+        CachedScheduleApi(YandexTvApi()),
         CachedScheduleApi(MatchTvApi()),
         CachedWeatherApi(GismeteoApi()),
         CachedWeatherApi(OpenWeatherApi()),
@@ -24,8 +26,8 @@ app = build_app(api)
 def run():
     uvicorn.run(
         "gallery.main:app",
-        host="0.0.0.0",
+        host=environ.get("GALLERY_HOST", "0.0.0.0"),
-        port=8000,
+        port=int(environ.get("GALLERY_PORT", 8000)),
         log_config=str(Path(__file__).parent / "logging.yaml"),
         reload="DEBUG" in environ,
     )

gallery/painting/matchtv/api.py

@@ -4,8 +4,7 @@ import logging
 from bs4 import BeautifulSoup
 
 from gallery.sketch.schedule.api import ScheduleApi
-from gallery.sketch.schedule.catalog import ChannelId
+from gallery.sketch.schedule.model import Channel, ChannelId, Schedule, ScheduleValue
-from gallery.sketch.schedule.model import Channel, Schedule, ScheduleValue
 from gallery.sketch.source import ApiSource
 
 logger = logging.getLogger("matchtv")
@@ -15,7 +14,7 @@ class MatchTvApi(ScheduleApi):
     PROVIDER = "matchtv"
     SOURCE = ApiSource("https://matchtv.ru")
 
-    async def get_channels(self) -> list[str]:
+    async def get_channels(self) -> list[ChannelId]:
         return [
             ChannelId.MATCH_TV,
             ChannelId.MATCH_IGRA,
@@ -27,7 +26,7 @@ class MatchTvApi(ScheduleApi):
         ]
 
     async def get_channel_schedule(
-        self, channel_id: str, date: datetime.date
+        self, channel_id: ChannelId, date: datetime.date
     ) -> Schedule:
         endpoint = f"tvguide/{channel_id}?date={date:%Y%m%d}"
         data = await self.SOURCE.request(endpoint)
@@ -46,8 +45,12 @@ class MatchTvApi(ScheduleApi):
         for item in soup.select(
             ".p-tv-guide-schedule-channel-carcass__transmissions .p-tv-guide-schedule-channel-transmission"
         ):
-            title = item.select_one(".p-tv-guide-schedule-channel-transmission__title").text.strip()
+            title = item.select_one(
-            time_str = item.select_one(".p-tv-guide-schedule-channel-transmission__time-block").text.strip()
+                ".p-tv-guide-schedule-channel-transmission__title"
+            ).text.strip()
+            time_str = item.select_one(
+                ".p-tv-guide-schedule-channel-transmission__time-block"
+            ).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:

gallery/painting/yandextv/api.py

@@ -0,0 +1,82 @@
+import datetime
+import logging
+
+from bs4 import BeautifulSoup
+
+from gallery.sketch.schedule.api import ScheduleApi
+from gallery.sketch.schedule.model import Channel, ChannelId, Schedule, ScheduleValue
+from gallery.sketch.source import ApiSource
+
+logger = logging.getLogger("matchtv")
+
+CHANNELS_MAP: dict[ChannelId, str] = {
+    ChannelId.MATCH_TV: "match-tv-49",
+    ChannelId.MATCH_IGRA: "match-igra-1174",
+    ChannelId.MATCH_ARENA: "match-arena-1173",
+    ChannelId.MATCH_FUTBOL_1: "match-futbol-1-646",
+    ChannelId.MATCH_FUTBOL_2: "match-futbol-2-593",
+    ChannelId.MATCH_FUTBOL_3: "match-futbol-3-797",
+    ChannelId.MATCH_STRANA: "match-strana-1356",
+    ChannelId.MATCH_PLANETA: "match-planeta-1177",
+    ChannelId.EUROSPORT: "eurosport-677",
+    ChannelId.EUROSPORT_2: "eurosport-2-720",
+    ChannelId.START: "start-103",
+}
+
+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 get_channels(self) -> list[ChannelId]:
+        return list(CHANNELS_MAP.keys())
+
+    async def get_channel_schedule(
+        self, channel_id: ChannelId, date: datetime.date
+    ) -> Schedule:
+        endpoint = f"channel/{CHANNELS_MAP[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), date=date, values=values
+        )

gallery/painting/yandextv/mock/__init__.py

@@ -0,0 +1,5 @@
+from pathlib import Path
+
+from gallery.sketch.mock import MockData
+
+YANDEXTV_MOCK_DATA = MockData(Path(__file__).parent / "data")

gallery/sketch/bundle.py

@@ -1,5 +1,3 @@
-from typing import Type
-
 from .api import API, Api
 from .schedule.api import ScheduleApi
 from .weather.api import WeatherApi
@@ -15,7 +13,7 @@ class ApiBundle(list[Api]):
                 return value
         raise ValueError(provider)
 
-    def get_api_by_type(self, api_type: Type[API]) -> API:
+    def get_api_by_type(self, api_type: type[API]) -> API:
         for value in self:
             if isinstance(value, api_type):
                 return value

gallery/sketch/schedule/api.py

@@ -1,14 +1,14 @@
 import datetime
 
 from ..api import Api
-from .model import Schedule
+from .model import ChannelId, Schedule
 
 
 class ScheduleApi(Api):
-    async def get_channels(self) -> list[str]:
+    async def get_channels(self) -> list[ChannelId]:
         raise NotImplementedError
 
     async def get_channel_schedule(
-        self, channel_id: str, date: datetime.date
+        self, channel_id: ChannelId, date: datetime.date
     ) -> Schedule:
         raise NotImplementedError

gallery/sketch/schedule/cached.py

@@ -3,20 +3,22 @@ import datetime
 from aiocache import cached
 
 from gallery.sketch.cached import CachedApi
+from gallery.util import TimeUnit
 
 from .api import ScheduleApi
-from .model import Schedule
+from .model import ChannelId, Schedule
 
 
 class CachedScheduleApi(ScheduleApi, CachedApi[ScheduleApi]):
     CACHE_KEY = "schedule"
+    CACHE_TTL = TimeUnit.HOUR * 6
 
     @cached(
         key_builder=lambda fun, self: f"api.{self.CACHE_KEY}.{self.provider}.channels",
         alias=CachedApi.CACHE_ALIAS,
         ttl=CachedApi.CACHE_TTL,
     )
-    async def get_channels(self) -> list[str]:
+    async def get_channels(self) -> list[ChannelId]:
         return await self._api.get_channels()
 
     @cached(
@@ -27,6 +29,6 @@ class CachedScheduleApi(ScheduleApi, CachedApi[ScheduleApi]):
         ttl=CachedApi.CACHE_TTL,
     )
     async def get_channel_schedule(
-        self, channel_id: str, date: datetime.date
+        self, channel_id: ChannelId, date: datetime.date
     ) -> Schedule:
         return await self._api.get_channel_schedule(channel_id, date)

gallery/sketch/schedule/catalog.py

@@ -1,22 +1,6 @@
-from enum import Enum
-
 from gallery.sketch.catalog import CatalogBundle
 
-from .model import Channel
+from .model import Channel, ChannelId
-
-
-class ChannelId(str, Enum):
-    MATCH_TV = "matchtv"
-    MATCH_IGRA = "igra"
-    MATCH_ARENA = "arena"
-    MATCH_FUTBOL_1 = "futbol-1"
-    MATCH_FUTBOL_2 = "futbol-2"
-    MATCH_FUTBOL_3 = "futbol-3"
-    MATCH_STRANA = "strana"
-
-    def __str__(self) -> str:
-        return self.value
-
 
 BUNDLE = CatalogBundle(
     [
@@ -27,5 +11,11 @@ BUNDLE = CatalogBundle(
         Channel(id=ChannelId.MATCH_FUTBOL_2, name="Футбол 2"),
         Channel(id=ChannelId.MATCH_FUTBOL_3, name="Футбол 3"),
         Channel(id=ChannelId.MATCH_STRANA, name="Матч! Страна"),
+        Channel(id=ChannelId.MATCH_PLANETA, name="Матч! Планета"),
+        Channel(id=ChannelId.MATCH_PLANETA, name="Матч! Планета"),
+        Channel(id=ChannelId.EUROSPORT, name="Europsort"),
+        Channel(id=ChannelId.EUROSPORT_2, name="Europsort 2"),
+        Channel(id=ChannelId.START, name="Старт!"),
+        Channel(id=ChannelId.TEST, name="Тест"),
     ]
 )

gallery/sketch/schedule/model.py

@@ -1,4 +1,5 @@
 import datetime
+from enum import StrEnum
 
 from pydantic import BaseModel
 
@@ -8,8 +9,26 @@ class Model(BaseModel):
         use_enum_values = True
 
 
+class ChannelId(StrEnum):
+    MATCH_TV = "matchtv"
+    MATCH_IGRA = "igra"
+    MATCH_ARENA = "arena"
+    MATCH_FUTBOL_1 = "futbol-1"
+    MATCH_FUTBOL_2 = "futbol-2"
+    MATCH_FUTBOL_3 = "futbol-3"
+    MATCH_STRANA = "strana"
+    MATCH_PLANETA = "planeta"
+    EUROSPORT = "eurosport"
+    EUROSPORT_2 = "eurosport-2"
+    START = "start"
+    TEST = "test"
+
+    def __str__(self) -> str:
+        return self.value
+
+
 class Channel(Model):
-    id: str
+    id: ChannelId
     name: str
 
 

gallery/sketch/source.py

@@ -19,18 +19,18 @@ class ApiSource:
         user_agent: str = DEFAULT_USER_AGENT,
         timeout: float = DEFAULT_TIMEOUT,
         cookies: dict[str, str] | None = None,
+        headers: dict[str, str] | None = None,
     ):
         self._base_url = base_url
         self._user_agent = user_agent
         self._timeout = timeout
         self._cookies = cookies
+        self._headers = headers
 
     async def request(self, endpoint: str) -> str:
         url = f"{self._base_url}/{endpoint}"
         logger.info(url)
-        headers = {
+        headers = {"User-Agent": self._user_agent, **(self._headers or {})}
-            "User-Agent": self._user_agent,
-        }
         async with aiohttp.ClientSession(
             headers=headers,
             cookies=self._cookies,

tests/test_matchtv_api.py

@@ -4,6 +4,7 @@ import pytest
 
 from gallery.painting.matchtv.api import MatchTvApi
 from gallery.painting.matchtv.mock import MATCHTV_MOCK_DATA
+from gallery.sketch.schedule.model import ChannelId
 
 
 @pytest.fixture(name="matchtv_api", scope="module")
@@ -18,6 +19,8 @@ def matchtv_api_fixture() -> MatchTvApi:
 
 
 async def test_channel(matchtv_api: MatchTvApi):
-    result = await matchtv_api.get_channel_schedule("test", datetime.date.today())
+    result = await matchtv_api.get_channel_schedule(
+        ChannelId.TEST, datetime.date.today()
+    )
     assert result is not None
     assert len(result.values) > 0

tests/test_yandextv_api.py

@@ -0,0 +1,26 @@
+import datetime
+
+import pytest
+
+from gallery.painting.yandextv.api import YandexTvApi
+from gallery.painting.yandextv.mock import YANDEXTV_MOCK_DATA
+from gallery.sketch.schedule.model import ChannelId
+
+
+@pytest.fixture(name="yandextv_api", scope="module")
+def yandextv_api_fixture() -> YandexTvApi:
+    class MockSource:
+        async def request(self, endpoint: str):
+            return YANDEXTV_MOCK_DATA.get_html(endpoint.split("/")[1].split("?")[0])
+
+    api = YandexTvApi()
+    api.SOURCE = MockSource()
+    return api
+
+
+async def test_channel(yandextv_api: YandexTvApi):
+    result = await yandextv_api.get_channel_schedule(
+        ChannelId.TEST, datetime.date.today()
+    )
+    assert result is not None
+    assert len(result.values) > 0