fix(rp5): fix rp5 parser

2026-09-10 18:48:23 +03:00
parent fdf6e0eac3
commit 46e72feb82
7 changed files with 222 additions and 212 deletions
@@ -12,12 +12,22 @@ from gallery.sketch.weather.api import WeatherApi
 from gallery.sketch.weather.model import Location, WeatherResponse, WeatherValue
 from gallery.sketch.weather.util import merge_weather_values

-from . import datehelp
 from .parser import DAYS_PARSER, LOCATION_PARSER, ONE_DAY_PARSER, ROW_PARSERS

 logger = logging.getLogger("gismeteo")


+def format_date(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"
+
+
 class GismeteoApi(WeatherApi):
    PROVIDER = "gismeteo"

@@ -107,7 +117,7 @@ class GismeteoApi(WeatherApi):
            raise HTTPException(
                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}/{datehelp.dump(date)}")
+        data = await self.source.request(f"weather-{location_id}/{format_date(date)}")
        return self._parse_oneday(date, data)

    async def get_days(self, location_id: str, days: int) -> WeatherResponse:
@@ -1,31 +0,0 @@
-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
@@ -1,39 +0,0 @@
-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"
@@ -1,7 +1,7 @@
 import datetime
 import logging
 import re
-from typing import Iterable
+from typing import Generic, Iterable, Optional, TypeVar

 import dateparser
 from bs4 import Tag
@@ -14,10 +14,37 @@ from gallery.sketch.weather.model import (
 )
 from gallery.sketch.weather.parser import ParseError, parse_wind_direction

-from .core import BaseWidgetParser, RowParser
-
 logger = logging.getLogger("gismeteo")

+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
+
+
 ONE_DAY_PARSER = BaseWidgetParser(".widget.widget-oneday .widget-items")
 DAYS_PARSER = BaseWidgetParser(".widget.widget-days .widget-items")

@@ -1,4 +1,6 @@
 import re
+from enum import Enum, auto
+from typing import Iterable

 import dateparser
 from bs4 import BeautifulSoup, Tag
@@ -36,72 +38,109 @@ def _parse_wind_speed(cell: Tag) -> int:
    return 0


+class RowType(Enum):
+    DATE = auto()
+    TIME = auto()
+    CLOUDNESS = auto()
+    PRECIPITATION = auto()
+    TEMPERATURE = auto()
+    PRESSURE = auto()
+    WIND_SPEED = auto()
+    WIND_GUST = auto()
+    WIND_DIRECTION = auto()
+    HUMIDITY = auto()
+
+
+def _iter_table_rows(table: Tag) -> Iterable[tuple[RowType, Tag]]:
+    for row in table.select("tr"):
+        title = row.select_one("td").text.strip().lower()
+        if "сегодня" in title:
+            yield RowType.DATE, row
+        elif "местное время" in title:
+            yield RowType.TIME, row
+        elif "облачность" in title:
+            yield RowType.CLOUDNESS, row
+        elif "осадки" in title:
+            yield RowType.PRECIPITATION, row
+        elif "температура" in title:
+            yield RowType.TEMPERATURE, row
+        elif "давление" in title:
+            yield RowType.PRESSURE, row
+        elif "скорость" in title:
+            yield RowType.WIND_SPEED, row
+        elif "порывы" in title:
+            yield RowType.WIND_GUST, row
+        elif "направление" in title:
+            yield RowType.WIND_DIRECTION, row
+        elif "влажность" in title:
+            yield RowType.HUMIDITY, row
+
+
 # pylint:disable=too-many-branches
 def parse_page(page: BeautifulSoup) -> list[WeatherResponse]:
    location = page.select(".intoLeftNavi > .verticalBottom > a")[-1].text.strip()
    table = page.select_one("#forecastTable")
    if table is None:
        raise ParseError("#forecastTable not found")
-    (
-        date_row,
-        time_row,
-        cloudness_row,
-        precipitation_row,
-        temperature_row,
-        pressure_row,
-        wind_speed_row,
-        wind_gust_row,
-        wind_direction_row,
-        humidity_row,
-        *_,
-    ) = table.select("tr")
    result: list[WeatherResponse] = []
    values: list[WeatherValue] = []
-    for index, cell in enumerate(date_row.select("td")):
-        colspan = int(cell.attrs["colspan"])
-        if index == 0:
-            colspan -= 1
-        date_str = ",".join(cell.text.strip().split(",")[-2:])
-        date = dateparser.parse(date_str, languages=["ru"])
-        if date is None:
-            raise ParseError.invalid_date(date_str)
-        item_values = [build_weather_value(date) for _ in range(colspan)]
-        result.append(
-            WeatherResponse(
-                location=location,
-                date=date.date(),
-                period="day",
-                values=item_values,
-            )
-        )
-        values += item_values
-    for index, cell in enumerate(time_row.select("td")[1:]):
-        hour = int(cell.text.strip())
-        values[index].date = values[index].date.replace(hour=hour)
-    for index, cell in enumerate(cloudness_row.select("td")[1:]):
-        onmouseover = cell.select_one(".cc_0").next_element.attrs["onmouseover"]
-        cloudness_str = re.compile("<b>(.*?)</b>").search(onmouseover).group(1).strip()
-        values[index].sky.cloudness = CLOUDNESS_MAP[cloudness_str]
-    for index, cell in enumerate(precipitation_row.select("td")[1:]):
-        onmouseover = cell.select_one(".pr_0").attrs["onmouseover"]
-        precipitation_str = onmouseover.split(",")[1].strip(" '")
-        precipitation_value_matched = re.compile("(\\d+\\.\\d+) мм").search(precipitation_str)
-        if precipitation_value_matched:
-            # one hour value
-            values[index].precipitation = float(precipitation_value_matched.group(1)) * 6
-        if "(" in precipitation_str:
-            precipitation_str = precipitation_str.split(" (")[0]
-        values[index].sky.precipitation = PRECIPITATION_MAP[precipitation_str]
-    for index, cell in enumerate(temperature_row.select("td")[1:]):
-        values[index].temperature = [int(cell.select_one(".t_0").text.lstrip("+").strip())]
-    for index, cell in enumerate(pressure_row.select("td")[1:]):
-        values[index].pressure = [int(cell.select_one(".p_0").text.strip())]
-    for index, cell in enumerate(wind_speed_row.select("td")[1:]):
-        values[index].wind_speed = _parse_wind_speed(cell)
-    for index, cell in enumerate(wind_gust_row.select("td")[1:]):
-        values[index].wind_gust = _parse_wind_speed(cell)
-    for index, cell in enumerate(wind_direction_row.select("td")[1:]):
-        values[index].wind_direction = parse_wind_direction(cell.text.strip())
-    for index, cell in enumerate(humidity_row.select("td")[1:]):
-        values[index].humidity = int(cell.text.strip())
+    for row_type, row in _iter_table_rows(table):
+        match row_type:
+            case RowType.DATE:
+                for index, cell in enumerate(row.select("td")):
+                    colspan = int(cell.attrs["colspan"])
+                    if index == 0:
+                        colspan -= 1
+                    date_str = ",".join(cell.text.strip().split(",")[-2:])
+                    date = dateparser.parse(date_str, languages=["ru"])
+                    if date is None:
+                        raise ParseError.invalid_date(date_str)
+                    item_values = [build_weather_value(date) for _ in range(colspan)]
+                    result.append(
+                        WeatherResponse(
+                            location=location,
+                            date=date.date(),
+                            period="day",
+                            values=item_values,
+                        )
+                    )
+                    values += item_values
+            case RowType.TIME:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    hour = int(cell.text.strip())
+                    values[index].date = values[index].date.replace(hour=hour)
+            case RowType.CLOUDNESS:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    onmouseover = cell.select_one(".cc_0").next_element.attrs["onmouseover"]
+                    cloudness_str = re.compile("<b>(.*?)</b>").search(onmouseover).group(1).strip()
+                    values[index].sky.cloudness = CLOUDNESS_MAP[cloudness_str]
+            case RowType.PRECIPITATION:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    onmouseover = cell.select_one(".pr_0").attrs["onmouseover"]
+                    precipitation_str = onmouseover.split(",")[1].strip(" '")
+                    precipitation_value_matched = re.compile("(\\d+\\.\\d+) мм").search(precipitation_str)
+                    if precipitation_value_matched:
+                        # one hour value
+                        values[index].precipitation = float(precipitation_value_matched.group(1)) * 6
+                    if "(" in precipitation_str:
+                        precipitation_str = precipitation_str.split(" (")[0]
+                    values[index].sky.precipitation = PRECIPITATION_MAP[precipitation_str]
+            case RowType.TEMPERATURE:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    values[index].temperature = [int(cell.select_one(".t_0").text.lstrip("+").strip())]
+            case RowType.PRESSURE:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    values[index].pressure = [int(cell.select_one(".p_0").text.strip())]
+            case RowType.WIND_SPEED:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    values[index].wind_speed = _parse_wind_speed(cell)
+            case RowType.WIND_GUST:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    values[index].wind_gust = _parse_wind_speed(cell)
+            case RowType.WIND_DIRECTION:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    values[index].wind_direction = parse_wind_direction(cell.text.strip())
+            case RowType.HUMIDITY:
+                for index, cell in enumerate(row.select("td")[1:]):
+                    values[index].humidity = int(cell.text.strip())
    return result