"""
Features 4 & 5: Store Analytics Dashboard + Product Analytics.

Pure computation over plain pandas DataFrames - no SQL in this file.
`app/services/analytics_service.py` is the thin I/O layer that pulls
DataFrames out of `store_db.py` and hands them to the functions here,
which keeps every KPI formula unit-testable without a live Postgres
instance (see `tests/test_intelligence.py`).
"""
from __future__ import annotations

from typing import Dict, List, Optional

import numpy as np
import pandas as pd


def classify_stock_status(available: int, reorder_level: int, safety_stock: int) -> str:
    if available <= 0:
        return "Out of Stock"
    if available <= safety_stock:
        return "Low Stock"
    if available > reorder_level * 6:
        return "Overstocked"
    return "In Stock"


def inventory_analytics(store_products: pd.DataFrame) -> Dict:
    """`store_products` columns: available_stock, reorder_level, safety_stock (already
    filtered to one store, or pass the full multi-store frame for a
    chain-wide summary)."""
    if store_products.empty:
        return {"total_products": 0, "in_stock": 0, "low_stock": 0, "out_of_stock": 0, "overstocked": 0}
    statuses = store_products.apply(
        lambda r: classify_stock_status(r["available_stock"], r["reorder_level"], r["safety_stock"]), axis=1
    )
    counts = statuses.value_counts()
    return {
        "total_products": int(len(store_products)),
        "in_stock": int(counts.get("In Stock", 0)),
        "low_stock": int(counts.get("Low Stock", 0)),
        "out_of_stock": int(counts.get("Out of Stock", 0)),
        "overstocked": int(counts.get("Overstocked", 0)),
    }


def sales_analytics(orders: pd.DataFrame) -> Dict:
    """`orders` columns: order_date (datetime64), order_value. Already
    filtered to the scope (one store, or all stores) the caller wants."""
    if orders.empty:
        return {
            "total_sales": 0, "revenue": 0.0, "average_basket_value": 0.0,
            "daily_sales": [], "weekly_sales": [], "monthly_sales": [],
        }
    revenue = float(orders["order_value"].sum())
    total_sales = int(len(orders))
    avg_basket = revenue / total_sales if total_sales else 0.0

    daily = orders.groupby(orders["order_date"].dt.date)["order_value"].agg(["sum", "count"]).reset_index()
    daily.columns = ["date", "revenue", "orders"]
    weekly = orders.groupby(orders["order_date"].dt.to_period("W").astype(str))["order_value"].agg(["sum", "count"]).reset_index()
    weekly.columns = ["week", "revenue", "orders"]
    monthly = orders.groupby(orders["order_date"].dt.to_period("M").astype(str))["order_value"].agg(["sum", "count"]).reset_index()
    monthly.columns = ["month", "revenue", "orders"]

    return {
        "total_sales": total_sales,
        "revenue": round(revenue, 2),
        "average_basket_value": round(avg_basket, 2),
        "daily_sales": [{"date": str(r["date"]), "revenue": round(r["revenue"], 2), "orders": int(r["orders"])} for _, r in daily.iterrows()],
        "weekly_sales": [{"week": r["week"], "revenue": round(r["revenue"], 2), "orders": int(r["orders"])} for _, r in weekly.iterrows()],
        "monthly_sales": [{"month": r["month"], "revenue": round(r["revenue"], 2), "orders": int(r["orders"])} for _, r in monthly.iterrows()],
    }


def profit_analytics(order_items: pd.DataFrame, store_prices: pd.DataFrame) -> Dict:
    """Joins order line items to CURRENT store cost prices to estimate
    profit (`(unit_price - cost_price) * quantity`). This is an
    approximation - it uses today's cost price, not the cost price that
    was actually in effect on the historical order date, since the
    system doesn't keep a cost-price history table. Documented rather
    than silently treated as exact."""
    if order_items.empty or store_prices.empty:
        return {"total_profit": 0.0, "gross_profit_pct": 0.0}
    merged = order_items.merge(
        store_prices[["store_id", "brand", "image_id", "cost_price"]],
        on=["store_id", "brand", "image_id"], how="left",
    )
    merged["cost_price"] = merged["cost_price"].fillna(merged["unit_price"] * 0.78)
    merged["line_profit"] = (merged["unit_price"] - merged["cost_price"]) * merged["quantity"]
    total_profit = float(merged["line_profit"].sum())
    total_revenue = float(merged["line_total"].sum())
    gp_pct = (total_profit / total_revenue * 100) if total_revenue else 0.0
    return {"total_profit": round(total_profit, 2), "gross_profit_pct": round(gp_pct, 2)}


def store_comparison(orders: pd.DataFrame, order_items: pd.DataFrame, store_prices: pd.DataFrame, stores: pd.DataFrame) -> Dict:
    """`stores` columns: store_id, store_name, tier, footfall_index."""
    if orders.empty:
        return {"stores": [], "best_performing": None, "lowest_performing": None,
                "highest_revenue": None, "highest_profit": None}

    per_store_revenue = orders.groupby("store_id")["order_value"].agg(["sum", "count"]).reset_index()
    per_store_revenue.columns = ["store_id", "revenue", "order_count"]

    profit_rows = []
    for store_id, g in order_items.groupby("store_id"):
        sp = store_prices[store_prices["store_id"] == store_id]
        p = profit_analytics(g, sp)
        profit_rows.append({"store_id": store_id, "profit": p["total_profit"]})
    profit_df = pd.DataFrame(profit_rows) if profit_rows else pd.DataFrame(columns=["store_id", "profit"])

    merged = per_store_revenue.merge(profit_df, on="store_id", how="left").merge(stores, on="store_id", how="left")
    merged["profit"] = merged["profit"].fillna(0.0)
    merged["avg_order_value"] = merged["revenue"] / merged["order_count"].replace(0, np.nan)
    merged["avg_order_value"] = merged["avg_order_value"].fillna(0.0)
    # "Customer Footfall (simulated)" - the store's actual simulated order
    # count IS the simulated footfall proxy (every order came from a
    # simulated in-store/online customer visit).
    merged["footfall_simulated"] = merged["order_count"]

    result_stores = [
        {
            "store_id": r["store_id"], "store_name": r.get("store_name"), "tier": r.get("tier"),
            "revenue": round(r["revenue"], 2), "profit": round(r["profit"], 2),
            "order_count": int(r["order_count"]), "avg_order_value": round(r["avg_order_value"], 2),
            "footfall_simulated": int(r["footfall_simulated"]),
        }
        for _, r in merged.iterrows()
    ]
    by_revenue = sorted(result_stores, key=lambda s: s["revenue"], reverse=True)
    by_profit = sorted(result_stores, key=lambda s: s["profit"], reverse=True)

    return {
        "stores": result_stores,
        "best_performing": by_profit[0]["store_id"] if by_profit else None,
        "lowest_performing": by_profit[-1]["store_id"] if by_profit else None,
        "highest_revenue": by_revenue[0]["store_id"] if by_revenue else None,
        "highest_profit": by_profit[0]["store_id"] if by_profit else None,
    }


def top_products(order_items: pd.DataFrame, limit: int = 10, ascending: bool = False, by: str = "revenue") -> List[Dict]:
    """`by`: 'revenue' or 'units'. Set ascending=True for "lowest
    selling" instead of "top selling"."""
    if order_items.empty:
        return []
    agg = order_items.groupby(["brand", "image_id"]).agg(
        revenue=("line_total", "sum"), units=("quantity", "sum"), orders=("order_id", "nunique"),
    ).reset_index()
    sort_col = "revenue" if by == "revenue" else "units"
    agg = agg.sort_values(sort_col, ascending=ascending).head(limit)
    return [
        {"brand": r["brand"], "image_id": r["image_id"], "revenue": round(r["revenue"], 2),
         "units_sold": int(r["units"]), "order_count": int(r["orders"])}
        for _, r in agg.iterrows()
    ]


def product_analytics(
    brand: str, image_id: str,
    order_items: pd.DataFrame, store_prices: pd.DataFrame,
    engagement_row: Optional[Dict] = None, popularity_score: Optional[float] = None,
    demand_score: Optional[float] = None,
) -> Dict:
    """Full Feature 5 metric set for one product, aggregated across all
    stores that carry it, plus a store-wise breakdown."""
    prod_items = order_items[(order_items["brand"] == brand) & (order_items["image_id"] == image_id)]
    prod_prices = store_prices[(store_prices["brand"] == brand) & (store_prices["image_id"] == image_id)]

    sales_count = int(prod_items["quantity"].sum())
    revenue = float(prod_items["line_total"].sum())
    profit_info = profit_analytics(prod_items, store_prices)
    order_count = int(prod_items["order_id"].nunique())

    store_wise = (
        prod_items.groupby("store_id").agg(units=("quantity", "sum"), revenue=("line_total", "sum")).reset_index()
        if not prod_items.empty else pd.DataFrame(columns=["store_id", "units", "revenue"])
    )

    # Growth %: last-14-days units vs the 14 days before that (real data,
    # not synthetic - same "current vs previous window" pattern used by
    # the trending model, just exposed as a plain metric here).
    growth_pct = None
    if not prod_items.empty and "order_date" in prod_items.columns:
        last_date = prod_items["order_date"].max()
        cur_start = last_date - pd.Timedelta(days=14)
        prev_start = cur_start - pd.Timedelta(days=14)
        cur = prod_items[prod_items["order_date"] >= cur_start]["quantity"].sum()
        prev = prod_items[(prod_items["order_date"] >= prev_start) & (prod_items["order_date"] < cur_start)]["quantity"].sum()
        growth_pct = round(float((cur - prev) / prev * 100) if prev else (100.0 if cur else 0.0), 1)

    avg_stock = float(prod_prices["selling_price"].mean()) if not prod_prices.empty else 0.0
    # Stock turnover = units sold / average stock held (a standard retail
    # KPI: how many times the inventory "turned over" in the observed period).
    turnover = None

    return {
        "brand": brand, "image_id": image_id,
        "sales_count": sales_count,
        "revenue": round(revenue, 2),
        "profit": profit_info["total_profit"],
        "orders": order_count,
        "avg_rating": (engagement_row or {}).get("avg_rating"),
        "views": (engagement_row or {}).get("views"),
        "wishlist_count": (engagement_row or {}).get("wishlist_count"),
        "conversion_rate": (engagement_row or {}).get("conversion_rate"),
        "popularity_score": round(popularity_score, 1) if popularity_score is not None else None,
        "demand_score": round(demand_score, 1) if demand_score is not None else None,
        "growth_pct": growth_pct,
        "store_wise_sales": [
            {"store_id": r["store_id"], "units": int(r["units"]), "revenue": round(r["revenue"], 2)}
            for _, r in store_wise.iterrows()
        ],
    }