48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import datetime
|
|
from typing import Any, Dict, List, NamedTuple
|
|
|
|
import aiohttp
|
|
from bs4 import BeautifulSoup
|
|
from . import dateutil
|
|
|
|
from .location import LOCATION_BUNDLE
|
|
from .parser import ROW_PARSERS, ONE_DAY_PARSER
|
|
|
|
|
|
class WeatherValue(NamedTuple):
|
|
date: datetime.datetime
|
|
cloudness: str
|
|
temperature: int
|
|
wind_speed: int
|
|
wind_gust: int
|
|
wind_direction: str
|
|
precipitation: float
|
|
pressure: int
|
|
humidity: int
|
|
|
|
|
|
class GismeteoApi:
|
|
BASE_URL = "https://www.gismeteo.ru"
|
|
|
|
async def _request(self, endpoint: str) -> str:
|
|
url = f"{self.BASE_URL}/{endpoint}"
|
|
async with aiohttp.ClientSession(raise_for_status=True) as session:
|
|
async with session.request("GET", url) as response:
|
|
return await response.text()
|
|
|
|
def _parse_oneday(self, data: str) -> List[WeatherValue]:
|
|
result: List[Dict[str, Any]] = []
|
|
soup = BeautifulSoup(data, features="html.parser")
|
|
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({})
|
|
result[index][parser.KEY] = value
|
|
return [WeatherValue(**item) for item in result]
|
|
|
|
async def get_day(self, location_id: str, date: datetime.date) -> List[WeatherValue]:
|
|
location = LOCATION_BUNDLE.parse(location_id)
|
|
data = await self._request(f"weather-{location}/{dateutil.dump(date)}")
|
|
return self._parse_oneday(data)
|