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 pprint import pformat
from time import time

import deepdiff  # type: ignore
import flask
import requests

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.

{pformat(differences)}
    """

    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()