mazda/nasa_export.py

import pandas as pd
from sqlalchemy import create_engine


def conn_string(dsn: dict[str, str]):
    return f"mssql+pyodbc://{dsn['user']}:{dsn['pass']}@{dsn['server']}/{dsn['database']}?driver=SQL+Server+Native+Client+11.0"


def load_data(config: dict[str, str], source: str, period: str):
    year = period[:4]
    month = period[4:6]

    select_befehl_auftraege = f"SELECT * FROM [Auftraege_NASA_gruppiert] WHERE Periode = '{period}'"
    select_befehl_mitarbeiter = f"SELECT * FROM [Mitarbeiter_NASA] WHERE Periode = '{period}'"

    source_auftraege = "data/Auftraege_NASA_gruppiert.csv"
    source_mitarbeiter = "data/Mitarbeiter_NASA.csv"

    payload = {
        "HaendlerNr": config["client_id"],
        "Filiale": config["client_id"],
        "Jahr": year,
        "Monat": month,
        "Fabrikat": "Mazda",
        "AnzahlMitarbeiter": 0,
        "AnzahlProduktiv": 0.0,
        "WerkstattDurchlaeufe": 0,
        "Token": config["credentials"]["token"],
    }

    if source == "database":
        source_db = create_engine(conn_string(config["source_dsn"]))
        df = pd.read_sql(select_befehl_auftraege, con=source_db)
        rename_from = ["AuftragsArt", "AuftragsTyp"]
        rename_to = ["AuftragsArtId_Name", "AuftragsArt"]
        df = df.rename(columns=dict(zip(rename_from, rename_to)))

    else:
        df = pd.read_csv(source_auftraege, sep=";", encoding="latin-1", decimal=",")
        df = df[df["Periode"] == period]

    # AuftragsArt = ["Inspektion", "Karosseriearbeit", "Lackierung", "Verschleißteile", "Sonstiges"]
    # AuftragsArtId = {"1": "Extern", "2": "Garantie", "3": "Intern", "4": "Theke"]

    columns = [
        "AuftragsArt",
        "AuftragsArtId",
        "TeileUmsatz",
        "LohnUmsatz",
        "SonstigeUmsatz",
        "GesamtUmsatz",
        "AnzahlAuftraege",
    ]

    df = df[columns]

    df.to_csv(
        f"{config['export_dir']}/csv/{period}_auftraege.csv",
        sep=";",
        encoding="latin-1",
        decimal=",",
        index=False,
    )

    payload["WerkstattDurchlaeufe"] = int(df["AnzahlAuftraege"].sum())
    payload["AfterSalesPositionen"] = df.to_dict("records")

    # Mitarbeiter gesamt und produktiv
    if source == "database":
        df = pd.read_sql(select_befehl_mitarbeiter, con=source_db)
    else:
        df = pd.read_csv(source_mitarbeiter, sep=";", encoding="latin-1", decimal=",")

    df.to_csv(
        f"{config['export_dir']}/csv/{period}_mitarbeiter.csv",
        sep=";",
        encoding="latin-1",
        decimal=",",
        index=False,
    )

    payload["AnzahlMitarbeiter"] = df.shape[0]
    payload["AnzahlProduktiv"] = int(df["produktiv"].sum())
    return payload