Update main.py

main.py

@@ -4,8 +4,7 @@
 # - Deterministic time windows (Harare); explicit start/end on API calls
 # - KPI engine never uses LLM for numbers (LLM is narration-only fallback)
 # - JSON-safe snapshot; deep DEBUG logs (optional mirror to Firebase)
-# - Drop-in Firebase + AI wiring identical in spirit to your prior server
-
+# - Drop-in Firebase + AI wiring identical in spirit to prior server
 from __future__ import annotations
 
 import os, io, re, json, time, uuid, base64, logging
@@ -49,7 +48,6 @@ try:
     credentials_json_string = os.environ.get("FIREBASE")
     if not credentials_json_string:
         raise ValueError("FIREBASE env var is not set")
-
     credentials_json = json.loads(credentials_json_string)
     firebase_db_url = os.environ.get("Firebase_DB")
     if not firebase_db_url:
@@ -113,16 +111,11 @@ logger.info(f"Chart export path set to: {user_defined_path}")
 # -----------------------------------------------------------------------------
 # Admin API client (client-supplied credentials; holistic admin scope)
 # -----------------------------------------------------------------------------
-# --- Auth + Request client (robust: bearer OR cookie session) ---
-
-import requests
-from typing import Dict, Optional
-
 SC_BASE_URL = os.getenv("SC_BASE_URL", "https://delta-api.pricelyst.co.zw").rstrip("/")
 
 class SCAuth:
     """Caches a requests.Session per admin email; supports bearer or cookie sessions."""
-    _cache: Dict[str, Dict[str, any]] = {}
+    _cache: Dict[str, Dict[str, Any]] = {}
 
     @classmethod
     def invalidate(cls, email: str) -> None:
@@ -137,7 +130,6 @@ class SCAuth:
     def _extract_token(cls, js: dict) -> Optional[str]:
         if not isinstance(js, dict):
             return None
-        # Try common token shapes
         candidates = [
             js.get("token"),
             js.get("access_token"),
@@ -152,42 +144,38 @@ class SCAuth:
         return None
 
     @classmethod
-    def login(cls, email: str, password: str) -> Dict[str, any]:
+    def login(cls, email: str, password: str) -> Dict[str, Any]:
         s = requests.Session()
         s.headers.update({"Accept": "application/json"})
         url = f"{SC_BASE_URL}/api/auth/admin/login"
-        # IMPORTANT: many APIs expect JSON, not form data
         resp = s.post(url, json={"email": email, "password": password}, timeout=30)
 
-        # Try JSON parse (may fail if empty body)
-        body_text = ""
-        body_json = {}
+        body_text, body_json = "", {}
         try:
             body_json = resp.json() or {}
         except Exception:
-            body_text = (resp.text or "")[:800]  # keep it short for logs
-
+            body_text = (resp.text or "")[:800]
         token = cls._extract_token(body_json)
 
         if token:
             s.headers.update({"Authorization": f"Bearer {token}"})
             entry = {"session": s, "auth": "bearer", "token": token}
             cls._cache[email] = entry
+            logger.debug("Admin login (bearer) OK")
             return entry
 
-        # If no token but we got Set-Cookie, assume cookie session auth
         if resp.cookies and (resp.status_code // 100) == 2:
             entry = {"session": s, "auth": "cookie"}
             cls._cache[email] = entry
+            logger.debug("Admin login (cookie) OK")
             return entry
 
-        # Neither token nor cookie: raise, but include short body for diagnosis
         snippet = body_text or (str(body_json)[:800])
         raise RuntimeError(f"Login did not return a token or cookie session. HTTP {resp.status_code}. Body≈ {snippet}")
 
 def sc_request(method: str, path: str, email: str, password: str, *,
                params: dict = None, json_body: dict = None, timeout: int = 30):
-    """Authenticated request with 401 auto-refresh (once)."""
+    """Authenticated request with 401 auto-refresh (once). Logs a compact sample on success."""
     if not path.startswith("/"):
         path = "/" + path
     url = f"{SC_BASE_URL}{path}"
@@ -195,7 +183,6 @@ def sc_request(method: str, path: str, email: str, password: str, *,
     def _do(s: requests.Session):
         return s.request(method.upper(), url, params=params, json=json_body, timeout=timeout)
 
-    # get or create session
     entry = SCAuth._cache.get(email)
     if not entry:
         entry = SCAuth.login(email, password)
@@ -203,77 +190,83 @@ def sc_request(method: str, path: str, email: str, password: str, *,
 
     resp = _do(s)
     if resp.status_code == 401:
-        # refresh & retry once
         SCAuth.invalidate(email)
         entry = SCAuth.login(email, password)
         s = entry["session"]
         resp = _do(s)
 
-    # Raise for other errors
     try:
         resp.raise_for_status()
     except Exception as e:
-        # include small snippet to aid debugging
         snippet = (getattr(resp, "text", "") or "")[:800]
         raise RuntimeError(f"SC request error {method.upper()} {path}: HTTP {resp.status_code} – {snippet}") from e
 
-    # Return JSON or text as appropriate
+    payload: Any
     try:
-        return resp.json()
+        payload = resp.json()
     except Exception:
-        return resp.text
-# -----------------------------------------------------------------------------
-# Temporal helpers
-# -----------------------------------------------------------------------------
+        payload = resp.text
+
+    # ---- Compact sample logging for every endpoint ----
+    sample = None
+    if isinstance(payload, dict):
+        d = payload.get("data", payload)
+        if isinstance(d, dict):
+            # try common array keys
+            for key in ("sales_over_time", "orders", "transactions", "items", "list", "rows", "data"):
+                v = d.get(key)
+                if isinstance(v, list) and v:
+                    sample = {key: v[:2]}  # first 2 rows
+                    break
+            if sample is None:
+                # fallback: first 10 keys
+                sample = {k: ("[list]" if isinstance(v, list) else v) for k, v in list(d.items())[:10]}
+        elif isinstance(d, list):
+            sample = d[:2]
+    elif isinstance(payload, list):
+        sample = payload[:2]
+    else:
+        sample = str(payload)[:300]
+
+    logger.debug("SAMPLE %s %s -> %s", method.upper(), path, json.dumps(sample, default=str))
+    return payload
+
 # -----------------------------------------------------------------------------
-# Timezone Configuration (fix for "_TZ is not defined")
+# Timezone & temporal helpers
 # -----------------------------------------------------------------------------
-import os
-import pandas as pd
-
-# Canonical timezone for Brave Retail Insights
 TZ = os.getenv("APP_TZ", "Africa/Harare")
-
-# Backward-compatible alias for older references
-_TZ = TZ
+_TZ = TZ  # backward-compatible alias
 
 def now_harare() -> pd.Timestamp:
-    """Return the current timestamp in Harare timezone."""
     return pd.Timestamp.now(tz=TZ)
 
-def week_bounds_from(ts: pd.Timestamp) -> tuple[pd.Timestamp, pd.Timestamp]:
-    """Return Monday–Sunday bounds for the given timestamp."""
+def week_bounds_from(ts: pd.Timestamp) -> Tuple[pd.Timestamp, pd.Timestamp]:
     monday = ts.tz_convert(TZ).normalize() - pd.Timedelta(days=ts.weekday())
     sunday = monday + pd.Timedelta(days=6, hours=23, minutes=59, seconds=59)
     return monday, sunday
 
-def this_month_bounds(ts: pd.Timestamp) -> tuple[pd.Timestamp, pd.Timestamp]:
-    """Return start and end of the current month."""
+def this_month_bounds(ts: pd.Timestamp) -> Tuple[pd.Timestamp, pd.Timestamp]:
     first_this = ts.normalize().replace(day=1)
-    next_month = first_this.replace(
-        year=first_this.year + 1, month=1
-    ) if first_this.month == 12 else first_this.replace(month=first_this.month + 1)
-    last_this = next_month - pd.Timedelta(seconds=1)
+    if first_this.month == 12:
+        first_next = first_this.replace(year=first_this.year + 1, month=1)
+    else:
+        first_next = first_this.replace(month=first_this.month + 1)
+    last_this = first_next - pd.Timedelta(seconds=1)
     return first_this, last_this
-    
 
 def period_to_bounds(period: str) -> Tuple[pd.Timestamp, pd.Timestamp, str]:
     p = (period or "week").strip().lower()
     now = now_harare()
     if p == "today":
         start = now.normalize()
-        end = start + pd.Timedelta(hours=23, minutes=59, seconds=59)
-        lbl = "Today"
+        end = start + pd.Timedelta(hours=23, minutes=59, seconds=59); lbl = "Today"
     elif p in ("week", "this_week"):
-        start, end = week_bounds_from(now)
-        lbl = "This Week"
+        start, end = week_bounds_from(now); lbl = "This Week"
     elif p in ("month", "this_month"):
-        start, end = this_month_bounds(now)
-        lbl = "This Month"
+        start, end = this_month_bounds(now); lbl = "This Month"
     elif p in ("year", "this_year"):
         start = now.normalize().replace(month=1, day=1, hour=0, minute=0, second=0)
-        end = now.normalize().replace(month=12, day=31, hour=23, minute=59, second=59)
-        lbl = "This Year"
+        end = now.normalize().replace(month=12, day=31, hour=23, minute=59, second=59); lbl = "This Year"
     else:
         start, end = week_bounds_from(now); lbl = "This Week"
     return start, end, lbl
@@ -338,32 +331,20 @@ def sanitize_answer(ans) -> str:
     if tb in s: s = s.split(tb, 1)[0]
     return (s or "").strip()
 
-
-
-# -------------------- Robust normalizers (paste once, near your engine class) --------------------
-from typing import Any, Iterable, List, Dict, Optional
-import json
-import pandas as pd
-import numpy as np
-
+# -----------------------------------------------------------------------------
+# Robust normalizers
+# -----------------------------------------------------------------------------
 def _to_list(x: Any) -> List[Any]:
-    """Return x as a list. Accept list/dict/JSON-string/None gracefully."""
-    if x is None:
-        return []
-    if isinstance(x, list):
-        return x
-    if isinstance(x, dict):
-        return [x]
+    if x is None: return []
+    if isinstance(x, list): return x
+    if isinstance(x, dict): return [x]
     if isinstance(x, str):
-        # try to parse JSON strings like '[{...}]' or '{"a":1}'
         try:
             j = json.loads(x)
-            if isinstance(j, list):
-                return j
-            if isinstance(j, dict):
-                return [j]
+            if isinstance(j, list): return j
+            if isinstance(j, dict): return [j]
         except Exception:
-            return [x]  # treat as scalar string row (will be skipped later)
+            return [x]
     return [x]
 
 def _to_float(x: Any) -> Optional[float]:
@@ -382,13 +363,10 @@ def _to_int(x: Any) -> Optional[int]:
         return None
 
 def _coerce_date(s: Any) -> Optional[pd.Timestamp]:
-    if s is None:
-        return None
+    if s is None: return None
     try:
         dt = pd.to_datetime(s, errors="coerce")
-        if pd.isna(dt):
-            return None
-        # align to Harare if naive; respects your global TZ
+        if pd.isna(dt): return None
         try:
             return dt.tz_localize(TZ, nonexistent="shift_forward", ambiguous="NaT")
         except Exception:
@@ -396,10 +374,186 @@ def _coerce_date(s: Any) -> Optional[pd.Timestamp]:
     except Exception:
         return None
 
-# -------------------- Hardened sales series (drop-in replacement) --------------------
+# -----------------------------------------------------------------------------
+# Admin raw transactions extractor (row-level for PandasAI) + sample logging
+# -----------------------------------------------------------------------------
+def _paginate(sc_get, email, password, path, params=None, page_param="page", per_page=200, max_pages=50):
+    """Generic paginator for endpoints with page/per_page/meta"""
+    params = dict(params or {})
+    params.setdefault(page_param, 1)
+    params.setdefault("per_page", per_page)
+    page = 1
+    for _ in range(max_pages):
+        params[page_param] = page
+        raw = sc_get("GET", path, email, password, params=params)
+        yield raw
+        try:
+            meta = (raw or {}).get("meta") or {}
+            last_page = int(meta.get("last_page") or 0)
+            cur = int(meta.get("current_page") or page)
+            if last_page and cur >= last_page:
+                break
+            if not last_page and not raw:
+                break
+        except Exception:
+            break
+        page += 1
+
+def _normalize_line(order, item, tz=TZ) -> dict:
+    g = lambda o, *ks, default=None: next((o[k] for k in ks if isinstance(o, dict) and k in o), default)
+    to_f = lambda x: _to_float(x) or 0.0
+    to_i = lambda x: _to_int(x) or 0
+
+    order_id   = g(order, "id", "order_id", "uuid", "reference")
+    created_at = g(order, "created_at", "date", "ordered_at", "timestamp")
+    customer   = g(order, "customer_name", "customer", "buyer_name", "customer_reference")
+    payment    = g(order, "payment_method", "payment", "money_type")
+    branch     = g(order, "shop_name", "shop", "branch", "store")
+    status     = g(order, "status")
+    currency   = g(order, "currency")
+
+    prod_id    = g(item, "product_id", "item_id", "sku_id", "id")
+    prod_name  = g(item, "product_name", "name", "title", "sku")
+    qty        = to_i(g(item, "quantity", "qty", "units"))
+    unit_price = to_f(g(item, "unit_price", "price", "unitPrice"))
+    line_total = to_f(g(item, "line_total", "total", "amount", "revenue"))
+    cost_price = _to_float(g(item, "unit_cost", "cost_price", "cost"))  # optional
+
+    dt = _coerce_date(created_at)
+    revenue = line_total if line_total else (qty * unit_price)
+    gp = None
+    if cost_price is not None:
+        gp = float(revenue - qty * (cost_price or 0.0))
+
+    return {
+        "order_id": order_id,
+        "datetime": dt,
+        "date": dt.tz_convert(tz).date().isoformat() if dt is not None else None,
+        "customer": customer,
+        "payment_method": payment,
+        "branch": branch,
+        "status": status,
+        "currency": currency,
+        "product_id": prod_id,
+        "product": prod_name,
+        "quantity": qty,
+        "unit_price": unit_price,
+        "line_total": revenue,
+        "unit_cost": float(cost_price) if cost_price is not None else None,
+        "gross_profit": float(gp) if gp is not None else None,
+    }
+
+def fetch_transactions_df(email: str, password: str, t_start: pd.Timestamp, t_end: pd.Timestamp) -> pd.DataFrame:
+    """
+    Pull row-level order lines. Tries multiple likely endpoints, logs a sample for each,
+    flattens nested items, returns a clean DataFrame suitable for PandasAI.
+    """
+    CANDIDATES: Tuple[Tuple[str, str, str], ...] = (
+        ("/api/analytics/orders", "orders", "items"),
+        ("/api/orders", "data", "items"),                  # many APIs wrap orders under "data"
+        ("/api/analytics/transactions", "transactions", "items"),
+        ("/api/sales/transactions", "transactions", "lines"),
+    )
+    params = {
+        "start_date": t_start.strftime("%Y-%m-%d"),
+        "end_date": t_end.strftime("%Y-%m-%d"),
+        "include": "items",
+        "per_page": 200,
+    }
+
+    rows: List[dict] = []
+    for path, orders_key, items_key in CANDIDATES:
+        try:
+            # Non-paginated attempt
+            raw = sc_request("GET", path, email, password, params=params)
+            # Log a sharper sample for this endpoint (top-level)
+            logger.debug("TXN_PROBE_RAW %s -> keys=%s", path, list(raw.keys())[:10] if isinstance(raw, dict) else type(raw))
+
+            payload = raw.get("data") if isinstance(raw, dict) and isinstance(raw.get("data"), (dict, list)) else raw
+            orders = payload.get(orders_key) if isinstance(payload, dict) else payload
+            if orders:
+                orders_list = _to_list(orders)
+                if orders_list:
+                    # sample one order + items
+                    o0 = orders_list[0] if isinstance(orders_list[0], dict) else {}
+                    i0 = _to_list((o0 or {}).get(items_key))
+                    logger.debug("TXN_SAMPLE %s -> order_keys=%s; first_item_keys=%s",
+                                 path,
+                                 list(o0.keys())[:15] if isinstance(o0, dict) else type(o0),
+                                 (list(i0[0].keys())[:15] if i0 and isinstance(i0[0], dict) else "N/A"))
+                for o in orders_list:
+                    for it in _to_list((o or {}).get(items_key)):
+                        if isinstance(o, dict) and isinstance(it, dict):
+                            rows.append(_normalize_line(o, it))
+                if rows:
+                    break
+
+            # Try paginated shape
+            collected = 0
+            for page_raw in _paginate(sc_request, email, password, path, params=params):
+                logger.debug("TXN_PAGE %s meta=%s", path, (page_raw or {}).get("meta") if isinstance(page_raw, dict) else "N/A")
+                page_data = page_raw.get("data") if isinstance(page_raw, dict) and isinstance(page_raw.get("data"), (dict, list)) else page_raw
+                page_orders = page_data.get(orders_key) if isinstance(page_data, dict) else page_data
+                for o in _to_list(page_orders):
+                    for it in _to_list((o or {}).get(items_key)):
+                        if isinstance(o, dict) and isinstance(it, dict):
+                            rows.append(_normalize_line(o, it))
+                            collected += 1
+                if collected and collected >= 5000:  # safety cap
+                    break
+            if rows:
+                # Log a compact sample of flattened rows
+                logger.debug("TXN_FLAT_SAMPLE %s -> %s", path, json.dumps(rows[:2], default=str))
+                break
+        except Exception as e:
+            logger.debug(f"fetch_transactions_df: {path} probe failed: {e}")
+
+    if not rows:
+        logger.warning("No row-level endpoint found; returning an empty transactions frame (schema only).")
+        schema = {
+            "datetime": pd.Series(dtype="datetime64[ns]"),
+            "date": pd.Series(dtype="object"),
+            "order_id": pd.Series(dtype="object"),
+            "status": pd.Series(dtype="object"),
+            "customer": pd.Series(dtype="object"),
+            "branch": pd.Series(dtype="object"),
+            "payment_method": pd.Series(dtype="object"),
+            "currency": pd.Series(dtype="object"),
+            "product_id": pd.Series(dtype="object"),
+            "product": pd.Series(dtype="object"),
+            "quantity": pd.Series(dtype="float"),
+            "unit_price": pd.Series(dtype="float"),
+            "line_total": pd.Series(dtype="float"),
+            "unit_cost": pd.Series(dtype="float"),
+            "gross_profit": pd.Series(dtype="float"),
+        }
+        return pd.DataFrame(schema)
+
+    df = pd.DataFrame(rows)
+    df["datetime"] = pd.to_datetime(df["datetime"], errors="coerce")
+    try:
+        # Keep tz-naive for some plotting libs but deterministic in Harare
+        df["datetime"] = df["datetime"].dt.tz_convert(TZ).dt.tz_localize(None)
+    except Exception:
+        pass
+
+    for c in ("quantity", "unit_price", "line_total", "unit_cost", "gross_profit"):
+        if c in df.columns:
+            df[c] = pd.to_numeric(df[c], errors="coerce")
+
+    cols = [
+        "datetime", "date", "order_id", "status", "customer", "branch",
+        "payment_method", "currency", "product_id", "product",
+        "quantity", "unit_price", "line_total", "unit_cost", "gross_profit",
+    ]
+    df = df[[c for c in cols if c in df.columns]]
+
+    logger.debug("TXN_DF_COLUMNS %s", df.columns.tolist())
+    logger.debug("TXN_DF_HEAD %s", json.dumps(df.head(3).to_dict(orient="records"), default=str))
+    return df
 
 # -----------------------------------------------------------------------------
-# Admin KPI Engine (holistic view)
+# Admin KPI Engine (holistic view) — logs sample after each endpoint
 # -----------------------------------------------------------------------------
 class AdminAnalyticsEngine:
     """Single-tenant holistic admin analytics. No shop/brand filters; admin sees entire dataset."""
@@ -410,44 +564,27 @@ class AdminAnalyticsEngine:
         self.period = (period or "week").lower().strip()
         self.t_start, self.t_end, self.period_label = period_to_bounds(self.period)
 
-    # -------------------- helpers --------------------
     @staticmethod
     def _unwrap_data(payload: dict) -> dict:
         if isinstance(payload, dict):
-            # common backend pattern: {"status":"success","data":{...}}
             return payload.get("data") if isinstance(payload.get("data"), dict) else payload
         return {}
 
-    # -------------------- API pulls (no shop/brand params at all) --------------------
     def _dashboard(self) -> dict:
-        raw = sc_request(
-            "GET",
-            "/api/analytics/dashboard",
-            self.email,
-            self.password,
-            params={"period": self.period},
-        )
+        raw = sc_request("GET", "/api/analytics/dashboard", self.email, self.password, params={"period": self.period})
         data = self._unwrap_data(raw)
         emit_kpi_debug(self.tenant_key, "dashboard", data or raw or {})
+        # Log a friendly sample view:
+        logger.debug("SAMPLE /api/analytics/dashboard -> %s", json.dumps({k: data.get(k) for k in list(data.keys())[:10]}, default=str))
         return data or {}
 
     def _sales_series(self) -> pd.DataFrame:
-        """
-        Fetches /api/analytics/sales and returns a tidy daily series:
-        columns: ['_date','total_sales','total_orders','aov']
-        - Robust to strings/dicts/mixed payloads
-        - Converts numeric strings to floats/ints
-        - Skips malformed rows instead of crashing
-        """
         params = {
             "start_date": self.t_start.strftime("%Y-%m-%d"),
             "end_date": self.t_end.strftime("%Y-%m-%d"),
             "group_by": "day",
         }
         raw = sc_request("GET", "/api/analytics/sales", self.email, self.password, params=params)
-
-        # Expect raw like:
-        # {"status":"success","data":{"sales_over_time":[{"date":"YYYY-MM-DD","total_sales":"141.11","total_orders":3}], ...}}
         data = {}
         if isinstance(raw, dict):
             data = (raw.get("data") or raw) if isinstance(raw.get("data"), (dict, list)) else raw
@@ -458,78 +595,105 @@ class AdminAnalyticsEngine:
             except Exception:
                 data = {}
 
+        # log samples from top-level keys we expect
+        try:
+            so = data.get("sales_over_time")
+            pm = data.get("sales_by_payment_method")
+            cat = data.get("sales_by_category")
+            logger.debug("SAMPLE /api/analytics/sales -> sales_over_time[:2]=%s; sales_by_payment_method[:2]=%s; sales_by_category[:2]=%s",
+                         json.dumps((so or [])[:2]), json.dumps((pm or [])[:2]), json.dumps((cat or [])[:2]))
+        except Exception:
+            pass
+
         series = []
-        sales_ot = data.get("sales_over_time")
-        for r in _to_list(sales_ot):
+        for r in _to_list(data.get("sales_over_time")):
             if not isinstance(r, dict):
                 continue
             date_str = r.get("date") or r.get("day") or r.get("period")
             dt = _coerce_date(date_str)
             if dt is None:
                 continue
-
             total_sales = _to_float(r.get("total_sales") or r.get("total") or r.get("revenue"))
             total_orders = _to_int(r.get("total_orders") or r.get("orders") or r.get("count"))
             aov = _to_float(r.get("average_order_value") or r.get("aov"))
-
             if aov is None and total_sales is not None and (total_orders or 0) > 0:
                 aov = float(total_sales) / int(total_orders)
-
             series.append({
                 "_date": dt,
                 "total_sales": float(total_sales) if total_sales is not None else 0.0,
                 "total_orders": int(total_orders) if total_orders is not None else 0,
                 "aov": float(aov) if aov is not None else None,
             })
-
         df = pd.DataFrame(series)
         if df.empty:
             return pd.DataFrame(columns=["_date", "total_sales", "total_orders", "aov"])
-
         df = df.sort_values("_date").reset_index(drop=True)
         emit_kpi_debug(self.tenant_key, "sales_series_raw", (raw if isinstance(raw, dict) else {"raw": raw}))
+        logger.debug("SAMPLE sales_series_df.head -> %s", json.dumps(df.head(3).to_dict(orient="records"), default=str))
+        return df
+
+    def transactions_df(self) -> pd.DataFrame:
+        df = fetch_transactions_df(self.email, self.password, self.t_start, self.t_end)
+        emit_kpi_debug(self.tenant_key, "transactions_df_meta", {
+            "rows": int(len(df)),
+            "cols": list(df.columns),
+            "period": {"start": self.t_start.isoformat(), "end": self.t_end.isoformat()}
+        })
+        # already logged columns + head in fetch_transactions_df()
         return df
 
     def _products(self) -> dict:
         raw = sc_request(
-            "GET",
-            "/api/analytics/products",
-            self.email,
-            self.password,
-            params={"start_date": self.t_start.strftime("%Y-%m-%d"), "end_date": self.t_end.strftime("%Y-%m-%d")},
+            "GET", "/api/analytics/products", self.email, self.password,
+            params={"start_date": self.t_start.strftime("%Y-%m-%d"), "end_date": self.t_end.strftime("%Y-%m-%d")}
         )
         data = self._unwrap_data(raw)
         emit_kpi_debug(self.tenant_key, "products", data or raw or {})
+        # log sample leaderboards if present
+        keys = ["top_by_revenue","top_by_units","top_by_margin_value","top_by_margin_pct","bottom_by_revenue","loss_makers"]
+        sample = {k: (data.get(k) or [])[:2] for k in keys if isinstance(data.get(k), list)}
+        logger.debug("SAMPLE /api/analytics/products -> %s", json.dumps(sample))
         return data or {}
 
     def _customers(self) -> dict:
         raw = sc_request(
-            "GET",
-            "/api/analytics/customers",
-            self.email,
-            self.password,
-            params={"start_date": self.t_start.strftime("%Y-%m-%d"), "end_date": self.t_end.strftime("%Y-%m-%d")},
+            "GET", "/api/analytics/customers", self.email, self.password,
+            params={"start_date": self.t_start.strftime("%Y-%m-%d"), "end_date": self.t_end.strftime("%Y-%m-%d")}
         )
         data = self._unwrap_data(raw)
         emit_kpi_debug(self.tenant_key, "customers", data or raw or {})
+        # sample common shapes
+        sample = {
+            "top_customers_by_gp": (data.get("top_customers_by_gp") or [])[:2],
+            "at_risk": (data.get("at_risk") or [])[:2],
+            "new_customers": (data.get("new_customers") or [])[:2],
+            "summary": data.get("summary"),
+        }
+        logger.debug("SAMPLE /api/analytics/customers -> %s", json.dumps(sample))
         return data or {}
 
     def _inventory(self) -> dict:
         raw = sc_request("GET", "/api/analytics/inventory", self.email, self.password)
         data = self._unwrap_data(raw)
         emit_kpi_debug(self.tenant_key, "inventory", data or raw or {})
+        try:
+            items = data.get("products") or data.get("items") or data.get("snapshot") or []
+            logger.debug("SAMPLE /api/analytics/inventory -> %s", json.dumps((items or [])[:2], default=str))
+        except Exception:
+            pass
         return data or {}
 
     def _comparisons(self) -> dict:
         raw = sc_request(
-            "GET",
-            "/api/analytics/comparisons",
-            self.email,
-            self.password,
-            params={"start_date": self.t_start.strftime("%Y-%m-%d"), "end_date": self.t_end.strftime("%Y-%m-%d")},
+            "GET", "/api/analytics/comparisons", self.email, self.password,
+            params={"start_date": self.t_start.strftime("%Y-%m-%d"), "end_date": self.t_end.strftime("%Y-%m-%d")}
         )
         data = self._unwrap_data(raw)
         emit_kpi_debug(self.tenant_key, "comparisons", data or raw or {})
+        try:
+            logger.debug("SAMPLE /api/analytics/comparisons -> keys=%s", list(data.keys())[:15])
+        except Exception:
+            pass
         return data or {}
 
     # -------------------- deterministic snapshot --------------------
@@ -541,8 +705,6 @@ class AdminAnalyticsEngine:
         inv = self._inventory()
         comps = self._comparisons()
 
-        # Dashboard numbers can be at the top level or under 'metrics' etc.
-        # We’ll try a few keys before falling back to sums from sales_df.
         def _get_num(d: dict, *keys, default=0.0):
             for k in keys:
                 v = d.get(k)
@@ -558,10 +720,8 @@ class AdminAnalyticsEngine:
         transactions = int(_get_num(dash, "transactions", "orders", default=0.0))
 
         if (total_revenue == 0.0 or transactions == 0) and isinstance(sales_df, pd.DataFrame) and not sales_df.empty:
-            # Fallback: use the sales series we just fetched
             total_revenue = float(sales_df["total_sales"].sum())
             transactions = int(sales_df["total_orders"].sum())
-            # We don't have GP in this endpoint; leave as 0.0 if not provided by dashboard
 
         product_lb = {
             "top_by_revenue": prods.get("top_by_revenue") or prods.get("topRevenue") or [],
@@ -622,32 +782,21 @@ class AdminAnalyticsEngine:
         return json_safe(snapshot)
 
     def _temporal_patterns_from_sales(self, df: pd.DataFrame) -> Dict[str, Any]:
-        """
-        Build simple temporal readouts from the sales series returned by _sales_series():
-        - series: [{date, total_sales, total_orders, aov}]
-        - best_day_by_sales: highest total_sales by weekday
-        """
         if df is None or df.empty:
             return {"series": [], "best_day_by_sales": None}
-
         d = df.copy()
-        # Derive DOW and safe date strings
         d["dow"] = d["_date"].dt.day_name()
         d["date"] = d["_date"].dt.strftime("%Y-%m-%d")
-
-        # Aggregate by DOW using total_sales as revenue proxy
         g = d.groupby("dow", dropna=False).agg(
             total_sales=("total_sales", "sum"),
             total_orders=("total_orders", "sum"),
         ).reset_index()
-
         best_row = None if g.empty else g.loc[g["total_sales"].idxmax()]
         best_day = None if g.empty else {
             "day": str(best_row["dow"]),
             "total_sales": float(best_row["total_sales"]),
             "total_orders": int(best_row["total_orders"]),
         }
-
         series = d[["date", "total_sales", "total_orders", "aov"]].to_dict(orient="records")
         return {"series": series, "best_day_by_sales": best_day}
 
@@ -664,6 +813,7 @@ class AdminAnalyticsEngine:
             return sanitize_answer(text)
         except Exception:
             return "### Business Snapshot\n\n```\n" + json.dumps(json_safe(snapshot), indent=2) + "\n```"
+
 # -----------------------------------------------------------------------------
 # /chat — PandasAI first on sales series, else deterministic snapshot + narration
 # -----------------------------------------------------------------------------
@@ -686,15 +836,22 @@ def chat():
             return jsonify({"error": "Missing 'email' or 'password'."}), 400
 
         engine = AdminAnalyticsEngine(tenant_key, email, password, period)
+
+        # Build transactions_df now and place it in meta logs (useful for PandasAI later)
+        tdf = engine.transactions_df()
+
+        # For simple Q&A we still start with sales_df (fast + stable)
         sales_df = engine._sales_series()
-        if sales_df.empty:
+        if sales_df.empty and tdf.empty:
             snapshot = engine.build_snapshot()
             answer = engine.narrate(snapshot, user_question)
             return jsonify({"answer": sanitize_answer(answer), "meta": {"source": "analyst_fallback"}})
 
         try:
             logger.info(f"[{rid}] PandasAI attempt …")
-            pandas_agent = SmartDataframe(sales_df, config={
+            # If the question references products/items explicitly, switch to transactions_df
+            use_df = tdf if re.search(r"\b(product|sku|item|category|top\s*5|top\s*ten|by\s*revenue|by\s*units)\b", user_question, re.I) and not tdf.empty else sales_df
+            pandas_agent = SmartDataframe(use_df, config={
                 "llm": llm,
                 "response_parser": FlaskResponse,
                 "security": "none",
@@ -726,7 +883,6 @@ def chat():
                 return jsonify({"answer": data_uri, "meta": {"source": "pandasai"}})
 
             return jsonify({"answer": sanitize_answer(answer), "meta": {"source": "pandasai"}})
-
         except Exception:
             snapshot = engine.build_snapshot()
             answer = engine.narrate(snapshot, user_question)
@@ -747,8 +903,7 @@ def report():
         payload = request.get_json() or {}
         tenant_key = str(payload.get("tenant_key") or "admin")
         period = (payload.get("period") or "week").strip().lower()
-        email = payload.get("email")
-        password = payload.get("password")
+        email = payload.get("email"); password = payload.get("password")
         if not email or not password:
             return jsonify({"error": "Missing 'email' or 'password'."}), 400
 
@@ -773,8 +928,7 @@ def marketing():
         payload = request.get_json() or {}
         tenant_key = str(payload.get("tenant_key") or "admin")
         period = (payload.get("period") or "week").strip().lower()
-        email = payload.get("email")
-        password = payload.get("password")
+        email = payload.get("email"); password = payload.get("password")
         if not email or not password:
             return jsonify({"error": "Missing 'email' or 'password'."}), 400
 
@@ -799,8 +953,7 @@ def notify():
         payload = request.get_json() or {}
         tenant_key = str(payload.get("tenant_key") or "admin")
         period = (payload.get("period") or "week").strip().lower()
-        email = payload.get("email")
-        password = payload.get("password")
+        email = payload.get("email"); password = payload.get("password")
         if not email or not password:
             return jsonify({"error": "Missing 'email' or 'password'."}), 400
 
@@ -871,7 +1024,6 @@ def get_call_briefing():
     if not email or not password:
         return jsonify({"error": "Missing 'email' or 'password'."}), 400
     try:
-        # 1) Summarize call history
         call_history = []
         try:
             transcripts = db_ref.child(f"transcripts/{profile_id}").get()
@@ -880,7 +1032,6 @@ def get_call_briefing():
             logger.warning(f"Transcript fetch failed for '{profile_id}': {e}")
         memory_summary = _synthesize_history_summary(call_history)
 
-        # 2) Admin holistic KPI snapshot
         engine = AdminAnalyticsEngine(profile_id or "admin", email, password, period)
         kpi_snapshot = engine.build_snapshot()