"""Canonical taxonomy path aliases and node merge operations."""

from __future__ import annotations

from typing import Any, Dict, List, Optional, Sequence

from sqlalchemy import select, update
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.channel_listing_path import upsert_listing_path
from db.master_taxonomy_hierarchy import _upsert_channel_link
from db.master_taxonomy_sync import _links_for_node
from db.models import (
    ChannelListingPath,
    MasterTaxonomyChannelLink,
    MasterTaxonomyNode,
    MasterTaxonomyPathAlias,
    ProductListingPathAssignment,
)
from db.product_channel_taxonomy import bulk_assign_sku_taxonomy

ACTIVE = "active"
INACTIVE = "inactive"


def load_path_alias_map(session: Session) -> Dict[str, str]:
    """Preload alias_path_slug → active canonical path_slug for hot taxonomy paths."""
    rows = session.execute(
        select(
            MasterTaxonomyPathAlias.alias_path_slug,
            MasterTaxonomyNode.path_slug,
            MasterTaxonomyNode.is_active,
        ).join(
            MasterTaxonomyNode,
            MasterTaxonomyNode.id == MasterTaxonomyPathAlias.canonical_node_id,
        )
    ).all()
    out: Dict[str, str] = {}
    for alias_slug, canonical_slug, is_active in rows:
        if not is_active:
            continue
        key = normalize_plp_path(str(alias_slug or ""))
        value = normalize_plp_path(str(canonical_slug or ""))
        if key and value:
            out[key] = value
    return out


def resolve_canonical_path_slug(
    session: Session,
    path_slug: str,
    *,
    alias_map: Optional[Dict[str, str]] = None,
) -> str:
    """Follow alias chain to the canonical taxonomy path slug."""
    slug = normalize_plp_path(path_slug)
    if not slug:
        return ""
    seen: set[str] = set()
    while slug and slug not in seen:
        seen.add(slug)
        if alias_map is not None:
            next_slug = alias_map.get(slug)
            if not next_slug or next_slug == slug:
                return slug
            slug = next_slug
            continue
        alias = session.scalar(
            select(MasterTaxonomyPathAlias)
            .where(MasterTaxonomyPathAlias.alias_path_slug == slug)
            .limit(1)
        )
        if alias is None:
            return slug
        canonical = session.get(MasterTaxonomyNode, int(alias.canonical_node_id))
        if canonical is None or not canonical.is_active:
            return slug
        slug = normalize_plp_path(canonical.path_slug)
    return slug


def resolve_canonical_node(
    session: Session,
    *,
    node_id: Optional[int] = None,
    path_slug: Optional[str] = None,
) -> Optional[MasterTaxonomyNode]:
    """Resolve a taxonomy node id or path slug to its canonical active node."""
    node: Optional[MasterTaxonomyNode] = None
    if node_id is not None:
        node = session.get(MasterTaxonomyNode, int(node_id))
    elif path_slug:
        canonical_slug = resolve_canonical_path_slug(session, path_slug)
        if canonical_slug:
            node = session.scalar(
                select(MasterTaxonomyNode)
                .where(MasterTaxonomyNode.path_slug == canonical_slug)
                .where(MasterTaxonomyNode.is_active.is_(True))
                .limit(1)
            )
    if node is None or not node.is_active:
        return None
    canonical_slug = resolve_canonical_path_slug(session, node.path_slug)
    if canonical_slug != node.path_slug:
        return session.scalar(
            select(MasterTaxonomyNode)
            .where(MasterTaxonomyNode.path_slug == canonical_slug)
            .where(MasterTaxonomyNode.is_active.is_(True))
            .limit(1)
        )
    return node


def collection_names_for_path_slug(
    session: Session,
    path_slug: str,
    *,
    fallback_collection: Optional[str] = None,
) -> List[str]:
    """All master_product.collection values that belong to one canonical PLP path."""
    from db.collection_landing_pages import canonical_collection

    canonical_slug = resolve_canonical_path_slug(session, path_slug)
    names: set[str] = set()

    def _add_name(value: Optional[str]) -> None:
        coll = canonical_collection(value)
        if coll:
            names.add(coll.lower().replace("grey", "gray"))

    if fallback_collection:
        _add_name(fallback_collection)

    node = resolve_canonical_node(session, path_slug=canonical_slug)
    if node is not None:
        _add_name(node.name)
        alias_rows = session.scalars(
            select(MasterTaxonomyPathAlias).where(
                MasterTaxonomyPathAlias.canonical_node_id == node.id
            )
        ).all()
        slugs = {canonical_slug, *(row.alias_path_slug for row in alias_rows)}
    else:
        slugs = {canonical_slug}

    for slug in slugs:
        _add_name(_collection_name_from_path_slug(slug))
        inactive = session.scalar(
            select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == slug).limit(1)
        )
        if inactive is not None:
            _add_name(inactive.name)

    return sorted(names)


def _collection_name_from_path_slug(path_slug: str) -> Optional[str]:
    slug = normalize_plp_path(path_slug)
    if not slug or "/" not in slug:
        return None
    return slug.rsplit("/", 1)[-1].replace("-", " ").title()


def list_taxonomy_path_aliases(
    session: Session,
    *,
    canonical_node_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    stmt = select(MasterTaxonomyPathAlias).order_by(MasterTaxonomyPathAlias.alias_path_slug)
    if canonical_node_id is not None:
        stmt = stmt.where(MasterTaxonomyPathAlias.canonical_node_id == int(canonical_node_id))
    rows = session.scalars(stmt).all()
    canonical_ids = {row.canonical_node_id for row in rows}
    nodes = {
        node.id: node
        for node in session.scalars(
            select(MasterTaxonomyNode).where(MasterTaxonomyNode.id.in_(canonical_ids))
        ).all()
    }
    return [
        {
            "id": row.id,
            "alias_path_slug": row.alias_path_slug,
            "canonical_node_id": row.canonical_node_id,
            "canonical_path_slug": nodes[row.canonical_node_id].path_slug if row.canonical_node_id in nodes else None,
            "canonical_name": nodes[row.canonical_node_id].name if row.canonical_node_id in nodes else None,
            "notes": row.notes,
        }
        for row in rows
    ]


def register_taxonomy_path_alias(
    session: Session,
    *,
    alias_path_slug: str,
    canonical_node_id: int,
    notes: Optional[str] = None,
) -> Dict[str, Any]:
    alias_slug = normalize_plp_path(alias_path_slug)
    if not alias_slug:
        raise ValueError("alias_path_slug is required")
    canonical = session.get(MasterTaxonomyNode, int(canonical_node_id))
    if canonical is None or not canonical.is_active:
        raise ValueError(f"Unknown canonical taxonomy node: {canonical_node_id}")
    if alias_slug == normalize_plp_path(canonical.path_slug):
        raise ValueError("Alias path cannot equal the canonical path")

    existing_node = session.scalar(
        select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == alias_slug).limit(1)
    )
    if existing_node is not None and existing_node.is_active and existing_node.id != canonical.id:
        raise ValueError(
            f"Active taxonomy node already uses alias path {alias_slug!r} — merge the node instead"
        )

    row = session.scalar(
        select(MasterTaxonomyPathAlias).where(MasterTaxonomyPathAlias.alias_path_slug == alias_slug).limit(1)
    )
    if row is None:
        row = MasterTaxonomyPathAlias(
            alias_path_slug=alias_slug,
            canonical_node_id=canonical.id,
            notes=notes,
        )
        session.add(row)
    else:
        row.canonical_node_id = canonical.id
        row.notes = notes
    session.flush()
    return {
        "alias_path_slug": alias_slug,
        "canonical_node_id": canonical.id,
        "canonical_path_slug": canonical.path_slug,
    }


def suggest_taxonomy_merge_groups(
    session: Session,
    *,
    parent_path_slug: Optional[str] = None,
    limit: int = 200,
) -> List[Dict[str, Any]]:
    """Suggest sibling nodes where one path leaf extends another."""
    nodes = list(
        session.scalars(
            select(MasterTaxonomyNode)
            .where(MasterTaxonomyNode.is_active.is_(True))
            .order_by(MasterTaxonomyNode.path_slug)
        ).all()
    )
    if parent_path_slug:
        parent_slug = normalize_plp_path(parent_path_slug)
        nodes = [n for n in nodes if _parent_path_slug(n.path_slug) == parent_slug]
    by_parent: Dict[Optional[int], List[MasterTaxonomyNode]] = {}
    for node in nodes:
        by_parent.setdefault(node.parent_id, []).append(node)

    suggestions: List[Dict[str, Any]] = []
    for siblings in by_parent.values():
        if len(siblings) < 2:
            continue
        siblings.sort(key=lambda row: (len(row.path_slug), row.path_slug))
        for idx, canonical in enumerate(siblings):
            canonical_leaf = _leaf_slug(canonical.path_slug)
            for alias in siblings[idx + 1 :]:
                alias_leaf = _leaf_slug(alias.path_slug)
                if not alias_leaf.startswith(f"{canonical_leaf}-"):
                    continue
                suggestions.append(
                    {
                        "reason": "extended_leaf_slug",
                        "canonical_node_id": canonical.id,
                        "canonical_path_slug": canonical.path_slug,
                        "canonical_name": canonical.name,
                        "alias_node_id": alias.id,
                        "alias_path_slug": alias.path_slug,
                        "alias_name": alias.name,
                    }
                )
                if len(suggestions) >= limit:
                    return suggestions
    return suggestions


def merge_taxonomy_nodes(
    session: Session,
    *,
    canonical_node_id: int,
    alias_node_ids: Sequence[int],
    dry_run: bool = True,
    deactivate_alias_nodes: bool = True,
) -> Dict[str, Any]:
    """Merge alias taxonomy nodes into one canonical node."""
    canonical = session.get(MasterTaxonomyNode, int(canonical_node_id))
    if canonical is None or not canonical.is_active:
        raise ValueError(f"Unknown canonical taxonomy node: {canonical_node_id}")

    alias_ids = sorted({int(node_id) for node_id in alias_node_ids if int(node_id) != canonical.id})
    if not alias_ids:
        raise ValueError("alias_node_ids is required")

    alias_nodes = []
    for node_id in alias_ids:
        node = session.get(MasterTaxonomyNode, node_id)
        if node is None or not node.is_active:
            raise ValueError(f"Unknown alias taxonomy node: {node_id}")
        alias_nodes.append(node)

    stats: Dict[str, Any] = {
        "dry_run": dry_run,
        "canonical_node_id": canonical.id,
        "canonical_path_slug": canonical.path_slug,
        "aliases": [],
        "listing_assignments_moved": 0,
        "channel_links_copied": 0,
        "path_aliases_registered": 0,
        "alias_nodes_deactivated": 0,
    }

    canonical_links = _links_for_node(session, canonical.id)
    canonical_listing = _get_listing_path(session, canonical.path_slug)

    for alias in alias_nodes:
        alias_result: Dict[str, Any] = {
            "alias_node_id": alias.id,
            "alias_path_slug": alias.path_slug,
            "alias_name": alias.name,
        }
        if dry_run:
            alias_result["would_register_alias"] = True
            alias_result["would_move_assignments"] = _count_listing_assignments(session, alias.path_slug)
            stats["aliases"].append(alias_result)
            continue

        merged_skus = _listing_path_skus(session, alias.path_slug)
        moved = _migrate_listing_assignments(session, alias.path_slug, canonical.path_slug)
        stats["listing_assignments_moved"] += moved
        alias_result["listing_assignments_moved"] = moved

        alias_links = _links_for_node(session, alias.id)
        copied = _copy_missing_channel_links(session, canonical, alias_links, canonical_links)
        stats["channel_links_copied"] += copied
        alias_result["channel_links_copied"] = copied

        if merged_skus and canonical_links:
            for channel_code, link in canonical_links.items():
                if not link.remote_id:
                    continue
                bulk_assign_sku_taxonomy(
                    session,
                    channel_code=channel_code,
                    remote_ids=[link.remote_id],
                    taxonomy_kind=link.taxonomy_kind,
                    connection_id=link.connection_id,
                    skus=sorted(merged_skus),
                    remote_path=link.remote_path or canonical.name,
                    replace_existing=False,
                    dry_run=False,
                )

        alias_path_slug = alias.path_slug
        if deactivate_alias_nodes:
            alias.is_active = False
            alias.notes = (alias.notes or "") + f" merged into node {canonical.id}"
            _deactivate_listing_path(session, alias.path_slug)
            _deactivate_collection_landing_page(session, alias.path_slug)
            stats["alias_nodes_deactivated"] += 1

        register_taxonomy_path_alias(
            session,
            alias_path_slug=alias_path_slug,
            canonical_node_id=canonical.id,
            notes=f"merged from node {alias.id}",
        )
        stats["path_aliases_registered"] += 1

        stats["aliases"].append(alias_result)

    if not dry_run:
        if canonical_listing is None:
            upsert_listing_path(
                session,
                path_slug=canonical.path_slug,
                title=canonical.name,
                path_kind=canonical.node_kind,
                parent_path_slug=_parent_path_slug(canonical.path_slug),
                notes="canonical merge target",
            )
        from db.start_shopping_config_prune import prune_start_shopping_configs

        merge_rewrites = {
            str(alias.get("alias_path_slug") or ""): canonical.path_slug
            for alias in stats["aliases"]
            if alias.get("alias_path_slug")
        }
        stats["start_shopping_prune"] = prune_start_shopping_configs(
            session,
            path_rewrites=merge_rewrites,
            dry_run=False,
            refresh_intersections=True,
        )
        session.flush()
    return stats


def _leaf_slug(path_slug: str) -> str:
    slug = normalize_plp_path(path_slug)
    return slug.rsplit("/", 1)[-1] if slug else ""


def _parent_path_slug(path_slug: str) -> Optional[str]:
    slug = normalize_plp_path(path_slug)
    if not slug or "/" not in slug:
        return None
    return slug.rsplit("/", 1)[0]


def _get_listing_path(session: Session, path_slug: str) -> Optional[ChannelListingPath]:
    slug = normalize_plp_path(path_slug)
    if not slug:
        return None
    return session.scalar(select(ChannelListingPath).where(ChannelListingPath.path_slug == slug).limit(1))


def _listing_path_skus(session: Session, path_slug: str) -> set[str]:
    path = _get_listing_path(session, path_slug)
    if path is None:
        return set()
    rows = session.scalars(
        select(ProductListingPathAssignment.master_sku)
        .where(ProductListingPathAssignment.listing_path_id == path.id)
        .where(ProductListingPathAssignment.assignment_status == ACTIVE)
    ).all()
    return {str(sku) for sku in rows}


def _count_listing_assignments(session: Session, path_slug: str) -> int:
    return len(_listing_path_skus(session, path_slug))


def _migrate_listing_assignments(session: Session, from_slug: str, to_slug: str) -> int:
    from_path = _get_listing_path(session, from_slug)
    if from_path is None:
        return 0
    if _get_listing_path(session, to_slug) is None:
        canonical_node = session.scalar(
            select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == normalize_plp_path(to_slug)).limit(1)
        )
        upsert_listing_path(
            session,
            path_slug=to_slug,
            title=canonical_node.name if canonical_node else to_slug,
            path_kind=canonical_node.node_kind if canonical_node else "category",
            parent_path_slug=_parent_path_slug(to_slug),
            notes="canonical merge target",
        )
    to_path = _get_listing_path(session, to_slug)
    if to_path is None:
        return 0

    moved = 0
    for row in session.scalars(
        select(ProductListingPathAssignment).where(
            ProductListingPathAssignment.listing_path_id == from_path.id,
            ProductListingPathAssignment.assignment_status == ACTIVE,
        )
    ).all():
        # Unique is (master_sku, listing_path_id) regardless of status — an inactive
        # row on the canonical path still blocks UPDATE. Merge must deactivate the
        # alias assignment instead of moving when any target row already exists.
        existing = session.scalar(
            select(ProductListingPathAssignment)
            .where(ProductListingPathAssignment.listing_path_id == to_path.id)
            .where(ProductListingPathAssignment.master_sku == row.master_sku)
            .limit(1)
        )
        if existing is None:
            row.listing_path_id = to_path.id
            row.match_source = "taxonomy_merge"
            moved += 1
        else:
            if existing.assignment_status != ACTIVE:
                existing.assignment_status = ACTIVE
                existing.match_source = "taxonomy_merge"
            row.assignment_status = INACTIVE
            row.match_source = "taxonomy_merge_duplicate"
    session.flush()
    return moved


def _copy_missing_channel_links(
    session: Session,
    canonical: MasterTaxonomyNode,
    alias_links: Dict[str, MasterTaxonomyChannelLink],
    canonical_links: Dict[str, MasterTaxonomyChannelLink],
) -> int:
    copied = 0
    for channel_code, alias_link in alias_links.items():
        if channel_code in canonical_links and canonical_links[channel_code].remote_id:
            continue
        if not alias_link.remote_id:
            continue
        _upsert_channel_link(
            session,
            node_id=canonical.id,
            channel_code=alias_link.channel_code,
            connection_id=alias_link.connection_id,
            taxonomy_kind=alias_link.taxonomy_kind,
            remote_id=alias_link.remote_id,
            remote_parent_id=alias_link.remote_parent_id,
            remote_path=alias_link.remote_path,
        )
        alias_link.is_active = False
        copied += 1
    session.flush()
    return copied


def _deactivate_listing_path(session: Session, path_slug: str) -> None:
    path = _get_listing_path(session, path_slug)
    if path is None:
        return
    path.is_active = False
    session.execute(
        update(ProductListingPathAssignment)
        .where(ProductListingPathAssignment.listing_path_id == path.id)
        .values(assignment_status=INACTIVE)
    )
    session.flush()


def _deactivate_collection_landing_page(session: Session, path_slug: str) -> None:
    from db.models import CollectionLandingPage

    slug = normalize_plp_path(path_slug)
    if not slug:
        return
    row = session.scalar(
        select(CollectionLandingPage).where(CollectionLandingPage.path_slug == slug).limit(1)
    )
    if row is None:
        return
    row.is_active = False
    row.notes = ((row.notes or "").strip() + " merged into canonical taxonomy path").strip()
    session.flush()