""" GemmaSheet — foglio Gemma (read-only). Usa QTableView + QAbstractTableModel per gestire 51K righe senza creare QTableWidgetItem per ogni cella — Qt renderizza solo le righe visibili. """ import logging from PyQt6.QtCore import ( Qt, QAbstractTableModel, QModelIndex, QThread, pyqtSignal, QObject, QUrl, ) from PyQt6.QtGui import ( QColor, QFont, QPainter, QKeySequence, QShortcut, QDesktopServices, QAction, QBrush, ) from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QTableView, QHeaderView, QAbstractItemView, QApplication, QLabel, QFileDialog, QMessageBox, QToolButton, QMenu, ) from app.modules.linker.src.gui.grid_filler import ( FlatBgDelegate, paint_std_header_section, HEADER_HEIGHT, HEADER_BORDER_COLOR, SelectableVHeader, HEADER_SEL_COLOR, ) from app.modules.linker.src.gui.filter_bar import FilterBar from app.modules.linker.src.gui.delegates import RiferLinkDelegate logger = logging.getLogger(__name__) # ── Colonne ─────────────────────────────────────────────────────────────────── _COLS = [ # (header, parquet_field | None, width) ('GEMMA', 'A_ID', 45), ('IMDB', 'A_COD_IMDB', 90), ('TITOLO', 'A_TITOLO', 200), ('REGIA', 'A_AUTORE_REGISTA', 100), ('CAST', 'A_CAST', 100), ('DATA', 'A_DATA', 70), ('DISTRIBUTORE', 'A_DISTRIBUTORE', 120), ('TIPOLOGIA', 'A_TIPOLOGIA', 80), ('SUPPORTO', 'A_SUPPORTO', 90), ('NOTE SUPP.', 'A_NOTE_SUPPORTO', 100), ('DEADLINE', 'A_DEADLINE', 70), ('NOTE PUBB.', 'A_NOTE_PUBBLICHE', 150), ('NOTE PRIV.', 'A_NOTE_PRIVATE', 150), ('ACQ/VEND', 'A_ACQUISTATO_VENDUTO', 80), ('STATO', 'A_STATO', 80), ('RDA', 'V_RDA', 40), ('C5', 'V_C5', 40), ('I1', 'V_I1', 40), ('R4', 'V_R4', 40), ('LA5', 'V_LA5', 40), ('I2', 'V_I2', 40), ('IRIS', 'V_IRIS', 40), ('TOP', 'V_TOP', 40), ('FOC', 'V_FOC', 40), ('C20', 'V_C20', 40), ('CI34', 'V_CI34', 40), ('C27', 'V_C27', 40), ('INF', 'V_INF', 40), ('STATO RIC.', 'V_STATO', 80), ('RETE', 'em_rete', 50), ('DATA EM.', 'em_data', 70), ('HI', 'em_hi', 40), ('FA', 'em_fa', 40), ('AUD.', 'em_aud', 50), ('SHA', 'em_sha', 50), ] _HEADERS = [c[0] for c in _COLS] _FIELDS = [c[1] for c in _COLS] _WIDTHS = [c[2] for c in _COLS] _N_COLS = len(_COLS) _GEMMA_FIELDS = [f for f in _FIELDS if f is not None and not f.startswith('em_')] _VAL_INDICES = list(range(15, 29)) # RDA…STATO RIC. _EM_INDICES = list(range(29, 35)) # RETE…SHA COL_IMDB = 1 COL_ACQVEND = 13 COL_EM_RETE = 29 _MEDIASET_RETI = { 'B5','B6','BP','C5','FU','I1','I2','KA','KB','KI','KQ','LA','LB','LT','R4','TS', } _BG_VENDUTO = QColor('#ffcdd2') _BG_ACQUISTATO = QColor('#c8e6c9') _BG_CONCORRENZA = _BG_VENDUTO _HC_DATA = QColor('#BDD7EE') _HC_VAL = QColor('#C8B0E0') _HC_EM = QColor('#B4EAD7') _CELL_FONT = QFont('Calibri', 8) _TOOLBAR_COLOR = '#9DC3E6' # ── Model ───────────────────────────────────────────────────────────────────── class GemmaModel(QAbstractTableModel): """Modello dati virtuale — restituisce valori on-demand, nessun QTableWidgetItem.""" def __init__(self, parent=None): super().__init__(parent) self._rows: list[list[str]] = [] def set_rows(self, rows: list[list[str]]): self.beginResetModel() self._rows = rows self.endResetModel() def rowCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self._rows) def columnCount(self, parent=QModelIndex()): return 0 if parent.isValid() else _N_COLS def data(self, index, role=Qt.ItemDataRole.DisplayRole): if not index.isValid(): return None row, col = index.row(), index.column() if row >= len(self._rows): return None if role == Qt.ItemDataRole.DisplayRole: return self._rows[row][col] if role == Qt.ItemDataRole.BackgroundRole: rete = self._rows[row][COL_EM_RETE] if rete and rete not in _MEDIASET_RETI: return QBrush(_BG_CONCORRENZA) acqvend = self._rows[row][COL_ACQVEND].lower() if acqvend == 'venduto': return QBrush(_BG_VENDUTO) if acqvend == 'acquistato': return QBrush(_BG_ACQUISTATO) return None if role == Qt.ItemDataRole.FontRole: return _CELL_FONT return None def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole): if role != Qt.ItemDataRole.DisplayRole: return None if orientation == Qt.Orientation.Horizontal: return _HEADERS[section] return str(section + 1) def get_row(self, row_idx: int) -> list[str]: return self._rows[row_idx] if row_idx < len(self._rows) else [] # ── Header colorato (orizzontale) ───────────────────────────────────────────── class _GemmaHeader(QHeaderView): def __init__(self, view): super().__init__(Qt.Orientation.Horizontal, view) self._view = view self._sort_col = -1 self._sort_asc = True self.setSectionsClickable(True) self.setSectionResizeMode(QHeaderView.ResizeMode.Interactive) self.setFixedHeight(HEADER_HEIGHT) def set_sort(self, col: int, asc: bool): self._sort_col = col self._sort_asc = asc self.viewport().update() def paintSection(self, painter: QPainter, rect, logicalIndex: int): painter.save() sel = self._view.selectionModel() if sel and sel.columnIntersectsSelection(logicalIndex, QModelIndex()): bg = HEADER_SEL_COLOR elif logicalIndex in _VAL_INDICES: bg = _HC_VAL elif logicalIndex in _EM_INDICES: bg = _HC_EM else: bg = _HC_DATA text = self.model().headerData( logicalIndex, Qt.Orientation.Horizontal, Qt.ItemDataRole.DisplayRole ) paint_std_header_section(painter, rect, str(text) if text else '', bg, HEADER_BORDER_COLOR) if self._sort_col == logicalIndex: arrow = '▲' if self._sort_asc else '▼' painter.drawText( rect.adjusted(0, 0, -2, 0), Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignTop, arrow, ) painter.restore() # ── Worker caricamento ──────────────────────────────────────────────────────── class _LoadWorker(QObject): finished = pyqtSignal(list) # list[list[str]] def __init__(self, parquet_path: str): super().__init__() self._parquet_path = parquet_path def run(self): try: import duckdb import os as _os base = self._parquet_path.replace('\\', '/') gemma_f = base + '/gemma.parquet' imdb_f = base + '/imdb.parquet' emesso_f = base + '/emesso.parquet' agg_f = base + '/../parquet_agg/emesso_agg.parquet' gemma_cols = ', '.join( f"COALESCE(CAST(g.{f} AS VARCHAR), '')" for f in _GEMMA_FIELDS ) # prima emissione (qualsiasi rete): usa emesso_agg se disponibile if _os.path.exists(agg_f): em_subq = f""" SELECT RIFER AS prodotto, rete_first AS rete, data_first AS data_emissione, printf('%06d', CAST(ora_first AS INTEGER)) AS ora_inizio, fascia_first AS fascia, aud_first AS audience, sha_first AS share FROM read_parquet('{agg_f}') QUALIFY ROW_NUMBER() OVER ( PARTITION BY RIFER ORDER BY data_first ASC NULLS LAST ) = 1 """ ora_expr = "COALESCE(substr(e.ora_inizio, 1, 2) || ':' || substr(e.ora_inizio, 3, 2), '')" else: em_subq = f""" SELECT CAST(prodotto AS VARCHAR) AS prodotto, rete, data_emissione, CAST(ora_inizio AS VARCHAR) AS ora_inizio, fascia, audience, share FROM read_parquet('{emesso_f}') QUALIFY ROW_NUMBER() OVER ( PARTITION BY CAST(prodotto AS VARCHAR) ORDER BY data_emissione ASC, ora_inizio ASC ) = 1 """ ora_expr = "COALESCE(substr(e.ora_inizio, 1, 5), '')" rows_raw = duckdb.connect().execute(f""" SELECT {gemma_cols}, COALESCE(e.rete, '') AS em_rete, CASE WHEN e.data_emissione IS NOT NULL THEN strftime(CAST(e.data_emissione AS DATE), '%d/%m/%Y') ELSE '' END AS em_data, {ora_expr} AS em_hi, COALESCE(e.fascia, '') AS em_fa, CASE WHEN e.audience IS NOT NULL THEN CAST(CAST(ROUND(e.audience) AS BIGINT) AS VARCHAR) ELSE '' END AS em_aud, CASE WHEN e.share IS NOT NULL THEN printf('%.2f', e.share) ELSE '' END AS em_sha FROM read_parquet('{gemma_f}') g LEFT JOIN ( SELECT riferimento_imdb, CAST(codice AS VARCHAR) AS codice FROM read_parquet('{imdb_f}') WHERE riferimento_imdb IS NOT NULL AND riferimento_imdb != '' QUALIFY ROW_NUMBER() OVER ( PARTITION BY riferimento_imdb ORDER BY codice ) = 1 ) i ON i.riferimento_imdb = g.A_COD_IMDB LEFT JOIN ({em_subq}) e ON e.prodotto = i.codice ORDER BY CAST(g.A_ID AS INTEGER) ASC NULLS LAST """).fetchall() rows: list[list[str]] = [list(raw) for raw in rows_raw] self.finished.emit(rows) except Exception as e: logger.error(f'GemmaSheet _LoadWorker: {e}', exc_info=True) self.finished.emit([]) # ── GemmaSheet ──────────────────────────────────────────────────────────────── class GemmaSheet(QWidget): app_type = 'gemma' def __init__(self, parquet_path: str, parent=None): super().__init__(parent) self._parquet_path = parquet_path self._all_rows: list[list[str]] = [] self._sort_col = -1 self._sort_asc = True self._model = GemmaModel(self) self._load_thread = None self._load_worker = None self._setup_ui() self._start_loading() # ── UI ──────────────────────────────────────────────────────────────── def _setup_ui(self): layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) layout.addWidget(self._build_toolbar()) self.view = QTableView() self.view.setModel(self._model) self.view.setHorizontalHeader(_GemmaHeader(self.view)) self.view.setFont(_CELL_FONT) self.view.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.view.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectItems) self.view.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection) self.view.setAlternatingRowColors(False) self.view.setWordWrap(False) self.view.setItemDelegate(FlatBgDelegate(self.view)) self.view.setItemDelegateForColumn(COL_IMDB, RiferLinkDelegate(self.view, center=True)) vh = SelectableVHeader(self.view) self.view.setVerticalHeader(vh) vh.setVisible(True) self.view.selectionModel().selectionChanged.connect( lambda *_: (vh.viewport().update(), self.view.horizontalHeader().viewport().update()) ) self.view.setHorizontalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel) self.view.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel) self.view.setStyleSheet(""" QTableView { gridline-color: #C0C0C0; } QTableView::item { padding: 0px 3px; } QTableView::item:selected { background-color: #CCE8FF; color: #0078D7; } """) for i, w in enumerate(_WIDTHS): self.view.setColumnWidth(i, w) self.view.doubleClicked.connect(self._on_double_click) self.view.horizontalHeader().sectionClicked.connect(self._on_header_clicked) self.view.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) self.view.customContextMenuRequested.connect(self._on_context_menu) hh = self.view.horizontalHeader() hh.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) hh.customContextMenuRequested.connect(self._on_hheader_context_menu) vh = self.view.verticalHeader() vh.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) vh.customContextMenuRequested.connect(self._on_vheader_context_menu) self._loading_label = QLabel('Caricamento...') self._loading_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self._loading_label.setStyleSheet('font-size:14px;color:#555;background:white;') # FilterBar — creata dopo view perché richiede il widget già inizializzato self._filter_bar = FilterBar(self.view, parent=self) self._filter_bar.filter_changed.connect(self._apply_filters) self._filter_bar.hide() layout.addWidget(self._filter_bar) layout.addWidget(self.view) layout.addWidget(self._loading_label) QShortcut(QKeySequence('Ctrl+F'), self).activated.connect(self._toggle_filter_bar) QShortcut(QKeySequence.StandardKey.Copy, self.view).activated.connect(self._copy_selection) def _build_toolbar(self) -> QWidget: bar = QWidget() bar.setFixedHeight(32) bar.setStyleSheet(f'background-color: {_TOOLBAR_COLOR};') h = QHBoxLayout(bar) h.setContentsMargins(6, 2, 6, 2) h.setSpacing(6) menu = QMenu(bar) act_export = QAction('Esporta XLS', bar) act_export.triggered.connect(self._esporta_sheet) menu.addAction(act_export) menu.addSeparator() self._act_filtro = QAction('Filtro colonne (Ctrl+F)', bar) self._act_filtro.setCheckable(True) self._act_filtro.triggered.connect(lambda c: self._set_filter_bar_visible(c)) menu.addAction(self._act_filtro) btn = QToolButton(bar) btn.setText('≡') btn.setFont(QFont('Calibri', 11)) btn.setFixedSize(30, 26) btn.setStyleSheet( 'QToolButton{background:#2E74B5;color:white;border:1px solid #1F4E79;' 'border-radius:3px;padding:0 4px;}' 'QToolButton:hover{background:#1F5C9A;}' 'QToolButton::menu-indicator{width:0;}') btn.setMenu(menu) btn.setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup) h.addWidget(btn) h.addSpacing(6) self._count_label = QLabel('') self._count_label.setFont(QFont('Calibri', 8)) self._count_label.setStyleSheet('color:#1F3864;') h.addWidget(self._count_label) h.addStretch() return bar # ── Loading ─────────────────────────────────────────────────────────── def _start_loading(self): self.view.hide() self._loading_label.show() self._load_worker = _LoadWorker(self._parquet_path) self._load_thread = QThread() self._load_worker.moveToThread(self._load_thread) self._load_thread.started.connect(self._load_worker.run) self._load_worker.finished.connect(self._on_load_done) self._load_worker.finished.connect(self._load_thread.quit) self._load_worker.finished.connect(self._load_worker.deleteLater) self._load_thread.finished.connect(self._load_thread.deleteLater) self._load_thread.start() def _on_load_done(self, rows: list): self._loading_label.hide() self.view.show() if not rows: self._count_label.setText('0 record') return self._all_rows = rows self._model.set_rows(rows) self._filter_bar._build_inputs() self._filter_bar._update_geometries() self._count_label.setText(f'{len(rows):,} record'.replace(',', '.')) # ── Filtri ──────────────────────────────────────────────────────────── def _toggle_filter_bar(self): self._set_filter_bar_visible(not self._filter_bar.isVisible()) def _set_filter_bar_visible(self, visible: bool): self._filter_bar.setVisible(visible) self._act_filtro.setChecked(visible) if not visible: self._filter_bar.clear_all() @staticmethod def _filter_match(cell: str, f: str) -> bool: if f.startswith('='): return cell.lower() == f[1:] return f in cell.lower() def _apply_filters(self): filters = self._filter_bar.get_filters() # {col_idx: text_lower} if not filters: filtered = self._all_rows else: filtered = [] for row in self._all_rows: if all(self._filter_match(row[col], f) for col, f in filters.items()): filtered.append(row) if self._sort_col >= 0: col = self._sort_col filtered = sorted(filtered, key=lambda r: self._sort_key(r[col]), reverse=not self._sort_asc) self._model.set_rows(filtered) total = len(self._all_rows) shown = len(filtered) if shown == total: self._count_label.setText(f'{total:,} record'.replace(',', '.')) else: self._count_label.setText(f'{shown:,} / {total:,} record'.replace(',', '.')) # ── Sort ───────────────────────────────────────────────────────────── def _on_header_clicked(self, col: int): if self._sort_col == col: self._sort_asc = not self._sort_asc else: self._sort_col = col self._sort_asc = True self.view.horizontalHeader().set_sort(col, self._sort_asc) self._apply_sort() @staticmethod def _sort_key(val: str): import re v = val or '' if not v: return (3, '') # gg/mm/aaaa → aaaa-mm-gg (ordinamento cronologico corretto) m = re.match(r'^(\d{2})/(\d{2})/(\d{4})$', v) if m: return (0, f'{m.group(3)}-{m.group(2)}-{m.group(1)}') # aaaa-mm-gg già ordinabile come stringa if re.match(r'^\d{4}-\d{2}-\d{2}$', v): return (0, v) try: return (1, float(v.replace(',', '.'))) except (ValueError, AttributeError): return (2, v.lower()) def _apply_sort(self): if self._sort_col < 0: return col = self._sort_col rows = sorted(self._model._rows, key=lambda r: self._sort_key(r[col]), reverse=not self._sort_asc) self._model.set_rows(rows) # ── Interazione ─────────────────────────────────────────────────────── def _on_hheader_context_menu(self, pos): col = self.view.horizontalHeader().logicalIndexAt(pos) if col < 0: return sel_cols = sorted({i.column() for i in self.view.selectionModel().selectedIndexes()}) if not sel_cols: sel_cols = [col] label = f'Copia colonna "{_HEADERS[sel_cols[0]]}"' if len(sel_cols) == 1 else f'Copia {len(sel_cols)} colonne' menu = QMenu(self) act = menu.addAction(label) if menu.exec(self.view.horizontalHeader().mapToGlobal(pos)) == act: self._copy_columns(sel_cols) def _on_vheader_context_menu(self, pos): row = self.view.verticalHeader().logicalIndexAt(pos) if row < 0 or row >= len(self._model._rows): return sel_rows = sorted({i.row() for i in self.view.selectionModel().selectedIndexes()}) if not sel_rows: sel_rows = [row] label = 'Copia riga' if len(sel_rows) == 1 else f'Copia {len(sel_rows)} righe' menu = QMenu(self) act = menu.addAction(label) if menu.exec(self.view.verticalHeader().mapToGlobal(pos)) == act: self._copy_rows(sel_rows) def _copy_columns(self, cols: list[int]): rows = self._model._rows header = '\t'.join(_HEADERS[c] for c in cols) lines = [header] + ['\t'.join(row[c] for c in cols) for row in rows] QApplication.clipboard().setText('\n'.join(lines)) def _copy_rows(self, rows: list[int]): data = self._model._rows lines = ['\t'.join(data[r]) for r in rows if r < len(data)] QApplication.clipboard().setText('\n'.join(lines)) def _on_double_click(self, index: QModelIndex): if index.column() != COL_IMDB: return val = self._model.data(index, Qt.ItemDataRole.DisplayRole) or '' if val.strip(): QDesktopServices.openUrl(QUrl(f'https://www.imdb.com/title/{val.strip()}/')) def _on_context_menu(self, pos): menu = QMenu(self) act = menu.addAction('Copia') act.setEnabled(bool(self.view.selectionModel().selectedIndexes())) if menu.exec(self.view.viewport().mapToGlobal(pos)) == act: self._copy_selection() def _copy_selection(self): indexes = self.view.selectionModel().selectedIndexes() if not indexes: return rows = sorted({i.row() for i in indexes}) cols = sorted({i.column() for i in indexes}) lines = [] for r in rows: cells = [self._model.data(self._model.index(r, c), Qt.ItemDataRole.DisplayRole) or '' for c in cols] lines.append('\t'.join(cells)) QApplication.clipboard().setText('\n'.join(lines)) # ── Export ──────────────────────────────────────────────────────────── def _esporta_sheet(self): rows = self._model._rows if not rows: QMessageBox.information(self, 'Esporta', 'Nessun dato da esportare.') return path, _ = QFileDialog.getSaveFileName( self, 'Esporta Gemma', 'gemma_export.xlsx', 'Excel (*.xlsx)' ) if not path: return try: import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter def _hdr_color(col_idx: int) -> str: if col_idx in _VAL_INDICES: return 'C8B0E0' if col_idx in _EM_INDICES: return 'B4EAD7' return 'BDD7EE' _side = Side(style='thin', color='A0A0A0') _border = Border(left=_side, right=_side, top=_side, bottom=_side) _cell_font = Font(name='Calibri', size=8) _hdr_font = Font(name='Calibri', size=8, bold=True) _center = Alignment(horizontal='center', vertical='center') _hdr_align = Alignment(horizontal='center', vertical='center', wrap_text=True) wb = openpyxl.Workbook() ws = wb.active ws.title = 'Gemma' visible = [i for i in range(_N_COLS) if not self.view.isColumnHidden(i)] ws.row_dimensions[1].height = round(33 * 0.75, 2) for ci, col_i in enumerate(visible, 1): cell = ws.cell(row=1, column=ci, value=_HEADERS[col_i]) cell.font = _hdr_font cell.fill = PatternFill('solid', fgColor=_hdr_color(col_i)) cell.alignment = _hdr_align cell.border = _border px = self.view.columnWidth(col_i) ws.column_dimensions[get_column_letter(ci)].width = max(4.0, round(px / 7, 2)) _row_h = round(17 * 0.75, 2) for ri, row in enumerate(rows, 2): ws.row_dimensions[ri].height = _row_h bg_brush = self._model.data( self._model.index(ri - 2, 0), Qt.ItemDataRole.BackgroundRole ) row_fill = None if bg_brush: hex_bg = bg_brush.color().name()[1:].upper() if hex_bg not in ('000000', 'FFFFFF', ''): row_fill = PatternFill('solid', fgColor=hex_bg) for ci, col_i in enumerate(visible, 1): cell = ws.cell(row=ri, column=ci, value=row[col_i]) cell.font = _cell_font cell.alignment = _center cell.border = _border if row_fill: cell.fill = row_fill ws.freeze_panes = 'A2' ws.auto_filter.ref = ws.dimensions wb.save(path) QMessageBox.information(self, 'Esporta', f'File salvato:\n{path}') except Exception as e: QMessageBox.critical(self, 'Errore esportazione', str(e)) logger.error(f'GemmaSheet._esporta_sheet: {e}', exc_info=True)