agenda/update.py

#!/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"],
        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 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()