import logging import socket import sys import pymysql from PySide6 import QtWidgets from PySide6.QtWidgets import QApplication, QMainWindow, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, \ QTextEdit, QPushButton, QHBoxLayout, QLineEdit, QMessageBox, QSizePolicy class TableWidgetExample(QMainWindow): def __init__(self): super().__init__() self.logger = None self.table = None self.button_connection = None self.connection_pwd = None self.connection_user = None self.connection_dbname = None self.connection_port = None self.connection_host = None self.input_text = None self.connection = None self.cursor = None self.is_connect = False # self.button_add = None # self.button_execute = None # self.input_text = None # self.connection_host = None # self.connection_port = None # self.connection_user = None # self.connection_pwd = None self.initLogger() self.initUI() def initUI(self): self.setWindowTitle("采集") self.setGeometry(100, 100, 800, 600) central_widget = QWidget() center_layout = QVBoxLayout(central_widget) self.setCentralWidget(central_widget) top_widget = QWidget() top_layout = QVBoxLayout(top_widget) self.table = QTableWidget() self.table.setColumnCount(2) # 设置水平表头标签 self.table.setHorizontalHeaderLabels(["URL", "状态"]) self.table.setEditTriggers(QTableWidget.NoEditTriggers) header = self.table.horizontalHeader() # header.setSectionResizeMode(QtWidgets.QHeaderView.ResizeToContents) header.setSectionResizeMode(QtWidgets.QHeaderView.Stretch) top_layout.addWidget(self.table) center_layout.addWidget(top_widget) # 下部分布局 bottom_widget = QWidget() bottom_widget.setFixedHeight(200) bottom_layout = QVBoxLayout(bottom_widget) self.input_text = QTextEdit() self.input_text.setPlaceholderText("请输入链接，一行一个") bottom_layout.addWidget(self.input_text) connection_layout = QHBoxLayout() self.connection_host = QLineEdit() self.connection_host.setPlaceholderText("地址") self.connection_port = QLineEdit() self.connection_port.setPlaceholderText("端口") self.connection_dbname = QLineEdit() self.connection_dbname.setPlaceholderText("数据库") self.connection_user = QLineEdit() self.connection_user.setPlaceholderText("用户名") self.connection_pwd = QLineEdit() self.connection_pwd.setPlaceholderText("密码") self.connection_pwd.setEchoMode(QLineEdit.Password) self.button_connection = QPushButton("连接") self.button_connection.clicked.connect(self.connect_to_database) connection_layout.addWidget(self.connection_host) connection_layout.addWidget(self.connection_port) connection_layout.addWidget(self.connection_dbname) connection_layout.addWidget(self.connection_user) connection_layout.addWidget(self.connection_pwd) connection_layout.addWidget(self.button_connection) bottom_layout.addLayout(connection_layout) action_layout = QHBoxLayout() button_add = QPushButton("添加链接") button_add.clicked.connect(self.add_data_to_table) action_layout.addWidget(button_add) button_execute = QPushButton("开始采集") action_layout.addWidget(button_execute) bottom_layout.addLayout(action_layout) center_layout.addWidget(bottom_widget) def initLogger(self): logging.basicConfig(level=logging.DEBUG) self.logger = logging.getLogger("dmuk") def connect_to_database(self): # 数据库基本信息 host = self.connection_host.text() port = self.connection_port.text() user = self.connection_user.text() pwd = self.connection_pwd.text() dbname = self.connection_dbname.text() if not host or not port or not dbname or not user or not pwd: self.statusBar().showMessage("主机、端口、数据库名、用户名和密码不能为空!") QMessageBox.critical(self, "Error", "主机、端口、数据库名、用户名和密码不能为空!") return try: socket.gethostbyname(host) except socket.error: self.statusBar().showMessage("无效的主机!") QMessageBox.critical(self, "Error", "无效的主机!") return else: # 主机有效 print("有效的主机") # 校验端口 try: port = int(port) if 0 < port <= 65535: print("有效的端口") else: QMessageBox.critical(self, "Error", "端口超出范围!") self.statusBar().showMessage("端口超出范围!") return except ValueError: QMessageBox.critical(self, "Error", "无效的端口!") self.statusBar().showMessage("无效的端口!") return try: # 连接数据库 self.connection = pymysql.connect(host=host, port=port, user=user, password=pwd, db=dbname) self.cursor = self.connection.cursor() # 链接成功 self.is_connect = True self.statusBar().showMessage("Connection succees!") except pymysql.Error as e: # 异常 连接失败 self.is_connect = False QMessageBox.critical(self, "Error", str(e)) self.statusBar().showMessage("Connection failed") def closeEvent(self, event): if self.connection: self.connection.close() event.accept() def add_data_to_table(self): self.logger.debug("Add button clicked") # data = [["数据1", "数据2"], # ["数据4", "数据5"]] # self.table.setRowCount(len(data)) # # for row, row_data in enumerate(data): # for col, cell_data in enumerate(row_data): # item = QTableWidgetItem(cell_data) # self.table.setItem(row, col, item) # 从 if __name__ == "__main__": app = QApplication(sys.argv) window = TableWidgetExample() window.show() sys.exit(app.exec())