gallery/gallery/painting/gismeteo/parser.py

import datetime
import logging
import re
from typing import Iterable

import dateparser
from bs4 import Tag

from gallery.sketch.weather.model import (
    Cloudness,
    Precipitation,
    Sky,
    WindDirection,
)
from gallery.sketch.weather.parser import ParseError, parse_wind_direction

from .core import BaseWidgetParser, RowParser

logger = logging.getLogger("gismeteo")

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:
        match = self.PATTERN.search(data)
        if match:
            return match.group(1)
        return None


LOCATION_PARSER = LocationParser()


class DateParser(RowParser[datetime.datetime]):
    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


class SkyParser(RowParser[Sky]):
    KEY = "sky"

    CLOUDNESS_MAP: dict[str, Cloudness] = {
        "ясно": Cloudness.CLEAR,
        "безоблачно": Cloudness.CLEAR,
        "малооблачно": Cloudness.PARTLY_CLOUDY,
        "облачно": Cloudness.CLOUDY,
        "пасмурно": Cloudness.MAINLY_CLOUDY,
    }

    PRECIPITATION_MAP: dict[str, Precipitation] = {
        "без осадков": Precipitation.NO,
        "без существенных осадков": Precipitation.NO,
        "небольшой дождь": Precipitation.SMALL_RAIN,
        "сильный дождь": Precipitation.HEAVY_RAIN,
        "очень сильный дождь": Precipitation.HEAVY_RAIN,
        "ливневый дождь": Precipitation.SHOWER,
        "дождь": Precipitation.RAIN,
        "ливень": Precipitation.SHOWER,
        "снег": Precipitation.SNOW,
        "небольшой снег": Precipitation.SNOW,
        "сильный снег": Precipitation.HEAVY_SNOW,
        "мокрый снег": Precipitation.SNOW,
        "снег с дождём": Precipitation.SNOW,
        "сильный снег с дождём": Precipitation.HEAVY_SNOW,
        "небольшой снег с дождём": Precipitation.SNOW,
        "небольшой мокрый снег": Precipitation.SNOW,
        "град": Precipitation.HAIL,
    }

    THUNDER = {"гроза"}
    FOG = {"дымка", "туман"}

    @classmethod
    def _detect_flag(cls, values: set[str], flag_values: set[str]) -> tuple[set[str], bool]:
        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,
            )


class TemperatureParser(RowParser[list[int]]):
    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")]


class WindSpeedParser(RowParser[int]):
    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"])


class WindGustParser(RowParser[int]):
    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


class WindDirectionParser(RowParser[WindDirection]):
    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)


class PrecipitationParser(RowParser[float]):
    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


class PressureParser(RowParser[list[int]]):
    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")]


class HumidityParser(RowParser[int]):
    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)


ROW_PARSERS: list[RowParser] = [
    DateParser(),
    SkyParser(),
    TemperatureParser(),
    WindSpeedParser(),
    WindGustParser(),
    WindDirectionParser(),
    PrecipitationParser(),
    PressureParser(),
    HumidityParser(),
]

ROW_PARSERS_MAP: dict[str, RowParser] = {parser.KEY: parser for parser in ROW_PARSERS}