refactor(yandexweather): rework yandexweather parser

2026-09-17 20:59:29 +03:00
parent ece1a58a27
commit 2e9d04c10e
3 changed files with 176 additions and 102 deletions
@@ -1,9 +1,7 @@
 import datetime
-import json
 import logging

 from aiocache import cached
-from bs4 import BeautifulSoup

 from gallery.sketch.source import ApiSource
 from gallery.sketch.weather.api import WeatherApi
@@ -14,7 +12,7 @@ from gallery.sketch.weather.model import (
 from gallery.sketch.weather.util import merge_weather_values
 from gallery.util import TimeUnit

-from .parser import parse_article
+from .parser import YandexForecastParser, YandexLocationParser

 logger = logging.getLogger("yandexweather")

@@ -22,6 +20,9 @@ logger = logging.getLogger("yandexweather")
 class YandexWeatherApi(WeatherApi):
    PROVIDER = "yandex"

+    forecast_parser = YandexForecastParser()
+    location_parser = YandexLocationParser()
+
    def _build_source(self) -> ApiSource:
        return ApiSource("https://yandex.ru")

@@ -33,29 +34,12 @@ class YandexWeatherApi(WeatherApi):
    async def _get_location_forecast(self, location_id: str) -> list[WeatherResponse]:
        endpoint = f"pogoda/ru/{location_id}/details"
        content = await self.source.request(endpoint, mode="browser")
-        soup = BeautifulSoup(content, features="html.parser")
-        location = soup.select_one("[class^=Suggest_placeholder]").text.strip()
-        articles = soup.select("article[class^=AppForecastDay]")
-        return [parse_article(article, location) for article in articles]
+        return self.forecast_parser.parse(content)

    async def find_locations(self, query: str) -> list[Location]:
        endpoint = f"weather/api/suggest?part={query}&lang=ru&type=weather"
        response = await self.source.request(endpoint)
-        data = json.loads(response)
-        return [
-            Location(
-                id=str(item["geoId"]),
-                name=item["title"],
-                provider=self.provider,
-                lat=item["coords"]["lat"],
-                lon=item["coords"]["lon"],
-                country="ru",
-                country_code="ru",
-                district=item["description"],
-                subdistrict="",
-            )
-            for item in data
-        ]
+        return self.location_parser.parse(response)

    async def get_day(self, location_id: str, date: datetime.date) -> WeatherResponse:
        forecast = await self._get_location_forecast(location_id)
@@ -1,91 +1,181 @@
+import datetime
+import json
 import logging
-from enum import Enum

-import dateparser
-from bs4 import Tag
+from bs4 import BeautifulSoup, Tag

-from gallery.sketch.weather.model import Cloudness, Precipitation, Sky, WeatherResponse
-from gallery.sketch.weather.parser import ParseError, parse_wind_direction
-from gallery.sketch.weather.util import build_weather_value
+from gallery.sketch.parse.core import Parser
+from gallery.sketch.parse.date import parse_date
+from gallery.sketch.parse.html import TableParser, TableValueParser, TagParser
+from gallery.sketch.weather.model import (
+    Cloudness,
+    Location,
+    Precipitation,
+    Sky,
+    WeatherResponse,
+    WeatherValue,
+    WindDirection,
+)
+from gallery.sketch.weather.parser import parse_wind_direction

 logger = logging.getLogger("yandexweather")


-class DayPart(Enum):
-    MORNING = ("Утром", 6)
-    AFTERNOON = ("Днём", 12)
-    EVENING = ("Вечером", 18)
-    NIGHT = ("Ночью", 0)
+class LocationNameParser(TagParser[str]):
+    selector = "[class^=Suggest_placeholder]"

-    def __init__(self, title: str, hour: int):
-        super().__init__()
-        self.title = title
-        self.hour = hour
+    def extract(self, data: Tag) -> str:
+        return data.text.strip()


-def parse_sky(sky: str) -> Sky:
-    cloudness = Cloudness.CLEAR
-    precipitation = Precipitation.NO
-    thunder = False
-    fog = False
-    if sky == "ясно":
-        pass
-    elif sky == "облачно с прояснениями":
-        cloudness = Cloudness.PARTLY_CLOUDY
-    elif sky == "малооблачно":
-        cloudness = Cloudness.CLOUDY
-    elif sky == "пасмурно":
-        cloudness = Cloudness.MAINLY_CLOUDY
-    elif sky == "небольшой дождь":
-        precipitation = Precipitation.SMALL_RAIN
-    elif sky == "дождь":
-        precipitation = Precipitation.RAIN
-    elif sky == "ливень":
-        precipitation = Precipitation.SHOWER
-    else:
-        logger.warning("unknown sky value: %s", sky)
-    return Sky(
-        cloudness=cloudness,
-        precipitation=precipitation,
-        thunder=thunder,
-        fog=fog,
-    )
+class DateParser(TagParser[datetime.date]):
+    selector = "[class^=AppForecastDayHeader_dayTitle]"
+
+    def extract(self, data: Tag) -> datetime.date:
+        return parse_date(data.text.split(",")[-1].strip())


-def parse_article(article: Tag, location: str) -> WeatherResponse:
-    date_str = article.select_one("[class^=AppForecastDayHeader_dayTitle]").text.split(",")[-1].strip()
-    date = dateparser.parse(date_str, languages=["ru"])
-    if date is None:
-        raise ParseError.invalid_date(date_str)
-    values = []
-    for day_part in DayPart:
-        value = build_weather_value(date.replace(hour=day_part.hour))
-        day_part_tag = article.find(string=day_part.title)
-        temperature_tag = day_part_tag.next_element
-        temperature = int(temperature_tag.text.strip().rstrip("°"))
-        value.temperature.append(temperature)
-        sky_tag = temperature_tag.next_element.next_element.next_element
-        sky_str = sky_tag.text.strip()
-        value.sky = parse_sky(sky_str)
-        temperature_tag = sky_tag.next_element.next_element
-        temperature = int(temperature_tag.text.strip().rstrip("°"))
-        value.temperature.append(temperature)
-        wind_speed_tag = temperature_tag.next_element.next_element
-        wind_speed = int(wind_speed_tag.text.strip())
-        value.wind.speed = wind_speed
-        wind_direction_tag = wind_speed_tag.next_element.next_element.next_element.next_element
-        wind_direction_str = wind_direction_tag.text.strip()
-        value.wind.direction = parse_wind_direction(wind_direction_str)
-        humidity_tag = wind_direction_tag.next_element.next_element.next_element.next_element
-        humidity = int(humidity_tag.text.strip().rstrip("%"))
-        value.humidity = humidity
-        pressure_tag = humidity_tag.next_element.next_element
-        pressure = int(pressure_tag.text.strip())
-        value.pressure.append(pressure)
-        values.append(value)
-    return WeatherResponse(
-        location=location,
-        date=date,
-        period="day",
-        values=values,
-    )
+class TimeParser(TableValueParser[datetime.datetime]):
+    selector = "[style$='-part']"
+    model_key = "date"
+
+    HOUR_MAP = {
+        "Утром": 6,
+        "Днём": 12,
+        "Вечером": 18,
+        "Ночью": 0,
+    }
+
+    def extract(self, data: Tag) -> datetime.datetime:
+        hour = self.HOUR_MAP[data.text.strip()]
+        return datetime.datetime.min.replace(hour=hour)
+
+
+class TemperatureParser(TableValueParser[list[int]]):
+    selector = "[style$='-temp']"
+    model_key = "temperature"
+
+    def extract(self, data: Tag) -> list[int]:
+        return [int(data.text.strip().rstrip("°"))]
+
+
+class SkyParser(TableValueParser[Sky]):
+    selector = "[style$='-text']"
+    model_key = "sky"
+
+    def extract(self, data: Tag) -> Sky:
+        sky = data.text.strip()
+        cloudness = Cloudness.CLEAR
+        precipitation = Precipitation.NO
+        thunder = False
+        fog = False
+        if sky == "ясно":
+            pass
+        elif sky == "облачно с прояснениями":
+            cloudness = Cloudness.PARTLY_CLOUDY
+        elif sky == "малооблачно":
+            cloudness = Cloudness.CLOUDY
+        elif sky == "пасмурно":
+            cloudness = Cloudness.MAINLY_CLOUDY
+        elif sky == "небольшой дождь":
+            precipitation = Precipitation.SMALL_RAIN
+        elif sky == "дождь":
+            precipitation = Precipitation.RAIN
+        elif sky == "ливень":
+            precipitation = Precipitation.SHOWER
+        else:
+            logger.warning("unknown sky value: %s", sky)
+        return Sky(
+            cloudness=cloudness,
+            precipitation=precipitation,
+            thunder=thunder,
+            fog=fog,
+        )
+
+
+class WindSpeedParser(TableValueParser[int]):
+    selector = "[style$='-wind']:not([class*=AppForecastDayPart_wind_formatted])"
+    model_key = "wind.speed"
+
+    def extract(self, data: Tag) -> int:
+        return int(data.text.strip())
+
+
+class WindDirectionParser(TableValueParser[WindDirection | None]):
+    selector = "[style$='-dir']"
+    model_key = "wind.direction"
+
+    def extract(self, data: Tag) -> WindDirection | None:
+        return parse_wind_direction(data.text.strip())
+
+
+class HumidityParser(TableValueParser[int]):
+    selector = "[style$='-hum']"
+    model_key = "humidity"
+
+    def extract(self, data: Tag) -> int:
+        return int(data.text.strip().rstrip("%"))
+
+
+class PressureParser(TableValueParser[list[int]]):
+    selector = "[style$='-press']"
+    model_key = "pressure"
+
+    def extract(self, data: Tag) -> list[int]:
+        return [int(data.text.strip().rstrip("%"))]
+
+
+class YandexTableParser(TableParser[WeatherValue]):
+    selector = ""
+    value_parsers = [
+        TimeParser(),
+        TemperatureParser(),
+        SkyParser(),
+        WindSpeedParser(),
+        WindDirectionParser(),
+        HumidityParser(),
+        PressureParser(),
+    ]
+
+    def create_model(self, data: dict) -> WeatherValue:
+        return WeatherValue.model_validate(data)
+
+
+class YandexForecastParser(Parser[str, list[WeatherResponse]]):
+    table_parser = YandexTableParser()
+    date_parser = DateParser()
+    location_parser = LocationNameParser()
+
+    def parse(self, data: str) -> list[WeatherResponse]:
+        soup = BeautifulSoup(data, features="html.parser")
+        location = self.location_parser.parse(soup)
+        articles = soup.select("article[class^=AppForecastDay]")
+        return [
+            WeatherResponse(
+                location=location,
+                date=self.date_parser.parse(item),
+                period="day",
+                values=self.table_parser.parse(item),
+            )
+            for item in articles
+        ]
+
+
+class YandexLocationParser(Parser[str, list[Location]]):
+
+    def parse(self, data: str) -> list[Location]:
+        content = json.loads(data)
+        return [
+            Location(
+                id=str(item["geoId"]),
+                name=item["title"],
+                provider="yandex",  # TODO: provider module const
+                lat=item["coords"]["lat"],
+                lon=item["coords"]["lon"],
+                country="ru",
+                country_code="ru",
+                district=item["description"],
+                subdistrict="",
+            )
+            for item in content
+        ]
@@ -13,7 +13,7 @@ class TagParser(Parser[Tag, R], Generic[R]):
        raise NotImplementedError

    def parse(self, data: Tag) -> R:
-        tag = data.select_one(self.selector)
+        tag = data.select_one(self.selector) if self.selector else data
        if tag is None:
            raise ParseError("Tag not found", self.selector)
        return self.extract(tag)