from pyspark.sql import functions as F

# Fabric: lakehouse.schemat, Databricks: katalog.schemat
PREFIKS = "workspace.nordvella_fvd"

sprzedaz = spark.read.table(f"{PREFIKS}.sprzedaz")
kalendarz = spark.read.table(f"{PREFIKS}.kalendarz")

miesiecznie = (
    sprzedaz.filter(F.col("CzyZwrot") == 0)
    .join(kalendarz.select("DataID", "Rok", "Miesiac", "RokMiesiac"), "DataID")
    .groupBy("Rok", "Miesiac", "RokMiesiac")
    .agg(
        F.round(F.sum("WartoscNetto"), 2).alias("SprzedazNetto"),
        F.round(F.sum("MarzaNetto"), 2).alias("Marza"),
        F.count("*").alias("Transakcje"),
    )
    .orderBy("Rok", "Miesiac")
)

miesiecznie.write.mode("overwrite").saveAsTable(f"{PREFIKS}.sprzedaz_miesieczna")
display(miesiecznie)