Get waste collection dates from the Omrin API and convert this into an ics file.

omrin2ics

#!/usr/bin/env -S uv run --script

# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "requests", 
#   "rsa", 
#   "pycryptodome", 
#   "icalendar",
# ]
# ///

# Get waste collection dates from the Omrin API and convert this into an ics file.

import uuid
import requests
import base64
import json
import sys
import argparse
from datetime import datetime, timedelta
from rsa import pkcs1
from Crypto.PublicKey import RSA
from icalendar import Calendar, Event

def parse_arguments():
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Fetch waste collection dates from Omrin API and generate an iCal file."
    )
    parser.add_argument(
        "postal_code",
        help="The postal code for the location (e.g., 8921HA)."
    )
    parser.add_argument(
        "house_number",
        type=int,
        help="The house number for the location (e.g., 123)."
    )
    parser.add_argument(
        "-o", "--output",
        help="File path to save the iCal file (e.g., omrin.ics). If not provided, prints to stdout.",
        type=str
    )
    return parser.parse_args()

def get_omrin_token(base_url, app_id):
    """Fetch the Omrin API token."""
    response = requests.post(f"{base_url}/GetToken/", json={
        'AppId': app_id,
        'AppVersion': '',
        'OsVersion': '',
        'Platform': 'iCal'
    }).json()
    if not response.get('PublicKey'):
        raise ValueError("Failed to retrieve public key from Omrin API.")
    return response

def encrypt_request(public_key, request_body):
    """Encrypt the request body using RSA encryption."""
    rsa_public_key = RSA.importKey(base64.b64decode(public_key))
    encrypted_request = pkcs1.encrypt(json.dumps(request_body).encode(), rsa_public_key)
    return base64.b64encode(encrypted_request).decode("utf-8")

def fetch_calendar_data(base_url, app_id, encrypted_request):
    """Fetch waste collection calendar data from Omrin API."""
    response = requests.post(
        f"{base_url}/FetchAccount/{app_id}",
        '"' + encrypted_request + '"',
        timeout=60
    ).json()
    if not response.get('Success'):
        raise ValueError("Failed to fetch data from the Omrin API.")
    return response['CalendarV2'] if response.get('CalendarV2') else []

def create_calendar(calendar_data, datetime_format):
    """Create an iCal calendar from the retrieved data."""
    cal = Calendar()
    cal.add('prodid', '-//WasteCollection//omrin.nl//')
    cal.add('version', '2.0')

    for item in calendar_data:
        date = datetime.strptime(item['Datum'], datetime_format).date()
        summary = item['Omschrijving']
        description = item['Info']

        # Create an event for each waste collection moment
        event = Event()
        event.add('summary', summary)
        event.add('description', description)
        event.add('dtstart', date)
        event.add('transp', 'TRANSPARENT')  # Make events not show up as 'busy'

        cal.add_component(event)

    return cal

def main():
    args = parse_arguments()
    postal_code = args.postal_code
    house_number = args.house_number
    output_file = args.output

    # Constants
    base_url = "https://api-omrin.freed.nl/Account"
    app_id = uuid.uuid1().__str__()
    datetime_format = "%Y-%m-%dT%H:%M:%S%z"

    try:
        # Fetch API token and public key
        token_response = get_omrin_token(base_url, app_id)

        # Prepare and encrypt the request
        request_body = {
            'a': False,
            'Email': None,
            'Password': None,
            'PostalCode': postal_code,
            'HouseNumber': house_number
        }
        encrypted_request = encrypt_request(token_response['PublicKey'], request_body)

        # Fetch calendar data
        calendar_data = fetch_calendar_data(base_url, app_id, encrypted_request)

        # Create the calendar
        cal = create_calendar(calendar_data, datetime_format)
        ical_data = cal.to_ical().decode('utf-8')

        if output_file:
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(ical_data)
            print(f"iCal file saved to {output_file}")
        else:
            print(ical_data)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()