diff --git a/gallery/painting/yandexweather/api.py b/gallery/painting/yandexweather/api.py index c9e36af..08c1e6a 100644 --- a/gallery/painting/yandexweather/api.py +++ b/gallery/painting/yandexweather/api.py @@ -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) diff --git a/gallery/painting/yandexweather/parser.py b/gallery/painting/yandexweather/parser.py index 300d4e1..53090ba 100644 --- a/gallery/painting/yandexweather/parser.py +++ b/gallery/painting/yandexweather/parser.py @@ -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 + ] diff --git a/gallery/sketch/parse/html.py b/gallery/sketch/parse/html.py index 9821a49..8342b05 100644 --- a/gallery/sketch/parse/html.py +++ b/gallery/sketch/parse/html.py @@ -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)