parent
95bab9a62b
commit
bc31db6143
|
@ -301,11 +301,11 @@ async def waste_collection_events() -> list[Event]:
|
||||||
return events
|
return events
|
||||||
|
|
||||||
|
|
||||||
def bristol_waste_collection_events(start_date: date) -> list[Event]:
|
async def bristol_waste_collection_events(start_date: date) -> list[Event]:
|
||||||
"""Waste colllection events."""
|
"""Waste colllection events."""
|
||||||
uprn = "358335"
|
uprn = "358335"
|
||||||
|
|
||||||
return waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)
|
return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)
|
||||||
|
|
||||||
|
|
||||||
async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||||||
|
@ -325,12 +325,14 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||||||
bank_holiday,
|
bank_holiday,
|
||||||
rockets,
|
rockets,
|
||||||
backwell_bins,
|
backwell_bins,
|
||||||
|
bristol_bins,
|
||||||
) = await asyncio.gather(
|
) = await asyncio.gather(
|
||||||
fx.get_gbpusd(config),
|
fx.get_gbpusd(config),
|
||||||
gwr.advance_ticket_date(data_dir),
|
gwr.advance_ticket_date(data_dir),
|
||||||
get_next_bank_holiday(last_year, next_year),
|
get_next_bank_holiday(last_year, next_year),
|
||||||
thespacedevs.get_launches(rocket_dir, limit=40),
|
thespacedevs.get_launches(rocket_dir, limit=40),
|
||||||
waste_collection_events(),
|
waste_collection_events(),
|
||||||
|
bristol_waste_collection_events(today),
|
||||||
)
|
)
|
||||||
|
|
||||||
reply = {
|
reply = {
|
||||||
|
@ -385,7 +387,7 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
|
||||||
events += get_accommodation(os.path.join(my_data, "accommodation.yaml"))
|
events += get_accommodation(os.path.join(my_data, "accommodation.yaml"))
|
||||||
events += travel.all_events(today, config["data"]["personal-data"])
|
events += travel.all_events(today, config["data"]["personal-data"])
|
||||||
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
|
events += conference.get_list(os.path.join(my_data, "conferences.yaml"))
|
||||||
events += backwell_bins + bristol_waste_collection_events(today)
|
events += backwell_bins + bristol_bins
|
||||||
|
|
||||||
next_up_series = Event(
|
next_up_series = Event(
|
||||||
date=date(2026, 6, 1),
|
date=date(2026, 6, 1),
|
||||||
|
|
|
@ -6,9 +6,8 @@ import typing
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
|
|
||||||
import lxml.html
|
|
||||||
import requests
|
|
||||||
import httpx
|
import httpx
|
||||||
|
import lxml.html
|
||||||
|
|
||||||
from .types import Event
|
from .types import Event
|
||||||
|
|
||||||
|
@ -91,7 +90,7 @@ def parse(root: lxml.html.HtmlElement) -> list[Event]:
|
||||||
BristolSchedule = list[dict[str, typing.Any]]
|
BristolSchedule = list[dict[str, typing.Any]]
|
||||||
|
|
||||||
|
|
||||||
def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule:
|
async def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule:
|
||||||
"""Get Bristol Waste schedule, with cache."""
|
"""Get Bristol Waste schedule, with cache."""
|
||||||
now = datetime.now()
|
now = datetime.now()
|
||||||
waste_dir = os.path.join(data_dir, "waste")
|
waste_dir = os.path.join(data_dir, "waste")
|
||||||
|
@ -112,7 +111,7 @@ def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule:
|
||||||
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
||||||
filename = f"{waste_dir}/{now_str}_{uprn}.json"
|
filename = f"{waste_dir}/{now_str}_{uprn}.json"
|
||||||
|
|
||||||
r = get_bristol_gov_uk_data(uprn)
|
r = await get_bristol_gov_uk_data(uprn)
|
||||||
|
|
||||||
with open(filename, "wb") as out:
|
with open(filename, "wb") as out:
|
||||||
out.write(r.content)
|
out.write(r.content)
|
||||||
|
@ -120,7 +119,7 @@ def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule:
|
||||||
return typing.cast(BristolSchedule, r.json()["data"])
|
return typing.cast(BristolSchedule, r.json()["data"])
|
||||||
|
|
||||||
|
|
||||||
def get_bristol_gov_uk_data(uprn: str) -> requests.Response:
|
async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response:
|
||||||
"""Get JSON from Bristol City Council."""
|
"""Get JSON from Bristol City Council."""
|
||||||
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
|
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
|
||||||
HEADERS = {
|
HEADERS = {
|
||||||
|
@ -139,11 +138,11 @@ def get_bristol_gov_uk_data(uprn: str) -> requests.Response:
|
||||||
}
|
}
|
||||||
|
|
||||||
_uprn = str(uprn).zfill(12)
|
_uprn = str(uprn).zfill(12)
|
||||||
s = requests.Session()
|
|
||||||
|
|
||||||
|
async with httpx.AsyncClient() as client:
|
||||||
# Initialise form
|
# Initialise form
|
||||||
payload = {"servicetypeid": "7dce896c-b3ba-ea11-a812-000d3a7f1cdc"}
|
payload = {"servicetypeid": "7dce896c-b3ba-ea11-a812-000d3a7f1cdc"}
|
||||||
response = s.get(
|
response = await client.get(
|
||||||
"https://bristolcouncil.powerappsportals.com/completedynamicformunauth/",
|
"https://bristolcouncil.powerappsportals.com/completedynamicformunauth/",
|
||||||
headers=HEADERS,
|
headers=HEADERS,
|
||||||
params=payload,
|
params=payload,
|
||||||
|
@ -153,7 +152,7 @@ def get_bristol_gov_uk_data(uprn: str) -> requests.Response:
|
||||||
|
|
||||||
# Set the search criteria
|
# Set the search criteria
|
||||||
payload = {"Uprn": "UPRN" + _uprn}
|
payload = {"Uprn": "UPRN" + _uprn}
|
||||||
response = s.post(
|
response = await client.post(
|
||||||
f"https://{host}/bcprdfundyna001-llpg/DetailedLLPG",
|
f"https://{host}/bcprdfundyna001-llpg/DetailedLLPG",
|
||||||
headers=HEADERS,
|
headers=HEADERS,
|
||||||
json=payload,
|
json=payload,
|
||||||
|
@ -161,17 +160,18 @@ def get_bristol_gov_uk_data(uprn: str) -> requests.Response:
|
||||||
|
|
||||||
# Retrieve the schedule
|
# Retrieve the schedule
|
||||||
payload = {"uprn": _uprn}
|
payload = {"uprn": _uprn}
|
||||||
response = s.post(
|
response = await client.post(
|
||||||
f"https://{host}/bcprdfundyna001-alloy/NextCollectionDates",
|
f"https://{host}/bcprdfundyna001-alloy/NextCollectionDates",
|
||||||
headers=HEADERS,
|
headers=HEADERS,
|
||||||
json=payload,
|
json=payload,
|
||||||
)
|
)
|
||||||
|
|
||||||
return response
|
return response
|
||||||
|
|
||||||
|
|
||||||
def get_bristol_gov_uk(start_date: date, data_dir: str, uprn: str) -> list[Event]:
|
async def get_bristol_gov_uk(start_date: date, data_dir: str, uprn: str) -> list[Event]:
|
||||||
"""Get waste collection schedule from Bristol City Council."""
|
"""Get waste collection schedule from Bristol City Council."""
|
||||||
data = get_bristol_data(data_dir, uprn)
|
data = await get_bristol_data(data_dir, uprn)
|
||||||
|
|
||||||
by_date: defaultdict[date, list[str]] = defaultdict(list)
|
by_date: defaultdict[date, list[str]] = defaultdict(list)
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue