Python-Projekte/mazda/mazda_upload.py

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


@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


@dataclass
class Token:
    access_token: str
    token_type: str
    refresh_token: str
    expires_in: int
    sub: str
    iss: str
    iat: int
    defaultDomain: str
    userGuid: str
    organisations: list[str]
    jti: str
    scope: list[str]
    expires_at: float


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='88888/MMD'
)

redirect_uri = 'https://localhost/'
base_dir = '/home/robert/projekte/python/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'


# After updating the token you will most likely want to save it.
def token_save(token):
    json.dump(token, open(base_dir + 'token.json', 'w'), indent=2)


def token_load() -> Token:
    try:
        return json.load(open(base_dir + 'token.json', 'r'))
    except FileNotFoundError:
        return None


def convert_csv(csv_file, json_file, year, month):
    date_min = datetime(year, month, 1, 0, 0, 0)
    date_max = datetime(year, month + 1, 1, 0, 0, 0)

    date_cols = ['invoiceDate', 'orderDate', 'orderCompletionDate', 'vehicleIntakeDate', 'nextMotDueDate']
    df = pd.read_csv(csv_file, 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'])

        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': item['serialNumber'],
                        'unit': 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(json_file, 'w'), indent=2)
    return res


def upload(data):
    headers = {
        'accept': 'application/vnd.mazdaeur.dms.v4+json',
        'x-mme-organisation': cfg.dealer_number,
        'X-mazda-org': cfg.dealer_number,
        'Content-Type': 'application/json',
        # 'Authorization': 'Bearer ' + token
    }
    extra = {
        'client_id': cfg.client_id,
        'client_secret': cfg.client_secret
    }

    token = token_load()
    oauth: OAuth2Session = None

    if token is not None:
        oauth = OAuth2Session(cfg.client_id, token=token)
        if token['expires_at'] < datetime.now().timestamp():
            # token = oauth.refresh_token(cfg.domain + cfg.auth_url, auth=(cfg.client_id, cfg.client_secret),
            #                            refresh_token=token['refresh_token'], redirect_response=None)
            # token_save(token)
            token = None

    if token is None or oauth is None:
        oauth = OAuth2Session(cfg.client_id, redirect_uri=redirect_uri)
        authorization_url, state = oauth.authorization_url(cfg.domain + cfg.auth_url)
        print('Please go here and authorize: ' + authorization_url)
        redirect_response = input('Paste the full redirect URL here:')
        token = oauth.fetch_token(cfg.domain + cfg.token_url, client_secret=cfg.client_secret, authorization_response=redirect_response)
        token_save(token)

    invoices = data['invoices']
    orders = data['orders']
    data['orders'] = []
    for i in invoices:
        data['invoices'] = [i]
        r = oauth.post(cfg.domain + cfg.webservice + cfg.module, json.dumps(data), headers=headers)
        print(f"{i['invoiceNumber']} => {r.status_code}")
    data['invoices'] = []
    for o in orders:
        data['orders'] = [o]
        r = oauth.post(cfg.domain + cfg.webservice + cfg.module, json.dumps(data), headers=headers)
        print(f"{o['orderNumber']} => {r.status_code}")

    with open(base_dir + 'post_error.log', 'w') as fwh:
        fwh.write(r.text)


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


if __name__ == '__main__':
    main()