parking-expenses/main.py

#!/usr/bin/env python3

import PyPDF2
from openpyxl import load_workbook
import requests
import json
from datetime import datetime
import io
import subprocess
import tempfile
import argparse
import os
import calendar

FILTER_TAG_ID = 1
FILTER_CORRESPONDENT_ID = 1


def request_headers():
    token = os.environ.get("PAPERLESS_TOKEN")
    return {"Authorization": f"Token {token}"}


def get_start_date(year, month):
    if month == 1:
        year, month = (year - 1, 12)
    else:
        year, month = (year, month - 1)

    day = calendar.monthrange(year, month)[1]

    return (year, month, day)


def get_end_date(year, month):
    day = calendar.monthrange(year, month)[1]

    return (year, month, day)


def retrieve_parking_tickets(paperless_ngx_url, year, month):
    (start_year, start_month, start_day) = get_start_date(year, month)
    (end_year, end_month, end_day) = get_end_date(year, month)

    start_date = f"{start_year}{str(start_month).zfill(2)}{start_day}"
    end_date = f"{end_year}{str(end_month).zfill(2)}{end_day}"

    print(
        f"Fetching parking tickets from {start_day}-{start_month}-{start_year} to {end_day}-{end_month}-{end_year}."
    )

    response = requests.get(
        f"{paperless_ngx_url}/api/documents/?page_size=50&query=created:[{start_date} TO {end_date}]&tags__id__all={FILTER_TAG_ID}&correspondent__id__in={FILTER_CORRESPONDENT_ID}",
        headers=request_headers(),
    )

    if response.status_code != 200:
        print(
            f"HTTP {response.status_code} error while retrieving parking ticket: {response.text}"
        )
        exit(1)

    data = json.loads(response.text)
    print("Successfully retrieved " + str(len(data["results"])) + " parking tickets.")

    tickets = list()

    for ticket in data["results"]:
        amount1 = float(ticket["custom_fields"][0]["value"].removeprefix("EUR"))
        amount2 = float(ticket["custom_fields"][1]["value"].removeprefix("EUR"))

        if amount1 > amount2:
            price = amount1
            btw = amount2
        else:
            price = amount2
            btw = amount1

        date = datetime.strptime(ticket["created_date"], "%Y-%m-%d")
        tickets.append({"date": date, "price": price, "btw": btw, "id": ticket["id"]})

    return sorted(tickets, key=lambda t: t["date"])


def find_last_used_row(ws):
    for row in range(ws.max_row, 0, -1):
        # Iterate through the columns in the current row
        for column in range(1, ws.max_column + 1):
            # Check if the cell is not empty
            if ws.cell(row=row, column=column).value:
                # Found the last row with data, print and exit
                return row


def insert_declaration(number_plate, ws, row, date, price, btw):
    ws.cell(row=row, column=1, value=date.strftime("%d-%m-%Y"))
    ws.cell(row=row, column=2, value="Parkeerkosten P+R Noord")
    ws.cell(row=row, column=3, value=number_plate)
    ws.cell(row=row, column=4, value=(price - btw))
    ws.cell(row=row, column=5, value=btw)
    ws.cell(row=row, column=6, value=price)


def fill_template(tickets, template, number_plate):
    print("Creating expenses overview.")

    wb = load_workbook(filename=template)
    ws = wb.active
    last_row = find_last_used_row(ws)

    price_total = 0
    btw_total = 0

    for ticket in tickets:
        price_total += ticket["price"]
        btw_total += ticket["btw"]

        ws.insert_rows(last_row + 1)
        insert_declaration(
            number_plate, ws, last_row + 1, ticket["date"], ticket["price"], ticket["btw"]
        )
        last_row += 1

    # Insert empty row
    ws.insert_rows(last_row + 1)
    last_row += 1

    ws.insert_rows(last_row + 1)
    ws.cell(row=last_row + 1, column=1, value="Totaal onkosten")
    ws.cell(row=last_row + 1, column=4, value=(price_total - btw_total))
    ws.cell(row=last_row + 1, column=5, value=btw_total)
    ws.cell(row=last_row + 1, column=6, value=price_total)

    print("Expenses overview created.")

    return wb


def merge_spreadsheet_and_tickets(paperless_ngx_url, spreadsheet, tickets):
    print("Creating final PDF.")

    pdf_merger = PyPDF2.PdfMerger()

    print("Converting expenses report to PDF.")
    with tempfile.NamedTemporaryFile(delete=True, suffix=".xlsx") as xlsx_file:
        spreadsheet.save(xlsx_file)

        with tempfile.NamedTemporaryFile(delete=True, suffix=".pdf") as pdf_file:
            subprocess.run(
                f"unoconv -f pdf -o {pdf_file.name} {xlsx_file.name}",
                shell=True,
                capture_output=True,
                text=True,
            )

            pdf_merger.append(pdf_file.name)

    print("Adding parking tickets photos to PDF.")
    for index, ticket in enumerate(tickets):
        id = ticket["id"]
        print(
            f"Adding parking ticket (ID={id}) photo to PDF ({index+1}/{len(tickets)})."
        )
        response = requests.get(
            f"{paperless_ngx_url}/api/documents/{id}/download/",
            headers=request_headers(),
        )

        pdf_merger.append(io.BytesIO(response.content))

    return pdf_merger


def write_pdf_to_file(pdf_merger, output):
    print("Writing final PDF to file.")
    with open(output, "wb") as output_file:
        pdf_merger.write(output_file)
    print(f"Written expense report to {output}.")


def main():
    parser = argparse.ArgumentParser(
        description="Create a monthly travel expense report"
    )

    parser.add_argument(
        "-y",
        "--year",
        type=int,
        required=True,
        help="The year of the expenses.",
    )
    parser.add_argument(
        "-m",
        "--month",
        type=int,
        required=True,
        help="The month of the expenses (1-12).",
    )
    parser.add_argument(
        "-p",
        "--number-plate",
        type=str,
        required=True,
        help="Number plate of the parked vehicle."
    )
    parser.add_argument(
        "-u",
        "--paperless-ngx-url",
        type=str,
        required=True,
        help="Base URL of the Paperless-ngx instance."
    )
    parser.add_argument(
        "-o", "--output", help="Output file of resulting PDF", default="expenses.pdf"
    )
    parser.add_argument(
        "-t",
        "--template",
        help="XLSX template for the expense report",
        default="template.xlsx",
    )

    args = parser.parse_args()

    tickets = retrieve_parking_tickets(args.paperless_ngx_url, args.year, args.month)
    spreadsheet = fill_template(tickets, args.template, args.number_plate)
    pdf_merger = merge_spreadsheet_and_tickets(args.paperless_ngx_url, spreadsheet, tickets)
    write_pdf_to_file(pdf_merger, args.output)


if __name__ == "__main__":
    main()