Edward Betts
1806785700
Replace technical YAML diff output with human-readable bullet points that clearly explain what changed. Added structured email format with mission details, rocket emoji, and formatted dates. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
281 lines
9.7 KiB
Python
Executable file
281 lines
9.7 KiB
Python
Executable file
#!/usr/bin/python3
|
|
"""Combined update script for various data sources."""
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import typing
|
|
from datetime import date, datetime
|
|
from time import time
|
|
|
|
import deepdiff
|
|
import flask
|
|
import requests
|
|
import yaml
|
|
|
|
import agenda.bristol_waste
|
|
import agenda.fx
|
|
import agenda.geomob
|
|
import agenda.gwr
|
|
import agenda.mail
|
|
import agenda.thespacedevs
|
|
import agenda.types
|
|
import agenda.uk_holiday
|
|
from agenda.types import StrDict
|
|
from web_view import app
|
|
|
|
|
|
async def update_bank_holidays(config: flask.config.Config) -> None:
|
|
"""Update cached copy of UK Bank holidays."""
|
|
t0 = time()
|
|
events = await agenda.uk_holiday.get_holiday_list(config["DATA_DIR"])
|
|
time_taken = time() - t0
|
|
if not sys.stdin.isatty():
|
|
return
|
|
print(len(events), "bank holidays in list")
|
|
print(f"took {time_taken:.1f} seconds")
|
|
|
|
|
|
async def update_bristol_bins(config: flask.config.Config) -> None:
|
|
"""Update waste schedule from Bristol City Council."""
|
|
t0 = time()
|
|
events = await agenda.bristol_waste.get(
|
|
date.today(),
|
|
config["DATA_DIR"],
|
|
config["BRISTOL_UPRN"],
|
|
cache="refresh",
|
|
)
|
|
time_taken = time() - t0
|
|
if not sys.stdin.isatty():
|
|
return
|
|
for event in events:
|
|
print(event)
|
|
print(f"took {time_taken:.1f} seconds")
|
|
|
|
|
|
def update_gwr_advance_ticket_date(config: flask.config.Config) -> None:
|
|
"""Update GWR advance ticket date cache."""
|
|
filename = os.path.join(config["DATA_DIR"], "advance-tickets.html")
|
|
existing_html = open(filename).read()
|
|
|
|
existing_dates = agenda.gwr.extract_dates(existing_html)
|
|
assert existing_dates
|
|
assert list(existing_dates.keys()) == ["Weekdays", "Saturdays", "Sundays"]
|
|
|
|
new_html = requests.get(agenda.gwr.url).text
|
|
|
|
new_dates = agenda.gwr.extract_dates(new_html)
|
|
if not new_dates:
|
|
subject = "Error parsing GWR advance ticket booking dates"
|
|
body = new_html
|
|
agenda.mail.send_mail(config, subject, body)
|
|
return
|
|
|
|
assert new_dates
|
|
assert list(new_dates.keys()) == ["Weekdays", "Saturdays", "Sundays"]
|
|
|
|
if existing_dates == new_dates:
|
|
if sys.stdin.isatty():
|
|
print(filename)
|
|
print(agenda.gwr.url)
|
|
print("dates haven't changed:", existing_dates)
|
|
return
|
|
|
|
open(filename, "w").write(new_html)
|
|
|
|
subject = (
|
|
"New GWR advance ticket booking date: "
|
|
+ f'{new_dates["Weekdays"].strftime("%d %b %Y")} (Weekdays)'
|
|
)
|
|
body = f"""
|
|
{"\n".join(f'{key}: {when.strftime("%d %b %Y")}' for key, when in new_dates.items())}
|
|
|
|
{agenda.gwr.url}
|
|
|
|
Agenda: https://edwardbetts.com/agenda/
|
|
"""
|
|
|
|
if sys.stdin.isatty():
|
|
print(filename)
|
|
print(agenda.gwr.url)
|
|
print()
|
|
print("dates have changed")
|
|
print("old:", existing_dates)
|
|
print("new:", new_dates)
|
|
print()
|
|
print(subject)
|
|
print(body)
|
|
|
|
agenda.mail.send_mail(config, subject, body)
|
|
|
|
|
|
def format_launch_changes(differences: dict) -> str:
|
|
"""Convert deepdiff output to human-readable format."""
|
|
changes = []
|
|
# Handle value changes
|
|
if "values_changed" in differences:
|
|
for path, change in differences["values_changed"].items():
|
|
field = path.replace("root['", "").replace("']", "").replace("root.", "")
|
|
old_val = change["old_value"]
|
|
new_val = change["new_value"]
|
|
# Format specific fields nicely
|
|
if field == "net":
|
|
try:
|
|
old_dt = datetime.fromisoformat(old_val.replace('Z', '+00:00'))
|
|
new_dt = datetime.fromisoformat(new_val.replace('Z', '+00:00'))
|
|
changes.append(f"Launch time changed from {old_dt.strftime('%d %b %Y at %H:%M UTC')} to {new_dt.strftime('%d %b %Y at %H:%M UTC')}")
|
|
except:
|
|
changes.append(f"Launch time changed from {old_val} to {new_val}")
|
|
elif field == "name":
|
|
changes.append(f"Mission name changed from '{old_val}' to '{new_val}'")
|
|
elif field == "probability":
|
|
if old_val is None:
|
|
changes.append(f"Launch probability set to {new_val}%")
|
|
elif new_val is None:
|
|
changes.append("Launch probability removed")
|
|
else:
|
|
changes.append(f"Launch probability changed from {old_val}% to {new_val}%")
|
|
elif "status" in field:
|
|
changes.append(f"Status changed from '{old_val}' to '{new_val}'")
|
|
else:
|
|
changes.append(f"{field.replace('_', ' ').title()} changed from '{old_val}' to '{new_val}'")
|
|
# Handle additions
|
|
if "dictionary_item_added" in differences:
|
|
for path in differences["dictionary_item_added"]:
|
|
field = path.replace("root['", "").replace("']", "").replace("root.", "")
|
|
changes.append(f"New field added: {field.replace('_', ' ').title()}")
|
|
# Handle removals
|
|
if "dictionary_item_removed" in differences:
|
|
for path in differences["dictionary_item_removed"]:
|
|
field = path.replace("root['", "").replace("']", "").replace("root.", "")
|
|
changes.append(f"Field removed: {field.replace('_', ' ').title()}")
|
|
# Handle type changes
|
|
if "type_changes" in differences:
|
|
for path, change in differences["type_changes"].items():
|
|
field = path.replace("root['", "").replace("']", "").replace("root.", "")
|
|
changes.append(f"{field.replace('_', ' ').title()} type changed from {change['old_type'].__name__} to {change['new_type'].__name__}")
|
|
return "\n".join(f"• {change}" for change in changes) if changes else "No specific changes detected"
|
|
|
|
|
|
def report_space_launch_change(
|
|
config: flask.config.Config, prev_launch: StrDict | None, cur_launch: StrDict | None
|
|
) -> None:
|
|
"""Send mail to announce change to space launch data."""
|
|
if cur_launch:
|
|
name = cur_launch["name"]
|
|
status = cur_launch.get("status", {}).get("name", "Unknown") if isinstance(cur_launch.get("status"), dict) else "Unknown"
|
|
launch_date = cur_launch.get("net", "Unknown")
|
|
location = cur_launch.get("pad", {}).get("location", {}).get("name", "Unknown") if isinstance(cur_launch.get("pad"), dict) else "Unknown"
|
|
else:
|
|
assert prev_launch
|
|
name = prev_launch["name"]
|
|
status = "Cancelled/Removed"
|
|
launch_date = prev_launch.get("net", "Unknown")
|
|
location = prev_launch.get("pad", {}).get("location", {}).get("name", "Unknown") if isinstance(prev_launch.get("pad"), dict) else "Unknown"
|
|
subject = f"Space Launch Update: {name}"
|
|
|
|
differences = deepdiff.DeepDiff(prev_launch, cur_launch)
|
|
changes_text = format_launch_changes(differences)
|
|
# Format launch date nicely
|
|
formatted_date = "Unknown"
|
|
if launch_date and launch_date != "Unknown":
|
|
try:
|
|
dt = datetime.fromisoformat(launch_date.replace('Z', '+00:00'))
|
|
formatted_date = dt.strftime('%d %b %Y at %H:%M UTC')
|
|
except:
|
|
formatted_date = launch_date
|
|
|
|
body = f"""🚀 Space Launch Update
|
|
|
|
Mission: {name}
|
|
Status: {status}
|
|
Launch Date: {formatted_date}
|
|
Location: {location}
|
|
|
|
Changes:
|
|
{changes_text}
|
|
|
|
View all launches: https://edwardbetts.com/agenda/launches
|
|
"""
|
|
|
|
agenda.mail.send_mail(config, subject, body)
|
|
|
|
|
|
def get_launch_by_slug(data: StrDict, slug: str) -> StrDict | None:
|
|
"""Find last update for space launch."""
|
|
return {item["slug"]: typing.cast(StrDict, item) for item in data["results"]}.get(
|
|
slug
|
|
)
|
|
|
|
|
|
def update_thespacedevs(config: flask.config.Config) -> None:
|
|
"""Update cache of space launch API."""
|
|
rocket_dir = os.path.join(config["DATA_DIR"], "thespacedevs")
|
|
|
|
existing_data = agenda.thespacedevs.load_cached_launches(rocket_dir)
|
|
assert existing_data
|
|
prev_launches = {
|
|
slug: get_launch_by_slug(existing_data, slug)
|
|
for slug in config["FOLLOW_LAUNCHES"]
|
|
}
|
|
|
|
t0 = time()
|
|
data = agenda.thespacedevs.next_launch_api_data(rocket_dir)
|
|
if not data:
|
|
return # thespacedevs API call failed
|
|
cur_launches = {
|
|
slug: get_launch_by_slug(data, slug) for slug in config["FOLLOW_LAUNCHES"]
|
|
}
|
|
|
|
for slug in config["FOLLOW_LAUNCHES"]:
|
|
prev, cur = prev_launches[slug], cur_launches[slug]
|
|
if prev is None and cur is None:
|
|
continue
|
|
if prev and cur and prev["last_updated"] == cur["last_updated"]:
|
|
continue
|
|
report_space_launch_change(config, prev, cur)
|
|
|
|
time_taken = time() - t0
|
|
if not sys.stdin.isatty():
|
|
return
|
|
rockets = [agenda.thespacedevs.summarize_launch(item) for item in data["results"]]
|
|
print(len(rockets), "launches")
|
|
print(f"took {time_taken:.1f} seconds")
|
|
|
|
|
|
def update_gandi(config: flask.config.Config) -> None:
|
|
"""Retrieve list of domains from gandi.net."""
|
|
url = "https://api.gandi.net/v5/domain/domains"
|
|
headers = {"authorization": "Bearer " + config["GANDI_TOKEN"]}
|
|
filename = os.path.join(config["DATA_DIR"], "gandi_domains.json")
|
|
|
|
r = requests.request("GET", url, headers=headers)
|
|
items = r.json()
|
|
assert isinstance(items, list)
|
|
assert all(item["fqdn"] and item["dates"]["registry_ends_at"] for item in items)
|
|
with open(filename, "w") as out:
|
|
out.write(r.text)
|
|
|
|
|
|
def main() -> None:
|
|
"""Update caches."""
|
|
now = datetime.now()
|
|
hour = now.hour
|
|
|
|
with app.app_context():
|
|
if hour % 3 == 0:
|
|
asyncio.run(update_bank_holidays(app.config))
|
|
asyncio.run(update_bristol_bins(app.config))
|
|
update_gwr_advance_ticket_date(app.config)
|
|
# TODO: debug why update gandi fails
|
|
# update_gandi(app.config)
|
|
agenda.geomob.update(app.config)
|
|
|
|
agenda.fx.get_rates(app.config)
|
|
|
|
update_thespacedevs(app.config)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|