refactor: rename project to gallery

gallery/painting/gismeteo/api.py

@@ -0,0 +1,96 @@
+import datetime
+import logging
+from typing import Any, Dict, List
+
+import aiohttp
+from aiocache import cached
+from bs4 import BeautifulSoup
+
+from gallery.sketch.weather.api import WeatherApi
+from gallery.sketch.weather.model import WeatherResponse, WeatherValue
+
+from . import datehelp
+from .parser import DAYS_PARSER, LOCATION_PARSER, ONE_DAY_PARSER, ROW_PARSERS
+
+logger = logging.getLogger("gismeteo")
+
+
+class GismeteoApi(WeatherApi):
+    BASE_URL = "https://www.gismeteo.ru"
+    CACHE_TTL = 10 * 60
+
+    USER_AGENT = (
+        "Mozilla/5.0 (X11; Linux x86_64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/126.0.0.0 Safari/537.36"
+    )
+    COOKIE = (
+        "cf_clearance=U28mYVC0ENu88vorlL_CWmWOoevvXp0vb4xCqfqYC9s-"
+        "1722273367-1.0.1.1-"
+        "IDV73azTHY0V.NAnmEvok3zf5HHEkvF098pmya7IiqRRB5nk3FhbLCb0AeWm_kpTFqi1niFk2mYN_ramGTSl0A"
+    )
+
+    async def _request(self, endpoint: str) -> str:
+        url = f"{self.BASE_URL}/{endpoint}"
+        logger.info(url)
+        async with aiohttp.ClientSession(
+            headers={
+                "User-Agent": self.USER_AGENT,
+                "Cookie": self.COOKIE,
+            },
+            raise_for_status=True,
+        ) as session:
+            async with session.request("GET", url) as response:
+                return await response.text()
+
+    def _parse_oneday(self, date: datetime.date, data: str) -> WeatherResponse:
+        result: List[Dict[str, Any]] = []
+        soup = BeautifulSoup(data, features="html.parser")
+        location = LOCATION_PARSER.parse_location(data)
+        widget = ONE_DAY_PARSER.parse_widget(soup)
+        for parser in ROW_PARSERS:
+            for index, value in enumerate(parser.parse_row(widget)):
+                while len(result) < index + 1:
+                    result.append({})
+                result[index][parser.KEY] = value
+        values = [WeatherValue(**item) for item in result]
+        return WeatherResponse(
+            location=location or "n/a",
+            date=date,
+            period="day",
+            values=values,
+        )
+
+    def _parse_manydays(self, data: str) -> WeatherResponse:
+        result: List[Dict[str, Any]] = []
+        soup = BeautifulSoup(data, features="html.parser")
+        location = LOCATION_PARSER.parse_location(data)
+        widget = DAYS_PARSER.parse_widget(soup)
+        for parser in ROW_PARSERS:
+            for index, value in enumerate(parser.parse_row(widget)):
+                while len(result) < index + 1:
+                    result.append({})
+                result[index][parser.KEY] = value
+        values = [WeatherValue(**item) for item in result]
+        return WeatherResponse(
+            location=location or "n/a",
+            date=datetime.date.today(),
+            period="days",
+            values=values,
+        )
+
+    async def get_locations(self) -> list[str]:
+        return [
+            "orel-4432",
+            "zmiyevka-184640",
+        ]
+
+    @cached(ttl=CACHE_TTL)
+    async def get_day(self, location_id: str, date: datetime.date) -> WeatherResponse:
+        data = await self._request(f"weather-{location_id}/{datehelp.dump(date)}")
+        return self._parse_oneday(date, data)
+
+    @cached(ttl=CACHE_TTL)
+    async def get_days(self, location_id: str, days: int) -> WeatherResponse:
+        data = await self._request(f"weather-{location_id}/{days}-days")
+        return self._parse_manydays(data)

gallery/painting/gismeteo/core.py

@@ -0,0 +1,30 @@
+from typing import Generic, Iterable, Optional, TypeVar
+
+from bs4 import Tag
+
+T = TypeVar("T")
+
+class WidgetParser:
+    def parse_widget(self, tag: Tag) -> Tag:
+        raise NotImplementedError
+
+
+class BaseWidgetParser(WidgetParser):
+    SELECT: str
+
+    def __init__(self, select: Optional[str] = None):
+        super().__init__()
+        self._select = select or self.SELECT
+
+    def parse_widget(self, tag: Tag) -> Tag:
+        widget = tag.select_one(self._select)
+        if widget is None:
+            raise ValueError(self._select)
+        return widget
+
+
+class RowParser(Generic[T]):
+    KEY: str
+
+    def parse_row(self, tag: Tag) -> Iterable[T]:
+        raise NotImplementedError

gallery/painting/gismeteo/datehelp.py

@@ -0,0 +1,39 @@
+import datetime
+from enum import Enum
+
+import dateparser
+import dateparser.date_parser
+
+
+class Day(str, Enum):
+    TODAY = "today"
+    TOMORROW = "tomorrow"
+    DAY_3 = "3-day"
+    DAY_4 = "4-day"
+    DAY_5 = "5-day"
+
+
+def parse(value: str) -> datetime.date:
+    if value in ["today", "mock"]:
+        return datetime.date.today()
+    elif value == "tomorrow":
+        return datetime.date.today() + datetime.timedelta(days=1)
+    elif value.endswith("-day"):
+        days = int(value.split("-")[0]) - 1
+        return datetime.date.today() + datetime.timedelta(days=days)
+    else:
+        date = dateparser.parse(value)
+        if date is None:
+            raise ValueError(value)
+        return date.date()
+
+
+def dump(date: datetime.date) -> str:
+    today = datetime.date.today()
+    days = (date - today).days
+    if days == 0:
+        return "today"
+    elif days == 1:
+        return "tomorrow"
+    else:
+        return f"{days + 1}-day"

gallery/painting/gismeteo/mock/__init__.py

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

gallery/painting/gismeteo/parser.py

@@ -0,0 +1,186 @@
+import datetime
+import re
+from typing import Iterable
+
+import dateparser
+from bs4 import Tag
+
+from gallery.sketch.weather.model import Cloudness, Precipitation, Sky, WindDirection
+
+from .core import BaseWidgetParser, RowParser
+
+ONE_DAY_PARSER = BaseWidgetParser(".widget.widget-oneday .widget-items")
+DAYS_PARSER = BaseWidgetParser(".widget.widget-days .widget-items")
+
+
+class LocationParser:
+    PATTERN = re.compile('{"ru":{"city":{"name":"(.*?)"')
+
+    def parse_location(self, data: str) -> str | None:
+        match = self.PATTERN.search(data)
+        if match:
+            return match.group(1)
+        return None
+
+
+LOCATION_PARSER = LocationParser()
+
+
+class DateParser(RowParser[datetime.datetime]):
+    KEY = "date"
+
+    def parse_row(self, tag: Tag) -> Iterable[datetime.datetime]:
+        datetime_date_tag = tag.select_one(
+            ".widget-row.widget-row-datetime-date > .row-item"
+        )
+        if datetime_date_tag:
+            date_str = datetime_date_tag.find(text=True, recursive=False).text
+            date = dateparser.parse(date_str, languages=["ru"])
+            for item in tag.select(".widget-row.widget-row-datetime-time > .row-item"):
+                time_str = item.text
+                time = dateparser.parse(time_str, languages=["ru"])
+                time = time.replace(year=date.year, month=date.month, day=date.day)
+                yield time
+        else:
+            for item in tag.select(".widget-row.widget-row-date > .row-item"):
+                date_str = item.text
+                date = dateparser.parse(date_str, languages=["ru"])
+                yield date
+
+
+class SkyParser(RowParser[Sky]):
+    KEY = "sky"
+
+    CLOUDNESS_MAP: dict[str, Cloudness] = {
+        "ясно": Cloudness.CLEAR,
+        "малооблачно": Cloudness.PARTLY_CLOUDY,
+        "облачно": Cloudness.CLOUDY,
+        "пасмурно": Cloudness.MAINLY_CLOUDY,
+    }
+
+    PRECIPITATION_MAP: dict[str, Precipitation] = {
+        "без осадков": Precipitation.NO,
+        "небольшой  дождь": Precipitation.SMALL_RAIN,
+        "дождь": Precipitation.RAIN,
+        "ливень": Precipitation.SHOWER,
+    }
+
+    def parse_row(self, tag: Tag) -> Iterable[Sky]:
+        for item in tag.select(".widget-row[data-row=icon-tooltip] > .row-item"):
+            sky_str = item.attrs["data-tooltip"]
+            values = {item.strip().lower() for item in sky_str.split(",")}
+            cloudness = Cloudness.CLEAR
+            precipitation = Precipitation.NO
+            thunder = "гроза" in values
+            fog = "дымка" in values
+            for k, v in self.CLOUDNESS_MAP.items():
+                if k in values:
+                    cloudness = v
+                    break
+            for k, v in self.PRECIPITATION_MAP.items():
+                if k in values:
+                    precipitation = v
+                    break
+            yield Sky(
+                cloudness=cloudness,
+                precipitation=precipitation,
+                thunder=thunder,
+                fog=fog,
+            )
+
+
+class TemperatureParser(RowParser[list[int]]):
+    KEY = "temperature"
+
+    def parse_row(self, tag: Tag) -> Iterable[list[int]]:
+        for item in tag.select(
+            ".widget-row-chart[data-row=temperature-air] > .chart > .values > .value"
+        ):
+            yield [
+                int(value.attrs["value"]) for value in item.select("temperature-value")
+            ]
+
+
+class WindSpeedParser(RowParser[int]):
+    KEY = "wind_speed"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(
+            ".widget-row[data-row=wind-speed] > .row-item > speed-value"
+        ):
+            yield int(item.attrs["value"])
+
+
+class WindGustParser(RowParser[int]):
+    KEY = "wind_gust"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(".widget-row[data-row=wind-gust] > .row-item"):
+            value = item.select_one("speed-value")
+            yield int(value.attrs["value"]) if value else 0
+
+
+class WindDirectionParser(RowParser[WindDirection]):
+    KEY = "wind_direction"
+
+    WIND_DIRECTION_MAP: dict[str, WindDirection] = {
+        "штиль": WindDirection.CALM,
+        "с": WindDirection.N,
+        "св": WindDirection.NO,
+        "в": WindDirection.O,
+        "юв": WindDirection.SO,
+        "ю": WindDirection.S,
+        "юз": WindDirection.SW,
+        "з": WindDirection.W,
+        "сз": WindDirection.NW,
+    }
+
+    def parse_row(self, tag: Tag) -> Iterable[WindDirection]:
+        for item in tag.select(
+            ".widget-row[data-row=wind-direction] > .row-item > .direction"
+        ):
+            wind_direction_str = item.text.lower()
+            yield self.WIND_DIRECTION_MAP[wind_direction_str]
+
+
+class WindPrecipitationParser(RowParser[float]):
+    KEY = "precipitation"
+
+    def parse_row(self, tag: Tag) -> Iterable[float]:
+        for item in tag.select(
+            ".widget-row[data-row=precipitation-bars] > .row-item > .item-unit"
+        ):
+            yield float(item.text.replace(",", "."))
+
+
+class PressureParser(RowParser[list[int]]):
+    KEY = "pressure"
+
+    def parse_row(self, tag: Tag) -> Iterable[list[int]]:
+        for item in tag.select(
+            ".widget-row-chart[data-row=pressure] > .chart > .values > .value"
+        ):
+            yield [int(value.attrs["value"]) for value in item.select("pressure-value")]
+
+
+class HumidityParser(RowParser[int]):
+    KEY = "humidity"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(".widget-row[data-row=humidity] > .row-item"):
+            yield int(item.text)
+
+
+ROW_PARSERS: list[RowParser] = [
+    DateParser(),
+    SkyParser(),
+    TemperatureParser(),
+    WindSpeedParser(),
+    WindGustParser(),
+    WindDirectionParser(),
+    WindPrecipitationParser(),
+    PressureParser(),
+    HumidityParser(),
+]
+
+ROW_PARSERS_MAP: dict[str, RowParser] = {parser.KEY: parser for parser in ROW_PARSERS}