meal-order-manager/scripts/upload_menu.py

"""Call the IAM-protected weekly-menu publication API with SigV4."""

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

PROJECT_ROOT = Path(__file__).resolve().parents[1]
OUTPUT_DIR = PROJECT_ROOT / "output"


def signed_request(
    api_url: str,
    path: str,
    method: str = "GET",
    body: dict | None = None,
    region: str = "us-east-1",
) -> dict:
    if not api_url.startswith(("http://", "https://")):
        raise ValueError(f"api_url must use http(s) scheme: {api_url!r}")
    data = json.dumps(body).encode() if body is not None else None
    headers = {"Content-Type": "application/json"} if data is not None else {}
    request = AWSRequest(
        method=method,
        url=f"{api_url.rstrip('/')}{path}",
        data=data,
        headers=headers,
    )
    credentials = boto3.Session().get_credentials()
    if credentials is None:
        raise RuntimeError("AWS credentials are required")
    SigV4Auth(credentials.get_frozen_credentials(), "execute-api", region).add_auth(
        request
    )

    prepared = request.prepare()
    http_request = urllib.request.Request(
        prepared.url,
        data=prepared.body,
        headers=dict(prepared.headers),
        method=method,
    )
    try:
        with urllib.request.urlopen(http_request, timeout=30) as response:
            return json.loads(response.read())
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode(errors="replace")
        raise RuntimeError(
            f"Publication API returned HTTP {exc.code}: {detail}"
        ) from exc


def get_settings(api_url: str, region: str) -> dict:
    return signed_request(api_url, "/api/publish/settings", method="GET", region=region)


def publish_menu(api_url: str, region: str) -> dict:
    files = sorted(OUTPUT_DIR.glob("menu-*.json"), reverse=True)
    if not files:
        raise RuntimeError("No menu JSON found. Run scrape_menu.py first.")

    with files[0].open() as menu_file:
        menu = json.load(menu_file)
    return signed_request(
        api_url,
        "/api/publish/menu",
        method="POST",
        body=menu,
        region=region,
    )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=("settings", "publish"))
    parser.add_argument("--api-url", required=True)
    parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-east-1"))
    args = parser.parse_args()

    try:
        result = (
            get_settings(args.api_url, args.region)
            if args.action == "settings"
            else publish_menu(args.api_url, args.region)
        )
    except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    print(json.dumps(result))
    return 0


if __name__ == "__main__":
    sys.exit(main())