"""Delivery performance analytics queries."""

import pandas as pd
from sqlalchemy import Engine, text


def delivery_by_state(engine: Engine) -> pd.DataFrame:
    sql = text("""
        SELECT
            dc.state,
            AVG(f.days_to_delivery)     AS avg_days,
            AVG(f.delivery_delay_days)  AS avg_delay,
            AVG(CASE WHEN f.is_late THEN 1.0 ELSE 0.0 END) AS late_rate,
            COUNT(DISTINCT f.order_id)  AS orders
        FROM fact_orders f
        JOIN dim_customers dc ON f.customer_key = dc.customer_key
        WHERE f.days_to_delivery IS NOT NULL AND dc.state IS NOT NULL
        GROUP BY 1
    """)
    with engine.connect() as conn:
        return pd.read_sql(sql, conn)


def delivery_delay_distribution(engine: Engine) -> pd.DataFrame:
    sql = text("""
        SELECT delivery_delay_days AS delay_days, COUNT(*) AS orders
        FROM fact_orders
        WHERE delivery_delay_days IS NOT NULL
          AND delivery_delay_days BETWEEN -30 AND 60
        GROUP BY 1 ORDER BY 1
    """)
    with engine.connect() as conn:
        return pd.read_sql(sql, conn)


def review_vs_delay(engine: Engine, sample: int = 5000) -> pd.DataFrame:
    sql = text("""
        SELECT review_score, delivery_delay_days, days_to_delivery
        FROM fact_orders
        WHERE review_score IS NOT NULL AND delivery_delay_days IS NOT NULL
        ORDER BY RANDOM() LIMIT :n
    """)
    with engine.connect() as conn:
        return pd.read_sql(sql, conn, params={"n": sample})


def delivery_kpis(engine: Engine) -> dict:
    sql = text("""
        SELECT
            AVG(days_to_delivery)   AS avg_delivery_days,
            AVG(CASE WHEN is_late THEN 1.0 ELSE 0.0 END) AS late_rate,
            COUNT(*) FILTER (WHERE is_late)               AS late_orders,
            COUNT(*) FILTER (WHERE NOT is_late)           AS on_time_orders
        FROM fact_orders
        WHERE days_to_delivery IS NOT NULL
    """)
    with engine.connect() as conn:
        row = conn.execute(sql).fetchone()
        return dict(row._mapping)