mazda/mazda_webservice.py

import json
import os
from dataclasses import dataclass

from flask import Flask, redirect, request, session
from oauthlib.oauth2.rfc6749.errors import OAuth2Error
from requests_oauthlib import OAuth2Session

import mazda_upload

# from datetime import datetime


@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__)

PROD = True

cfg: mazda_upload.MazdaConfig = mazda_upload.cfg_prod if PROD else mazda_upload.cfg_debug
base_dir = "C:/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 home():
    token = get_token()

    if token is None:
        return redirect("/login")

    oauth = OAuth2Session(cfg.client_id, token=token, auto_refresh_kwargs={"grant_type": "refresh_token"})

    if oauth.authorized:
        token_save(oauth.token)
    else:
        return redirect("/refresh")

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

    return redirect("/login")


@app.route("/refresh")
def refresh():
    token = get_token()
    oauth = OAuth2Session(cfg.client_id)
    new_token = oauth.refresh_token(token.iss, token.refresh_token)

    if new_token.authorized:
        token_save(new_token)

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


@app.route("/login")
def login():
    redirect_uri = request.base_url.rsplit("/", 1)[0] + "/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=get_token(), auto_refresh_kwargs={"grant_type": "refresh_token"})

    if not oauth.authorized:
        return redirect("/")

    try:
        res = mazda_upload.upload_files(cfg, oauth)
    except OAuth2Error as e:
        print(e.error, e.description)
        return redirect("/")
    return app.response_class(response=json.dumps(res, 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/global-cube.com.crt",
            "config/cert/global-cube.com.key",
            # "config/cert/intermediate.crt",
        ),
        debug=True,
    )