"""Decode HTML entities stored in product names and brands.

One feed ships names like "Bacardí" and they were stored verbatim,
rendering literally on every surface. Ingest now decodes at the boundary; this
repairs the stored rows, recomputes the match keys that depend on the text, and
re-merges any twins the cleanup reunites.

Revision ID: a3b4c5d6e7f8
Revises: f2a3b4c5d6e7
"""

import html

from alembic import op
from sqlalchemy import text

revision = "a3b4c5d6e7f8"
down_revision = "f2a3b4c5d6e7"
branch_labels = None
depends_on = None


def upgrade() -> None:
    from app.services.normalize import match_key

    conn = op.get_bind()
    rows = conn.execute(
        text(
            "SELECT id, brand, name, size_ml FROM products "
            "WHERE name ~ '&[a-zA-Z]+;|&#[0-9]+;' OR brand ~ '&[a-zA-Z]+;|&#[0-9]+;'"
        )
    ).all()
    for pid, brand, name, size_ml in rows:
        new_name = html.unescape(name)
        new_brand = html.unescape(brand) if brand else brand
        conn.execute(
            text("UPDATE products SET name=:n, brand=:b, match_key=:k WHERE id=:i"),
            {"n": new_name, "b": new_brand, "k": match_key(new_brand, new_name, size_ml), "i": pid},
        )

    # A decoded name can now equal a clean twin's; fold barcode-less twins in.
    groups = conn.execute(
        text(
            "SELECT match_key, array_agg(id ORDER BY (gtin IS NOT NULL) DESC, id) "
            "FROM products GROUP BY match_key HAVING count(*) > 1"
        )
    ).all()
    for _, ids in groups:
        canonical, rest = ids[0], ids[1:]
        gtins = dict(
            conn.execute(
                text("SELECT id, gtin FROM products WHERE id = ANY(:ids)"), {"ids": ids}
            ).all()
        )
        mergeable = [i for i in rest if gtins.get(i) is None]
        if not mergeable:
            continue
        conn.execute(
            text("UPDATE listings SET product_id=:c WHERE product_id = ANY(:ids)"),
            {"c": canonical, "ids": mergeable},
        )
        conn.execute(
            text(
                "UPDATE awards a SET product_id=:c WHERE a.product_id = ANY(:ids) "
                "AND NOT EXISTS (SELECT 1 FROM awards b WHERE b.product_id=:c "
                "AND b.competition=a.competition AND b.year IS NOT DISTINCT FROM a.year)"
            ),
            {"c": canonical, "ids": mergeable},
        )
        conn.execute(text("DELETE FROM awards WHERE product_id = ANY(:ids)"), {"ids": mergeable})
        conn.execute(text("DELETE FROM products WHERE id = ANY(:ids)"), {"ids": mergeable})


def downgrade() -> None:
    pass