104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
import datetime
|
|
import logging
|
|
|
|
import aiohttp
|
|
from aiocache import cached
|
|
from bs4 import BeautifulSoup
|
|
|
|
from gallery.sketch.source import ApiSource
|
|
from gallery.sketch.weather.api import WeatherApi
|
|
from gallery.sketch.weather.model import (
|
|
Location,
|
|
WeatherResponse,
|
|
)
|
|
from gallery.sketch.weather.util import merge_weather_values
|
|
from gallery.util import TimeUnit
|
|
|
|
from .parser import parse_page
|
|
|
|
logger = logging.getLogger("rp5")
|
|
|
|
HEADERS: dict[str, str] = {
|
|
"Accept": "application/json, text/javascript, */*; q=0.01",
|
|
"Accept-Encoding": "gzip, deflate, br, zstd",
|
|
"Accept-Language": "en-US,en;q=0.9,ru-RU;q=0.8,ru;q=0.7,uk;q=0.6",
|
|
"Connection": "keep-alive",
|
|
"Host": "rp5.ru",
|
|
"Referer": "https://rp5.ru/",
|
|
"Sec-Fetch-Dest": "empty",
|
|
"Sec-Fetch-Mode": "cors",
|
|
"Sec-Fetch-Site": "same-origin",
|
|
"User-Agent": ApiSource.DEFAULT_USER_AGENT,
|
|
"X-Requested-With": "XMLHttpRequest",
|
|
"sec-ch-ua": '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"Linux"',
|
|
}
|
|
|
|
|
|
class RP5WeatherApi(WeatherApi):
|
|
PROVIDER = "rp5"
|
|
|
|
def _build_source(self) -> ApiSource:
|
|
return ApiSource(
|
|
"https://rp5.ru",
|
|
headers=HEADERS,
|
|
)
|
|
|
|
async def find_locations(self, query: str) -> list[Location]:
|
|
endpoint = "docs/search/ru"
|
|
data = aiohttp.FormData({"searchStr": query})
|
|
response = await self.source.request(endpoint, method="POST", data=data)
|
|
soup = BeautifulSoup(response, features="html.parser")
|
|
table = soup.select_one(".searchResults")
|
|
result = []
|
|
for row in table.select("tr")[1:]:
|
|
cells = row.select("td")
|
|
name = cells[0].text.strip()
|
|
location_id = cells[0].select_one("a").attrs["href"].lstrip("/")
|
|
country = cells[1].text.strip()
|
|
district = cells[2].text.strip()
|
|
subdistrict = cells[3].text.strip()
|
|
result.append(
|
|
Location(
|
|
id=location_id,
|
|
name=name,
|
|
provider=self.provider,
|
|
lat=0,
|
|
lon=0,
|
|
country=country,
|
|
country_code="ru", # TODO: contry_code from country
|
|
district=district,
|
|
subdistrict=subdistrict,
|
|
)
|
|
)
|
|
return result
|
|
|
|
@cached(
|
|
key_builder=lambda fun, self, location_id: f"api.weather.{self.provider}.source.{location_id}.forecast",
|
|
alias="redis",
|
|
ttl=TimeUnit.HOUR,
|
|
)
|
|
async def _get_location_forecast(self, location_id: str) -> list[WeatherResponse]:
|
|
endpoint = f"{location_id}"
|
|
content = await self.source.request(endpoint)
|
|
soup = BeautifulSoup(content, features="html.parser")
|
|
return parse_page(soup)
|
|
|
|
async def get_day(self, location_id: str, date: datetime.date) -> WeatherResponse:
|
|
forecast = await self._get_location_forecast(location_id)
|
|
for response in forecast:
|
|
if response.date == date:
|
|
return response
|
|
raise ValueError(f"Unsupported date: {date}")
|
|
|
|
async def get_days(self, location_id: str, days: int) -> WeatherResponse:
|
|
forecast = await self._get_location_forecast(location_id)
|
|
values = [merge_weather_values(item.date, item.values) for item in forecast]
|
|
return WeatherResponse(
|
|
location=forecast[0].location,
|
|
date=datetime.date.today(),
|
|
period="days",
|
|
values=values,
|
|
)
|