184 lines
5.1 KiB
Python
184 lines
5.1 KiB
Python
import datetime
|
|
import json
|
|
import logging
|
|
|
|
from bs4 import BeautifulSoup, Tag
|
|
|
|
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
|
|
|
|
from . import PROVIDER
|
|
|
|
logger = logging.getLogger(PROVIDER)
|
|
|
|
|
|
class LocationNameParser(TagParser[str]):
|
|
selector = "[class^=Suggest_placeholder]"
|
|
|
|
def extract(self, data: Tag) -> str:
|
|
return data.text.strip()
|
|
|
|
|
|
class DateParser(TagParser[datetime.date]):
|
|
selector = "[class^=AppForecastDayHeader_dayTitle]"
|
|
|
|
def extract(self, data: Tag) -> datetime.date:
|
|
return parse_date(data.text.split(",")[-1].strip())
|
|
|
|
|
|
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=PROVIDER,
|
|
lat=item["coords"]["lat"],
|
|
lon=item["coords"]["lon"],
|
|
country="ru",
|
|
country_code="ru",
|
|
district=item["description"],
|
|
subdistrict="",
|
|
)
|
|
for item in content
|
|
]
|