"""Photo Assets By SKU wide-CSV transform (pure domain — no SQLAlchemy).

Expected columns (any casing / spacing):
  SKU, Type, Image_Default, Image_Open, Image_Closed, Image_Generic,
  Collection_Hero_Image, Collection_Vignette_Images

Multi-value cells (vignettes) are split on ';' or '|'.
Family vignette fill copies vignette URLs to style siblings that lack them.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import PurePosixPath
from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
from urllib.parse import unquote, urlparse

import pandas as pd

from db.image_urls import normalize_external_image_url
from db.source_imports import normalize_column_key
from db.tribeca_sku_parse import split_styled_sku

# Normalized column → (slot_key, view_suffix, multi)
PHOTO_ASSET_SLOTS: Tuple[Tuple[str, str, Optional[str], bool], ...] = (
    ("image_default", "default", None, False),
    ("image_closed", "closed", "closed", False),
    ("image_open", "open", "open", False),
    ("image_generic", "generic", None, False),
    ("collection_hero_image", "collection_hero", None, False),
    ("collection_vignette_images", "vignette", None, True),
)

PRODUCT_SHOT_SLOTS = frozenset({"default", "closed", "open", "generic"})
VIGNETTE_SLOT = "vignette"


@dataclass
class PhotoAssetSlot:
    slot_key: str
    url: str
    file_name: str
    view_suffix: Optional[str]


@dataclass
class PhotoAssetSkuRow:
    sku: str
    row_type: str = ""
    slots: List[PhotoAssetSlot] = field(default_factory=list)
    from_family_vignette: bool = False

    @property
    def has_product_shot(self) -> bool:
        return any(s.slot_key in PRODUCT_SHOT_SLOTS for s in self.slots)

    @property
    def has_vignette(self) -> bool:
        return any(s.slot_key == VIGNETTE_SLOT for s in self.slots)

    @property
    def vignette_urls(self) -> List[str]:
        return [s.url for s in self.slots if s.slot_key == VIGNETTE_SLOT]


def _clean(value: object) -> str:
    if value is None:
        return ""
    return str(value).strip()


def is_photo_assets_by_sku_csv(df: pd.DataFrame) -> bool:
    """Detect wide Photo_Assets_By_SKU sheets (SKU + image slot columns)."""
    if df is None:
        return False
    cols = {normalize_column_key(c) for c in df.columns}
    has_sku = bool(cols & {"sku", "master_sku", "item", "item_sku"})
    has_slot = bool(
        cols
        & {
            "image_default",
            "image_open",
            "image_closed",
            "image_generic",
            "collection_hero_image",
            "collection_vignette_images",
        }
    )
    return has_sku and has_slot


def split_image_cell(value: object) -> List[str]:
    """Split multi-URL cells on ';' or '|' (Photo Assets vignette lists)."""
    text = _clean(value)
    if not text:
        return []
    parts: List[str] = []
    for chunk in text.replace("|", ";").split(";"):
        url = normalize_external_image_url(chunk.strip())
        if url:
            parts.append(url)
    return parts


def file_name_from_url(url: str, *, fallback: str) -> str:
    """Prefer the URL basename; fall back to a stable synthetic name."""
    try:
        path = unquote(urlparse(url).path or "")
        name = PurePosixPath(path.replace("\\", "/")).name
    except Exception:
        name = ""
    if name:
        return name
    return fallback


def style_code_for_sku(sku: str) -> str:
    style, _ = split_styled_sku(_clean(sku))
    return style


def _slot_file_fallback(sku: str, slot_key: str, index: int = 0) -> str:
    safe_sku = _clean(sku).replace("/", "-") or "UNKNOWN"
    if slot_key == VIGNETTE_SLOT:
        return f"{safe_sku}-VIGNETTE-{index + 1}.jpg"
    return f"{safe_sku}-{slot_key.upper().replace('_', '-')}.jpg"


def parse_photo_assets_row(raw: Dict[str, Any]) -> Optional[PhotoAssetSkuRow]:
    normalized = {normalize_column_key(k): v for k, v in raw.items()}
    sku = _clean(
        normalized.get("sku")
        or normalized.get("master_sku")
        or normalized.get("item")
        or normalized.get("item_sku")
    )
    if not sku:
        return None
    row_type = _clean(normalized.get("type"))
    slots: List[PhotoAssetSlot] = []
    for col_key, slot_key, view_suffix, multi in PHOTO_ASSET_SLOTS:
        cell = normalized.get(col_key)
        urls = split_image_cell(cell) if multi else ([normalize_external_image_url(_clean(cell))] if _clean(cell) else [])
        for idx, url in enumerate(urls):
            if not url:
                continue
            file_name = file_name_from_url(
                url,
                fallback=_slot_file_fallback(sku, slot_key, idx),
            )
            slots.append(
                PhotoAssetSlot(
                    slot_key=slot_key,
                    url=url,
                    file_name=file_name,
                    view_suffix=view_suffix,
                )
            )
    return PhotoAssetSkuRow(sku=sku, row_type=row_type, slots=slots)


def fill_family_vignettes(
    rows_by_sku: Dict[str, PhotoAssetSkuRow],
) -> Tuple[Dict[str, PhotoAssetSkuRow], int]:
    """Copy vignette URLs to style siblings that have none.

    Groups by style prefix (ACH, ASW, …). Within each group, the first stable
    vignette bundle (by SKU sort) is used as the family source.
    """
    by_style: Dict[str, List[str]] = {}
    for sku in rows_by_sku:
        style = style_code_for_sku(sku)
        if not style:
            continue
        by_style.setdefault(style, []).append(sku)

    filled = 0
    for style, skus in by_style.items():
        ordered = sorted(skus)
        source_slots: List[PhotoAssetSlot] = []
        for sku in ordered:
            row = rows_by_sku[sku]
            if row.has_vignette:
                source_slots = [s for s in row.slots if s.slot_key == VIGNETTE_SLOT]
                break
        if not source_slots:
            continue
        for sku in ordered:
            row = rows_by_sku[sku]
            if row.has_vignette:
                continue
            # Re-bind file names to family URLs (shared basename from source).
            for slot in source_slots:
                row.slots.append(
                    PhotoAssetSlot(
                        slot_key=slot.slot_key,
                        url=slot.url,
                        file_name=slot.file_name,
                        view_suffix=slot.view_suffix,
                    )
                )
            row.from_family_vignette = True
            filled += 1
    return rows_by_sku, filled


def photo_assets_to_long_records(
    df: pd.DataFrame,
    *,
    known_skus: Optional[Set[str]] = None,
    fill_family: bool = True,
) -> Tuple[List[Dict[str, str]], Dict[str, Any]]:
    """Melt wide Photo Assets rows into long {sku, name, url} records for image import."""
    rows_by_sku: Dict[str, PhotoAssetSkuRow] = {}
    blank_sku = 0
    empty_rows = 0
    unknown_skus: List[str] = []
    slot_counts: Dict[str, int] = {spec[1]: 0 for spec in PHOTO_ASSET_SLOTS}

    for raw in df.to_dict(orient="records"):
        parsed = parse_photo_assets_row(raw)
        if parsed is None:
            blank_sku += 1
            continue
        if known_skus is not None and parsed.sku not in known_skus:
            # Still keep the row so family fill can use its vignettes for known siblings;
            # unknown SKUs are dropped at emit time unless they match known_skus.
            unknown_skus.append(parsed.sku)
        if not parsed.slots:
            empty_rows += 1
            # Keep empty shell so family fill can still target this SKU.
            rows_by_sku.setdefault(parsed.sku, parsed)
            continue
        existing = rows_by_sku.get(parsed.sku)
        if existing is None:
            rows_by_sku[parsed.sku] = parsed
        else:
            existing.slots.extend(parsed.slots)
            if parsed.row_type and not existing.row_type:
                existing.row_type = parsed.row_type

    family_filled = 0
    if fill_family:
        rows_by_sku, family_filled = fill_family_vignettes(rows_by_sku)

    # Also fill known master SKUs that are missing from the CSV but share a style
    # with a vignette-bearing CSV row (true family fan-out into catalog gaps).
    catalog_family_filled = 0
    if fill_family and known_skus:
        style_vignettes: Dict[str, List[PhotoAssetSlot]] = {}
        for sku, row in rows_by_sku.items():
            if not row.has_vignette:
                continue
            style = style_code_for_sku(sku)
            if style and style not in style_vignettes:
                style_vignettes[style] = [s for s in row.slots if s.slot_key == VIGNETTE_SLOT]
        for master_sku in sorted(known_skus):
            if master_sku in rows_by_sku and rows_by_sku[master_sku].has_vignette:
                continue
            style = style_code_for_sku(master_sku)
            source = style_vignettes.get(style)
            if not source:
                continue
            if master_sku not in rows_by_sku:
                rows_by_sku[master_sku] = PhotoAssetSkuRow(sku=master_sku, from_family_vignette=True)
            row = rows_by_sku[master_sku]
            if row.has_vignette:
                continue
            for slot in source:
                row.slots.append(
                    PhotoAssetSlot(
                        slot_key=slot.slot_key,
                        url=slot.url,
                        file_name=slot.file_name,
                        view_suffix=slot.view_suffix,
                    )
                )
            row.from_family_vignette = True
            catalog_family_filled += 1

    long_records: List[Dict[str, str]] = []
    # Full gallery replace only when the CSV supplies product shots for that SKU.
    # Vignette/hero-only (incl. family fill) merges so existing renders are kept.
    replace_skus: Set[str] = set()
    merge_skus: Set[str] = set()
    skipped_unknown = 0

    for sku, row in sorted(rows_by_sku.items()):
        if known_skus is not None and sku not in known_skus:
            skipped_unknown += 1
            continue
        if not row.slots:
            continue
        if row.has_product_shot:
            replace_skus.add(sku)
        else:
            merge_skus.add(sku)
        for slot in row.slots:
            slot_counts[slot.slot_key] = slot_counts.get(slot.slot_key, 0) + 1
            long_records.append(
                {
                    "sku": sku,
                    "name": slot.file_name,
                    "url": slot.url,
                    "image_role": slot.slot_key,
                }
            )

    stats: Dict[str, Any] = {
        "input_rows": int(len(df)),
        "blank_sku_rows": blank_sku,
        "empty_image_rows": empty_rows,
        "family_vignettes_filled_skus": family_filled,
        "catalog_family_vignettes_filled_skus": catalog_family_filled,
        "long_records": len(long_records),
        "replace_sku_count": len(replace_skus),
        "merge_sku_count": len(merge_skus),
        "unknown_sku_rows": len(set(unknown_skus)),
        "skipped_unknown_skus": skipped_unknown,
        "sample_unknown_skus": sorted(set(unknown_skus))[:20],
        "slot_counts": slot_counts,
        "replace_skus": sorted(replace_skus),
        "merge_skus": sorted(merge_skus),
    }
    return long_records, stats


def long_records_to_dataframe(records: Sequence[Dict[str, str]]) -> pd.DataFrame:
    if not records:
        return pd.DataFrame(columns=["sku", "name", "url", "image_role"])
    return pd.DataFrame(list(records))