147 lines
6.2 KiB
Python
147 lines
6.2 KiB
Python
import re
|
|
from enum import Enum, auto
|
|
from typing import Iterable
|
|
|
|
import dateparser
|
|
from bs4 import BeautifulSoup, Tag
|
|
|
|
from gallery.sketch.weather.model import (
|
|
Cloudness,
|
|
Precipitation,
|
|
WeatherResponse,
|
|
WeatherValue,
|
|
)
|
|
from gallery.sketch.weather.parser import ParseError, parse_wind_direction
|
|
from gallery.sketch.weather.util import build_weather_value
|
|
|
|
CLOUDNESS_MAP = {
|
|
"Ясно": Cloudness.CLEAR,
|
|
"Малооблачно": Cloudness.PARTLY_CLOUDY,
|
|
"Облачно с прояснениями": Cloudness.PARTLY_CLOUDY,
|
|
"Переменная облачность": Cloudness.PARTLY_CLOUDY,
|
|
"Небольшая облачность": Cloudness.CLOUDY,
|
|
"Облачно": Cloudness.CLOUDY,
|
|
"Значительная облачность": Cloudness.MAINLY_CLOUDY,
|
|
"Пасмурная погода": Cloudness.MAINLY_CLOUDY,
|
|
}
|
|
|
|
PRECIPITATION_MAP = {
|
|
"Явления погоды отсутствуют": Precipitation.NO,
|
|
"Cлабый дождь": Precipitation.SMALL_RAIN,
|
|
}
|
|
|
|
|
|
def _parse_wind_speed(cell: Tag) -> int:
|
|
element = cell.select_one(".wv_0")
|
|
if element:
|
|
return int(element.text.strip() or "0")
|
|
return 0
|
|
|
|
|
|
class RowType(Enum):
|
|
DATE = auto()
|
|
TIME = auto()
|
|
CLOUDNESS = auto()
|
|
PRECIPITATION = auto()
|
|
TEMPERATURE = auto()
|
|
PRESSURE = auto()
|
|
WIND_SPEED = auto()
|
|
WIND_GUST = auto()
|
|
WIND_DIRECTION = auto()
|
|
HUMIDITY = auto()
|
|
|
|
|
|
def _iter_table_rows(table: Tag) -> Iterable[tuple[RowType, Tag]]:
|
|
for row in table.select("tr"):
|
|
title = row.select_one("td").text.strip().lower()
|
|
if "сегодня" in title or "завтра" in title:
|
|
yield RowType.DATE, row
|
|
elif "местное время" in title:
|
|
yield RowType.TIME, row
|
|
elif "облачность" in title:
|
|
yield RowType.CLOUDNESS, row
|
|
elif "осадки" in title:
|
|
yield RowType.PRECIPITATION, row
|
|
elif "температура" in title:
|
|
yield RowType.TEMPERATURE, row
|
|
elif "давление" in title:
|
|
yield RowType.PRESSURE, row
|
|
elif "скорость" in title:
|
|
yield RowType.WIND_SPEED, row
|
|
elif "порывы" in title:
|
|
yield RowType.WIND_GUST, row
|
|
elif "направление" in title:
|
|
yield RowType.WIND_DIRECTION, row
|
|
elif "влажность" in title:
|
|
yield RowType.HUMIDITY, row
|
|
|
|
|
|
# pylint:disable=too-many-branches
|
|
def parse_page(page: BeautifulSoup) -> list[WeatherResponse]:
|
|
location = page.select(".intoLeftNavi > .verticalBottom > a")[-1].text.strip()
|
|
table = page.select_one("#forecastTable")
|
|
if table is None:
|
|
raise ParseError("#forecastTable not found")
|
|
result: list[WeatherResponse] = []
|
|
values: list[WeatherValue] = []
|
|
for row_type, row in _iter_table_rows(table):
|
|
match row_type:
|
|
case RowType.DATE:
|
|
for index, cell in enumerate(row.select("td")):
|
|
colspan = int(cell.attrs["colspan"])
|
|
if index == 0:
|
|
colspan -= 1
|
|
date_str = ",".join(cell.text.strip().split(",")[-2:])
|
|
date = dateparser.parse(date_str, languages=["ru"])
|
|
if date is None:
|
|
raise ParseError.invalid_date(date_str)
|
|
item_values = [build_weather_value(date) for _ in range(colspan)]
|
|
result.append(
|
|
WeatherResponse(
|
|
location=location,
|
|
date=date.date(),
|
|
period="day",
|
|
values=item_values,
|
|
)
|
|
)
|
|
values += item_values
|
|
case RowType.TIME:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
hour = int(cell.text.strip())
|
|
values[index].date = values[index].date.replace(hour=hour)
|
|
case RowType.CLOUDNESS:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
onmouseover = cell.select_one(".cc_0").next_element.attrs["onmouseover"]
|
|
cloudness_str = re.compile("<b>(.*?)</b>").search(onmouseover).group(1).strip()
|
|
values[index].sky.cloudness = CLOUDNESS_MAP[cloudness_str]
|
|
case RowType.PRECIPITATION:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
onmouseover = cell.select_one(".pr_0").attrs["onmouseover"]
|
|
precipitation_str = onmouseover.split(",")[1].strip(" '")
|
|
precipitation_value_matched = re.compile("(\\d+\\.\\d+) мм").search(precipitation_str)
|
|
if precipitation_value_matched:
|
|
# one hour value
|
|
values[index].precipitation = float(precipitation_value_matched.group(1)) * 6
|
|
if "(" in precipitation_str:
|
|
precipitation_str = precipitation_str.split(" (")[0]
|
|
values[index].sky.precipitation = PRECIPITATION_MAP[precipitation_str]
|
|
case RowType.TEMPERATURE:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
values[index].temperature = [int(cell.select_one(".t_0").text.lstrip("+").strip())]
|
|
case RowType.PRESSURE:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
values[index].pressure = [int(cell.select_one(".p_0").text.strip())]
|
|
case RowType.WIND_SPEED:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
values[index].wind.speed = _parse_wind_speed(cell)
|
|
case RowType.WIND_GUST:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
values[index].wind.gust = _parse_wind_speed(cell)
|
|
case RowType.WIND_DIRECTION:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
values[index].wind.direction = parse_wind_direction(cell.text.strip())
|
|
case RowType.HUMIDITY:
|
|
for index, cell in enumerate(row.select("td")[1:]):
|
|
values[index].humidity = int(cell.text.strip())
|
|
return result
|