# C:\PYTHON\MyICR_Suite\app\main_window.py
import sys
import subprocess
from pathlib import Path
from functools import partial
from PyQt6.QtWidgets import (
QMainWindow, QWidget, QStackedWidget, QLabel, QMessageBox,
QVBoxLayout, QHBoxLayout, QPushButton, QFrame, QApplication
)
from PyQt6.QtCore import Qt, QTimer, QThreadPool
from PyQt6.QtGui import QAction, QFont, QCloseEvent
from loguru import logger
from app.config_loader import is_debug, get_app_root, get_sync_config
from app.widgets.loading_page import LoadingPageWidget
from app.features.navigation import NavigationPage
from app.utils.worker import Worker
from app.features.dash_tipologie import DashboardTipologiePage
from app.features.dash_tipologie.filter_panel_widget import FilterPanel as FilterPanelTipologie
from app.features.dash_tipologie.fullscreen_page import FullScreenTablePage as FullScreenTipologie
from app.features.dash_em_pivot import DashboardEmPivotPage
from app.features.dash_em_pivot.filter_panel_widget import FilterPanelEmPivot
from app.features.dash_em_pivot.fullscreen_page import FullScreenTablePage as FullScreenPivot
from app.features.dash_diritti import DashboardDirittiPage
from app.features.dash_diritti.filter_panel_widget import FilterPanelDiritti
from app.features.dash_diritti.fullscreen_page import FullScreenTablePage as FullScreenDiritti
class MainWindow(QMainWindow):
def __init__(self, config):
super().__init__()
self.page_registry = {
'navigation_page': {'class': NavigationPage, 'type': 'core', 'title': '🏠 Home'},
'dash_tipologie': {
'class': DashboardTipologiePage, 'type': 'container', 'title': "Dashboard Tipologie",
'async_loader': FilterPanelTipologie.get_filter_data
},
'dash_em_pivot': {
'class': DashboardEmPivotPage, 'type': 'container', 'title': "Dashboard Pivot Emissioni",
'async_loader': FilterPanelEmPivot.get_filter_data
},
'dash_diritti': {
'class': DashboardDirittiPage,
'type': 'container',
'title': "Dashboard Diritti",
'async_loader': FilterPanelDiritti.get_filter_data,
},
'fullscreen_tipologie': { 'class': FullScreenTipologie, 'type': 'sub-container' },
'fullscreen_pivot': { 'class': FullScreenPivot, 'type': 'sub-container' },
'fullscreen_diritti': { 'class': FullScreenDiritti, 'type': 'sub-container' }
}
self.config = config
self.theme = self.config.get('theme', {})
self.open_pages = {}
self.tab_widgets = {}
self.tab_buttons = {}
self.page_counter = 0
self.current_page_widget = None
self.threadpool = QThreadPool()
self.mediatrack_process = None # Traccia il processo MediaTrack
logger.info(f"Multithreading con un massimo di {self.threadpool.maxThreadCount()} thread.")
self.init_ui()
def init_ui(self):
# Costruisci titolo
from app.config_loader import get_version
_ver = get_version()
base_title = f"MyICR Suite v{_ver}" if _ver else "MyICR Suite"
window_title = f"{base_title} - Development" if is_debug() else base_title
self.setWindowTitle(window_title)
self.resize(1600, 1000)
self._create_main_menu_bar()
self.status_bar = self.statusBar()
self.status_bar.setStyleSheet("padding: 3px;")
# Aggiungi indicatore ambiente permanente nella status bar
self._add_environment_indicator()
central_widget = QWidget()
self.setCentralWidget(central_widget)
main_layout = QVBoxLayout(central_widget)
main_layout.setContentsMargins(0, 0, 0, 0)
main_layout.setSpacing(0)
self.central_stacked_widget = QStackedWidget()
self.central_stacked_widget.currentChanged.connect(self._on_current_page_changed)
self.tab_bar_widget = QWidget()
self.tab_bar_layout = QHBoxLayout(self.tab_bar_widget)
self.tab_bar_layout.setContentsMargins(5, 2, 5, 2)
self.tab_bar_layout.setSpacing(2)
self.tab_bar_layout.setAlignment(Qt.AlignmentFlag.AlignLeft)
self.tab_bar_widget.setStyleSheet("background-color: #E1E1E1; border-top: 1px solid #C6C6C6;")
main_layout.addWidget(self.central_stacked_widget, stretch=1)
main_layout.addWidget(self.tab_bar_widget)
self.open_page('navigation_page', is_singleton=True, is_closable=False)
def closeEvent(self, event: QCloseEvent):
current_widget = self.central_stacked_widget.currentWidget()
if not current_widget:
event.accept()
return
page_id = current_widget.property("page_id")
is_closable = self.tab_widgets.get(page_id) and len(self.tab_widgets[page_id].findChildren(QPushButton, "x")) > 0
if len(self.open_pages) > 1 and is_closable:
logger.info(f"Evento di chiusura intercettato. Chiusura della tab: {page_id}")
self.close_page(page_id)
event.ignore()
else:
reply = QMessageBox.question(self, 'Conferma Uscita',
"Sei sicuro di voler chiudere l'applicazione?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No)
if reply == QMessageBox.StandardButton.Yes:
logger.info("Chiusura applicazione confermata dall'utente.")
# Termina il processo MediaTrack se in esecuzione
if self.mediatrack_process is not None:
if self.mediatrack_process.poll() is None:
logger.info("Terminazione processo MediaTrack...")
self.mediatrack_process.terminate()
try:
self.mediatrack_process.wait(timeout=3)
logger.info("Processo MediaTrack terminato correttamente")
except subprocess.TimeoutExpired:
logger.warning("Processo MediaTrack non risponde, forzatura...")
self.mediatrack_process.kill()
event.accept()
else:
logger.info("Chiusura applicazione annullata dall'utente.")
event.ignore()
def _create_main_menu_bar(self):
menu_bar = self.menuBar()
primary_color = self.theme.get('primary_color', '#0078D4')
primary_hover_color = self.theme.get('primary_hover', '#005A9E')
text_white = self.theme.get('text_white', '#FFFFFF')
menu_bar.setStyleSheet(f"""
QMenuBar {{
background-color: {primary_color};
color: {text_white};
font-size: 13px;
padding: 4px;
}}
QMenuBar::item {{
padding: 4px 8px;
background-color: transparent;
}}
QMenuBar::item:selected {{
background-color: {primary_hover_color};
}}
QMenu {{
background-color: #FFFFFF;
border: 1px solid #D2D0CE;
}}
QMenu::item:selected {{
background-color: #DEECF9;
}}
""")
file_menu = menu_bar.addMenu("&File")
exit_action = QAction("Esci", self)
exit_action.triggered.connect(self.close)
file_menu.addAction(exit_action)
help_menu = menu_bar.addMenu("&Guida")
about_action = QAction("Informazioni su...", self)
about_action.triggered.connect(self.show_about_dialog)
help_menu.addAction(about_action)
def _add_environment_indicator(self):
"""Mostra SYNC OFF nella status bar se push_enabled è False."""
sync_config = get_sync_config()
if not sync_config.get("push_enabled", True):
sync_label = QLabel(" SYNC OFF ")
sync_label.setStyleSheet("""
QLabel {
background-color: #FF8C00;
color: white;
padding: 2px 8px;
border-radius: 3px;
font-weight: bold;
font-size: 11px;
}
""")
self.status_bar.addPermanentWidget(sync_label)
def show_about_dialog(self):
QMessageBox.about(self,
"Informazioni su MyICR Suite",
"MyICR Suite
"
"Versione 1.0.0 (build fittizia)
"
"Applicazione desktop per l'analisi dati.
"
"Sviluppata internamente.")
def launch_mediatrack(self):
"""Lancia MediaTrack come modulo integrato."""
try:
# Controlla se MediaTrack è già in esecuzione
if self.mediatrack_process is not None:
# Verifica se il processo è ancora vivo
if self.mediatrack_process.poll() is None:
# Processo ancora in esecuzione
logger.info("MediaTrack già in esecuzione")
QMessageBox.information(
self,
"MediaTrack già aperto",
"MediaTrack è già in esecuzione.\n\n"
"Controlla le finestre aperte o attendi che si carichi."
)
return
else:
# Processo terminato, resetta
self.mediatrack_process = None
# Mostra cursore di attesa
QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
self.show_status_message("Avvio MediaTrack in corso...", is_temporary=False)
# Determina il percorso del modulo MediaTrack (ora integrato in app/modules)
mediatrack_path = Path(__file__).resolve().parent / "modules" / "mediatrack"
run_script = mediatrack_path / "run.py"
requirements_file = mediatrack_path / "requirements.txt"
if not run_script.exists():
logger.error(f"Script MediaTrack non trovato: {run_script}")
QApplication.restoreOverrideCursor()
QMessageBox.warning(
self,
"MediaTrack non disponibile",
f"L'applicazione MediaTrack non è stata trovata.\n\n"
f"Percorso cercato: {run_script}"
)
return
# Verifica se requirements.txt esiste (dipendenze installate)
if requirements_file.exists():
logger.info("Verifico dipendenze MediaTrack...")
# Lancia MediaTrack come processo separato
logger.info(f"Lancio MediaTrack da: {run_script}")
# Determina Python executable: venv se presente (dev), altrimenti sys.executable
venv_python = get_app_root().parent / "venv" / "Scripts" / "python.exe"
if venv_python.exists():
python_exe = venv_python
else:
python_dir = Path(sys.executable).parent
python_exe = python_dir / "python.exe"
if not python_exe.exists():
python_exe = Path(sys.executable)
logger.warning(f"python.exe non trovato, uso sys.executable: {python_exe}")
logger.info(f"Python executable: {python_exe}")
logger.info(f"Working directory: {mediatrack_path}")
# Su Windows, nascondi la console usando CREATE_NO_WINDOW flag
if sys.platform == 'win32':
# CREATE_NO_WINDOW nasconde la console senza bisogno di pythonw.exe
# Disponibile da Python 3.7+ / Windows
CREATE_NO_WINDOW = 0x08000000
process = subprocess.Popen(
[python_exe, str(run_script)],
cwd=str(mediatrack_path),
creationflags=CREATE_NO_WINDOW,
# Capture stdout/stderr per debug (non blocca subprocess)
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
else:
# Su Linux/Mac lancia normalmente
process = subprocess.Popen(
[python_exe, str(run_script)],
cwd=str(mediatrack_path),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# Salva il processo per prevenire lanci multipli
self.mediatrack_process = process
logger.info(f"Processo MediaTrack avviato con PID: {process.pid}")
# Spawn thread per monitorare stderr (per debug errori avvio)
import threading
def monitor_stderr():
try:
for line in process.stderr:
if line.strip():
logger.error(f"[MediaTrack stderr] {line.strip()}")
except Exception:
pass # Subprocess terminato, ignora errori lettura
stderr_thread = threading.Thread(target=monitor_stderr, daemon=True)
stderr_thread.start()
# Ripristina cursore normale
QApplication.restoreOverrideCursor()
self.show_status_message("MediaTrack avviato", is_temporary=True, duration_ms=2000)
except Exception as e:
logger.error(f"Errore durante il lancio di MediaTrack: {e}", exc_info=True)
# Assicurati di ripristinare il cursore anche in caso di errore
QApplication.restoreOverrideCursor()
QMessageBox.critical(
self,
"Errore",
f"Impossibile avviare MediaTrack.\n\nErrore: {e}"
)
def launch_linker(self):
"""Lancia Linker come QMainWindow floating."""
from app.modules.linker.launcher import launch_linker_window
launch_linker_window(parent=self)
def open_page(self, page_id_prefix: str, title: str = None, is_singleton: bool = True, is_closable: bool = True, data=None):
# Gestione speciale per MediaTrack - lancia come applicazione separata
if page_id_prefix == 'mediatrack':
self.launch_mediatrack()
return
# Gestione speciale per Linker - lancia come QMainWindow floating
if page_id_prefix == 'linker':
self.launch_linker()
return
page_id = page_id_prefix
if is_singleton and page_id in self.open_pages:
self.focus_page(page_id)
return
if not is_singleton:
self.page_counter += 1
page_id = f"{page_id_prefix}_{self.page_counter}"
page_info = self.page_registry.get(page_id_prefix)
if not page_info:
logger.error(f"Tentativo di aprire una pagina non registrata: {page_id_prefix}")
QMessageBox.warning(self, "Funzionalità non disponibile", f"La pagina '{page_id_prefix}' non è stata ancora implementata o registrata.")
return
display_title = title or page_info.get('title', page_id_prefix.replace('_', ' ').title())
if page_info.get('async_loader'):
self._open_async_page(page_id, page_id_prefix, display_title, is_closable)
else:
page_widget = self._create_page_instance(page_id_prefix, data)
if page_widget:
self._add_page_to_tabs(page_widget, page_id, display_title, is_closable)
def _open_async_page(self, page_id, page_id_prefix, title, is_closable):
loading_page = LoadingPageWidget(text="Aggiornamento in corso...")
self._add_page_to_tabs(loading_page, page_id, f"{title} (loading...)", is_closable)
page_info = self.page_registry.get(page_id_prefix)
loader_func = page_info.get('async_loader')
if not loader_func:
self._on_async_page_error((None, f"Configurazione async loader per '{page_id_prefix}' mancante."), page_id)
return
worker = Worker(loader_func)
worker.signals.result.connect(lambda d: self._on_async_page_loaded(d, page_id, page_id_prefix, title))
worker.signals.error.connect(lambda e: self._on_async_page_error(e, page_id))
self.threadpool.start(worker)
def _on_async_page_loaded(self, data, page_id, page_id_prefix, title):
logger.info(f"Dati caricati per {page_id}, creo pagina.")
loading_widget = self.open_pages.pop(page_id, None)
if loading_widget:
self.central_stacked_widget.removeWidget(loading_widget)
loading_widget.deleteLater()
page_widget = self._create_page_instance(page_id_prefix, data)
if page_widget:
tab_button = self.tab_buttons.get(page_id)
if tab_button:
tab_button.setText(title)
self.central_stacked_widget.addWidget(page_widget)
self.open_pages[page_id] = page_widget
self.focus_page(page_id)
else:
self.close_page(page_id)
def _on_async_page_error(self, error_tuple, page_id):
error_msg = error_tuple[1] if len(error_tuple) > 1 else 'Errore sconosciuto'
logger.error(f"Errore caricamento pagina {page_id}: {error_msg}")
self.close_page(page_id)
QMessageBox.critical(self, "Errore Caricamento", f"Impossibile caricare i dati necessari.\\n\\n{error_msg}")
def _add_page_to_tabs(self, page_widget, page_id, title, is_closable):
page_widget.setProperty("page_id", page_id)
if hasattr(page_widget, 'status_updated'):
page_widget.status_updated.connect(self.show_status_message)
self.central_stacked_widget.addWidget(page_widget)
self.open_pages[page_id] = page_widget
self._create_tab_button_widget(page_id, title, is_closable)
self.tab_bar_layout.addWidget(self.tab_widgets[page_id])
self.focus_page(page_id)
def _create_page_instance(self, page_id_prefix: str, data=None) -> QWidget:
page_info = self.page_registry.get(page_id_prefix)
PageClass = page_info['class']
try:
if page_info['type'] == 'core':
instance = PageClass(main_window=self)
elif page_info['type'] == 'container':
instance = PageClass(main_window=self, filter_data=data)
elif page_info['type'] == 'sub-container':
if data is None or 'dataframe' not in data:
logger.error(f"La pagina sub-container '{page_id_prefix}' richiede un dataframe nei dati.")
return None
instance = PageClass(main_window=self, data=data['dataframe'], theme=self.theme)
else:
instance = PageClass()
if hasattr(instance, 'open_new_tab_requested'):
instance.open_new_tab_requested.connect(self.on_open_new_tab_requested)
if hasattr(instance, 'section_selected'):
instance.section_selected.connect(lambda pid: self.open_page(pid, is_singleton=True))
display_title = data.get('title') if isinstance(data, dict) else page_info.get('title', page_id_prefix)
if display_title:
instance.setWindowTitle(display_title)
return instance
except Exception as e:
logger.error(f"Errore durante l'istanza della pagina {page_id_prefix}: {e}", exc_info=True)
QMessageBox.critical(self, "Errore Creazione Pagina", f"Impossibile creare la pagina '{page_id_prefix}'.\\n\\n{e}")
return None
def _create_tab_button_widget(self, page_id: str, title: str, is_closable: bool):
tab_widget = QWidget()
tab_layout = QHBoxLayout(tab_widget)
tab_layout.setContentsMargins(0, 0, 0, 0)
tab_layout.setSpacing(5)
title_button = QPushButton(title)
title_button.setCheckable(True)
title_button.setAutoExclusive(True)
title_button.clicked.connect(lambda: self.focus_page(page_id))
self.tab_widgets[page_id] = tab_widget
self.tab_buttons[page_id] = title_button
tab_layout.addWidget(title_button)
if is_closable:
close_button = QPushButton("x")
close_button.setObjectName("x")
close_button.setFixedSize(20, 20)
close_button.clicked.connect(lambda: self.close_page(page_id))
tab_layout.addWidget(close_button)
close_button.setStyleSheet("QPushButton { border: none; font-weight: bold; } QPushButton:hover { color: red; }")
title_button.setStyleSheet("""
QPushButton { text-align: center; padding: 6px 10px; border: 1px solid #C6C6C6; border-top: none; background-color: #E1E1E1; }
QPushButton:hover { background-color: #CACACA; }
QPushButton:checked { background-color: #FFFFFF; border-color: #C6C6C6; }
""")
def focus_page(self, page_id: str):
if page_id in self.open_pages:
self.central_stacked_widget.setCurrentWidget(self.open_pages[page_id])
if page_id in self.tab_buttons:
self.tab_buttons[page_id].setChecked(True)
def close_page(self, page_id: str):
if page_id not in self.open_pages: return
logger.info(f"Chiusura pagina: {page_id}")
page_widget = self.open_pages.pop(page_id, None)
tab_widget = self.tab_widgets.pop(page_id, None)
self.tab_buttons.pop(page_id, None)
if page_widget:
self.central_stacked_widget.removeWidget(page_widget)
page_widget.deleteLater()
if tab_widget:
self.tab_bar_layout.removeWidget(tab_widget)
tab_widget.deleteLater()
if self.open_pages:
self.focus_page(list(self.open_pages.keys())[-1])
def _on_current_page_changed(self, index: int):
new_widget = self.central_stacked_widget.widget(index)
if self.current_page_widget and hasattr(self.current_page_widget, 'page_deactivated'):
self.current_page_widget.page_deactivated()
if new_widget and hasattr(new_widget, 'page_activated'):
new_widget.page_activated()
self.current_page_widget = new_widget
def show_toolbar(self):
pass
def hide_toolbar(self):
pass
def show_status_message(self, message: str, is_temporary: bool = False, duration_ms: int = 4000):
self.statusBar().clearMessage()
if is_temporary:
self.status_bar.showMessage(message, duration_ms)
else:
self.status_bar.showMessage(message)
def create_placeholder_widget(self, title: str):
label = QLabel(f"{title}
Contenuto in sviluppo...
", alignment=Qt.AlignmentFlag.AlignCenter)
return label
def on_open_new_tab_requested(self, page_id_prefix: str, title: str, is_singleton: bool, data: dict):
logger.info(f"Ricevuta richiesta di apertura tab per '{page_id_prefix}' con dati: {data is not None}")
self.open_page(
page_id_prefix=page_id_prefix,
title=title,
is_singleton=is_singleton,
is_closable=True,
data=data
)