"""Data repair from the 2026-08-23 quality audit.

Each step corrects rows written under a bug that is now fixed in code:

1. unaccent extension, so search stops being accent-sensitive.
2. Multipack sizes ("24x330ml" was stored as 330) and the match keys that
   depend on them.
3. Listings attached to a product of a materially different size (retailer
   barcode errors; the ingest now vetoes these). Deleted rather than moved --
   the next collection recreates them under the fixed resolution rules.
4. Duplicate products created by the fallback-matching bug (one new twin per
   airport per run): barcode-less twins merge into the canonical row.
5. Avolta observations that recorded the small-variant price against the
   full-size bottle (fake 3-6x savings). Deleted with their listings; the
   fixed collector plus the ingest quarantine keep them out.
6. Same-timestamp duplicate observations from the catalogue collector.
7. Junk feed strings stored as categories ("View All", "Summer Drinks").
8. Exclusive flags for products whose own name declares them travel-exclusive.
9. One corrupted product name from a source feed.
10. was_price recorded as zero.

Revision ID: d0e1f2a3b4c5
Revises: c9d0e1f2a3b4
"""

import re

from alembic import op
from sqlalchemy import text

revision = "d0e1f2a3b4c5"
down_revision = "c9d0e1f2a3b4"
branch_labels = None
depends_on = None

_URL_SIZE_X = re.compile(r"(?<![0-9a-z])(\d{1,2})\s?x\s?(\d+(?:\.\d+)?)\s?(l|cl|ml)(?![a-z0-9])")
_URL_SIZE = re.compile(r"(?<![0-9a-z])(\d{1,3}(?:\.\d{1,2})?)\s?(l|cl|ml)(?![a-z0-9])")
_FACTORS = {"l": 1000.0, "cl": 10.0, "ml": 1.0}


def _size_from_url(url: str | None) -> int | None:
    if not url:
        return None
    # The size token can sit in any path segment (some shops end the URL with
    # a numeric product code), so scan the whole path.
    seg = url.split("://")[-1].split("?")[0].lower()
    seg = seg.split("/", 1)[1] if "/" in seg else seg
    seg = re.sub(r"(\d)-(\d)", r"\1.\2", seg).replace("-", " ").replace("/", " ")
    pack = _URL_SIZE_X.search(seg)
    if pack:
        value = int(pack.group(1)) * float(pack.group(2)) * _FACTORS[pack.group(3)]
        return round(value) if 10 <= value <= 50000 else None
    match = _URL_SIZE.search(seg)
    if not match:
        return None
    value, unit = float(match.group(1)), match.group(2)
    if unit == "cl" and value < 1:
        unit = "l"
    millilitres = round(value * _FACTORS[unit])
    return millilitres if 10 <= millilitres <= 50000 else None


def _disagree(a: int | None, b: int | None) -> bool:
    if not a or not b:
        return False
    return abs(a - b) / max(a, b) > 0.12


def upgrade() -> None:
    from app.services.normalize import match_key, parse_size_ml
    from app.services.taxonomy import CATEGORY_RULES, classify

    conn = op.get_bind()

    # 1. Accent-insensitive search support.
    conn.execute(text("CREATE EXTENSION IF NOT EXISTS unaccent"))

    # 2. Multipack sizes and dependent match keys.
    for pid, brand, name, size_ml in conn.execute(
        text("SELECT id, brand, name, size_ml FROM products")
    ).all():
        parsed = parse_size_ml(name)
        new_size = parsed if parsed and parsed != size_ml else size_ml
        new_key = match_key(brand, name, new_size)
        if new_size != size_ml:
            conn.execute(
                text("UPDATE products SET size_ml=:s, match_key=:k WHERE id=:i"),
                {"s": new_size, "k": new_key, "i": pid},
            )

    # 3. Listings whose own URL names a materially different size than the
    #    product they are attached to (wrong-barcode merges).
    doomed: list[int] = []
    for lid, url, product_size in conn.execute(
        text(
            "SELECT l.id, l.url, p.size_ml FROM listings l "
            "JOIN products p ON p.id = l.product_id"
        )
    ).all():
        if _disagree(product_size, _size_from_url(url)):
            doomed.append(lid)
    # The one confirmed wrong cross-operator join on a restricted-circulation
    # barcode: a limited edition attached to the regular bottle.
    for (lid,) in conn.execute(
        text("SELECT id FROM listings WHERE url LIKE '%ciroc-blue-stone%'")
    ).all():
        doomed.append(lid)
    if doomed:
        conn.execute(
            text("DELETE FROM price_observations WHERE listing_id = ANY(:ids)"),
            {"ids": doomed},
        )
        conn.execute(text("DELETE FROM listings WHERE id = ANY(:ids)"), {"ids": doomed})

    # 4. Merge barcode-less duplicate products into their canonical twin.
    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()
    merged = 0
    for _, ids in groups:
        canonical, rest = ids[0], ids[1:]
        # Never merge two barcode-bearing rows: different barcodes with one
        # name are different bottles.
        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(
                "UPDATE products c SET thumb_url = d.thumb_url, image_url = d.image_url, "
                "image_source = d.image_source FROM products d "
                "WHERE c.id=:c AND c.thumb_url IS NULL AND d.id = ANY(:ids) "
                "AND d.thumb_url IS NOT NULL AND d.id = "
                "(SELECT min(id) FROM products WHERE id = ANY(:ids) AND thumb_url IS NOT NULL)"
            ),
            {"c": canonical, "ids": mergeable},
        )
        conn.execute(
            text("DELETE FROM products WHERE id = ANY(:ids)"), {"ids": mergeable}
        )
        merged += len(mergeable)

    # 5. Avolta small-variant prices: any Avolta listing whose latest price sits
    #    under 0.45x the median of the product's other shops (3+ of them).
    bad_avolta = [
        row[0]
        for row in conn.execute(
            text(
                """
                WITH latest AS (
                  SELECT DISTINCT ON (po.listing_id) po.listing_id, po.price_usd
                  FROM price_observations po
                  ORDER BY po.listing_id, po.observed_at DESC, po.id DESC
                ),
                peers AS (
                  SELECT l.id AS listing_id,
                    (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY l2latest.price_usd)
                     FROM listings l2
                     JOIN latest l2latest ON l2latest.listing_id = l2.id
                     WHERE l2.product_id = l.product_id AND l2.location_id <> l.location_id
                    ) AS med,
                    (SELECT count(DISTINCT l2.location_id) FROM listings l2
                     WHERE l2.product_id = l.product_id AND l2.location_id <> l.location_id
                    ) AS n
                  FROM listings l
                  JOIN locations loc ON loc.id = l.location_id
                  JOIN retailers r ON r.id = loc.retailer_id
                  WHERE r.slug = 'avolta'
                )
                SELECT p.listing_id FROM peers p
                JOIN latest lt ON lt.listing_id = p.listing_id
                WHERE p.n >= 3 AND p.med IS NOT NULL AND lt.price_usd < 0.45 * p.med
                """
            )
        ).all()
    ]
    if bad_avolta:
        conn.execute(
            text("DELETE FROM price_observations WHERE listing_id = ANY(:ids)"),
            {"ids": bad_avolta},
        )
        conn.execute(
            text("DELETE FROM listings WHERE id = ANY(:ids)"), {"ids": bad_avolta}
        )

    # 5b. Products left with no listings at all (their only listing was wrong
    #     and deleted above): empty husks that would render as "no prices".
    conn.execute(
        text(
            "DELETE FROM awards WHERE product_id IN "
            "(SELECT p.id FROM products p WHERE NOT EXISTS "
            "(SELECT 1 FROM listings l WHERE l.product_id = p.id))"
        )
    )
    conn.execute(
        text(
            "DELETE FROM products p WHERE NOT EXISTS "
            "(SELECT 1 FROM listings l WHERE l.product_id = p.id)"
        )
    )

    # 6. Same-timestamp duplicate observations: keep the first-written row.
    conn.execute(
        text(
            "DELETE FROM price_observations a USING price_observations b "
            "WHERE a.listing_id = b.listing_id AND a.observed_at = b.observed_at "
            "AND a.id > b.id"
        )
    )

    # 7. Junk feed strings stored as categories.
    canonical_categories = {name for name, _ in CATEGORY_RULES}
    for pid, brand, name, category in conn.execute(
        text("SELECT id, brand, name, category FROM products WHERE category IS NOT NULL")
    ).all():
        if category in canonical_categories:
            continue
        conn.execute(
            text("UPDATE products SET category=:c WHERE id=:i"),
            {"c": classify(name, brand, []), "i": pid},
        )

    # 8. Names that declare themselves travel-exclusive.
    conn.execute(
        text(
            "UPDATE products SET is_exclusive = TRUE WHERE is_exclusive = FALSE "
            "AND name ~* 'travel\\s*(retail\\s*)?exclusive|airport\\s+exclusive'"
        )
    )

    # 9. A source feed shipped 'Kentucky' with 'ck' replaced by a brand link.
    conn.execute(
        text(
            "UPDATE products SET name = replace(name, 'KentuCalvin Kleiny', 'Kentucky') "
            "WHERE name LIKE '%KentuCalvin Kleiny%'"
        )
    )

    # 10. A zero was-price is an absent fact, not a discount claim.
    conn.execute(text("UPDATE price_observations SET was_price = NULL WHERE was_price <= 0"))


def downgrade() -> None:
    # Deleted rows were wrong records of the world; there is nothing to restore.
    pass