kmftools/delfland/ppo-insight/ppo_insight.py

import os
import io
from datetime import datetime

import pandas as pd
from dotenv import load_dotenv
from flask import Flask, request, render_template, redirect, url_for, flash, Response
from werkzeug.utils import secure_filename

from database import get_db_connection

env_path = "./.env"
load_dotenv(env_path)

UPLOAD_FOLDER = os.getenv("UPLOAD_FOLDER")
ALLOWED_EXTENSIONS = {"xlsx"}

app = Flask(__name__)
app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
app.secret_key = "9f2a1d3e4c5b6a7d8e9f0a1b2c3d4e5f"


def allowed_file(filename):
    return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS


@app.route("/", methods=["GET", "POST"])
def upload_file():
    if request.method == "POST":
        if "file" not in request.files:
            flash("Geen bestand geselecteerd", "danger")
            return redirect(request.url)

        file = request.files["file"]
        if file.filename == "":
            flash("Geen bestand geselecteerd", "danger")
            return redirect(request.url)

        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            filepath = os.path.join(app.config["UPLOAD_FOLDER"], filename)
            file.save(filepath)
            process_excel(filepath)
            flash(
                "Bestand geüpload en verwerkt! <a href='" +
                url_for("show_result") + "'>Bekijk de resultaten</a>",
                "success",
            )
            return redirect(url_for("upload_file"))

    return render_template("index.html")


@app.route("/extrapolate", methods=["GET", "POST"])
def show_result():
    x_str = request.form.get("selected_date")
    print(x_str)
    try:
        if x_str is not None:
            x = datetime.strptime(x_str, "%Y-%m-%d")
        else:
            # Handle the case where x_str is None, for example:
            x = None  # or raise an exception or use a default date
            flash("Ongeldige datum ingevoerd.", "danger")
    except ValueError:
        flash("Ongeldige datum ingevoerd.", "danger")
        return redirect(url_for("upload_file"))

    result = getextrapolation(x)
    return render_template("resultaat.html", result=result, selected_date=x_str)


@app.route("/download", methods=["GET"])
def download_excel():
    x_str = request.args.get("x")
    x = datetime.strptime(x_str, "%Y-%m-%d")
    result = getextrapolation(x)

    df = pd.DataFrame(result)
    output = io.BytesIO()
    with pd.ExcelWriter(output, engine="openpyxl") as writer:
        df.to_excel(writer, index=False,
                    sheet_name=f"Extrapolatie_{x.strftime('%Y-%m-%d')}")
    output.seek(0)

    filename = f"extrapolatie_{x.strftime('%Y-%m-%d')}.xlsx"
    return Response(
        output,
        mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename={filename}"},
    )


def extrapolate_vervaldatum(vervaldatum, uom, frequentie, x):
    if vervaldatum is None or pd.isna(vervaldatum):
        print("Ongeldige vervaldatum:", vervaldatum)
        return []  # Geef een lege lijst terug in plaats van een fout te veroorzaken

    vervaldatum = pd.to_datetime(str(vervaldatum), errors="coerce")

    if pd.isna(vervaldatum):  # Extra controle als de conversie mislukt
        print("Kon datum niet omzetten:", vervaldatum)
        return []

    result = []

    while vervaldatum <= x:
        result.append(vervaldatum.strftime("%Y-%m-%d"))

        if uom == "D":
            vervaldatum += pd.DateOffset(days=frequentie)
        elif uom == "W":
            vervaldatum += pd.DateOffset(weeks=frequentie)
        elif uom == "M":
            vervaldatum += pd.DateOffset(months=frequentie)
        elif uom == "Y":
            vervaldatum += pd.DateOffset(years=frequentie)
        else:
            print("Onbekende eenheid:", uom)
            break  # Voorkom oneindige lus bij een onbekende eenheid

    return result


def getextrapolation(x):
    if x is None:
        print("Geen geldige datum opgegeven voor extrapolatie.")
        return []
    conn, cursor = get_db_connection()
    cursor.execute("""
        SELECT Object_code, Object_omschrijving, Afdeling_object, PO_Code, Omschrijving_PO, Vervaldatum, Frequentie, UOM, 
               PO_schema_Niet_gebruikt, Cluster, Locatie, Locatie_omschrijving, Klasse_object, Categorie
        FROM onderhoud
    """)

    columns = [desc[0] for desc in cursor.description]
    rows = cursor.fetchall()
    cursor.close()
    conn.close()

    raw_data = [dict(zip(columns, row)) for row in rows]
    result = []

    for row in raw_data:
        vervaldatum = row.get("Vervaldatum")
        uom = row.get("UOM")
        frequentie = row.get("Frequentie", 0)

        if vervaldatum:
            extrapolated_dates = extrapolate_vervaldatum(
                vervaldatum, uom, frequentie, x)
            for date in extrapolated_dates:
                new_row = row.copy()
                new_row["Vervaldatum"] = date
                result.append(new_row)

    return result


if __name__ == "__main__":
    os.makedirs(UPLOAD_FOLDER, exist_ok=True)
    app.run(debug=True)