"""Publish (Stream K6; plan W18): the facts, the redirects and the candidate rules. What it cost
before: the brand floor decided whether a brand page EXISTED, so a quarter of the visible product
pages would have linked a 404; a rename back to an old address would have left a redirect
pointing at itself; an indexed page renamed at confirm loses the address an engine holds.
SQLite kit, no network."""
from __future__ import annotations

from datetime import UTC, datetime

import pytest
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.models import (Account, Base, Brand, CollectionRun, LEDGER_TABLES, Listing, PriceObservation, ProductLine, ProductVariant,
                        Redirect, Retailer, Shop, Source, Suggestion)
from app.models.places import Place, ShopPlace
from app.models.quality import VerificationCheck
from app.services import publish

TABLES = [Account.__table__, Source.__table__, CollectionRun.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__,
          Retailer.__table__, Shop.__table__, Listing.__table__, PriceObservation.__table__, Suggestion.__table__, Place.__table__,
          ShopPlace.__table__, VerificationCheck.__table__, *LEDGER_TABLES]
NOW = datetime(2026, 9, 17, tzinfo=UTC)


def price(s, variant_id: int, shop_id: int) -> None:
    listing = Listing(variant_id=variant_id, shop_id=shop_id, source_sku=f"v{variant_id}s{shop_id}", listed_name="x")
    s.add(listing); s.flush()
    s.add(PriceObservation(listing_id=listing.id, price=10, currency="USD", price_usd=10, observed_at=NOW))


def variant(s, id_: int, brand_id: int, line_id: int | None, name: str = "v") -> None:
    s.add(ProductVariant(id=id_, name=f"{name} {id_}", brand="B", brand_id=brand_id, vertical="liquor", category="Whisky",
                         match_key=f"k{id_}", product_line_id=line_id, form="single", attributes={}))
    s.flush()


@pytest.fixture
def db():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine, tables=TABLES)
    with sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)() as s:
        s.add_all([Retailer(id=1, slug="r", name="R"), Brand(id=1, slug="macallan", name="The Macallan"),
                   Brand(id=2, slug="thin", name="Thin"), Brand(id=3, slug="macallan-distillery", name="Macallan Distillery", alias_of_id=1)])
        s.flush()
        s.add_all([Shop(id=1, retailer_id=1, code="LHR-T2", iata="LHR", name="Heathrow T2", city="London", currency="GBP", visible=True),
                   Shop(id=2, retailer_id=1, code="LHR-T5", iata="LHR", name="Heathrow T5", city="London", currency="GBP", visible=True),
                   Shop(id=3, retailer_id=1, code="CDG", iata="CDG", name="Paris CDG", city="Paris", currency="EUR", visible=True)])
        s.add_all([ProductLine(id=1, brand_id=1, key="12", slug="macallan-12", name="12 Year Old"),
                   ProductLine(id=2, brand_id=1, key="18", slug="macallan-18", name="18 Year Old"),
                   ProductLine(id=3, brand_id=2, key="one", slug="thin-one", name="One")])
        s.add_all([Place(id=1, slug="lhr-london", kind="airport", name="Heathrow"), Place(id=2, slug="cdg-paris", kind="airport", name="CDG")])
        s.flush()
        s.add_all([ShopPlace(shop_id=1, place_id=1, role="primary"), ShopPlace(shop_id=2, place_id=1, role="primary"),
                   ShopPlace(shop_id=3, place_id=2, role="primary")])
        # Macallan: three variants; variant 1 at LHR and CDG (compared), 2 and 3 at two Heathrow shops only (one place).
        for v, line in ((1, 1), (2, 2), (3, 2)):
            variant(s, v, 1, line)
        price(s, 1, 1); price(s, 1, 3); price(s, 2, 1); price(s, 2, 2); price(s, 3, 1)
        variant(s, 4, 2, 3); price(s, 4, 3)
        s.commit()
        yield s


def test_the_three_facts_read_the_materialised_decision_and_hidden_wins(db):
    brand = db.get(Brand, 1)
    assert not publish.is_indexed(brand) and not publish.is_hidden(brand), "noindex and reachable from discovery"
    brand.indexed = True
    assert publish.is_indexed(brand)
    brand.hidden = True
    assert publish.is_hidden(brand) and not publish.is_indexed(brand), "a hidden page is never indexed"
    alias = db.get(Brand, 3)
    alias.indexed = True
    assert not publish.is_indexed(alias), "a forwarding row answers a redirect"
    assert not publish.is_indexed(None)


def test_a_hidden_page_forwards_to_the_nearest_right_page(db):
    assert publish.hidden_forward(db.get(ProductLine, 1)) == "/brands/macallan"
    assert publish.hidden_forward(db.get(Brand, 1)) == "/products?brand=The%20Macallan"
    assert publish.hidden_forward(db.get(Place, 1)) == "/airports"
    db.get(Brand, 1).hidden = True
    assert publish.hidden_forward(db.get(ProductLine, 1)) == "/products", "never onto another hidden page"
    assert publish.page_path(db.get(Place, 1)) == "/airports/lhr-london" and publish.page_path(db.get(ProductLine, 1)) == "/products/macallan-12"


def test_redirects_flatten_at_write_and_never_point_at_themselves(db):
    publish.redirect_write(db, "product_line", "a", "b")
    publish.redirect_write(db, "product_line", "b", "c")
    assert publish.redirect_for("a", "product_line", db) == "c" and publish.redirect_for("b", "product_line", db) == "c"
    assert publish.redirect_for("a", "brand", db) is None, "a redirect belongs to its kind"
    # Back to an old address: it is live again and stops forwarding; nothing points at itself.
    publish.redirect_write(db, "product_line", "c", "a")
    rows = {r.from_slug: r.to_slug for r in db.scalars(select(Redirect))}
    assert rows == {"b": "a", "c": "a"}
    assert publish.redirect_write(db, "product_line", "a", "a") is None
    # The table's key is the slug alone: another kind's row is kept, not overwritten.
    assert publish.redirect_write(db, "brand", "b", "z") is None and publish.redirect_for("b", "product_line", db) == "a"
    with pytest.raises(ValueError):
        publish.redirect_write(db, "variant", "x", "y")


def test_a_brand_alias_row_is_the_forwarding_address(db):
    assert publish.redirect_for("macallan-distillery", "brand", db) == "macallan"
    assert publish.redirect_for("macallan", "brand", db) is None
    assert publish.find(db, "brand", "macallan-distillery").id == 1 and publish.find(db, "brand", 3).id == 1
    assert publish.find(db, "product_line", "nothing") is None


def test_the_final_slug_is_fixed_at_confirm_and_a_second_alias_flattens(db):
    line = db.get(ProductLine, 1)
    assert publish.slug_at_confirm(line, "Twelve Years Old") == "macallan-twelve-years-old"
    assert publish.slug_at_confirm(line, "Twelve Years Old") is None, "nothing moved"
    assert publish.slug_at_confirm(line, "Macallan Double Cask 12") == "macallan-double-cask-12", "the brand is not said twice"
    assert publish.redirect_for("macallan-12", "product_line", db) == "macallan-double-cask-12"
    assert publish.redirect_for("macallan-twelve-years-old", "product_line", db) == "macallan-double-cask-12"
    assert publish.find(db, "product_line", "macallan-12").id == 1
    # Another line may not take a live slug nor one that forwards elsewhere.
    other = db.get(ProductLine, 2)
    assert publish.slug_at_confirm(other, "Double Cask 12") == "macallan-double-cask-12-2"
    assert publish.slug_at_confirm(other, "12") == "macallan-12-2"
    # A brand keeps its slug: it is the fold key ingest finds the row by.
    assert publish.slug_at_confirm(db.get(Brand, 1), "Macallan Whisky") is None and db.get(Brand, 1).slug == "macallan"


def test_an_indexed_page_is_not_renamed_unless_the_decision_says_so(db):
    line = db.get(ProductLine, 1)
    line.indexed = True
    assert publish.slug_at_confirm(line, "Twelve") is None and line.slug == "macallan-12"
    assert db.scalar(select(func.count()).select_from(Redirect)) == 0
    assert publish.slug_at_confirm(line, "Twelve", rename_indexed=True) == "macallan-twelve"


def test_the_candidate_rules(db):
    found = {(c.kind, c.slug) for c in publish.candidates(db)}
    # Macallan: three variants across two places. Thin: one variant at one place.
    assert ("brand", "macallan") in found and ("brand", "thin") not in found
    # Line 1 has a variant compared across LHR and CDG; line 2's variants sit at two shops of ONE place.
    assert ("product_line", "macallan-12") in found and ("product_line", "macallan-18") not in found
    assert not [c for c in found if c[0] == "place"], "three variants are under the floor of fifteen"
    publish.PLACE_MIN_VARIANTS, keep = 3, publish.PLACE_MIN_VARIANTS
    try:
        assert {c.slug for c in publish.place_candidates(db)} == {"lhr-london"}
    finally:
        publish.PLACE_MIN_VARIANTS = keep
    db.get(Brand, 1).hidden = True
    db.flush()
    assert not publish.brand_candidates(db), "a hidden page is never a candidate"


def test_suggest_is_idempotent_answers_stand_and_removal_is_only_ever_suggested(db):
    assert publish.suggest(db, check=True)["suggested"] == 2 and not publish.open_suggestions(db), "a check writes nothing"
    assert publish.suggest(db)["suggested"] == 2
    again = publish.suggest(db)
    assert again["suggested"] == 0 and again["refreshed"] == 0 and len(publish.open_suggestions(db)) == 2
    # A dismissed suggestion never resurfaces.
    row = next(s for s in publish.open_suggestions(db) if s.level == "index:brand")
    row.decision = "dismissed"
    db.flush()
    assert publish.suggest(db) == {"suggested": 0, "suggested_removal": 0, "refreshed": 0, "withdrawn": 0, "kept": 1}
    # An indexed page under its rule: a suggestion to remove; the page stays indexed.
    thin = db.get(Brand, 2)
    thin.indexed = True
    db.flush()
    assert publish.suggest(db)["suggested_removal"] == 1 and db.get(Brand, 2).indexed is True
    removal = next(s for s in publish.open_suggestions(db) if s.reason == publish.REASON_REMOVE)
    assert removal.left_id == 2 and removal.detail["slug"] == "thin"
    # The page grows back over its rule... here: is de-indexed by a person; the open removal is withdrawn.
    thin.indexed = False
    db.flush()
    assert publish.suggest(db)["withdrawn"] == 1


def test_an_alias_of_an_alias_answers_one_hop(db):
    """On the staging copy `/brands/glen-moray` answered 301 to its first brand and that page a
    second 301 to the brand it had since become an alias of. A chain is followed to its end."""
    from app.services import catalog_queries

    db.get(Brand, 1).alias_of_id = 2  # macallan-distillery -> macallan -> thin
    db.flush()
    assert catalog_queries.brand_by_slug(db, "macallan-distillery").slug == "thin"
    assert publish.redirect_for("macallan-distillery", "brand", db) == "thin"


def test_a_generated_page_says_noindex_follow_until_a_person_approves_it(db):
    from app.services import seo

    shell = "<html><head><title>x</title></head><body></body></html>"
    head = seo._published(db, "brand", "macallan", seo.Head("t", "d", canonical_path="/brands/macallan"))
    assert '<meta name="robots" content="noindex, follow" />' in head.apply(shell, "https://s.example")
    db.get(Brand, 1).indexed = True
    db.flush()
    assert "robots" not in seo._published(db, "brand", "macallan", seo.Head("t", "d")).apply(shell, "")
    # An airport is its place row, found by its code; one with no place row yet is not indexed.
    db.get(Place, 1).identifiers = [{"scheme": "iata", "value": "LHR"}]
    db.get(Place, 1).indexed = True
    db.flush()
    db.info.pop("places_by_identifier", None)
    assert not seo._published(db, "place", "LHR", seo.Head("t", "d")).noindex
    assert seo._published(db, "place", "JFK", seo.Head("t", "d")).noindex
    # The internal pages keep `noindex, nofollow`; a stub session reads as not approved.
    assert 'content="noindex, nofollow"' in seo.STATIC_HEADS["/review"].apply(shell, "")
    assert seo._published(None, "brand", "macallan", seo.Head("t", "d")).noindex


def test_index_decisions_are_batches_a_removal_is_a_standing_no_and_an_approval_is_not(db):
    from app.models import Decision, DecisionBatch

    db.add(Account(id=1, username="rian", display_name="rian")); db.commit()
    publish.suggest(db); db.commit()
    result = publish.set_indexed(db, [("brand", "macallan"), ("product_line", "macallan-12"), ("brand", "nothing"), ("brand", "macallan-distillery")], True, "rian")
    db.commit()
    # The alias names its brand (already in the list), so two pages change and one is "already so".
    assert result["counts"] == {"changed": 2, "unchanged": 1, "refused": 1} and result["mode"] == "bulk"
    assert result["ping"] == ["/brands/macallan", "/products/macallan-12"] and result["refusals"][0]["error_code"] == "PAGE_NOT_FOUND"
    assert db.get(Brand, 1).indexed and db.get(ProductLine, 1).indexed and not publish.open_suggestions(db)
    batch = db.scalar(select(DecisionBatch))
    assert str(batch.uid) == result["batch_uid"] and batch.kind == "cli" and batch.closed_at is not None
    assert db.scalar(select(func.count()).select_from(Decision).where(Decision.batch_id == batch.id, Decision.field == "indexed")) == 2
    # Nothing to record opens no batch.
    assert publish.set_indexed(db, [("brand", "macallan")], True, "rian")["batch_uid"] is None
    assert db.scalar(select(func.count()).select_from(DecisionBatch)) == 1
    # A hidden page is refused.
    db.get(Brand, 2).hidden = True; db.flush()
    assert publish.set_indexed(db, [("brand", "thin")], True, "rian")["refusals"][0]["error_code"] == "PAGE_HIDDEN"
    # An approved page that falls under its rule is still suggested for removal...
    db.get(Brand, 2).hidden = False
    for listing_id in db.scalars(select(Listing.id).where(Listing.shop_id == 3)):
        db.execute(PriceObservation.__table__.delete().where(PriceObservation.listing_id == listing_id))
    db.flush()
    assert publish.suggest(db)["suggested_removal"] == 2  # Paris gone: macallan is at one place, its line compared nowhere
    # ...and a person removing it is a standing no: the rule does not offer it again when it recovers.
    publish.set_indexed(db, [("brand", "macallan")], False, "rian"); db.commit()
    assert not db.get(Brand, 1).indexed
    listing = db.scalar(select(Listing).where(Listing.variant_id == 1, Listing.shop_id == 3))
    db.add(PriceObservation(listing_id=listing.id, price=10, currency="USD", price_usd=10, observed_at=NOW)); db.flush()
    counts = publish.suggest(db)
    assert counts["suggested"] == 0 and counts["kept"] >= 1
    assert not [s for s in publish.open_suggestions(db) if s.level == "index:brand" and s.left_id == 1]
    # The desk row carries the three facts.
    publish.set_indexed(db, [("brand", "thin")], True, "rian"); publish.suggest(db); db.commit()
    row = next(r for r in publish.desk(db)["rows"])
    assert (row["kind"], row["slug"], row["action"], row["indexed"], row["reachable"]) == ("brand", "thin", "remove", True, True)
    assert {"reachable", "quality", "indexed", "hidden", "why", "path", "action"} <= set(row)
    assert publish.dismiss(db, [row["suggestion_id"]], "rian") == 1 and row["suggestion_id"] not in {s.id for s in publish.open_suggestions(db)}


def test_a_hidden_brand_is_absent_everywhere_and_its_address_forwards(db):
    """The brief's acceptance: a hidden brand answers 302 from its page and is in neither the
    index, the sitemap nor a product card's link."""
    from app.services import catalog_queries, seo

    brand = db.get(Brand, 1)
    brand.indexed = True
    db.flush()
    assert "macallan" in seo.indexed_brand_slugs(db) and catalog_queries.brand_page_slugs(db, [1, 3]) == {1: "macallan", 3: "macallan"}
    brand.hidden = True
    db.flush()
    db.info.clear()
    assert publish.hidden_target(db, "brand", "macallan") == "/products?brand=The%20Macallan"
    assert publish.hidden_target(db, "brand", "macallan-distillery") == "/products?brand=The%20Macallan", "its old address too"
    assert publish.hidden_target(db, "brand", "thin") is None and publish.hidden_target(db, "brand", "nothing") is None
    assert catalog_queries.brand_detail(db, "macallan") is None
    assert catalog_queries.brand_page_slugs(db, [1, 3]) == {}, "a card's brand name links a search, not a redirect"
    assert "macallan" not in {b.slug for b in catalog_queries.list_brands(db, 1, 1)}
    assert "macallan" not in seo.indexed_brand_slugs(db), "indexed and hidden: hidden wins"


def test_a_new_product_line_is_reachable_and_noindex_and_only_an_approved_priced_line_is_listed(db):
    from app.services import catalog_queries, seo

    db.add(ProductLine(id=9, brand_id=2, key="new", slug="thin-new", name="New"))
    db.flush()
    line, forward = catalog_queries.line_by_slug(db, "thin-new")
    assert line is not None and forward is None and not publish.is_indexed(line), "a page from the moment the row exists"
    assert seo.indexed_line_rows(db) == []
    db.get(ProductLine, 1).indexed = True
    line.indexed = True  # approved, but nothing priced on it: no address goes to an engine
    db.flush()
    # SQLite hands the timestamp back naive, Postgres aware: the address and the day are the claim.
    assert [(path, last.date()) for path, last in seo.indexed_line_rows(db)] == [("/products/macallan-12", NOW.date())]
    assert seo.indexed_line_rows(db, since=NOW) == [], "IndexNow: only a line that gained an observation after `since`"
    # An airport is approved through its place row.
    db.get(Place, 2).identifiers = [{"scheme": "iata", "value": "cdg"}]
    db.get(Place, 2).indexed = True
    db.flush()
    assert seo.indexed_airport_codes(db) == {"CDG"}


def test_nothing_is_pinged_until_a_person_approves(db, monkeypatch):
    """IndexNow pings only on approval: a suggestion is not a decision, and asking an engine to
    fetch a noindex page spends the quota on a page it must drop."""
    import argparse

    from app import cli_index
    from app.services import indexnow

    db.add(Account(id=1, username="rian", display_name="rian")); db.commit()
    pinged: list[list[str]] = []
    monkeypatch.setattr(indexnow, "submit", lambda urls, base, key, post=None: pinged.append(urls) or [(len(urls), 202)])
    monkeypatch.setattr(cli_index.settings, "public_base_url", "https://s.example")
    monkeypatch.setattr(cli_index.settings, "indexnow_key", "abc123def456")

    class _Ctx:
        def __enter__(self): return db
        def __exit__(self, *exc): return None

    monkeypatch.setattr(cli_index, "SessionLocal", lambda: _Ctx())
    assert cli_index.cmd_suggest(argparse.Namespace(check=False)) == 0 and pinged == []
    args = argparse.Namespace(pages=["brand:macallan"], all_suggested=False, by="rian", reason=None)
    assert cli_index._decide(args, True) == 0
    assert pinged == [["https://s.example/sitemap.xml", "https://s.example/brands/macallan"]]
    assert cli_index._decide(args, True) == 0 and len(pinged) == 1, "already indexed: no second ping"
    assert cli_index._decide(args, False) == 0 and len(pinged) == 1, "a removal pings nothing"