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/mazda/"


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


def token_load() -> Token:
    try:
        with open(base_dir + "temp/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 + "data/Workshop_Order_Report.csv", base_dir + "temp/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=("config/cert/cert.pem", "config/cert/key.pem"), debug=True)