parent
2a12e14e89
commit
2483fe8d60
|
@ -1,5 +1,4 @@
|
||||||
import configparser
|
import configparser
|
||||||
import dataclasses
|
|
||||||
import json
|
import json
|
||||||
import operator
|
import operator
|
||||||
import os
|
import os
|
||||||
|
@ -14,6 +13,7 @@ import dateutil
|
||||||
import dateutil.parser
|
import dateutil.parser
|
||||||
import exchange_calendars
|
import exchange_calendars
|
||||||
import holidays
|
import holidays
|
||||||
|
import lxml
|
||||||
import pandas
|
import pandas
|
||||||
import pytz
|
import pytz
|
||||||
import requests
|
import requests
|
||||||
|
@ -22,6 +22,9 @@ from dateutil.easter import easter
|
||||||
|
|
||||||
from agenda import thespacedevs
|
from agenda import thespacedevs
|
||||||
|
|
||||||
|
from . import waste_schedule
|
||||||
|
from .types import Event
|
||||||
|
|
||||||
warnings.simplefilter(action="ignore", category=FutureWarning)
|
warnings.simplefilter(action="ignore", category=FutureWarning)
|
||||||
|
|
||||||
|
|
||||||
|
@ -39,7 +42,6 @@ next_us_presidential_election = date(2024, 11, 5)
|
||||||
|
|
||||||
xmas_last_posting_dates = {"first": date(2023, 12, 20), "second": date(2023, 12, 18)}
|
xmas_last_posting_dates = {"first": date(2023, 12, 20), "second": date(2023, 12, 18)}
|
||||||
|
|
||||||
|
|
||||||
config_filename = os.path.join(os.path.dirname(__file__), "..", "config")
|
config_filename = os.path.join(os.path.dirname(__file__), "..", "config")
|
||||||
|
|
||||||
assert os.path.exists(config_filename)
|
assert os.path.exists(config_filename)
|
||||||
|
@ -51,15 +53,6 @@ access_key = config.get("exchangerate", "access_key")
|
||||||
data_dir = config.get("data", "dir")
|
data_dir = config.get("data", "dir")
|
||||||
|
|
||||||
|
|
||||||
@dataclasses.dataclass
|
|
||||||
class Event:
|
|
||||||
"""Event."""
|
|
||||||
|
|
||||||
name: str
|
|
||||||
date: date
|
|
||||||
title: str | None = None
|
|
||||||
|
|
||||||
|
|
||||||
def next_uk_mothers_day(input_date: date) -> date:
|
def next_uk_mothers_day(input_date: date) -> date:
|
||||||
"""Calculate the date of the next UK Mother's Day from the current date."""
|
"""Calculate the date of the next UK Mother's Day from the current date."""
|
||||||
current_year = input_date.year
|
current_year = input_date.year
|
||||||
|
@ -137,8 +130,9 @@ def get_next_bank_holiday(input_date: date) -> list[Event]:
|
||||||
return hols
|
return hols
|
||||||
|
|
||||||
|
|
||||||
def get_gbpusd(now: datetime) -> Decimal:
|
def get_gbpusd() -> Decimal:
|
||||||
"""Get the current value for GBPUSD, with caching."""
|
"""Get the current value for GBPUSD, with caching."""
|
||||||
|
now = datetime.now()
|
||||||
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
||||||
fx_dir = os.path.join(data_dir, "fx")
|
fx_dir = os.path.join(data_dir, "fx")
|
||||||
existing_data = os.listdir(fx_dir)
|
existing_data = os.listdir(fx_dir)
|
||||||
|
@ -326,6 +320,17 @@ def get_birthdays(from_date: date, config: configparser.ConfigParser) -> list[Ev
|
||||||
return events
|
return events
|
||||||
|
|
||||||
|
|
||||||
|
def waste_collection_events() -> list[Event]:
|
||||||
|
"""Waste colllection events."""
|
||||||
|
postcode = "BS48 3HG"
|
||||||
|
uprn = "24071046"
|
||||||
|
|
||||||
|
html = waste_schedule.get_html(data_dir, postcode, uprn)
|
||||||
|
root = lxml.html.fromstring(html)
|
||||||
|
events = waste_schedule.parse(root)
|
||||||
|
return events
|
||||||
|
|
||||||
|
|
||||||
def get_data(now: datetime) -> dict[str, str | object]:
|
def get_data(now: datetime) -> dict[str, str | object]:
|
||||||
"""Get data to display on agenda dashboard."""
|
"""Get data to display on agenda dashboard."""
|
||||||
rocket_dir = os.path.join(data_dir, "thespacedevs")
|
rocket_dir = os.path.join(data_dir, "thespacedevs")
|
||||||
|
@ -333,7 +338,7 @@ def get_data(now: datetime) -> dict[str, str | object]:
|
||||||
|
|
||||||
reply = {
|
reply = {
|
||||||
"now": now,
|
"now": now,
|
||||||
"gbpusd": get_gbpusd(now),
|
"gbpusd": get_gbpusd(),
|
||||||
"next_economist": next_economist(today),
|
"next_economist": next_economist(today),
|
||||||
"bank_holiday": get_next_bank_holiday(today),
|
"bank_holiday": get_next_bank_holiday(today),
|
||||||
"us_holiday": get_us_holidays(today),
|
"us_holiday": get_us_holidays(today),
|
||||||
|
@ -366,6 +371,7 @@ def get_data(now: datetime) -> dict[str, str | object]:
|
||||||
|
|
||||||
events += get_birthdays(today, config)
|
events += get_birthdays(today, config)
|
||||||
events += get_conferences(today, "conferences.yaml")
|
events += get_conferences(today, "conferences.yaml")
|
||||||
|
events += waste_collection_events()
|
||||||
|
|
||||||
next_up_series = Event(
|
next_up_series = Event(
|
||||||
date=date(2026, 6, 1),
|
date=date(2026, 6, 1),
|
||||||
|
|
13
agenda/types.py
Normal file
13
agenda/types.py
Normal file
|
@ -0,0 +1,13 @@
|
||||||
|
"""Types."""
|
||||||
|
|
||||||
|
import dataclasses
|
||||||
|
from datetime import date
|
||||||
|
|
||||||
|
|
||||||
|
@dataclasses.dataclass
|
||||||
|
class Event:
|
||||||
|
"""Event."""
|
||||||
|
|
||||||
|
name: str
|
||||||
|
date: date
|
||||||
|
title: str | None = None
|
70
agenda/waste_schedule.py
Executable file
70
agenda/waste_schedule.py
Executable file
|
@ -0,0 +1,70 @@
|
||||||
|
#!/usr/bin/python3
|
||||||
|
|
||||||
|
import os
|
||||||
|
from datetime import date, datetime, timedelta
|
||||||
|
|
||||||
|
import lxml.html
|
||||||
|
import requests
|
||||||
|
|
||||||
|
from .types import Event
|
||||||
|
|
||||||
|
|
||||||
|
def get_html(data_dir: str, postcode: str, uprn: str) -> str:
|
||||||
|
"""Get waste schedule."""
|
||||||
|
now = datetime.now()
|
||||||
|
waste_dir = os.path.join(data_dir, "waste")
|
||||||
|
if not os.path.exists(waste_dir):
|
||||||
|
os.mkdir(waste_dir)
|
||||||
|
existing_data = os.listdir(waste_dir)
|
||||||
|
existing = [f for f in existing_data if f.endswith(".html")]
|
||||||
|
if existing:
|
||||||
|
recent_filename = max(existing)
|
||||||
|
recent = datetime.strptime(recent_filename, "%Y-%m-%d_%H:%M.html")
|
||||||
|
delta = now - recent
|
||||||
|
|
||||||
|
if existing and delta < timedelta(hours=6):
|
||||||
|
return open(os.path.join(waste_dir, recent_filename)).read()
|
||||||
|
|
||||||
|
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
||||||
|
filename = f"{waste_dir}/{now_str}.html"
|
||||||
|
|
||||||
|
r = requests.post(
|
||||||
|
"https://forms.n-somerset.gov.uk/Waste/CollectionSchedule",
|
||||||
|
data={
|
||||||
|
"PreviousHouse": "",
|
||||||
|
"PreviousPostcode": "-",
|
||||||
|
"Postcode": postcode,
|
||||||
|
"SelectedUprn": uprn,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
html = r.text
|
||||||
|
open(filename, "w").write(html)
|
||||||
|
return html
|
||||||
|
|
||||||
|
|
||||||
|
def parse_waste_schedule_date(day_and_month: str) -> date:
|
||||||
|
"""Parse waste schedule date."""
|
||||||
|
today = date.today()
|
||||||
|
this_year = today.year
|
||||||
|
date_format = "%A %d %B %Y"
|
||||||
|
d = datetime.strptime(f"{day_and_month} {this_year}", date_format).date()
|
||||||
|
if d < today:
|
||||||
|
d = datetime.strptime(f"{day_and_month} {this_year + 1}", date_format).date()
|
||||||
|
return d
|
||||||
|
|
||||||
|
|
||||||
|
def parse(root: lxml.html.HtmlElement) -> list[Event]:
|
||||||
|
"""Parse waste schedule."""
|
||||||
|
events = []
|
||||||
|
tbody = root.find(".//table/tbody")
|
||||||
|
assert tbody is not None
|
||||||
|
for e_service, e_next_date, e_following in tbody:
|
||||||
|
assert e_service.text and e_next_date.text and e_following.text
|
||||||
|
service = e_service.text
|
||||||
|
next_date = parse_waste_schedule_date(e_next_date.text)
|
||||||
|
following_date = parse_waste_schedule_date(e_following.text)
|
||||||
|
|
||||||
|
events.append(Event(name="waste_schedule", date=next_date, title=service))
|
||||||
|
events.append(Event(name="waste_schedule", date=following_date, title=service))
|
||||||
|
|
||||||
|
return events
|
|
@ -23,6 +23,7 @@
|
||||||
"xmas_last_first": "Christmas last posting 1st class",
|
"xmas_last_first": "Christmas last posting 1st class",
|
||||||
"xmas_day": "Christmas day",
|
"xmas_day": "Christmas day",
|
||||||
"next_up_series": "Next Up documentary",
|
"next_up_series": "Next Up documentary",
|
||||||
|
"waste_schedule": "Waste schedule",
|
||||||
}
|
}
|
||||||
%}
|
%}
|
||||||
|
|
||||||
|
@ -35,6 +36,7 @@
|
||||||
<ul>
|
<ul>
|
||||||
<li>Today is {{now.strftime("%A, %-d %b %Y")}}</li>
|
<li>Today is {{now.strftime("%A, %-d %b %Y")}}</li>
|
||||||
<li>GBPUSD: {{"{:,.3f}".format(gbpusd)}}</li>
|
<li>GBPUSD: {{"{:,.3f}".format(gbpusd)}}</li>
|
||||||
|
|
||||||
{# <li>lock down:
|
{# <li>lock down:
|
||||||
{{"{:.1f}".format(lockdown_days)}} days
|
{{"{:.1f}".format(lockdown_days)}} days
|
||||||
({{"{:.2f}".format(lockdown_days / 7)}} weeks) so far</li>
|
({{"{:.2f}".format(lockdown_days / 7)}} weeks) so far</li>
|
||||||
|
|
Loading…
Reference in a new issue