"""Modal dialog used to unload one or more UDCs from a multi-occupancy cell."""

from __future__ import annotations

import json
import logging
import sys
import tkinter as tk
from dataclasses import dataclass
from datetime import datetime
from functools import wraps
from pathlib import Path
from typing import Any, Callable

import customtkinter as ctk
from tkinter import messagebox, ttk

from gestione_aree import AsyncRunner
from audit_log import log_user_action
from busy_overlay import InlineBusyOverlay
from locale_text import load_locale_catalog, text as loc_text
from ui_theme import theme_color, theme_font, theme_section, theme_value
from ui_tables import style_treeview, zebra_tag
from user_session import UserSession
from version_info import module_version, versioned_title

__version__ = module_version(__name__)

try:
    from loguru import logger
except Exception:  # pragma: no cover - fallback used only when Loguru is not available
    class _FallbackLogger:
        """Minimal adapter used only when Loguru is not installed yet."""

        def __init__(self):
            self._logger = logging.getLogger(MODULE_LOG_NAME if "MODULE_LOG_NAME" in globals() else __name__)
            self._logger.setLevel(logging.DEBUG)
            self._logger.propagate = False

        def bind(self, **_kwargs):
            return self

        def add(self, sink, level="INFO", format=None, encoding="utf-8", **_kwargs):
            handler: logging.Handler
            if hasattr(sink, "write"):
                handler = logging.StreamHandler(sink)
            else:
                handler = logging.FileHandler(str(sink), encoding=encoding)
            handler.setLevel(getattr(logging, str(level).upper(), logging.INFO))
            handler.setFormatter(
                logging.Formatter("%(asctime)s | %(levelname)-8s | %(name)s | %(message)s")
            )
            self._logger.addHandler(handler)
            return 0

        def log(self, level, message):
            getattr(self._logger, str(level).lower(), self._logger.info)(message)

        def debug(self, message):
            self._logger.debug(message)

        def info(self, message):
            self._logger.info(message)

        def exception(self, message):
            self._logger.exception(message)

    logger = _FallbackLogger()


SCARICO_LOG_MODE = "INFO"  # "OFF" | "INFO" | "DEBUG"
MODULE_LOG_NAME = Path(__file__).stem
MODULE_LOG_PATH = Path(__file__).with_suffix(".log")
_MODULE_LOG_ENABLED = SCARICO_LOG_MODE.upper() != "OFF"
_MODULE_LOG_LEVEL = "DEBUG" if SCARICO_LOG_MODE.upper() == "DEBUG" else "INFO"
_MODULE_LOGGER = logger.bind(warehouse_module=MODULE_LOG_NAME)
_MODULE_LOGGING_CONFIGURED = False
DEFAULT_SCARICO_USER = "warehouse_ui"
SHIPPED_IDCELLA = 9999
SHIPPED_BARCODE = "9000000"
SHIPPED_LABEL = "7G.1.1"
NON_SCAFF_IDCELLA = 1000
NON_SCAFF_BARCODE = "9001000"
NON_SCAFF_LABEL = "5E1.1"


def _session_login(session: UserSession | None, fallback: str | None = None) -> str:
    """Return the current application login, falling back to a technical user."""

    if session and str(session.login or "").strip():
        return str(session.login).strip()
    return str((fallback or DEFAULT_SCARICO_USER) or DEFAULT_SCARICO_USER).strip()


def _configure_module_logger():
    """Configure console and file logging for this module."""
    global _MODULE_LOGGING_CONFIGURED
    if _MODULE_LOGGING_CONFIGURED:
        return
    if not _MODULE_LOG_ENABLED:
        _MODULE_LOGGING_CONFIGURED = True
        return

    record_filter = lambda record: record["extra"].get("warehouse_module") == MODULE_LOG_NAME
    logger.add(
        sys.stderr,
        level=_MODULE_LOG_LEVEL,
        colorize=True,
        filter=record_filter,
        format=(
            "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
            "<level>{level: <8}</level> | "
            "<cyan>" + MODULE_LOG_NAME + "</cyan> | "
            "<level>{message}</level>"
        ),
    )
    logger.add(
        MODULE_LOG_PATH,
        level=_MODULE_LOG_LEVEL,
        colorize=False,
        encoding="utf-8",
        filter=record_filter,
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | " + MODULE_LOG_NAME + " | {message}",
    )
    _MODULE_LOGGING_CONFIGURED = True


def _format_payload(payload: Any) -> str:
    """Serialize payloads for human-readable logging."""
    try:
        return json.dumps(payload, ensure_ascii=False, indent=2, default=str)
    except Exception:
        return repr(payload)


def _log_call(level: str | None = None):
    """Trace entry, exit and failure of selected high-level functions."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            effective_level = level or _MODULE_LOG_LEVEL
            _MODULE_LOGGER.log(
                effective_level,
                f"CALL {func.__qualname__} args={_format_payload(args[1:] if args else ())} kwargs={_format_payload(kwargs)}",
            )
            try:
                result = func(*args, **kwargs)
            except Exception:
                _MODULE_LOGGER.exception(f"FAIL {func.__qualname__}")
                raise
            _MODULE_LOGGER.log(effective_level, f"RETURN {func.__qualname__}")
            return result

        return wrapper

    return decorator


def _log_sql(query_name: str, sql: str, params: dict[str, Any]):
    """Log one SQL statement and its parameters."""
    _MODULE_LOGGER.log(_MODULE_LOG_LEVEL, f"SQL {query_name} params={_format_payload(params)}")
    _MODULE_LOGGER.debug(f"SQL {query_name} text:\n{sql.strip()}")


def _log_dataset(query_name: str, rows: list[Any]):
    """Log query results at summary or full-debug level depending on the flag."""
    _MODULE_LOGGER.log(_MODULE_LOG_LEVEL, f"SQL {query_name} returned {len(rows)} rows")
    if SCARICO_LOG_MODE.upper() == "DEBUG":
        _MODULE_LOGGER.debug(f"SQL {query_name} dataset:\n{_format_payload(rows)}")


_configure_module_logger()
if _MODULE_LOG_ENABLED:
    _MODULE_LOGGER.info(
        f"Logging inizializzato su {MODULE_LOG_PATH.name} livello={_MODULE_LOG_LEVEL} mode={SCARICO_LOG_MODE.upper()}"
    )


SQL_UDC_IN_CELLA = """
WITH cell_pallets AS (
    SELECT DISTINCT
        g.BarcodePallet
    FROM dbo.XMag_GiacenzaPallet g
    WHERE g.IDCella = :idcella
),
last_in_cell AS (
    SELECT
        mp.Attributo AS BarcodePallet,
        MAX(mp.ID) AS LastID
    FROM dbo.MagazziniPallet mp
    JOIN cell_pallets cp
      ON cp.BarcodePallet COLLATE Latin1_General_CI_AS =
         mp.Attributo COLLATE Latin1_General_CI_AS
    WHERE mp.IDCella = :idcella
      AND mp.Tipo = 'V'
    GROUP BY mp.Attributo
),
last_move AS (
    SELECT
        mp.Attributo AS BarcodePallet,
        mp.ID,
        mp.DataMagazzino
    FROM dbo.MagazziniPallet mp
    JOIN last_in_cell lic ON lic.LastID = mp.ID
),
closed_plist AS (
    SELECT DISTINCT
        shipped.BarcodePallet
    FROM dbo.XMag_GiacenzaPalletPlistChiuse shipped
    JOIN cell_pallets cp
      ON cp.BarcodePallet COLLATE Latin1_General_CI_AS =
         shipped.BarcodePallet COLLATE Latin1_General_CI_AS
)
SELECT
    cp.BarcodePallet AS UDC,
    lm.ID AS SourceID,
    lm.DataMagazzino AS LastEventAt,
    :idcella AS CurrentIDCella,
    CASE
        WHEN :idcella <> 9999
         AND closed_plist.BarcodePallet IS NOT NULL
        THEN CAST(1 AS int)
        ELSE CAST(0 AS int)
    END AS IsMissingUnload
FROM cell_pallets cp
LEFT JOIN last_move lm
  ON lm.BarcodePallet COLLATE Latin1_General_CI_AS =
     cp.BarcodePallet COLLATE Latin1_General_CI_AS
LEFT JOIN closed_plist
  ON closed_plist.BarcodePallet COLLATE Latin1_General_CI_AS =
     cp.BarcodePallet COLLATE Latin1_General_CI_AS
ORDER BY
    lm.ID DESC,
    cp.BarcodePallet DESC;
"""


SQL_SCARICA_UDC = """
SET NOCOUNT ON;

DECLARE @Now datetime = GETDATE();
DECLARE @SourceID int = 0;
DECLARE @NumeroPallet int = 0;
DECLARE @PesoUnitario float = 1;
DECLARE @Tara float = 0;
DECLARE @SourceIDCella int = 0;

SELECT TOP (1)
    @SourceID = src.ID,
    @NumeroPallet = ISNULL(src.NumeroPallet, 0),
    @PesoUnitario = ISNULL(NULLIF(src.PesoUnitario, 0), 1),
    @Tara = ISNULL(src.Tara, 0),
    @SourceIDCella = ISNULL(src.IDCella, 0)
FROM dbo.MagazziniPallet src
WHERE src.Attributo = :barcode_pallet
  AND src.Tipo = 'V'
  AND src.PesoUnitario > 0
ORDER BY src.ID DESC;

IF @SourceID > 0
BEGIN
    UPDATE dbo.MagazziniPallet
       SET ModUtente = :utente,
           ModDataOra = @Now
     WHERE ID = @SourceID;

    INSERT INTO dbo.MagazziniPallet (
        Tipo,
        IDRiferimento,
        NumeroPallet,
        Attributo,
        IDMagazzino,
        IDArea,
        IDCella,
        DataMagazzino,
        PesoUnitario,
        Tara,
        InsUtente,
        InsDataOra
    )
    SELECT
        'P',
        @SourceID,
        ISNULL(src.NumeroPallet, 0),
        src.Attributo,
        src.IDMagazzino,
        src.IDArea,
        src.IDCella,
        @Now,
        ISNULL(NULLIF(src.PesoUnitario, 0), 1),
        ISNULL(src.Tara, 0),
        :utente,
        @Now
    FROM dbo.MagazziniPallet src
    WHERE src.ID = @SourceID;

    UPDATE dbo.Celle
       SET IDStato = 0,
           ModUtente = :utente,
           ModDataOra = @Now
     WHERE ID = @SourceIDCella;
END;

INSERT INTO dbo.MagazziniPallet (
    Tipo,
    IDRiferimento,
    NumeroPallet,
    Attributo,
    IDMagazzino,
    IDArea,
    IDCella,
    DataMagazzino,
    PesoUnitario,
    Tara,
    InsUtente,
    InsDataOra
)
SELECT
    'V',
    @SourceID,
    @NumeroPallet,
    :barcode_pallet,
    target.IDMagazzino,
    target.IDArea,
    target.ID,
    @Now,
    @PesoUnitario,
    @Tara,
    :utente,
    @Now
FROM (
    SELECT c.ID, c.IDArea, a.IDMagazzino
    FROM dbo.Celle c
    JOIN dbo.Aree a ON a.ID = c.IDArea
    WHERE c.ID = :target_idcella
) AS target;

EXEC dbo.py_sp_ControllaPrenotazionePackingListPalletNew;

SELECT
    CAST(1 AS int) AS Ok,
    @SourceID AS SourceID,
    :target_idcella AS TargetIDCella,
    :target_barcode_cella AS TargetBarcodeCella;
"""


async def move_pallet_async(
    db_client,
    *,
    barcode_pallet: str,
    target_idcella: int,
    target_barcode_cella: str,
    utente: str | None = None,
) -> dict[str, Any]:
    """Move one pallet to a target cell using the same movement semantics as the legacy app.

    The original C# application delegates load, transfer and unload to
    ``sp_xMagGestioneMagazziniPallet``. The Python app mirrors the same
    behavior with an explicit SQL batch that:
    1. finds the latest positive row for the pallet,
    2. registers a compensating ``P`` move on the source cell,
    3. frees the previous cell reservation,
    4. inserts a new ``V`` move on the target cell,
    5. re-runs the packing-list reservation check.
    """
    params = {
        "barcode_pallet": str(barcode_pallet or "").strip(),
        "target_idcella": int(target_idcella),
        "target_barcode_cella": str(target_barcode_cella or "").strip(),
        "utente": str((utente or DEFAULT_SCARICO_USER) or "warehouse_ui").strip(),
    }
    _log_sql("move_pallet", SQL_SCARICA_UDC, params)
    res = await db_client.query_json(SQL_SCARICA_UDC, params, commit=True)
    rows = res.get("rows", []) if isinstance(res, dict) else []
    _log_dataset("move_pallet", rows)
    first = rows[0] if rows else [1, 0, params["target_idcella"], params["target_barcode_cella"]]
    return {
        "ok": int(first[0] or 0),
        "source_id": int(first[1] or 0),
        "target_idcella": int(first[2] or 0),
        "target_barcode_cella": str(first[3] or ""),
        "barcode_pallet": params["barcode_pallet"],
    }


@dataclass
class ScaricoRow:
    """View-model describing one unloadable UDC currently present in a cell."""

    udc: str
    source_id: int | None
    last_event_at: str
    current_idcella: int
    diagnostic_note: str
    selected: tk.BooleanVar


def _build_diagnostic_note(is_missing_unload: int | bool = False) -> str:
    """Translate low-level anomaly flags into one operator-facing note."""

    if bool(is_missing_unload):
        return "Motivo: spedita ma non scaricata dalla cella"
    return ""


def _is_missing_unload_row(row: ScaricoRow) -> bool:
    """Return True when the row must be corrected only with shipped unload."""

    return "spedita ma non scaricata dalla cella" in str(row.diagnostic_note or "").lower()


class ScaricoDialog(ctk.CTkToplevel):
    """Modal dialog that allows unloading selected UDCs from one cell."""

    CHECKBOX_COL_W = 56
    UDC_COL_W = 130
    DATE_COL_W = 180
    DIAG_COL_W = 320

    @_log_call()
    def __init__(
        self,
        parent: tk.Misc,
        *,
        db_client,
        idcella: int,
        ubicazione: str,
        on_completed: Callable[[], None] | None = None,
        session: UserSession | None = None,
    ):
        super().__init__(parent)
        self.parent = parent
        self.db_client = db_client
        self.idcella = idcella
        self.ubicazione = ubicazione
        self.on_completed = on_completed
        self.session = session
        self.rows: list[ScaricoRow] = []
        self._theme = theme_section("scarico_dialog", {})
        self._locale_catalog = load_locale_catalog()
        self._busy = InlineBusyOverlay(self, self._theme)
        self._async = AsyncRunner(self)
        self.rows_tree: ttk.Treeview | None = None

        self.title(versioned_title(loc_text("scarico.title", catalog=self._locale_catalog, default="Scarica {ubicazione}").format(ubicazione=ubicazione), __name__))
        self.resizable(False, False)
        self.transient(parent)
        self.protocol("WM_DELETE_WINDOW", self._close)

        self.grid_columnconfigure(0, weight=1)
        self.grid_rowconfigure(1, weight=1)

        self._build_ui()
        self._load_rows()

        self.update_idletasks()
        self.grab_set()
        try:
            self.wait_visibility()
        except Exception:
            pass
        if self.winfo_exists():
            self.lift()
            self.focus_force()

    def _build_ui(self):
        """Build the compact modal layout."""
        top = ctk.CTkFrame(self)
        top.grid(row=0, column=0, sticky="ew", padx=10, pady=(10, 6))
        top.grid_columnconfigure(0, weight=1)
        ctk.CTkLabel(
            top,
            text=loc_text("scarico.label.location", catalog=self._locale_catalog, default="Ubicazione: {ubicazione}").format(ubicazione=self.ubicazione),
            font=theme_font(self._theme, "header_font", ("Segoe UI", 13, "bold")),
        ).grid(
            row=0, column=0, sticky="w"
        )
        ctk.CTkLabel(
            top,
            text=loc_text("scarico.label.select", catalog=self._locale_catalog, default="Seleziona le UDC da scaricare"),
            anchor="w",
            font=theme_font(self._theme, "body_font", ("Segoe UI", 10)),
        ).grid(
            row=1, column=0, sticky="w", pady=(4, 0)
        )

        table = ctk.CTkFrame(self)
        table.grid(row=1, column=0, sticky="nsew", padx=10, pady=(0, 8))
        table.grid_rowconfigure(0, weight=1)
        table.grid_columnconfigure(0, weight=1)

        tree_host = tk.Frame(table, bd=0, highlightthickness=0, background="#DBDBDB")
        tree_host.grid(row=0, column=0, sticky="nsew", padx=8, pady=(8, 8))
        tree_host.grid_rowconfigure(0, weight=1)
        tree_host.grid_columnconfigure(0, weight=1)

        self.rows_tree = ttk.Treeview(
            tree_host,
            columns=("sel", "udc", "last", "diag"),
            show="headings",
            style="Scarico.Treeview",
            selectmode="none",
        )
        style_treeview(
            self.rows_tree,
            style_name="Scarico.Treeview",
            rowheight=28,
            font=theme_font(self._theme, "tree_body_font", ("Segoe UI", 10)),
            heading_font=theme_font(self._theme, "tree_heading_font", ("Segoe UI", 10, "bold")),
        )
        self.rows_tree.heading("sel", text="Sel")
        self.rows_tree.heading("udc", text=loc_text("scarico.col.udc", catalog=self._locale_catalog, default="UDC"))
        self.rows_tree.heading("last", text=loc_text("scarico.col.last_insert", catalog=self._locale_catalog, default="Ultimo inserimento"))
        self.rows_tree.heading("diag", text=loc_text("scarico.col.diagnostic", catalog=self._locale_catalog, default="Diagnostica"))
        self.rows_tree.column("sel", width=self.CHECKBOX_COL_W, stretch=False, anchor="center")
        self.rows_tree.column("udc", width=self.UDC_COL_W, stretch=False, anchor="w")
        self.rows_tree.column("last", width=self.DATE_COL_W, stretch=False, anchor="w")
        self.rows_tree.column("diag", width=self.DIAG_COL_W, stretch=True, anchor="w")
        self.rows_tree.grid(row=0, column=0, sticky="nsew")
        self.rows_tree.bind("<Button-1>", self._on_tree_click, add="+")

        tree_scroll = ttk.Scrollbar(tree_host, orient="vertical", command=self.rows_tree.yview)
        tree_scroll.grid(row=0, column=1, sticky="ns")
        self.rows_tree.configure(yscrollcommand=tree_scroll.set)

        actions = ctk.CTkFrame(self)
        actions.grid(row=2, column=0, sticky="ew", padx=10, pady=(0, 10))
        actions.grid_columnconfigure(0, weight=1)
        ctk.CTkButton(
            actions,
            text=loc_text("scarico.button.shipped", catalog=self._locale_catalog, default="Scarico come spedita"),
            command=self._on_scarica_spedita,
            font=theme_font(self._theme, "button_font", ("Segoe UI", 10, "bold")),
        ).grid(
            row=0, column=1, padx=(8, 0), pady=8
        )
        ctk.CTkButton(
            actions,
            text=loc_text("scarico.button.non_shelved", catalog=self._locale_catalog, default="Scarico come non scaff."),
            command=self._on_scarica_non_scaff,
            font=theme_font(self._theme, "button_font", ("Segoe UI", 10, "bold")),
        ).grid(
            row=0, column=2, padx=(8, 0), pady=8
        )
        ctk.CTkButton(
            actions,
            text=loc_text("scarico.button.close", catalog=self._locale_catalog, default="Chiudi"),
            command=self._close,
            font=theme_font(self._theme, "button_font", ("Segoe UI", 10, "bold")),
        ).grid(
            row=0, column=3, padx=(8, 8), pady=8
        )

    def _render_rows(self):
        """Recreate the compact list of unloadable UDC rows."""
        if self.rows_tree is None:
            return
        for item in self.rows_tree.get_children():
            self.rows_tree.delete(item)

        for idx, row in enumerate(self.rows):
            self.rows_tree.insert(
                "",
                "end",
                iid=str(idx),
                values=(
                    "[x]" if row.selected.get() else "[ ]",
                    row.udc,
                    row.last_event_at,
                    row.diagnostic_note or "",
                ),
                tags=(zebra_tag(idx),),
            )

        self.update_idletasks()
        width = max(900, min(1040, self.winfo_reqwidth() + 20))
        total_height = max(210, min(460, self.winfo_reqheight() + 8))
        self.geometry(f"{width}x{total_height}")

    def _on_tree_click(self, event):
        """Toggle the pseudo-checkbox when the operator clicks the first column."""

        if self.rows_tree is None:
            return
        region = self.rows_tree.identify("region", event.x, event.y)
        if region != "cell":
            return
        column = self.rows_tree.identify_column(event.x)
        item_id = self.rows_tree.identify_row(event.y)
        if column != "#1" or not item_id:
            return
        try:
            row = self.rows[int(item_id)]
        except Exception:
            return
        row.selected.set(not row.selected.get())
        self.rows_tree.set(item_id, "sel", "[x]" if row.selected.get() else "[ ]")
        return "break"

    @_log_call()
    def _load_rows(self):
        """Load the list of current UDCs ordered from newest to oldest."""
        params = {"idcella": self.idcella}
        _log_sql("scarico_load_rows", SQL_UDC_IN_CELLA, params)

        def _ok(res):
            rows = res.get("rows", []) if isinstance(res, dict) else []
            _log_dataset("scarico_load_rows", rows)
            self.rows = []
            for (
                udc,
                source_id,
                last_event_at,
                current_idcella,
                is_missing_unload,
            ) in rows:
                if isinstance(last_event_at, datetime):
                    last_event = last_event_at.strftime("%d/%m/%Y %H:%M:%S")
                else:
                    last_event = str(last_event_at or "")
                self.rows.append(
                    ScaricoRow(
                        udc=str(udc or ""),
                        source_id=int(source_id) if source_id is not None else None,
                        last_event_at=last_event,
                        current_idcella=int(current_idcella or 0),
                        diagnostic_note=_build_diagnostic_note(is_missing_unload),
                        selected=tk.BooleanVar(value=False),
                    )
                )
            self._render_rows()
            self._busy.hide()

        def _err(ex):
            self._busy.hide()
            _MODULE_LOGGER.exception(f"Errore caricamento righe scarico idcella={self.idcella}: {ex}")
            messagebox.showerror(
                loc_text("scarico.msg.title", catalog=self._locale_catalog, default="Scarica"),
                loc_text("scarico.msg.load_error", catalog=self._locale_catalog, default="Caricamento UDC fallito:\n{error}").format(error=ex),
                parent=self,
            )
            self._close()

        self._async.run(
            self.db_client.query_json(SQL_UDC_IN_CELLA, params),
            _ok,
            _err,
            busy=self._busy,
            message="Carico UDC...",
        )

    @_log_call()
    def _selected_rows_or_warn(self) -> list[ScaricoRow] | None:
        """Return selected rows or show the standard selection warning."""

        selected = [row for row in self.rows if row.selected.get()]
        if not selected:
            messagebox.showinfo(
                loc_text("scarico.msg.title", catalog=self._locale_catalog, default="Scarica"),
                loc_text("scarico.msg.select_one", catalog=self._locale_catalog, default="Seleziona almeno una UDC da scaricare."),
                parent=self,
            )
        return selected

    @_log_call()
    def _on_scarica_spedita(self):
        """Unload selected UDCs to the shipped conventional location."""

        self._scarica_selected_to_target(
            target_idcella=SHIPPED_IDCELLA,
            target_barcode_cella=SHIPPED_BARCODE,
            target_label=SHIPPED_LABEL,
            title="Scarico come spedita",
            confirm_text="Scaricare come spedite {count} UDC da {ubicazione}?",
            busy_message="Scarico UDC come spedite...",
            block_shipped_for_non_scaff=False,
        )

    @_log_call()
    def _on_scarica_non_scaff(self):
        """Unload selected UDCs to the non-shelved conventional location."""

        self._scarica_selected_to_target(
            target_idcella=NON_SCAFF_IDCELLA,
            target_barcode_cella=NON_SCAFF_BARCODE,
            target_label=NON_SCAFF_LABEL,
            title="Scarico come non scaff.",
            confirm_text="Scaricare come non scaffalate {count} UDC da {ubicazione}?",
            busy_message="Scarico UDC come non scaffalate...",
            block_shipped_for_non_scaff=True,
        )

    def _scarica_selected_to_target(
        self,
        *,
        target_idcella: int,
        target_barcode_cella: str,
        target_label: str,
        title: str,
        confirm_text: str,
        busy_message: str,
        block_shipped_for_non_scaff: bool,
    ):
        """Move selected UDCs to one conventional unload location."""

        selected = self._selected_rows_or_warn()
        if not selected:
            return

        shipped_rows: list[ScaricoRow] = []
        missing_unload_rows: list[ScaricoRow] = []
        movable_rows = list(selected)
        if block_shipped_for_non_scaff:
            shipped_rows = [
                row for row in selected
                if int(row.current_idcella or 0) == SHIPPED_IDCELLA
            ]
            missing_unload_rows = [row for row in selected if _is_missing_unload_row(row)]
            blocked_udcs = {row.udc for row in shipped_rows + missing_unload_rows}
            movable_rows = [row for row in selected if row.udc not in blocked_udcs]

        if (shipped_rows or missing_unload_rows) and not movable_rows:
            message_lines = ["Nessuna UDC movimentata."]
            if shipped_rows:
                message_lines.append(f"UDC gia' in {SHIPPED_LABEL}: " + ", ".join(row.udc for row in shipped_rows))
            if missing_unload_rows:
                message_lines.append(
                    "UDC spedite ma non scaricate: "
                    + ", ".join(row.udc for row in missing_unload_rows)
                )
                message_lines.append("Usare 'Scarico come spedita'.")
            messagebox.showwarning(
                title,
                "\n".join(message_lines),
                parent=self,
            )
            return

        if not messagebox.askyesno(
            title,
            confirm_text.format(count=len(movable_rows), ubicazione=self.ubicazione),
            parent=self,
        ):
            return

        async def _job():
            results: list[dict[str, Any]] = []
            for row in movable_rows:
                result = await move_pallet_async(
                    self.db_client,
                    barcode_pallet=row.udc,
                    target_idcella=target_idcella,
                    target_barcode_cella=target_barcode_cella,
                    utente=_session_login(self.session),
                )
                results.append({"udc": row.udc, "affected": int(result.get("ok") or 0)})
            return results

        def _ok(results):
            _log_dataset("scarica_udc", results)
            done = [item["udc"] for item in results if int(item.get("affected") or 0) > 0]
            skipped = [item["udc"] for item in results if int(item.get("affected") or 0) <= 0]
            if not done:
                messagebox.showwarning(
                    title,
                    "Nessuna UDC e' stata scaricata. Verifica che le unita' siano ancora presenti in cella.",
                    parent=self,
                )
                return
            if skipped:
                shipped_note = (
                    f"\nGia' in {SHIPPED_LABEL} non toccate: " + ", ".join(row.udc for row in shipped_rows)
                    if shipped_rows
                    else ""
                )
                missing_unload_note = (
                    "\nSpedite ma non scaricate, usare 'Scarico come spedita': "
                    + ", ".join(row.udc for row in missing_unload_rows)
                    if missing_unload_rows
                    else ""
                )
                target_note = (
                    f"\nUDC in {target_label}: " + ", ".join(done)
                    if done
                    else ""
                )
                messagebox.showwarning(
                    title,
                    "Scarico parziale.\nCompletate: "
                    + ", ".join(done)
                    + "\nNon scaricate: "
                    + ", ".join(skipped)
                    + shipped_note
                    + missing_unload_note
                    + target_note,
                    parent=self,
                )
            else:
                msg = f"Scarico completato verso {target_label} per:\n" + "\n".join(done)
                if shipped_rows:
                    msg += f"\n\nUDC gia' in {SHIPPED_LABEL} non toccate:\n" + "\n".join(row.udc for row in shipped_rows)
                    msg += f"\n\nUDC in {target_label}: " + ", ".join(done)
                if missing_unload_rows:
                    msg += (
                        "\n\nUDC spedite ma non scaricate non toccate:\n"
                        + "\n".join(row.udc for row in missing_unload_rows)
                        + "\nUsare 'Scarico come spedita'."
                    )
                messagebox.showinfo(title, msg, parent=self)
            log_user_action(
                self.session,
                module=MODULE_LOG_NAME,
                action="layout.scarico.spedita" if int(target_idcella) == SHIPPED_IDCELLA else "layout.scarico.non_scaff",
                outcome="ok",
                target=self.ubicazione,
                details={
                    "scaricate": done,
                    "saltate": skipped,
                    "target_idcella": target_idcella,
                    "target_barcode_cella": target_barcode_cella,
                    "gia_spedite": [row.udc for row in shipped_rows],
                    "spedite_non_scaricate": [row.udc for row in missing_unload_rows],
                },
            )
            if self.on_completed:
                self.on_completed()
            self._close()

        def _err(ex):
            _MODULE_LOGGER.exception(f"Errore scarico idcella={self.idcella}: {ex}")
            log_user_action(
                self.session,
                module=MODULE_LOG_NAME,
                action="layout.scarico.spedita" if int(target_idcella) == SHIPPED_IDCELLA else "layout.scarico.non_scaff",
                outcome="error",
                target=self.ubicazione,
                details={"error": str(ex)},
            )
            messagebox.showerror(
                loc_text("scarico.msg.title", catalog=self._locale_catalog, default="Scarica"),
                loc_text("scarico.msg.exec_error", catalog=self._locale_catalog, default="Scarico fallito:\n{error}").format(error=ex),
                parent=self,
            )

        self._async.run(
            _job(),
            _ok,
            _err,
            busy=self._busy,
            message=busy_message,
        )

    @_log_call()
    def _close(self):
        """Release the modal grab and close the dialog safely."""
        try:
            self.grab_release()
        except Exception:
            pass
        try:
            self.destroy()
        except Exception:
            pass


@_log_call()
def open_scarico_dialog(
    parent: tk.Misc,
    *,
    db_client,
    idcella: int,
    ubicazione: str,
    on_completed: Callable[[], None] | None = None,
    session: UserSession | None = None,
) -> ScaricoDialog:
    """Create and return the modal unload dialog for one multi-UDC cell."""
    return ScaricoDialog(
        parent,
        db_client=db_client,
        idcella=idcella,
        ubicazione=ubicazione,
        on_completed=on_completed,
        session=session,
    )