mazda/mazda_upload.py

import pandas as pd
import numpy as np
import json
from requests_oauthlib import OAuth2Session
from dataclasses import dataclass
from datetime import datetime
from dateutil.relativedelta import relativedelta


@dataclass
class MazdaConfig:
    domain: str
    webservice: str
    module: str
    auth_url: str
    token_url: str
    client_id: str
    client_secret: str
    username: str
    password: str
    dealer_number: str


cfg = MazdaConfig(
    domain="https://mappsacc.mazdaeur.com",
    webservice="/dogma-restapi-dms/api",
    module="/vehicles/workshop/order-report",
    auth_url="/oauth/authorize",
    token_url="/oauth/token",
    client_id="E7FC943B-B73F-F48E-B71A-419EA4CD4AC7",
    client_secret="^bH=rk@c58zrr^Apc#9fzy$c",
    username="mmd88888.cdk",
    password="MazdaCX30",
    dealer_number="15972/MMD",
)

redirect_uri = "https://localhost:8085/"
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": str(item["unit"]),
                        },
                    }
                )
            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 upload(oauth: OAuth2Session, data):
    headers = {
        "accept": "application/vnd.mazdaeur.dms.v5+json",
        "x-mme-organisation": cfg.dealer_number,
        "X-mazda-org": cfg.dealer_number,
        "Content-Type": "application/json",
        # 'Authorization': 'Bearer ' + token
    }

    invoices = data["invoices"]
    orders = data["orders"]
    data["orders"] = []
    for i in invoices:
        data["invoices"] = [i]
        order_no = [item["orderNumber"] for item in i["invoiceItems"]]
        data["orders"] = [o for o in orders if o["orderNumber"] in order_no]
        r = oauth.post(cfg.domain + cfg.webservice + cfg.module, json.dumps(data), headers=headers)
        print(f"{i['invoiceNumber']} => {r.status_code}")
        if r.status_code not in [200, 202]:
            with open(base_dir + f"logs/invoice_{i['invoiceNumber']}.log", "wb") as fwh:
                fwh.write(r.content)


def main():
    data = convert_csv(base_dir + "data/Workshop_Order_Report.csv", base_dir + "temp/mazda_export.json", 2021, 6)
    # data = json.load(open(base_dir + 'mazda_export.json', 'r'))
    upload(None, data)


if __name__ == "__main__":
    main()