parent
ae8ed755c4
commit
4ee81de1fd
|
@ -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),
|
||||
|
|
|
@ -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()
|
||||
]
|
||||
|
|
Loading…
Reference in a new issue