diff --git a/agenda/__init__.py b/agenda/__init__.py index 46bd82d..24fa7de 100644 --- a/agenda/__init__.py +++ b/agenda/__init__.py @@ -594,20 +594,28 @@ def waste_collection_events() -> list[Event]: return events -def bristol(): +def bristol_waste_collection_events(start_date: date) -> list[Event]: + """Waste colllection events.""" + uprn = "358335" + + return waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn) + + +def bristol() -> ephem.Observer: + """Location of Bristol.""" observer = ephem.Observer() observer.lat, observer.lon = "51.4545", "-2.5879" return observer -def sunrise(observer): +def sunrise(observer: ephem.Observer) -> datetime: """Sunrise.""" - return observer.next_rising(ephem.Sun(observer)).datetime() + return typing.cast(datetime, observer.next_rising(ephem.Sun(observer)).datetime()) -def sunset(observer): +def sunset(observer: ephem.Observer) -> datetime: """Sunrise.""" - return observer.next_setting(ephem.Sun(observer)).datetime() + return typing.cast(datetime, observer.next_setting(ephem.Sun(observer)).datetime()) def get_data(now: datetime) -> typing.Mapping[str, str | object]: @@ -663,7 +671,7 @@ def get_data(now: datetime) -> typing.Mapping[str, str | object]: events += get_accommodation(today, os.path.join(my_data, "accommodation.yaml")) events += get_all_travel_events(today) events += get_conferences(today, os.path.join(my_data, "conferences.yaml")) - events += waste_collection_events() + events += waste_collection_events() + bristol_waste_collection_events(today) next_up_series = Event( date=date(2026, 6, 1), diff --git a/agenda/waste_schedule.py b/agenda/waste_schedule.py index f656f55..7be18d8 100755 --- a/agenda/waste_schedule.py +++ b/agenda/waste_schedule.py @@ -1,6 +1,8 @@ #!/usr/bin/python3 +import json import os +import typing from collections import defaultdict from datetime import date, datetime, timedelta @@ -9,13 +11,23 @@ import requests from .types import Event +ttl_hours = 6 + + +def make_waste_dir(data_dir: str) -> None: + """Make waste dir if missing.""" + waste_dir = os.path.join(data_dir, "waste") + if not os.path.exists(waste_dir): + os.mkdir(waste_dir) + 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) + + make_waste_dir(data_dir) + existing_data = os.listdir(waste_dir) existing = [f for f in existing_data if f.endswith(".html")] if existing: @@ -23,7 +35,7 @@ def get_html(data_dir: str, postcode: str, uprn: str) -> str: recent = datetime.strptime(recent_filename, "%Y-%m-%d_%H:%M.html") delta = now - recent - if existing and delta < timedelta(hours=6): + if existing and delta < timedelta(hours=ttl_hours): return open(os.path.join(waste_dir, recent_filename)).read() now_str = now.strftime("%Y-%m-%d_%H:%M") @@ -69,6 +81,110 @@ def parse(root: lxml.html.HtmlElement) -> list[Event]: by_date[following_date].append(service) return [ - Event(name="waste_schedule", date=d, title=", ".join(services)) + Event(name="waste_schedule", date=d, title="Backwell: " + ", ".join(services)) + for d, services in by_date.items() + ] + + +BristolSchedule = list[dict[str, typing.Any]] + + +def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule: + """Get Bristol Waste schedule, with cache.""" + now = datetime.now() + waste_dir = os.path.join(data_dir, "waste") + + make_waste_dir(data_dir) + + existing_data = os.listdir(waste_dir) + existing = [f for f in existing_data if f.endswith(f"_{uprn}.json")] + if existing: + recent_filename = max(existing) + recent = datetime.strptime(recent_filename, f"%Y-%m-%d_%H:%M_{uprn}.json") + delta = now - recent + + if existing and delta < timedelta(hours=ttl_hours): + json_data = json.load(open(os.path.join(waste_dir, recent_filename))) + return typing.cast(BristolSchedule, json_data["data"]) + + now_str = now.strftime("%Y-%m-%d_%H:%M") + filename = f"{waste_dir}/{now_str}_{uprn}.json" + + r = get_bristol_gov_uk_data(uprn) + + with open(filename, "wb") as out: + out.write(r.content) + + return typing.cast(BristolSchedule, r.json()["data"]) + + +def get_bristol_gov_uk_data(uprn: str) -> requests.Response: + """Get JSON from Bristol City Council.""" + UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" + HEADERS = { + "Accept": "*/*", + "Accept-Language": "en-GB,en;q=0.9", + "Connection": "keep-alive", + "Ocp-Apim-Subscription-Key": "47ffd667d69c4a858f92fc38dc24b150", + "Ocp-Apim-Trace": "true", + "Origin": "https://bristolcouncil.powerappsportals.com", + "Referer": "https://bristolcouncil.powerappsportals.com/", + "Sec-Fetch-Dest": "empty", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Site": "cross-site", + "Sec-GPC": "1", + "User-Agent": UA, + } + + _uprn = str(uprn).zfill(12) + s = requests.Session() + + # Initialise form + payload = {"servicetypeid": "7dce896c-b3ba-ea11-a812-000d3a7f1cdc"} + response = s.get( + "https://bristolcouncil.powerappsportals.com/completedynamicformunauth/", + headers=HEADERS, + params=payload, + ) + + host = "bcprdapidyna002.azure-api.net" + + # Set the search criteria + payload = {"Uprn": "UPRN" + _uprn} + response = s.post( + f"https://{host}/bcprdfundyna001-llpg/DetailedLLPG", + headers=HEADERS, + json=payload, + ) + + # Retrieve the schedule + payload = {"uprn": _uprn} + response = s.post( + f"https://{host}/bcprdfundyna001-alloy/NextCollectionDates", + headers=HEADERS, + json=payload, + ) + return response + + +def get_bristol_gov_uk(start_date: date, data_dir: str, uprn: str) -> list[Event]: + """Get waste collection schedule from Bristol City Council.""" + data = get_bristol_data(data_dir, uprn) + + by_date: defaultdict[date, list[str]] = defaultdict(list) + + for item in data: + service = item["containerName"] + service = "Recycling" if "Recycling" in service else service.partition(" ")[2] + for collection in item["collection"]: + for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: + d = date.fromisoformat(collection[collection_date_key][:10]) + if d < start_date: + continue + if service not in by_date[d]: + by_date[d].append(service) + + return [ + Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services)) for d, services in by_date.items() ]