from __future__ import annotations

from typing import Any, Dict, Iterable, List, Optional, Set, Tuple

import pandas as pd
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from channel.url_canonical import (
    build_plp_url,
    build_pdp_slug,
    build_pdp_url,
    infer_path_kind,
    normalize_plp_path,
)
from db.channel_exports import resolve_pipeline_channel_code
from db.models import (
    ChannelListingPath,
    ChannelListingPathTarget,
    MasterProduct,
    ProductListingPathAssignment,
)


INACTIVE_STATUS = "inactive"
ACTIVE_STATUS = "active"


def count_active_listing_assignments_except(
    session: Session,
    *,
    listing_path_id: int,
    keep_skus: Set[str],
) -> int:
    keep = {str(sku).strip() for sku in keep_skus if str(sku).strip()}
    rows = session.scalars(
        select(ProductListingPathAssignment.master_sku)
        .where(ProductListingPathAssignment.listing_path_id == listing_path_id)
        .where(ProductListingPathAssignment.assignment_status == ACTIVE_STATUS)
    ).all()
    return sum(1 for sku in rows if str(sku) not in keep)


def deactivate_listing_assignments_except(
    session: Session,
    *,
    listing_path_id: int,
    keep_skus: Set[str],
) -> int:
    keep = {str(sku).strip() for sku in keep_skus if str(sku).strip()}
    deactivated = 0
    for row in session.scalars(
        select(ProductListingPathAssignment)
        .where(ProductListingPathAssignment.listing_path_id == listing_path_id)
        .where(ProductListingPathAssignment.assignment_status == ACTIVE_STATUS)
    ).all():
        if str(row.master_sku) not in keep:
            row.assignment_status = INACTIVE_STATUS
            deactivated += 1
    if deactivated:
        session.flush()
    return deactivated


def list_listing_paths(
    session: Session,
    *,
    search: Optional[str] = None,
    path_kind: Optional[str] = None,
    hub_path_slug: Optional[str] = None,
    is_active: bool = True,
    limit: Optional[int] = None,
) -> List[Dict[str, Any]]:
    stmt = select(ChannelListingPath).order_by(ChannelListingPath.path_slug)
    if is_active:
        stmt = stmt.where(ChannelListingPath.is_active.is_(True))
    if path_kind:
        stmt = stmt.where(ChannelListingPath.path_kind == path_kind.strip())
    if hub_path_slug:
        stmt = stmt.where(ChannelListingPath.hub_path_slug == normalize_plp_path(hub_path_slug))
    if search:
        pattern = f"%{search.strip()}%"
        stmt = stmt.where(
            (ChannelListingPath.path_slug.ilike(pattern)) | (ChannelListingPath.title.ilike(pattern))
        )
    if limit:
        stmt = stmt.limit(limit)
    paths = list(session.scalars(stmt).all())
    targets_by_path = _targets_by_listing_path(session, [p.id for p in paths])
    return [_listing_path_dict(path, targets_by_path.get(path.id, [])) for path in paths]


def upsert_listing_path(
    session: Session,
    *,
    path_slug: str,
    title: str,
    path_kind: Optional[str] = None,
    parent_path_slug: Optional[str] = None,
    hub_path_slug: Optional[str] = None,
    facet_terms: Optional[List[str]] = None,
    shopping_facet_attribute: Optional[str] = None,
    notes: Optional[str] = None,
    is_active: bool = True,
    flush: bool = True,
    existing_by_slug: Optional[Dict[str, ChannelListingPath]] = None,
) -> Dict[str, Any]:
    slug = normalize_plp_path(path_slug)
    if not slug:
        raise ValueError("path_slug is required")
    title = str(title or "").strip() or slug
    kind = path_kind or infer_path_kind(slug)
    parent = normalize_plp_path(parent_path_slug) if parent_path_slug else None
    hub = normalize_plp_path(hub_path_slug) if hub_path_slug else None
    if not hub and parent:
        hub = parent.split("/")[0] if "/" not in parent else parent.rsplit("/", 1)[0]

    if existing_by_slug is not None:
        row = existing_by_slug.get(slug)
    else:
        row = session.scalar(select(ChannelListingPath).where(ChannelListingPath.path_slug == slug))
    current_facet_payload = row.facet_terms if row is not None and isinstance(row.facet_terms, dict) else {}
    facet_payload = dict(current_facet_payload)
    if facet_terms is not None:
        facet_payload["terms"] = facet_terms
    elif "terms" not in facet_payload:
        facet_payload["terms"] = []
    if shopping_facet_attribute is not None:
        attr = str(shopping_facet_attribute or "").strip()
        if attr:
            facet_payload["shopping_facet_attribute"] = attr
        else:
            facet_payload.pop("shopping_facet_attribute", None)

    payload = {
        "path_slug": slug,
        "path_kind": kind,
        "title": title,
        "parent_path_slug": parent,
        "hub_path_slug": hub,
        "facet_terms": facet_payload or None,
        "notes": notes,
        "is_active": is_active,
    }
    if row is None:
        row = ChannelListingPath(**payload)
        session.add(row)
        if existing_by_slug is not None:
            existing_by_slug[slug] = row
    else:
        for key, value in payload.items():
            setattr(row, key, value)
    if flush:
        session.flush()
    return _listing_path_dict(row, [])


def upsert_listing_path_target(
    session: Session,
    *,
    path_slug: str,
    channel_code: str,
    remote_id: str,
    taxonomy_kind: str = "category",
    connection_id: Optional[int] = None,
    remote_path: Optional[str] = None,
    flush: bool = True,
    existing_paths_by_slug: Optional[Dict[str, ChannelListingPath]] = None,
    existing_targets: Optional[Dict[Tuple[int, str, Optional[int], str, str], ChannelListingPathTarget]] = None,
) -> Dict[str, Any]:
    slug = normalize_plp_path(path_slug)
    if existing_paths_by_slug is not None:
        path = existing_paths_by_slug.get(slug)
    else:
        path = session.scalar(
            select(ChannelListingPath).where(ChannelListingPath.path_slug == slug)
        )
    if path is None:
        raise ValueError(f"Unknown listing path: {path_slug}")
    channel = resolve_pipeline_channel_code(channel_code)
    remote_id = str(remote_id or "").strip()
    if not remote_id:
        raise ValueError("remote_id is required")

    target_key = (int(path.id), channel, connection_id, taxonomy_kind, remote_id)
    if existing_targets is not None:
        existing = existing_targets.get(target_key)
        if existing is None:
            # Also match by (path, channel, connection, kind) ignoring remote_id for upsert-in-place.
            for key, row in existing_targets.items():
                if key[0] == int(path.id) and key[1] == channel and key[2] == connection_id and key[3] == taxonomy_kind:
                    existing = row
                    break
    else:
        existing = session.scalar(
            select(ChannelListingPathTarget)
            .where(ChannelListingPathTarget.listing_path_id == path.id)
            .where(ChannelListingPathTarget.channel_code == channel)
            .where(ChannelListingPathTarget.taxonomy_kind == taxonomy_kind)
            .where(ChannelListingPathTarget.remote_id == remote_id)
            .where(
                ChannelListingPathTarget.connection_id.is_(None)
                if connection_id is None
                else ChannelListingPathTarget.connection_id == connection_id
            )
        )
    payload = {
        "listing_path_id": path.id,
        "channel_code": channel,
        "connection_id": connection_id,
        "taxonomy_kind": taxonomy_kind,
        "remote_id": remote_id,
        "remote_path": remote_path,
        "is_active": True,
    }
    if existing is None:
        row = ChannelListingPathTarget(**payload)
        session.add(row)
        if existing_targets is not None and path.id is not None:
            existing_targets[(int(path.id), channel, connection_id, taxonomy_kind, remote_id)] = row
        if flush:
            session.flush()
        return _target_dict(row)
    for key, value in payload.items():
        setattr(existing, key, value)
    if flush:
        session.flush()
    return _target_dict(existing)


def bulk_assign_listing_paths(
    session: Session,
    *,
    path_slugs: List[str],
    skus: Optional[Iterable[str]] = None,
    only_assigned: bool = False,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    search: Optional[str] = None,
    master_filters: Optional[Any] = None,
    dry_run: bool = False,
    replace_existing: bool = False,
    limit: Optional[int] = None,
) -> Dict[str, Any]:
    from db.product_channel_taxonomy import resolve_filtered_master_skus

    slugs = [normalize_plp_path(s) for s in path_slugs if normalize_plp_path(s)]
    if not slugs:
        raise ValueError("path_slugs is required")
    paths = {
        row.path_slug: row
        for row in session.scalars(
            select(ChannelListingPath).where(ChannelListingPath.path_slug.in_(slugs))
        ).all()
    }
    missing = [s for s in slugs if s not in paths]
    if missing:
        raise ValueError(f"Unknown listing paths: {', '.join(missing[:5])}")

    master_skus = resolve_filtered_master_skus(
        session,
        skus=skus,
        channel_code=None,
        only_assigned=only_assigned,
        category_l1=category_l1,
        category_l2=category_l2,
        category_l3=category_l3,
        collection=collection,
        product_family=product_family,
        search=search,
        master_filters=master_filters,
        limit=limit,
    )
    if not master_skus:
        return {"sku_count": 0, "assignment_count": 0, "path_slugs": slugs}

    if replace_existing and not dry_run:
        for master_sku in master_skus:
            for row in session.scalars(
                select(ProductListingPathAssignment).where(ProductListingPathAssignment.master_sku == master_sku)
            ).all():
                row.assignment_status = INACTIVE_STATUS
        session.flush()

    candidates: List[Dict[str, Any]] = []
    for master_sku in master_skus:
        for slug in slugs:
            candidates.append(
                {
                    "master_sku": master_sku,
                    "listing_path_id": paths[slug].id,
                    "path_slug": slug,
                    "match_source": "bulk_assign",
                }
            )

    upserted = 0
    if not dry_run:
        for item in candidates:
            _upsert_listing_assignment(
                session,
                master_sku=item["master_sku"],
                listing_path_id=item["listing_path_id"],
                match_source=item.get("match_source") or "bulk_assign",
            )
            upserted += 1
    return {
        "sku_count": len(master_skus),
        "path_slugs": slugs,
        "assignment_count": len(candidates),
        "upserted": upserted if not dry_run else 0,
        "would_upsert": len(candidates) if dry_run else upserted,
        "samples": candidates[:25],
    }


def bulk_remove_listing_paths(
    session: Session,
    *,
    path_slugs: List[str],
    skus: Optional[Iterable[str]] = None,
    only_assigned: bool = False,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    search: Optional[str] = None,
    master_filters: Optional[Any] = None,
    dry_run: bool = False,
    limit: Optional[int] = None,
) -> Dict[str, Any]:
    from db.product_channel_taxonomy import resolve_filtered_master_skus

    slugs = [normalize_plp_path(s) for s in path_slugs if normalize_plp_path(s)]
    if not slugs:
        raise ValueError("path_slugs is required")
    paths = {
        row.path_slug: row
        for row in session.scalars(
            select(ChannelListingPath).where(ChannelListingPath.path_slug.in_(slugs))
        ).all()
    }
    missing = [s for s in slugs if s not in paths]
    if missing:
        raise ValueError(f"Unknown listing paths: {', '.join(missing[:5])}")

    master_skus = resolve_filtered_master_skus(
        session,
        skus=skus,
        channel_code=None,
        only_assigned=only_assigned,
        category_l1=category_l1,
        category_l2=category_l2,
        category_l3=category_l3,
        collection=collection,
        product_family=product_family,
        search=search,
        master_filters=master_filters,
        limit=limit,
    )
    if not master_skus:
        return {"sku_count": 0, "removed_count": 0, "path_slugs": slugs}

    path_ids = {paths[slug].id for slug in slugs}
    removed = 0
    for row in session.scalars(
        select(ProductListingPathAssignment)
        .where(ProductListingPathAssignment.master_sku.in_(master_skus))
        .where(ProductListingPathAssignment.listing_path_id.in_(path_ids))
        .where(ProductListingPathAssignment.assignment_status == ACTIVE_STATUS)
    ).all():
        removed += 1
        if not dry_run:
            row.assignment_status = INACTIVE_STATUS
    if removed and not dry_run:
        session.flush()
    return {
        "sku_count": len(master_skus),
        "path_slugs": slugs,
        "removed_count": removed,
        "would_remove": removed if dry_run else removed,
        "dry_run": dry_run,
    }


def resolve_listing_path_ids_for_sku(session: Session, master_sku: str) -> List[int]:
    return [
        row.listing_path_id
        for row in session.scalars(
            select(ProductListingPathAssignment)
            .where(ProductListingPathAssignment.master_sku == master_sku)
            .where(ProductListingPathAssignment.assignment_status == ACTIVE_STATUS)
            .order_by(ProductListingPathAssignment.sort_order, ProductListingPathAssignment.id)
        ).all()
    ]


def resolve_listing_paths_for_sku(session: Session, master_sku: str) -> List[Dict[str, Any]]:
    path_ids = resolve_listing_path_ids_for_sku(session, master_sku)
    if not path_ids:
        return []
    paths = list(
        session.scalars(
            select(ChannelListingPath)
            .where(ChannelListingPath.id.in_(path_ids))
            .where(ChannelListingPath.is_active.is_(True))
        ).all()
    )
    targets_by_path = _targets_by_listing_path(session, [p.id for p in paths])
    by_id = {p.id: p for p in paths}
    out: List[Dict[str, Any]] = []
    for path_id in path_ids:
        path = by_id.get(path_id)
        if path is None:
            continue
        out.append(_listing_path_dict(path, targets_by_path.get(path.id, [])))
    return out


def upsert_product_listing_path_assignment(
    session: Session,
    *,
    master_sku: str,
    listing_path_id: int,
    match_source: str = "manual",
    sort_order: int = 0,
    notes: Optional[str] = None,
) -> Dict[str, Any]:
    row = _upsert_listing_assignment(
        session,
        master_sku=master_sku,
        listing_path_id=listing_path_id,
        match_source=match_source,
        sort_order=sort_order,
        notes=notes,
    )
    return {
        "id": row.id,
        "master_sku": row.master_sku,
        "listing_path_id": row.listing_path_id,
        "assignment_status": row.assignment_status,
        "match_source": row.match_source,
        "sort_order": row.sort_order,
        "notes": row.notes,
    }


def expand_listing_path_targets(
    session: Session,
    *,
    master_sku: str,
    channel_code: str,
    connection_id: Optional[int] = None,
) -> Tuple[List[Dict[str, Any]], List[str]]:
    """Return channel taxonomy targets + PLP path slugs for a SKU."""
    channel = resolve_pipeline_channel_code(channel_code)
    plp_paths = resolve_listing_paths_for_sku(session, master_sku)
    path_slugs = [p["path_slug"] for p in plp_paths]
    targets: List[Dict[str, Any]] = []
    seen: Set[Tuple[str, str]] = set()
    for path in plp_paths:
        for target in path.get("targets") or []:
            if target.get("channel_code") != channel:
                continue
            if connection_id is not None and target.get("connection_id") not in (connection_id, None):
                continue
            key = (target.get("taxonomy_kind") or "", target.get("remote_id") or "")
            if key in seen:
                continue
            seen.add(key)
            targets.append({**target, "plp_path_slug": path["path_slug"], "plp_url": path["plp_url"]})
    return targets, path_slugs


def import_listing_paths_csv(session: Session, df: pd.DataFrame) -> Dict[str, int]:
    inserted = 0
    updated = 0
    skipped = 0
    for _, row in df.iterrows():
        path_slug = _cell(row, "path_slug", "plp_path", "path")
        title = _cell(row, "title", "name")
        if not path_slug or not title:
            skipped += 1
            continue
        existing = session.scalar(
            select(ChannelListingPath.id).where(ChannelListingPath.path_slug == normalize_plp_path(path_slug))
        )
        facet_raw = _cell(row, "facet_terms", "facets")
        facet_terms = [t.strip() for t in str(facet_raw or "").split("|") if t.strip()] or None
        upsert_listing_path(
            session,
            path_slug=path_slug,
            title=title,
            path_kind=_cell(row, "path_kind", "kind"),
            parent_path_slug=_cell(row, "parent_path_slug", "parent"),
            hub_path_slug=_cell(row, "hub_path_slug", "hub"),
            facet_terms=facet_terms,
            notes=_cell(row, "notes"),
        )
        if existing is None:
            inserted += 1
        else:
            updated += 1

        magento_id = _cell(row, "magento_category_id", "magento_remote_id")
        shopify_collection = _cell(row, "shopify_collection_id", "shopify_remote_id")
        connection_id = _optional_int(row.get("connection_id"))
        if magento_id:
            upsert_listing_path_target(
                session,
                path_slug=path_slug,
                channel_code="magento",
                remote_id=magento_id,
                taxonomy_kind="category",
                connection_id=connection_id,
                remote_path=_cell(row, "magento_category_path"),
            )
        if shopify_collection:
            upsert_listing_path_target(
                session,
                path_slug=path_slug,
                channel_code="shopify",
                remote_id=shopify_collection,
                taxonomy_kind="collection",
                connection_id=connection_id,
                remote_path=_cell(row, "shopify_collection_handle"),
            )
    session.flush()
    return {"inserted": inserted, "updated": updated, "skipped": skipped}


def _upsert_listing_assignment(
    session: Session,
    *,
    master_sku: str,
    listing_path_id: int,
    match_source: str = "manual",
    sort_order: int = 0,
    notes: Optional[str] = None,
) -> ProductListingPathAssignment:
    existing = session.scalar(
        select(ProductListingPathAssignment)
        .where(ProductListingPathAssignment.master_sku == master_sku)
        .where(ProductListingPathAssignment.listing_path_id == listing_path_id)
    )
    if existing is None:
        row = ProductListingPathAssignment(
            master_sku=master_sku,
            listing_path_id=listing_path_id,
            assignment_status=ACTIVE_STATUS,
            match_source=match_source,
            sort_order=sort_order,
            notes=notes,
        )
        session.add(row)
        session.flush()
        return row
    existing.assignment_status = ACTIVE_STATUS
    existing.match_source = match_source
    existing.sort_order = sort_order
    existing.notes = notes
    session.flush()
    return existing


def bulk_upsert_listing_assignments(
    session: Session,
    *,
    assignments: List[Tuple[str, int]],
    match_source: str = "bulk_assign",
    flush_every: int = 2000,
) -> int:
    """Activate many (master_sku, listing_path_id) pairs with batched reads/flushes.

    Much faster than calling ``_upsert_listing_assignment`` per row (which flushes
    on every write and is what made brochure reconcile look "stuck").
    """
    if not assignments:
        return 0
    unique: Dict[Tuple[str, int], None] = {}
    for master_sku, listing_path_id in assignments:
        sku = str(master_sku or "").strip()
        if not sku or not listing_path_id:
            continue
        unique[(sku, int(listing_path_id))] = None
    pairs = list(unique.keys())
    if not pairs:
        return 0

    skus = sorted({sku for sku, _ in pairs})
    path_ids = sorted({path_id for _, path_id in pairs})
    existing_by_key: Dict[Tuple[str, int], ProductListingPathAssignment] = {}
    # Chunk IN clauses to stay under DB parameter limits.
    chunk = 500
    for i in range(0, len(skus), chunk):
        sku_chunk = skus[i : i + chunk]
        for row in session.scalars(
            select(ProductListingPathAssignment)
            .where(ProductListingPathAssignment.master_sku.in_(sku_chunk))
            .where(ProductListingPathAssignment.listing_path_id.in_(path_ids))
        ).all():
            existing_by_key[(row.master_sku, int(row.listing_path_id))] = row

    upserted = 0
    pending = 0
    for sku, path_id in pairs:
        row = existing_by_key.get((sku, path_id))
        if row is None:
            session.add(
                ProductListingPathAssignment(
                    master_sku=sku,
                    listing_path_id=path_id,
                    assignment_status=ACTIVE_STATUS,
                    match_source=match_source,
                    sort_order=0,
                    notes=None,
                )
            )
        else:
            row.assignment_status = ACTIVE_STATUS
            row.match_source = match_source
        upserted += 1
        pending += 1
        if pending >= flush_every:
            session.flush()
            pending = 0
    if pending:
        session.flush()
    return upserted


def deactivate_listing_assignments_for_skus(
    session: Session,
    *,
    skus: Iterable[str],
) -> int:
    """Mark all listing-path assignments inactive for the given master SKUs."""
    from sqlalchemy import update

    cleaned = sorted({str(sku or "").strip() for sku in skus if str(sku or "").strip()})
    if not cleaned:
        return 0
    total = 0
    chunk = 500
    for i in range(0, len(cleaned), chunk):
        sku_chunk = cleaned[i : i + chunk]
        result = session.execute(
            update(ProductListingPathAssignment)
            .where(ProductListingPathAssignment.master_sku.in_(sku_chunk))
            .where(ProductListingPathAssignment.assignment_status != INACTIVE_STATUS)
            .values(assignment_status=INACTIVE_STATUS)
        )
        total += int(result.rowcount or 0)
    session.flush()
    return total


def _targets_by_listing_path(session: Session, path_ids: List[int]) -> Dict[int, List[Dict[str, Any]]]:
    if not path_ids:
        return {}
    out: Dict[int, List[Dict[str, Any]]] = {}
    for row in session.scalars(
        select(ChannelListingPathTarget)
        .where(ChannelListingPathTarget.listing_path_id.in_(path_ids))
        .where(ChannelListingPathTarget.is_active.is_(True))
    ).all():
        out.setdefault(row.listing_path_id, []).append(_target_dict(row))
    return out


def _listing_path_dict(path: ChannelListingPath, targets: List[Dict[str, Any]]) -> Dict[str, Any]:
    facet_terms = []
    shopping_facet_attribute = None
    if isinstance(path.facet_terms, dict):
        facet_terms = path.facet_terms.get("terms") or []
        shopping_facet_attribute = path.facet_terms.get("shopping_facet_attribute")
    return {
        "id": path.id,
        "path_slug": path.path_slug,
        "plp_url": build_plp_url(path.path_slug),
        "path_kind": path.path_kind,
        "title": path.title,
        "parent_path_slug": path.parent_path_slug,
        "hub_path_slug": path.hub_path_slug,
        "facet_terms": facet_terms,
        "shopping_facet_attribute": shopping_facet_attribute,
        "is_active": path.is_active,
        "notes": path.notes,
        "targets": targets,
    }


def _target_dict(row: ChannelListingPathTarget) -> Dict[str, Any]:
    return {
        "id": row.id,
        "listing_path_id": row.listing_path_id,
        "channel_code": row.channel_code,
        "connection_id": row.connection_id,
        "taxonomy_kind": row.taxonomy_kind,
        "remote_id": row.remote_id,
        "remote_path": row.remote_path,
        "is_active": row.is_active,
    }


def build_pdp_preview(session: Session, product: MasterProduct) -> Dict[str, str]:
    from db.models import MasterProductAttributeValue

    existing = session.scalar(
        select(MasterProductAttributeValue.value)
        .where(MasterProductAttributeValue.product_id == product.id)
        .where(MasterProductAttributeValue.attribute_code == "product_url_slug")
        .limit(1)
    )
    slug = build_pdp_slug(product.name, product.sku, existing_slug=existing)
    return {
        "pdp_slug": slug,
        "pdp_url": build_pdp_url(slug),
        "shopify_handle": slug,
        "magento_url_key": slug,
    }


def _cell(row: pd.Series, *names: str) -> Optional[str]:
    for name in names:
        if name in row.index:
            text = str(row.get(name) or "").strip()
            if text and text.lower() not in {"nan", "none", "null"}:
                return text
    return None


def _optional_int(value: Any) -> Optional[int]:
    text = str(value or "").strip()
    if not text:
        return None
    try:
        return int(text)
    except (TypeError, ValueError):
        return None