Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1bffafd6a3 | ||
|
|
46e72feb82 |
@@ -12,12 +12,22 @@ from gallery.sketch.weather.api import WeatherApi
|
||||
from gallery.sketch.weather.model import Location, WeatherResponse, WeatherValue
|
||||
from gallery.sketch.weather.util import merge_weather_values
|
||||
|
||||
from . import datehelp
|
||||
from .parser import DAYS_PARSER, LOCATION_PARSER, ONE_DAY_PARSER, ROW_PARSERS
|
||||
|
||||
logger = logging.getLogger("gismeteo")
|
||||
|
||||
|
||||
def format_date(date: datetime.date) -> str:
|
||||
today = datetime.date.today()
|
||||
days = (date - today).days
|
||||
if days == 0:
|
||||
return "today"
|
||||
elif days == 1:
|
||||
return "tomorrow"
|
||||
else:
|
||||
return f"{days + 1}-day"
|
||||
|
||||
|
||||
class GismeteoApi(WeatherApi):
|
||||
PROVIDER = "gismeteo"
|
||||
|
||||
@@ -107,7 +117,7 @@ class GismeteoApi(WeatherApi):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail={"max_date": max_date.strftime("%Y-%m-%d")}
|
||||
)
|
||||
data = await self.source.request(f"weather-{location_id}/{datehelp.dump(date)}")
|
||||
data = await self.source.request(f"weather-{location_id}/{format_date(date)}")
|
||||
return self._parse_oneday(date, data)
|
||||
|
||||
async def get_days(self, location_id: str, days: int) -> WeatherResponse:
|
||||
|
||||
@@ -1,31 +0,0 @@
|
||||
from typing import Generic, Iterable, Optional, TypeVar
|
||||
|
||||
from bs4 import Tag
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class WidgetParser:
|
||||
def parse_widget(self, tag: Tag) -> Tag:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class BaseWidgetParser(WidgetParser):
|
||||
SELECT: str
|
||||
|
||||
def __init__(self, select: Optional[str] = None):
|
||||
super().__init__()
|
||||
self._select = select or self.SELECT
|
||||
|
||||
def parse_widget(self, tag: Tag) -> Tag:
|
||||
widget = tag.select_one(self._select)
|
||||
if widget is None:
|
||||
raise ValueError(self._select)
|
||||
return widget
|
||||
|
||||
|
||||
class RowParser(Generic[T]):
|
||||
KEY: str
|
||||
|
||||
def parse_row(self, tag: Tag) -> Iterable[T]:
|
||||
raise NotImplementedError
|
||||
@@ -1,39 +0,0 @@
|
||||
import datetime
|
||||
from enum import Enum
|
||||
|
||||
import dateparser
|
||||
import dateparser.date_parser
|
||||
|
||||
|
||||
class Day(str, Enum):
|
||||
TODAY = "today"
|
||||
TOMORROW = "tomorrow"
|
||||
DAY_3 = "3-day"
|
||||
DAY_4 = "4-day"
|
||||
DAY_5 = "5-day"
|
||||
|
||||
|
||||
def parse(value: str) -> datetime.date:
|
||||
if value in ["today", "mock"]:
|
||||
return datetime.date.today()
|
||||
elif value == "tomorrow":
|
||||
return datetime.date.today() + datetime.timedelta(days=1)
|
||||
elif value.endswith("-day"):
|
||||
days = int(value.split("-")[0]) - 1
|
||||
return datetime.date.today() + datetime.timedelta(days=days)
|
||||
else:
|
||||
date = dateparser.parse(value)
|
||||
if date is None:
|
||||
raise ValueError(value)
|
||||
return date.date()
|
||||
|
||||
|
||||
def dump(date: datetime.date) -> str:
|
||||
today = datetime.date.today()
|
||||
days = (date - today).days
|
||||
if days == 0:
|
||||
return "today"
|
||||
elif days == 1:
|
||||
return "tomorrow"
|
||||
else:
|
||||
return f"{days + 1}-day"
|
||||
@@ -1,7 +1,7 @@
|
||||
import datetime
|
||||
import logging
|
||||
import re
|
||||
from typing import Iterable
|
||||
from typing import Generic, Iterable, Optional, TypeVar
|
||||
|
||||
import dateparser
|
||||
from bs4 import Tag
|
||||
@@ -14,10 +14,37 @@ from gallery.sketch.weather.model import (
|
||||
)
|
||||
from gallery.sketch.weather.parser import ParseError, parse_wind_direction
|
||||
|
||||
from .core import BaseWidgetParser, RowParser
|
||||
|
||||
logger = logging.getLogger("gismeteo")
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class WidgetParser:
|
||||
def parse_widget(self, tag: Tag) -> Tag:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class BaseWidgetParser(WidgetParser):
|
||||
SELECT: str
|
||||
|
||||
def __init__(self, select: Optional[str] = None):
|
||||
super().__init__()
|
||||
self._select = select or self.SELECT
|
||||
|
||||
def parse_widget(self, tag: Tag) -> Tag:
|
||||
widget = tag.select_one(self._select)
|
||||
if widget is None:
|
||||
raise ValueError(self._select)
|
||||
return widget
|
||||
|
||||
|
||||
class RowParser(Generic[T]):
|
||||
KEY: str
|
||||
|
||||
def parse_row(self, tag: Tag) -> Iterable[T]:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
ONE_DAY_PARSER = BaseWidgetParser(".widget.widget-oneday .widget-items")
|
||||
DAYS_PARSER = BaseWidgetParser(".widget.widget-days .widget-items")
|
||||
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import re
|
||||
from enum import Enum, auto
|
||||
from typing import Iterable
|
||||
|
||||
import dateparser
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
@@ -36,28 +38,56 @@ def _parse_wind_speed(cell: Tag) -> int:
|
||||
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:
|
||||
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")
|
||||
(
|
||||
date_row,
|
||||
time_row,
|
||||
cloudness_row,
|
||||
precipitation_row,
|
||||
temperature_row,
|
||||
pressure_row,
|
||||
wind_speed_row,
|
||||
wind_gust_row,
|
||||
wind_direction_row,
|
||||
humidity_row,
|
||||
*_,
|
||||
) = table.select("tr")
|
||||
result: list[WeatherResponse] = []
|
||||
values: list[WeatherValue] = []
|
||||
for index, cell in enumerate(date_row.select("td")):
|
||||
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
|
||||
@@ -75,14 +105,17 @@ def parse_page(page: BeautifulSoup) -> list[WeatherResponse]:
|
||||
)
|
||||
)
|
||||
values += item_values
|
||||
for index, cell in enumerate(time_row.select("td")[1:]):
|
||||
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)
|
||||
for index, cell in enumerate(cloudness_row.select("td")[1:]):
|
||||
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]
|
||||
for index, cell in enumerate(precipitation_row.select("td")[1:]):
|
||||
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)
|
||||
@@ -92,16 +125,22 @@ def parse_page(page: BeautifulSoup) -> list[WeatherResponse]:
|
||||
if "(" in precipitation_str:
|
||||
precipitation_str = precipitation_str.split(" (")[0]
|
||||
values[index].sky.precipitation = PRECIPITATION_MAP[precipitation_str]
|
||||
for index, cell in enumerate(temperature_row.select("td")[1:]):
|
||||
case RowType.TEMPERATURE:
|
||||
for index, cell in enumerate(row.select("td")[1:]):
|
||||
values[index].temperature = [int(cell.select_one(".t_0").text.lstrip("+").strip())]
|
||||
for index, cell in enumerate(pressure_row.select("td")[1:]):
|
||||
case RowType.PRESSURE:
|
||||
for index, cell in enumerate(row.select("td")[1:]):
|
||||
values[index].pressure = [int(cell.select_one(".p_0").text.strip())]
|
||||
for index, cell in enumerate(wind_speed_row.select("td")[1:]):
|
||||
case RowType.WIND_SPEED:
|
||||
for index, cell in enumerate(row.select("td")[1:]):
|
||||
values[index].wind_speed = _parse_wind_speed(cell)
|
||||
for index, cell in enumerate(wind_gust_row.select("td")[1:]):
|
||||
case RowType.WIND_GUST:
|
||||
for index, cell in enumerate(row.select("td")[1:]):
|
||||
values[index].wind_gust = _parse_wind_speed(cell)
|
||||
for index, cell in enumerate(wind_direction_row.select("td")[1:]):
|
||||
case RowType.WIND_DIRECTION:
|
||||
for index, cell in enumerate(row.select("td")[1:]):
|
||||
values[index].wind_direction = parse_wind_direction(cell.text.strip())
|
||||
for index, cell in enumerate(humidity_row.select("td")[1:]):
|
||||
case RowType.HUMIDITY:
|
||||
for index, cell in enumerate(row.select("td")[1:]):
|
||||
values[index].humidity = int(cell.text.strip())
|
||||
return result
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
name = "gallery"
|
||||
description = ""
|
||||
version = "0.8.0"
|
||||
version = "0.8.1"
|
||||
readme = "README.md"
|
||||
authors = [{ name = "shmyga", email = "shmyga.z@gmail.com" }]
|
||||
requires-python = '>=3.14,<4.0'
|
||||
|
||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "gallery",
|
||||
"version": "0.8.0",
|
||||
"version": "0.8.1",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "gallery",
|
||||
"version": "0.8.0",
|
||||
"version": "0.8.1",
|
||||
"license": "ISC",
|
||||
"dependencies": {
|
||||
"@popperjs/core": "^2.11.8",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "gallery",
|
||||
"version": "0.8.0",
|
||||
"version": "0.8.1",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"build": "webpack --mode production",
|
||||
|
||||
+79
-75
File diff suppressed because one or more lines are too long
+2
-2
@@ -18,10 +18,10 @@ async def test_search(rp5_api: RP5WeatherApi):
|
||||
|
||||
|
||||
async def test_day(rp5_api: RP5WeatherApi):
|
||||
result = await rp5_api.get_day("test", datetime.date(2026, 9, 2))
|
||||
result = await rp5_api.get_day("test", datetime.date(2026, 9, 11))
|
||||
assert len(result.values) == 4
|
||||
|
||||
|
||||
async def test_days(rp5_api: RP5WeatherApi):
|
||||
result = await rp5_api.get_days("test", 10)
|
||||
assert len(result.values) == 7
|
||||
assert len(result.values) == 8
|
||||
|
||||
Reference in New Issue
Block a user