from __future__ import annotations

import hashlib
import json
import math
import re
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

import pandas as pd

from db.ports import IngestRepository
from db.schemas import (
    IngestRunCreate,
    MagentoProductSourceSnapshotCreate,
    ProductSupplementalAttributeValueCreate,
    PlytixProductSourceSnapshotCreate,
    ShopifyProductSourceSnapshotCreate,
)


def stable_payload_hash(payload: Dict[str, Any]) -> str:
    blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()


def dataframe_to_source_records(df: pd.DataFrame) -> List[Dict[str, Any]]:
    records: List[Dict[str, Any]] = []
    for row in df.to_dict(orient="records"):
        records.append({str(k): _clean_json_value(v) for k, v in row.items()})
    return records


def import_magento_source_dataframe(
    repo: IngestRepository,
    df: pd.DataFrame,
    *,
    file_name: str,
    file_hash: str,
    connection_id: Optional[int] = None,
) -> Tuple[int, Dict[str, int]]:
    started_at = datetime.utcnow()
    records = dataframe_to_source_records(df)
    ingest_id = repo.create_ingest_run(
        IngestRunCreate(
            source="magento_source",
            file_name=file_name,
            file_hash=file_hash,
            started_at=started_at,
            status="started",
            row_count=len(records),
        )
    )

    by_sku = _dedupe_records_by_sku(records)
    missing_sku = _count_missing_sku(records)
    current = repo.load_current_magento_source_index(list(by_sku), connection_id=connection_id)
    close_ids: List[int] = []
    inserts: List[MagentoProductSourceSnapshotCreate] = []
    unchanged = 0

    for sku, payload in by_sku.items():
        row_hash = stable_payload_hash(payload)
        existing = current.get(sku)
        if existing and existing.get("row_hash") == row_hash:
            unchanged += 1
            continue
        if existing:
            close_ids.append(int(existing["id"]))
        inserts.append(
            MagentoProductSourceSnapshotCreate(
                connection_id=connection_id,
                load_id=ingest_id,
                sku=sku,
                valid_from=started_at,
                magento_product_id=_to_int(_first(payload, "id", "entity_id", "product_id", "magento_product_id")),
                product_type=_as_text(_first(payload, "type_id", "product_type", "type")),
                attribute_set_id=_to_int(_first(payload, "attribute_set_id")),
                status=_as_text(_first(payload, "status")),
                visibility=_as_text(_first(payload, "visibility")),
                price=_to_float(_first(payload, "price")),
                qty=_to_float(_first(payload, "qty", "quantity")),
                is_in_stock=_to_bool(_first(payload, "is_in_stock", "stock_status")),
                payload=payload,
                column_names={"columns": [str(c) for c in df.columns]},
                row_hash=row_hash,
            )
        )

    repo.close_magento_source_snapshots(close_ids, valid_to=started_at)
    repo.add_magento_source_snapshots(inserts)
    repo.update_ingest_run(ingest_id, status="success", row_count=len(records))
    return ingest_id, {
        "input_rows": len(records),
        "distinct_skus": len(by_sku),
        "inserted": len(inserts),
        "closed_previous": len(close_ids),
        "unchanged": unchanged,
        "missing_sku": missing_sku,
        "duplicate_skus_overwritten": max(len(records) - missing_sku - len(by_sku), 0),
    }


def import_plytix_source_dataframe(
    repo: IngestRepository,
    df: pd.DataFrame,
    *,
    file_name: str,
    file_hash: str,
) -> Tuple[int, Dict[str, int]]:
    from plytix.product_normalize import normalize_plytix_flat_record

    started_at = datetime.utcnow()
    records = [
        normalize_plytix_flat_record(row)
        for row in dataframe_to_source_records(df)
    ]
    ingest_id = repo.create_ingest_run(
        IngestRunCreate(
            source="plytix_source",
            file_name=file_name,
            file_hash=file_hash,
            started_at=started_at,
            status="started",
            row_count=len(records),
        )
    )

    by_sku = _dedupe_records_by_sku(records)
    missing_sku = _count_missing_sku(records)
    current = repo.load_current_plytix_source_index(list(by_sku))
    close_ids: List[int] = []
    inserts: List[PlytixProductSourceSnapshotCreate] = []
    unchanged = 0

    for sku, payload in by_sku.items():
        row_hash = stable_payload_hash(payload)
        existing = current.get(sku)
        if existing and existing.get("row_hash") == row_hash:
            unchanged += 1
            continue
        if existing:
            close_ids.append(int(existing["id"]))
        inserts.append(
            PlytixProductSourceSnapshotCreate(
                load_id=ingest_id,
                sku=sku,
                valid_from=started_at,
                product_websites=_as_text(_first(payload, "product_websites", "product websites", "websites")),
                product_type=_as_text(_first(payload, "product_type", "type")),
                status=_as_text(_first(payload, "status", "product_online")),
                price=_to_float(_first(payload, "price")),
                payload=payload,
                column_names={"columns": [str(c) for c in df.columns]},
                row_hash=row_hash,
            )
        )

    repo.close_plytix_source_snapshots(close_ids, valid_to=started_at)
    repo.add_plytix_source_snapshots(inserts)
    repo.update_ingest_run(ingest_id, status="success", row_count=len(records))
    return ingest_id, {
        "input_rows": len(records),
        "distinct_skus": len(by_sku),
        "inserted": len(inserts),
        "closed_previous": len(close_ids),
        "unchanged": unchanged,
        "missing_sku": missing_sku,
        "duplicate_skus_overwritten": max(len(records) - missing_sku - len(by_sku), 0),
    }


def import_shopify_source_dataframe(
    repo: IngestRepository,
    df: pd.DataFrame,
    *,
    file_name: str,
    file_hash: str,
    connection_id: Optional[int] = None,
) -> Tuple[int, Dict[str, int]]:
    started_at = datetime.utcnow()
    records = dataframe_to_source_records(df)
    ingest_id = repo.create_ingest_run(
        IngestRunCreate(
            source="shopify_source",
            file_name=file_name,
            file_hash=file_hash,
            started_at=started_at,
            status="started",
            row_count=len(records),
        )
    )

    by_sku = _dedupe_records_by_sku(records)
    missing_sku = _count_missing_sku(records)
    current = repo.load_current_shopify_source_index(list(by_sku), connection_id=connection_id)
    close_ids: List[int] = []
    inserts: List[ShopifyProductSourceSnapshotCreate] = []
    unchanged = 0

    for sku, payload in by_sku.items():
        row_hash = stable_payload_hash(payload)
        existing = current.get(sku)
        if existing and existing.get("row_hash") == row_hash:
            unchanged += 1
            continue
        if existing:
            close_ids.append(int(existing["id"]))
        inserts.append(
            ShopifyProductSourceSnapshotCreate(
                connection_id=connection_id,
                load_id=ingest_id,
                sku=sku,
                valid_from=started_at,
                shopify_product_id=_as_text(_first(payload, "shopify_product_id", "product_id")),
                shopify_variant_id=_as_text(_first(payload, "shopify_variant_id", "variant_id")),
                product_type=_as_text(_first(payload, "product_type", "productType")),
                status=_as_text(_first(payload, "status")),
                price=_to_float(_first(payload, "price", "variant_price")),
                payload=payload,
                column_names={"columns": [str(c) for c in df.columns]},
                row_hash=row_hash,
            )
        )

    repo.close_shopify_source_snapshots(close_ids, valid_to=started_at)
    repo.add_shopify_source_snapshots(inserts)
    repo.update_ingest_run(ingest_id, status="success", row_count=len(records))
    return ingest_id, {
        "input_rows": len(records),
        "distinct_skus": len(by_sku),
        "inserted": len(inserts),
        "closed_previous": len(close_ids),
        "unchanged": unchanged,
        "missing_sku": missing_sku,
        "duplicate_skus_overwritten": max(len(records) - missing_sku - len(by_sku), 0),
    }


def import_supplemental_attribute_dataframe(
    repo: IngestRepository,
    df: pd.DataFrame,
    *,
    file_name: str,
    file_hash: str,
    source_label: str,
    sku_column: Optional[str] = None,
) -> Tuple[int, Dict[str, int]]:
    started_at = datetime.utcnow()
    records = dataframe_to_source_records(df)
    ingest_id = repo.create_ingest_run(
        IngestRunCreate(
            source="supplemental_attributes",
            file_name=file_name,
            file_hash=file_hash,
            started_at=started_at,
            status="started",
            row_count=len(records),
            notes=f"source_label={source_label}",
        )
    )

    rows: List[ProductSupplementalAttributeValueCreate] = []
    missing_sku = 0
    empty_values = 0
    sku_col = _find_sku_column(df, preferred=sku_column)
    source = source_label.strip() or "csv_upload"
    columns = [str(c) for c in df.columns]

    for row_num, payload in enumerate(records, start=1):
        sku = _as_text(payload.get(sku_col)) if sku_col else _as_text(_first(payload, "sku", "SKU"))
        if not sku:
            missing_sku += 1
            continue
        for column_name in columns:
            if column_name == sku_col:
                continue
            value_text = _as_text(payload.get(column_name))
            if value_text is None:
                empty_values += 1
                continue
            column_key = normalize_column_key(column_name)
            rows.append(
                ProductSupplementalAttributeValueCreate(
                    load_id=ingest_id,
                    source_label=source,
                    file_name=file_name,
                    row_num=row_num,
                    sku=sku,
                    column_name=column_name,
                    column_key=column_key,
                    value=value_text,
                    destination_hint=infer_destination_hint(column_key),
                )
            )

    repo.add_supplemental_attribute_values(rows)
    repo.update_ingest_run(ingest_id, status="success", row_count=len(records))
    return ingest_id, {
        "input_rows": len(records),
        "columns": len(columns),
        "inserted_values": len(rows),
        "missing_sku": missing_sku,
        "empty_values_skipped": empty_values,
        "sku_column": sku_col or "",
    }


def _dedupe_records_by_sku(records: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
    by_sku: Dict[str, Dict[str, Any]] = {}
    for payload in records:
        sku = _as_text(_first(payload, "sku", "SKU"))
        if sku:
            by_sku[sku] = payload
    return by_sku


def normalize_column_key(column_name: str) -> str:
    text = str(column_name or "").strip().replace("\ufeff", "").lower()
    text = re.sub(r"[/\s]+", "_", text)
    text = re.sub(r"[^a-z0-9_]+", "_", text)
    return re.sub(r"_+", "_", text).strip("_")


def infer_destination_hint(column_key: str) -> Optional[str]:
    key = normalize_column_key(column_key)
    if key.startswith("shopify_") or key in {"shopify_description", "shopify_title"}:
        return "shopify"
    if key.startswith("magento_"):
        return "magento"
    if key.startswith("seo_") or key.startswith("meta_") or key in {"title_tag", "meta_title", "meta_description"}:
        return "seo"
    return None


def _find_sku_column(df: pd.DataFrame, preferred: Optional[str] = None) -> Optional[str]:
    if preferred:
        wanted = normalize_column_key(preferred)
        for column in df.columns:
            if normalize_column_key(str(column)) == wanted:
                return str(column)
        for column in df.columns:
            if str(column).strip().lower() == str(preferred).strip().lower():
                return str(column)

    sku_aliases = {
        "sku",
        "product_sku",
        "item_sku",
        "variant_sku",
        "child_sku",
        "parent_sku",
        "magento_sku",
        "shopify_sku",
        "plytix_sku",
        "item_number",
        "item_no",
        "item",
        "product_code",
        "code",
    }
    for column in df.columns:
        key = normalize_column_key(str(column))
        if key in sku_aliases:
            return str(column)
    return None


def _count_missing_sku(records: List[Dict[str, Any]]) -> int:
    return sum(1 for payload in records if not _as_text(_first(payload, "sku", "SKU")))


def _first(payload: Dict[str, Any], *keys: str) -> Any:
    norm = {str(k).strip().lower().replace("_", " "): v for k, v in payload.items()}
    for key in keys:
        if key in payload:
            return payload[key]
        normalized = str(key).strip().lower().replace("_", " ")
        if normalized in norm:
            return norm[normalized]
    return None


def _clean_json_value(value: Any) -> Any:
    if value is None:
        return None
    if isinstance(value, dict):
        return {str(k): _clean_json_value(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_clean_json_value(v) for v in value]
    if isinstance(value, float):
        if math.isnan(value) or math.isinf(value):
            return None
        return value
    if isinstance(value, (str, int, bool)):
        return value
    return str(value)


def _as_text(value: Any) -> Optional[str]:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _to_float(value: Any) -> Optional[float]:
    text = _as_text(value)
    if text is None:
        return None
    try:
        return float(text.replace(",", ""))
    except (TypeError, ValueError):
        return None


def _to_int(value: Any) -> Optional[int]:
    number = _to_float(value)
    if number is None:
        return None
    return int(number)


def _to_bool(value: Any) -> Optional[bool]:
    text = _as_text(value)
    if text is None:
        return None
    lowered = text.lower()
    if lowered in {"1", "true", "yes", "y", "in stock"}:
        return True
    if lowered in {"0", "false", "no", "n", "out of stock"}:
        return False
    return None