agenda/update.py

203 lines
5.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 # type: ignore
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"], refresh=True
)
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)
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 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"]
else:
assert prev_launch
name = prev_launch["name"]
subject = f"Change to {name}"
differences = deepdiff.DeepDiff(prev_launch, cur_launch)
body = f"""
A space launch of interest was updated.
{yaml.dump(differences)}
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)
update_gandi(app.config)
agenda.geomob.update(app.config)
agenda.fx.get_rates(app.config)
update_thespacedevs(app.config)
if __name__ == "__main__":
main()