src/rag_core.py

"""
Amazon RAG Core — v3
Extracted from notebook, importable module.
"""

import os
import re
import requests
import numpy as np
import pandas as pd
import faiss
from pathlib import Path
from typing import Optional, Dict, Any, List, Tuple
from sqlalchemy import create_engine, text as sql_text
from sentence_transformers import SentenceTransformer
from huggingface_hub import InferenceClient


# ═══════════════════════════════════════════
# CONFIG
# ═══════════════════════════════════════════
# Veri dosyaları bu .py dosyasının yanındaki klasörde aranır (src/)
_THIS_DIR = Path(__file__).resolve().parent
DB_PATH     = str(_THIS_DIR / "company_data.db")
INDEX_PATH  = str(_THIS_DIR / "rag.index")
CHUNKS_PATH = str(_THIS_DIR / "rag_chunks.parquet")

EMBED_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"

# Hugging Face Inference API config
# HF_TOKEN ortam değişkeninden alınır (Space'te "Secrets" olarak tanımlanmalı)
HF_MODEL = os.getenv("HF_MODEL", "Qwen/Qwen2.5-7B-Instruct")
HF_TOKEN = os.getenv("HF_TOKEN")
LLM_MODEL = HF_MODEL  # app.py ile uyumluluk için takma isim
# Legacy aliases (eski kodla uyumluluk)
OLLAMA_MODEL = HF_MODEL
OLLAMA_URL   = "https://api-inference.huggingface.co"

TABLES = [
    "amazon_orders_2023",
    "products_campaign_report",
    "business_reports_raw",
    "sp_advertised_product_report",
    "sp_search_terms",
    "df_time_series",
    "keepa_product_links",
]

ALLOWED_VIEWS = {
    "amazon_orders_2023__clean",
    "products_campaign_report__clean",
    "business_reports_raw__clean",
    "sp_advertised_product_report__clean",
    "sp_search_terms__clean",
    "df_time_series__clean",
    "keepa_product_links__clean",
}

SQLITE_BANNED_FUNCS = [
    r"\byear\s*\(",
    r"\bmonth\s*\(",
    r"\bdatepart\s*\(",
    r"\bextract\s*\(",
    r"\bto_char\s*\(",
]


# ═══════════════════════════════════════════
# EMBEDDER SINGLETON
# ═══════════════════════════════════════════
_embedder_cache: Dict[str, SentenceTransformer] = {}


def get_embedder(model_name: str = EMBED_MODEL_NAME) -> SentenceTransformer:
    if model_name not in _embedder_cache:
        _embedder_cache[model_name] = SentenceTransformer(model_name)
    return _embedder_cache[model_name]


# ═══════════════════════════════════════════
# LLM (Hugging Face Inference API) CHECK
# ═══════════════════════════════════════════
_hf_client: Optional[InferenceClient] = None


def get_hf_client() -> Optional[InferenceClient]:
    """HF Inference Client'ı tekil (singleton) olarak döndürür."""
    global _hf_client
    if _hf_client is None:
        if not HF_TOKEN:
            return None
        _hf_client = InferenceClient(model=HF_MODEL, token=HF_TOKEN, timeout=180)
    return _hf_client


def ensure_llm_up() -> bool:
    """LLM API'sinin erişilebilir olup olmadığını kontrol eder."""
    if not HF_TOKEN:
        return False
    try:
        client = get_hf_client()
        if client is None:
            return False
        _ = client.chat_completion(
            messages=[{"role": "user", "content": "ping"}],
            max_tokens=5,
        )
        return True
    except Exception:
        return False


# Eski isimle de çağrılabilsin (backward compat)
def ensure_ollama_up() -> bool:
    return ensure_llm_up()


# ═══════════════════════════════════════════
# SQL SAFETY
# ═══════════════════════════════════════════
def sanitize_sql_value(value: str) -> str:
    if value is None:
        return ""
    return re.sub(r"[^a-zA-Z0-9_\-]", "", str(value))


def validate_asin(value: str) -> Optional[str]:
    if not value:
        return None
    m = re.match(r"^B[0-9A-Z]{9}$", value.upper())
    return m.group(0) if m else None


def extract_cte_names(sql: str) -> set:
    s = sql.strip()
    if not re.match(r"(?is)^\s*with\b", s):
        return set()
    names = re.findall(
        r'(?is)\bwith\s+([a-zA-Z_][a-zA-Z0-9_]*)\s+as\s*\(|,\s*([a-zA-Z_][a-zA-Z0-9_]*)\s+as\s*\(',
        s,
    )
    out = set()
    for a, b in names:
        if a:
            out.add(a.lower())
        if b:
            out.add(b.lower())
    return out


# ═══════════════════════════════════════════
# CLEAN VIEWS
# ═══════════════════════════════════════════
def sanitize_col(name: str) -> str:
    s = name.strip().lower()
    s = re.sub(r"[^0-9a-zA-Z]+", "_", s)
    s = re.sub(r"_+", "_", s).strip("_")
    if not s:
        s = "col"
    if s[0].isdigit():
        s = "c_" + s
    return s


def build_clean_views(engine):
    mapping = {}
    with engine.connect() as conn:
        for t in TABLES:
            cols = conn.execute(sql_text(f"PRAGMA table_info('{t}')")).fetchall()
            orig_cols = [c[1] for c in cols]

            clean_map = {}
            used = set()
            for oc in orig_cols:
                base = sanitize_col(oc)
                cc = base
                k = 2
                while cc in used:
                    cc = f"{base}_{k}"
                    k += 1
                used.add(cc)
                clean_map[cc] = oc

            view = f"{t}__clean"
            conn.execute(sql_text(f'DROP VIEW IF EXISTS "{view}";'))
            select_parts = [f'[{orig}] AS "{clean}"' for clean, orig in clean_map.items()]
            sql = f'CREATE VIEW "{view}" AS SELECT {", ".join(select_parts)} FROM "{t}";'
            conn.execute(sql_text(sql))
            mapping[t] = clean_map
    return mapping


# ═══════════════════════════════════════════
# DATE PARSING
# ═══════════════════════════════════════════
# Turkish month names (kept for backward compatibility with Turkish queries)
MONTH_MAP_TR = {
    "ocak": 1, "şubat": 2, "mart": 3, "nisan": 4, "mayis": 5, "mayıs": 5,
    "haziran": 6, "temmuz": 7, "agustos": 8, "ağustos": 8, "eylul": 9, "eylül": 9,
    "ekim": 10, "kasım": 11, "aralık": 12,
}

# English month names
MONTH_MAP_EN = {
    "january": 1, "jan": 1,
    "february": 2, "feb": 2,
    "march": 3, "mar": 3,
    "april": 4, "apr": 4,
    "may": 5,
    "june": 6, "jun": 6,
    "july": 7, "jul": 7,
    "august": 8, "aug": 8,
    "september": 9, "sep": 9, "sept": 9,
    "october": 10, "oct": 10,
    "november": 11, "nov": 11,
    "december": 12, "dec": 12,
}

MONTH_MAP = {**MONTH_MAP_TR, **MONTH_MAP_EN}


def _find_year(q: str) -> Optional[int]:
    m = re.search(r"\b(20\d{2})\b", q)
    return int(m.group(1)) if m else None


def _find_quarter(q: str) -> Optional[int]:
    m = re.search(r"\bq([1-4])\b", q)
    if m:
        return int(m.group(1))
    # Turkish: "1. çeyrek"
    m2 = re.search(r"\b([1-4])\s*\.?\s*(çeyrek|ceyrek|quarter)\b", q)
    if m2:
        return int(m2.group(1))
    return None


def _find_last_n(q: str) -> Dict[str, Any]:
    # English: "last N days/weeks/months"
    m_en = re.search(r"\blast\s+(\d+)\s*(day|days|week|weeks|month|months)\b", q)
    if m_en:
        n = int(m_en.group(1))
        raw = m_en.group(2).lower()
        unit = "day" if "day" in raw else ("week" if "week" in raw else "month")
        return {"type": "last_n", "n": max(1, n), "unit": unit}
    # Turkish: "son N gün/hafta/ay"
    m_tr = re.search(r"\bson\s+(\d+)\s*(gün|gun|hafta|ay)\b", q)
    if not m_tr:
        if re.search(r"\b(last month|son ay)\b", q):
            return {"type": "last_n", "n": 1, "unit": "month"}
        if re.search(r"\b(last week|son hafta)\b", q):
            return {"type": "last_n", "n": 1, "unit": "week"}
        if re.search(r"\b(yesterday|son gün|son gun)\b", q):
            return {"type": "last_n", "n": 1, "unit": "day"}
        return {}
    n = int(m_tr.group(1))
    unit_raw = m_tr.group(2)
    unit = "day" if unit_raw in ["gün", "gun"] else ("week" if unit_raw == "hafta" else "month")
    return {"type": "last_n", "n": max(1, n), "unit": unit}


def _find_months(q: str) -> Dict[str, Any]:
    nums = [int(x) for x in re.findall(r"\b([1-9]|1[0-2])\b", q)]
    if (re.search(r"\b(ve|and|,)\b", q)) and len(nums) >= 2:
        seen = []
        for n in nums:
            if 1 <= n <= 12 and n not in seen:
                seen.append(n)
        if len(seen) >= 2:
            return {"type": "in", "months": seen[:12]}

    m = re.search(r"\b([1-9]|1[0-2])\s*[-–]\s*([1-9]|1[0-2])\b", q)
    if m:
        a, b = int(m.group(1)), int(m.group(2))
        return {"type": "between", "start": min(a, b), "end": max(a, b)}

    for name, idx in MONTH_MAP.items():
        if re.search(rf"\b{name}\b", q):
            return {"type": "single", "month": idx}

    # Turkish: "6. ay"
    m2 = re.search(r"\b([1-9]|1[0-2])\s*\.\s*ay\b", q)
    if m2:
        return {"type": "single", "month": int(m2.group(1))}

    # English: "first half" / "first 6 months"
    if re.search(r"\b(first half|h1|first 6 months)\b", q):
        return {"type": "between", "start": 1, "end": 6}
    if re.search(r"\b(second half|h2|last 6 months)\b", q):
        return {"type": "between", "start": 7, "end": 12}

    return {}


def parse_date_spec(question: str) -> Dict[str, Any]:
    q = (question or "").lower()
    year = _find_year(q)

    last_n = _find_last_n(q)
    if last_n:
        return {"type": "last_n", **last_n}

    qtr = _find_quarter(q)
    if year and qtr:
        start = (qtr - 1) * 3 + 1
        end   = start + 2
        return {"type": "absolute", "year": year, "months": {"type": "between", "start": start, "end": end}}

    # Turkish: "ilk 6 ay" / "ilk yarı"
    if year and re.search(r"\b(ilk 6 ay|ilk yarı|first half|h1)\b", q):
        return {"type": "absolute", "year": year, "months": {"type": "between", "start": 1, "end": 6}}

    months = _find_months(q)
    if year:
        return {"type": "absolute", "year": year, "months": months}
    return {"type": "absolute", "year": None, "months": months}


def build_where_from_date_spec(date_col: str, spec: Dict[str, Any], source_name: Optional[str] = None) -> str:
    if not spec:
        return ""

    stype = spec.get("type", "absolute")

    if stype == "last_n":
        n    = int(spec.get("n", 30))
        unit = spec.get("unit", "day")
        if unit == "day":
            mod = f"-{n} days"
        elif unit == "week":
            mod = f"-{n*7} days"
        else:
            mod = f"-{n} months"
        if not source_name:
            return ""
        max_expr = f"(SELECT MAX(date({date_col})) FROM {source_name})"
        return f"WHERE date({date_col}) BETWEEN date({max_expr}, '{mod}') AND date({max_expr})"

    clauses = []
    year   = spec.get("year")
    months = spec.get("months") or {}

    if year:
        clauses.append(f"strftime('%Y', date({date_col})) = '{year}'")

    if months.get("type") == "single":
        mm = f"{int(months['month']):02d}"
        clauses.append(f"strftime('%m', date({date_col})) = '{mm}'")
    elif months.get("type") == "between":
        a = int(months["start"])
        b = int(months["end"])
        clauses.append(f"CAST(strftime('%m', date({date_col})) AS INTEGER) BETWEEN {a} AND {b}")
    elif months.get("type") == "in":
        ms = [f"'{int(m):02d}'" for m in months["months"] if 1 <= int(m) <= 12]
        if ms:
            clauses.append(f"strftime('%m', date({date_col})) IN ({','.join(ms)})")

    if not clauses:
        return ""
    return "WHERE " + " AND ".join(clauses)


def detect_time_grain(q: str) -> Optional[str]:
    ql = (q or "").lower()
    if any(k in ql for k in ["daily", "per day", "day by day", "günlük", "gün baz", "güne göre"]):
        return "day"
    if any(k in ql for k in ["weekly", "per week", "week by week", "haftalık", "haftaya göre", "hafta baz"]):
        return "week"
    if any(k in ql for k in ["monthly", "per month", "month by month", "aylık", "aya göre", "ay baz"]):
        return "month"
    return None


def date_bucket_expr(date_col: str, grain: str) -> str:
    if grain == "day":
        return f"date({date_col})"
    if grain == "week":
        return f"(strftime('%Y', date({date_col})) || '-W' || strftime('%W', date({date_col})))"
    if grain == "month":
        return f"strftime('%Y-%m', date({date_col}))"
    return f"date({date_col})"


# ═══════════════════════════════════════════
# LocalRAG CLASS
# ═══════════════════════════════════════════
class LocalRAG:
    def __init__(self, engine):
        self.engine   = engine
        self.embedder = get_embedder(EMBED_MODEL_NAME)
        self.index    = faiss.read_index(INDEX_PATH)
        self.chunks   = pd.read_parquet(CHUNKS_PATH)
        self._cols_cache: Dict[str, set] = {}

    # ── Schema ──────────────────────────────
    def get_view_cols(self, view_name: str) -> set:
        if view_name in self._cols_cache:
            return self._cols_cache[view_name]
        try:
            with self.engine.connect() as conn:
                rows = conn.execute(sql_text(f"PRAGMA table_info({view_name});")).mappings().all()
            cols = {r["name"] for r in rows if "name" in r}
        except Exception:
            cols = set()
        self._cols_cache[view_name] = cols
        return cols

    def has_col(self, view_name: str, col: str) -> bool:
        return col in self.get_view_cols(view_name)

    # ── Orders Union CTE ─────────────────────
    def build_orders_union_cte(self, alias: str = "orders_union") -> Tuple[str, List[str]]:
        left_view  = "amazon_orders_2023__clean"
        right_view = "df_time_series__clean"
        left_cols  = self.get_view_cols(left_view)
        right_cols = self.get_view_cols(right_view)

        canonical = [
            "purchase_date", "amazon_order_id", "quantity", "item_price",
            "item_tax", "shipping_price", "ship_country", "ship_city",
            "ship_state", "order_status", "fulfillment_channel",
            "ship_service_level", "asin", "sku", "sales_channel",
            "order_channel", "is_business_order", "buyer_company_name",
            "unit_cost", "total_cost", "estimated_profit",
        ]

        def sel_list(view_cols: set) -> str:
            parts = []
            for c in canonical:
                parts.append(c if c in view_cols else f"NULL AS {c}")
            return ",\n        ".join(parts)

        cte = f"""WITH {alias} AS (
    SELECT
        {sel_list(left_cols)}
    FROM {left_view}

    UNION ALL

    SELECT
        {sel_list(right_cols)}
    FROM {right_view}
)"""
        return cte, canonical

    # ── RAG Retrieve ─────────────────────────
    def retrieve(self, query: str, k: int = 6) -> List[Dict]:
        q      = self.embedder.encode([query], convert_to_numpy=True, normalize_embeddings=True).astype(np.float32)
        scores, ids = self.index.search(q, k)
        out = []
        for score, idx in zip(scores[0], ids[0]):
            if idx < 0:
                continue
            row = self.chunks.iloc[int(idx)]
            out.append({"score": float(score), "table": row["table"], "text": row["text"]})
        return out

    # ── LLM (HF Inference API) ───────────────
    def ask_ollama(self, prompt: str, temperature: float = 0.1, force_json: bool = False) -> str:
        """
        Hugging Face Inference API üzerinden LLM'ye istek atar.
        İsim 'ask_ollama' olarak kaldı — kodun geri kalanı bu ismi kullanıyor.
        """
        client = get_hf_client()
        if client is None:
            return "[LLM error: HF_TOKEN tanımlı değil. Space Settings > Secrets'a HF_TOKEN ekleyin.]"

        # force_json için prompt'a ek talimat veriyoruz (HF'de native JSON mode yok)
        if force_json:
            prompt = (
                prompt
                + "\n\nIMPORTANT: Respond ONLY with valid JSON. No markdown, no explanation, no code fences."
            )

        try:
            out = client.chat_completion(
                messages=[{"role": "user", "content": prompt}],
                temperature=max(temperature, 0.01),  # bazı providerlar 0'ı kabul etmiyor
                max_tokens=1024,
            )
            return (out.choices[0].message.content or "").strip()
        except Exception as e:
            return f"[LLM error: {e}]"

    # Yeni isim (okunabilirlik için)
    def ask_llm(self, prompt: str, temperature: float = 0.1, force_json: bool = False) -> str:
        return self.ask_ollama(prompt, temperature=temperature, force_json=force_json)

    # ── Intent Detection ─────────────────────
    def looks_like_sql_question(self, q: str) -> bool:
        ql = (q or "").lower()
        agg = [
            # English
            "total", "sum", "count", "how many", "number of", "quantity",
            "average", "avg", "max", "min", "median",
            "ratio", "percent", "%", "share", "rate",
            "trend", "daily", "weekly", "monthly", "yearly", "annual",
            "increase", "decrease", "change", "delta", "growth",
            "top", "rank", "ranking", "sort", "order by",
            "breakdown", "distribution", "segment", "group", "group by",
            "filter", "where", "between", "last", "first", "rolling", "moving",
            # Turkish (kept for compatibility)
            "toplam", "kaç", "adet", "miktar", "sayı", "sayısı",
            "ortalama", "maks", "oran", "yüzde", "pay",
            "artış", "azalış", "değişim", "fark",
            "en çok", "en az", "sıralama",
            "kırılım", "dağılım", "gruplama",
            "filtre", "arasında", "son", "ilk",
            "aylık", "haftalık", "günlük", "yıl",
        ]
        domain = [
            # English
            "order", "orders", "fulfillment", "revenue", "sales", "profit",
            "cost", "margin", "shipping",
            "sku", "asin", "product",
            "sessions", "page views", "buy box", "featured offer",
            "ordered_product_sales", "total_order_items", "units_ordered",
            "campaign", "portfolio", "impressions", "clicks", "ctr", "cpc", "spend",
            "acos", "roas", "conversion",
            "business", "b2b", "b2c",
            "ship city", "ship country", "ship state", "state", "country", "city",
            "parent_asin", "child_asin",
            # Turkish (kept for compatibility)
            "sipariş", "ciro", "gelir", "kâr", "kar",
            "maliyet", "gider", "masraf", "marj",
            "kârlılık", "karlılık", "kazanç",
        ]
        return any(k in ql for k in agg) or any(k in ql for k in domain)

    # ── Metric Expressions ───────────────────
    def _revenue_expr(self)       -> str: return "SUM(COALESCE(item_price,0))"
    def _shipping_expr(self)      -> str: return "SUM(COALESCE(shipping_price,0))"
    def _total_revenue_expr(self) -> str: return f"({self._revenue_expr()} + {self._shipping_expr()})"
    def _orders_expr(self)        -> str: return "COUNT(DISTINCT amazon_order_id)"
    def _units_expr(self)         -> str: return "SUM(COALESCE(quantity,0))"
    def _aov_expr(self)           -> str: return f"{self._revenue_expr()} / NULLIF({self._orders_expr()},0)"
    def _asp_expr(self)           -> str: return f"{self._revenue_expr()} / NULLIF({self._units_expr()},0)"
    def _profit_expr(self)        -> str: return "SUM(COALESCE(estimated_profit,0))"
    def _total_cost_expr(self)    -> str: return "SUM(COALESCE(total_cost,0))"
    def _profit_margin_expr(self) -> str:
        return f"ROUND({self._profit_expr()} * 100.0 / NULLIF({self._revenue_expr()},0), 2)"

    # ── Dimension Detection ──────────────────
    def _dim_in_question(self, q: str) -> Optional[str]:
        ql = (q or "").lower()

        def has_word(w: str) -> bool:
            return re.search(rf"\b{re.escape(w)}\b", ql) is not None

        if has_word("country") or "by country" in ql or "ülkeye göre" in ql or has_word("ülke"):
            return "ship_country"
        if has_word("city") or "by city" in ql or "şehre göre" in ql or has_word("şehir"):
            return "ship_city"
        if has_word("state") or "by state" in ql or has_word("eyalet") or "eyalete göre" in ql:
            return "ship_state"
        if has_word("fulfillment") or has_word("fba") or has_word("fbm"):
            return "fulfillment_channel"
        if "shipping method" in ql or "service level" in ql or "kargo yöntemi" in ql:
            return "ship_service_level"
        if has_word("channel") or "by channel" in ql or "kanala göre" in ql or has_word("kanal"):
            return "sales_channel"
        if has_word("sku"):
            return "sku"
        if has_word("asin"):
            return "asin"
        return None

    def _safe_dim(self, dim: str, canonical_cols: List[str]) -> Optional[str]:
        if dim is None:
            return None
        if dim in canonical_cols:
            return dim
        if dim == "sales_channel":
            if "order_channel" in canonical_cols:
                return "order_channel"
            if "fulfillment_channel" in canonical_cols:
                return "fulfillment_channel"
            return None
        return None

    def _build_breakdown_sql(self, cte_sql, where_sql, metric_selects, group_cols, order_by, limit=50):
        select_parts = list(group_cols)
        for alias, expr in metric_selects:
            select_parts.append(f"{expr} AS {alias}")
        select_clause = ",\n       ".join(select_parts)
        group_clause  = "GROUP BY " + ", ".join(group_cols) if group_cols else ""
        lim           = f"LIMIT {int(limit)}" if (limit and limit > 0) else ""
        return f"""{cte_sql}
SELECT {select_clause}
FROM orders_union
{where_sql}
{group_clause}
ORDER BY {order_by}
{lim}""".strip()

    # ── Business Reports Helpers ─────────────
    def _br_bucket_in_question(self, q: str) -> Optional[str]:
        ql = (q or "").lower()
        if any(k in ql for k in ["daily", "per day", "günlük", "gün gün"]):
            return "strftime('%Y-%m-%d', date(report_date))"
        if any(k in ql for k in ["weekly", "per week", "haftalık", "week"]):
            return "strftime('%Y-%W', date(report_date))"
        if any(k in ql for k in ["monthly", "per month", "aylık", "ay ay", "month"]):
            return "strftime('%Y-%m', date(report_date))"
        return None

    def _br_dim_in_question(self, q: str) -> Optional[str]:
        ql = (q or "").lower()
        if any(k in ql for k in ["child asin", "child_asin", "variant asin", "varyant asin"]):
            return "child_asin"
        if any(k in ql for k in ["parent asin", "parent_asin", "parent", "ana asin"]):
            return "parent_asin"
        if "sku" in ql:
            return "sku"
        if any(k in ql for k in ["title", "product title", "başlık", "ürün adı"]):
            return "title"
        return None

    def _br_device_mode(self, q: str) -> str:
        ql = (q or "").lower()
        if any(k in ql for k in ["mobile", "app", "mobil", "mobile_app"]):
            return "mobile_app"
        if any(k in ql for k in ["browser", "web", "tarayıcı"]):
            return "browser"
        return "total"

    def _br_is_b2b(self, q: str) -> bool:
        return any(k in (q or "").lower() for k in ["b2b", "business"])

    def _br_extract_asin(self, question: str) -> Optional[str]:
        m = re.search(r"\bB[0-9A-Z]{9}\b", (question or "").upper())
        return validate_asin(m.group(0)) if m else None

    def _where_and(self, where_sql: str, extra_clause: str) -> str:
        if not extra_clause:
            return where_sql or ""
        if where_sql and where_sql.strip().lower().startswith("where"):
            return where_sql + " AND " + extra_clause
        return "WHERE " + extra_clause

    def _br_metric_sql(self, metric: str, device_mode: str, b2b: bool) -> str:
        suffix = "_b2b" if b2b else ""
        dev    = device_mode
        if metric == "sessions":
            col = f"sessions_{dev}{suffix}" if dev != "total" else f"sessions_total{suffix}"
            return f"SUM(COALESCE({col},0))"
        if metric == "pageviews":
            col = f"page_views_{dev}{suffix}" if dev != "total" else f"page_views_total{suffix}"
            return f"SUM(COALESCE({col},0))"
        if metric == "buybox":
            return f"AVG(COALESCE(featured_offer_buy_box_percentage{suffix},0))"
        if metric == "units":
            return f"SUM(COALESCE(units_ordered{suffix},0))"
        if metric == "sales":
            return f"SUM(COALESCE(ordered_product_sales{suffix},0))"
        if metric == "order_items":
            return f"SUM(COALESCE(total_order_items{suffix},0))"
        if metric == "conversion":
            return f"AVG(COALESCE(unit_session_percentage{suffix},0))"
        return "NULL"

    def _br_build_breakdown_sql(self, where_sql, select_exprs, group_col=None, order_by=None, limit=50):
        sel_parts     = []
        group_by_cols = []
        if group_col:
            sel_parts.append(group_col)
            if " AS " in group_col.upper():
                group_by_cols.append(group_col.split(" AS ")[0].strip())
            else:
                group_by_cols.append(group_col.strip())
        for alias, expr in select_exprs:
            sel_parts.append(f"{expr} AS {alias}")
        select_sql = ",\n  ".join(sel_parts)
        sql = f"""SELECT
  {select_sql}
FROM business_reports_raw__clean
{where_sql}""".strip()
        if group_by_cols:
            sql += f"\nGROUP BY {', '.join(group_by_cols)}"
        if order_by:
            sql += f"\nORDER BY {order_by}"
        if limit and limit > 0:
            sql += f"\nLIMIT {int(limit)}"
        return sql

    # ── Template SQL ─────────────────────────
    def template_sql(self, question: str) -> Optional[str]:
        q = (question or "").lower().strip()
        orders_union_cte, canonical_cols = self.build_orders_union_cte()
        spec         = parse_date_spec(question)
        where_orders = build_where_from_date_spec("purchase_date", spec, source_name="orders_union")

        # Smart filtering based on query intent
        _cancel_keywords = [
            "return", "refund", "cancel", "cancelled", "canceled",
            "iade", "iptal", "iade edilen", "iptal edilen",
        ]
        _is_cancel_query = any(k in q for k in _cancel_keywords)

        if _is_cancel_query:
            # Returns/cancellations → show non-shipped orders
            where_orders = self._where_and(where_orders, "order_status != 'Shipped'")
        else:
            # Revenue/orders → shipped only + price filters + outlier cap
            where_orders = self._where_and(where_orders, "order_status = 'Shipped'")
            where_orders = self._where_and(where_orders, "item_price > 0")
            where_orders = self._where_and(where_orders, "item_price <= 324.55")

        grain  = detect_time_grain(question)
        bucket = date_bucket_expr("purchase_date", grain) if grain else None
        dim_raw = self._dim_in_question(question)
        dim     = self._safe_dim(dim_raw, canonical_cols)

        # ── BUSINESS REPORTS ──────────────────
        br_keywords = [
            "session", "sessions", "oturum", "trafik",
            "page view", "pageviews", "sayfa", "görüntülenme",
            "buy box", "buybox", "featured offer",
            "unit session", "conversion", "dönüşüm",
            "ordered_product_sales",
            "total_order_items", "sipariş kalemi",
            "units_ordered",
            "parent_asin", "child_asin",
        ]
        if any(k in q for k in br_keywords):
            where_br  = build_where_from_date_spec("report_date", spec, source_name="business_reports_raw__clean")
            dim_br    = self._br_dim_in_question(question)
            bucket_br = self._br_bucket_in_question(question)
            dev       = self._br_device_mode(question)
            is_b2b    = self._br_is_b2b(question)

            asin_val = self._br_extract_asin(question)
            if asin_val:
                where_br = self._where_and(where_br, f"child_asin = '{sanitize_sql_value(asin_val)}'")

            if any(k in q for k in ["page views per session", "pv per session", "views per session", "pv/session"]):
                pv   = self._br_metric_sql("pageviews", dev, is_b2b)
                ss   = self._br_metric_sql("sessions", dev, is_b2b)
                expr = f"({pv}) * 1.0 / NULLIF(({ss}),0)"
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("pageviews_per_session", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("pageviews_per_session", expr)], group_col=dim_br, order_by="pageviews_per_session DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("pageviews_per_session", expr)])

            if any(k in q for k in ["revenue per session", "sales per session", "revenue/session"]):
                sales = self._br_metric_sql("sales", "total", is_b2b)
                ss    = self._br_metric_sql("sessions", dev, is_b2b)
                expr  = f"({sales}) * 1.0 / NULLIF(({ss}),0)"
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("revenue_per_session", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("revenue_per_session", expr)], group_col=dim_br, order_by="revenue_per_session DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("revenue_per_session", expr)])

            if any(k in q for k in ["b2b revenue share", "b2b sales share", "b2b revenue %"]):
                b2b_sales = self._br_metric_sql("sales", "total", True)
                tot_sales = self._br_metric_sql("sales", "total", False)
                expr      = f"100.0 * ({b2b_sales}) / NULLIF(({tot_sales}),0)"
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("b2b_revenue_share_pct", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("b2b_revenue_share_pct", expr)], group_col=dim_br, order_by="b2b_revenue_share_pct DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("b2b_revenue_share_pct", expr)])

            if any(k in q for k in ["session", "sessions", "oturum", "trafik"]):
                expr = self._br_metric_sql("sessions", dev, is_b2b)
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("sessions", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("sessions", expr)], group_col=dim_br, order_by="sessions DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("sessions", expr)])

            if any(k in q for k in ["page view", "pageviews", "sayfa", "görüntülenme"]):
                expr = self._br_metric_sql("pageviews", dev, is_b2b)
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("page_views", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("page_views", expr)], group_col=dim_br, order_by="page_views DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("page_views", expr)])

            if any(k in q for k in ["unit session", "conversion", "dönüşüm", "unit_session_percentage"]):
                expr = self._br_metric_sql("conversion", "total", is_b2b)
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("unit_session_percentage", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("unit_session_percentage", expr)], group_col=dim_br, order_by="unit_session_percentage DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("unit_session_percentage", expr)])

            if any(k in q for k in ["buy box", "buybox", "featured offer"]):
                expr = self._br_metric_sql("buybox", "total", is_b2b)
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("buybox_avg", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("buybox_avg", expr)], group_col=dim_br, order_by="buybox_avg DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("buybox_avg", expr)])

            if any(k in q for k in ["ordered_product_sales", "sales", "satış"]):
                expr = self._br_metric_sql("sales", "total", is_b2b)
                if bucket_br:
                    return self._br_build_breakdown_sql(where_br, [("ordered_product_sales", expr)], group_col=f"{bucket_br} AS period", order_by="period ASC", limit=0)
                if dim_br:
                    return self._br_build_breakdown_sql(where_br, [("ordered_product_sales", expr)], group_col=dim_br, order_by="ordered_product_sales DESC", limit=50)
                return self._br_build_breakdown_sql(where_br, [("ordered_product_sales", expr)])

            return None

        # ── ORDERS: Profit ────────────────────
        if any(k in q for k in ["profit", "margin", "net income", "earnings", "kâr", "kar", "kârlılık", "karlılık", "kazanç", "marj"]):
            if any(k in q for k in ["margin", "marj", "ratio", "percent", "oran"]):
                if dim:
                    return self._build_breakdown_sql(orders_union_cte, where_orders, [("profit_margin_pct", self._profit_margin_expr())], [dim], "profit_margin_pct DESC", 50)
                if bucket:
                    return self._build_breakdown_sql(orders_union_cte, where_orders, [("profit_margin_pct", self._profit_margin_expr())], [f"{bucket} AS period"], "period ASC", 0)
                return f"""{orders_union_cte}
SELECT {self._profit_margin_expr()} AS profit_margin_pct
FROM orders_union
{where_orders}""".strip()
            if dim:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("profit", self._profit_expr())], [dim], "profit DESC", 50)
            if bucket:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("profit", self._profit_expr())], [f"{bucket} AS period"], "period ASC", 0)
            return f"""{orders_union_cte}
SELECT {self._profit_expr()} AS profit,
       {self._revenue_expr()} AS revenue,
       {self._total_cost_expr()} AS total_cost,
       {self._profit_margin_expr()} AS profit_margin_pct
FROM orders_union
{where_orders}""".strip()

        # ── ORDERS: Cost ──────────────────────
        if any(k in q for k in ["cost", "costs", "expense", "maliyet", "gider", "masraf"]):
            if dim:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("total_cost", self._total_cost_expr())], [dim], "total_cost DESC", 50)
            if bucket:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("total_cost", self._total_cost_expr())], [f"{bucket} AS period"], "period ASC", 0)
            return f"""{orders_union_cte}
SELECT {self._total_cost_expr()} AS total_cost
FROM orders_union
{where_orders}""".strip()

        # ── ORDERS: Revenue ───────────────────
        if any(k in q for k in ["revenue", "sales amount", "income", "ciro", "gelir", "satış tutarı"]):
            asin_match = re.search(r"\bB[0-9A-Z]{9}\b", question.upper())
            if asin_match:
                asin_val = validate_asin(asin_match.group(0))
                if asin_val:
                    safe_asin  = sanitize_sql_value(asin_val)
                    where_asin = (where_orders + f" AND asin = '{safe_asin}'") if where_orders else f"WHERE asin = '{safe_asin}'"
                    if any(k in q for k in ["b2b", "business"]) and ("is_business_order" in canonical_cols or "buyer_company_name" in canonical_cols):
                        b2b_filter = "COALESCE(is_business_order,0)=1" if "is_business_order" in canonical_cols else "buyer_company_name IS NOT NULL AND TRIM(buyer_company_name) <> ''"
                        where_asin = where_asin + " AND " + b2b_filter
                    return f"""{orders_union_cte}
SELECT
  '{safe_asin}' AS asin,
  COALESCE({self._revenue_expr()}, 0) AS revenue,
  COALESCE({self._units_expr()}, 0) AS units,
  {self._orders_expr()} AS orders
FROM orders_union
{where_asin}""".strip()

            if any(k in q for k in ["b2b", "business"]) and ("is_business_order" in canonical_cols or "buyer_company_name" in canonical_cols):
                b2b_filter = "COALESCE(is_business_order,0)=1" if "is_business_order" in canonical_cols else "buyer_company_name IS NOT NULL AND TRIM(buyer_company_name) <> ''"
                where_b2b  = (where_orders + " AND " + b2b_filter) if where_orders else ("WHERE " + b2b_filter)
                if dim:
                    dim_final = dim if dim != "sales_channel" else ("sales_channel" if "sales_channel" in canonical_cols else ("order_channel" if "order_channel" in canonical_cols else "fulfillment_channel"))
                    return self._build_breakdown_sql(orders_union_cte, where_b2b, [("b2b_revenue", self._revenue_expr())], [dim_final], "b2b_revenue DESC", 50)
                if bucket:
                    return self._build_breakdown_sql(orders_union_cte, where_b2b, [("b2b_revenue", self._revenue_expr())], [f"{bucket} AS period"], "period ASC", 0)
                return f"""{orders_union_cte}
SELECT {self._revenue_expr()} AS b2b_revenue
FROM orders_union
{where_b2b}""".strip()

            if dim:
                dim_final = dim if dim != "sales_channel" else ("sales_channel" if "sales_channel" in canonical_cols else ("order_channel" if "order_channel" in canonical_cols else "fulfillment_channel"))
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("revenue", self._revenue_expr())], [dim_final], "revenue DESC", 50)
            if bucket:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("revenue", self._revenue_expr())], [f"{bucket} AS period"], "period ASC", 0)
            return f"""{orders_union_cte}
SELECT {self._revenue_expr()} AS revenue
FROM orders_union
{where_orders}""".strip()

        # Total Revenue (incl. shipping)
        if any(k in q for k in ["total revenue", "gross revenue", "shipping included", "shipping dahil", "toplam gelir", "toplam ciro"]):
            if bucket:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("total_revenue", self._total_revenue_expr())], [f"{bucket} AS period"], "period ASC", 0)
            return f"""{orders_union_cte}
SELECT {self._total_revenue_expr()} AS total_revenue
FROM orders_union
{where_orders}""".strip()

        # AOV
        if any(k in q for k in ["aov", "average order value", "order value", "ortalama sepet", "sipariş başı gelir"]):
            return f"""{orders_union_cte}
SELECT {self._aov_expr()} AS aov
FROM orders_union
{where_orders}""".strip()

        # ASP
        if any(k in q for k in ["asp", "average selling price", "average price", "ortalama satış fiyatı"]):
            return f"""{orders_union_cte}
SELECT {self._asp_expr()} AS asp
FROM orders_union
{where_orders}""".strip()

        # Shipping
        if any(k in q for k in ["shipping revenue", "shipping total", "shipping cost", "kargo geliri", "kargo ücreti"]):
            return f"""{orders_union_cte}
SELECT {self._shipping_expr()} AS shipping_total
FROM orders_union
{where_orders}""".strip()

        # Orders count
        if any(k in q for k in ["order count", "number of orders", "how many orders", "sipariş", "order", "kaç sipariş", "sipariş sayısı"]):
            if dim:
                dim_final = dim if dim != "sales_channel" else ("sales_channel" if "sales_channel" in canonical_cols else ("order_channel" if "order_channel" in canonical_cols else "fulfillment_channel"))
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("orders", self._orders_expr())], [dim_final], "orders DESC", 50)
            if bucket:
                return self._build_breakdown_sql(orders_union_cte, where_orders, [("orders", self._orders_expr())], [f"{bucket} AS period"], "period ASC", 0)
            return f"""{orders_union_cte}
SELECT {self._orders_expr()} AS orders
FROM orders_union
{where_orders}""".strip()

        # Units
        if any(k in q for k in ["units", "quantity", "items sold", "adet", "miktar", "birim"]):
            return f"""{orders_union_cte}
SELECT {self._units_expr()} AS units
FROM orders_union
{where_orders}""".strip()

        # ── ADS ───────────────────────────────
        where_ads = build_where_from_date_spec("start_date_date", spec, source_name="products_campaign_report__clean")
        where_sp  = build_where_from_date_spec("start_date_date", spec, source_name="sp_search_terms__clean")

        if any(k in q for k in ["spend", "ad spend", "advertising spend", "harcama", "reklam harcaması"]):
            return f"SELECT SUM(spend) AS spend_total FROM products_campaign_report__clean {where_ads}".strip()
        if any(k in q for k in ["impressions", "impression", "gösterim"]):
            return f"SELECT SUM(impressions) AS impressions_total FROM products_campaign_report__clean {where_ads}".strip()
        if any(k in q for k in ["clicks", "click", "tıklama"]):
            return f"SELECT SUM(clicks) AS clicks_total FROM products_campaign_report__clean {where_ads}".strip()
        if any(k in q for k in ["ctr", "click-through rate", "click thru rate"]):
            return f"SELECT AVG(click_thru_rate_ctr) AS ctr_avg FROM products_campaign_report__clean {where_ads}".strip()
        if any(k in q for k in ["cpc", "cost per click"]):
            return f"SELECT AVG(cost_per_click_cpc) AS cpc_avg FROM products_campaign_report__clean {where_ads}".strip()
        if "acos" in q:
            return f"SELECT AVG(total_advertising_cost_of_sales_acos) AS acos_avg FROM products_campaign_report__clean {where_ads}".strip()
        if "roas" in q:
            return f"SELECT AVG(total_return_on_advertising_spend_roas) AS roas_avg FROM products_campaign_report__clean {where_ads}".strip()
        if any(k in q for k in ["search term", "search terms", "customer search", "arama terimi"]):
            return f"""SELECT customer_search_term,
       SUM(spend) AS spend_total,
       SUM(day_7_total_sales) AS sales_7d
FROM sp_search_terms__clean
{where_sp}
GROUP BY customer_search_term
ORDER BY spend_total DESC
LIMIT 50""".strip()

        # ── KEEPA ─────────────────────────────
        if any(k in q for k in ["review", "rating", "star", "yorum", "puan"]):
            return """SELECT AVG(reviews_rating) AS avg_rating,
       AVG(reviews_review_count) AS avg_review_count
FROM keepa_product_links__clean""".strip()
        if any(k in q for k in ["sales rank", "bsr", "best seller rank", "satış sıralaması"]):
            return "SELECT AVG(sales_rank_current) AS sales_rank_avg FROM keepa_product_links__clean"
        if any(k in q for k in ["buy box current", "buybox current", "buy box price"]):
            return "SELECT AVG(buy_box_current) AS buybox_current_avg FROM keepa_product_links__clean"

        return None

    # ── SQL Validation ────────────────────────
    def validate_sql(self, sql: str):
        if not sql:
            return "Empty SQL"
        s = sql.strip()
        if not (re.match(r"(?is)^\s*select\b", s) or re.match(r"(?is)^\s*with\b", s)):
            return "Only SELECT/WITH SELECT allowed"
        low = s.lower()
        for pat in SQLITE_BANNED_FUNCS:
            if re.search(pat, low):
                return "SQLite unsupported function used (use strftime/date instead)"
        cte_names = extract_cte_names(s)
        refs      = re.findall(r'(?is)\bfrom\s+("?[a-zA-Z0-9_]+"?)|\bjoin\s+("?[a-zA-Z0-9_]+"?)', low)
        mentioned = set()
        for a, b in refs:
            t = (a or b or "").strip().strip('"')
            if t:
                mentioned.add(t)
        allowed_lower = {v.lower() for v in ALLOWED_VIEWS}
        for t in mentioned:
            if t in cte_names:
                continue
            if t.endswith("__clean"):
                if t not in allowed_lower:
                    return f"Disallowed table/view: {t}"
            else:
                if t not in cte_names and t != "orders_union":
                    return f"Disallowed reference: {t}"
        return None

    # ── SQL Execute ───────────────────────────
    def run_sql_safe(self, sql: str):
        err = self.validate_sql(sql)
        if err:
            return {"error": err, "sql": sql}
        s = sql.strip().rstrip(";")
        try:
            with self.engine.connect() as conn:
                rows = conn.execute(sql_text(s)).mappings().all()
                rows = [dict(r) for r in rows]
            return {"sql": s, "result": rows}
        except Exception as e:
            return {"error": str(e), "sql": s}

    # ── Result Summary ────────────────────────
    def summarize_sql_result(self, question, sql, result) -> str:
        if not result:
            return "No matching records found."
        if len(result) == 1 and isinstance(result[0], dict) and len(result[0]) == 1:
            k = list(result[0].keys())[0]
            return self._format_metric(k, result[0][k])
        if len(result) == 1 and isinstance(result[0], dict):
            return " | ".join(self._format_metric(k, v) for k, v in result[0].items())
        if len(result) <= 10:
            header = list(result[0].keys())
            lines  = [" | ".join(str(r.get(h, "")) for h in header) for r in result]
            return f"{len(result)} rows:\n" + " | ".join(header) + "\n" + "\n".join(lines)
        header  = list(result[0].keys())
        first_3 = [" | ".join(str(r.get(h, "")) for h in header) for r in result[:3]]
        return (
            f"{len(result)} rows returned. First 3:\n"
            + " | ".join(header) + "\n"
            + "\n".join(first_3)
            + f"\n... and {len(result)-3} more rows."
        )

    def _format_metric(self, key, value) -> str:
        if value is None:
            return f"{key}: No data"
        if isinstance(value, float):
            if abs(value) >= 1_000_000:
                return f"{key}: {value:,.2f} ({value/1_000_000:.2f}M)"
            if abs(value) >= 1_000:
                return f"{key}: {value:,.2f} ({value/1_000:.1f}K)"
            return f"{key}: {value:,.2f}"
        if isinstance(value, int) and abs(value) >= 1_000:
            return f"{key}: {value:,}"
        return f"{key}: {value}"

    # ── RAG Prompt ────────────────────────────
    def _build_rag_prompt(self, question, contexts):
        ctx_text = "\n\n".join(
            f"[Source: {c['table']} | Score: {c['score']:.3f}]\n{c['text']}"
            for c in contexts
        )
        return f"""Below are context snippets retrieved from the Amazon e-commerce database.
Use this information to answer the user's question.
If the context does not contain enough information, say so clearly.
Respond in English.

=== CONTEXT ===
{ctx_text}

=== QUESTION ===
{question}

=== ANSWER ==="""

    # ── Main Answer ───────────────────────────
    def answer(self, question: str, k: int = 6) -> Dict[str, Any]:
        if self.looks_like_sql_question(question):
            tpl = self.template_sql(question)
            if tpl:
                out = self.run_sql_safe(tpl)
                if "error" not in out:
                    return {
                        "mode":   "template_sql",
                        "sql":    out["sql"],
                        "result": out["result"],
                        "answer": self.summarize_sql_result(question, out["sql"], out["result"]),
                    }
                return {"mode": "template_sql_error", "sql": out.get("sql"), "error": out.get("error")}

        contexts = self.retrieve(question, k=k)
        if not contexts:
            return {"mode": "rag", "contexts": [], "answer": "No relevant data found."}

        prompt     = self._build_rag_prompt(question, contexts)
        llm_answer = self.ask_ollama(prompt, temperature=0.2)
        return {"mode": "rag", "contexts": contexts, "answer": llm_answer}