"""Phase 5 ETL — fact_returns из intake returns.

WB: резолв nm_id → dim_product_identifier (nmID).
Ozon: резолв product.offer_id из return_json → dim_product / identifier offer_id.
Lamoda: резолв offer_id из lamoda_status_dates → dim_product / identifier seller_sku.
YM: резолв order_id → fact_orders (mp∈{6,8}) → model_id + per-unit revenue.
Идемпотентно. Запуск: python -m kernel.etl.build_fact_returns

2026-07-09: order_date/order_id (возвраты по дате ЗАКАЗА, требует миграции
2026-07-09_fact_returns_order_date.sql): WB — srid/g_number/odid возврата →
fact_orders.order_id (mp 1,2); Ozon — posting_number → fact_orders.order_id
(mp 3,4,5). Lamoda/YM не заполняются (NULL). Нет матча заказа → order_date NULL.

Семантика amount/quantity (epic-05/07, вычет возвратов net-of-returns):
  * amount   — ПОЛОЖИТЕЛЬНАЯ величина денег возврата (total по всем возвращённым
    единицам строки), в ТОЙ ЖЕ денежной базе, что числитель продаж соответствующего МП:
      - WB     = ABS(for_pay) из wb_returns (после комиссии; rating/turnover WB тоже for_pay).
                 wb_returns.for_pay в источнике ОТРИЦАТЕЛЕН (сторно выплаты) → берём модуль.
                 total_price (розница) НЕ используем — вычитать розницу из net-выручки нельзя.
      - Ozon   = ABS(price.price × product.quantity) — решение владельца 2026-08-06:
                 все площадки по модулю. Гейт unit_price < 0 ловит отрицательную цену
                 до DML, поэтому в живой проекции ABS холостой; защита от краевых случаев.
      - Lamoda = ABS((fact_orders.revenue / quantity) × число ТОЧНЫХ item_id возврата) —
                 решение владельца 2026-08-06; если fact_order.revenue отрицателен
                 (теоретический случай), модуль предохраняет знак.
                 Ежедневные повторы одного item_id выбираются latest-row, а две единицы
                 одного SKU остаются двумя единицами. Исторический NULL-item fallback = 1.
      - YM     = ABS(SUM((fact_orders.revenue / quantity) × items[].count)) — решение
                 владельца 2026-08-06; Яндекс присылает возвраты отрицательными,
                 модуль приводит к положительному виду.
                 (Та же база, что YM-продажи = prices.total).
  * quantity — число возвращённых единиц строки (WB = 1; Lamoda = COUNT(item_id),
    Ozon/YM из JSON; Lamoda legacy NULL-item fallback = 1).
    Вычет: net_qty = qty_sold − SUM(returns.quantity); net_revenue = revenue − SUM(returns.amount).
  * Тайминг вычета — accrual по return_date (см. build_fact_rating/turnover). return_date IS NULL
    → НЕ вычитается (BETWEEN исключает NULL); алармятся ТОЛЬКО новые NULL-даты — legacy-backfill
    (source_payload_id IS NULL, в т.ч. 863 YM) списан решением владельца 2026-07-03 (F-44).

Ozon 'Cancellation' (канон 2026-06-30, live-evidence 2026-07-21):
  /v1/returns/list отдаёт type ∈ {ClientReturn, Cancellation, SellerReturn}.
  Cancellation = НЕВЫКУП (отмена до выкупа: отказ при вручении / отмена покупателем,
  продавцом или Ozon), а НЕ возврат реализованной единицы: её posting НИКОГДА не
  status='delivered' в fact_orders (live-сверка 2026-07-21: 891 mp3 + 306 mp5 строк
  fact_orders по Cancellation-постингам; статусы cancelled/delivering/
  awaiting_packaging, delivered=0), т.е. единица УЖЕ исключена из realized-числителя
  продаж. Вычесть её ещё и как возврат = двойное вычитание (занижение net): в окне
  рана 3171 Cancellation = 210 шт / 3 026 244 ₽ против 26 шт / 472 008 ₽
  легитимных ClientReturn. Legacy-канон тот же: UI считает возвраты только по
  type='ClientReturn' (OzonSalesResource), невыкупы — отдельно по заказам
  (status='cancelled'); docs/data-api/MARKETPLACE_STATS_FIX_PLAN_2026-06-30.md:
  «фильтровать Ozon по return_json.type='ClientReturn', не включать Cancellation».
  Поэтому Cancellation структурно ВАЛИДИРУЕТСЯ (ключи + точный тип), НЕ попадает в
  fact_returns и учитывается метрикой cancellation_rows. Переключатель владельца —
  env V3_RETURNS_OZON_CANCELLATION_MODE: exclude_validated (DEFAULT) | fail_closed
  (поведение до 2026-07-21: любой тип вне ClientReturn/SellerReturn рушит билд).

Ozon non-events (F-63, live evidence 2026-08-04):
  A ``ClientReturn`` with one of four terminal visual statuses below is not a
  return.  The buyer cancelled it or Ozon/seller rejected it before the return
  logistics started.  Those rows are structurally attested and counted as
  ``non_event_rows``, but never reach ``fact_returns``.  An unknown ClientReturn
  status fails closed before DML; do not repair its date to make the gate pass.
"""
from __future__ import annotations

import logging
import json
import hashlib
import os
import sys
from collections.abc import Mapping
from pathlib import Path
from typing import Any

from db.connection import get_cursor
from kernel.etl.build_input_projection import (
    approved_payload_join,
    require_build_input_projection,
)
from scripts.runtime_data_fence import ensure_kernel_writer_fences

logger = logging.getLogger("v3-kernel.etl.fact_returns")

K = "gwptd_kernel"
I = "gwptd_intake"  # noqa: E741 - established short schema alias in kernel ETL

WB_APPROVED = approved_payload_join(
    # A reviewed legacy recovery supplements the fresh rolling collection.  It
    # keeps its own endpoint/provenance; it must never be presented as a fresh
    # ProductSpec run just to pass the approved-input gate.
    "r", ("wb.returns", "wb.returns.legacy_recovery"), "approved_wb_returns"
)
OZON_APPROVED = approved_payload_join(
    "r", "ozon.returns", "approved_ozon_returns"
)
LAMODA_EXACT_APPROVED = approved_payload_join(
    "sd", "lamoda.status_dates", "approved_lamoda_exact"
)
LAMODA_FALLBACK_APPROVED = approved_payload_join(
    "sd", "lamoda.status_dates", "approved_lamoda_fallback"
)
LAMODA_HISTORY_APPROVED = approved_payload_join(
    "history", "lamoda.status_dates", "approved_lamoda_history"
)
YM_APPROVED = approved_payload_join(
    "r", ("ym.returns", "ym.returns.legacy_recovery"), "approved_ym_returns"
)

_REPO_ROOT = Path(__file__).resolve().parents[3]
ACKNOWLEDGED_YM_LOSS_LEDGER_PATH = (
    _REPO_ROOT / "specs/kernel/ym-acknowledged-missing-returns.json"
)
ACKNOWLEDGED_YM_LOSS_ENTRY_COUNT = 79
ACKNOWLEDGED_YM_LOSS_LEDGER_SHA256 = (
    "484cda6ab4ee16dd8cb38845f18ced6907c425f92c98a683d62483b9a4f9491e"
)


def _load_acknowledged_ym_loss_ledger() -> tuple[dict[str, object], ...]:
    """Load the reviewed finite exception set; reject a permissive rewrite."""
    try:
        raw = ACKNOWLEDGED_YM_LOSS_LEDGER_PATH.read_bytes()
        document = json.loads(raw)
    except (OSError, ValueError) as exc:
        raise RuntimeError("YM acknowledged-loss ledger is unreadable") from exc
    if hashlib.sha256(raw).hexdigest() != ACKNOWLEDGED_YM_LOSS_LEDGER_SHA256:
        raise RuntimeError("YM acknowledged-loss ledger differs from the reviewed file")
    if not isinstance(document, dict) or set(document) != {
        "schemaVersion", "ledgerId", "sourceEndpoint", "observedAt", "proof",
        "expectedEntryCount", "entries",
    }:
        raise RuntimeError("YM acknowledged-loss ledger has an invalid schema")
    entries = document.get("entries")
    if (
        document.get("schemaVersion") != 1
        or document.get("ledgerId") != "ym.returns.acknowledged_missing_orders.v1"
        or document.get("sourceEndpoint") != "ym.returns.legacy_recovery"
        or document.get("expectedEntryCount") != ACKNOWLEDGED_YM_LOSS_ENTRY_COUNT
        or not isinstance(entries, list)
        or len(entries) != ACKNOWLEDGED_YM_LOSS_ENTRY_COUNT
    ):
        raise RuntimeError("YM acknowledged-loss ledger is not the reviewed finite set")
    required = {
        "legacyReturnId", "mpId", "returnId", "orderId", "lineNo", "shopSku",
        "itemCount", "modelId",
    }
    seen: set[tuple[int, str, str, int, str, int, int]] = set()
    normalized: list[dict[str, object]] = []
    for entry in entries:
        if not isinstance(entry, dict) or set(entry) != required:
            raise RuntimeError("YM acknowledged-loss ledger entry has an invalid schema")
        try:
            legacy_return_id = int(entry["legacyReturnId"])
            mp_id = int(entry["mpId"])
            line_no = int(entry["lineNo"])
            item_count = int(entry["itemCount"])
            model_id = int(entry["modelId"])
        except (TypeError, ValueError) as exc:
            raise RuntimeError("YM acknowledged-loss ledger entry has an invalid numeric key") from exc
        return_id = str(entry["returnId"]).strip()
        order_id = str(entry["orderId"]).strip()
        shop_sku = str(entry["shopSku"]).strip()
        key = (mp_id, return_id, order_id, line_no, shop_sku, item_count, model_id)
        if (
            legacy_return_id <= 0 or mp_id != 6 or not return_id or not order_id
            or line_no <= 0 or not shop_sku or item_count <= 0 or model_id <= 0 or key in seen
        ):
            raise RuntimeError("YM acknowledged-loss ledger entry is invalid or duplicated")
        seen.add(key)
        normalized.append(
            {
                "legacyReturnId": legacy_return_id,
                "mpId": mp_id,
                "returnId": return_id,
                "orderId": order_id,
                "lineNo": line_no,
                "shopSku": shop_sku,
                "itemCount": item_count,
                "modelId": model_id,
            }
        )
    return tuple(normalized)


ACKNOWLEDGED_YM_LOSS_ENTRIES = _load_acknowledged_ym_loss_ledger()

WB_CURRENT = "tmp_fact_returns_wb_current"


def _wb_returns_plan() -> tuple[tuple[str, ...], str, str, tuple[str, ...]]:
    """Select one coherent WB row per return and prove its fact projection."""
    prepare = (
        f"DROP TEMPORARY TABLE IF EXISTS {WB_CURRENT}",
        f"""
CREATE TEMPORARY TABLE {WB_CURRENT} (
    source_row_id BIGINT UNSIGNED NOT NULL,
    mp_id TINYINT UNSIGNED NULL,
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    model_id INT UNSIGNED NULL,
    identity_matches INT UNSIGNED NOT NULL,
    return_date DATE NULL,
    order_date DATE NULL,
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    amount DECIMAL(14,2) NULL,
    source_payload_id BIGINT UNSIGNED NULL,
    PRIMARY KEY (source_row_id),
    KEY idx_fact_key (mp_id, return_id, model_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {WB_CURRENT}
  (source_row_id, mp_id, return_id, model_id, identity_matches, return_date,
   order_date, order_id, amount, source_payload_id)
SELECT latest.source_row_id,
       latest.mp_id,
       latest.return_id,
       identity_map.model_id,
       COALESCE(identity_map.identity_matches, 0),
       DATE(latest.return_date_mp),
       matched_order.order_date,
       latest.order_id,
       ABS(latest.for_pay),
       latest.payload_id
FROM (
    SELECT normalized.*,
           ROW_NUMBER() OVER (
               PARTITION BY normalized.mp_id, normalized.return_id
               ORDER BY normalized.collected_at DESC,
                        normalized.payload_id DESC,
                        normalized.source_row_id DESC
           ) AS rn
    FROM (
        SELECT r.id AS source_row_id,
               r.payload_id,
               r.collected_at,
               r.mp_id,
               NULLIF(TRIM(r.return_id), '') AS return_id,
               r.nm_id,
               r.return_date_mp,
               r.for_pay,
               COALESCE(
                   NULLIF(TRIM(r.srid), ''),
                   NULLIF(TRIM(r.g_number), ''),
                   CAST(r.odid AS CHAR)
               ) AS order_id
        FROM {I}.wb_returns r
        {WB_APPROVED}
    ) normalized
) latest
LEFT JOIN (
    SELECT identifier_value,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product_identifier
    WHERE mp_id IN (1, 2)
      AND identifier_type = 'nmID'
      AND is_current = 1
    GROUP BY identifier_value
) identity_map
  ON identity_map.identifier_value = CAST(latest.nm_id AS CHAR)
       COLLATE utf8mb4_unicode_ci
LEFT JOIN (
    SELECT fo.mp_id,
           fo.order_id,
           fo.model_id,
           MAX(fo.order_date) AS order_date
    FROM {K}.fact_orders fo
    WHERE fo.mp_id IN (1, 2)
    GROUP BY fo.mp_id, fo.order_id, fo.model_id
) matched_order
  ON matched_order.mp_id = latest.mp_id
 AND matched_order.order_id = latest.order_id COLLATE utf8mb4_unicode_ci
 AND matched_order.model_id = identity_map.model_id
WHERE latest.rn = 1
""",
    )
    # MySQL cannot reopen a TEMPORARY table twice in one statement (ERROR
    # 1137): the duplicate-group count is folded into the same single pass over
    # {WB_CURRENT} via window functions instead of a second derived read.
    # key_rows counts rows per natural key (window PARTITION groups NULLs
    # exactly like GROUP BY); key_rn = 1 counts each duplicate GROUP once,
    # preserving the old COUNT-of-groups semantics.
    attest = f"""
SELECT COUNT(*) AS selected_rows,
       COALESCE(SUM(
         CASE WHEN current_return.mp_id NOT IN (1, 2)
                OR current_return.return_id IS NULL
                OR current_return.source_payload_id IS NULL
              THEN 1 ELSE 0 END
       ), 0) AS invalid_keys,
       COALESCE(SUM(current_return.return_date IS NULL), 0) AS invalid_dates,
       COALESCE(SUM(current_return.amount IS NULL), 0) AS invalid_amounts,
       COALESCE(SUM(current_return.identity_matches <> 1), 0) AS identity_failures,
       COALESCE(SUM(
         CASE WHEN current_return.model_id IS NOT NULL
               AND current_return.key_rows <> 1
               AND current_return.key_rn = 1
              THEN 1 ELSE 0 END
       ), 0) AS natural_key_duplicates
FROM (
    SELECT wb_current.*,
           COUNT(*) OVER (
               PARTITION BY wb_current.mp_id,
                            wb_current.return_id,
                            wb_current.model_id
           ) AS key_rows,
           ROW_NUMBER() OVER (
               PARTITION BY wb_current.mp_id,
                            wb_current.return_id,
                            wb_current.model_id
               ORDER BY wb_current.source_row_id
           ) AS key_rn
    FROM {WB_CURRENT} wb_current
) current_return
"""
    upsert = f"""
INSERT INTO {K}.fact_returns
  (model_id, mp_id, return_id, return_date, order_date, order_id, amount, quantity, status,
   source_payload_id)
SELECT current_return.model_id,
       current_return.mp_id,
       current_return.return_id,
       current_return.return_date,
       current_return.order_date,
       current_return.order_id,
       current_return.amount,
       1,
       NULL,
       current_return.source_payload_id
FROM {WB_CURRENT} current_return
ON DUPLICATE KEY UPDATE
  amount=VALUES(amount), quantity=VALUES(quantity), return_date=VALUES(return_date),
  order_date=VALUES(order_date), order_id=VALUES(order_id),
  source_payload_id=VALUES(source_payload_id)
"""
    cleanup = (f"DROP TEMPORARY TABLE IF EXISTS {WB_CURRENT}",)
    return prepare, attest, upsert, cleanup


WB_PREPARE_SQLS, WB_ATTEST_SQL, WB_SQL, WB_CLEANUP_SQLS = _wb_returns_plan()

OZON_CURRENT = "tmp_fact_returns_ozon_current"

# Переключатель семантики Ozon Cancellation (канон — см. шапку модуля).
# exclude_validated (DEFAULT) — Cancellation структурно валидируется
#   (ключи + точный тип), исключается из fact_returns и считается метрикой
#   cancellation_rows; любой ДРУГОЙ отсутствующий/новый тип по-прежнему рушит
#   билд (гейт НЕ ослаблен).
# fail_closed — поведение до 2026-07-21: любой тип вне
#   ('ClientReturn', 'SellerReturn') считается invalid_types и рушит билд.
OZON_CANCELLATION_MODE_ENV = "V3_RETURNS_OZON_CANCELLATION_MODE"
OZON_CANCELLATION_MODE_EXCLUDE = "exclude_validated"
OZON_CANCELLATION_MODE_FAIL_CLOSED = "fail_closed"
_OZON_CANCELLATION_MODES = frozenset(
    {OZON_CANCELLATION_MODE_EXCLUDE, OZON_CANCELLATION_MODE_FAIL_CLOSED}
)

# F-63: these are terminal visual statuses whose business meaning is that no
# return occurred.  They are deliberately distinct from Cancellation: Ozon
# emits them with return_type=ClientReturn, so the type-only rule is insufficient.
OZON_NON_EVENT_RETURN_STATUSES = (
    "Отменена покупателем",
    "Отклонена вами",
    "Отклонена, спор не открыт",
    "Отклонена Ozon",
)

# Reviewed ClientReturn status vocabulary must stay explicit.  A new
# status is a source-semantic change, not permission to publish a money fact.
#
# Measured on the live intake 2026-08-04 (gwptd_intake.ozon_returns, all rows
# grouped by visual.status.display_name × type).  The four non-event statuses
# above exist ONLY in the legacy mirror: the current collector has never seen
# them.  The live ClientReturn vocabulary is the list below — it was NOT
# derivable from the 168 legacy rows that started this investigation, and a
# dictionary built from those alone would have failed the build on the very
# first real run.  Counts at the time of measurement, ClientReturn only:
#   Получен 1044 · Едет к вам 425 · На складе Ozon 337 · В пункте выдачи 243
#   Ожидает отправки 157 · Едет на склад Ozon 57 · Отказано в компенсации 8
OZON_EVENT_RETURN_STATUSES = (
    # legacy-only spellings, kept so historical rows stay classifiable
    "Одобрена",
    "Возвращена часть денег",
    "Получен продавцом",
    "Возвращён в Ozon",
    # live vocabulary
    "Получен",
    "Едет к вам",
    "На складе Ozon",
    "В пункте выдачи",
    "Ожидает отправки",
    "Едет на склад Ozon",
    "Отказано в компенсации",
)
OZON_CLIENT_RETURN_STATUS_VOCABULARY = (
    *OZON_NON_EVENT_RETURN_STATUSES,
    *OZON_EVENT_RETURN_STATUSES,
)


def _sql_string_literals(values: tuple[str, ...]) -> str:
    """Render a fixed reviewed vocabulary as MySQL string literals."""
    return ", ".join(f"'{value.replace(chr(39), chr(39) * 2)}'" for value in values)


OZON_NON_EVENT_RETURN_STATUS_SQL = _sql_string_literals(OZON_NON_EVENT_RETURN_STATUSES)
OZON_CLIENT_RETURN_STATUS_VOCABULARY_SQL = _sql_string_literals(
    OZON_CLIENT_RETURN_STATUS_VOCABULARY
)


def _ozon_attest_sql(mode: str) -> str:
    """Ozon projection attestation for one Cancellation-handling mode.

    Both modes validate every selected row's keys, dates, prices, quantities,
    identity and natural-key uniqueness — the gate is never weakened.  They
    differ only in whether the known невыкуп type ``Cancellation`` belongs to
    the accepted type vocabulary (it is never fact-published in any mode).
    ``cancellation_rows`` is observability in both modes, never a failure.
    """
    if mode == OZON_CANCELLATION_MODE_EXCLUDE:
        accepted_types = "'ClientReturn', 'SellerReturn', 'Cancellation'"
    elif mode == OZON_CANCELLATION_MODE_FAIL_CLOSED:
        accepted_types = "'ClientReturn', 'SellerReturn'"
    else:
        raise ValueError(f"unknown Ozon cancellation mode: {mode!r}")
    # MySQL cannot reopen a TEMPORARY table twice in one statement (ERROR
    # 1137): the duplicate-group count shares the single pass over
    # {OZON_CURRENT} via window functions (same key_rows/key_rn group-count
    # semantics as the former second derived read).
    return f"""
SELECT COUNT(*) AS selected_rows,
       COALESCE(SUM(
         CASE WHEN current_return.return_id IS NULL
                OR current_return.offer_id IS NULL
                OR current_return.source_payload_id IS NULL
              THEN 1 ELSE 0 END
       ), 0) AS invalid_keys,
       COALESCE(SUM(
         current_return.return_type IS NULL
         OR current_return.return_type NOT IN ({accepted_types})
       ), 0) AS invalid_types,
       COALESCE(SUM(current_return.return_type = 'Cancellation'), 0)
         AS cancellation_rows,
       COALESCE(SUM(
         current_return.return_type = 'ClientReturn'
         AND (
           current_return.status IS NULL
           OR current_return.status NOT IN ({OZON_CLIENT_RETURN_STATUS_VOCABULARY_SQL})
         )
       ), 0) AS invalid_statuses,
       COALESCE(SUM(
         current_return.return_type = 'ClientReturn'
         AND current_return.status IN ({OZON_NON_EVENT_RETURN_STATUS_SQL})
       ), 0) AS non_event_rows,
       COALESCE(SUM(
         current_return.return_date IS NULL
         AND NOT (
           current_return.return_type = 'ClientReturn'
           AND current_return.status IN ({OZON_NON_EVENT_RETURN_STATUS_SQL})
         )
       ), 0) AS invalid_dates,
       COALESCE(SUM(
         current_return.unit_price IS NULL OR current_return.unit_price < 0
       ), 0) AS invalid_prices,
       COALESCE(SUM(
         current_return.quantity IS NULL OR current_return.quantity <= 0
       ), 0) AS invalid_quantities,
       COALESCE(SUM(current_return.identity_matches <> 1), 0) AS identity_failures,
       COALESCE(SUM(
         CASE WHEN current_return.return_id IS NOT NULL
               AND current_return.model_id IS NOT NULL
               AND current_return.key_rows <> 1
               AND current_return.key_rn = 1
              THEN 1 ELSE 0 END
       ), 0) AS natural_key_duplicates
FROM (
    SELECT ozon_current.*,
           COUNT(*) OVER (
               PARTITION BY ozon_current.return_id, ozon_current.model_id
           ) AS key_rows,
           ROW_NUMBER() OVER (
               PARTITION BY ozon_current.return_id, ozon_current.model_id
               ORDER BY ozon_current.source_row_id
           ) AS key_rn
    FROM {OZON_CURRENT} ozon_current
) current_return
"""


def _ozon_returns_plan() -> tuple[tuple[str, ...], str, str, str, tuple[str, ...]]:
    """Deterministic Ozon snapshot projection and natural-key retraction.

    ``ozon_returns`` is an append-only journal.  Independent ``MAX()`` values used
    to combine status/reason/date/price/payload from different journal rows.  The
    temporary table below is the single latest *whole row* for each
    ``(return_id, offer_id)`` (tie-break: collected_at, payload_id, id).

    Type semantics (канон 2026-06-30): ``ClientReturn`` publishes a fact and
    ``SellerReturn`` retracts an S3-owned fact.  ``Cancellation`` is a validated
    невыкуп (pre-buyout cancellation, never a realized sale): the gate proves its
    keys/type, counts it as ``cancellation_rows`` and the DML type filters keep it
    out of ``fact_returns`` — excluding it is safe because its posting never
    reaches status='delivered' in fact_orders.  Missing or novel types are
    semantic failures in every mode, never deletion signals.

    F-63 adds a second non-return class: a ClientReturn with one of the reviewed
    terminal rejected/cancelled visual statuses is counted but never published.
    A previously published S3-owned row is retracted if its latest snapshot enters
    that class.  Unknown ClientReturn statuses are attestation failures, never a
    date-based fallback or a deletion signal.
    """
    prepare = (
        f"DROP TEMPORARY TABLE IF EXISTS {OZON_CURRENT}",
        f"""
CREATE TEMPORARY TABLE {OZON_CURRENT} (
    source_row_id BIGINT UNSIGNED NOT NULL,
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    offer_id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    model_id INT UNSIGNED NULL,
    identity_matches INT UNSIGNED NOT NULL,
    return_type VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    return_date DATE NULL,
    order_date DATE NULL,
    posting_number VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    unit_price DECIMAL(14,2) NULL,
    quantity INT NULL,
    reason VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    source_payload_id BIGINT UNSIGNED NULL,
    PRIMARY KEY (source_row_id),
    KEY idx_natural_key (return_id, offer_id),
    KEY idx_type_key (return_type, return_id, model_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {OZON_CURRENT}
  (source_row_id, return_id, offer_id, model_id, identity_matches, return_type,
   return_date, posting_number, unit_price, quantity, reason, status,
   source_payload_id)
SELECT latest.source_row_id,
       latest.return_id,
       latest.offer_id,
       CASE
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 0
         THEN idn.model_id
         WHEN COALESCE(dp.identity_matches, 0) = 1
          AND COALESCE(idn.identity_matches, 0) = 0
         THEN dp.model_id
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 1
          AND idn.model_id = dp.model_id
         THEN idn.model_id
       END AS model_id,
       CASE
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 0
         THEN 1
         WHEN COALESCE(dp.identity_matches, 0) = 1
          AND COALESCE(idn.identity_matches, 0) = 0
         THEN 1
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 1
          AND idn.model_id = dp.model_id
         THEN 1
         ELSE COALESCE(idn.identity_matches, 0) + COALESCE(dp.identity_matches, 0)
       END AS identity_matches,
       latest.return_type,
       latest.return_date,
       latest.posting_number,
       CASE
         WHEN latest.unit_price_text IS NOT NULL
          AND CHAR_LENGTH(latest.unit_price_text) <= 32
          AND REGEXP_LIKE(
                latest.unit_price_text,
                '^[0-9]+([.][0-9]+)?$',
                'c'
              )
         THEN CAST(latest.unit_price_text AS DECIMAL(14,2))
       END AS unit_price,
       CASE
         WHEN latest.quantity_text IS NOT NULL
          AND CHAR_LENGTH(latest.quantity_text) <= 9
          AND REGEXP_LIKE(latest.quantity_text, '^[1-9][0-9]*$', 'c')
         THEN CAST(latest.quantity_text AS UNSIGNED)
       END AS quantity,
       latest.reason,
       latest.status,
       latest.payload_id
FROM (
    SELECT normalized.*,
           ROW_NUMBER() OVER (
               PARTITION BY normalized.return_id, normalized.offer_id
               ORDER BY normalized.collected_at DESC,
                        normalized.payload_id DESC,
                        normalized.source_row_id DESC
           ) AS rn
    FROM (
        SELECT r.id AS source_row_id,
               r.payload_id,
               r.collected_at,
               NULLIF(TRIM(CAST(r.return_id AS CHAR)), '') AS return_id,
               COALESCE(
                   NULLIF(TRIM(r.offer_id), ''),
                   NULLIF(TRIM(JSON_UNQUOTE(JSON_EXTRACT(r.return_json, '$.product.offer_id'))), '')
               ) AS offer_id,
               COALESCE(
                   NULLIF(TRIM(r.return_type), ''),
                   NULLIF(TRIM(JSON_UNQUOTE(JSON_EXTRACT(r.return_json, '$.type'))), '')
               ) AS return_type,
               DATE(COALESCE(
                   CASE
                     WHEN REGEXP_LIKE(
                         JSON_UNQUOTE(JSON_EXTRACT(
                             r.return_json, '$.logistic.return_date'
                         )),
                         '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}([.][0-9]+)?Z$',
                         'c'
                     )
                     THEN STR_TO_DATE(
                         -- Ozon emits both ``...:ssZ`` and
                         -- ``...:ss.ffffffZ``.  Remove only a proven UTC
                         -- fractional suffix; malformed values stay NULL and
                         -- use the typed intake fallbacks below.
                         REGEXP_REPLACE(
                             JSON_UNQUOTE(JSON_EXTRACT(
                                 r.return_json, '$.logistic.return_date'
                             )),
                             '[.][0-9]+Z$',
                             'Z'
                         ),
                         '%Y-%m-%dT%H:%i:%sZ'
                     )
                   END,
                   r.return_date_mp,
                   r.visual_status_change_moment_mp
               )) AS return_date,
               r.posting_number,
               NULLIF(TRIM(COALESCE(
                   CAST(r.price AS CHAR),
                   JSON_UNQUOTE(JSON_EXTRACT(
                       r.return_json, '$.product.price.price'
                   ))
               )), '') AS unit_price_text,
               NULLIF(TRIM(COALESCE(
                   r.quantity,
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(
                       r.return_json, '$.product.quantity'
                   )) AS CHAR)
               )), '') AS quantity_text,
               r.return_reason AS reason,
               r.return_status AS status
        FROM {I}.ozon_returns r
        {OZON_APPROVED}
    ) normalized
) latest
LEFT JOIN (
    SELECT identifier_value,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product_identifier
    WHERE mp_id = 3
      AND identifier_type = 'offer_id'
      AND is_current = 1
    GROUP BY identifier_value
) idn
  ON idn.identifier_value = latest.offer_id COLLATE utf8mb4_unicode_ci
LEFT JOIN (
    SELECT artikul_upper,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product
    GROUP BY artikul_upper
) dp
  ON dp.artikul_upper = UPPER(TRIM(latest.offer_id)) COLLATE utf8mb4_unicode_ci
WHERE latest.rn = 1
""",
        f"""
UPDATE {OZON_CURRENT} current_return
LEFT JOIN (
    SELECT fo.order_id, fo.model_id, MAX(fo.order_date) AS order_date
    FROM {K}.fact_orders fo
    WHERE fo.mp_id IN (3, 4, 5)
    GROUP BY fo.order_id, fo.model_id
) matched_order
  ON matched_order.order_id = current_return.posting_number COLLATE utf8mb4_unicode_ci
 AND matched_order.model_id = current_return.model_id
SET current_return.order_date = matched_order.order_date
""",
    )
    attest = _ozon_attest_sql(OZON_CANCELLATION_MODE_EXCLUDE)
    cleanup_non_client = f"""
DELETE fact_return
FROM {OZON_CURRENT} current_return
JOIN {K}.fact_returns fact_return
  ON fact_return.mp_id = 3
 AND fact_return.return_id = current_return.return_id COLLATE utf8mb4_unicode_ci
 AND fact_return.model_id = current_return.model_id
WHERE (
      current_return.return_type = 'SellerReturn'
   OR (
        current_return.return_type = 'ClientReturn'
    AND current_return.status IN ({OZON_NON_EVENT_RETURN_STATUS_SQL})
   )
)
  AND current_return.model_id IS NOT NULL
  AND fact_return.source_payload_id IS NOT NULL
"""
    upsert = f"""
INSERT INTO {K}.fact_returns
  (model_id, mp_id, return_id, return_date, order_date, order_id, amount, quantity, reason,
   status, source_payload_id)
SELECT current_return.model_id,
       3 AS mp_id,
       current_return.return_id,
       current_return.return_date,
       current_return.order_date,
       current_return.posting_number,
       ABS(current_return.unit_price * current_return.quantity) AS amount,
       current_return.quantity,
       current_return.reason,
       current_return.status,
       current_return.source_payload_id
FROM {OZON_CURRENT} current_return
LEFT JOIN {K}.fact_returns legacy_return
  ON legacy_return.mp_id = 3
 AND legacy_return.return_id = current_return.return_id COLLATE utf8mb4_unicode_ci
 AND legacy_return.model_id = current_return.model_id
 AND legacy_return.source_payload_id IS NULL
WHERE current_return.return_type = 'ClientReturn'
  AND current_return.status NOT IN ({OZON_NON_EVENT_RETURN_STATUS_SQL})
  AND current_return.model_id IS NOT NULL
  AND legacy_return.id IS NULL
ON DUPLICATE KEY UPDATE
  return_date=VALUES(return_date),
  order_date=VALUES(order_date),
  order_id=VALUES(order_id),
  amount=VALUES(amount),
  quantity=VALUES(quantity),
  reason=VALUES(reason),
  status=VALUES(status),
  source_payload_id=VALUES(source_payload_id)
"""
    cleanup = (f"DROP TEMPORARY TABLE IF EXISTS {OZON_CURRENT}",)
    return prepare, attest, cleanup_non_client, upsert, cleanup


(
    OZON_PREPARE_SQLS,
    OZON_ATTEST_SQL,
    OZON_NON_CLIENT_CLEANUP_SQL,
    OZON_SQL,
    OZON_CLEANUP_SQLS,
) = _ozon_returns_plan()

# Строгий гейт до 2026-07-21 — выполняется только при явном
# V3_RETURNS_OZON_CANCELLATION_MODE=fail_closed.
OZON_ATTEST_SQL_FAIL_CLOSED = _ozon_attest_sql(OZON_CANCELLATION_MODE_FAIL_CLOSED)

# Lamoda: единственный живой источник — lamoda_status_dates (B2B-контур, статусы
# Returned/Claimed*). Второй контур — FBS Seller Center — собирается коллектором
# lamoda.returns в {I}.lamoda_returns / lamoda_return_boxes, но live-проверка
# 2026-07-09 показала: FBS-очередь возвратов у продавца ПУСТА с 2023 года (возвраты
# идут B2B-контуром). Ветку ETL из lamoda_returns НЕ пишем, пока нет живых данных —
# маппинг полей (returnType/price-копейки/externalSku) не на чем проверить.
# Как только в {I}.lamoda_returns появятся строки — добавить сюда LAMODA_FBS_SQL
# (резолв external_sku → dim_product_identifier seller_sku, natural key return_item_id).
#
# B2B detail-item contract (2026-07-14): order-level status='Returned' решает,
# является ли строка возвратом. item_status сохраняется только как intake lineage и
# НЕ фильтрует позиции: две detail-позиции одного SKU должны дать quantity=2 даже
# при разных item_status. Ежедневные повторы выбираются latest по стабильному
# (order_id,item_id). Исторические строки без item_id дают ровно одну fallback-
# единицу на order:offer, только если точных позиций для того же ключа ещё нет.
LAMODA_UNITS = "tmp_fact_returns_lamoda_units"
LAMODA_EXACT_KEYS = "tmp_fact_returns_lamoda_exact_keys"
LAMODA_EXPECTED = "tmp_fact_returns_lamoda_expected"
LAMODA_SCOPE = "tmp_fact_returns_lamoda_scope"

LAMODA_TYPED_ATTEST_SQL = f"""
SELECT COALESCE(SUM(sd.status = 'Returned'), 0) AS return_rows,
       COALESCE(SUM(
         sd.status = 'Returned'
         AND (
           sd.mp_id IS NULL OR sd.mp_id <> 9
           OR sd.order_id IS NULL OR TRIM(sd.order_id) = ''
           OR sd.offer_id IS NULL OR TRIM(sd.offer_id) = ''
           OR (sd.item_id IS NOT NULL AND TRIM(sd.item_id) = '')
           OR COALESCE(sd.return_date, sd.status_date) IS NULL
           OR sd.payload_id IS NULL
         )
       ), 0) AS invalid_rows
FROM {I}.lamoda_status_dates sd
{LAMODA_EXACT_APPROVED}
"""


def _lamoda_returns_plan() -> tuple[tuple[str, ...], str, str, str, tuple[str, ...]]:
    """Build the latest Lamoda item set and reconcile only S3-owned facts."""
    prepare = (
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_SCOPE}",
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_EXPECTED}",
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_EXACT_KEYS}",
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_UNITS}",
        f"""
CREATE TEMPORARY TABLE {LAMODA_UNITS} (
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    offer_id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    unit_key VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    is_exact TINYINT(1) NOT NULL,
    model_id INT UNSIGNED NULL,
    identity_matches INT UNSIGNED NOT NULL,
    return_date DATE NULL,
    order_status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    item_status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    source_payload_id BIGINT UNSIGNED NOT NULL,
    PRIMARY KEY (order_id, is_exact, unit_key),
    KEY idx_order_offer (order_id, offer_id),
    KEY idx_model (model_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {LAMODA_UNITS}
  (order_id, offer_id, unit_key, is_exact, model_id, identity_matches, return_date,
   order_status, item_status, source_payload_id)
SELECT latest.order_id,
       latest.offer_id,
       latest.item_id,
       1 AS is_exact,
       CASE
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 0
         THEN idn.model_id
         WHEN COALESCE(dp.identity_matches, 0) = 1
          AND COALESCE(idn.identity_matches, 0) = 0
         THEN dp.model_id
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 1
          AND idn.model_id = dp.model_id
         THEN idn.model_id
       END AS model_id,
       CASE
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 0
         THEN 1
         WHEN COALESCE(dp.identity_matches, 0) = 1
          AND COALESCE(idn.identity_matches, 0) = 0
         THEN 1
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 1
          AND idn.model_id = dp.model_id
         THEN 1
         ELSE COALESCE(idn.identity_matches, 0) + COALESCE(dp.identity_matches, 0)
       END AS identity_matches,
       latest.return_date,
       latest.order_status,
       latest.item_status,
       latest.payload_id
FROM (
    SELECT normalized.*,
           ROW_NUMBER() OVER (
               PARTITION BY normalized.order_id, normalized.item_id
               ORDER BY normalized.return_date DESC,
                        normalized.payload_id DESC,
                        normalized.source_row_id DESC
           ) AS rn
    FROM (
        SELECT sd.id AS source_row_id,
               sd.payload_id,
               TRIM(sd.order_id) AS order_id,
               TRIM(sd.offer_id) AS offer_id,
               TRIM(sd.item_id) AS item_id,
               COALESCE(sd.return_date, sd.status_date) AS return_date,
               sd.status AS order_status,
               sd.item_status
        FROM {I}.lamoda_status_dates sd
        {LAMODA_EXACT_APPROVED}
        WHERE sd.status = 'Returned'
          AND sd.item_id IS NOT NULL
          AND TRIM(sd.item_id) <> ''
          AND sd.order_id IS NOT NULL
          AND TRIM(sd.order_id) <> ''
          AND sd.offer_id IS NOT NULL
          AND TRIM(sd.offer_id) <> ''
    ) normalized
) latest
LEFT JOIN (
    SELECT identifier_value,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product_identifier
    WHERE mp_id = 9
      AND identifier_type = 'seller_sku'
      AND is_current = 1
    GROUP BY identifier_value
) idn
  ON idn.identifier_value = latest.offer_id COLLATE utf8mb4_unicode_ci
LEFT JOIN (
    SELECT artikul_upper,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product
    GROUP BY artikul_upper
) dp
  ON dp.artikul_upper = UPPER(latest.offer_id) COLLATE utf8mb4_unicode_ci
WHERE latest.rn = 1
""",
        f"""
CREATE TEMPORARY TABLE {LAMODA_EXACT_KEYS} (
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    offer_id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    PRIMARY KEY (order_id, offer_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {LAMODA_EXACT_KEYS} (order_id, offer_id)
SELECT DISTINCT exact_item.order_id, exact_item.offer_id
FROM {LAMODA_UNITS} exact_item
WHERE exact_item.is_exact = 1
""",
        f"""
INSERT INTO {LAMODA_UNITS}
  (order_id, offer_id, unit_key, is_exact, model_id, identity_matches, return_date,
   order_status, item_status, source_payload_id)
SELECT latest.order_id,
       latest.offer_id,
       latest.offer_id AS unit_key,
       0 AS is_exact,
       CASE
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 0
         THEN idn.model_id
         WHEN COALESCE(dp.identity_matches, 0) = 1
          AND COALESCE(idn.identity_matches, 0) = 0
         THEN dp.model_id
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 1
          AND idn.model_id = dp.model_id
         THEN idn.model_id
       END AS model_id,
       CASE
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 0
         THEN 1
         WHEN COALESCE(dp.identity_matches, 0) = 1
          AND COALESCE(idn.identity_matches, 0) = 0
         THEN 1
         WHEN COALESCE(idn.identity_matches, 0) = 1
          AND COALESCE(dp.identity_matches, 0) = 1
          AND idn.model_id = dp.model_id
         THEN 1
         ELSE COALESCE(idn.identity_matches, 0) + COALESCE(dp.identity_matches, 0)
       END AS identity_matches,
       latest.return_date,
       latest.order_status,
       latest.item_status,
       latest.payload_id
FROM (
    SELECT normalized.*,
           ROW_NUMBER() OVER (
               PARTITION BY normalized.order_id, normalized.offer_id
               ORDER BY normalized.return_date DESC,
                        normalized.payload_id DESC,
                        normalized.source_row_id DESC
           ) AS rn
    FROM (
        SELECT sd.id AS source_row_id,
               sd.payload_id,
               TRIM(sd.order_id) AS order_id,
               TRIM(sd.offer_id) AS offer_id,
               COALESCE(sd.return_date, sd.status_date) AS return_date,
               sd.status AS order_status,
               sd.item_status
        FROM {I}.lamoda_status_dates sd
        {LAMODA_FALLBACK_APPROVED}
        WHERE sd.status = 'Returned'
          AND sd.item_id IS NULL
          AND sd.order_id IS NOT NULL
          AND TRIM(sd.order_id) <> ''
          AND sd.offer_id IS NOT NULL
          AND TRIM(sd.offer_id) <> ''
    ) normalized
) latest
LEFT JOIN {LAMODA_EXACT_KEYS} exact_item
  ON exact_item.order_id = latest.order_id
 AND exact_item.offer_id = latest.offer_id
LEFT JOIN (
    SELECT identifier_value,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product_identifier
    WHERE mp_id = 9
      AND identifier_type = 'seller_sku'
      AND is_current = 1
    GROUP BY identifier_value
) idn
  ON idn.identifier_value = latest.offer_id COLLATE utf8mb4_unicode_ci
LEFT JOIN (
    SELECT artikul_upper,
           COUNT(DISTINCT model_id) AS identity_matches,
           CASE WHEN COUNT(DISTINCT model_id) = 1 THEN MAX(model_id) END AS model_id
    FROM {K}.dim_product
    GROUP BY artikul_upper
) dp
  ON dp.artikul_upper = UPPER(latest.offer_id) COLLATE utf8mb4_unicode_ci
WHERE latest.rn = 1
  AND exact_item.order_id IS NULL
""",
        f"""
CREATE TEMPORARY TABLE {LAMODA_EXPECTED} (
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    model_id INT UNSIGNED NOT NULL,
    return_date DATE NULL,
    amount DECIMAL(14,2) NULL,
    quantity INT NOT NULL,
    reason VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    source_payload_id BIGINT UNSIGNED NOT NULL,
    order_matches INT UNSIGNED NOT NULL,
    order_revenue DECIMAL(14,2) NULL,
    order_quantity INT NULL,
    PRIMARY KEY (return_id, model_id),
    KEY idx_order (order_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {LAMODA_EXPECTED}
  (return_id, order_id, model_id, return_date, amount, quantity, reason, status,
   source_payload_id, order_matches, order_revenue, order_quantity)
SELECT projected_return.return_id,
       projected_return.order_id,
       projected_return.model_id,
       projected_return.return_date,
       ABS(CASE
         WHEN COALESCE(fact_order.order_matches, 0) = 1
          AND fact_order.revenue IS NOT NULL
          AND fact_order.quantity > 0
         THEN (fact_order.revenue / fact_order.quantity) * projected_return.quantity
       END) AS amount,
       projected_return.quantity,
       projected_return.reason,
       projected_return.status,
       projected_return.source_payload_id,
       COALESCE(fact_order.order_matches, 0),
       fact_order.revenue,
       fact_order.quantity
FROM (
    SELECT CONCAT(return_unit.order_id, ':', return_unit.offer_id) AS return_id,
           return_unit.order_id,
           return_unit.model_id,
           MAX(return_unit.return_date) AS return_date,
           COUNT(*) AS quantity,
           MAX(return_unit.order_status) AS reason,
           MAX(return_unit.order_status) AS status,
           MAX(return_unit.source_payload_id) AS source_payload_id
    FROM {LAMODA_UNITS} return_unit
    WHERE return_unit.identity_matches = 1
      AND return_unit.model_id IS NOT NULL
    GROUP BY return_unit.order_id, return_unit.offer_id, return_unit.model_id
) projected_return
LEFT JOIN (
    SELECT fo.order_id,
           fo.model_id,
           COUNT(*) AS order_matches,
           CASE WHEN COUNT(*) = 1 THEN MAX(fo.revenue) END AS revenue,
           CASE WHEN COUNT(*) = 1 THEN MAX(fo.quantity) END AS quantity
    FROM {K}.fact_orders fo
    WHERE fo.mp_id = 9
    GROUP BY fo.order_id, fo.model_id
) fact_order
  ON fact_order.order_id = projected_return.order_id COLLATE utf8mb4_unicode_ci
 AND fact_order.model_id = projected_return.model_id
""",
        f"""
CREATE TEMPORARY TABLE {LAMODA_SCOPE} (
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    PRIMARY KEY (return_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {LAMODA_SCOPE} (return_id)
SELECT DISTINCT expected_return.return_id
FROM {LAMODA_EXPECTED} expected_return
""",
        f"""
INSERT IGNORE INTO {LAMODA_SCOPE} (return_id)
SELECT DISTINCT CONCAT(TRIM(history.order_id), ':', TRIM(history.offer_id)) AS return_id
FROM {I}.lamoda_status_dates history
{LAMODA_HISTORY_APPROVED}
JOIN {LAMODA_UNITS} current_item
  ON current_item.is_exact = 1
 AND current_item.model_id IS NOT NULL
 AND current_item.order_id = TRIM(history.order_id) COLLATE utf8mb4_unicode_ci
 AND current_item.unit_key = TRIM(history.item_id) COLLATE utf8mb4_unicode_ci
WHERE history.status = 'Returned'
  AND history.item_id IS NOT NULL
  AND TRIM(history.item_id) <> ''
  AND history.offer_id IS NOT NULL
  AND TRIM(history.offer_id) <> ''
""",
    )
    # MySQL cannot reopen a TEMPORARY table more than once in one statement
    # (ERROR 1137): per-metric scalar subqueries would reopen {LAMODA_UNITS}
    # and {LAMODA_EXPECTED} several times.  Each surface is aggregated in one
    # single-pass derived read (always exactly one row) and the two rows are
    # combined with CROSS JOIN — one open per temporary table per statement.
    attest = f"""
SELECT units_agg.selected_units,
       units_agg.invalid_keys,
       units_agg.invalid_dates,
       units_agg.identity_failures,
       expected_agg.projected_returns,
       expected_agg.projected_units,
       expected_agg.order_failures
FROM (
    SELECT COUNT(*) AS selected_units,
           COALESCE(SUM(
              return_unit.order_id IS NULL OR return_unit.order_id = ''
              OR return_unit.offer_id IS NULL OR return_unit.offer_id = ''
              OR return_unit.unit_key IS NULL OR return_unit.unit_key = ''
              OR return_unit.source_payload_id IS NULL
            ), 0) AS invalid_keys,
           COALESCE(SUM(return_unit.return_date IS NULL), 0) AS invalid_dates,
           COALESCE(SUM(return_unit.identity_matches <> 1), 0)
             AS identity_failures
    FROM {LAMODA_UNITS} return_unit
) units_agg
CROSS JOIN (
    SELECT COUNT(*) AS projected_returns,
           COALESCE(SUM(expected_return.quantity), 0) AS projected_units,
           COALESCE(SUM(
              expected_return.order_matches <> 1
              OR expected_return.order_revenue IS NULL
              OR expected_return.order_quantity IS NULL
              OR expected_return.order_quantity <= 0
              OR expected_return.amount IS NULL
            ), 0) AS order_failures
    FROM {LAMODA_EXPECTED} expected_return
) expected_agg
"""
    retract_stale = f"""
DELETE fact_return
FROM {LAMODA_SCOPE} active_scope
JOIN {K}.fact_returns fact_return
  ON fact_return.mp_id = 9
 AND fact_return.return_id = active_scope.return_id COLLATE utf8mb4_unicode_ci
LEFT JOIN {LAMODA_EXPECTED} expected_return
  ON expected_return.return_id = fact_return.return_id
 AND expected_return.model_id = fact_return.model_id
WHERE fact_return.source_payload_id IS NOT NULL
  AND expected_return.model_id IS NULL
"""
    upsert = f"""
INSERT INTO {K}.fact_returns
  (model_id, mp_id, return_id, return_date, amount, quantity, reason, status,
   source_payload_id)
SELECT expected_return.model_id,
       9 AS mp_id,
       expected_return.return_id,
       expected_return.return_date,
       expected_return.amount,
       expected_return.quantity,
       expected_return.reason,
       expected_return.status,
       expected_return.source_payload_id
FROM {LAMODA_EXPECTED} expected_return
LEFT JOIN {K}.fact_returns legacy_return
  ON legacy_return.mp_id = 9
 AND legacy_return.return_id = expected_return.return_id COLLATE utf8mb4_unicode_ci
 AND legacy_return.model_id = expected_return.model_id
 AND legacy_return.source_payload_id IS NULL
WHERE legacy_return.id IS NULL
ON DUPLICATE KEY UPDATE
  return_date=VALUES(return_date),
  amount=VALUES(amount),
  quantity=VALUES(quantity),
  reason=VALUES(reason),
  status=VALUES(status),
  source_payload_id=VALUES(source_payload_id)
"""
    cleanup = (
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_SCOPE}",
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_EXPECTED}",
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_EXACT_KEYS}",
        f"DROP TEMPORARY TABLE IF EXISTS {LAMODA_UNITS}",
    )
    return prepare, attest, retract_stale, upsert, cleanup


(
    LAMODA_PREPARE_SQLS,
    LAMODA_ATTEST_SQL,
    LAMODA_STALE_CLEANUP_SQL,
    LAMODA_SQL,
    LAMODA_CLEANUP_SQLS,
) = _lamoda_returns_plan()

# YM: возвраты /v2/campaigns/{id}/returns → intake.ym_returns (mp 6 FBS / 8 FBY).
# Важно: return_id — заголовок возврата, а товарный состав живёт в
# return_json.items[]. Нельзя джойнить заголовок ко всем моделям заказа: это
# создаёт fan-out и завышает возвраты. План ниже разбирает items[], резолвит
# каждый shopSku и только потом матчит ту же модель в exact mp/order.
YM_CURRENT = "tmp_fact_returns_ym_current"
YM_ITEMS = "tmp_fact_returns_ym_items"
YM_EXPECTED = "tmp_fact_returns_ym_expected"
YM_ACKNOWLEDGED = "tmp_fact_returns_ym_acknowledged"


def _sql_literal(value: str) -> str:
    """Quote a reviewed ledger string; its values never come from runtime input."""
    return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'"


def _acknowledged_ym_values_sql() -> str:
    values = []
    for entry in ACKNOWLEDGED_YM_LOSS_ENTRIES:
        values.append(
            "(" + ", ".join(
                (
                    str(entry["mpId"]),
                    _sql_literal(str(entry["returnId"])),
                    _sql_literal(str(entry["orderId"])),
                    str(entry["lineNo"]),
                    _sql_literal(str(entry["shopSku"])),
                    str(entry["itemCount"]),
                    str(entry["modelId"]),
                )
            ) + ")"
        )
    return ",\n       ".join(values)


class ReturnsProjectionError(RuntimeError):
    """A selected return set cannot be projected without data loss."""


class WBReturnsProjectionError(ReturnsProjectionError):
    """WB return projection failed its semantic attestation."""


class OzonReturnsProjectionError(ReturnsProjectionError):
    """Ozon return projection failed its semantic attestation."""


class LamodaReturnsProjectionError(ReturnsProjectionError):
    """Lamoda return projection failed its semantic attestation."""


class YMReturnsProjectionError(ReturnsProjectionError):
    """The selected YM return set cannot be projected without data loss."""


YM_TYPED_ATTEST_SQL = f"""
SELECT COUNT(*) AS source_rows,
       SUM(CASE
             WHEN r.mp_id NOT IN (6, 8)
               OR r.return_id IS NULL OR TRIM(r.return_id) = ''
               OR r.order_id IS NULL OR TRIM(r.order_id) = ''
               OR r.return_date_mp IS NULL
               OR r.return_json IS NULL
               OR JSON_VALID(r.return_json) = 0
               OR JSON_TYPE(r.return_json) <> 'OBJECT'
               OR JSON_TYPE(JSON_EXTRACT(r.return_json, '$.items')) <> 'ARRAY'
               OR JSON_LENGTH(JSON_EXTRACT(r.return_json, '$.items')) <= 0
             THEN 1 ELSE 0
           END) AS invalid_rows
FROM {I}.ym_returns r
{YM_APPROVED}
"""


def _ym_returns_plan() -> tuple[tuple[str, ...], str, str, str, str, tuple[str, ...]]:
    """Build and attest the complete latest YM item set before fact DML.

    The approved input is first projected into session-local temporary tables.
    Every selected return must contain a non-empty ``items`` array; every array
    member must have one non-empty ``shopSku``, one positive integer ``count``,
    one exact product identity, and one exact order fact in the return's own
    marketplace.  The caller attests those invariants before *any* persistent
    ``fact_returns`` statement is allowed to run.
    """
    prepare = (
        f"DROP TEMPORARY TABLE IF EXISTS {YM_EXPECTED}",
        f"DROP TEMPORARY TABLE IF EXISTS {YM_ITEMS}",
        f"DROP TEMPORARY TABLE IF EXISTS {YM_CURRENT}",
        f"DROP TEMPORARY TABLE IF EXISTS {YM_ACKNOWLEDGED}",
        f"""
CREATE TEMPORARY TABLE {YM_CURRENT} (
    mp_id TINYINT UNSIGNED NOT NULL,
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    return_date DATE NULL,
    reason VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    source_payload_id BIGINT UNSIGNED NOT NULL,
    return_json JSON NULL,
    items_authoritative TINYINT(1) NOT NULL DEFAULT 0,
    PRIMARY KEY (mp_id, return_id),
    KEY idx_order (mp_id, order_id)
) ENGINE=InnoDB
""",
        f"""
CREATE TEMPORARY TABLE {YM_ACKNOWLEDGED} (
    mp_id TINYINT UNSIGNED NOT NULL,
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    line_no INT UNSIGNED NOT NULL,
    shop_sku VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    item_count INT UNSIGNED NOT NULL,
    model_id INT UNSIGNED NOT NULL,
    PRIMARY KEY (mp_id, return_id, line_no)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {YM_ACKNOWLEDGED}
  (mp_id, return_id, order_id, line_no, shop_sku, item_count, model_id)
VALUES {_acknowledged_ym_values_sql()}
""",
        f"""
INSERT INTO {YM_CURRENT}
  (mp_id, return_id, order_id, return_date, reason, status, source_payload_id,
   return_json, items_authoritative)
SELECT latest.mp_id,
       latest.return_id,
       latest.order_id,
       DATE(latest.return_date_mp),
       latest.reason_type,
       latest.status,
       latest.payload_id,
       latest.return_json,
       CASE
           WHEN latest.return_json IS NOT NULL
            AND JSON_TYPE(latest.return_json) = 'OBJECT'
            AND JSON_TYPE(JSON_EXTRACT(latest.return_json, '$.items')) = 'ARRAY'
           THEN 1 ELSE 0
       END
FROM (
    SELECT r.*,
           ROW_NUMBER() OVER (
               PARTITION BY r.mp_id, r.return_id
               ORDER BY r.collected_at DESC, r.payload_id DESC, r.id DESC
           ) AS rn
    FROM {I}.ym_returns r
    {YM_APPROVED}
    WHERE r.return_id IS NOT NULL
      AND TRIM(r.return_id) <> ''
      AND r.order_id IS NOT NULL
      AND TRIM(r.order_id) <> ''
) latest
WHERE latest.rn = 1
""",
        f"""
CREATE TEMPORARY TABLE {YM_ITEMS} (
    mp_id TINYINT UNSIGNED NOT NULL,
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    line_no INT UNSIGNED NOT NULL,
    order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    return_date DATE NULL,
    reason VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    source_payload_id BIGINT UNSIGNED NOT NULL,
    shop_sku VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    item_count INT UNSIGNED NULL,
    model_id INT UNSIGNED NULL,
    identity_matches INT UNSIGNED NOT NULL,
    order_matches INT UNSIGNED NOT NULL,
    order_revenue DECIMAL(14,2) NULL,
    order_quantity INT NULL,
    acknowledged_listed TINYINT(1) NOT NULL DEFAULT 0,
    PRIMARY KEY (mp_id, return_id, line_no),
    KEY idx_order (mp_id, order_id),
    KEY idx_model (model_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {YM_ITEMS}
  (mp_id, return_id, line_no, order_id, return_date, reason, status,
   source_payload_id, shop_sku, item_count, model_id, identity_matches,
   order_matches, order_revenue, order_quantity, acknowledged_listed)
SELECT item_scope.mp_id,
       item_scope.return_id,
       item_scope.line_no,
       item_scope.order_id,
       item_scope.return_date,
       item_scope.reason,
       item_scope.status,
       item_scope.source_payload_id,
       item_scope.shop_sku,
       CASE
         WHEN item_scope.item_count_text IS NOT NULL
          AND CHAR_LENGTH(item_scope.item_count_text) <= 9
          AND REGEXP_LIKE(item_scope.item_count_text, '^[1-9][0-9]*$', 'c')
         THEN CAST(item_scope.item_count_text AS UNSIGNED)
         ELSE NULL
       END AS item_count,
       CASE
         WHEN COUNT(DISTINCT item_scope.candidate_model_id) = 1
         THEN MAX(item_scope.candidate_model_id)
         ELSE NULL
       END AS model_id,
       COUNT(DISTINCT item_scope.candidate_model_id) AS identity_matches,
       COUNT(DISTINCT fact_order.id) AS order_matches,
       CASE WHEN COUNT(DISTINCT fact_order.id) = 1
            THEN MAX(fact_order.revenue) ELSE NULL END AS order_revenue,
       CASE WHEN COUNT(DISTINCT fact_order.id) = 1
            THEN MAX(fact_order.quantity) ELSE NULL END AS order_quantity,
       MAX(acknowledged.model_id IS NOT NULL) AS acknowledged_listed
FROM (
    SELECT current_return.mp_id,
           current_return.return_id,
           current_return.order_id,
           current_return.return_date,
           current_return.reason,
           current_return.status,
           current_return.source_payload_id,
           return_item.line_no,
           NULLIF(TRIM(return_item.shop_sku), '') AS shop_sku,
           return_item.item_count_text,
           CASE
             WHEN idn.model_id IS NOT NULL AND dp.model_id IS NOT NULL
              AND idn.model_id <> dp.model_id THEN NULL
             ELSE COALESCE(idn.model_id, dp.model_id)
           END AS candidate_model_id
    FROM {YM_CURRENT} current_return
    CROSS JOIN JSON_TABLE(
        CASE WHEN current_return.items_authoritative = 1
             THEN current_return.return_json ELSE JSON_OBJECT() END,
        -- COLLATE ОБЪЯВЛЯЕТСЯ ЗДЕСЬ, А НЕ В КАЖДОМ СРАВНЕНИИ (10.08.2026).
        --
        -- JSON_TABLE отдаёт значения в коллации СОЕДИНЕНИЯ (замер: без
        -- объявления — utf8mb4_0900_ai_ci, coercibility 2). Всё остальное ядро
        -- живёт в utf8mb4_unicode_ci, и любое сравнение с ним падает:
        --
        --   1267 Illegal mix of collations
        --        (utf8mb4_unicode_ci,IMPLICIT) and (utf8mb4_0900_ai_ci,IMPLICIT)
        --
        -- Раньше это лечили припиской COLLATE к КАЖДОМУ сравнению. Приписок
        -- вышло три, а сравнений — четыре: соединение с
        -- tmp_fact_returns_ym_acknowledged по shop_sku осталось без неё, и шаг
        -- fact_returns падал. Найдено 10.08 после того, как сняли барьеры в
        -- fact_finance и fact_orders и сборка впервые до него дошла.
        --
        -- Объявление в источнике снимает класс целиком: значение рождается уже
        -- в канонической коллации, и следующее сравнение не потребует помнить
        -- про приписку. Проверено на живой базе: с объявлением JOIN работает,
        -- без — падает.
        '$.items[*]' COLUMNS (
            line_no FOR ORDINALITY,
            shop_sku VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
                     PATH '$.shopSku' NULL ON EMPTY NULL ON ERROR,
            item_count_text VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
                     PATH '$.count' NULL ON EMPTY NULL ON ERROR
        )
    ) return_item
    LEFT JOIN {K}.dim_product_identifier idn
      ON idn.mp_id = 6
     AND idn.identifier_type = 'shopSku'
     AND idn.identifier_value = return_item.shop_sku COLLATE utf8mb4_unicode_ci
     AND idn.is_current = 1
    LEFT JOIN {K}.dim_product dp
      ON dp.artikul_upper = UPPER(TRIM(return_item.shop_sku)) COLLATE utf8mb4_unicode_ci
) item_scope
LEFT JOIN {YM_ACKNOWLEDGED} acknowledged
  ON acknowledged.mp_id = item_scope.mp_id
 AND acknowledged.return_id = item_scope.return_id
 AND acknowledged.order_id = item_scope.order_id
 AND acknowledged.line_no = item_scope.line_no
 AND acknowledged.shop_sku = item_scope.shop_sku
 AND acknowledged.item_count = CASE
       WHEN item_scope.item_count_text IS NOT NULL
        AND CHAR_LENGTH(item_scope.item_count_text) <= 9
        AND REGEXP_LIKE(item_scope.item_count_text, '^[1-9][0-9]*$', 'c')
       THEN CAST(item_scope.item_count_text AS UNSIGNED)
     END
 AND acknowledged.model_id = item_scope.candidate_model_id
LEFT JOIN {K}.fact_orders fact_order
  ON fact_order.mp_id = item_scope.mp_id
 AND fact_order.order_id = item_scope.order_id COLLATE utf8mb4_unicode_ci
 AND fact_order.model_id = item_scope.candidate_model_id
 AND fact_order.quantity > 0
 AND fact_order.revenue IS NOT NULL
GROUP BY item_scope.mp_id,
         item_scope.return_id,
         item_scope.line_no,
         item_scope.order_id,
         item_scope.return_date,
         item_scope.reason,
         item_scope.status,
         item_scope.source_payload_id,
         item_scope.shop_sku,
         item_scope.item_count_text
""",
        f"""
CREATE TEMPORARY TABLE {YM_EXPECTED} (
    mp_id TINYINT UNSIGNED NOT NULL,
    return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    model_id INT UNSIGNED NOT NULL,
    return_date DATE NULL,
    amount DECIMAL(14,2) NULL,
    quantity INT NOT NULL,
    reason VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    status VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
    source_payload_id BIGINT UNSIGNED NOT NULL,
    PRIMARY KEY (mp_id, return_id, model_id)
) ENGINE=InnoDB
""",
        f"""
INSERT INTO {YM_EXPECTED}
  (mp_id, return_id, model_id, return_date, amount, quantity, reason, status,
   source_payload_id)
SELECT current_return.mp_id,
       current_return.return_id,
       current_return.model_id,
       current_return.return_date,
       ABS(SUM(
           (current_return.order_revenue / current_return.order_quantity)
           * current_return.item_count
       )) AS amount,
       SUM(current_return.item_count) AS quantity,
       current_return.reason,
       current_return.status,
       current_return.source_payload_id
FROM {YM_ITEMS} current_return
WHERE current_return.shop_sku IS NOT NULL
  AND current_return.item_count IS NOT NULL
  AND current_return.identity_matches = 1
  AND current_return.order_matches = 1
GROUP BY current_return.mp_id,
         current_return.return_id,
         current_return.model_id,
         current_return.return_date,
         current_return.reason,
         current_return.status,
         current_return.source_payload_id
""",
        f"""
INSERT INTO {YM_EXPECTED}
  (mp_id, return_id, model_id, return_date, amount, quantity, reason, status,
   source_payload_id)
SELECT current_return.mp_id,
       current_return.return_id,
       current_return.model_id,
       current_return.return_date,
       NULL AS amount,
       SUM(current_return.item_count) AS quantity,
       current_return.reason,
       current_return.status,
       current_return.source_payload_id
FROM {YM_ITEMS} current_return
WHERE current_return.shop_sku IS NOT NULL
  AND current_return.item_count IS NOT NULL
  AND current_return.identity_matches = 1
  AND current_return.order_matches = 0
  AND current_return.acknowledged_listed = 1
GROUP BY current_return.mp_id,
         current_return.return_id,
         current_return.model_id,
         current_return.return_date,
         current_return.reason,
         current_return.status,
         current_return.source_payload_id
""",
    )
    # MySQL cannot reopen a TEMPORARY table more than once in one statement
    # (ERROR 1137).  The return-level metrics below open {YM_CURRENT} once and
    # {YM_ITEMS} once (per-return parsed counts); the item-level metrics need
    # their own whole-table pass over {YM_ITEMS} and therefore live in a
    # separate statement (YM_ITEMS_ATTEST_SQL).  SUM(boolean) reproduces the
    # old COUNT(*)-with-WHERE semantics term for term (NULL terms stay
    # uncounted in both forms).
    attest = f"""
SELECT COUNT(*) AS selected_returns,
       COALESCE(SUM(
           current_return.mp_id NOT IN (6, 8)
           OR current_return.items_authoritative <> 1
           OR JSON_LENGTH(JSON_EXTRACT(current_return.return_json, '$.items')) = 0
       ), 0) AS invalid_return_shapes,
       COALESCE(SUM(
           CASE WHEN current_return.items_authoritative = 1
                THEN JSON_LENGTH(JSON_EXTRACT(current_return.return_json, '$.items'))
                ELSE 0 END
       ), 0) AS expected_items,
       COALESCE(SUM(
           current_return.items_authoritative = 1
           AND COALESCE(parsed_count.parsed_count, 0)
               <> JSON_LENGTH(JSON_EXTRACT(current_return.return_json, '$.items'))
       ), 0) AS item_count_mismatches
FROM {YM_CURRENT} current_return
LEFT JOIN (
    SELECT parsed.mp_id, parsed.return_id, COUNT(*) AS parsed_count
    FROM {YM_ITEMS} parsed
    GROUP BY parsed.mp_id, parsed.return_id
) parsed_count
  ON parsed_count.mp_id = current_return.mp_id
 AND parsed_count.return_id = current_return.return_id
"""
    items_attest = f"""
SELECT COUNT(*) AS parsed_items,
       COALESCE(SUM(
           parsed.shop_sku IS NULL OR parsed.item_count IS NULL
       ), 0) AS invalid_items,
       COALESCE(SUM(parsed.identity_matches <> 1), 0) AS identity_failures,
       COALESCE(SUM(
           parsed.identity_matches = 1 AND parsed.order_matches <> 1
           AND parsed.acknowledged_listed = 0
       ), 0) AS order_failures,
       COALESCE(SUM(
           parsed.identity_matches = 1 AND parsed.order_matches = 0
           AND parsed.acknowledged_listed = 1
       ), 0) AS acknowledged_missing,
       COALESCE(SUM(
           parsed.shop_sku IS NOT NULL
           AND parsed.item_count IS NOT NULL
           AND parsed.identity_matches = 1
           AND parsed.order_matches = 1
       ), 0) AS projected_items
FROM {YM_ITEMS} parsed
"""
    retract_stale = f"""
DELETE fact_return
FROM {YM_CURRENT} current_return
JOIN {K}.fact_returns fact_return
  ON fact_return.mp_id = current_return.mp_id
 AND fact_return.return_id = current_return.return_id COLLATE utf8mb4_unicode_ci
LEFT JOIN {YM_EXPECTED} expected_return
  ON expected_return.mp_id = fact_return.mp_id
 AND expected_return.return_id = fact_return.return_id
 AND expected_return.model_id = fact_return.model_id
WHERE current_return.items_authoritative = 1
  AND fact_return.source_payload_id IS NOT NULL
  AND expected_return.model_id IS NULL
"""
    upsert = f"""
INSERT INTO {K}.fact_returns
  (model_id, mp_id, return_id, return_date, amount, quantity, reason, status,
   source_payload_id)
SELECT expected_return.model_id,
       expected_return.mp_id,
       expected_return.return_id,
       expected_return.return_date,
       expected_return.amount,
       expected_return.quantity,
       expected_return.reason,
       expected_return.status,
       expected_return.source_payload_id
FROM {YM_EXPECTED} expected_return
LEFT JOIN {K}.fact_returns legacy_return
  ON legacy_return.mp_id = expected_return.mp_id
 AND legacy_return.return_id = expected_return.return_id COLLATE utf8mb4_unicode_ci
 AND legacy_return.model_id = expected_return.model_id
 AND legacy_return.source_payload_id IS NULL
WHERE legacy_return.id IS NULL
ON DUPLICATE KEY UPDATE
  return_date=VALUES(return_date),
  amount=VALUES(amount),
  quantity=VALUES(quantity),
  reason=VALUES(reason),
  status=VALUES(status),
  source_payload_id=VALUES(source_payload_id)
"""
    cleanup = (
        f"DROP TEMPORARY TABLE IF EXISTS {YM_EXPECTED}",
        f"DROP TEMPORARY TABLE IF EXISTS {YM_ITEMS}",
        f"DROP TEMPORARY TABLE IF EXISTS {YM_CURRENT}",
        f"DROP TEMPORARY TABLE IF EXISTS {YM_ACKNOWLEDGED}",
    )
    return prepare, attest, items_attest, retract_stale, upsert, cleanup


(
    YM_PREPARE_SQLS,
    YM_ATTEST_SQL,
    YM_ITEMS_ATTEST_SQL,
    YM_STALE_CLEANUP_SQL,
    YM_SQL,
    YM_CLEANUP_SQLS,
) = _ym_returns_plan()


def _read_attestation_metrics(
    cursor: Any,
    sql: str,
    fields: tuple[str, ...],
    *,
    error_type: type[ReturnsProjectionError],
    label: str,
) -> dict[str, int]:
    cursor.execute(sql)
    row = cursor.fetchone()
    if not isinstance(row, Mapping):
        raise error_type(f"{label} projection attestation returned no row")
    try:
        return {field: int(row[field] or 0) for field in fields}
    except (KeyError, TypeError, ValueError) as exc:
        raise error_type(f"{label} projection attestation is incomplete") from exc


def _require_wb_projection(cursor: Any) -> dict[str, int]:
    fields = (
        "selected_rows",
        "invalid_keys",
        "invalid_dates",
        "invalid_amounts",
        "identity_failures",
        "natural_key_duplicates",
    )
    metrics = _read_attestation_metrics(
        cursor,
        WB_ATTEST_SQL,
        fields,
        error_type=WBReturnsProjectionError,
        label="WB returns",
    )
    if any(metrics[field] for field in fields[1:]):
        raise WBReturnsProjectionError(
            "WB returns projection is incomplete; persistent DML is forbidden: "
            f"{metrics}"
        )
    return metrics


def _ozon_cancellation_mode() -> str:
    """Resolve the owner-flippable Ozon Cancellation gate mode (fail closed).

    Unset/empty env -> ``exclude_validated`` (канон 2026-06-30).  An unknown
    value is a configuration defect and must abort the step, never silently
    fall through to a permissive path.
    """
    raw = os.getenv(OZON_CANCELLATION_MODE_ENV, "").strip()
    if not raw:
        return OZON_CANCELLATION_MODE_EXCLUDE
    if raw not in _OZON_CANCELLATION_MODES:
        raise OzonReturnsProjectionError(
            f"{OZON_CANCELLATION_MODE_ENV}={raw!r} is not one of "
            f"{sorted(_OZON_CANCELLATION_MODES)}; persistent DML is forbidden"
        )
    return raw


def _require_ozon_projection(cursor: Any) -> dict[str, int]:
    mode = _ozon_cancellation_mode()
    fields = (
        "selected_rows",
        "invalid_keys",
        "invalid_types",
        "invalid_statuses",
        "invalid_dates",
        "invalid_prices",
        "invalid_quantities",
        "identity_failures",
        "natural_key_duplicates",
        "cancellation_rows",
        "non_event_rows",
    )
    # cancellation_rows — наблюдаемость, НЕ повод для падения: валидированный
    # невыкуп исключается из fact DML самими type-фильтрами upsert/cleanup.
    failure_fields = fields[1:-2]
    attest_sql = (
        OZON_ATTEST_SQL
        if mode == OZON_CANCELLATION_MODE_EXCLUDE
        else OZON_ATTEST_SQL_FAIL_CLOSED
    )
    metrics = _read_attestation_metrics(
        cursor,
        attest_sql,
        fields,
        error_type=OzonReturnsProjectionError,
        label="Ozon returns",
    )
    if any(metrics[field] for field in failure_fields):
        raise OzonReturnsProjectionError(
            "Ozon returns projection is incomplete; persistent DML is forbidden: "
            f"{metrics}"
        )
    return metrics


def _require_lamoda_typed_input(cursor: Any) -> None:
    metrics = _read_attestation_metrics(
        cursor,
        LAMODA_TYPED_ATTEST_SQL,
        ("return_rows", "invalid_rows"),
        error_type=LamodaReturnsProjectionError,
        label="Lamoda returns typed input",
    )
    if metrics["invalid_rows"] != 0:
        raise LamodaReturnsProjectionError(
            "Lamoda returns typed input is malformed; persistent DML is forbidden: "
            f"{metrics}"
        )


def _require_lamoda_projection(cursor: Any) -> dict[str, int]:
    fields = (
        "selected_units",
        "invalid_keys",
        "invalid_dates",
        "identity_failures",
        "projected_returns",
        "projected_units",
        "order_failures",
    )
    metrics = _read_attestation_metrics(
        cursor,
        LAMODA_ATTEST_SQL,
        fields,
        error_type=LamodaReturnsProjectionError,
        label="Lamoda returns",
    )
    if (
        any(
            metrics[field]
            for field in (
                "invalid_keys",
                "invalid_dates",
                "identity_failures",
                "order_failures",
            )
        )
        or metrics["selected_units"] != metrics["projected_units"]
    ):
        raise LamodaReturnsProjectionError(
            "Lamoda returns projection is incomplete; persistent DML is forbidden: "
            f"{metrics}"
        )
    return metrics


def _require_ym_typed_input(cursor: Any) -> None:
    """Prove every approved typed row is a complete return before projection."""
    cursor.execute(YM_TYPED_ATTEST_SQL)
    row = cursor.fetchone() or {}
    try:
        source_rows = int(row["source_rows"])
        invalid_rows = int(row["invalid_rows"] or 0)
    except (KeyError, TypeError, ValueError) as exc:
        raise YMReturnsProjectionError(
            "YM returns typed-input attestation is incomplete"
        ) from exc
    if source_rows <= 0 or invalid_rows != 0:
        raise YMReturnsProjectionError(
            "YM returns typed input is empty or malformed; persistent DML is "
            f"forbidden: source_rows={source_rows}, invalid_rows={invalid_rows}"
        )


def _require_ym_projection(cursor: Any) -> dict[str, int]:
    """Fail closed unless the selected YM set has a lossless projection."""
    cursor.execute(YM_ATTEST_SQL)
    row = cursor.fetchone()
    if not isinstance(row, Mapping):
        raise YMReturnsProjectionError("YM returns projection attestation returned no row")
    cursor.execute(YM_ITEMS_ATTEST_SQL)
    items_row = cursor.fetchone()
    if not isinstance(items_row, Mapping):
        raise YMReturnsProjectionError(
            "YM returns item attestation returned no row"
        )

    return_fields = (
        "selected_returns",
        "invalid_return_shapes",
        "expected_items",
        "item_count_mismatches",
    )
    item_fields = (
        "parsed_items",
        "invalid_items",
        "identity_failures",
        "order_failures",
        "acknowledged_missing",
        "projected_items",
    )
    try:
        metrics = {field: int(row[field]) for field in return_fields}
        metrics.update({field: int(items_row[field]) for field in item_fields})
    except (KeyError, TypeError, ValueError) as exc:
        raise YMReturnsProjectionError(
            "YM returns projection attestation is incomplete"
        ) from exc

    failures = {
        key: metrics[key]
        for key in (
            "invalid_return_shapes",
            "item_count_mismatches",
            "invalid_items",
            "identity_failures",
            "order_failures",
        )
        if metrics[key] != 0
    }
    if (
        failures
        or metrics["expected_items"] != metrics["parsed_items"]
        or metrics["parsed_items"]
        != metrics["projected_items"] + metrics["acknowledged_missing"]
    ):
        raise YMReturnsProjectionError(
            "YM returns projection is incomplete; persistent DML is forbidden: "
            f"{metrics}"
        )
    return metrics


def build() -> None:
    with get_cursor() as cur:
        require_build_input_projection(cur)

        # Every source is materialized and semantically attested before the
        # first persistent fact statement.  Ozon/Lamoda/YM can retract facts,
        # so a malformed source must abort the entire step, not leave WB or a
        # preceding marketplace partially published.
        _require_lamoda_typed_input(cur)
        _require_ym_typed_input(cur)

        for statement in WB_PREPARE_SQLS:
            cur.execute(statement)
        for statement in OZON_PREPARE_SQLS:
            cur.execute(statement)
        for statement in LAMODA_PREPARE_SQLS:
            cur.execute(statement)
        for statement in YM_PREPARE_SQLS:
            cur.execute(statement)

        _require_wb_projection(cur)
        ozon_metrics = _require_ozon_projection(cur)
        _require_lamoda_projection(cur)
        _require_ym_projection(cur)

        cur.execute(f"SELECT COUNT(*) c FROM {K}.fact_returns")
        before = cur.fetchone()["c"]
        cur.execute(WB_SQL)
        for statement in WB_CLEANUP_SQLS:
            cur.execute(statement)
        cur.execute(f"SELECT COUNT(*) c FROM {K}.fact_returns")
        after_wb = cur.fetchone()["c"]

        cur.execute(OZON_NON_CLIENT_CLEANUP_SQL)
        ozon_removed = cur.rowcount
        cur.execute(OZON_SQL)
        for statement in OZON_CLEANUP_SQLS:
            cur.execute(statement)
        cur.execute(f"SELECT COUNT(*) c FROM {K}.fact_returns")
        after_ozon = cur.fetchone()["c"]

        cur.execute(LAMODA_STALE_CLEANUP_SQL)
        lamoda_removed = cur.rowcount
        cur.execute(LAMODA_SQL)
        for statement in LAMODA_CLEANUP_SQLS:
            cur.execute(statement)
        cur.execute(f"SELECT COUNT(*) c FROM {K}.fact_returns")
        after_lamoda = cur.fetchone()["c"]

        cur.execute(YM_STALE_CLEANUP_SQL)
        ym_removed = cur.rowcount
        cur.execute(YM_SQL)
        for statement in YM_CLEANUP_SQLS:
            cur.execute(statement)
        cur.execute(f"SELECT COUNT(*) c FROM {K}.fact_returns")
        after_ym = cur.fetchone()["c"]
        # Аларм качества данных (epic-05/07): return_date IS NULL НЕ вычитается из net
        # (accrual по дате возврата; BETWEEN исключает NULL). Считаем и сигналим.
        # F-44 (DEFECT-LEDGER, решение владельца 2026-07-03): legacy-backfill строки
        # (признак source_payload_id IS NULL — сверено live: ВСЕ NULL-даты на 2026-07-03,
        # включая 863 YM, имеют source_payload_id IS NULL) СПИСАНЫ официально — вне accrual,
        # не чинить, из аларма исключены. Аларм — ТОЛЬКО на НОВЫЕ NULL-даты
        # (source_payload_id IS NOT NULL): это реальные проблемы сбора даты.
        cur.execute(
            f"SELECT COALESCE(SUM(source_payload_id IS NOT NULL), 0) c_new, "
            f"COALESCE(SUM(source_payload_id IS NULL), 0) c_legacy "
            f"FROM {K}.fact_returns WHERE return_date IS NULL"
        )
        row = cur.fetchone()
        null_dates_new, null_dates_legacy = int(row["c_new"]), int(row["c_legacy"])
    logger.info(
        "fact_returns: rows before=%d after_wb=%d (+%d) "
        "ozon_non_client_removed=%d after_ozon=%d (%+d) after_lamoda=%d (%+d) "
        "lamoda_stale_removed=%d ym_stale_removed=%d after_ym=%d (%+d)",
        before,
        after_wb,
        after_wb - before,
        ozon_removed,
        after_ozon,
        after_ozon - after_wb,
        after_lamoda,
        after_lamoda - after_ozon,
        lamoda_removed,
        ym_removed,
        after_ym,
        after_ym - after_lamoda,
    )
    logger.info(
        "fact_returns: ozon cancellation gate mode=%s env=%s cancellation_rows=%d "
        "non_event_rows=%d (Cancellation=невыкуп; F-63 status=несостоявшийся "
        "возврат; оба не fact_returns)",
        _ozon_cancellation_mode(),
        OZON_CANCELLATION_MODE_ENV,
        ozon_metrics["cancellation_rows"],
        ozon_metrics["non_event_rows"],
    )
    if null_dates_legacy:
        logger.info(
            "fact_returns: %d legacy-backfill строк с return_date IS NULL "
            "(source_payload_id IS NULL) — вне accrual, из них 863 legacy-YM. "
            "F-44 ПЕРЕОТКРЫТ 2026-08-03: прежнее «списано, не чинить» было "
            "неточным — тогда не знали, КАК чинить, и отложили. Чинить надо: "
            "суммы этих возвратов не вычитаются из выручки. Путь — разовый "
            "добор YM-возвратов за исторический период; внутренним JOIN не "
            "восстановить, легаси-идентификатор NNN_NNN не совпадает ни с одним "
            "заказом в fact_orders.",
            null_dates_legacy,
        )
    if null_dates_new:
        logger.warning(
            "fact_returns: %d НОВЫХ строк с return_date IS NULL (source_payload_id "
            "IS NOT NULL) — НЕ вычитаются из net (accrual по дате возврата). "
            "Реальная проблема сбора даты — проверь источники возвратов (F-44).",
            null_dates_new,
        )


def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    try:
        ensure_kernel_writer_fences()
        build()
        return 0
    except Exception:
        logger.exception("fact_returns ETL failed")
        return 1


if __name__ == "__main__":
    sys.exit(main())