Supply Chain Alimentare: ETL Pattern da Farm a Retailer

# Architettura logica della supply chain alimentare
# (dati generati ad ogni step)

FARM SYSTEMS
    |-- Sensori IoT (MQTT): temp suolo, umidita, pH, EC
    |-- Sistema ERP agricolo: colture pianificate, costi input
    |-- Registro trattamenti fitosanitari (LIMS)
    |-- Geo-dati GPS macchinari agricoli
    |-- API meteo (OpenMeteo, Copernicus)
    |
    v
PROCESSING PLANTS (Centri di Lavorazione)
    |-- MES (Manufacturing Execution System): ordini produzione, rese
    |-- LIMS: analisi chimiche, microbiologiche, allergeni
    |-- ERP SAP/Oracle: distinta base, lotti, tracciabilita
    |-- WMS (Warehouse Management): magazzino materie prime/finiti
    |-- Sensori cold chain: temperatura/umidita durante lavorazione
    |
    v
DISTRIBUTION CENTERS
    |-- WMS: gestione stock, picking, slotting
    |-- TMS (Transport Management): spedizioni, veicoli, routing
    |-- Sensori cold chain: logger temperatura nei camion frigoriferi
    |-- EDI: ordini da retailer (EDIFACT ORDERS, DESADV)
    |-- Tracker GPS: posizione veicoli in tempo reale
    |
    v
RETAIL POS
    |-- POS data: vendite per SKU, store, ora
    |-- Gestione shelf life: prodotti prossimi scadenza
    |-- EDI: conferme ricezione (RECADV), fatture (INVOIC)
    |-- Inventario store: stock disponibile per punto vendita
    |
    v
CONSUMER (segnali indiretti)
    |-- App loyalty: acquisti per cliente, frequenza
    |-- Reviews prodotto: qualità percepita
    |-- Social sentiment: trend di consumo

# Architettura Medallion per Food Supply Chain

BRONZE LAYER (Raw, immutabile)
    |-- Tutti i dati sorgente caricati as-is
    |-- Partitioned by: source_system, ingestion_date
    |-- Retention: 7 anni (compliance FSMA/EU)
    |-- Formato: Parquet su S3/GCS o Delta Lake
    |
    v
SILVER LAYER (Cleansed, deduplicato)
    |-- Dbt models: normalizzazione schemi eterogenei
    |-- Deduplicazione lotti, SKU, location codes
    |-- Type casting: date, temperature (C/F), pesi (kg/lb)
    |-- NULL handling: sensori offline, ERP null fields
    |-- PII masking: dati operatori, conducenti
    |
    v
GOLD LAYER (KPI pronti per business)
    |-- Aggregazioni: fill rate, waste rate, OTIF per SKU
    |-- Metriche cold chain: violazioni temperatura per lotto
    |-- Dashboard BI: Power BI, Tableau, Grafana
    |-- API ML: feature store per demand forecasting
    |
    v
DATA MART
    |-- Retail mart: sell-through, shelf life at receipt
    |-- Operations mart: OEE impianti, rendimenti produzione
    |-- Finance mart: costo per lotto, margine per SKU
    |-- Compliance mart: tracciabilita FSMA, recall readiness

# food_supply_chain_dag.py
from __future__ import annotations

import logging
from datetime import datetime, timedelta
from typing import Any

from airflow import DAG
from airflow.decorators import task
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.http.sensors.http import HttpSensor
from airflow.utils.trigger_rule import TriggerRule

logger = logging.getLogger(__name__)

DEFAULT_ARGS = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
}

with DAG(
    dag_id="food_supply_chain_daily_etl",
    description="Pipeline ETL giornaliera: farm -> processing -> distribution -> retail",
    default_args=DEFAULT_ARGS,
    schedule_interval="0 2 * * *",  # ogni notte alle 02:00 UTC
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=["foodtech", "supply-chain", "etl", "production"],
    max_active_runs=1,
    doc_md="""
    ## Food Supply Chain Daily ETL
    Estrae dati da: SAP ERP, MES impianti, WMS, TMS, POS retail.
    Carica su Bronze S3, poi lancia dbt per Silver e Gold.
    """,
) as dag:

    # ----------------------------------------------------------------
    # STEP 1: Health checks sulle sorgenti
    # ----------------------------------------------------------------
    check_sap_available = HttpSensor(
        task_id="check_sap_api_available",
        http_conn_id="sap_erp_api",
        endpoint="/api/health",
        timeout=30,
        poke_interval=60,
        mode="reschedule",
        soft_fail=False,
    )

    check_mes_available = HttpSensor(
        task_id="check_mes_api_available",
        http_conn_id="mes_plant_api",
        endpoint="/health",
        timeout=30,
        poke_interval=60,
        mode="reschedule",
    )

    # ----------------------------------------------------------------
    # STEP 2: Estrazione SAP ERP (lotti, ordini, inventario)
    # ----------------------------------------------------------------
    @task(task_id="extract_sap_erp", retries=3)
    def extract_sap_erp(**context: Any) -> dict:
        """Estrae dati da SAP ERP via RFC/BAPI wrapper REST."""
        from src.connectors.sap_connector import SAPConnector

        execution_date = context["ds"]  # 'YYYY-MM-DD'
        sap = SAPConnector(conn_id="sap_erp_api")

        # Estrai: lotti produzione del giorno precedente
        batches = sap.get_production_batches(
            date=execution_date,
            plants=["PL001", "PL002", "PL003"],
        )

        # Estrai: movimenti magazzino
        stock_movements = sap.get_stock_movements(
            date=execution_date,
            movement_types=["101", "261", "501", "601"],
        )

        # Estrai: ordini di vendita confermati
        sales_orders = sap.get_sales_orders(
            date=execution_date,
            status=["CONF", "DLVR"],
        )

        # Salva su S3 Bronze layer
        s3 = S3Hook(aws_conn_id="aws_s3")
        prefix = f"bronze/sap_erp/dt={execution_date}"

        s3.load_string(
            string_data=batches.to_json(orient="records", lines=True),
            key=f"{prefix}/batches.jsonl",
            bucket_name="food-data-lake",
            replace=True,
        )
        s3.load_string(
            string_data=stock_movements.to_json(orient="records", lines=True),
            key=f"{prefix}/stock_movements.jsonl",
            bucket_name="food-data-lake",
            replace=True,
        )

        logger.info(
            "SAP extraction OK: %d batches, %d movements, %d orders",
            len(batches),
            len(stock_movements),
            len(sales_orders),
        )
        return {
            "batches_count": len(batches),
            "movements_count": len(stock_movements),
            "orders_count": len(sales_orders),
        }

    # ----------------------------------------------------------------
    # STEP 3: Estrazione dati cold chain (IoT sensori temperatura)
    # ----------------------------------------------------------------
    @task(task_id="extract_cold_chain_iot")
    def extract_cold_chain_iot(**context: Any) -> dict:
        """Estrae letture temperatura da InfluxDB (IoT cold chain)."""
        from influxdb_client import InfluxDBClient

        execution_date = context["ds"]

        client = InfluxDBClient(
            url="{{ var.value.influxdb_url }}",
            token="{{ var.value.influxdb_token }}",
            org="foodtech-org",
        )
        query_api = client.query_api()

        # Query Flux: temperature readings per lotto di trasporto
        flux_query = f"""
        from(bucket: "cold-chain")
            |> range(start: {execution_date}T00:00:00Z,
                       stop: {execution_date}T23:59:59Z)
            |> filter(fn: (r) => r._measurement == "temperature")
            |> filter(fn: (r) => r._field == "celsius")
            |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
            |> pivot(rowKey: ["_time", "lot_id"], columnKey: ["_field"], valueColumn: "_value")
        """

        result = query_api.query_data_frame(flux_query)

        # Calcola violazioni: temperatura fuori range per categoria prodotto
        violations = result[
            (result["celsius"] < result["min_temp_required"]) |
            (result["celsius"] > result["max_temp_required"])
        ]

        s3 = S3Hook(aws_conn_id="aws_s3")
        prefix = f"bronze/cold_chain/dt={execution_date}"

        s3.load_string(
            string_data=result.to_json(orient="records", lines=True),
            key=f"{prefix}/temperature_readings.jsonl",
            bucket_name="food-data-lake",
            replace=True,
        )

        logger.info(
            "Cold chain extraction OK: %d readings, %d violations",
            len(result),
            len(violations),
        )

        if len(violations) > 0:
            logger.warning(
                "COLD CHAIN VIOLATIONS: %d eventi fuori temperatura!",
                len(violations),
            )

        return {
            "readings_count": len(result),
            "violations_count": len(violations),
        }

    # ----------------------------------------------------------------
    # STEP 4: Estrazione EDI dai partner retailer
    # ----------------------------------------------------------------
    @task(task_id="extract_edi_partners")
    def extract_edi_partners(**context: Any) -> dict:
        """Processa file EDI EDIFACT da SFTP partners."""
        from src.connectors.edi_connector import EDIFACTConnector

        execution_date = context["ds"]
        edi = EDIFACTConnector(sftp_conn_id="edi_sftp_partners")

        # Download e parsing ORDERS (EDIFACT ORDERS D.01B)
        orders = edi.process_orders(date=execution_date)

        # Download e parsing DESADV (Despatch Advice)
        despatch_advices = edi.process_desadv(date=execution_date)

        # Download e parsing RECADV (Receiving Advice)
        receiving_advices = edi.process_recadv(date=execution_date)

        s3 = S3Hook(aws_conn_id="aws_s3")
        prefix = f"bronze/edi_partners/dt={execution_date}"

        s3.load_string(
            string_data=orders.to_json(orient="records", lines=True),
            key=f"{prefix}/orders.jsonl",
            bucket_name="food-data-lake",
            replace=True,
        )

        return {
            "orders_count": len(orders),
            "desadv_count": len(despatch_advices),
            "recadv_count": len(receiving_advices),
        }

    # ----------------------------------------------------------------
    # STEP 5: Estrazione POS retail
    # ----------------------------------------------------------------
    @task(task_id="extract_pos_data")
    def extract_pos_data(**context: Any) -> dict:
        """Estrae dati vendite POS da database retailer partner."""
        from src.connectors.retail_connector import RetailPOSConnector

        execution_date = context["ds"]
        pos = RetailPOSConnector(conn_id="retail_pos_db")

        # Vendite per SKU/store con shelf life info
        sales = pos.get_daily_sales(
            date=execution_date,
            include_promotions=True,
            include_markdowns=True,
        )

        # Near-expiry: prodotti a <3 giorni scadenza ancora in shelf
        near_expiry = pos.get_near_expiry_stock(
            date=execution_date,
            days_threshold=3,
        )

        s3 = S3Hook(aws_conn_id="aws_s3")
        prefix = f"bronze/pos_retail/dt={execution_date}"

        s3.load_string(
            string_data=sales.to_json(orient="records", lines=True),
            key=f"{prefix}/sales.jsonl",
            bucket_name="food-data-lake",
            replace=True,
        )

        logger.info("POS extraction OK: %d transazioni", len(sales))
        return {
            "sales_count": len(sales),
            "near_expiry_skus": len(near_expiry),
        }

    # ----------------------------------------------------------------
    # STEP 6: dbt run (Silver + Gold transformations)
    # ----------------------------------------------------------------
    dbt_run_silver = BashOperator(
        task_id="dbt_run_silver_layer",
        bash_command=(
            "cd /opt/dbt/food_supply_chain && "
            "dbt run --select tag:silver --target prod "
            "--vars '{\"execution_date\": \"{{ ds }}\", \"source_env\": \"prod\"}'"
        ),
        retries=2,
    )

    dbt_run_gold = BashOperator(
        task_id="dbt_run_gold_layer",
        bash_command=(
            "cd /opt/dbt/food_supply_chain && "
            "dbt run --select tag:gold --target prod "
            "--vars '{\"execution_date\": \"{{ ds }}\", \"source_env\": \"prod\"}'"
        ),
        retries=2,
    )

    # ----------------------------------------------------------------
    # STEP 7: dbt test (data quality checks)
    # ----------------------------------------------------------------
    dbt_test = BashOperator(
        task_id="dbt_test_all",
        bash_command=(
            "cd /opt/dbt/food_supply_chain && "
            "dbt test --select tag:silver tag:gold --target prod"
        ),
        trigger_rule=TriggerRule.ALL_SUCCESS,
    )

    # ----------------------------------------------------------------
    # STEP 8: Great Expectations validation
    # ----------------------------------------------------------------
    @task(task_id="run_great_expectations_validation")
    def run_ge_validation(**context: Any) -> dict:
        """Lancia Great Expectations per validare data quality cold chain."""
        from great_expectations.data_context import get_context

        context_ge = get_context(project_root_dir="/opt/great_expectations")
        execution_date = context["ds"]

        results = {}
        suites_to_run = [
            ("cold_chain_temperature", "bronze_cold_chain_checkpoint"),
            ("lot_traceability", "silver_lots_checkpoint"),
            ("kpi_supply_chain", "gold_kpi_checkpoint"),
        ]

        for suite_name, checkpoint_name in suites_to_run:
            result = context_ge.run_checkpoint(
                checkpoint_name=checkpoint_name,
                batch_request={
                    "runtime_parameters": {"path": f"s3://food-data-lake/bronze/cold_chain/dt={execution_date}"},
                    "batch_identifiers": {"execution_date": execution_date},
                },
            )
            results[suite_name] = result.success

        failed = [name for name, ok in results.items() if not ok]
        if failed:
            raise ValueError(f"GE validation FAILED per suite: {failed}")

        logger.info("Great Expectations: tutte le suite OK")
        return results

    # ----------------------------------------------------------------
    # STEP 9: KPI alert (email se waste rate > soglia)
    # ----------------------------------------------------------------
    @task(task_id="check_kpi_alerts", trigger_rule=TriggerRule.ALL_SUCCESS)
    def check_kpi_alerts(**context: Any) -> None:
        """Controlla KPI critici e invia alert se fuori soglia."""
        from src.services.kpi_alert_service import KPIAlertService

        execution_date = context["ds"]
        alert_service = KPIAlertService(conn_id="snowflake_dw")

        kpis = alert_service.get_daily_kpis(date=execution_date)

        alerts_sent = []

        # Waste rate > 5%: alert critico
        if kpis["waste_rate_pct"] > 5.0:
            alert_service.send_alert(
                severity="HIGH",
                message=f"Waste rate {kpis['waste_rate_pct']:.1f}% supera soglia 5%",
                kpi="waste_rate",
                value=kpis["waste_rate_pct"],
            )
            alerts_sent.append("waste_rate")

        # OTIF < 95%: alert warning
        if kpis["otif_pct"] < 95.0:
            alert_service.send_alert(
                severity="MEDIUM",
                message=f"OTIF {kpis['otif_pct']:.1f}% sotto soglia 95%",
                kpi="otif",
                value=kpis["otif_pct"],
            )
            alerts_sent.append("otif")

        # Cold chain violations > 0: alert immediato
        if kpis["cold_chain_violations"] > 0:
            alert_service.send_alert(
                severity="CRITICAL",
                message=f"{kpis['cold_chain_violations']} violazioni temperatura cold chain!",
                kpi="cold_chain_violations",
                value=kpis["cold_chain_violations"],
            )
            alerts_sent.append("cold_chain")

        logger.info("KPI check completato. Alert inviati: %s", alerts_sent)

    # ----------------------------------------------------------------
    # DAG wiring: dipendenze tra task
    # ----------------------------------------------------------------
    extract_sap = extract_sap_erp()
    extract_cold = extract_cold_chain_iot()
    extract_edi = extract_edi_partners()
    extract_pos = extract_pos_data()
    ge_validation = run_ge_validation()
    kpi_alerts = check_kpi_alerts()

    # Health checks prima di tutto
    [check_sap_available, check_mes_available] >> extract_sap

    # Estrazione parallela di tutte le sorgenti
    [extract_sap, extract_cold, extract_edi, extract_pos] >> dbt_run_silver

    # Silver prima di Gold
    dbt_run_silver >> dbt_run_gold

    # Test dopo Gold
    dbt_run_gold >> dbt_test

    # GE validation e KPI alerts dopo i test
    dbt_test >> ge_validation >> kpi_alerts

# models/silver/schema.yml
version: 2

models:
  - name: silver_lots
    description: "Lotti di produzione normalizzati da SAP ERP"
    config:
      tags: ["silver", "lots", "traceability"]
      materialized: incremental
      incremental_strategy: merge
      unique_key: lot_id
    columns:
      - name: lot_id
        description: "Identificativo univoco lotto (GS1 SGTIN)"
        tests:
          - not_null
          - unique
      - name: production_date
        tests:
          - not_null
      - name: expiry_date
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "expiry_date > production_date"
      - name: plant_code
        tests:
          - not_null
          - accepted_values:
              values: ["PL001", "PL002", "PL003", "PL004"]
      - name: temperature_class
        tests:
          - accepted_values:
              values: ["ambient", "chilled", "frozen", "ultra_frozen"]
      - name: net_weight_kg
        tests:
          - dbt_utils.expression_is_true:
              expression: "net_weight_kg > 0 AND net_weight_kg < 50000"

  - name: silver_shipments
    description: "Spedizioni normalizzate da TMS e EDI DESADV"
    config:
      tags: ["silver", "logistics"]
      materialized: incremental
      unique_key: shipment_id
    columns:
      - name: shipment_id
        tests: [not_null, unique]
      - name: planned_delivery_date
        tests: [not_null]
      - name: actual_delivery_date
        description: "NULL se non ancora consegnato"
      - name: otif_flag
        description: "True se consegnato On-Time and In-Full"
        tests:
          - accepted_values:
              values: [true, false, null]
              quote: false

  - name: gold_kpi_supply_chain
    description: "KPI giornalieri supply chain aggregati per SKU/plant/customer"
    config:
      tags: ["gold", "kpi", "dashboard"]
      materialized: table
    columns:
      - name: kpi_date
        tests: [not_null]
      - name: waste_rate_pct
        description: "% prodotto sprecato su prodotto totale"
        tests:
          - dbt_utils.expression_is_true:
              expression: "waste_rate_pct >= 0 AND waste_rate_pct <= 100"
      - name: otif_pct
        tests:
          - dbt_utils.expression_is_true:
              expression: "otif_pct >= 0 AND otif_pct <= 100"

-- models/silver/silver_lots.sql
-- {{ config(tags=["silver", "lots"], materialized="incremental") }}

WITH source_sap AS (
    SELECT
        lot_number                                          AS lot_id,
        material_number                                     AS sku_code,
        plant                                               AS plant_code,
        production_date                                     AS production_date,
        -- Normalizza data scadenza (SAP usa formato YYYYMMDD)
        TO_DATE(CAST(expiry_date_sap AS VARCHAR), 'YYYYMMDD') AS expiry_date,
        quantity_kg                                         AS net_weight_kg,
        -- Classi temperatura da tabella materiali SAP
        CASE
            WHEN temp_cond_sap = '0001' THEN 'ambient'
            WHEN temp_cond_sap = '0002' THEN 'chilled'
            WHEN temp_cond_sap = '0003' THEN 'frozen'
            WHEN temp_cond_sap = '0004' THEN 'ultra_frozen'
            ELSE 'unknown'
        END                                                 AS temperature_class,
        -- Range temperatura richiesto (gradi Celsius)
        CASE
            WHEN temp_cond_sap = '0001' THEN STRUCT(15.0 AS min_c, 25.0 AS max_c)
            WHEN temp_cond_sap = '0002' THEN STRUCT(2.0 AS min_c, 8.0 AS max_c)
            WHEN temp_cond_sap = '0003' THEN STRUCT(-25.0 AS min_c, -15.0 AS max_c)
            WHEN temp_cond_sap = '0004' THEN STRUCT(-60.0 AS min_c, -30.0 AS max_c)
        END                                                 AS temp_range,
        batch_status                                        AS lot_status,
        _ingestion_timestamp                                AS ingested_at
    FROM {{ source('bronze', 'sap_erp_batches') }}
    WHERE lot_number IS NOT NULL
        AND quantity_kg > 0

    {% if is_incremental() %}
        AND _ingestion_timestamp > (
            SELECT MAX(ingested_at) FROM {{ this }}
        )
    {% endif %}
),

deduplicated AS (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY lot_id
            ORDER BY ingested_at DESC
        ) AS row_num
    FROM source_sap
),

final AS (
    SELECT
        lot_id,
        sku_code,
        plant_code,
        production_date,
        expiry_date,
        net_weight_kg,
        temperature_class,
        temp_range,
        lot_status,
        -- Shelf life totale in giorni
        DATEDIFF('day', production_date, expiry_date)   AS shelf_life_days,
        -- Shelf life residua rispetto a oggi
        DATEDIFF('day', CURRENT_DATE, expiry_date)      AS shelf_life_remaining_days,
        ingested_at
    FROM deduplicated
    WHERE row_num = 1
        AND expiry_date > production_date
)

SELECT * FROM final

-- models/gold/gold_kpi_supply_chain.sql
-- {{ config(tags=["gold", "kpi"], materialized="table") }}

WITH lots AS (
    SELECT * FROM {{ ref('silver_lots') }}
),

shipments AS (
    SELECT * FROM {{ ref('silver_shipments') }}
),

pos_sales AS (
    SELECT * FROM {{ ref('silver_pos_sales') }}
),

inventory AS (
    SELECT * FROM {{ ref('silver_inventory_snapshot') }}
),

-- KPI 1: Waste Rate (% prodotto sprecato)
waste_kpi AS (
    SELECT
        production_date                                         AS kpi_date,
        plant_code,
        -- Prodotto scaduto o smaltito come spreco
        SUM(CASE WHEN lot_status IN ('WASTE', 'EXPIRED', 'DOWNGRADE')
                 THEN net_weight_kg ELSE 0 END)                 AS wasted_kg,
        SUM(net_weight_kg)                                      AS total_produced_kg,
        ROUND(
            100.0 * SUM(CASE WHEN lot_status IN ('WASTE', 'EXPIRED', 'DOWNGRADE')
                             THEN net_weight_kg ELSE 0 END)
            / NULLIF(SUM(net_weight_kg), 0),
            2
        )                                                       AS waste_rate_pct
    FROM lots
    GROUP BY 1, 2
),

-- KPI 2: OTIF (On-Time In-Full)
otif_kpi AS (
    SELECT
        DATE(planned_delivery_date)                             AS kpi_date,
        customer_code,
        COUNT(*)                                                AS total_orders,
        SUM(CASE WHEN otif_flag = TRUE THEN 1 ELSE 0 END)      AS otif_orders,
        ROUND(
            100.0 * SUM(CASE WHEN otif_flag = TRUE THEN 1 ELSE 0 END)
            / NULLIF(COUNT(*), 0),
            2
        )                                                       AS otif_pct
    FROM shipments
    WHERE actual_delivery_date IS NOT NULL
    GROUP BY 1, 2
),

-- KPI 3: Days of Supply (scorta in giorni)
dos_kpi AS (
    SELECT
        inv.snapshot_date                                       AS kpi_date,
        inv.sku_code,
        inv.location_code,
        inv.stock_qty_units,
        COALESCE(avg_daily_sales.avg_daily_units, 0)           AS avg_daily_demand,
        ROUND(
            inv.stock_qty_units
            / NULLIF(avg_daily_sales.avg_daily_units, 0),
            1
        )                                                       AS days_of_supply
    FROM inventory inv
    LEFT JOIN (
        SELECT sku_code, location_code,
               AVG(units_sold) AS avg_daily_units
        FROM pos_sales
        WHERE sale_date >= DATEADD('day', -30, CURRENT_DATE)
        GROUP BY 1, 2
    ) avg_daily_sales
        ON inv.sku_code = avg_daily_sales.sku_code
        AND inv.location_code = avg_daily_sales.location_code
),

-- KPI 4: Shelf Life at Receipt (qualità al ricevimento)
slr_kpi AS (
    SELECT
        DATE(s.actual_delivery_date)                            AS kpi_date,
        l.sku_code,
        -- % shelf life residua al momento della consegna
        ROUND(
            100.0 * l.shelf_life_remaining_days / NULLIF(l.shelf_life_days, 0),
            1
        )                                                       AS shelf_life_pct_at_receipt,
        -- Media per SKU
        AVG(
            100.0 * l.shelf_life_remaining_days / NULLIF(l.shelf_life_days, 0)
        ) OVER (PARTITION BY DATE(s.actual_delivery_date), l.sku_code)
                                                                AS avg_slr_pct
    FROM shipments s
    INNER JOIN lots l ON s.lot_id = l.lot_id
    WHERE s.actual_delivery_date IS NOT NULL
)

-- Join finale per dashboard KPI unificato
SELECT
    w.kpi_date,
    w.plant_code,
    w.waste_rate_pct,
    w.wasted_kg,
    w.total_produced_kg,
    o.otif_pct,
    o.total_orders,
    d.days_of_supply,
    d.avg_daily_demand,
    s.avg_slr_pct                                               AS shelf_life_at_receipt_pct,
    -- Score complessivo supply chain (0-100)
    ROUND(
        (
            (100.0 - COALESCE(w.waste_rate_pct, 0)) * 0.30 +
            COALESCE(o.otif_pct, 0) * 0.40 +
            LEAST(d.days_of_supply / 14.0 * 100, 100) * 0.15 +
            COALESCE(s.avg_slr_pct, 0) * 0.15
        ),
        1
    )                                                           AS supply_chain_health_score
FROM waste_kpi w
LEFT JOIN otif_kpi o ON w.kpi_date = o.kpi_date
LEFT JOIN dos_kpi d ON w.kpi_date = d.kpi_date
LEFT JOIN slr_kpi s ON w.kpi_date = s.kpi_date

# great_expectations/expectations/cold_chain_temperature_suite.py
import great_expectations as gx
from great_expectations.core.batch import BatchRequest

context = gx.get_context()

# Crea o aggiorna la suite di aspettative
suite = context.add_or_update_expectation_suite(
    expectation_suite_name="cold_chain_temperature_suite"
)

validator = context.get_validator(
    batch_request=BatchRequest(
        datasource_name="s3_bronze_datasource",
        data_connector_name="cold_chain_connector",
        data_asset_name="temperature_readings",
    ),
    expectation_suite_name="cold_chain_temperature_suite",
)

# REGOLA 1: Temperatura mai NULL per prodotti refrigerati/surgelati
validator.expect_column_values_to_not_be_null(
    column="celsius",
    mostly=0.99,  # tolleranza 1% per sensori offline temporanei
    meta={
        "notes": "Sensori IoT cold chain: max 1% di letture mancanti",
        "regulatory_reference": "EU Reg 37/2005 cold chain monitoring",
    }
)

# REGOLA 2: Range temperatura prodotti refrigerati (2-8°C)
validator.expect_column_values_to_be_between(
    column="celsius",
    min_value=-30.0,
    max_value=35.0,
    meta={
        "notes": "Range assoluto: include frozen (-25°C) e ambient (25°C)",
    }
)

# REGOLA 3: Lot ID sempre presente e formato GS1 SGTIN valido
validator.expect_column_values_to_match_regex(
    column="lot_id",
    regex=r"^[0-9]{14}$",  # GS1 GTIN-14 format
    meta={"notes": "GS1 SGTIN-14 standard per tracciabilita alimentare"}
)

# REGOLA 4: Timestamp monotonicamente crescente per sensore
validator.expect_column_values_to_be_increasing(
    column="reading_timestamp",
    strictly=False,  # allow duplicates (es. batch upload)
    parse_strings_as_datetimes=True,
    meta={"notes": "Verifico che i timestamp non siano retroattivi"}
)

# REGOLA 5: Numero di letture per lotto (almeno 1 ogni 30 min)
validator.expect_table_row_count_to_be_between(
    min_value=48,     # 24h * 2 letture/ora = 48 minimo
    max_value=10000,  # max ragionevole per 24h
    meta={"notes": "Minimo 1 lettura ogni 30 min per cold chain attiva"}
)

# REGOLA 6: Distribuzione temperatura per classe prodotto
# Prodotti refrigerati: mediana deve essere 2-8°C
validator.expect_column_median_to_be_between(
    column="celsius",
    min_value=1.0,
    max_value=9.0,
    meta={
        "notes": "Solo per lotti con temperature_class='chilled'",
        "applies_to": "filtered_batches_chilled",
    }
)

# Salva le aspettative
validator.save_expectation_suite(discard_failed_expectations=False)

# Checkpoint per Airflow
checkpoint = context.add_or_update_checkpoint(
    name="bronze_cold_chain_checkpoint",
    config_version=1,
    template_name=None,
    module_name="great_expectations.checkpoint",
    class_name="Checkpoint",
    run_name_template="%Y%m%d-%H%M%S-cold-chain",
    expectation_suite_name="cold_chain_temperature_suite",
    action_list=[
        {
            "name": "store_validation_result",
            "action": {"class_name": "StoreValidationResultAction"},
        },
        {
            "name": "send_slack_notification_on_failure",
            "action": {
                "class_name": "SlackNotificationAction",
                "slack_webhook": "{{ var.value.slack_webhook_cold_chain }}",
                "notify_on": "failure",
                "renderer": {
                    "module_name": "great_expectations.render.renderer",
                    "class_name": "SlackRenderer",
                },
            },
        },
        {
            "name": "update_data_docs",
            "action": {"class_name": "UpdateDataDocsAction"},
        },
    ],
)

# Architettura Lambda per Food Supply Chain
# Speed layer (real-time) + Batch layer (accuracy)

# SPEED LAYER: Kafka + Apache Flink
# Processa eventi in milliseconds: cold chain, recall alerts

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.common.serialization import SimpleStringSchema

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)

kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_topics("cold-chain-readings") \
    .set_group_id("flink-cold-chain-consumer") \
    .set_starting_offsets(KafkaOffsetsInitializer.latest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

cold_chain_stream = env.from_source(
    kafka_source,
    watermark_strategy=WatermarkStrategy.for_bounded_out_of_orderness(
        Duration.of_seconds(30)
    ),
    source_name="cold-chain-kafka",
)

# Filtra violazioni temperatura in tempo reale
violations_stream = cold_chain_stream \
    .map(lambda msg: json.loads(msg)) \
    .filter(lambda r: (
        r.get("temperature_class") == "chilled" and
        (r["celsius"] < 2.0 or r["celsius"] > 8.0)
    )) \
    .map(lambda r: {
        **r,
        "violation_severity": "CRITICAL" if r["celsius"] > 12.0 else "WARNING",
        "violation_detected_at": datetime.utcnow().isoformat(),
    })

# Scrivi violations su Kafka alert topic
violations_stream.sink_to(
    KafkaSink.builder()
        .set_bootstrap_servers("kafka:9092")
        .set_record_serializer(
            KafkaRecordSerializationSchema.builder()
                .set_topic("cold-chain-violations")
                .set_value_serialization_schema(SimpleStringSchema())
                .build()
        )
        .build()
)

env.execute("food-cold-chain-monitoring")

# src/connectors/sap_connector.py
"""
Connettore SAP ERP via pyrfc (SAP RFC/BAPI wrapper).
Richiede: SAP NetWeaver RFC SDK + pyrfc library.
"""
import pyrfc
import pandas as pd
from dataclasses import dataclass
from typing import Optional
import logging

logger = logging.getLogger(__name__)

@dataclass
class SAPConnectionConfig:
    ashost: str
    sysnr: str
    client: str
    user: str
    passwd: str
    lang: str = "IT"

class SAPConnector:
    """Wrapper per SAP RFC/BAPI calls per dati supply chain."""

    def __init__(self, config: SAPConnectionConfig):
        self._config = config
        self._conn: Optional[pyrfc.Connection] = None

    def __enter__(self):
        self._conn = pyrfc.Connection(
            ashost=self._config.ashost,
            sysnr=self._config.sysnr,
            client=self._config.client,
            user=self._config.user,
            passwd=self._config.passwd,
            lang=self._config.lang,
        )
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._conn:
            self._conn.close()

    def get_production_batches(
        self,
        date_from: str,
        date_to: str,
        plants: list[str],
    ) -> pd.DataFrame:
        """
        Estrae lotti di produzione da SAP tramite BAPI_BATCH_GET_DETAIL.
        Args:
            date_from: Data inizio formato YYYYMMDD
            date_to: Data fine formato YYYYMMDD
            plants: Lista codici stabilimento SAP
        """
        all_batches = []

        for plant in plants:
            try:
                # Chiama BAPI SAP per lista lotti
                result = self._conn.call(
                    "BAPI_BATCH_GET_DETAIL",
                    PLANT=plant,
                    DATE_FROM=date_from,
                    DATE_TO=date_to,
                    BATCH_STATUS="",  # tutti gli stati
                )

                batches = result.get("BATCH_DETAIL_LIST", [])
                logger.info(
                    "SAP plant %s: %d lotti estratti",
                    plant,
                    len(batches),
                )

                for batch in batches:
                    all_batches.append({
                        "lot_id": f"{batch['MATNR']}-{batch['CHARG']}",
                        "material_number": batch["MATNR"].strip(),
                        "lot_number": batch["CHARG"].strip(),
                        "plant": plant,
                        "production_date": batch.get("HSDAT"),
                        "expiry_date_sap": batch.get("VFDAT"),
                        "batch_status": batch.get("ZUSTD"),
                        "quantity_kg": float(batch.get("CLABS", 0)),
                        "temp_cond_sap": batch.get("MHDRZ", "0001"),
                        "_source": "SAP_ERP",
                        "_extracted_at": pd.Timestamp.utcnow().isoformat(),
                    })

            except pyrfc.ABAPApplicationError as e:
                logger.error(
                    "SAP BAPI error per plant %s: %s",
                    plant,
                    str(e),
                )
                # Non fallisce tutta l'estrazione per un singolo plant
                continue

        return pd.DataFrame(all_batches)

    def get_stock_movements(
        self,
        date_from: str,
        date_to: str,
        movement_types: list[str],
    ) -> pd.DataFrame:
        """
        Estrae movimenti di magazzino SAP via MB51 BAPI equivalent.
        movement_types: ['101'=GR, '261'=GI to production, '601'=GI to customer]
        """
        result = self._conn.call(
            "BAPI_GOODSMVT_GETDETAIL",
            GOODSMVT_DATE_FROM=date_from,
            GOODSMVT_DATE_TO=date_to,
        )

        movements = []
        for item in result.get("GOODSMVT_ITEMS", []):
            if item["BWART"] in movement_types:
                movements.append({
                    "movement_id": f"{item['MBLNR']}-{item['ZEILE']}",
                    "material": item["MATNR"].strip(),
                    "plant": item["WERKS"],
                    "movement_type": item["BWART"],
                    "quantity": float(item.get("MENGE", 0)),
                    "unit": item.get("MEINS", "KG"),
                    "lot_number": item.get("CHARG", ""),
                    "posting_date": item["BUDAT"],
                    "_source": "SAP_MOVEMENTS",
                })

        return pd.DataFrame(movements)

# debezium-config/oracle-erp-connector.json
# Configurazione Debezium per CDC da Oracle EBS (ERP legacy)
{
  "name": "oracle-erp-food-supply-chain",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "tasks.max": "1",
    "database.hostname": "oracle-erp.internal",
    "database.port": "1521",
    "database.user": "debezium_cdc",
    "database.password": "${ORACLE_CDC_PASSWORD}",
    "database.dbname": "FOOD_ERP",
    "database.pdb.name": "FOOD_PDB",
    "database.server.name": "oracle-erp",
    "table.include.list": "FOOD_ERP.INV_TRANSACTIONS,FOOD_ERP.MTL_LOT_NUMBERS,FOOD_ERP.WSH_DELIVERY_DETAILS",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "dbhistory.oracle.food_erp",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schemahistory.oracle.food_erp",
    "transforms": "route,addFields",
    "transforms.route.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.addFields.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addFields.static.field": "_cdc_source",
    "transforms.addFields.static.value": "oracle-erp",
    "include.schema.changes": "true",
    "snapshot.mode": "initial_only",
    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    "tombstones.on.delete": "true",
    "topic.prefix": "food-cdc"
  }
}

# src/connectors/edi_connector.py
"""
Parser EDI EDIFACT per messaggi ORDERS, DESADV, RECADV (GS1 subset).
Usa pydifact library per parsing EDIFACT messages.
"""
import pandas as pd
from pydifact.segmentcollection import SegmentCollection
from pydifact.segments import Segment
from pathlib import Path
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def parse_edifact_orders(file_path: str | Path) -> pd.DataFrame:
    """
    Parsa un file EDI EDIFACT ORDERS D.01B (GS1 EANCOM).
    Restituisce DataFrame con righe ordine normalizzate.

    Struttura messaggio ORDERS EDIFACT:
    UNB (Interchange header)
    UNH (Message header)
    BGM (Order message identifier)
    DTM (Date/time)
    NAD+BY (Buyer = retailer)
    NAD+SU (Supplier = nostro sistema)
    LIN (Line item)
      +QTY (Quantity ordered)
      +DTM (Requested delivery date)
      +PIA (Product code GTIN)
    UNT (Message trailer)
    """
    file_content = Path(file_path).read_text(encoding="latin-1")
    collection = SegmentCollection.from_str(file_content)

    orders = []
    current_order: dict = {}
    current_line: dict = {}

    for segment in collection.segments:
        tag = segment.tag

        if tag == "BGM":
            # Inizio nuovo ordine
            current_order = {
                "order_number": segment.elements[1] if len(segment.elements) > 1 else None,
                "document_type": segment.elements[0][0] if segment.elements else None,
                "lines": [],
            }

        elif tag == "DTM" and current_order:
            # Data ordine o data consegna richiesta
            qualifier = segment.elements[0][0]
            date_str = segment.elements[0][1]
            date_fmt = segment.elements[0][2]

            if qualifier == "137":  # Document date
                try:
                    current_order["order_date"] = datetime.strptime(
                        date_str, "%Y%m%d"
                    ).date().isoformat()
                except ValueError:
                    logger.warning("DTM parse error: %s", date_str)

            elif qualifier == "2":  # Requested delivery date
                try:
                    current_order["requested_delivery_date"] = datetime.strptime(
                        date_str, "%Y%m%d"
                    ).date().isoformat()
                except ValueError:
                    pass

        elif tag == "NAD":
            qualifier = segment.elements[0]
            party_id = segment.elements[1][0] if len(segment.elements) > 1 else None
            party_name = segment.elements[3] if len(segment.elements) > 3 else None

            if qualifier == "BY":  # Buyer (retailer)
                current_order["buyer_gln"] = party_id
                current_order["buyer_name"] = party_name
            elif qualifier == "SU":  # Supplier
                current_order["supplier_gln"] = party_id

        elif tag == "LIN":
            # Salva riga precedente se esiste
            if current_line and current_order:
                current_order["lines"].append(current_line)

            current_line = {
                "line_number": segment.elements[0],
                "gtin": None,
                "qty_ordered": None,
                "delivery_date": None,
            }

        elif tag == "PIA" and current_line:
            # Product identification: GTIN-13 o GTIN-14
            if len(segment.elements) > 1:
                qualifier = segment.elements[1][1] if len(segment.elements[1]) > 1 else ""
                if qualifier in ["SRV", "GE1"]:  # GTIN qualifiers
                    current_line["gtin"] = segment.elements[1][0]

        elif tag == "QTY" and current_line:
            qualifier = segment.elements[0][0]
            if qualifier == "21":  # Ordered quantity
                current_line["qty_ordered"] = float(segment.elements[0][1])
                current_line["qty_unit"] = segment.elements[0][2] if len(segment.elements[0]) > 2 else "PCE"

    # Normalizza output
    for order in orders:
        for line in order.get("lines", []):
            orders.append({
                "order_number": order.get("order_number"),
                "order_date": order.get("order_date"),
                "requested_delivery_date": order.get("requested_delivery_date"),
                "buyer_gln": order.get("buyer_gln"),
                "buyer_name": order.get("buyer_name"),
                "supplier_gln": order.get("supplier_gln"),
                "line_number": line.get("line_number"),
                "gtin": line.get("gtin"),
                "qty_ordered": line.get("qty_ordered"),
                "qty_unit": line.get("qty_unit"),
                "_source": "EDI_EDIFACT_ORDERS",
                "_parsed_at": datetime.utcnow().isoformat(),
            })

    return pd.DataFrame(orders)

# Inventario sorgenti dati: output assessment
SOURCE_INVENTORY = {
    "farm_iot": {
        "count": 8000,
        "protocol": "MQTT v3.1.1",
        "broker": "Mosquitto on-premise",
        "format": "JSON custom",
        "issues": [
            "Schema non standardizzato tra vendor IoT diversi",
            "15% sensori con timestamp errati (clock drift)",
            "3 farm su 500 senza connettivita stabile (4G intermittente)",
        ],
        "recommended_fix": "Normalizzazione schema + edge buffer (Mosquitto local)",
    },
    "sap_erp": {
        "version": "SAP R/3 4.7 (anno 2005)",
        "tables_relevant": ["MCHA", "MCH1", "MSEG", "MKPF", "VBAK", "VBAP"],
        "issues": [
            "No REST API: solo RFC/BAPI",
            "Batch extraction window: 01:00-03:00 (finestra ristretta)",
            "Codifica caratteri: ISO-8859-1 (non UTF-8)",
            "Date in formato SAP (YYYYMMDD come intero)",
        ],
        "recommended_fix": "pyrfc + ETL incremental con delta timestamp",
    },
    "edi_partners": {
        "partners": 15,
        "standards": ["EANCOM D.01B", "X12 850/856", "GS1 XML 3.1"],
        "transport": "SFTP (12 partner), AS2 (3 partner)",
        "issues": [
            "3 retailer usano EDIFACT non-standard (varianti proprietarie)",
            "Mancanza di RECADV da 5 retailer: no conferma ricezione",
        ],
    },
    "pos_retail": {
        "vendors": ["Cassa Easy", "NCR Aloha", "Custom PHP legacy"],
        "issues": [
            "3 formati CSV diversi",
            "Nessun campo shelf_life nei dati POS",
            "Granularità: ora (non transazione)",
        ],
    },
}

# Stack tecnico implementato per la cooperativa

INFRASTRUCTURE = {
    "data_lake": "AWS S3 (Medallion: Bronze/Silver/Gold)",
    "compute": "AWS EMR Serverless (Spark per bulk load iniziale)",
    "warehouse": "Snowflake (pay-per-query, snowpark per ML features)",
    "orchestration": "Apache Airflow 2.9 su AWS MWAA (managed)",
    "transformation": "dbt Cloud (Team plan: 8 developers)",
    "data_quality": "Great Expectations + Slack alerts",
    "streaming": "Apache Kafka su MSK (managed) per cold chain",
    "monitoring": "Grafana + Prometheus (Airflow metrics, DWH queries)",
    "ci_cd": "GitHub Actions (dbt CI: test su ogni PR)",
    "secret_management": "AWS Secrets Manager",
}

TIMELINE = [
    {"week": "1-4",  "milestone": "Assessment sorgenti + architettura design"},
    {"week": "5-7",  "milestone": "Bronze layer: connettori SAP + IoT MQTT"},
    {"week": "8-10", "milestone": "Bronze layer: EDI + POS + LIMS"},
    {"week": "11-12","milestone": "Silver layer: dbt models + data quality"},
    {"week": "13-14","milestone": "Gold layer: KPI dashboard + alerting"},
    {"week": "15-16","milestone": "UAT + go-live + team training"},
}

TEAM = {
    "data_engineers": 3,
    "dbt_developers": 2,
    "sap_consultant": 1,  # part-time per RFC/BAPI
    "project_manager": 1,
}