From 2483fe8d60e07f47ea8ee2f496db6dacca68fc72 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Sun, 8 Oct 2023 19:53:44 +0100 Subject: [PATCH] Include waste collection schedule Closes: #11 --- agenda/__init__.py | 32 ++++++++++-------- agenda/types.py | 13 ++++++++ agenda/waste_schedule.py | 70 ++++++++++++++++++++++++++++++++++++++++ templates/index.html | 2 ++ 4 files changed, 104 insertions(+), 13 deletions(-) create mode 100644 agenda/types.py create mode 100755 agenda/waste_schedule.py diff --git a/agenda/__init__.py b/agenda/__init__.py index 360fffa..6ca9505 100644 --- a/agenda/__init__.py +++ b/agenda/__init__.py @@ -1,5 +1,4 @@ import configparser -import dataclasses import json import operator import os @@ -14,6 +13,7 @@ import dateutil import dateutil.parser import exchange_calendars import holidays +import lxml import pandas import pytz import requests @@ -22,6 +22,9 @@ from dateutil.easter import easter from agenda import thespacedevs +from . import waste_schedule +from .types import Event + 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)} - config_filename = os.path.join(os.path.dirname(__file__), "..", "config") assert os.path.exists(config_filename) @@ -51,15 +53,6 @@ access_key = config.get("exchangerate", "access_key") 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: """Calculate the date of the next UK Mother's Day from the current date.""" current_year = input_date.year @@ -137,8 +130,9 @@ def get_next_bank_holiday(input_date: date) -> list[Event]: return hols -def get_gbpusd(now: datetime) -> Decimal: +def get_gbpusd() -> Decimal: """Get the current value for GBPUSD, with caching.""" + now = datetime.now() now_str = now.strftime("%Y-%m-%d_%H:%M") fx_dir = os.path.join(data_dir, "fx") existing_data = os.listdir(fx_dir) @@ -326,6 +320,17 @@ def get_birthdays(from_date: date, config: configparser.ConfigParser) -> list[Ev 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]: """Get data to display on agenda dashboard.""" rocket_dir = os.path.join(data_dir, "thespacedevs") @@ -333,7 +338,7 @@ def get_data(now: datetime) -> dict[str, str | object]: reply = { "now": now, - "gbpusd": get_gbpusd(now), + "gbpusd": get_gbpusd(), "next_economist": next_economist(today), "bank_holiday": get_next_bank_holiday(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_conferences(today, "conferences.yaml") + events += waste_collection_events() next_up_series = Event( date=date(2026, 6, 1), diff --git a/agenda/types.py b/agenda/types.py new file mode 100644 index 0000000..6877b49 --- /dev/null +++ b/agenda/types.py @@ -0,0 +1,13 @@ +"""Types.""" + +import dataclasses +from datetime import date + + +@dataclasses.dataclass +class Event: + """Event.""" + + name: str + date: date + title: str | None = None diff --git a/agenda/waste_schedule.py b/agenda/waste_schedule.py new file mode 100755 index 0000000..6c66a4f --- /dev/null +++ b/agenda/waste_schedule.py @@ -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 diff --git a/templates/index.html b/templates/index.html index fed683a..1f15653 100644 --- a/templates/index.html +++ b/templates/index.html @@ -23,6 +23,7 @@ "xmas_last_first": "Christmas last posting 1st class", "xmas_day": "Christmas day", "next_up_series": "Next Up documentary", + "waste_schedule": "Waste schedule", } %} @@ -35,6 +36,7 @@