refactor: add base parse module

gallery/painting/gismeteo/api.py

@@ -2,17 +2,14 @@ import datetime
 import json
 import logging
 from os import environ
-from typing import Any
 
-from bs4 import BeautifulSoup
 from fastapi import HTTPException, status
 
 from gallery.sketch.source import ApiSource
 from gallery.sketch.weather.api import WeatherApi
-from gallery.sketch.weather.model import Location, WeatherResponse, WeatherValue
+from gallery.sketch.weather.model import Location, WeatherResponse
-from gallery.sketch.weather.util import merge_weather_values
 
-from .parser import DAYS_PARSER, LOCATION_PARSER, ONE_DAY_PARSER, ROW_PARSERS
+from .parser import Context, GismeteoParser
 
 logger = logging.getLogger("gismeteo")
 
@@ -28,17 +25,9 @@ def format_date(date: datetime.date) -> str:
         return f"{days + 1}-day"
 
 
-def set_dict_nested_value(target: dict, key: str, value: Any):
-    parts = key.split(".")
-    for part in parts[:-1]:
-        if part not in target:
-            target[part] = {}
-        target = target[part]
-    target[parts[-1]] = value
-
-
 class GismeteoApi(WeatherApi):
     PROVIDER = "gismeteo"
+    parser = GismeteoParser()
 
     def _build_source(self) -> ApiSource:
         return ApiSource(
@@ -46,48 +35,6 @@ class GismeteoApi(WeatherApi):
             cookies={"xauth": auth} if (auth := environ.get("GISMETEO_AUTH")) else None,
         )
 
-    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({})
-                set_dict_nested_value(result[index], parser.KEY, value)
-        values = [WeatherValue(**item) for item in result]
-        # group values by 3 hours
-        if len(values) > 8:
-            n = 3
-            values = [
-                merge_weather_values(values[i].date, values[i : i + n], "average") for i in range(0, len(values), n)
-            ]
-        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({})
-                set_dict_nested_value(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 find_locations(self, query: str) -> list[Location]:
         geo = "ru"
         latitude = 52.968498
@@ -127,8 +74,10 @@ class GismeteoApi(WeatherApi):
                 status_code=status.HTTP_400_BAD_REQUEST, detail={"max_date": max_date.strftime("%Y-%m-%d")}
             )
         data = await self.source.request(f"weather-{location_id}/{format_date(date)}")
-        return self._parse_oneday(date, data)
+        context = Context(date, "day", data, 8)
+        return self.parser.parse(context)
 
     async def get_days(self, location_id: str, days: int) -> WeatherResponse:
         data = await self.source.request(f"weather-{location_id}/{days}-days")
-        return self._parse_manydays(data)
+        context = Context(datetime.date.today(), "days", data)
+        return self.parser.parse(context)

gallery/painting/gismeteo/parser.py

@@ -1,92 +1,57 @@
 import datetime
 import logging
 import re
-from typing import Generic, Iterable, Optional, TypeVar
+from typing import Literal, NamedTuple
 
-import dateparser
+from bs4 import BeautifulSoup, Tag
-from bs4 import Tag
 
+from gallery.sketch.parse.core import ParseError, Parser
+from gallery.sketch.parse.date import parse_date
+from gallery.sketch.parse.html import TableParser, TableValueParser
 from gallery.sketch.weather.model import (
     Cloudness,
     Precipitation,
     Sky,
+    WeatherResponse,
+    WeatherValue,
     WindDirection,
 )
-from gallery.sketch.weather.parser import ParseError, parse_wind_direction
+from gallery.sketch.weather.parser import parse_wind_direction
+from gallery.sketch.weather.util import build_weather_value, merge_weather_values
 
 logger = logging.getLogger("gismeteo")
 
-T = TypeVar("T")
 
-
+class LocationParser(Parser[str, str]):
-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
-
-
-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:
+    def parse(self, data: str) -> str:
         match = self.PATTERN.search(data)
         if match:
             return match.group(1)
-        return None
+        raise ParseError("Location not found")
 
 
-LOCATION_PARSER = LocationParser()
+class TimeParser(TableValueParser[datetime.datetime]):
+    selector = ".widget-row.widget-row-datetime-time > .row-item > time-value"
+    model_key = "date"
+
+    def parse(self, data: Tag) -> datetime.datetime:
+        timestamp = int(data.attrs["timestamp"])
+        return datetime.datetime.fromtimestamp(timestamp)
 
 
-class DateParser(RowParser[datetime.datetime]):
+class DateParser(TableValueParser[datetime.datetime]):
-    KEY = "date"
+    selector = ".widget-row.widget-row-date > .row-item"
+    model_key = "date"
 
-    def parse_row(self, tag: Tag) -> Iterable[datetime.datetime]:
+    def parse(self, data: Tag) -> datetime.datetime:
-        datetime_time_row = tag.select_one(".widget-row.widget-row-datetime-time")
+        return parse_date(data.text)
-        if datetime_time_row:
-            for item in datetime_time_row.select(".row-item > time-value"):
-                timestamp = int(item.attrs["timestamp"])
-                time = datetime.datetime.fromtimestamp(timestamp)
-                yield time
-        else:
-            month = None
-            for item in tag.select(".widget-row.widget-row-date > .row-item"):
-                date_str = item.text
-                date = dateparser.parse(date_str, languages=["ru"])
-                if date is None:
-                    raise ParseError.invalid_date(date_str)
-                if month and month > date.month:
-                    date = date.replace(month=month)
-                month = date.month
-                yield date
 
 
-class SkyParser(RowParser[Sky]):
+class SkyParser(TableValueParser[Sky]):
-    KEY = "sky"
+    selector = ".widget-row[data-row=icon-tooltip] > .row-item"
+    model_key = "sky"
 
     CLOUDNESS_MAP: dict[str, Cloudness] = {
         "ясно": Cloudness.CLEAR,
@@ -124,9 +89,8 @@ class SkyParser(RowParser[Sky]):
         result_values = values - flag_values
         return result_values, len(result_values) < len(values)
 
-    def parse_row(self, tag: Tag) -> Iterable[Sky]:
+    def parse(self, data: Tag) -> Sky:
-        for item in tag.select(".widget-row[data-row=icon-tooltip] > .row-item"):
+        sky_str = data.attrs["data-tooltip"]
-            sky_str = item.attrs["data-tooltip"]
         values = {item.strip().lower() for item in sky_str.split(",")}
         cloudness = Cloudness.CLEAR
         precipitation = Precipitation.NO
@@ -144,7 +108,7 @@ class SkyParser(RowParser[Sky]):
                 break
         if values:
             logger.warning("unknown sky values: %s:", values)
-            yield Sky(
+        return Sky(
             cloudness=cloudness,
             precipitation=precipitation,
             thunder=thunder,
@@ -152,71 +116,70 @@ class SkyParser(RowParser[Sky]):
         )
 
 
-class TemperatureParser(RowParser[list[int]]):
+class TemperatureParser(TableValueParser[list[int]]):
-    KEY = "temperature"
+    selector = ".widget-row-chart[data-row=temperature-air] > .chart > .values > .value"
+    model_key = "temperature"
 
-    def parse_row(self, tag: Tag) -> Iterable[list[int]]:
+    def parse(self, data: Tag) -> list[int]:
-        for item in tag.select(".widget-row-chart[data-row=temperature-air] > .chart > .values > .value"):
+        return [int(value.attrs["value"]) for value in data.select("temperature-value")]
-            yield [int(value.attrs["value"]) for value in item.select("temperature-value")]
 
 
-class WindSpeedParser(RowParser[int]):
+class WindSpeedParser(TableValueParser[int]):
-    KEY = "wind.speed"
+    selector = ".widget-row-wind > .row-item > .wind-speed"
+    model_key = "wind.speed"
 
-    def parse_row(self, tag: Tag) -> Iterable[int]:
+    def parse(self, data: Tag) -> int:
-        for item in tag.select(".widget-row-wind > .row-item > .wind-speed > speed-value"):
+        value = data.select_one("speed-value")
-            yield int(item.attrs["value"])
+        return int(value.attrs["value"]) if value else 0
 
 
-class WindGustParser(RowParser[int]):
+class WindGustParser(TableValueParser[int]):
-    KEY = "wind.gust"
+    selector = ".widget-row-wind > .row-item > .wind-gust"
+    model_key = "wind.gust"
 
-    def parse_row(self, tag: Tag) -> Iterable[int]:
+    def parse(self, data: Tag) -> int:
-        for item in tag.select(".widget-row-wind > .row-item > .wind-gust"):
+        value = data.select_one("speed-value")
-            value = item.select_one("speed-value")
+        return int(value.attrs["value"]) if value else 0
-            yield int(value.attrs["value"]) if value else 0
 
 
-class WindDirectionParser(RowParser[WindDirection]):
+class WindDirectionParser(TableValueParser[WindDirection | None]):
-    KEY = "wind.direction"
+    selector = ".widget-row-wind > .row-item > .wind-speed > .wind-direction"
+    model_key = "wind.direction"
 
-    def parse_row(self, tag: Tag) -> Iterable[float]:
+    def parse(self, data: Tag) -> WindDirection | None:
-        for item in tag.select(".widget-row-wind > .row-item > .wind-speed > .wind-direction"):
+        return parse_wind_direction(data.text.strip())
-            wind_direction_str = item.text.lower().strip()
-            yield parse_wind_direction(wind_direction_str)
 
 
-class PrecipitationParser(RowParser[float]):
+class PrecipitationParser(TableValueParser[float]):
-    KEY = "precipitation"
+    selector = ".widget-row[data-row=precipitation-bars] > .row-item"
+    model_key = "precipitation"
 
-    def parse_row(self, tag: Tag) -> Iterable[float]:
+    def parse(self, data: Tag) -> float:
-        for item in tag.select(".widget-row[data-row=precipitation-bars] > .row-item"):
+        value = data.select_one("precipitation-value")
-            value = item.select_one("precipitation-value")
+        return float(value.attrs["value"]) if value else 0
-            if value:
-                yield float(value.attrs["value"])
-            else:
-                yield 0
 
 
-class PressureParser(RowParser[list[int]]):
+class PressureParser(TableValueParser[list[int]]):
-    KEY = "pressure"
+    selector = ".widget-row-chart[data-row=pressure] > .chart > .values > .value"
+    model_key = "pressure"
 
-    def parse_row(self, tag: Tag) -> Iterable[list[int]]:
+    def parse(self, data: Tag) -> list[int]:
-        for item in tag.select(".widget-row-chart[data-row=pressure] > .chart > .values > .value"):
+        return [int(value.attrs["value"]) for value in data.select("pressure-value")]
-            yield [int(value.attrs["value"]) for value in item.select("pressure-value")]
 
 
-class HumidityParser(RowParser[int]):
+class HumidityParser(TableValueParser[int]):
-    KEY = "humidity"
+    selector = ".widget-row[data-row=humidity] > .row-item, .widget-row[data-row=humidity-avg] > .row-item"
+    model_key = "humidity"
 
-    def parse_row(self, tag: Tag) -> Iterable[int]:
+    def parse(self, data: Tag) -> int:
-        for item in tag.select(
+        return int(data.text)
-            ".widget-row[data-row=humidity] > .row-item, .widget-row[data-row=humidity-avg] > .row-item"
-        ):
-            yield int(item.text)
 
 
-ROW_PARSERS: list[RowParser] = [
+class GismeteoTableParser(TableParser[WeatherValue]):
+    selector = ".widget .widget-items"
+    model_factory = lambda cls: build_weather_value()
+    value_parsers = [
+        TimeParser(),
         DateParser(),
         SkyParser(),
         TemperatureParser(),
@@ -226,6 +189,32 @@ ROW_PARSERS: list[RowParser] = [
         PrecipitationParser(),
         PressureParser(),
         HumidityParser(),
-]
+    ]
 
-ROW_PARSERS_MAP: dict[str, RowParser] = {parser.KEY: parser for parser in ROW_PARSERS}
+
+class Context(NamedTuple):
+    date: datetime.date
+    period: Literal["day", "days"]
+    data: str
+    values: int | None = None
+
+
+class GismeteoParser(Parser[Context, WeatherResponse]):
+    location_parser = LocationParser()
+    table_parser = GismeteoTableParser()
+
+    def parse(self, data: Context) -> WeatherResponse:
+        location = self.location_parser.parse(data.data)
+        soup = BeautifulSoup(data.data, features="html.parser")
+        values = self.table_parser.parse(soup)
+        if data.values and len(values) > data.values:
+            n = int(len((values)) / data.values)
+            values = [
+                merge_weather_values(values[i].date, values[i : i + n], "average") for i in range(0, len(values), n)
+            ]
+        return WeatherResponse(
+            location=location,
+            date=data.date,
+            period=data.period,
+            values=values,
+        )

gallery/sketch/parse/core.py

@@ -0,0 +1,13 @@
+from typing import Generic, TypeVar
+
+D = TypeVar("D")
+R = TypeVar("R")
+
+
+class ParseError(Exception):
+    pass
+
+
+class Parser(Generic[D, R]):
+    def parse(self, data: D) -> R:
+        raise NotImplementedError

gallery/sketch/parse/date.py

@@ -0,0 +1,12 @@
+import datetime
+
+import dateparser
+
+from .core import ParseError
+
+
+def parse_date(value: str) -> datetime.datetime:
+    date = dateparser.parse(value, languages=["ru"])
+    if date is None:
+        raise ParseError("Invalid date", value)
+    return date

gallery/sketch/parse/html.py

@@ -0,0 +1,36 @@
+from typing import Callable, Generic
+
+from bs4 import Tag
+
+from .core import ParseError, Parser, R
+from .util import set_nested_attr
+
+
+class TableValueParser(Parser[Tag, R], Generic[R]):
+    selector: str
+    model_key: str
+
+    def parse(self, data: Tag) -> R:
+        raise NotImplementedError
+
+
+class TableParser(Parser[Tag, R], Generic[R]):
+    selector: str
+    model_factory: Callable[[], R]
+    value_parsers: list[TableValueParser]
+
+    def create_model(self) -> R:
+        return self.model_factory()
+
+    def parse(self, data: Tag) -> list[R]:
+        table = data.select_one(self.selector)
+        if table is None:
+            raise ParseError("Table not found", self.selector)
+        result = []
+        for parser in self.value_parsers:
+            for index, tag in enumerate(table.select(parser.selector)):
+                value = parser.parse(tag)
+                while index > len(result) - 1:
+                    result.append(self.create_model())
+                set_nested_attr(result[index], parser.model_key, value)
+        return result

gallery/sketch/parse/util.py

@@ -0,0 +1,17 @@
+from typing import Any
+
+
+def set_nested_attr(target: Any, key: str, value: Any):
+    parts = key.split(".")
+    for part in parts[:-1]:
+        target = getattr(target, part)
+    setattr(target, parts[-1], value)
+
+
+def set_dict_nested_value(target: dict, key: str, value: Any):
+    parts = key.split(".")
+    for part in parts[:-1]:
+        if part not in target:
+            target[part] = {}
+        target = target[part]
+    target[parts[-1]] = value

gallery/sketch/weather/util.py

@@ -7,9 +7,9 @@ from .model import Cloudness, Precipitation, Sky, WeatherValue, Wind
 MergeMode = Literal["interval", "average"]
 
 
-def build_weather_value(date: datetime.datetime) -> WeatherValue:
+def build_weather_value(date: datetime.datetime | None = None) -> WeatherValue:
     return WeatherValue(
-        date=date,
+        date=date or datetime.datetime.now(),
         sky=Sky(
             cloudness=Cloudness.CLEAR,
             precipitation=Precipitation.NO,