544 lines
20 KiB
Python
544 lines
20 KiB
Python
"""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("<Escape>", lambda _e: win.destroy())
|
|
return win
|