from __future__ import annotations

from collections import defaultdict
from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple

from sqlalchemy import select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.master_taxonomy_product_paths import (
    canonical_master_taxonomy_path_slugs,
    leaf_taxonomy_path_slugs,
)
from db.models import (
    ChannelListingPath,
    ChannelListingPathTarget,
    MasterProduct,
    ProductListingPathAssignment,
)
from db.product_channel_taxonomy import bulk_remove_sku_taxonomy, resolve_filtered_master_skus


def _split_sku_prefixes(values: Any) -> List[str]:
    if isinstance(values, (list, tuple, set)):
        items = values
    elif values is None:
        items = []
    else:
        items = str(values).replace(";", "\n").replace(",", "\n").splitlines()
    normalized: List[str] = []
    seen = set()
    for item in items:
        prefix = str(item or "").strip().upper()
        if prefix and prefix not in seen:
            seen.add(prefix)
            normalized.append(prefix)
    return normalized


def _path_covered(path_slug: str, candidates: Set[str]) -> bool:
    slug = normalize_plp_path(path_slug)
    if not slug:
        return False
    if slug in candidates:
        return True
    prefix = f"{slug}/"
    return any(other.startswith(prefix) for other in candidates)


def _attrs_by_sku(session: Session, products: Sequence[MasterProduct]) -> Dict[str, Dict[str, str]]:
    if not products:
        return {}
    from db.master_taxonomy_hierarchy import _attrs_by_sku as _load_attrs_by_sku

    return _load_attrs_by_sku(session, [p.id for p in products])


def resolve_stale_listing_path_scope(
    session: Session,
    *,
    skus: Optional[Iterable[str]] = None,
    search: Optional[str] = None,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    master_filters: Optional[Any] = None,
    limit: Optional[int] = None,
    rta_only: bool = False,
    shared_only: bool = False,
    sku_prefixes: Optional[Sequence[str]] = None,
    all_products: bool = False,
) -> Dict[str, Any]:
    prefixes = _split_sku_prefixes(sku_prefixes or [])
    if not any(
        [
            all_products,
            rta_only,
            shared_only,
            prefixes,
            skus,
            search,
            category_l1,
            category_l2,
            category_l3,
            collection,
            product_family,
            master_filters,
        ]
    ):
        raise ValueError(
            "Provide a scope such as all_products=true, rta_only=true, sku_prefixes, skus, search, or other filters."
        )

    resolved = resolve_filtered_master_skus(
        session,
        skus=skus,
        channel_code=None,
        only_assigned=False,
        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 resolved:
        return {
            "requested_skus": [str(s).strip() for s in (skus or []) if str(s).strip()],
            "sku_prefixes": prefixes,
            "rta_only": bool(rta_only),
            "shared_only": bool(shared_only),
            "all_products": bool(all_products),
            "target_skus": [],
            "target_sku_count": 0,
        }

    products = {
        p.sku: p
        for p in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(resolved))).all()
    }
    out: List[str] = []
    for sku in resolved:
        product = products.get(sku)
        if product is None:
            continue
        sku_upper = str(product.sku or "").strip().upper()
        if rta_only and not sku_upper.startswith("RTA-"):
            continue
        if shared_only and str(getattr(product, "membership_mode", "") or "").strip().lower() != "shared":
            continue
        if prefixes and not any(sku_upper.startswith(prefix) for prefix in prefixes):
            continue
        out.append(product.sku)

    return {
        "requested_skus": [str(s).strip() for s in (skus or []) if str(s).strip()],
        "sku_prefixes": prefixes,
        "rta_only": bool(rta_only),
        "shared_only": bool(shared_only),
        "all_products": bool(all_products),
        "target_skus": sorted(out),
        "target_sku_count": len(out),
    }


def prune_stale_listing_paths(
    session: Session,
    *,
    skus: Optional[Iterable[str]] = None,
    search: Optional[str] = None,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    master_filters: Optional[Any] = None,
    limit: Optional[int] = None,
    rta_only: bool = False,
    shared_only: bool = False,
    sku_prefixes: Optional[Sequence[str]] = None,
    all_products: bool = False,
    prune_channel_taxonomy: bool = True,
    dry_run: bool = True,
) -> Dict[str, Any]:
    scope = resolve_stale_listing_path_scope(
        session,
        skus=skus,
        search=search,
        category_l1=category_l1,
        category_l2=category_l2,
        category_l3=category_l3,
        collection=collection,
        product_family=product_family,
        master_filters=master_filters,
        limit=limit,
        rta_only=rta_only,
        shared_only=shared_only,
        sku_prefixes=sku_prefixes,
        all_products=all_products,
    )
    target_skus = list(scope["target_skus"])
    if not target_skus:
        return {
            "status": "ok",
            "dry_run": bool(dry_run),
            "scope": scope,
            "sku_count": 0,
            "stale_sku_count": 0,
            "stale_assignment_count": 0,
            "deactivated_assignment_count": 0,
            "channel_taxonomy": {
                "target_count": 0,
                "would_deactivate": 0,
                "deactivated": 0,
            },
            "samples": [],
        }

    products = {
        p.sku: p
        for p in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(target_skus))).all()
    }
    attrs_by_sku = _attrs_by_sku(session, list(products.values()))

    assignment_rows = session.scalars(
        select(ProductListingPathAssignment)
        .where(ProductListingPathAssignment.master_sku.in_(target_skus))
        .where(ProductListingPathAssignment.assignment_status == "active")
        .order_by(ProductListingPathAssignment.master_sku, ProductListingPathAssignment.sort_order, ProductListingPathAssignment.id)
    ).all()
    path_ids = sorted({int(row.listing_path_id) for row in assignment_rows})
    paths_by_id = {
        row.id: row
        for row in session.scalars(
            select(ChannelListingPath)
            .where(ChannelListingPath.id.in_(path_ids))
            .where(ChannelListingPath.is_active.is_(True))
        ).all()
    }
    targets_by_path_id: Dict[int, List[ChannelListingPathTarget]] = defaultdict(list)
    if path_ids:
        for row in session.scalars(
            select(ChannelListingPathTarget)
            .where(ChannelListingPathTarget.listing_path_id.in_(path_ids))
            .where(ChannelListingPathTarget.is_active.is_(True))
        ).all():
            targets_by_path_id[int(row.listing_path_id)].append(row)

    assignments_by_sku: Dict[str, List[Tuple[ProductListingPathAssignment, str]]] = defaultdict(list)
    for row in assignment_rows:
        path = paths_by_id.get(int(row.listing_path_id))
        slug = normalize_plp_path(path.path_slug if path is not None else "")
        if slug:
            assignments_by_sku[str(row.master_sku)].append((row, slug))

    stale_rows: List[ProductListingPathAssignment] = []
    taxonomy_targets: Dict[Tuple[str, str, Optional[int]], Set[str]] = defaultdict(set)
    samples: List[Dict[str, Any]] = []
    stale_sku_count = 0
    for sku in target_skus:
        product = products.get(sku)
        if product is None:
            continue
        assigned_pairs = assignments_by_sku.get(sku, [])
        assigned_raw = {slug for _, slug in assigned_pairs}
        expected_raw = {
            normalize_plp_path(slug)
            for slug in canonical_master_taxonomy_path_slugs(
                session,
                product,
                attrs_by_sku.get(sku, {}),
            )
            if normalize_plp_path(slug)
        }
        assigned_leaves = set(leaf_taxonomy_path_slugs(list(assigned_raw)))
        expected_leaves = set(leaf_taxonomy_path_slugs(list(expected_raw)))
        stale_leafs = sorted(
            slug for slug in assigned_leaves
            if _path_covered(slug, assigned_raw) and not (slug in expected_leaves or _path_covered(slug, expected_raw))
        )
        if not stale_leafs:
            continue
        stale_sku_count += 1
        for row, slug in assigned_pairs:
            if slug not in stale_leafs:
                continue
            stale_rows.append(row)
            for target in targets_by_path_id.get(int(row.listing_path_id), []):
                key = (
                    str(target.channel_code or "").strip().lower(),
                    str(target.taxonomy_kind or "").strip(),
                    target.connection_id,
                )
                remote_id = str(target.remote_id or "").strip()
                if key[0] and key[1] and remote_id:
                    taxonomy_targets[key].add(remote_id)
        if len(samples) < 25:
            samples.append(
                {
                    "sku": sku,
                    "stale_paths": stale_leafs,
                    "expected_paths": sorted(expected_leaves),
                    "assigned_paths": sorted(assigned_leaves),
                }
            )

    deactivated_assignments = 0
    if stale_rows and not dry_run:
        for row in stale_rows:
            row.assignment_status = "inactive"
            deactivated_assignments += 1

    channel_taxonomy_stats = {
        "target_count": sum(len(ids) for ids in taxonomy_targets.values()),
        "would_deactivate": sum(len(ids) for ids in taxonomy_targets.values()) if dry_run and prune_channel_taxonomy else 0,
        "deactivated": 0,
    }
    if taxonomy_targets and prune_channel_taxonomy and not dry_run:
        for (channel_code, taxonomy_kind, connection_id), remote_ids in taxonomy_targets.items():
            result = bulk_remove_sku_taxonomy(
                session,
                channel_code=channel_code,
                taxonomy_kind=taxonomy_kind,
                remote_ids=sorted(remote_ids),
                connection_id=connection_id,
                skus=target_skus,
                only_assigned=False,
                dry_run=False,
            )
            channel_taxonomy_stats["deactivated"] += int(result.get("deactivated") or 0)

    if stale_rows and not dry_run:
        session.flush()

    return {
        "status": "ok",
        "dry_run": bool(dry_run),
        "scope": scope,
        "sku_count": len(target_skus),
        "stale_sku_count": stale_sku_count,
        "stale_assignment_count": len(stale_rows),
        "deactivated_assignment_count": deactivated_assignments,
        "channel_taxonomy": channel_taxonomy_stats,
        "samples": samples,
    }