# -*- coding: utf-8 -*- """数据库查看器""" from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QSplitter, QGroupBox, QLabel, QPushButton, QLineEdit, QPlainTextEdit, QTableWidget, QTableWidgetItem, QTreeWidget, QTreeWidgetItem, QHeaderView, QComboBox, QTabWidget, QMessageBox, QFrame ) from PySide6.QtCore import Qt, Signal from PySide6.QtGui import QFont from ..workers.db_worker import DBWorker from ..utils.config_helper import ConfigHelper # 常用查询模板 QUERY_TEMPLATES = { "ODS 行数统计": """ SELECT table_name, (xpath('/row/cnt/text()', query_to_xml('SELECT COUNT(*) AS cnt FROM ' || table_schema || '.' || table_name, false, false, '')) )[1]::text::bigint AS row_count FROM information_schema.tables WHERE table_schema = 'billiards_ods' ORDER BY table_name; """, "DWD 行数统计": """ SELECT table_name, (xpath('/row/cnt/text()', query_to_xml('SELECT COUNT(*) AS cnt FROM ' || table_schema || '.' || table_name, false, false, '')) )[1]::text::bigint AS row_count FROM information_schema.tables WHERE table_schema = 'billiards_dwd' ORDER BY table_name; """, "ETL 游标状态": """ SELECT task_code, last_start, last_end, last_run_id, updated_at FROM etl_admin.etl_cursor ORDER BY task_code; """, "最近运行记录": """ SELECT run_id, task_code, status, started_at, finished_at, EXTRACT(EPOCH FROM (finished_at - started_at))::int AS duration_sec, rows_affected FROM etl_admin.run_tracker ORDER BY started_at DESC LIMIT 50; """, "ODS 最新入库时间": """ SELECT 'payment_transactions' AS table_name, MAX(fetched_at) AS max_fetched_at FROM billiards_ods.payment_transactions UNION ALL SELECT 'member_profiles', MAX(fetched_at) FROM billiards_ods.member_profiles UNION ALL SELECT 'settlement_records', MAX(fetched_at) FROM billiards_ods.settlement_records UNION ALL SELECT 'recharge_settlements', MAX(fetched_at) FROM billiards_ods.recharge_settlements ORDER BY table_name; """, } class DBViewer(QWidget): """数据库查看器""" # 信号 connection_changed = Signal(bool, str) # 连接状态变化 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._load_dsn_from_env() def _init_ui(self): """初始化界面""" layout = QVBoxLayout(self) layout.setContentsMargins(16, 16, 16, 16) layout.setSpacing(16) # 标题 title = QLabel("数据库查看器") title.setProperty("heading", True) layout.addWidget(title) # 连接配置 conn_group = QGroupBox("数据库连接") conn_layout = QHBoxLayout(conn_group) conn_layout.addWidget(QLabel("DSN:")) self.dsn_edit = QLineEdit() self.dsn_edit.setPlaceholderText("postgresql://user:password@host:5432/dbname") self.dsn_edit.setEchoMode(QLineEdit.Password) conn_layout.addWidget(self.dsn_edit, 1) self.show_dsn_btn = QPushButton("显示") self.show_dsn_btn.setProperty("secondary", True) self.show_dsn_btn.setCheckable(True) self.show_dsn_btn.setFixedWidth(60) conn_layout.addWidget(self.show_dsn_btn) self.connect_btn = QPushButton("连接") self.connect_btn.setFixedWidth(80) conn_layout.addWidget(self.connect_btn) self.disconnect_btn = QPushButton("断开") self.disconnect_btn.setProperty("secondary", True) self.disconnect_btn.setFixedWidth(80) self.disconnect_btn.setEnabled(False) conn_layout.addWidget(self.disconnect_btn) layout.addWidget(conn_group) # 主分割器 main_splitter = QSplitter(Qt.Horizontal) layout.addWidget(main_splitter, 1) # 左侧：表浏览器 left_widget = self._create_table_browser() main_splitter.addWidget(left_widget) # 右侧：查询和结果 right_widget = self._create_query_area() main_splitter.addWidget(right_widget) # 设置分割比例 main_splitter.setSizes([300, 700]) def _create_table_browser(self) -> QWidget: """创建表浏览器""" widget = QWidget() layout = QVBoxLayout(widget) layout.setContentsMargins(0, 0, 8, 0) # 标题和刷新按钮 header_layout = QHBoxLayout() header_layout.addWidget(QLabel("表结构")) self.refresh_tables_btn = QPushButton("刷新") self.refresh_tables_btn.setProperty("secondary", True) self.refresh_tables_btn.setEnabled(False) header_layout.addWidget(self.refresh_tables_btn) layout.addLayout(header_layout) # 表树形视图 self.table_tree = QTreeWidget() self.table_tree.setHeaderLabels(["名称", "行数", "最后更新"]) self.table_tree.header().setSectionResizeMode(0, QHeaderView.Stretch) self.table_tree.setColumnWidth(1, 80) self.table_tree.setColumnWidth(2, 130) layout.addWidget(self.table_tree, 1) return widget def _create_query_area(self) -> QWidget: """创建查询区域""" widget = QWidget() layout = QVBoxLayout(widget) layout.setContentsMargins(8, 0, 0, 0) # 查询输入区 query_group = QGroupBox("SQL 查询") query_layout = QVBoxLayout(query_group) # 模板选择 template_layout = QHBoxLayout() template_layout.addWidget(QLabel("常用查询:")) self.template_combo = QComboBox() self.template_combo.addItem("-- 选择模板 --") for name in QUERY_TEMPLATES.keys(): self.template_combo.addItem(name) template_layout.addWidget(self.template_combo, 1) query_layout.addLayout(template_layout) # SQL 编辑器 self.sql_editor = QPlainTextEdit() self.sql_editor.setObjectName("sqlEditor") self.sql_editor.setPlaceholderText("输入 SQL 查询语句...") self.sql_editor.setFont(QFont("Consolas", 11)) self.sql_editor.setMaximumHeight(150) query_layout.addWidget(self.sql_editor) # 执行按钮 exec_layout = QHBoxLayout() exec_layout.addStretch() self.exec_btn = QPushButton("执行查询 (Ctrl+Enter)") self.exec_btn.setEnabled(False) exec_layout.addWidget(self.exec_btn) query_layout.addLayout(exec_layout) layout.addWidget(query_group) # 结果区域 result_group = QGroupBox("查询结果") result_layout = QVBoxLayout(result_group) # 结果表格 self.result_table = QTableWidget() self.result_table.setAlternatingRowColors(True) self.result_table.horizontalHeader().setStretchLastSection(True) result_layout.addWidget(self.result_table, 1) # 结果统计 self.result_label = QLabel("就绪") self.result_label.setProperty("subheading", True) result_layout.addWidget(self.result_label) layout.addWidget(result_group, 1) return widget def _connect_signals(self): """连接信号""" # 连接按钮 self.show_dsn_btn.toggled.connect(self._toggle_dsn_visibility) self.connect_btn.clicked.connect(self._connect_db) self.disconnect_btn.clicked.connect(self._disconnect_db) self.refresh_tables_btn.clicked.connect(self._refresh_tables) # 模板选择 self.template_combo.currentIndexChanged.connect(self._on_template_selected) # 执行查询 self.exec_btn.clicked.connect(self._execute_query) # 表双击 self.table_tree.itemDoubleClicked.connect(self._on_table_double_clicked) # 工作线程信号 self.db_worker.connection_status.connect(self._on_connection_status) self.db_worker.tables_loaded.connect(self._on_tables_loaded) self.db_worker.query_finished.connect(self._on_query_finished) self.db_worker.query_error.connect(self._on_query_error) def _load_dsn_from_env(self): """从环境变量加载 DSN""" env_vars = self.config_helper.load_env() dsn = env_vars.get("PG_DSN", "") if dsn: self.dsn_edit.setText(dsn) def _toggle_dsn_visibility(self, checked: bool): """切换 DSN 可见性""" self.dsn_edit.setEchoMode( QLineEdit.Normal if checked else QLineEdit.Password ) self.show_dsn_btn.setText("隐藏" if checked else "显示") def _connect_db(self): """连接数据库""" dsn = self.dsn_edit.text().strip() if not dsn: QMessageBox.warning(self, "提示", "请输入数据库连接字符串") return self.connect_btn.setEnabled(False) self.connect_btn.setText("连接中...") self.db_worker.connect_db(dsn) def _disconnect_db(self): """断开数据库连接""" self.db_worker.disconnect_db() def _refresh_tables(self): """刷新表列表""" self.db_worker.load_tables() def _on_connection_status(self, connected: bool, message: str): """处理连接状态变化""" self._connected = connected self.connect_btn.setEnabled(not connected) self.connect_btn.setText("连接") self.disconnect_btn.setEnabled(connected) self.refresh_tables_btn.setEnabled(connected) self.exec_btn.setEnabled(connected) self.connection_changed.emit(connected, message) if connected: # 自动加载表列表 self._refresh_tables() def _on_tables_loaded(self, tables_dict: dict): """处理表列表加载完成""" self.table_tree.clear() for schema, tables in tables_dict.items(): schema_item = QTreeWidgetItem([schema, "", ""]) schema_item.setExpanded(True) for table_name, row_count, updated_at in tables: table_item = QTreeWidgetItem([table_name, str(row_count), updated_at]) table_item.setData(0, Qt.UserRole, f"{schema}.{table_name}") schema_item.addChild(table_item) self.table_tree.addTopLevelItem(schema_item) def _on_template_selected(self, index: int): """模板选择变化""" if index <= 0: return template_name = self.template_combo.currentText() if template_name in QUERY_TEMPLATES: self.sql_editor.setPlainText(QUERY_TEMPLATES[template_name].strip()) # 重置选择 self.template_combo.setCurrentIndex(0) def _on_table_double_clicked(self, item: QTreeWidgetItem, column: int): """表双击事件""" full_name = item.data(0, Qt.UserRole) if full_name: # 生成预览查询 sql = f"SELECT * FROM {full_name} LIMIT 100;" self.sql_editor.setPlainText(sql) self._execute_query() def _execute_query(self): """执行查询""" sql = self.sql_editor.toPlainText().strip() if not sql: QMessageBox.warning(self, "提示", "请输入 SQL 语句") return self.exec_btn.setEnabled(False) self.exec_btn.setText("执行中...") self.result_label.setText("正在查询...") self.db_worker.execute_query(sql) def _on_query_finished(self, columns: list, rows: list): """查询完成""" self.exec_btn.setEnabled(True) self.exec_btn.setText("执行查询 (Ctrl+Enter)") # 更新结果表格 self.result_table.clear() self.result_table.setColumnCount(len(columns)) self.result_table.setRowCount(len(rows)) self.result_table.setHorizontalHeaderLabels(columns) for row_idx, row_data in enumerate(rows): for col_idx, col_name in enumerate(columns): value = row_data.get(col_name, "") item = QTableWidgetItem(str(value) if value is not None else "NULL") if value is None: item.setForeground(Qt.gray) self.result_table.setItem(row_idx, col_idx, item) # 更新统计 self.result_label.setText(f"返回 {len(rows)} 行, {len(columns)} 列") def _on_query_error(self, error: str): """查询错误""" self.exec_btn.setEnabled(True) self.exec_btn.setText("执行查询 (Ctrl+Enter)") self.result_label.setText(f"错误: {error}") QMessageBox.critical(self, "查询错误", error) def close_connection(self): """关闭连接""" if self._connected: self.db_worker.disconnect_db() def keyPressEvent(self, event): """键盘事件""" # Ctrl+Enter 执行查询 if event.modifiers() == Qt.ControlModifier and event.key() == Qt.Key_Return: if self._connected: self._execute_query() else: super().keyPressEvent(event)