import pandas as pd
import pytest
from sqlalchemy import select

from db.channel_catalog_provision import _link_provision_listing_path_targets_magento
from db.collection_landing_push import describe_collection_channel_targets
from db.master_catalog_intent import (
    WRITE_MODE_PREVIEW_ONLY,
    WRITE_MODE_STAGED_IMPORT,
    confirm_intent_run,
    create_intent_preview_only_from_dataframe,
    create_intent_staged_import_from_dataframe,
)
from db.master_taxonomy_registry import (
    backfill_master_taxonomy_from_products,
    list_collections,
    upsert_brand,
    upsert_collection_registry,
)
from db.models import ChannelListingPathTarget, MasterProduct


def _sample_master_csv_df() -> pd.DataFrame:
    return pd.DataFrame(
        [
            {
                "SKU": "ASG-B15",
                "Display Name": "B15 - Anna Stone Gray - Base",
                "Category": "Kitchen Cabinets/Port & Bell/Anna Stone Gray",
                "Product Category": "Kitchen Cabinets",
                "Cabinet Brand": "Port & Bell",
                "Product Collection/Series Name": "Anna Stone Gray",
                "Item Style": "Anna Stone Gray - Assembled",
                "Assembled or RTA": "Assembled",
            }
        ]
    )


def test_preview_only_does_not_persist_skus(catalog_intent_session):
    session = catalog_intent_session
    result = create_intent_preview_only_from_dataframe(session, _sample_master_csv_df(), source_label="test_preview")
    session.commit()

    assert result["write_mode"] == WRITE_MODE_PREVIEW_ONLY
    assert result["valid_skus"] == 1
    assert session.scalar(select(MasterProduct).limit(1)) is None
    assert result["preview"]["write_mode"] == WRITE_MODE_PREVIEW_ONLY


def test_staged_import_confirm_links_registry_and_listing_target(catalog_intent_session):
    session = catalog_intent_session
    # Fixed registry: invent is frozen; confirm only links to existing actives.
    upsert_brand(session, {"name": "Port & Bell", "code": "port-bell"})
    registry = upsert_collection_registry(
        session,
        {
            "name": "Anna Stone Gray",
            "code": "anna-stone-gray",
            "category_l1": "Kitchen Cabinets",
        },
        allow_invent=True,
    )
    session.flush()

    staged = create_intent_staged_import_from_dataframe(session, _sample_master_csv_df(), source_label="test_staged")
    assert staged["write_mode"] == WRITE_MODE_STAGED_IMPORT

    product = session.scalar(select(MasterProduct).where(MasterProduct.sku == "ASG-B15"))
    assert product is not None
    # Import may already FK-link when registry exists; confirm must keep/repair links.
    assert product.collection in {None, "Anna Stone Gray"}

    confirm = confirm_intent_run(
        session,
        staged["intent_run_id"],
        create_registry_rows=False,
        link_products=True,
        run_backfill=False,
        allow_taxonomy_invent=False,
    )
    session.flush()

    session.refresh(product)
    assert confirm["products_linked"] >= 1
    assert product.collection_registry_id == registry["id"]

    linked = _link_provision_listing_path_targets_magento(
        session,
        category_ids_by_path={"Kitchen Cabinets/Anna Stone Gray": 501},
        connection_id=1,
    )
    session.flush()
    assert linked == 1

    target = session.scalar(
        select(ChannelListingPathTarget).where(ChannelListingPathTarget.remote_id == "501").limit(1)
    )
    assert target is not None
    assert target.channel_code == "magento"
    assert target.connection_id == 1


def test_confirm_with_invent_unlock_creates_missing_collection(catalog_intent_session):
    session = catalog_intent_session
    staged = create_intent_staged_import_from_dataframe(session, _sample_master_csv_df(), source_label="test_invent")
    product = session.scalar(select(MasterProduct).where(MasterProduct.sku == "ASG-B15"))
    assert product is not None
    # Import clears unresolved free-text; restore candidate so unlock invent can create it.
    product.collection = "Anna Stone Gray"
    session.flush()

    confirm = confirm_intent_run(
        session,
        staged["intent_run_id"],
        create_registry_rows=True,
        link_products=True,
        run_backfill=True,
        allow_taxonomy_invent=True,
    )
    session.flush()
    session.refresh(product)

    assert confirm.get("invent_allowed") is True
    assert product.collection_registry_id is not None
    assert any(c["name"] == "Anna Stone Gray" for c in list_collections(session))


def test_registry_upsert_preserves_aliases_when_omitted(catalog_intent_session):
    session = catalog_intent_session
    created = upsert_brand(
        session,
        {"name": "Port & Bell", "code": "port-bell", "aliases": ["P&B", "Port and Bell"]},
    )
    updated = upsert_brand(session, {"id": created["id"], "name": "Port & Bell", "code": "port-bell"})
    session.flush()

    assert updated["aliases"] == ["P&B", "Port and Bell"]


def test_preview_only_confirm_requires_imported_skus(catalog_intent_session):
    session = catalog_intent_session
    preview = create_intent_preview_only_from_dataframe(session, _sample_master_csv_df())
    session.flush()

    with pytest.raises(ValueError, match="preview_only"):
        confirm_intent_run(session, preview["intent_run_id"], dry_run=False)


def test_collection_target_resolution_uses_listing_path_target(catalog_intent_session, monkeypatch):
    session = catalog_intent_session
    upsert_brand(session, {"name": "Port & Bell", "code": "port-bell"})
    registry = upsert_collection_registry(
        session,
        {
            "name": "Anna Stone Gray",
            "code": "anna-stone-gray",
            "category_l1": "Kitchen Cabinets",
        },
        allow_invent=True,
    )
    from db.collection_landing_pages import upsert_collection_landing_page

    landing = upsert_collection_landing_page(
        session,
        {
            "path_slug": registry["path_slug"],
            "category_l1": "Kitchen Cabinets",
            "collection": "Anna Stone Gray",
            "title": "Anna Stone Gray",
            "collection_registry_id": registry["id"],
        },
    )
    _link_provision_listing_path_targets_magento(
        session,
        category_ids_by_path={"Kitchen Cabinets/Anna Stone Gray": 777},
        connection_id=2,
    )
    session.flush()

    monkeypatch.setattr(
        "db.collection_landing_push._get_shopify_connection",
        lambda *args, **kwargs: None,
    )
    monkeypatch.setattr(
        "db.collection_landing_push._build_magento_api",
        lambda *args, **kwargs: (object(), 1),
    )

    payload = describe_collection_channel_targets(
        session,
        landing,
        channel_code="magento",
        connection_id=2,
    )
    magento_target = payload["targets"][0]
    assert magento_target["linked"] is True
    assert magento_target["remote_id"] == "777"
    assert payload["registry"]["consistent"] is True


def test_backfill_dry_run_matches_live_create_counts(catalog_intent_session):
    session = catalog_intent_session
    session.add(
        MasterProduct(
            sku="ASG-B15",
            name="Base",
            brand="Port & Bell",
            product_family="Cabinets",
            category_l1="Kitchen Cabinets",
            collection="Anna Stone Gray",
            row_hash="hash-1",
            is_active=True,
        )
    )
    session.flush()

    preview = backfill_master_taxonomy_from_products(session, dry_run=True, allow_invent=True)
    assert preview["brands_created"] == 1
    assert preview["families_created"] == 1
    assert preview["collections_created"] == 1

    applied = backfill_master_taxonomy_from_products(session, dry_run=False, allow_invent=True)
    session.flush()

    assert applied["brands_created"] == 1
    assert applied["families_created"] == 1
    assert applied["collections_created"] == 1
    assert applied["products_linked"] == 1
    assert len(list_collections(session)) == 1

    # Idempotent second run
    again = backfill_master_taxonomy_from_products(session, dry_run=False, allow_invent=True)
    assert again["brands_created"] == 0
    assert again["collections_created"] == 0