接受

etl_billiards/gui/widgets/status_panel.py

@@ -0,0 +1,406 @@
+# -*- coding: utf-8 -*-
+"""ETL 状态面板"""
+
+from datetime import datetime
+from typing import Dict, List, Optional, Any
+
+from PySide6.QtWidgets import (
+    QWidget, QVBoxLayout, QHBoxLayout, QGridLayout,
+    QGroupBox, QLabel, QPushButton, QTableWidget, QTableWidgetItem,
+    QHeaderView, QFrame, QScrollArea, QMessageBox
+)
+from PySide6.QtCore import Qt, Signal, QTimer
+from PySide6.QtGui import QColor
+
+from ..workers.db_worker import DBWorker
+from ..utils.config_helper import ConfigHelper
+
+
+class StatusCard(QFrame):
+    """状态卡片"""
+    
+    def __init__(self, title: str, parent=None):
+        super().__init__(parent)
+        self.setProperty("card", True)
+        self.setFrameShape(QFrame.StyledPanel)
+        
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(16, 12, 16, 12)
+        layout.setSpacing(8)
+        
+        # 标题
+        self.title_label = QLabel(title)
+        self.title_label.setProperty("subheading", True)
+        layout.addWidget(self.title_label)
+        
+        # 值
+        self.value_label = QLabel("-")
+        self.value_label.setStyleSheet("font-size: 24px; font-weight: bold;")
+        layout.addWidget(self.value_label)
+        
+        # 描述
+        self.desc_label = QLabel("")
+        self.desc_label.setProperty("subheading", True)
+        layout.addWidget(self.desc_label)
+    
+    def set_value(self, value: str, description: str = "", status: str = ""):
+        """设置值"""
+        self.value_label.setText(value)
+        self.desc_label.setText(description)
+        
+        if status:
+            self.value_label.setProperty("status", status)
+            self.value_label.style().unpolish(self.value_label)
+            self.value_label.style().polish(self.value_label)
+
+
+class StatusPanel(QWidget):
+    """ETL 状态面板"""
+    
+    def __init__(self, parent=None):
+        super().__init__(parent)
+        self.config_helper = ConfigHelper()
+        self.db_worker = DBWorker(self)
+        self._connected = False
+        
+        self._init_ui()
+        self._connect_signals()
+        
+        # 定时刷新
+        self.refresh_timer = QTimer(self)
+        self.refresh_timer.timeout.connect(self._auto_refresh)
+    
+    def _init_ui(self):
+        """初始化界面"""
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(16, 16, 16, 16)
+        layout.setSpacing(16)
+        
+        # 标题和按钮
+        header_layout = QHBoxLayout()
+        
+        title = QLabel("ETL 状态")
+        title.setProperty("heading", True)
+        header_layout.addWidget(title)
+        
+        header_layout.addStretch()
+        
+        self.auto_refresh_btn = QPushButton("自动刷新: 关")
+        self.auto_refresh_btn.setProperty("secondary", True)
+        self.auto_refresh_btn.setCheckable(True)
+        header_layout.addWidget(self.auto_refresh_btn)
+        
+        self.refresh_btn = QPushButton("刷新")
+        self.refresh_btn.clicked.connect(self._refresh_all)
+        header_layout.addWidget(self.refresh_btn)
+        
+        layout.addLayout(header_layout)
+        
+        # 连接状态
+        self.conn_status_label = QLabel("数据库: 未连接")
+        self.conn_status_label.setProperty("status", "warning")
+        layout.addWidget(self.conn_status_label)
+        
+        # 滚动区域
+        scroll_area = QScrollArea()
+        scroll_area.setWidgetResizable(True)
+        scroll_area.setFrameShape(QFrame.NoFrame)
+        layout.addWidget(scroll_area, 1)
+        
+        # 内容容器
+        content_widget = QWidget()
+        content_layout = QVBoxLayout(content_widget)
+        content_layout.setSpacing(16)
+        
+        # 概览卡片
+        cards_layout = QHBoxLayout()
+        
+        self.ods_card = StatusCard("ODS 表数量")
+        cards_layout.addWidget(self.ods_card)
+        
+        self.dwd_card = StatusCard("DWD 表数量")
+        cards_layout.addWidget(self.dwd_card)
+        
+        self.last_update_card = StatusCard("最后更新")
+        cards_layout.addWidget(self.last_update_card)
+        
+        self.task_count_card = StatusCard("今日任务")
+        cards_layout.addWidget(self.task_count_card)
+        
+        content_layout.addLayout(cards_layout)
+        
+        # ODS Cutoff 状态
+        cutoff_group = QGroupBox("ODS Cutoff 状态")
+        cutoff_layout = QVBoxLayout(cutoff_group)
+        
+        self.cutoff_table = QTableWidget()
+        self.cutoff_table.setColumnCount(4)
+        self.cutoff_table.setHorizontalHeaderLabels(["表名", "最新 fetched_at", "行数", "状态"])
+        self.cutoff_table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch)
+        self.cutoff_table.setMaximumHeight(250)
+        cutoff_layout.addWidget(self.cutoff_table)
+        
+        content_layout.addWidget(cutoff_group)
+        
+        # 最近运行记录
+        history_group = QGroupBox("最近运行记录")
+        history_layout = QVBoxLayout(history_group)
+        
+        self.history_table = QTableWidget()
+        self.history_table.setColumnCount(6)
+        self.history_table.setHorizontalHeaderLabels(["运行ID", "任务", "状态", "开始时间", "耗时", "影响行数"])
+        self.history_table.horizontalHeader().setSectionResizeMode(1, QHeaderView.Stretch)
+        self.history_table.setMaximumHeight(250)
+        history_layout.addWidget(self.history_table)
+        
+        content_layout.addWidget(history_group)
+        
+        # 弹性空间
+        content_layout.addStretch()
+        
+        scroll_area.setWidget(content_widget)
+    
+    def _connect_signals(self):
+        """连接信号"""
+        self.auto_refresh_btn.toggled.connect(self._toggle_auto_refresh)
+        self.db_worker.connection_status.connect(self._on_connection_status)
+        self.db_worker.query_finished.connect(self._on_query_finished)
+        self.db_worker.query_error.connect(self._on_query_error)
+    
+    def _toggle_auto_refresh(self, checked: bool):
+        """切换自动刷新"""
+        if checked:
+            self.auto_refresh_btn.setText("自动刷新: 开")
+            self.refresh_timer.start(30000)  # 30秒刷新一次
+            self._refresh_all()
+        else:
+            self.auto_refresh_btn.setText("自动刷新: 关")
+            self.refresh_timer.stop()
+    
+    def _auto_refresh(self):
+        """自动刷新"""
+        if self._connected:
+            self._refresh_all()
+    
+    def _refresh_all(self):
+        """刷新所有数据"""
+        # 尝试连接数据库
+        if not self._connected:
+            env_vars = self.config_helper.load_env()
+            dsn = env_vars.get("PG_DSN", "")
+            if dsn:
+                self.db_worker.connect_db(dsn)
+            else:
+                self.conn_status_label.setText("数据库: 未配置 DSN")
+                return
+        else:
+            self._load_status_data()
+    
+    def _on_connection_status(self, connected: bool, message: str):
+        """处理连接状态"""
+        self._connected = connected
+        
+        if connected:
+            self.conn_status_label.setText(f"数据库: 已连接")
+            self.conn_status_label.setProperty("status", "success")
+            self._load_status_data()
+        else:
+            self.conn_status_label.setText(f"数据库: {message}")
+            self.conn_status_label.setProperty("status", "error")
+        
+        self.conn_status_label.style().unpolish(self.conn_status_label)
+        self.conn_status_label.style().polish(self.conn_status_label)
+    
+    def _load_status_data(self):
+        """加载状态数据"""
+        # 加载表统计
+        self._current_query = "table_count"
+        self.db_worker.execute_query("""
+            SELECT 
+                table_schema,
+                COUNT(*) as table_count
+            FROM information_schema.tables
+            WHERE table_schema IN ('billiards_ods', 'billiards_dwd', 'billiards_dws')
+            GROUP BY table_schema
+        """)
+    
+    def _on_query_finished(self, columns: list, rows: list):
+        """处理查询结果"""
+        query_type = getattr(self, '_current_query', '')
+        
+        if query_type == "table_count":
+            self._process_table_count(rows)
+            # 继续加载 cutoff 数据
+            self._current_query = "cutoff"
+            self.db_worker.execute_query("""
+                SELECT 
+                    'payment_transactions' AS table_name, 
+                    MAX(fetched_at) AS max_fetched_at,
+                    COUNT(*) AS row_count
+                FROM billiards_ods.payment_transactions
+                UNION ALL
+                SELECT 'member_profiles', MAX(fetched_at), COUNT(*)
+                FROM billiards_ods.member_profiles
+                UNION ALL
+                SELECT 'settlement_records', MAX(fetched_at), COUNT(*)
+                FROM billiards_ods.settlement_records
+                UNION ALL
+                SELECT 'recharge_settlements', MAX(fetched_at), COUNT(*)
+                FROM billiards_ods.recharge_settlements
+                UNION ALL
+                SELECT 'assistant_service_records', MAX(fetched_at), COUNT(*)
+                FROM billiards_ods.assistant_service_records
+                ORDER BY table_name
+            """)
+        elif query_type == "cutoff":
+            self._process_cutoff_data(rows)
+            # 继续加载运行历史
+            self._current_query = "history"
+            self.db_worker.execute_query("""
+                SELECT 
+                    run_id,
+                    task_code,
+                    status,
+                    started_at,
+                    finished_at,
+                    rows_affected
+                FROM etl_admin.run_tracker
+                ORDER BY started_at DESC
+                LIMIT 20
+            """)
+        elif query_type == "history":
+            self._process_history_data(rows)
+            self._current_query = ""
+    
+    def _process_table_count(self, rows: list):
+        """处理表数量数据"""
+        ods_count = 0
+        dwd_count = 0
+        
+        for row in rows:
+            schema = row.get("table_schema", "")
+            count = row.get("table_count", 0)
+            
+            if schema == "billiards_ods":
+                ods_count = count
+            elif schema == "billiards_dwd":
+                dwd_count = count
+        
+        self.ods_card.set_value(str(ods_count), "个表")
+        self.dwd_card.set_value(str(dwd_count), "个表")
+    
+    def _process_cutoff_data(self, rows: list):
+        """处理 Cutoff 数据"""
+        self.cutoff_table.setRowCount(len(rows))
+        
+        latest_time = None
+        now = datetime.now()
+        
+        for row_idx, row in enumerate(rows):
+            table_name = row.get("table_name", "")
+            max_fetched = row.get("max_fetched_at")
+            row_count = row.get("row_count", 0)
+            
+            self.cutoff_table.setItem(row_idx, 0, QTableWidgetItem(table_name))
+            
+            if max_fetched:
+                time_str = str(max_fetched)[:19]
+                self.cutoff_table.setItem(row_idx, 1, QTableWidgetItem(time_str))
+                
+                # 更新最新时间
+                if latest_time is None or max_fetched > latest_time:
+                    latest_time = max_fetched
+                
+                # 计算状态
+                if isinstance(max_fetched, datetime):
+                    hours_ago = (now - max_fetched).total_seconds() / 3600
+                    if hours_ago < 1:
+                        status = "正常"
+                        status_color = QColor("#1e8e3e")
+                    elif hours_ago < 24:
+                        status = "较新"
+                        status_color = QColor("#1a73e8")
+                    else:
+                        status = f"落后 {int(hours_ago)}h"
+                        status_color = QColor("#f9ab00")
+                else:
+                    status = "-"
+                    status_color = QColor("#9aa0a6")
+            else:
+                self.cutoff_table.setItem(row_idx, 1, QTableWidgetItem("-"))
+                status = "无数据"
+                status_color = QColor("#d93025")
+            
+            self.cutoff_table.setItem(row_idx, 2, QTableWidgetItem(str(row_count)))
+            
+            status_item = QTableWidgetItem(status)
+            status_item.setForeground(status_color)
+            self.cutoff_table.setItem(row_idx, 3, status_item)
+        
+        # 更新最后更新时间卡片
+        if latest_time:
+            time_str = str(latest_time)[:16]
+            self.last_update_card.set_value(time_str, "")
+        else:
+            self.last_update_card.set_value("-", "无数据")
+    
+    def _process_history_data(self, rows: list):
+        """处理运行历史数据"""
+        self.history_table.setRowCount(len(rows))
+        
+        today_count = 0
+        today = datetime.now().date()
+        
+        for row_idx, row in enumerate(rows):
+            run_id = row.get("run_id", "")
+            task_code = row.get("task_code", "")
+            status = row.get("status", "")
+            started_at = row.get("started_at")
+            finished_at = row.get("finished_at")
+            rows_affected = row.get("rows_affected", 0)
+            
+            # 统计今日任务
+            if started_at and isinstance(started_at, datetime):
+                if started_at.date() == today:
+                    today_count += 1
+            
+            self.history_table.setItem(row_idx, 0, QTableWidgetItem(str(run_id)[:8] if run_id else "-"))
+            self.history_table.setItem(row_idx, 1, QTableWidgetItem(task_code))
+            
+            # 状态
+            status_item = QTableWidgetItem(status)
+            if status and "success" in status.lower():
+                status_item.setForeground(QColor("#1e8e3e"))
+            elif status and ("fail" in status.lower() or "error" in status.lower()):
+                status_item.setForeground(QColor("#d93025"))
+            self.history_table.setItem(row_idx, 2, status_item)
+            
+            # 开始时间
+            time_str = str(started_at)[:19] if started_at else "-"
+            self.history_table.setItem(row_idx, 3, QTableWidgetItem(time_str))
+            
+            # 耗时
+            if started_at and finished_at:
+                try:
+                    duration = (finished_at - started_at).total_seconds()
+                    if duration < 60:
+                        duration_str = f"{duration:.1f}秒"
+                    else:
+                        duration_str = f"{int(duration // 60)}分{int(duration % 60)}秒"
+                except:
+                    duration_str = "-"
+            else:
+                duration_str = "-"
+            self.history_table.setItem(row_idx, 4, QTableWidgetItem(duration_str))
+            
+            # 影响行数
+            self.history_table.setItem(row_idx, 5, QTableWidgetItem(str(rows_affected or 0)))
+        
+        # 更新今日任务卡片
+        self.task_count_card.set_value(str(today_count), "次执行")
+    
+    def _on_query_error(self, error: str):
+        """处理查询错误"""
+        self._current_query = ""
+        # 可能是表不存在，忽略错误继续
+        pass