Python-Projekte/mazda/mazda_webservice.py

from dataclasses import dataclass
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2.rfc6749.errors import OAuth2Error
from flask import Flask, redirect, request, session
from datetime import datetime
import json
import os
import mazda_upload


@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


app = Flask(__name__)

cfg: mazda_upload.MazdaConfig = mazda_upload.cfg
base_dir = '/home/robert/projekte/python/mazda/'


def token_save(token):
    session['oauth_token'] = token
    with open(base_dir + 'token.json', 'w') as fwh:
        json.dump(token, fwh, indent=2)


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


def get_token() -> Token:
    if session.get('oauth_token') is None:
        session['oauth_token'] = token_load()

        if session['oauth_token'] is not None and session['oauth_token']['expires_at'] < datetime.now().timestamp():
            session['oauth_token'] = None
    return session['oauth_token']


@app.route("/")
def demo():
    # if token := get_token() is not None:
    oauth = OAuth2Session(cfg.client_id, token=get_token())

    if oauth.authorized:
        return redirect('/profile')

    redirect_uri = request.base_url + 'callback'
    print(redirect_uri)
    oauth = OAuth2Session(cfg.client_id, redirect_uri=redirect_uri)
    authorization_url, state = oauth.authorization_url(cfg.domain + cfg.auth_url)
    session['oauth_state'] = state
    return redirect(authorization_url)


@app.route("/callback", methods=["GET"])
def callback():
    state = session.get('oauth_state')
    if state is None:
        return redirect('/')
    print(state)
    redirect_uri = request.base_url
    print(redirect_uri)
    oauth = OAuth2Session(cfg.client_id, state=state, redirect_uri=redirect_uri)
    try:
        token = oauth.fetch_token(cfg.domain + cfg.token_url, client_secret=cfg.client_secret, authorization_response=request.url)
        token_save(token)
        session['oauth_token'] = token
        return redirect('/profile')
    except OAuth2Error as e:
        print(e)
        return redirect('/')


@app.route("/profile", methods=["GET"])
def profile():
    oauth = OAuth2Session(cfg.client_id, token=session.get('oauth_token'))
    data = mazda_upload.convert_csv(base_dir + 'Workshop_Order_Report.csv', base_dir + 'mazda_export.json', 2021, 6)
    mazda_upload.upload(oauth, data)
    return app.response_class(
        response=json.dumps(data, indent=2),
        mimetype='application/json'
    )


if __name__ == "__main__":
    # os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = "1"
    app.secret_key = os.urandom(24)
    app.run(host='0.0.0.0', port='8085', ssl_context=('cert.pem', 'key.pem'), debug=True)