Compare commits

...
4 Commits
Author SHA1 Message Date
shmyga 41914d7f74 ci(version): 0.8.2 2026-09-12 15:27:15 +03:00
shmyga de466de805 fix(view): fix build_url function 2026-09-12 15:26:35 +03:00
shmyga 1bffafd6a3 ci(version): 0.8.1 2026-09-10 18:48:42 +03:00
shmyga 46e72feb82 fix(rp5): fix rp5 parser 2026-09-10 18:48:23 +03:00
11 changed files with 227 additions and 217 deletions
@@ -45,7 +45,7 @@ def build_url(
if query:
query_dict = {**dict(parse_qsl(parts.query)), **query}
parts = parts._replace(query=urlencode(query_dict))
return parts.geturl()
return "/".join([""] + parts.geturl().split("/")[3:])
def build_templates(
+12 -2
View File
@@ -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:
-31
View File
@@ -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
-39
View File
@@ -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"
+30 -3
View File
@@ -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")
+99 -60
View File
@@ -1,4 +1,6 @@
import re
from enum import Enum, auto
from typing import Iterable
import dateparser
from bs4 import BeautifulSoup, Tag
@@ -36,72 +38,109 @@ 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")):
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
for index, cell in enumerate(time_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:]):
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:]):
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]
for index, cell in enumerate(temperature_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:]):
values[index].pressure = [int(cell.select_one(".p_0").text.strip())]
for index, cell in enumerate(wind_speed_row.select("td")[1:]):
values[index].wind_speed = _parse_wind_speed(cell)
for index, cell in enumerate(wind_gust_row.select("td")[1:]):
values[index].wind_gust = _parse_wind_speed(cell)
for index, cell in enumerate(wind_direction_row.select("td")[1:]):
values[index].wind_direction = parse_wind_direction(cell.text.strip())
for index, cell in enumerate(humidity_row.select("td")[1:]):
values[index].humidity = int(cell.text.strip())
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
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "gallery"
description = ""
version = "0.8.0"
version = "0.8.2"
readme = "README.md"
authors = [{ name = "shmyga", email = "shmyga.z@gmail.com" }]
requires-python = '>=3.14,<4.0'
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "gallery",
"version": "0.8.0",
"version": "0.8.2",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "gallery",
"version": "0.8.0",
"version": "0.8.2",
"license": "ISC",
"dependencies": {
"@popperjs/core": "^2.11.8",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "gallery",
"version": "0.8.0",
"version": "0.8.2",
"type": "module",
"scripts": {
"build": "webpack --mode production",
File diff suppressed because one or more lines are too long
+2 -2
View File
@@ -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