gallery/gallery/painting/gismeteo/parser.py

import datetime
import logging
import re
from typing import Literal, NamedTuple

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 parse_wind_direction
from gallery.sketch.weather.util import merge_weather_values

logger = logging.getLogger("gismeteo")


class LocationNameParser(Parser[str, str]):
    PATTERN = re.compile('{"ru":{"city":{"name":"(.*?)"')

    def parse(self, data: str) -> str:
        match = self.PATTERN.search(data)
        if match:
            return match.group(1)
        raise ParseError("Location not found")


class TimeParser(TableValueParser[datetime.datetime]):
    selector = ".widget-row.widget-row-datetime-time > .row-item > time-value"
    model_key = "date"

    def extract(self, data: Tag) -> datetime.datetime:
        timestamp = int(data.attrs["timestamp"])
        return datetime.datetime.fromtimestamp(timestamp)


class DateParser(TableValueParser[datetime.datetime]):
    selector = ".widget-row.widget-row-date > .row-item"
    model_key = "date"

    def extract(self, data: Tag) -> datetime.datetime:
        return parse_date(data.text)


class SkyParser(TableValueParser[Sky]):
    selector = ".widget-row[data-row=icon-tooltip] > .row-item"
    model_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 extract(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(TableValueParser[list[int]]):
    selector = ".widget-row-chart[data-row=temperature-air] > .chart > .values > .value"
    model_key = "temperature"

    def extract(self, data: Tag) -> list[int]:
        return [int(value.attrs["value"]) for value in data.select("temperature-value")]


class WindSpeedParser(TableValueParser[int]):
    selector = ".widget-row-wind > .row-item > .wind-speed"
    model_key = "wind.speed"

    def extract(self, data: Tag) -> int:
        value = data.select_one("speed-value")
        return int(value.attrs["value"]) if value else 0


class WindGustParser(TableValueParser[int]):
    selector = ".widget-row-wind > .row-item > .wind-gust"
    model_key = "wind.gust"

    def extract(self, data: Tag) -> int:
        value = data.select_one("speed-value")
        return int(value.attrs["value"]) if value else 0


class WindDirectionParser(TableValueParser[WindDirection | None]):
    selector = ".widget-row-wind > .row-item > .wind-speed > .wind-direction"
    model_key = "wind.direction"

    def extract(self, data: Tag) -> WindDirection | None:
        return parse_wind_direction(data.text.strip())


class PrecipitationParser(TableValueParser[float]):
    selector = ".widget-row[data-row=precipitation-bars] > .row-item"
    model_key = "precipitation"

    def extract(self, data: Tag) -> float:
        value = data.select_one("precipitation-value")
        return float(value.attrs["value"]) if value else 0


class PressureParser(TableValueParser[list[int]]):
    selector = ".widget-row-chart[data-row=pressure] > .chart > .values > .value"
    model_key = "pressure"

    def extract(self, data: Tag) -> list[int]:
        return [int(value.attrs["value"]) for value in data.select("pressure-value")]


class HumidityParser(TableValueParser[int]):
    selector = ".widget-row[data-row=humidity] > .row-item, .widget-row[data-row=humidity-avg] > .row-item"
    model_key = "humidity"

    def extract(self, data: Tag) -> int:
        return int(data.text)


class GismeteoTableParser(TableParser[WeatherValue]):
    selector = ".widget .widget-items"
    value_parsers = [
        TimeParser(),
        DateParser(),
        SkyParser(),
        TemperatureParser(),
        WindSpeedParser(),
        WindGustParser(),
        WindDirectionParser(),
        PrecipitationParser(),
        PressureParser(),
        HumidityParser(),
    ]

    def create_model(self, data: dict) -> WeatherValue:
        return WeatherValue.model_validate(data)


class Context(NamedTuple):
    date: datetime.date
    period: Literal["day", "days"]
    data: str
    values: int | None = None


class GismeteoParser(Parser[Context, WeatherResponse]):
    location_parser = LocationNameParser()
    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,
        )