"""Excel upload + erp_transactions insertion.

Ports the column-detection logic from `update_sales.py:read_one_file` but
writes to Postgres instead of a CSV. New SKUs are auto-created in
`dim_products` (sku-only stub — name and category can be backfilled later).
After every successful batch, refreshes `v_sales_weekly` then
`v_sales_weekly_full` so downstream endpoints see the new data immediately.

Required columns (auto-detected, case-insensitive, multi-language):
    Datum / Date              → transaction_date
    Artikal / Artikl / SKU    → product_id (resolved)
    Količina / Quantity       → quantity
    Vrijednost / Value        → total_value
    Tip dok. / Doc type       → channel_map_id (resolved)
Optional:
    Naziv                     → dim_products.name (for new SKUs only)
    Naziv grupacije           → category (for new SKUs, resolved to dim_categories)
    RUC / Marža               → ruc_eur
"""
from __future__ import annotations

import io
from datetime import datetime
from pathlib import Path
from typing import Optional

import pandas as pd
from sqlalchemy import text

from backend.repositories.base import BaseRepository


VALID_TIPS = {"RCM", "WSA", "WSB", "WSC", "WSD", "TRC", "VPT", "VPB", "RAC", "RIZ", "RPE"}


def _norm_code_series(s: "pd.Series") -> "pd.Series":
    """Normalise an ERP code column (partner / store / sales-rep / document)
    read from Excel. Excel coerces numeric-looking codes to numbers, so a
    stored '05770' arrives as 5770 (int) → '5770', and a column with NaNs
    becomes float → '5770.0'. Strip whitespace, drop the trailing '.0', and
    map blanks/'nan' to None. Leading-zero re-padding for partner matching
    happens at lookup time (see _partner_key)."""
    out = s.astype(str).str.strip().str.replace(r"\.0$", "", regex=True)
    return out.where(~out.str.lower().isin(["", "nan", "none"]), None)


def _partner_key(code: Optional[str]) -> Optional[str]:
    """Match key for a partner code that is robust to leading-zero padding:
    dim_partners stores 5-char zero-padded codes ('05770') but Excel strips
    the zero ('5770'). Compare on the zero-stripped form so both collapse to
    the same key. ('0' → '0' so an all-zero code doesn't vanish.)"""
    if code is None:
        return None
    stripped = code.lstrip("0")
    return stripped if stripped else "0"


def _detect_columns(df: pd.DataFrame) -> dict[str, str]:
    """Ported verbatim from update_sales.read_one_file column-detection logic.
    Returns dict mapping canonical keys (date/sku/qty/value/tip/cat/name/ruc)
    to the actual column names in `df`."""
    col_map: dict[str, str] = {}

    # Pass 1: exact matches
    for col in df.columns:
        cl = str(col).lower().strip()
        if cl in ("datum", "date") and "date" not in col_map:
            col_map["date"] = col
        elif cl in ("artikal", "artikl", "sku") and "sku" not in col_map:
            col_map["sku"] = col
        elif cl in ("količina", "kolicina", "qty", "quantity") and "qty" not in col_map:
            col_map["qty"] = col
        elif cl in ("vrijednost €", "vrijednost", "value") and "value" not in col_map:
            col_map["value"] = col
        elif cl in ("tip dok.", "tip dok", "tip_dok", "doc type") and "tip" not in col_map:
            col_map["tip"] = col
        elif cl in ("naziv grupacije",) and "cat" not in col_map:
            col_map["cat"] = col
        elif cl == "naziv" and "name" not in col_map:
            col_map["name"] = col
        elif "ruc" not in col_map and "%" not in cl and (
            cl.startswith("ruc") or cl.startswith("marž")
            or cl.startswith("marza") or cl == "margin"
        ):
            col_map["ruc"] = col
        # ---- full-rekapitulacija fields (all optional) ----
        elif cl == "partner" and "partner" not in col_map:
            col_map["partner"] = col
        elif cl in ("naziv partnera", "partner name") and "partner_name" not in col_map:
            col_map["partner_name"] = col
        elif cl == "jedinica" and "store" not in col_map:
            col_map["store"] = col
        elif cl in ("naziv jedinice",) and "store_name" not in col_map:
            col_map["store_name"] = col
        elif cl == "dokument" and "document" not in col_map:
            col_map["document"] = col
        elif cl == "komercijalist" and "sales_rep" not in col_map:
            col_map["sales_rep"] = col
        elif cl.startswith("nabavna vrijednost") and "purchase_value" not in col_map:
            col_map["purchase_value"] = col
        elif cl in ("% ruc", "ruc %") and "ruc_pct" not in col_map:
            col_map["ruc_pct"] = col
        elif cl.startswith("porezna osnovica") and "tax_base" not in col_map:
            col_map["tax_base"] = col
        elif cl.startswith("pdv") and "vat" not in col_map:
            col_map["vat"] = col
        elif cl.startswith("odobreni rabat") and "discount" not in col_map:
            col_map["discount"] = col
        elif cl.startswith("loyalty") and "loyalty" not in col_map:
            col_map["loyalty"] = col
        elif cl in ("država", "drzava", "country") and "country" not in col_map:
            col_map["country"] = col

    # Pass 2: fuzzy fallbacks
    for col in df.columns:
        cl = str(col).lower().strip()
        if "date" not in col_map and ("datum" in cl or "date" in cl):
            col_map["date"] = col
        if "sku" not in col_map and ("artikal" in cl or "artikl" in cl or "sku" in cl):
            col_map["sku"] = col
        if "qty" not in col_map and ("količ" in cl or "kolic" in cl):
            col_map["qty"] = col
        if "value" not in col_map and "nabavn" not in cl and (
            "vrijednost" in cl or "vrednost" in cl or "value" in cl
        ):
            col_map["value"] = col
        if "tip" not in col_map and ("tip dok" in cl or "tip_dok" in cl):
            col_map["tip"] = col
        if "cat" not in col_map and "naziv grupac" in cl:
            col_map["cat"] = col
        if "ruc" not in col_map and "%" not in cl and (
            "ruc" in cl or "marž" in cl or "marz" in cl or "margin" in cl
        ):
            col_map["ruc"] = col

    return col_map


def _is_european_numbers(series_list: list["pd.Series"]) -> bool:
    """Decide a FILE's number format from all its numeric columns jointly. A
    single value's lone '.' is ambiguous (1.234 = 1234 in HR, 1.234 in anglo),
    and one column (e.g. integer quantities) may carry no decimals at all — so
    we look across every numeric column. European if any cell anywhere shows a
    decimal-comma signature (comma + 1-2 trailing digits)."""
    pat = r",\d{1,2}(?:\D|$)"
    for s in series_list:
        if s is None:
            continue
        if (s.dtype == object or pd.api.types.is_string_dtype(s)):
            if s.astype(str).str.contains(pat, regex=True, na=False).any():
                return True
    return False


def _to_numeric_eu(s: "pd.Series", european: bool) -> "pd.Series":
    """Coerce a Series to float using the file-level format decision from
    _is_european_numbers. Already-numeric Series (Excel cells) pass through.
    European → '.' is thousands, ',' is decimal; anglo → ',' is thousands."""
    if not (s.dtype == object or pd.api.types.is_string_dtype(s)):
        return pd.to_numeric(s, errors="coerce")

    def fix(v):
        t = str(v).strip()
        if t == "" or t.lower() == "nan":
            return None
        if european:
            return t.replace(".", "").replace(",", ".")
        return t.replace(",", "")

    return pd.to_numeric(s.astype(str).str.strip().map(fix), errors="coerce")


def parse_excel_file(file_bytes: bytes, *, filename: str = "upload.xlsx") -> dict:
    """Parse one Rekapitulacija Excel file. Returns a dict with:
        df: DataFrame [date, sku, qty, value, tip, ruc?, cat?, name?]
        col_map: detected column mapping
        warnings: list of strings (non-fatal issues)

    Raises ValueError on missing required columns or unreadable file.
    """
    warnings_out: list[str] = []
    bio = io.BytesIO(file_bytes)
    is_csv = filename.lower().endswith(".csv")

    def _looks_like_data(df) -> bool:
        cols_lower = [str(c).lower() for c in df.columns]
        has_date_col = any("datum" in c or "date" in c for c in cols_lower)
        has_sku_col  = any("artikal" in c or "artikl" in c or "sku" in c for c in cols_lower)
        return has_date_col or has_sku_col

    raw = None
    if is_csv:
        # CSV (e.g. emailed Rekapitulacija). Sniff delimiter (Croatian ERP
        # exports often use ';') and tolerate a title block via header offset.
        last_exc: Exception | None = None
        for sep in (";", ",", "\t"):
            for skip in range(0, 6):
                try:
                    bio.seek(0)
                    df = pd.read_csv(bio, sep=sep, header=skip,
                                     dtype=str, encoding="utf-8-sig",
                                     engine="python", on_bad_lines="skip")
                    if df.shape[1] < 2:
                        continue   # wrong delimiter — everything in one column
                    if _looks_like_data(df):
                        raw = df
                        if skip > 0:
                            warnings_out.append(f"{filename}: skipped {skip} header row(s)")
                        break
                except Exception as exc:
                    last_exc = exc
                    continue
            if raw is not None:
                break
        if raw is None:
            raise ValueError(
                f"Cannot read CSV '{filename}': no Datum/Artikal columns found "
                f"with ';', ',' or tab delimiter."
                + (f" Last error: {last_exc}" if last_exc else "")
            )
    else:
        try:
            import openpyxl
            wb_check = openpyxl.load_workbook(bio, read_only=True)
            sheet_names = wb_check.sheetnames
            wb_check.close()
        except Exception as exc:
            raise ValueError(f"Cannot read workbook '{filename}': {exc}")

        skip_sheets = {f"Sheet{i}" for i in range(1, 10)} | {f"sheet{i}" for i in range(1, 10)}
        data_sheets = [s for s in sheet_names if s not in skip_sheets]
        target_sheet = data_sheets[0] if data_sheets else sheet_names[0]

        bio.seek(0)
        for skip in range(0, 6):
            try:
                bio.seek(0)
                df = pd.read_excel(bio, sheet_name=target_sheet, header=skip)
                if _looks_like_data(df):
                    raw = df
                    if skip > 0:
                        warnings_out.append(f"{filename}: skipped {skip} header row(s)")
                    break
            except Exception:
                continue
        if raw is None:
            bio.seek(0)
            raw = pd.read_excel(bio, sheet_name=target_sheet)

    col_map = _detect_columns(raw)
    required = ["date", "sku", "qty", "value", "tip"]
    missing = [k for k in required if k not in col_map]
    if missing:
        raise ValueError(
            f"{filename}: missing columns {missing}. "
            f"Found: {list(raw.columns)}. "
            f"Need: Datum, Artikal, Količina, Vrijednost, Tip dok."
        )

    # Build a normalised DataFrame
    out = raw[[col_map["date"], col_map["sku"], col_map["qty"],
               col_map["value"], col_map["tip"]]].copy()
    out.columns = ["date", "sku", "qty", "value", "tip"]
    # Decide number format ONCE from all numeric columns in this file.
    _NUMERIC_OPT = ("ruc", "purchase_value", "ruc_pct", "tax_base", "vat", "discount")
    _num_src = [raw[col_map[k]] for k in ("qty", "value", *_NUMERIC_OPT) if k in col_map]
    european = _is_european_numbers(_num_src)

    if "ruc" in col_map:
        out["ruc"] = _to_numeric_eu(raw[col_map["ruc"]], european)
    else:
        out["ruc"] = pd.Series([None] * len(out), dtype=object)
    if "cat" in col_map:
        out["cat"] = raw[col_map["cat"]].astype(str).str.strip()
    else:
        out["cat"] = ""
    if "name" in col_map:
        out["name"] = raw[col_map["name"]].astype(str).str.strip()
    else:
        out["name"] = ""

    # ---- full-rekapitulacija fields (optional; absent → None) ----
    # Code/string columns: normalise Excel's numeric coercion (drop '.0').
    for key in ("partner", "partner_name", "store", "store_name",
                "document", "sales_rep", "country"):
        out[key] = (_norm_code_series(raw[col_map[key]]) if key in col_map
                    else pd.Series([None] * len(out), dtype=object))
    # Numeric financial columns.
    for key in ("purchase_value", "ruc_pct", "tax_base", "vat", "discount"):
        out[key] = (_to_numeric_eu(raw[col_map[key]], european) if key in col_map
                    else pd.Series([None] * len(out), dtype=object))
    # Loyalty: a long numeric card-id string when a card was used (see
    # migrate_sales _prepare_chunk); store just the boolean.
    if "loyalty" in col_map:
        lk = raw[col_map["loyalty"]].fillna("").astype(str).str.strip()
        out["has_loyalty"] = lk.str.match(r"^\d{6,}$")
    else:
        out["has_loyalty"] = False

    # Normalise
    out["sku"] = out["sku"].astype(str).str.strip()
    out["tip"] = out["tip"].astype(str).str.strip().str.upper()
    out["date"] = pd.to_datetime(out["date"], errors="coerce", dayfirst=True)
    out["qty"] = _to_numeric_eu(out["qty"], european)
    out["value"] = _to_numeric_eu(out["value"], european)

    n_before = len(out)
    out = out[out["date"].notna() & out["sku"].notna() & (out["sku"] != "")]
    out = out[out["qty"].notna() & out["value"].notna()]
    out = out[out["tip"].isin(VALID_TIPS)]
    out = out[out["qty"] > 0]   # exclude returns (matches update_sales behaviour)
    n_after = len(out)
    if n_after < n_before:
        warnings_out.append(f"{filename}: dropped {n_before - n_after} invalid/return rows")

    return {"df": out.reset_index(drop=True), "col_map": col_map, "warnings": warnings_out}


class UploadRepository(BaseRepository):

    def get_existing_skus(self) -> dict[str, int]:
        """Return {sku: product_id}."""
        rows = self.db.execute(text("SELECT sku, id FROM dim_products")).fetchall()
        return {r[0]: int(r[1]) for r in rows}

    def get_channel_map(self) -> dict[str, int]:
        """Return {doc_type: lookup_channel_map.id}."""
        rows = self.db.execute(text("SELECT doc_type, id FROM lookup_channel_map")).fetchall()
        return {r[0]: int(r[1]) for r in rows}

    def get_partner_map(self) -> dict[str, int]:
        """Return {zero-stripped code: partner_id} so Excel-depadded codes
        ('5770') match the DB's zero-padded codes ('05770'). See _partner_key."""
        rows = self.db.execute(text("SELECT code, id FROM dim_partners")).fetchall()
        out: dict[str, int] = {}
        for code, pid in rows:
            key = _partner_key(str(code).strip()) if code is not None else None
            if key is not None:
                out[key] = int(pid)
        return out

    def get_store_map(self) -> dict[str, int]:
        """Return {zero-stripped unit_code: store_id} (same depadding logic
        as partners — '08' and '8' collapse to one key)."""
        rows = self.db.execute(text("SELECT unit_code, id FROM dim_stores")).fetchall()
        out: dict[str, int] = {}
        for unit, sid in rows:
            key = _partner_key(str(unit).strip()) if unit is not None else None
            if key is not None:
                out[key] = int(sid)
        return out

    def create_partners_batch(self, new_partners: list[dict]) -> dict[str, int]:
        """Auto-discover partners. Each dict: {code, name, country}. Stores the
        code zero-padded to 5 chars to match the dominant DB convention.
        Returns {zero-stripped code: id} for the inserted rows."""
        out: dict[str, int] = {}
        for r in new_partners:
            code = (r.get("code") or "").strip()
            if not code:
                continue
            stored = code.zfill(5) if (code.isdigit() and len(code) < 5) else code
            row = self.db.execute(
                text("""
                    INSERT INTO dim_partners (code, name, country)
                    VALUES (:code, :name, :country)
                    ON CONFLICT (code) DO UPDATE SET code = EXCLUDED.code
                    RETURNING id
                """),
                {"code": stored, "name": r.get("name") or None,
                 "country": r.get("country") or None},
            ).first()
            if row:
                out[_partner_key(code)] = int(row[0])
        self.db.commit()
        return out

    def create_stores_batch(self, new_stores: list[dict]) -> dict[str, int]:
        """Auto-discover stores. Each dict: {unit_code, name, country}. Unit '1'
        (zero-stripped) is the central warehouse. Returns {zero-stripped
        unit_code: id} for the inserted rows."""
        out: dict[str, int] = {}
        for r in new_stores:
            unit = (r.get("unit_code") or "").strip()
            if not unit:
                continue
            row = self.db.execute(
                text("""
                    INSERT INTO dim_stores (unit_code, name, country, is_warehouse)
                    VALUES (:unit, :name, :country, :wh)
                    ON CONFLICT (unit_code, country) DO UPDATE SET unit_code = EXCLUDED.unit_code
                    RETURNING id
                """),
                {"unit": unit, "name": r.get("name") or None,
                 "country": (r.get("country") or "HR"),
                 "wh": _partner_key(unit) == "1"},
            ).first()
            if row:
                out[_partner_key(unit)] = int(row[0])
        self.db.commit()
        return out

    def ensure_category(self, *, name: str) -> Optional[int]:
        if not name:
            return None
        row = self.db.execute(
            text("SELECT id FROM dim_categories WHERE name = :n"),
            {"n": name},
        ).first()
        if row:
            return int(row[0])
        ins = self.db.execute(
            text("INSERT INTO dim_categories (name) VALUES (:n) RETURNING id"),
            {"n": name},
        ).first()
        return int(ins[0]) if ins else None

    def create_products_batch(
        self, *, new_skus: list[dict],
    ) -> dict[str, int]:
        """Create new products in dim_products. Each dict: {sku, name, category}.
        Returns {sku: id} for the newly inserted rows. Uses ON CONFLICT to
        keep the operation idempotent if another worker inserts the same SKU."""
        if not new_skus:
            return {}
        # Cache category resolution
        cat_cache: dict[str, Optional[int]] = {}
        for r in new_skus:
            c = (r.get("category") or "").strip()
            if c and c not in cat_cache:
                cat_cache[c] = self.ensure_category(name=c)

        # Insert one at a time (small N — only new SKUs) with ON CONFLICT
        out: dict[str, int] = {}
        for r in new_skus:
            sku = r["sku"]
            name = r.get("name") or ""
            cat_name = (r.get("category") or "").strip()
            cat_id = cat_cache.get(cat_name) if cat_name else None
            row = self.db.execute(
                text("""
                    INSERT INTO dim_products (sku, name, category_id, active)
                    VALUES (:sku, :name, :cat_id, true)
                    ON CONFLICT (sku) DO UPDATE SET sku = EXCLUDED.sku
                    RETURNING id
                """),
                {"sku": sku, "name": name, "cat_id": cat_id},
            ).first()
            if row:
                out[sku] = int(row[0])
        self.db.commit()
        return out

    # Every column the upload can populate. Missing keys default to NULL via
    # the per-row .get() normalisation in _normalise_tx_rows so a sparse
    # (date/sku/qty/value/tip-only) file still inserts cleanly.
    _TX_COLS = (
        "transaction_date", "document", "partner_id", "product_id", "store_id",
        "channel_map_id", "sales_rep", "quantity", "purchase_value", "ruc_eur",
        "ruc_pct", "tax_base", "vat", "total_value", "approved_discount",
        "has_loyalty",
    )

    @classmethod
    def _normalise_tx_rows(cls, rows: list[dict]) -> list[dict]:
        """Ensure every row dict carries every _TX_COLS key (NULL when absent),
        so the single parameterised INSERT binds consistently."""
        return [{c: r.get(c) for c in cls._TX_COLS} for r in rows]

    def insert_transactions(self, rows: list[dict]) -> int:
        """Bulk insert into erp_transactions. Required per row:
        transaction_date, product_id, channel_map_id, quantity, total_value.
        Optional (NULL when omitted): document, partner_id, store_id,
        sales_rep, purchase_value, ruc_eur, ruc_pct, tax_base, vat,
        approved_discount, has_loyalty. Returns count inserted."""
        if not rows:
            return 0
        cols = ", ".join(self._TX_COLS)
        binds = ", ".join(f":{c}" for c in self._TX_COLS)
        sql = text(f"INSERT INTO erp_transactions ({cols}) VALUES ({binds})")
        rows = self._normalise_tx_rows(rows)
        # Use executemany via SQLAlchemy's list-of-dicts form
        batch_size = 1000
        n = 0
        for i in range(0, len(rows), batch_size):
            chunk = rows[i:i + batch_size]
            self.db.execute(sql, chunk)
            n += len(chunk)
        self.db.commit()
        return n

    def refresh_sales_weekly(self) -> bool:
        """Refresh v_sales_weekly (erp_transactions window), then
        v_sales_weekly_full (UNION of v_sales_weekly + sales_clean_import).
        v_sales_weekly uses CONCURRENTLY (has unique index); v_sales_weekly_full
        uses plain refresh (no unique index on the UNION view)."""
        try:
            self.db.execute(text("REFRESH MATERIALIZED VIEW CONCURRENTLY v_sales_weekly"))
            self.db.commit()
        except Exception:
            self.db.rollback()
            try:
                self.db.execute(text("REFRESH MATERIALIZED VIEW v_sales_weekly"))
                self.db.commit()
            except Exception:
                self.db.rollback()
                return False
        # Refresh the full history view that unions v_sales_weekly
        try:
            self.db.execute(text("REFRESH MATERIALIZED VIEW v_sales_weekly_full"))
            self.db.commit()
        except Exception:
            self.db.rollback()
        return True