import pandas as pd import numpy as np import json from collections import namedtuple from requests_oauthlib import OAuth2Session from datetime import datetime MazdaConfig = namedtuple('MazdaConfig', 'domain webservice module auth_url token_url ' + 'client_id client_secret username password dealer_number') 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' class mazda_upload: 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 token_save(self, token): json.dump(token, open(self.base_dir + 'token.json', 'w'), indent=2) def token_load(self): try: return json.load(open(self.base_dir + 'token.json', 'r')) except FileNotFoundError: return None def convert_csv(self, 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(self.base_dir + 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(self.base_dir + json_file, 'w'), indent=2) return res def login(self): token = self.token_load() if token is None or token['expires_at'] < datetime.now().timestamp(): self.oauth = OAuth2Session(self.cfg.client_id, redirect_uri=self.redirect_uri) authorization_url, _ = self.oauth.authorization_url(self.cfg.domain + self.cfg.auth_url) return authorization_url else: extra = { 'client_id': self.cfg.client_id, 'client_secret': self.cfg.client_secret } self.oauth = OAuth2Session(self.cfg.client_id, token=token, auto_refresh_url=self.cfg.domain + self.cfg.token_url, auto_refresh_kwargs=extra, token_updater=self.token_save) return None def fetch_token(self, redirect_response): token = self.oauth.fetch_token(self.cfg.domain + self.cfg.token_url, client_secret=self.cfg.client_secret, authorization_response=redirect_response) self.token_save(token) def upload(self, data): headers = { 'accept': 'application/vnd.mazdaeur.dms.v4+json', 'x-mme-organisation': self.cfg.dealer_number, 'X-mazda-org': self.cfg.dealer_number, 'Content-Type': 'application/json', # 'Authorization': 'Bearer ' + token } r = self.oauth.post(self.cfg.domain + self.cfg.webservice + self.cfg.module, json.dumps(data), headers=headers) print(r.status_code) with open(self.base_dir + 'post_error.log', 'w') as fwh: fwh.write(r.text) def main(): mu = mazda_upload() data = mu.convert_csv('Workshop_Order_Report.csv', 'mazda_export.json', 2021, 6) # data = json.load(open(base_dir + 'mazda_export.json', 'r')) mu.login() mu.upload(data) if __name__ == '__main__': main()