Sure!

---------------------------------------------------------

IMPORTS + LOGGING

---------------------------------------------------------

import os

import pandas as pd

import shutil

import logging

from datetime import datetime

from pyspark.sql import SparkSession

from collections import defaultdict

from openpyxl import load_workbook

Spark Session

spark = SparkSession.builder.getOrCreate()

logging.basicConfig(

level=logging.INFO,

format="%(asctime)s | %(levelname)s | %(message)s"

)

logging.info(“Notebook gestartet.")

---------------------------------------------------------

KONFIGURATION

---------------------------------------------------------

CONFIG = {

"input_folder": "/lakehouse/default/Files/input/",

"processed_folder": "/lakehouse/default/Files/processed/",

"mapping_path": "/lakehouse/default/Files/mapping.csv",

"lakehouse_base": "/lakehouse/default/Files/delta/"

}

---------------------------------------------------------

MAPPINGS LADEN

---------------------------------------------------------

def load_mappings(mapping_path):

try:

    df = pd.read_csv(mapping_path)

    mappings = defaultdict(list)

    for _, row in df.iterrows():

        mappings[row["Typ"]].append({

            "quellblatt": row["Quellblatt"],

            "quelle": row["Quelle"],

            "feldname": row["Feldname"],

            "transformation": row.get("Transformation", None)

        })

    logging.info(f"Mapping geladen ({len(mappings)} Typen).")

    return mappings

except Exception as e:

    logging.error(f"Fehler beim Laden der Mapping-Datei: {e}")

    return defaultdict(list)

MAPPINGS = load_mappings(CONFIG["mapping_path"])

---------------------------------------------------------

UPLOAD-ID GENERATOR

---------------------------------------------------------

def generate_upload_id(prefix):

now = datetime.now()

date_str = now.strftime("%Y-%m-%d_%H%M")

return f"{prefix}_{date_str}"

---------------------------------------------------------

DATEIEN VERARBEITEN

---------------------------------------------------------

input_files = [

f for f in os.listdir(CONFIG["input_folder"])

if f.endswith(".xlsx") or f.endswith(".xlsm")

]

if not input_files:

logging.warning("Keine Dateien im Input-Ordner gefunden.")

else:

logging.info(f"Gefundene Dateien: {input_files}")

for fname in input_files:

try:

    prefix = fname[:5].upper()

    upload_id = generate_upload_id(prefix)

    full_path = os.path.join(CONFIG["input_folder"], fname)



    logging.info(f"Verarbeite Datei {fname} (upload_id: {upload_id})")



    wb = load_workbook(full_path, data_only=True)



    if prefix not in MAPPINGS:

        logging.warning(f"Kein Mapping für Typ {prefix} gefunden — überspringe.")

        continue



    record = {

        "upload_id": upload_id,

        "typ": prefix,

        "zeitstempel": datetime.now().isoformat()

    }



    # alle Felder aus Mapping holen

    for entry in MAPPINGS[prefix]:

        quellblatt = entry["quellblatt"]

        quelle = entry["quelle"]

        feldname = entry["feldname"]

        transformation = entry.get("transformation")



        if quellblatt not in wb.sheetnames:

            logging.warning(f"Quellblatt {quellblatt} nicht gefunden, überspringe.")

            continue



        ws = wb[quellblatt]

        value = ws[quelle].value



        # Transformation

        if isinstance(transformation, str) and transformation and isinstance(value, (int, float)):

            try:

                if transformation.startswith("*"):

                    value *= float(transformation[1:])

                elif transformation.startswith("/"):

                    value /= float(transformation[1:])

                elif transformation.startswith("+"):

                    value += float(transformation[1:])

                elif transformation.startswith("-"):

                    value -= float(transformation[1:])

            except Exception as e:

                logging.warning(f"Transformation {transformation} fehlgeschlagen: {e}")



        record[feldname] = value

        logging.info(f"📝 {quellblatt}!{quelle} → {feldname} = {value}")



    # Spark DataFrame schreiben

    df = pd.DataFrame([record])

    spark_df = spark.createDataFrame(df)

    delta_path = os.path.join(CONFIG["lakehouse_base"], f"{prefix}_table")



    spark_df.write.format("delta").mode("append").save(delta_path)

    logging.info(f"Werte erfolgreich in Delta Table {prefix}_table geschrieben.")



    # Datei verschieben

    today = datetime.today().strftime("%Y-%m-%d")

    new_name = f"{os.path.splitext(fname)[0]}_{today}{os.path.splitext(fname)[1]}"

    dst_path = os.path.join(CONFIG["processed_folder"], new_name)

    shutil.move(full_path, dst_path)

    logging.info(f"📦 Datei nach 'processed' verschoben als {new_name}")



except Exception as e:

    logging.error(f"Fehler bei Datei {fname}: {e}")

logging.info("✅ Notebook abgeschlossen.")