From 9ae9d1be1417cda9c107ce64149b0feec08283d1 Mon Sep 17 00:00:00 2001 From: shmyga Date: Thu, 17 Sep 2026 00:47:01 +0300 Subject: [PATCH] refactor: add base parse module --- gallery/painting/gismeteo/api.py | 65 +------ gallery/painting/gismeteo/parser.py | 275 +++++++++++++--------------- gallery/sketch/parse/__init__.py | 0 gallery/sketch/parse/core.py | 13 ++ gallery/sketch/parse/date.py | 12 ++ gallery/sketch/parse/html.py | 36 ++++ gallery/sketch/parse/util.py | 17 ++ gallery/sketch/weather/util.py | 4 +- 8 files changed, 219 insertions(+), 203 deletions(-) create mode 100644 gallery/sketch/parse/__init__.py create mode 100644 gallery/sketch/parse/core.py create mode 100644 gallery/sketch/parse/date.py create mode 100644 gallery/sketch/parse/html.py create mode 100644 gallery/sketch/parse/util.py diff --git a/gallery/painting/gismeteo/api.py b/gallery/painting/gismeteo/api.py index 327fbaf..b052c77 100644 --- a/gallery/painting/gismeteo/api.py +++ b/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.util import merge_weather_values +from gallery.sketch.weather.model import Location, WeatherResponse -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) diff --git a/gallery/painting/gismeteo/parser.py b/gallery/painting/gismeteo/parser.py index edc054c..9d03386 100644 --- a/gallery/painting/gismeteo/parser.py +++ b/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 Tag +from bs4 import BeautifulSoup, 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 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: +class LocationParser(Parser[str, str]): 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]): - KEY = "date" +class DateParser(TableValueParser[datetime.datetime]): + selector = ".widget-row.widget-row-date > .row-item" + model_key = "date" - def parse_row(self, tag: Tag) -> Iterable[datetime.datetime]: - datetime_time_row = tag.select_one(".widget-row.widget-row-datetime-time") - 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 + def parse(self, data: Tag) -> datetime.datetime: + return parse_date(data.text) -class SkyParser(RowParser[Sky]): - KEY = "sky" +class SkyParser(TableValueParser[Sky]): + selector = ".widget-row[data-row=icon-tooltip] > .row-item" + model_key = "sky" CLOUDNESS_MAP: dict[str, Cloudness] = { "ясно": Cloudness.CLEAR, @@ -124,108 +89,132 @@ 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]: - 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 - values, thunder = self._detect_flag(values, self.THUNDER) - values, fog = self._detect_flag(values, self.FOG) - for k, v in self.CLOUDNESS_MAP.items(): - if k in values: - cloudness = v - values.remove(k) - break - for k, v in self.PRECIPITATION_MAP.items(): - if k in values: - precipitation = v - values.remove(k) - break - if values: - logger.warning("unknown sky values: %s:", values) - yield Sky( - cloudness=cloudness, - precipitation=precipitation, - thunder=thunder, - fog=fog, - ) + def parse(self, data: Tag) -> Sky: + sky_str = data.attrs["data-tooltip"] + values = {item.strip().lower() for item in sky_str.split(",")} + cloudness = Cloudness.CLEAR + precipitation = Precipitation.NO + values, thunder = self._detect_flag(values, self.THUNDER) + values, fog = self._detect_flag(values, self.FOG) + for k, v in self.CLOUDNESS_MAP.items(): + if k in values: + cloudness = v + values.remove(k) + break + for k, v in self.PRECIPITATION_MAP.items(): + if k in values: + precipitation = v + values.remove(k) + break + if values: + logger.warning("unknown sky values: %s:", values) + return Sky( + cloudness=cloudness, + precipitation=precipitation, + thunder=thunder, + fog=fog, + ) -class TemperatureParser(RowParser[list[int]]): - KEY = "temperature" +class TemperatureParser(TableValueParser[list[int]]): + selector = ".widget-row-chart[data-row=temperature-air] > .chart > .values > .value" + model_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")] + def parse(self, data: Tag) -> list[int]: + return [int(value.attrs["value"]) for value in data.select("temperature-value")] -class WindSpeedParser(RowParser[int]): - KEY = "wind.speed" +class WindSpeedParser(TableValueParser[int]): + selector = ".widget-row-wind > .row-item > .wind-speed" + model_key = "wind.speed" - def parse_row(self, tag: Tag) -> Iterable[int]: - for item in tag.select(".widget-row-wind > .row-item > .wind-speed > speed-value"): - yield int(item.attrs["value"]) + def parse(self, data: Tag) -> int: + value = data.select_one("speed-value") + return int(value.attrs["value"]) if value else 0 -class WindGustParser(RowParser[int]): - KEY = "wind.gust" +class WindGustParser(TableValueParser[int]): + selector = ".widget-row-wind > .row-item > .wind-gust" + model_key = "wind.gust" - def parse_row(self, tag: Tag) -> Iterable[int]: - for item in tag.select(".widget-row-wind > .row-item > .wind-gust"): - value = item.select_one("speed-value") - yield int(value.attrs["value"]) if value else 0 + def parse(self, data: Tag) -> int: + value = data.select_one("speed-value") + return int(value.attrs["value"]) if value else 0 -class WindDirectionParser(RowParser[WindDirection]): - KEY = "wind.direction" +class WindDirectionParser(TableValueParser[WindDirection | None]): + selector = ".widget-row-wind > .row-item > .wind-speed > .wind-direction" + model_key = "wind.direction" - def parse_row(self, tag: Tag) -> Iterable[float]: - for item in tag.select(".widget-row-wind > .row-item > .wind-speed > .wind-direction"): - wind_direction_str = item.text.lower().strip() - yield parse_wind_direction(wind_direction_str) + def parse(self, data: Tag) -> WindDirection | None: + return parse_wind_direction(data.text.strip()) -class PrecipitationParser(RowParser[float]): - KEY = "precipitation" +class PrecipitationParser(TableValueParser[float]): + selector = ".widget-row[data-row=precipitation-bars] > .row-item" + model_key = "precipitation" - def parse_row(self, tag: Tag) -> Iterable[float]: - for item in tag.select(".widget-row[data-row=precipitation-bars] > .row-item"): - value = item.select_one("precipitation-value") - if value: - yield float(value.attrs["value"]) - else: - yield 0 + def parse(self, data: Tag) -> float: + value = data.select_one("precipitation-value") + return float(value.attrs["value"]) if value else 0 -class PressureParser(RowParser[list[int]]): - KEY = "pressure" +class PressureParser(TableValueParser[list[int]]): + selector = ".widget-row-chart[data-row=pressure] > .chart > .values > .value" + model_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")] + def parse(self, data: Tag) -> list[int]: + return [int(value.attrs["value"]) for value in data.select("pressure-value")] -class HumidityParser(RowParser[int]): - KEY = "humidity" +class HumidityParser(TableValueParser[int]): + 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]: - for item in tag.select( - ".widget-row[data-row=humidity] > .row-item, .widget-row[data-row=humidity-avg] > .row-item" - ): - yield int(item.text) + def parse(self, data: Tag) -> int: + return int(data.text) -ROW_PARSERS: list[RowParser] = [ - DateParser(), - SkyParser(), - TemperatureParser(), - WindSpeedParser(), - WindGustParser(), - WindDirectionParser(), - PrecipitationParser(), - PressureParser(), - HumidityParser(), -] +class GismeteoTableParser(TableParser[WeatherValue]): + selector = ".widget .widget-items" + model_factory = lambda cls: build_weather_value() + value_parsers = [ + TimeParser(), + DateParser(), + SkyParser(), + TemperatureParser(), + WindSpeedParser(), + WindGustParser(), + WindDirectionParser(), + 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, + ) diff --git a/gallery/sketch/parse/__init__.py b/gallery/sketch/parse/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/gallery/sketch/parse/core.py b/gallery/sketch/parse/core.py new file mode 100644 index 0000000..66c9240 --- /dev/null +++ b/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 diff --git a/gallery/sketch/parse/date.py b/gallery/sketch/parse/date.py new file mode 100644 index 0000000..d022463 --- /dev/null +++ b/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 diff --git a/gallery/sketch/parse/html.py b/gallery/sketch/parse/html.py new file mode 100644 index 0000000..fb00f59 --- /dev/null +++ b/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 diff --git a/gallery/sketch/parse/util.py b/gallery/sketch/parse/util.py new file mode 100644 index 0000000..058626a --- /dev/null +++ b/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 diff --git a/gallery/sketch/weather/util.py b/gallery/sketch/weather/util.py index ea1be53..bb79781 100644 --- a/gallery/sketch/weather/util.py +++ b/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,