#!/usr/bin/env python3 """ TopHub数据查看器 - PySide5界面应用程序 用于显示SQLite数据库中的TopHub抓取数据 """ import sys import os import sqlite3 import webbrowser from datetime import datetime from loguru import logger from PySide6.QtWidgets import ( QApplication, QMainWindow, QTableWidget, QTableWidgetItem, QVBoxLayout, QHBoxLayout, QWidget, QLabel, QLineEdit, QPushButton, QComboBox, QGroupBox, QStatusBar, QMenuBar, QMenu, QMessageBox, QHeaderView, QAbstractItemView, QDialog, QFormLayout, QTextEdit, QInputDialog ) from PySide6.QtCore import Qt, QUrl, QTimer, QEvent from PySide6.QtGui import QAction, QFont, QIcon, QDesktopServices, QClipboard class DatabaseViewer(QMainWindow): """主窗口类，用于显示数据库内容""" def __init__(self): super().__init__() # 获取当前脚本所在目录的数据库文件路径 script_dir = os.path.dirname(os.path.abspath(__file__)) self.db_path = os.path.join(script_dir, "tophub_data.db") # 检查数据库文件是否存在 if not os.path.exists(self.db_path): QMessageBox.critical(self, "错误", f"数据库文件不存在: {self.db_path}") sys.exit(1) self.init_ui() self.load_data() def init_ui(self): """初始化用户界面""" # 设置窗口属性 self.setWindowTitle("TopHub数据查看器") self.setGeometry(100, 100, 1200, 800) # 创建中央部件 central_widget = QWidget() self.setCentralWidget(central_widget) # 创建主布局 main_layout = QVBoxLayout(central_widget) # 创建搜索和筛选区域 filter_group = QGroupBox("搜索和筛选") filter_layout = QHBoxLayout(filter_group) # 搜索框 self.search_edit = QLineEdit() self.search_edit.setPlaceholderText("输入搜索关键词...") self.search_edit.textChanged.connect(self.filter_data) filter_layout.addWidget(QLabel("搜索:")) filter_layout.addWidget(self.search_edit) # 分类筛选 self.category_combo = QComboBox() self.category_combo.addItem("全部分类") self.category_combo.currentTextChanged.connect(self.filter_data) filter_layout.addWidget(QLabel("分类:")) filter_layout.addWidget(self.category_combo) # 刷新按钮 self.refresh_button = QPushButton("刷新数据") self.refresh_button.clicked.connect(self.load_data) filter_layout.addWidget(self.refresh_button) # 批量删除相关控件 self.select_by_keyword_button = QPushButton("按关键字选中") self.select_by_keyword_button.clicked.connect(self.select_by_keyword) filter_layout.addWidget(self.select_by_keyword_button) self.delete_selected_button = QPushButton("删除选中项") self.delete_selected_button.clicked.connect(self.delete_selected_items) filter_layout.addWidget(self.delete_selected_button) # 标记感兴趣按钮 self.mark_interested_button = QPushButton("标记为感兴趣") self.mark_interested_button.clicked.connect(self.mark_as_interested) filter_layout.addWidget(self.mark_interested_button) # 添加筛选区域到主布局 main_layout.addWidget(filter_group) # 创建分类统计显示区域 self.category_stats_group = QGroupBox("分类统计") self.category_stats_layout = QHBoxLayout(self.category_stats_group) self.category_stats_label = QLabel("暂无数据") self.category_stats_layout.addWidget(self.category_stats_label) main_layout.addWidget(self.category_stats_group) # 创建表格 self.table = QTableWidget() self.table.setColumnCount(6) # 保留6列，最后一列显示评分 self.table.setHorizontalHeaderLabels(["ID", "标题", "链接", "分类", "来源日期", "评分"]) # 设置表格属性 self.table.setAlternatingRowColors(True) self.table.setSelectionBehavior(QAbstractItemView.SelectRows) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.table.setSortingEnabled(True) # 设置表格选择模式 self.table.setSelectionMode(QAbstractItemView.SingleSelection) # 设置列宽 header = self.table.horizontalHeader() header.setSectionResizeMode(0, QHeaderView.ResizeToContents) # ID列 header.setSectionResizeMode(1, QHeaderView.Interactive) # 标题列 - 允许用户调整 self.table.setColumnWidth(1, 400) # 设置标题列默认宽度为400像素 header.setSectionResizeMode(2, QHeaderView.ResizeToContents) # 链接列 header.setSectionResizeMode(3, QHeaderView.ResizeToContents) # 分类列 header.setSectionResizeMode(4, QHeaderView.ResizeToContents) # 时间列 header.setSectionResizeMode(5, QHeaderView.ResizeToContents) # 评分列 # 启用链接点击 self.table.cellClicked.connect(self.on_cell_clicked) # 安装事件过滤器以处理链接点击 self.table.viewport().installEventFilter(self) # 启用右键菜单 self.table.setContextMenuPolicy(Qt.CustomContextMenu) self.table.customContextMenuRequested.connect(self.show_context_menu) # 添加表格到主布局 main_layout.addWidget(self.table) # 创建状态栏 self.status_bar = QStatusBar() self.setStatusBar(self.status_bar) # 创建菜单栏 self.create_menu_bar() def create_menu_bar(self): """创建菜单栏""" menubar = self.menuBar() # 文件菜单 file_menu = menubar.addMenu("文件") # 刷新动作 refresh_action = QAction("刷新数据", self) refresh_action.setShortcut("F5") refresh_action.triggered.connect(self.load_data) file_menu.addAction(refresh_action) # 退出动作 exit_action = QAction("退出", self) exit_action.setShortcut("Ctrl+Q") exit_action.triggered.connect(self.close) file_menu.addAction(exit_action) # 帮助菜单 help_menu = menubar.addMenu("帮助") # 关于动作 about_action = QAction("关于", self) about_action.triggered.connect(self.show_about) help_menu.addAction(about_action) def load_data(self): """从数据库加载数据""" try: # 连接数据库 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 检查表是否存在 cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='articles'") if not cursor.fetchone(): QMessageBox.critical(self, "错误", "数据库中不存在articles表") conn.close() return # 查询数据 - 修改为查询score字段而不是is_interested cursor.execute(''' SELECT id, title, url, category, source_date, score FROM articles ORDER BY id DESC ''') rows = cursor.fetchall() conn.close() # 更新表格 self.table.setRowCount(len(rows)) # 获取所有分类和统计信息 categories = set() category_counts = {} # 用于存储每个分类的数量 for row_idx, row in enumerate(rows): id_val, title, url, category, source_date, score = row # 添加到分类集合和统计字典 if category: categories.add(category) category_counts[category] = category_counts.get(category, 0) + 1 else: # 处理空分类的情况 category_counts["未分类"] = category_counts.get("未分类", 0) + 1 # 设置表格项 self.table.setItem(row_idx, 0, QTableWidgetItem(str(id_val))) self.table.setItem(row_idx, 1, QTableWidgetItem(title)) # 链接项 - 设置为蓝色并加下划线 link_item = QTableWidgetItem(url if url else "") if url: link_item.setForeground(Qt.blue) link_item.setFont(QFont("", -1, QFont.Bold)) self.table.setItem(row_idx, 2, link_item) self.table.setItem(row_idx, 3, QTableWidgetItem(category if category else "未分类")) self.table.setItem(row_idx, 4, QTableWidgetItem(source_date)) # 感兴趣状态项 score_item = QTableWidgetItem(str(score)) # 根据分数设置颜色 if score >= 8: score_item.setForeground(Qt.green) score_item.setFont(QFont("", -1, QFont.Bold)) elif score >= 6: score_item.setForeground(Qt.blue) elif score <= 3: score_item.setForeground(Qt.red) self.table.setItem(row_idx, 5, score_item) # 更新分类下拉框 current_category = self.category_combo.currentText() self.category_combo.clear() self.category_combo.addItem("全部分类") for cat in sorted(categories): self.category_combo.addItem(cat) # 恢复之前选择的分类 index = self.category_combo.findText(current_category) if index >= 0: self.category_combo.setCurrentIndex(index) # 更新分类统计显示 self.update_category_stats(category_counts) # 更新状态栏 self.status_bar.showMessage(f"已加载 {len(rows)} 条记录") except sqlite3.Error as e: logger.error(f"数据库操作出错: {str(e)}") QMessageBox.critical(self, "数据库错误", f"数据库操作出错: {str(e)}") self.status_bar.showMessage("加载数据失败") except Exception as e: logger.error(f"加载数据时出错: {str(e)}") QMessageBox.critical(self, "错误", f"加载数据时出错: {str(e)}") self.status_bar.showMessage("加载数据失败") def update_category_stats(self, category_counts): """更新分类统计显示""" if not category_counts: self.category_stats_label.setText("暂无数据") return # 按数量降序排列分类 sorted_categories = sorted(category_counts.items(), key=lambda x: x[1], reverse=True) # 构建统计信息文本 stats_text = " | ".join([f"{category}: {count}" for category, count in sorted_categories]) # 如果文本过长，进行截断并添加提示 if len(stats_text) > 200: stats_text = stats_text[:200] + "... (更多分类请查看完整数据)" self.category_stats_label.setText(stats_text) self.category_stats_label.setToolTip(" | ".join([f"{category}: {count}" for category, count in sorted_categories])) def update_category_stats_after_filter(self): """在筛选后更新分类统计显示""" # 统计可见行的分类 category_counts = {} for row in range(self.table.rowCount()): # 跳过隐藏的行 if self.table.isRowHidden(row): continue # 获取分类项 category_item = self.table.item(row, 3) if category_item: category = category_item.text() category_counts[category] = category_counts.get(category, 0) + 1 else: category_counts["未分类"] = category_counts.get("未分类", 0) + 1 # 更新分类统计显示 self.update_category_stats(category_counts) def filter_data(self): """根据搜索条件和分类筛选数据""" search_text = self.search_edit.text().lower() selected_category = self.category_combo.currentText() # 遍历所有行 for row in range(self.table.rowCount()): show_row = True # 检查搜索条件 if search_text: text_match = False for col in range(1, 6): # 检查标题、链接、分类、日期、感兴趣列 item = self.table.item(row, col) if item and search_text in item.text().lower(): text_match = True break show_row = show_row and text_match # 检查分类条件 if selected_category != "全部分类": category_item = self.table.item(row, 3) category_match = category_item and category_item.text() == selected_category show_row = show_row and category_match # 显示或隐藏行 self.table.setRowHidden(row, not show_row) # 计算可见行数 visible_count = sum(1 for row in range(self.table.rowCount()) if not self.table.isRowHidden(row)) self.status_bar.showMessage(f"显示 {visible_count}/{self.table.rowCount()} 条记录") # 重新计算并显示分类统计 self.update_category_stats_after_filter() def eventFilter(self, obj, event): """事件过滤器，用于处理链接点击而不触发行选择""" if obj == self.table.viewport() and event.type() == QEvent.MouseButtonPress: # 获取点击位置 pos = event.position() # 获取点击位置的行和列 row = self.table.rowAt(int(pos.y())) column = self.table.columnAt(int(pos.x())) # 如果点击的是链接列（第2列，索引为2） if column == 2 and row >= 0: item = self.table.item(row, column) if item and item.text() and item.text().startswith("http"): # 直接打开链接 webbrowser.open(item.text()) # 返回True表示事件已处理，不再传递给原始处理器 # 这样就不会触发行选择，避免鼠标跳动 return True # 其他事件交给原始处理器处理 return super().eventFilter(obj, event) def on_cell_clicked(self, row, column): """处理单元格点击事件""" # 链接列的点击已经由eventFilter处理，这里不再处理 # 只处理非链接列的点击，保持原有选择行为 if column != 2: # 可以在这里添加其他列的点击处理逻辑 pass def show_context_menu(self, position): """显示右键菜单""" # 获取点击位置的行 row = self.table.rowAt(position.y()) if row < 0: return # 选中该行 self.table.selectRow(row) # 创建右键菜单 menu = QMenu(self) # 添加"增加评分(+1)"动作 increase_score_action = QAction("增加评分(+1)", self) increase_score_action.triggered.connect(self.increase_score) menu.addAction(increase_score_action) # 添加"减少评分(-1)"动作 decrease_score_action = QAction("减少评分(-1)", self) decrease_score_action.triggered.connect(self.decrease_score) menu.addAction(decrease_score_action) # 添加分隔线 menu.addSeparator() # 添加"复制信息"动作 copy_info_action = QAction("复制信息", self) copy_info_action.triggered.connect(self.copy_info) menu.addAction(copy_info_action) # 添加分隔线 menu.addSeparator() # 添加"删除"动作 delete_action = QAction("删除选中项", self) delete_action.triggered.connect(self.delete_selected_items) menu.addAction(delete_action) # 显示菜单 menu.exec_(self.table.mapToGlobal(position)) def copy_info(self): """复制选中行的标题、链接、日期等信息""" # 获取选中的行 selected_rows = set() for item in self.table.selectedItems(): selected_rows.add(item.row()) # 如果没有选中的行，直接返回 if not selected_rows: QMessageBox.information(self, "提示", "请先选中要复制信息的行") return # 收集所有选中行的信息 all_info = [] for row in sorted(selected_rows): # 获取标题、链接、日期 title_item = self.table.item(row, 1) url_item = self.table.item(row, 2) date_item = self.table.item(row, 4) title = title_item.text() if title_item else "" url = url_item.text() if url_item else "" date = date_item.text() if date_item else "" # 按照要求的格式组合信息："日期 标题\n链接" info = f"{date} {title}\n{url}".strip() all_info.append(info) # 将所有信息用换行符连接 clipboard_text = "\n".join(all_info) # 复制到剪贴板 clipboard = QApplication.clipboard() clipboard.setText(clipboard_text) # 更新状态栏 self.status_bar.showMessage(f"已复制 {len(selected_rows)} 行信息到剪贴板") def show_about(self): """显示关于对话框""" about_text = """

TopHub数据查看器

版本: 1.0

用于查看TopHub网站抓取数据的PySide5应用程序

功能特性:

""" QMessageBox.about(self, "关于", about_text) def select_by_keyword(self): """按关键字选中行""" # 弹出输入对话框获取关键字 keyword, ok = QInputDialog.getText(self, "按关键字选中", "请输入关键字:") if not ok or not keyword: return keyword = keyword.lower() selected_count = 0 # 遍历所有可见行 for row in range(self.table.rowCount()): # 跳过隐藏的行 if self.table.isRowHidden(row): continue # 检查该行是否包含关键字 match = False for col in range(self.table.columnCount()): item = self.table.item(row, col) if item and keyword in item.text().lower(): match = True break # 如果匹配，则选中该行 if match: self.table.selectRow(row) selected_count += 1 # 更新状态栏 self.status_bar.showMessage(f"已选中 {selected_count} 行") def delete_selected_items(self): """删除选中的项目""" # 获取选中的行 selected_rows = set() for item in self.table.selectedItems(): selected_rows.add(item.row()) # 如果没有选中的行，直接返回 if not selected_rows: QMessageBox.information(self, "提示", "请先选中要删除的行") return # 弹出确认对话框 reply = QMessageBox.question( self, "确认删除", f"确定要删除选中的 {len(selected_rows)} 行数据吗？此操作不可撤销！", QMessageBox.Yes | QMessageBox.No, QMessageBox.No ) if reply == QMessageBox.No: return try: # 连接数据库 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 删除选中的行 deleted_count = 0 for row in sorted(selected_rows, reverse=True): # 从后往前删除，避免索引变化 # 获取ID id_item = self.table.item(row, 0) if id_item: article_id = id_item.text() # 从数据库中删除 cursor.execute("DELETE FROM articles WHERE id = ?", (article_id,)) # 从表格中移除行 self.table.removeRow(row) deleted_count += 1 # 提交更改 conn.commit() conn.close() # 更新状态栏 self.status_bar.showMessage(f"已删除 {deleted_count} 行数据") # 重新加载数据以更新分类统计 self.load_data() except sqlite3.Error as e: logger.error(f"删除数据时出错: {str(e)}") QMessageBox.critical(self, "数据库错误", f"删除数据时出错: {str(e)}") self.status_bar.showMessage("删除失败") except Exception as e: logger.error(f"删除数据时出错: {str(e)}") QMessageBox.critical(self, "错误", f"删除数据时出错: {str(e)}") self.status_bar.showMessage("删除失败") def mark_as_interested(self): """将选中的项目标记为感兴趣""" # 获取选中的行 selected_rows = set() for item in self.table.selectedItems(): selected_rows.add(item.row()) # 如果没有选中的行，直接返回 if not selected_rows: QMessageBox.information(self, "提示", "请先选中要标记的行") return # 弹出确认对话框 reply = QMessageBox.question( self, "确认标记", f"确定要将选中的 {len(selected_rows)} 行标记为感兴趣吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes ) if reply == QMessageBox.No: return try: # 连接数据库 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 更新选中的行 updated_count = 0 for row in selected_rows: # 获取ID id_item = self.table.item(row, 0) if id_item: article_id = id_item.text() # 更新数据库中的is_interested字段 cursor.execute("UPDATE articles SET is_interested = 1 WHERE id = ?", (article_id,)) # 更新表格中的显示 interested_item = QTableWidgetItem("是") interested_item.setForeground(Qt.green) interested_item.setFont(QFont("", -1, QFont.Bold)) self.table.setItem(row, 5, interested_item) updated_count += 1 # 提交更改 conn.commit() conn.close() # 更新状态栏 self.status_bar.showMessage(f"已标记 {updated_count} 行为感兴趣") except sqlite3.Error as e: logger.error(f"标记数据时出错: {str(e)}") QMessageBox.critical(self, "数据库错误", f"标记数据时出错: {str(e)}") self.status_bar.showMessage("标记失败") except Exception as e: logger.error(f"标记数据时出错: {str(e)}") QMessageBox.critical(self, "错误", f"标记数据时出错: {str(e)}") self.status_bar.showMessage("标记失败") def mark_as_not_interested(self): """将选中的项目标记为不感兴趣""" # 获取选中的行 selected_rows = set() for item in self.table.selectedItems(): selected_rows.add(item.row()) # 如果没有选中的行，直接返回 if not selected_rows: QMessageBox.information(self, "提示", "请先选中要标记的行") return # 弹出确认对话框 reply = QMessageBox.question( self, "确认标记", f"确定要将选中的 {len(selected_rows)} 行标记为不感兴趣吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes ) if reply == QMessageBox.No: return try: # 连接数据库 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 更新选中的行 updated_count = 0 for row in selected_rows: # 获取ID id_item = self.table.item(row, 0) if id_item: article_id = id_item.text() # 更新数据库中的is_interested字段 cursor.execute("UPDATE articles SET is_interested = 0 WHERE id = ?", (article_id,)) # 更新表格中的显示 interested_item = QTableWidgetItem("否") # 不感兴趣项使用普通字体和颜色 self.table.setItem(row, 5, interested_item) updated_count += 1 # 提交更改 conn.commit() conn.close() # 更新状态栏 self.status_bar.showMessage(f"已标记 {updated_count} 行为不感兴趣") except sqlite3.Error as e: logger.error(f"标记数据时出错: {str(e)}") QMessageBox.critical(self, "数据库错误", f"标记数据时出错: {str(e)}") self.status_bar.showMessage("标记失败") except Exception as e: logger.error(f"标记数据时出错: {str(e)}") QMessageBox.critical(self, "错误", f"标记数据时出错: {str(e)}") self.status_bar.showMessage("标记失败") def increase_score(self): """增加选中项目的评分(+1)""" # 获取选中的行 selected_rows = set() for item in self.table.selectedItems(): selected_rows.add(item.row()) # 如果没有选中的行，直接返回 if not selected_rows: QMessageBox.information(self, "提示", "请先选中要增加评分的行") return try: # 连接数据库 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 更新选中的行 updated_count = 0 for row in selected_rows: # 获取ID id_item = self.table.item(row, 0) if id_item: article_id = id_item.text() # 获取当前分数 cursor.execute("SELECT score FROM articles WHERE id = ?", (article_id,)) result = cursor.fetchone() if result: current_score = result[0] # 增加分数，但不超过10 new_score = min(current_score + 1, 10) # 更新数据库中的score字段 cursor.execute("UPDATE articles SET score = ? WHERE id = ?", (new_score, article_id)) # 更新表格中的显示 score_item = QTableWidgetItem(str(new_score)) # 根据分数设置颜色 if new_score >= 8: score_item.setForeground(Qt.green) score_item.setFont(QFont("", -1, QFont.Bold)) elif new_score >= 6: score_item.setForeground(Qt.blue) elif new_score <= 3: score_item.setForeground(Qt.red) self.table.setItem(row, 5, score_item) updated_count += 1 # 提交更改 conn.commit() conn.close() # 更新状态栏 self.status_bar.showMessage(f"已增加 {updated_count} 行的评分") except sqlite3.Error as e: logger.error(f"增加评分时出错: {str(e)}") QMessageBox.critical(self, "数据库错误", f"增加评分时出错: {str(e)}") self.status_bar.showMessage("增加评分失败") except Exception as e: logger.error(f"增加评分时出错: {str(e)}") QMessageBox.critical(self, "错误", f"增加评分时出错: {str(e)}") self.status_bar.showMessage("增加评分失败") def decrease_score(self): """减少选中项目的评分(-1)""" # 获取选中的行 selected_rows = set() for item in self.table.selectedItems(): selected_rows.add(item.row()) # 如果没有选中的行，直接返回 if not selected_rows: QMessageBox.information(self, "提示", "请先选中要减少评分的行") return try: # 连接数据库 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 更新选中的行 updated_count = 0 for row in selected_rows: # 获取ID id_item = self.table.item(row, 0) if id_item: article_id = id_item.text() # 获取当前分数 cursor.execute("SELECT score FROM articles WHERE id = ?", (article_id,)) result = cursor.fetchone() if result: current_score = result[0] # 减少分数，但不低于0 new_score = max(current_score - 1, 0) # 更新数据库中的score字段 cursor.execute("UPDATE articles SET score = ? WHERE id = ?", (new_score, article_id)) # 更新表格中的显示 score_item = QTableWidgetItem(str(new_score)) # 根据分数设置颜色 if new_score >= 8: score_item.setForeground(Qt.green) score_item.setFont(QFont("", -1, QFont.Bold)) elif new_score >= 6: score_item.setForeground(Qt.blue) elif new_score <= 3: score_item.setForeground(Qt.red) self.table.setItem(row, 5, score_item) updated_count += 1 # 提交更改 conn.commit() conn.close() # 更新状态栏 self.status_bar.showMessage(f"已减少 {updated_count} 行的评分") except sqlite3.Error as e: logger.error(f"减少评分时出错: {str(e)}") QMessageBox.critical(self, "数据库错误", f"减少评分时出错: {str(e)}") self.status_bar.showMessage("减少评分失败") except Exception as e: logger.error(f"减少评分时出错: {str(e)}") QMessageBox.critical(self, "错误", f"减少评分时出错: {str(e)}") self.status_bar.showMessage("减少评分失败") def main(): """主函数""" app = QApplication(sys.argv) # 设置应用程序属性 app.setApplicationName("TopHub数据查看器") app.setOrganizationName("TopHub") # 创建并显示主窗口 viewer = DatabaseViewer() viewer.show() # 运行应用程序 sys.exit(app.exec()) if __name__ == "__main__": main()