"""Operational diagnostic checks for warehouse data quadrature.""" from __future__ import annotations import tkinter as tk from datetime import datetime from tkinter import filedialog, messagebox, ttk from typing import Any import customtkinter as ctk from busy_overlay import InlineBusyOverlay from gestione_aree import AsyncRunner from locale_text import load_locale_catalog, text as loc_text from runtime_support import log_exception from ui_tables import style_treeview, zebra_tag from ui_theme import theme_color, theme_font, theme_section, theme_value from version_info import module_version, versioned_title from window_placement import place_window_fullsize_below_parent_later __version__ = module_version(__name__) SQL_SHIPPED_NOT_DISASSOCIATED = """ SELECT DISTINCT LTRIM(RTRIM(g.BarcodePallet)) AS UDC, CAST('Spedita ma non disassociata' AS varchar(64)) AS Diagnostica, CONCAT( RTRIM(c.Corsia), ' - ', RTRIM(CAST(c.Colonna AS varchar(32))), ' - ', RTRIM(CAST(c.Fila AS varchar(32))) ) AS NomeSimbolicoCella FROM dbo.XMag_GiacenzaPallet AS g INNER JOIN dbo.Celle AS c ON c.ID = g.IDCella INNER JOIN dbo.py_XMag_ViewPackingListStorico AS chiuse ON LTRIM(RTRIM(chiuse.Pallet)) COLLATE Latin1_General_CI_AS = LTRIM(RTRIM(g.BarcodePallet)) COLLATE Latin1_General_CI_AS WHERE chiuse.StatoDocumento = 'D' AND g.IDCella <> 9999 ORDER BY NomeSimbolicoCella, UDC; """ SQL_MULTIPLE_UDC_BY_CELL = """ WITH celle_multiple AS ( SELECT g.IDCella FROM dbo.XMag_GiacenzaPallet AS g WHERE g.IDCella NOT IN (1000, 9999) GROUP BY g.IDCella HAVING COUNT(DISTINCT LTRIM(RTRIM(g.BarcodePallet))) > 1 ) SELECT CONCAT( RTRIM(c.Corsia), ' - ', RTRIM(CAST(c.Colonna AS varchar(32))), ' - ', RTRIM(CAST(c.Fila AS varchar(32))) ) AS CELLA, LTRIM(RTRIM(g.BarcodePallet)) AS UDC FROM dbo.XMag_GiacenzaPallet AS g INNER JOIN celle_multiple AS cm ON cm.IDCella = g.IDCella INNER JOIN dbo.Celle AS c ON c.ID = g.IDCella ORDER BY CELLA, UDC; """ SQL_SAM_UDC_NOT_IN_WMS = """ WITH recent_lotser AS ( SELECT TOP (5000) ls.ID, LEFT(ls.NUMSER, 6) AS UDC, ls.NUMLOT AS Lotto, ls.IDARTICO FROM SAMA1.dbo.LOTSER AS ls WHERE NULLIF(LTRIM(RTRIM(LEFT(ls.NUMSER, 6))), '') IS NOT NULL AND TRY_CONVERT(int, LTRIM(RTRIM(LEFT(ls.NUMSER, 6)))) > 0 AND LEN(LTRIM(RTRIM(LEFT(ls.NUMSER, 6)))) = 6 AND ls.NUMLOT LIKE CONCAT('P', RIGHT(CONVERT(varchar(4), YEAR(GETDATE())), 2), '%') ORDER BY ls.ID DESC ), sam_udc AS ( SELECT LTRIM(RTRIM(r.UDC)) AS UDC, MIN(r.Lotto) AS Lotto, MIN(a.CODICE) AS Prodotto, MIN(a.DESCR) AS Descrizione, MAX(r.ID) AS UltimoIDLotSer FROM recent_lotser AS r INNER JOIN SAMA1.dbo.ARTICO AS a ON a.ID = r.IDARTICO GROUP BY LTRIM(RTRIM(r.UDC)) ) SELECT TOP (100) s.UDC, s.Lotto, s.Prodotto, s.Descrizione FROM sam_udc AS s WHERE NOT EXISTS ( SELECT 1 FROM dbo.MagazziniPallet AS mp WHERE LTRIM(RTRIM(mp.Attributo)) COLLATE Latin1_General_CI_AS = s.UDC COLLATE Latin1_General_CI_AS ) ORDER BY s.UltimoIDLotSer DESC; """ SHIPPED_NOT_DISASSOCIATED_COLUMNS = ( ("UDC", "diagnostics.col.udc", "UDC", 130, "w"), ( "Diagnostica", "diagnostics.col.shipped_not_disassociated", "Spedita ma non disassociata", 280, "w", ), ("NomeSimbolicoCella", "diagnostics.col.symbolic_cell", "Nome simbolico cella", 240, "w"), ) MULTIPLE_UDC_COLUMNS = ( ("CELLA", "diagnostics.col.cell", "CELLA", 220, "w"), ("UDC", "diagnostics.col.udc", "UDC", 160, "w"), ) SAM_UDC_NOT_IN_WMS_COLUMNS = ( ("UDC", "diagnostics.col.udc", "UDC", 130, "w"), ("Lotto", "diagnostics.col.lot", "Lotto", 160, "w"), ("Prodotto", "diagnostics.col.product", "Prodotto", 150, "w"), ("Descrizione", "diagnostics.col.description", "Descrizione", 360, "w"), ) def _rows_to_dicts(res: dict[str, Any] | None) -> list[dict[str, Any]]: if not isinstance(res, dict): return [] rows = res.get("rows") or [] cols = res.get("columns") or [] if rows and isinstance(rows[0], dict): return [row for row in rows if isinstance(row, dict)] out: list[dict[str, Any]] = [] for row in rows: if isinstance(row, (list, tuple)) and cols: out.append({str(cols[i]): row[i] for i in range(min(len(cols), len(row)))}) return out class DiagnosticaWindow(ctk.CTkToplevel): """Window collecting read-only diagnostic queries.""" def __init__(self, parent: tk.Widget, db_client, session=None): super().__init__(parent) self.db_client = db_client self.session = session self._theme = theme_section("diagnostics_window", theme_section("search_window", {})) self._locale_catalog = load_locale_catalog() self._async = AsyncRunner(self) self._busy = InlineBusyOverlay(self, self._theme) self._load_in_progress = False self._current_check = "" self.var_status = tk.StringVar(value="") self._last_export_slug = "diagnostica" self._current_columns = SHIPPED_NOT_DISASSOCIATED_COLUMNS self._column_titles: dict[str, str] = {} self._sort_column = "" self._sort_descending = False self.title( versioned_title( loc_text("diagnostics.title", catalog=self._locale_catalog, default="Diagnostica"), __name__, ) ) self.geometry(str(theme_value(self._theme, "window_geometry", "1100x680"))) minsize = theme_value(self._theme, "window_minsize", [900, 560]) self.minsize(int(minsize[0]), int(minsize[1])) try: self.configure(fg_color=theme_color(self._theme, "window_fg_color", ("#efefef", "#2f2f2f"))) except Exception: pass self._build_ui() def _is_alive(self) -> bool: try: return bool(self.winfo_exists()) except tk.TclError: return False def _hide_busy_safe(self) -> None: try: self._busy.hide() except Exception as exc: log_exception(__name__, exc, context="hide busy diagnostica") def _build_ui(self) -> None: self.grid_rowconfigure(2, weight=1) self.grid_columnconfigure(0, weight=1) top = ctk.CTkFrame( self, fg_color=theme_color(self._theme, "toolbar_frame_fg_color", ("#d7d7d7", "#3b3b3b")), ) top.grid(row=0, column=0, sticky="ew", padx=8, pady=8) top.grid_columnconfigure(3, weight=1) button_font = theme_font(self._theme, "toolbar_button_font", ("Segoe UI", 10, "bold")) ctk.CTkButton( top, text=loc_text( "diagnostics.button.shipped_not_disassociated", catalog=self._locale_catalog, default="UDC spedite non disassociate", ), command=self._load_shipped_not_disassociated, font=button_font, ).grid(row=0, column=0, sticky="w", padx=(0, 8)) ctk.CTkButton( top, text=loc_text( "diagnostics.button.multiple_udc_by_cell", catalog=self._locale_catalog, default="Celle con multiple UDC", ), command=self._load_multiple_udc_by_cell, font=button_font, ).grid(row=0, column=1, sticky="w", padx=(0, 8)) ctk.CTkButton( top, text=loc_text( "diagnostics.button.sam_udc_not_in_wms", catalog=self._locale_catalog, default="Ultime UDC SAM non in WMS", ), command=self._load_sam_udc_not_in_wms, font=button_font, ).grid(row=0, column=2, sticky="w", padx=(0, 8)) self.btn_export = ctk.CTkButton( top, text=loc_text("diagnostics.button.export", catalog=self._locale_catalog, default="Esporta XLSX"), command=self._export_xlsx, font=button_font, state="disabled", ) self.btn_export.grid(row=0, column=4, sticky="e") ctk.CTkLabel( self, textvariable=self.var_status, anchor="w", font=theme_font(self._theme, "status_font", ("Segoe UI", 10, "bold")), ).grid(row=1, column=0, sticky="ew", padx=10, pady=(0, 6)) wrap = ctk.CTkFrame(self) wrap.grid(row=2, column=0, sticky="nsew", padx=8, pady=(0, 8)) wrap.grid_rowconfigure(0, weight=1) wrap.grid_columnconfigure(0, weight=1) self.tree = ttk.Treeview(wrap, show="headings") self._configure_tree_columns(self._current_columns) style_treeview( self.tree, style_name="Diagnostics.Treeview", rowheight=22, font=("", 9), heading_font=("", 9, "bold"), ) sy = ttk.Scrollbar(wrap, orient="vertical", command=self.tree.yview) sx = ttk.Scrollbar(wrap, orient="horizontal", command=self.tree.xview) self.tree.configure(yscrollcommand=sy.set, xscrollcommand=sx.set) self.tree.grid(row=0, column=0, sticky="nsew") sy.grid(row=0, column=1, sticky="ns") sx.grid(row=1, column=0, sticky="ew") def _set_status(self, text: str) -> None: self.var_status.set(text) def _configure_tree_columns(self, columns) -> None: self.tree.delete(*self.tree.get_children("")) keys = tuple(str(col[0]) for col in columns) self.tree.configure(columns=keys, show="headings") self._sort_column = "" self._sort_descending = False self._column_titles = {} for key, locale_key, default, width, anchor in columns: title = loc_text(str(locale_key), catalog=self._locale_catalog, default=str(default)) self._column_titles[str(key)] = title self.tree.heading( key, text=title, command=lambda col=str(key): self._sort_by_column(col), ) self.tree.column(key, width=int(width), anchor=str(anchor), stretch=True) def _refresh_sort_headings(self) -> None: for key, title in self._column_titles.items(): suffix = "" if key == self._sort_column: suffix = " v" if self._sort_descending else " ^" self.tree.heading(key, text=f"{title}{suffix}", command=lambda col=key: self._sort_by_column(col)) def _sort_key(self, value: Any) -> tuple[int, Any]: text = str(value or "").strip() if not text: return (2, "") normalized = text.replace(".", "").replace(",", ".") try: return (0, float(normalized)) except ValueError: return (1, text.casefold()) def _sort_by_column(self, column: str) -> None: columns = list(self.tree["columns"]) if column not in columns: return descending = not self._sort_descending if self._sort_column == column else False col_index = columns.index(column) rows = [] for iid in self.tree.get_children(""): values = self.tree.item(iid, "values") value = values[col_index] if col_index < len(values) else "" rows.append((self._sort_key(value), iid)) rows.sort(key=lambda item: item[0], reverse=descending) for index, (_key, iid) in enumerate(rows): self.tree.move(iid, "", index) self.tree.item(iid, tags=(zebra_tag(index),)) self._sort_column = column self._sort_descending = descending self._refresh_sort_headings() def _load_shipped_not_disassociated(self) -> None: self._run_check( key="shipped_not_disassociated", sql=SQL_SHIPPED_NOT_DISASSOCIATED, columns=SHIPPED_NOT_DISASSOCIATED_COLUMNS, busy_text=loc_text( "diagnostics.busy.shipped_not_disassociated", catalog=self._locale_catalog, default="Cerco UDC spedite ma ancora in cella...", ), ) def _load_multiple_udc_by_cell(self) -> None: self._run_check( key="multiple_udc_by_cell", sql=SQL_MULTIPLE_UDC_BY_CELL, columns=MULTIPLE_UDC_COLUMNS, busy_text=loc_text( "diagnostics.busy.multiple_udc_by_cell", catalog=self._locale_catalog, default="Cerco celle con multiple UDC...", ), ) def _load_sam_udc_not_in_wms(self) -> None: self._run_check( key="sam_udc_not_in_wms", sql=SQL_SAM_UDC_NOT_IN_WMS, columns=SAM_UDC_NOT_IN_WMS_COLUMNS, busy_text=loc_text( "diagnostics.busy.sam_udc_not_in_wms", catalog=self._locale_catalog, default="Cerco le ultime UDC presenti in SAM ma non ancora in WMS...", ), ) def _run_check(self, *, key: str, sql: str, columns, busy_text: str) -> None: if self._load_in_progress or not self._is_alive(): return async def job(): return await self.db_client.query_json(sql, as_dict_rows=True) self._load_in_progress = True self._current_check = key self._last_export_slug = key self._current_columns = columns self._configure_tree_columns(columns) try: self.btn_export.configure(state="disabled") except Exception: pass try: self._busy.show(busy_text) self._async.run(job(), self._on_loaded, self._on_error) except Exception as exc: self._load_in_progress = False self._hide_busy_safe() log_exception(__name__, exc, context=f"avvio diagnostica {key}") messagebox.showerror( loc_text("diagnostics.title", catalog=self._locale_catalog, default="Diagnostica"), str(exc), parent=self if self._is_alive() else None, ) def _on_loaded(self, res: dict[str, Any] | None) -> None: self._load_in_progress = False self._hide_busy_safe() if not self._is_alive(): return try: rows = _rows_to_dicts(res) self.tree.delete(*self.tree.get_children("")) keys = [str(col[0]) for col in self._current_columns] for index, row in enumerate(rows): self.tree.insert( "", "end", values=tuple(row.get(key) or "" for key in keys), tags=(zebra_tag(index),), ) try: self.btn_export.configure(state="normal" if rows else "disabled") except Exception: pass self._set_status( loc_text( "diagnostics.status.rows", catalog=self._locale_catalog, default="Righe trovate: {count}", ).format(count=len(rows)) ) except Exception as exc: log_exception(__name__, exc, context="render diagnostica") messagebox.showerror( loc_text("diagnostics.title", catalog=self._locale_catalog, default="Diagnostica"), str(exc), parent=self, ) def _on_error(self, exc: BaseException) -> None: self._load_in_progress = False self._hide_busy_safe() log_exception(__name__, exc, context=f"query diagnostica {self._current_check}") if not self._is_alive(): return messagebox.showerror( loc_text("diagnostics.title", catalog=self._locale_catalog, default="Diagnostica"), str(exc), parent=self, ) def _export_xlsx(self) -> None: """Export the currently visible diagnostic grid to an Excel file.""" rows = [self.tree.item(iid, "values") for iid in self.tree.get_children("")] if not rows: messagebox.showinfo( loc_text("diagnostics.export.title", catalog=self._locale_catalog, default="Esporta"), loc_text( "diagnostics.export.empty", catalog=self._locale_catalog, default="Non ci sono righe da esportare.", ), parent=self, ) return try: from openpyxl import Workbook from openpyxl.styles import Alignment, Font from openpyxl.utils import get_column_letter except Exception as exc: messagebox.showerror( loc_text("diagnostics.export.title", catalog=self._locale_catalog, default="Esporta"), loc_text( "diagnostics.export.dep", catalog=self._locale_catalog, default="Per l'esportazione serve 'openpyxl' (pip install openpyxl).", ).format(error=exc), parent=self, ) return ts = datetime.now().strftime("%Y%m%d_%H%M") default_name = f"{self._last_export_slug}_{ts}.xlsx" path = filedialog.asksaveasfilename( parent=self, title=loc_text("diagnostics.export.title", catalog=self._locale_catalog, default="Esporta"), defaultextension=".xlsx", filetypes=[("Excel Workbook", "*.xlsx")], initialfile=default_name, ) if not path: return try: wb = Workbook() ws = wb.active ws.title = "Diagnostica" columns = list(self.tree["columns"]) headers = [str(self.tree.heading(col, "text") or col) for col in columns] for col_index, header in enumerate(headers, start=1): cell = ws.cell(row=1, column=col_index, value=header) cell.font = Font(bold=True) cell.alignment = Alignment(horizontal="center", vertical="center") for row_index, row in enumerate(rows, start=2): for col_index, value in enumerate(row, start=1): ws.cell(row=row_index, column=col_index, value=value) widths: dict[int, int] = {} for row in ws.iter_rows(values_only=True): for col_index, value in enumerate(row, start=1): widths[col_index] = max(widths.get(col_index, 0), len("" if value is None else str(value))) for col_index, width in widths.items(): ws.column_dimensions[get_column_letter(col_index)].width = min(max(width + 2, 12), 70) wb.save(path) messagebox.showinfo( loc_text("diagnostics.export.title", catalog=self._locale_catalog, default="Esporta"), loc_text("diagnostics.export.ok", catalog=self._locale_catalog, default="File creato:\n{path}").format(path=path), parent=self, ) except Exception as exc: log_exception(__name__, exc, context="export diagnostica xlsx") messagebox.showerror( loc_text("diagnostics.export.title", catalog=self._locale_catalog, default="Esporta"), loc_text( "diagnostics.export.error", catalog=self._locale_catalog, default="Errore durante l'esportazione:\n{error}", ).format(error=exc), parent=self, ) def open_diagnostica_window(parent: tk.Misc, db_client, session=None) -> tk.Misc: """Open the diagnostics window.""" win = DiagnosticaWindow(parent, db_client, session=session) place_window_fullsize_below_parent_later(parent, win) win.bind("", lambda _e: win.destroy()) return win