mazda/mazda_export.py

import json
from datetime import datetime

import numpy as np
import pandas as pd
from dateutil.relativedelta import relativedelta

base_dir = "C:/projekte/mazda/"


def date_format(d: datetime):
    if d == 0:
        return ""  # '0000-00-00T00:00:00.000Z'
    date_str = d.isoformat(sep="T")
    if len(date_str) == 19:
        return date_str + ".000Z"
    return date_str[:-3] + "Z"


def convert_csv(import_csv, export_json, year, month):
    date_min = datetime(year, month, 5, 0, 0, 0)
    date_max = datetime(year, month, 5, 0, 0, 0) + relativedelta(days=5)

    date_cols = ["invoiceDate", "orderDate", "orderCompletionDate", "vehicleIntakeDate", "nextMotDueDate"]
    df = pd.read_csv(import_csv, encoding="latin-1", decimal=",", sep=";", parse_dates=date_cols)
    df = df.fillna(0)  # [(df["invoiceDate"] >= date_min) & (df["invoiceDate"] <= date_max)]
    df = df.sort_values(by=["invoiceDate", "invoiceNumber", "orderNumber", "lineNumber"])
    df["vin"] = np.where(df["vin"] == 0, "0" * 17, df["vin"])
    # print(df[['currency','documentType','invoiceCategory','invoiceDate','invoiceNumber']].drop_duplicates().info())
    invoices_filter = ["currency", "documentType", "invoiceCategory", "invoiceDate", "invoiceNumber"]
    invoices = df[invoices_filter].drop_duplicates().to_dict("records")
    invoice_items_filter = [
        "invoiceNumber",
        "orderLineNumber",
        "orderNumber",
        "amount",
        "discount",
        "portion",
        "unitPrice",
    ]
    invoice_items = df[invoice_items_filter].groupby("invoiceNumber")

    for invoice in invoices:
        invoice["invoiceDate"] = date_format(invoice["invoiceDate"])
        items = invoice_items.get_group(invoice["invoiceNumber"])
        items.pop("invoiceNumber")
        invoice["invoiceItems"] = items.to_dict("records")

    orders = (
        df[["orderNumber", "orderDate", "orderCompletionDate", "vehicleIntakeDate"]]
        .drop_duplicates()
        .to_dict("records")
    )
    orders_vehicle_filter = ["orderNumber", "licensePlate", "nextMotDueDate", "odometer", "odometerUnit", "vin"]
    orders_vehicle = df[orders_vehicle_filter].drop_duplicates().groupby("orderNumber")
    orders_items = (
        df[
            [
                "orderNumber",
                "lineNumber",
                "orderItemType",
                "category",
                "descriptionOperation",
                "hours",
                "operationCode",
                "standardHours",
                "descriptionOther",
                "type",
                "descriptionPart",
                "isDamageCausal",
                "manufacturer",
                "partNumber",
                "quantity",
                "serialNumber",
                "unit",
                "company",
                "descriptionPurchase",
                "invoiceCode",
                "invoiceDate",
                "invoiceNumber",
            ]
        ]
        .drop_duplicates()
        .groupby("orderNumber")
    )

    for order in orders:
        order["vehicle"] = orders_vehicle.get_group(order["orderNumber"]).to_dict("records")[0]
        order["vehicle"]["nextMotDueDate"] = date_format(order["vehicle"]["nextMotDueDate"])
        # odo_str = str(order["vehicle"]["odometer"])
        order["vehicle"]["odometer"] = int(order["vehicle"]["odometer"])

        order["orderDate"] = date_format(order["orderDate"])
        order["orderCompletionDate"] = date_format(order["orderCompletionDate"])
        order["vehicleIntakeDate"] = date_format(order["vehicleIntakeDate"])

        items = orders_items.get_group(order["orderNumber"]).to_dict("records")
        order["items"] = []
        for item in items:
            if item["orderItemType"] == "operation":
                order["items"].append(
                    {
                        "lineNumber": item["lineNumber"],
                        "operation": {
                            "category": item["category"],
                            "description": item["descriptionOperation"],
                            "hours": item["hours"],
                            "operationCode": item["operationCode"],
                            "standardHours": item["standardHours"],
                        },
                    }
                )
            elif item["orderItemType"] == "part":
                order["items"].append(
                    {
                        "lineNumber": item["lineNumber"],
                        "part": {
                            "description": item["descriptionPart"],
                            "isDamageCausal": item["isDamageCausal"],
                            "manufacturer": item["manufacturer"],
                            "partNumber": item["partNumber"],
                            "quantity": item["quantity"],
                            "serialNumber": str(item["serialNumber"]),
                            "unit": "pcs",
                        },
                    }
                )
            elif item["orderItemType"] == "other":
                order["items"].append(
                    {
                        "lineNumber": item["lineNumber"],
                        "other": {"description": item["descriptionOther"], "type": item["type"]},
                    }
                )
            else:
                order["items"].append(
                    {
                        "lineNumber": item["lineNumber"],
                        "purchaseInvoice": {
                            "company": item["company"],
                            "description": item["descriptionPurchase"],
                            "invoiceCode": item["invoiceCode"],
                            "invoiceDate": date_format(item["invoiceDate"]),
                            "invoiceNumber": item["invoiceNumber"],
                        },
                    }
                )

    res = {
        "creationDate": date_format(datetime.now()),
        "invoices": invoices,
        "orders": orders,
        "timeRangeBegin": date_format(date_min),
        "timeRangeEnd": date_format(date_max),
    }

    # json.dump(res, open(export_json, "w"), indent=2)
    return res


def export_orders(data, export_dir):
    dt = datetime.now()
    timestamp = dt.strftime("%Y%m%d_%H%M%S")

    invoices = data["invoices"]
    orders = data["orders"]

    data["invoices"] = []
    data["orders"] = []

    for order in orders:
        order_date = order["orderDate"]
        data["creationDate"] = order_date
        data["timeRangeBegin"] = order_date
        data["timeRangeEnd"] = order_date
        period = order_date[:4] + order_date[5:7]
        data["orders"] = [order]
        order_no = order["orderNumber"]
        data["invoices"] = [i for i in invoices if i["invoiceItems"][0]["orderNumber"] == order_no]
        json.dump(data, open(f"{export_dir}/order-report_{order_no}_{period}_{timestamp}.json", "w"), indent=2)


def main():
    data = convert_csv(
        base_dir + "data/Workshop_Order_Report.csv", base_dir + "export/Mazda/temp/mazda_export.json", 2024, 1
    )
    export_orders(data, base_dir + "export/Mazda/temp")


if __name__ == "__main__":
    main()