{
 "nbformat": 4,
 "nbformat_minor": 5,
 "metadata": {
  "language_info": {
   "name": "python"
  },
  "kernel_info": {
   "name": "synapse_pyspark"
  },
  "kernelspec": {
   "name": "synapse_pyspark",
   "display_name": "Synapse PySpark"
  },
  "microsoft": {
   "language": "python",
   "language_group": "synapse_pyspark"
  }
 },
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Nordvella: z warstwy brązowej do srebrnej i złotej\n",
    "Notes czyta surowy plik `Files/surowe/sprzedaz.csv`, zapisuje oczyszczoną tabelę `sprzedaz` (warstwa srebrna)\n",
    "i tabelę `sprzedaz_miesieczna` z sumami po miesiącach, regionach i kategoriach (warstwa złota).\n",
    "Domyślny lakehouse: `nordvella_lakehouse`."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "# Warstwa brązowa: surowy plik CSV wgrany do folderu Files/surowe\n",
    "sales_raw = (spark.read\n",
    "    .option(\"header\", True)\n",
    "    .option(\"inferSchema\", True)\n",
    "    .csv(\"Files/surowe/sprzedaz.csv\"))\n",
    "\n",
    "print(f\"{sales_raw.count():,} wierszy\".replace(\",\", \" \"))\n",
    "sales_raw.printSchema()"
   ],
   "execution_count": null,
   "outputs": []
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "from pyspark.sql import functions as F\n",
    "\n",
    "# Warstwa srebrna: data zamiast klucza liczbowego, bez transakcji, które klient zwrócił\n",
    "sales = (sales_raw\n",
    "    .withColumn(\"Data\", F.to_date(F.col(\"DataID\").cast(\"string\"), \"yyyyMMdd\"))\n",
    "    .filter(F.col(\"CzyZwrot\") == 0)\n",
    "    .drop(\"CzyZwrot\"))\n",
    "\n",
    "sales.write.mode(\"overwrite\").format(\"delta\").saveAsTable(\"sprzedaz\")\n",
    "print(f\"Tabela sprzedaz: {sales.count():,} wierszy\".replace(\",\", \" \"))"
   ],
   "execution_count": null,
   "outputs": []
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "# Warstwa złota: sprzedaż netto, marża i liczba transakcji po miesiącach, regionach i kategoriach\n",
    "stores = spark.read.table(\"sklepy\").select(\"SklepID\", \"Region\")\n",
    "products = spark.read.table(\"produkty\").select(\"ProduktID\", \"Kategoria\")\n",
    "\n",
    "gold = (spark.read.table(\"sprzedaz\")\n",
    "    .join(stores, \"SklepID\")\n",
    "    .join(products, \"ProduktID\")\n",
    "    .withColumn(\"Miesiac\", F.trunc(\"Data\", \"month\"))\n",
    "    .groupBy(\"Miesiac\", \"Region\", \"Kategoria\")\n",
    "    .agg(F.round(F.sum(\"WartoscNetto\"), 2).alias(\"SprzedazNetto\"),\n",
    "         F.round(F.sum(\"MarzaNetto\"), 2).alias(\"Marza\"),\n",
    "         F.count(\"*\").alias(\"Transakcje\")))\n",
    "\n",
    "gold.write.mode(\"overwrite\").format(\"delta\").saveAsTable(\"sprzedaz_miesieczna\")\n",
    "display(gold.orderBy(F.desc(\"Miesiac\"), F.desc(\"SprzedazNetto\")))"
   ],
   "execution_count": null,
   "outputs": []
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Sprzedaż netto w kolejnych miesiącach, osobna linia dla każdego regionu\n",
    "pdf = (spark.read.table(\"sprzedaz_miesieczna\")\n",
    "    .groupBy(\"Miesiac\", \"Region\").agg(F.sum(\"SprzedazNetto\").alias(\"SprzedazNetto\"))\n",
    "    .toPandas()\n",
    "    .pivot(index=\"Miesiac\", columns=\"Region\", values=\"SprzedazNetto\")\n",
    "    .sort_index())\n",
    "\n",
    "ax = (pdf / 1e6).plot(figsize=(11, 4.5), linewidth=2)\n",
    "ax.set_title(\"Nordvella: sprzedaż netto miesięcznie według regionów (mln zł)\")\n",
    "ax.set_xlabel(\"\")\n",
    "ax.grid(alpha=0.3)\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ],
   "execution_count": null,
   "outputs": []
  }
 ]
}