"""Picking list management window. The module presents a master/detail UI for packing lists, supports reservation and unreservation through an async stored-procedure port and keeps rendering smooth by relying on deferred updates and lightweight progress indicators. """ from __future__ import annotations import tkinter as tk import customtkinter as ctk from tkinter import messagebox from typing import Optional, Any, Dict, List, Callable from dataclasses import dataclass # Usa overlay e runner "collaudati" from gestione_aree_frame_async import BusyOverlay, AsyncRunner # === IMPORT procedura async prenota/s-prenota (no pyodbc qui) === import asyncio try: from prenota_sprenota_sql import sp_xExePackingListPallet_async, SPResult except Exception: async def sp_xExePackingListPallet_async(*args, **kwargs): raise RuntimeError("sp_xExePackingListPallet_async non importabile: verifica prenota_sprenota_sql.py") class SPResult: def __init__(self, rc=-1, message="Procedura non disponibile", id_result=None): self.rc = rc; self.message = message; self.id_result = id_result # -------------------- SQL -------------------- SQL_PL = """ SELECT COUNT(DISTINCT Pallet) AS Pallet, COUNT(DISTINCT Lotto) AS Lotto, COUNT(DISTINCT Articolo) AS Articolo, COUNT(DISTINCT Descrizione) AS Descrizione, SUM(Qta) AS Qta, Documento, CodNazione, NAZIONE, Stato, MAX(PalletCella) AS PalletCella, MAX(Magazzino) AS Magazzino, MAX(Area) AS Area, MAX(Cella) AS Cella, MIN(Ordinamento) AS Ordinamento, MAX(IDStato) AS IDStato FROM dbo.XMag_ViewPackingList GROUP BY Documento, CodNazione, NAZIONE, Stato ORDER BY MIN(Ordinamento), Documento, NAZIONE, Stato; """ SQL_PL_DETAILS = """ SELECT * FROM ViewPackingListRestante WHERE Documento = :Documento ORDER BY Ordinamento; """ # -------------------- helpers -------------------- def _rows_to_dicts(res: Dict[str, Any]) -> List[Dict[str, Any]]: """ Converte il payload ritornato da query_json in lista di dict. Supporta: - res = [ {..}, {..} ] - res = { "rows": [..], "columns": [...] } - res = { "data": [..], "columns": [...] } - res = { "rows": [tuple,..], "columns": [...] } """ if res is None: return [] if isinstance(res, list): if not res: return [] if isinstance(res[0], dict): return res return [] if isinstance(res, dict): for rows_key in ("rows", "data", "result", "records"): if rows_key in res and isinstance(res[rows_key], list): rows = res[rows_key] if not rows: return [] if isinstance(rows[0], dict): return rows cols = res.get("columns") or res.get("cols") or [] out = [] for r in rows: if cols and isinstance(r, (list, tuple)): out.append({ (cols[i] if i < len(cols) else f"c{i}") : r[i] for i in range(min(len(cols), len(r))) }) else: if isinstance(r, (list, tuple)): out.append({ f"c{i}": r[i] for i in range(len(r)) }) return out if res and all(not isinstance(v, (list, tuple, dict)) for v in res.values()): return [res] return [] def _s(v) -> str: """Return a string representation, converting ``None`` to an empty string.""" return "" if v is None else str(v) def _first(d: Dict[str, Any], keys: List[str], default: str = ""): """Return the first non-empty value found among the provided keys.""" for k in keys: if k in d and d[k] not in (None, ""): return d[k] return default # -------------------- column specs -------------------- @dataclass class ColSpec: """Describe one logical column rendered in a ``ScrollTable``.""" title: str key: str width: int anchor: str # 'w' | 'e' | 'center' # Colonne PL (in alto) — include IDStato per la colorazione PL_COLS: List[ColSpec] = [ ColSpec("", "__check__", 36, "w"), ColSpec("Documento", "Documento", 120, "w"), ColSpec("NAZIONE", "NAZIONE", 240, "w"), ColSpec("Stato", "Stato", 110, "w"), ColSpec("IDStato", "IDStato", 80, "e"), # nuova colonna ColSpec("#Pallet", "Pallet", 100, "e"), ColSpec("#Lotti", "Lotto", 100, "e"), ColSpec("#Articoli", "Articolo", 110, "e"), ColSpec("Qta", "Qta", 120, "e"), ] DET_COLS: List[ColSpec] = [ ColSpec("UDC/Pallet", "Pallet", 150, "w"), ColSpec("Lotto", "Lotto", 130, "w"), ColSpec("Articolo", "Articolo", 150, "w"), ColSpec("Descrizione","Descrizione", 320, "w"), ColSpec("Qta", "Qta", 110, "e"), ColSpec("Ubicazione", "Ubicazione", 320, "w"), ] ROW_H = 28 # -------------------- Micro spinner (toolbar) -------------------- class ToolbarSpinner: """ Micro-animazione leggerissima per indicare attività: mostra una label con frame: ◐ ◓ ◑ ◒ ... finché è attivo. """ FRAMES = ("◐", "◓", "◑", "◒") def __init__(self, parent: tk.Widget): """Create the spinner label attached to the given parent widget.""" self.parent = parent self.lbl = ctk.CTkLabel(parent, text="", width=28) self._i = 0 self._active = False self._job = None def widget(self) -> ctk.CTkLabel: """Return the label widget hosting the spinner animation.""" return self.lbl def start(self, text: str = ""): """Start the animation and optionally show a short status message.""" if self._active: return self._active = True self.lbl.configure(text=f"{self.FRAMES[self._i]} {text}".strip()) self._tick() def stop(self): """Stop the animation and clear the label text.""" self._active = False if self._job is not None: try: self.parent.after_cancel(self._job) except Exception: pass self._job = None self.lbl.configure(text="") def _tick(self): """Advance the spinner animation frame.""" if not self._active: return self._i = (self._i + 1) % len(self.FRAMES) current = self.lbl.cget("text") # Mantieni l'eventuale testo dopo il simbolo txt_suffix = "" if isinstance(current, str) and len(current) > 2: txt_suffix = current[2:] self.lbl.configure(text=f"{self.FRAMES[self._i]}{txt_suffix}") self._job = self.parent.after(120, self._tick) # 8 fps soft # -------------------- Scrollable table -------------------- class ScrollTable(ctk.CTkFrame): GRID_COLOR = "#D0D5DD" PADX_L = 8 PADX_R = 8 PADY = 2 def __init__(self, master, columns: List[ColSpec]): """Create a fixed-header scrollable table rendered with Tk/CTk widgets.""" super().__init__(master) self.columns = columns self.total_w = sum(c.width for c in self.columns) self.grid_rowconfigure(1, weight=1) self.grid_columnconfigure(0, weight=1) # header self.h_canvas = tk.Canvas(self, height=ROW_H, highlightthickness=0, bd=0) self.h_inner = ctk.CTkFrame(self.h_canvas, fg_color="#f3f3f3", height=ROW_H, width=self.total_w) self.h_canvas.create_window((0,0), window=self.h_inner, anchor="nw", width=self.total_w, height=ROW_H) self.h_canvas.grid(row=0, column=0, sticky="ew") # body self.b_canvas = tk.Canvas(self, highlightthickness=0, bd=0) self.b_inner = ctk.CTkFrame(self.b_canvas, fg_color="transparent", width=self.total_w) self.body_window = self.b_canvas.create_window((0,0), window=self.b_inner, anchor="nw", width=self.total_w) self.b_canvas.grid(row=1, column=0, sticky="nsew") # scrollbars self.vbar = tk.Scrollbar(self, orient="vertical", command=self.b_canvas.yview) self.xbar = tk.Scrollbar(self, orient="horizontal", command=self._xscroll_both) self.vbar.grid(row=1, column=1, sticky="ns") self.xbar.grid(row=2, column=0, sticky="ew") # link scroll self.b_canvas.configure(yscrollcommand=self.vbar.set, xscrollcommand=self._xscroll_set_both) self.h_canvas.configure(xscrollcommand=self.xbar.set) # bind self.h_inner.bind("", lambda e: self._sync_header_width()) self.b_inner.bind("", lambda e: self._on_body_configure()) self._build_header() def _build_header(self): """Build the static header row using the configured columns.""" for w in self.h_inner.winfo_children(): w.destroy() row = ctk.CTkFrame(self.h_inner, fg_color="#f3f3f3", height=ROW_H, width=self.total_w) row.pack(fill="x", expand=False) row.pack_propagate(False) for col in self.columns: holder = ctk.CTkFrame( row, fg_color="#f3f3f3", width=col.width, height=ROW_H, border_width=1, border_color=self.GRID_COLOR ) holder.pack(side="left", fill="y") holder.pack_propagate(False) lbl = ctk.CTkLabel(holder, text=col.title, anchor="w") lbl.pack(fill="both", padx=(self.PADX_L, self.PADX_R), pady=self.PADY) self.h_inner.configure(width=self.total_w, height=ROW_H) self.h_canvas.configure(scrollregion=(0,0,self.total_w,ROW_H)) def _update_body_width(self): """Keep the scroll region aligned with the current body content width.""" self.b_canvas.itemconfigure(self.body_window, width=self.total_w) sr = self.b_canvas.bbox("all") if sr: self.b_canvas.configure(scrollregion=(0,0,max(self.total_w, sr[2]), sr[3])) else: self.b_canvas.configure(scrollregion=(0,0,self.total_w,0)) def _on_body_configure(self): """React to body resize events by syncing dimensions and header scroll.""" self._update_body_width() self._sync_header_width() def _sync_header_width(self): """Mirror the body horizontal scroll position on the header canvas.""" first, _ = self.b_canvas.xview() self.h_canvas.xview_moveto(first) def _xscroll_both(self, *args): """Scroll header and body together when the horizontal bar moves.""" self.h_canvas.xview(*args) self.b_canvas.xview(*args) def _xscroll_set_both(self, first, last): """Update the header viewport and scrollbar thumb in one place.""" self.h_canvas.xview_moveto(first) self.xbar.set(first, last) def clear_rows(self): """Remove all rendered body rows.""" for w in self.b_inner.winfo_children(): w.destroy() self._update_body_width() def add_row( self, values: List[str], row_index: int, anchors: Optional[List[str]] = None, checkbox_builder: Optional[Callable[[tk.Widget], ctk.CTkCheckBox]] = None, ): """Append one row to the table body.""" row = ctk.CTkFrame(self.b_inner, fg_color="transparent", height=ROW_H, width=self.total_w) row.pack(fill="x", expand=False) row.pack_propagate(False) for i, col in enumerate(self.columns): holder = ctk.CTkFrame( row, fg_color="transparent", width=col.width, height=ROW_H, border_width=1, border_color=self.GRID_COLOR ) holder.pack(side="left", fill="y") holder.pack_propagate(False) if col.key == "__check__": if checkbox_builder: cb = checkbox_builder(holder) cb.pack(padx=(self.PADX_L, self.PADX_R), pady=self.PADY, anchor="w") else: ctk.CTkLabel(holder, text="").pack(fill="both") else: anchor = (anchors[i] if anchors else col.anchor) ctk.CTkLabel(holder, text=values[i], anchor=anchor).pack( fill="both", padx=(self.PADX_L, self.PADX_R), pady=self.PADY ) self._update_body_width() # -------------------- PL row model -------------------- class PLRow: """State holder for one picking list row and its selection checkbox.""" def __init__(self, pl: Dict[str, Any], on_check): """Bind a picking list payload to a ``BooleanVar`` and callback.""" self.pl = pl self.var = ctk.BooleanVar(value=False) self._callback = on_check def is_checked(self) -> bool: """Return whether the row is currently selected.""" return self.var.get() def set_checked(self, val: bool): """Programmatically update the checkbox state.""" self.var.set(val) def build_checkbox(self, parent) -> ctk.CTkCheckBox: """Create the checkbox widget bound to this row model.""" return ctk.CTkCheckBox(parent, text="", variable=self.var, command=lambda: self._callback(self, self.var.get())) # -------------------- main frame (no-flicker + UX tuning + spinner) -------------------- class GestionePickingListFrame(ctk.CTkFrame): def __init__(self, master, *, db_client=None, conn_str=None): """Create the master/detail picking list frame.""" super().__init__(master) if db_client is None: raise ValueError("GestionePickingListFrame richiede un db_client condiviso.") self.db_client = db_client self.runner = AsyncRunner(self) # runner condiviso (usa loop globale) self.busy = BusyOverlay(self) # overlay collaudato self.rows_models: list[PLRow] = [] self._detail_cache: Dict[Any, list] = {} self.detail_doc = None self._first_loading: bool = False # flag per cursore d'attesa solo al primo load self._build_layout() # 🔇 Niente reload immediato: carichiamo quando la finestra è idle (= già resa) self.after_idle(self._first_show) def _first_show(self): """Chiamato a finestra già resa → evitiamo sfarfallio del primo paint e mostriamo wait-cursor.""" self._first_loading = True try: self.winfo_toplevel().configure(cursor="watch") except Exception: pass # spinner inizia self.spinner.start(" Carico…") self.reload_from_db(first=True) # ---------- UI ---------- def _build_layout(self): """Build toolbar, master table and detail table.""" for r in (1, 3): self.grid_rowconfigure(r, weight=1) self.grid_columnconfigure(0, weight=1) top = ctk.CTkFrame(self) top.grid(row=0, column=0, sticky="ew", padx=10, pady=(8,4)) for i, (text, cmd) in enumerate([ ("Ricarica", self.reload_from_db), ("Prenota", self.on_prenota), ("S-prenota", self.on_sprenota), ("Esporta XLSX", self.on_export) ]): ctk.CTkButton(top, text=text, command=cmd).grid(row=0, column=i, padx=6) # --- micro spinner a destra della toolbar --- self.spinner = ToolbarSpinner(top) self.spinner.widget().grid(row=0, column=10, padx=(8,0)) # largo spazio a destra self.pl_table = ScrollTable(self, PL_COLS) self.pl_table.grid(row=1, column=0, sticky="nsew", padx=10, pady=(4,8)) self.det_table = ScrollTable(self, DET_COLS) self.det_table.grid(row=3, column=0, sticky="nsew", padx=10, pady=(4,10)) self._draw_details_hint() def _draw_details_hint(self): """Render the placeholder row shown when no document is selected.""" self.det_table.clear_rows() self.det_table.add_row( values=["", "", "", "Seleziona una Picking List per vedere le UDC…", "", ""], row_index=0, anchors=["w"]*6 ) def _apply_row_colors(self, rows: List[Dict[str, Any]]): """Colorazione differita (after_idle) per evitare micro-jank durante l'inserimento righe.""" try: for idx, d in enumerate(rows): row_widget = self.pl_table.b_inner.winfo_children()[idx] if int(d.get("IDStato") or 0) == 1: row_widget.configure(fg_color="#ffe6f2") # rosa tenue else: row_widget.configure(fg_color="transparent") except Exception: pass def _refresh_mid_rows(self, rows: List[Dict[str, Any]]): """Rebuild the master table using the latest query results.""" self.pl_table.clear_rows() self.rows_models.clear() for r, d in enumerate(rows): model = PLRow(d, self.on_row_checked) self.rows_models.append(model) values = [ "", # checkbox _s(d.get("Documento")), _s(d.get("NAZIONE")), _s(d.get("Stato")), _s(d.get("IDStato")), # nuova colonna visibile _s(d.get("Pallet")), _s(d.get("Lotto")), _s(d.get("Articolo")), _s(d.get("Qta")), ] self.pl_table.add_row( values=values, row_index=r, anchors=[c.anchor for c in PL_COLS], checkbox_builder=model.build_checkbox ) # 🎯 Colora dopo che la UI è resa → no balzi visivi self.after_idle(lambda: self._apply_row_colors(rows)) # ----- helpers ----- def _get_selected_model(self) -> Optional[PLRow]: """Return the currently checked picking list row, if any.""" for m in self.rows_models: if m.is_checked(): return m return None def _recolor_row_by_documento(self, documento: str, idstato: int): """Aggiorna colore riga e cella IDStato per il Documento indicato.""" for idx, m in enumerate(self.rows_models): if _s(m.pl.get("Documento")) == _s(documento): m.pl["IDStato"] = idstato def _paint(): try: row_widget = self.pl_table.b_inner.winfo_children()[idx] row_widget.configure(fg_color="#ffe6f2" if idstato == 1 else "transparent") row_children = row_widget.winfo_children() if len(row_children) >= 5: holder = row_children[4] if holder.winfo_children(): lbl = holder.winfo_children()[0] if hasattr(lbl, "configure"): lbl.configure(text=str(idstato)) except Exception: pass # differisci la colorazione (smooth) self.after_idle(_paint) break def _reselect_documento_after_reload(self, documento: str): """(Opzionale) Dopo un reload DB, riseleziona la PL con lo stesso Documento.""" for m in self.rows_models: if _s(m.pl.get("Documento")) == _s(documento): m.set_checked(True) self.on_row_checked(m, True) break # ----- eventi ----- def on_row_checked(self, model: PLRow, is_checked: bool): """Handle row selection changes and refresh the detail section.""" # selezione esclusiva if is_checked: for m in self.rows_models: if m is not model and m.is_checked(): m.set_checked(False) self.detail_doc = model.pl.get("Documento") self.spinner.start(" Carico dettagli…") # spinner ON async def _job(): return await self.db_client.query_json(SQL_PL_DETAILS, {"Documento": self.detail_doc}) def _ok(res): self.spinner.stop() # spinner OFF self._detail_cache[self.detail_doc] = _rows_to_dicts(res) # differisci il render dei dettagli (più fluido) self.after_idle(self._refresh_details) def _err(ex): self.spinner.stop() messagebox.showerror("DB", f"Errore nel caricamento dettagli:\n{ex}") self.runner.run( _job(), on_success=_ok, on_error=_err, busy=self.busy, message=f"Carico UDC per Documento {self.detail_doc}…" ) else: if not any(m.is_checked() for m in self.rows_models): self.detail_doc = None self._refresh_details() # ----- load PL ----- def reload_from_db(self, first: bool = False): """Load or reload the picking list summary table from the database.""" self.spinner.start(" Carico…") # spinner ON async def _job(): return await self.db_client.query_json(SQL_PL, {}) def _on_success(res): rows = _rows_to_dicts(res) self._refresh_mid_rows(rows) self.spinner.stop() # spinner OFF # se era il primo load, ripristina il cursore standard if self._first_loading: try: self.winfo_toplevel().configure(cursor="") except Exception: pass self._first_loading = False def _on_error(ex): self.spinner.stop() if self._first_loading: try: self.winfo_toplevel().configure(cursor="") except Exception: pass self._first_loading = False messagebox.showerror("DB", f"Errore nel caricamento:\n{ex}") self.runner.run( _job(), on_success=_on_success, on_error=_on_error, busy=self.busy, message="Caricamento Picking List…" if first else "Aggiornamento…" ) def _refresh_details(self): """Render the detail table for the currently selected document.""" self.det_table.clear_rows() if not self.detail_doc: self._draw_details_hint() return rows = self._detail_cache.get(self.detail_doc, []) if not rows: self.det_table.add_row(values=["", "", "", "Nessuna UDC trovata.", "", ""], row_index=0, anchors=["w"]*6) return for r, d in enumerate(rows): pallet = _s(_first(d, ["Pallet", "UDC", "PalletID"])) lotto = _s(_first(d, ["Lotto"])) articolo = _s(_first(d, ["Articolo", "CodArticolo", "CodiceArticolo", "Art", "Codice"])) descr = _s(_first(d, ["Descrizione", "Descr", "DescrArticolo", "DescArticolo", "DesArticolo"])) qta = _s(_first(d, ["Qta", "Quantita", "Qty", "QTY"])) ubi_raw = _first(d, ["Ubicazione", "Cella", "PalletCella"]) loc = "Non scaffalata" if (ubi_raw is None or str(ubi_raw).strip()=="") else str(ubi_raw).strip() self.det_table.add_row( values=[pallet, lotto, articolo, descr, qta, loc], row_index=r, anchors=[c.anchor for c in DET_COLS] ) # ----- azioni ----- def on_prenota(self): """Reserve the selected picking list.""" model = self._get_selected_model() if not model: messagebox.showinfo("Prenota", "Seleziona una Picking List (checkbox) prima di prenotare.") return documento = _s(model.pl.get("Documento")) current = int(model.pl.get("IDStato") or 0) desired = 1 if current == desired: messagebox.showinfo("Prenota", f"La Picking List {documento} è già prenotata.") return id_operatore = 1 # TODO: recupera dal contesto reale self.spinner.start(" Prenoto…") async def _job(): return await sp_xExePackingListPallet_async(self.db_client, id_operatore, documento) def _ok(res: SPResult): self.spinner.stop() if res and res.rc == 0: self._recolor_row_by_documento(documento, desired) else: msg = (res.message if res else "Errore sconosciuto") messagebox.showerror("Prenota", f"Operazione non riuscita:\n{msg}") def _err(ex): self.spinner.stop() messagebox.showerror("Prenota", f"Errore:\n{ex}") self.runner.run( _job(), on_success=_ok, on_error=_err, busy=self.busy, message=f"Prenoto la Picking List {documento}…" ) def on_sprenota(self): """Unreserve the selected picking list.""" model = self._get_selected_model() if not model: messagebox.showinfo("S-prenota", "Seleziona una Picking List (checkbox) prima di s-prenotare.") return documento = _s(model.pl.get("Documento")) current = int(model.pl.get("IDStato") or 0) desired = 0 if current == desired: messagebox.showinfo("S-prenota", f"La Picking List {documento} è già NON prenotata.") return id_operatore = 1 # TODO: recupera dal contesto reale self.spinner.start(" S-prenoto…") async def _job(): return await sp_xExePackingListPallet_async(self.db_client, id_operatore, documento) def _ok(res: SPResult): self.spinner.stop() if res and res.rc == 0: self._recolor_row_by_documento(documento, desired) else: msg = (res.message if res else "Errore sconosciuto") messagebox.showerror("S-prenota", f"Operazione non riuscita:\n{msg}") def _err(ex): self.spinner.stop() messagebox.showerror("S-prenota", f"Errore:\n{ex}") self.runner.run( _job(), on_success=_ok, on_error=_err, busy=self.busy, message=f"S-prenoto la Picking List {documento}…" ) def on_export(self): """Placeholder for a future export implementation.""" messagebox.showinfo("Esporta", "Stub esportazione.") # factory per main def create_frame(parent, *, db_client=None, conn_str=None) -> 'GestionePickingListFrame': """Factory used by the launcher to build the picking list frame.""" ctk.set_appearance_mode("light") ctk.set_default_color_theme("green") return GestionePickingListFrame(parent, db_client=db_client) # =================== /gestione_pickinglist.py ===================