"""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; """ 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.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(1, 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)) 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=2, 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) cols = ("UDC", "Diagnostica", "NomeSimbolicoCella") self.tree = ttk.Treeview(wrap, columns=cols, show="headings") headings = { "UDC": ( loc_text("diagnostics.col.udc", catalog=self._locale_catalog, default="UDC"), 130, "w", ), "Diagnostica": ( loc_text( "diagnostics.col.shipped_not_disassociated", catalog=self._locale_catalog, default="Spedita ma non disassociata", ), 280, "w", ), "NomeSimbolicoCella": ( loc_text( "diagnostics.col.symbolic_cell", catalog=self._locale_catalog, default="Nome simbolico cella", ), 240, "w", ), } for col in cols: text, width, anchor = headings[col] self.tree.heading(col, text=text) self.tree.column(col, width=width, anchor=anchor, stretch=True) 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 _load_shipped_not_disassociated(self) -> None: self._run_check( key="shipped_not_disassociated", sql=SQL_SHIPPED_NOT_DISASSOCIATED, busy_text=loc_text( "diagnostics.busy.shipped_not_disassociated", catalog=self._locale_catalog, default="Cerco UDC spedite ma ancora in cella...", ), ) def _run_check(self, *, key: str, sql: str, 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 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("")) for index, row in enumerate(rows): self.tree.insert( "", "end", values=( row.get("UDC") or "", row.get("Diagnostica") or "", row.get("NomeSimbolicoCella") or "", ), 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