pyqt5和mysql编写数据库信息查看和删除修改功能
总体效果图
数据库全部信息查看与显示
数据库信息的修改
数据库信息查询
精准查询
模糊查询
删除用户
ui界面设计文件
https://download.csdn.net/download/mao_hui_fei/24710909
源码
'''**************************************************************************
版本:1.0
内容:数据库信息查看和管理界面
时间:2021.9.25
作者:狄云
***************************************************************************'''
import csv
import re
import cv2
import numpy as np
import pymysql
import xlwt as ExcelWrite
from xlwt import Borders, XFStyle, Pattern
from PyQt5.QtCore import pyqtSignal, QThread, Qt, QObject
from PyQt5.QtGui import QIcon, QTextCursor
from PyQt5.QtWidgets import QApplication, QWidget, QMessageBox, QTableWidgetItem, QAbstractItemView, QProgressBar, \
QDialog, QHeaderView
from PyQt5.uic import loadUi
import logging
import logging.config
import os
import shutil
import sys
import threading
import multiprocessing
from datetime import datetime
class RecordNotFound(Exception):
pass
class DataManageUI(QWidget):
logQueue = multiprocessing.Queue()
receiveLogSignal = pyqtSignal(str)
sql_name_map = ('id', 'name', 'pwd')
def __init__(self):
super(DataManageUI, self).__init__()
loadUi('./ui/DataManage.ui', self)
self.setWindowIcon(QIcon('./icons/icon.png'))
self.tableWidget.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.tableWidget.setAlternatingRowColors(True)
self.database = 'userinfo'
self.datasets = './datasets'
self.isDbReady = False
self.current_select = set()
self.initDbButton.clicked.connect(self.initDb)
self.tableWidget.itemClicked.connect(self.enable_delete_button)
self.queryUserButton.clicked.connect(self.queryUser)
self.deleteUserButton.clicked.connect(self.deleteUser)
self.CellChangeButton.clicked.connect(self.modify_line)
self.ExportExcelpushButton.clicked.connect(self.check_table)
self.receiveLogSignal.connect(lambda log: self.logOutput(log))
self.logOutputThread = threading.Thread(target=self.receiveLog, daemon=True)
self.logOutputThread.start()
self.enable_like_select = False
self.LikeSelectCheckBox.stateChanged.connect(
lambda: self.is_like_select(self.LikeSelectCheckBox))
def is_like_select(self, like_select_checkbox):
if like_select_checkbox.isChecked():
self.enable_like_select = True
else:
self.enable_like_select = False
@staticmethod
def table_exists(cur, table_name):
print('进入检查table是否存在函数')
sql = "show tables;"
cur.execute(sql)
tables = [cur.fetchall()]
print('显示数据库中所有表tables=',tables)
table_list = re.findall('(\'.*?\')', str(tables))
table_list = [re.sub("'", '', each) for each in table_list]
if table_name in table_list:
return True
else:
return False
def check_table(self):
self.export_excel_dialog = ExportExcelDialog()
self.export_excel_dialog.exec()
def cell_change(self, row, col):
print("进入数据修改函数")
try:
conn, cursor = self.connect_to_sql()
print("进入数据修改函数,数据库连接成功")
if not self.table_exists(cursor, self.database):
raise FileNotFoundError
item = self.tableWidget.item(row, col)
print("进入数据修改函数,item=",item)
stu_id = self.tableWidget.item(row, 0).text()
print("进入数据修改函数,stu_id=", stu_id)
after_change_txt = item.text()
print("进入数据修改函数,after_change_txt=", after_change_txt)
select_sql = 'SELECT * FROM userinfo WHERE id=%s' % stu_id
cursor.execute(select_sql)
ret = cursor.fetchall()
if not ret:
raise RecordNotFound
else:
before_change_txt = ret[0][col]
text = '确定将原数据<font color=blue> {} </font>修改为<font color=green> {} </font> 吗?<font color=red>该操作不可逆!</font>'.format(
before_change_txt, after_change_txt)
informativeText = '<b>是否继续?</b>'
ret = self.callDialog(QMessageBox.Warning, text, informativeText, QMessageBox.Yes | QMessageBox.No,
QMessageBox.No)
if ret == QMessageBox.Yes:
update_sql = 'UPDATE userinfo SET %s="%s" WHERE id=%s' % (
self.sql_name_map[col], after_change_txt, stu_id)
cursor.execute(update_sql)
self.logQueue.put('修改成功!')
else:
if self.CellChangeButton.text() == '禁用编辑':
self.tableWidget.cellChanged.disconnect()
self.tableWidget.setItem(row, col, QTableWidgetItem(str(before_change_txt)))
if self.CellChangeButton.text() == '禁用编辑':
self.enable_write_table()
except FileNotFoundError:
logging.error('系统找不到数据库表{}'.format(self.database))
self.isDbReady = False
self.initDbButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:未发现数据库,你可能未进行人脸采集')
except Exception as e:
print(e)
logging.error('读取数据库异常,无法完成数据库初始化')
self.isDbReady = False
self.initDbButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:读取数据库异常,初始化/刷新数据库失败')
else:
cursor.close()
conn.commit()
conn.close()
@staticmethod
def connect_to_sql():
try:
conn = pymysql.connect(host='localhost',
user='root',
password='root',
db='test_0505',
port=3306,
charset='utf8')
cursor = conn.cursor()
print('数据库连接成功!')
return conn, cursor
except pymysql.Error as e:
print('数据库连接失败Error: %s' % e)
def modify_line(self):
if self.CellChangeButton.text() == '禁用编辑':
self.tableWidget.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.CellChangeButton.setText('启用编辑')
self.enable_select_table()
else:
self.deleteUserButton.setEnabled(False)
self.tableWidget.setEditTriggers(QAbstractItemView.DoubleClicked)
self.CellChangeButton.setText('禁用编辑')
self.enable_write_table()
def enable_write_table(self):
row_count = self.tableWidget.rowCount()
for row in range(row_count):
self.tableWidget.item(row, 0).setFlags(Qt.ItemIsEnabled)
self.tableWidget.item(row, 1).setFlags(Qt.ItemIsEnabled)
self.tableWidget.cellChanged.connect(self.cell_change)
def enable_select_table(self):
self.tableWidget.cellChanged.disconnect()
row_count = self.tableWidget.rowCount()
for row in range(row_count):
self.tableWidget.item(row, 0).setFlags(Qt.ItemIsSelectable | Qt.ItemIsEnabled)
self.tableWidget.item(row, 1).setFlags(Qt.ItemIsSelectable | Qt.ItemIsEnabled)
def enable_delete_button(self, item):
self.current_select.clear()
select_items = self.tableWidget.selectedItems()[::self.tableWidget.columnCount()]
self.current_select.update(
map(lambda x: x.text(), select_items))
if self.current_select and self.CellChangeButton.text() != '禁用编辑':
self.deleteUserButton.setEnabled(True)
else:
self.deleteUserButton.setEnabled(False)
def print_to_table(self, stu_data):
print("进入print_to_table函数")
if self.CellChangeButton.text() == '禁用编辑':
self.tableWidget.cellChanged.disconnect()
print("进入print_to_table函数 281")
while self.tableWidget.rowCount() > 0:
self.tableWidget.removeRow(0)
print("进入print_to_table函数 286")
for row_index, row_data in enumerate(stu_data):
print("进入print_to_table函数 row_data=",row_data)
self.tableWidget.insertRow(row_index)
for col_index, col_data in enumerate(row_data):
self.tableWidget.setItem(row_index, col_index, QTableWidgetItem(str(col_data)))
if self.CellChangeButton.text() == '禁用编辑':
self.enable_write_table()
print("输出结果至界面表格成功")
def initDb(self):
try:
conn, cursor = self.connect_to_sql()
if not self.table_exists(cursor, self.database):
print('检查table不存在:' )
raise FileNotFoundError
print('检查table,存在:')
cursor.execute('SELECT * FROM userinfo')
conn.commit()
stu_data = cursor.fetchall()
print("stu_data=",stu_data)
self.print_to_table(stu_data)
cursor.execute('SELECT Count(*) FROM userinfo')
result = cursor.fetchone()
print("result=",result)
dbUserCount = result[0]
print("dbUserCount=",dbUserCount)
except FileNotFoundError:
logging.error('系统找不到数据库表{}'.format(self.database))
self.isDbReady = False
self.initDbButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:初始化时未发现数据库,你可能未进行人脸采集')
except Exception as e:
print(e)
logging.error('读取数据库异常,无法完成数据库初始化')
self.isDbReady = False
self.initDbButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:读取数据库异常,初始化/刷新数据库失败')
else:
cursor.close()
conn.close()
print("人数显示=", dbUserCount)
self.dbUserCountLcdNum.display(dbUserCount)
print("人数显示329=", dbUserCount)
if not self.isDbReady:
self.isDbReady = True
print("人数显示 332=", dbUserCount)
self.logQueue.put('Success:数据库初始化完成,发现用户数:{}'.format(dbUserCount))
print("人数显示 334=", dbUserCount)
self.initDbButton.setText('刷新数据库')
self.initDbButton.setIcon(QIcon('./icons/success.png'))
self.queryUserButton.setToolTip('')
self.queryUserButton.setEnabled(True)
self.CellChangeButton.setToolTip('')
self.CellChangeButton.setEnabled(True)
self.deleteUserButton.setToolTip('')
self.ExportExcelpushButton.setEnabled(True)
else:
self.logQueue.put('Success:刷新数据库成功,发现用户数:{}'.format(dbUserCount))
def queryUser(self):
print("进入用户查询函数")
select_data = dict()
select_data['id'] = self.querystuIDLineEdit.text().strip()
select_data['name'] = self.queryNameLineEdit.text().strip()
select_data['pwd'] = self.queryenNameLineEdit.text().strip()
print("进入用户查询函数",select_data['id'])
conn, cursor = self.connect_to_sql()
try:
select_sql = 'SELECT * FROM userinfo WHERE 1=1'
for key, value in select_data.items():
if value is not '':
if self.enable_like_select:
select_sql += ' AND %s LIKE "%%%s%%"' % (key, value)
else:
select_sql += ' AND %s LIKE "%s"' % (key, value)
cursor.execute(select_sql)
ret = cursor.fetchall()
if not ret:
raise RecordNotFound
self.print_to_table(ret)
except RecordNotFound:
self.queryUserButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:此用户不存在')
logging.warning('用户不存在{}'.format(str(select_data)))
text = 'Error!'
informativeText = '<b>此用户不存在。</b>'
self.callDialog(QMessageBox.Critical, text, informativeText, QMessageBox.Ok)
except Exception as e:
print(e)
logging.error('读取数据库异常,无法查询到{}的用户信息'.format(str(select_data)))
self.queryUserButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:读取数据库异常,查询失败')
else:
self.logQueue.put('查询成功!')
self.queryUserButton.setIcon(QIcon('./icons/success.png'))
self.deleteUserButton.setEnabled(True)
finally:
cursor.close()
conn.close()
def deleteUser(self):
del_user = tuple(self.current_select)
if len(del_user) == 1:
str_del_user = '(' + str(del_user[0]) + ')'
else:
str_del_user = str(del_user)
text = '已选择{}个用户。从数据库中删除选中用户,同时删除相应人脸数据,<font color=red>该操作不可逆!</font>'.format(len(del_user))
informativeText = '<b>是否继续?</b>'
ret = self.callDialog(QMessageBox.Warning, text, informativeText, QMessageBox.Yes | QMessageBox.No,
QMessageBox.No)
if ret == QMessageBox.Yes:
conn, cursor = self.connect_to_sql()
del_sql = 'DELETE FROM userinfo WHERE id IN %s' % str_del_user
try:
cursor.execute(del_sql)
except Exception as e:
print(e)
cursor.close()
logging.error('无法从数据库中删除{}'.format(del_user))
self.deleteUserButton.setIcon(QIcon('./icons/error.png'))
self.logQueue.put('Error:读写数据库异常,删除失败')
else:
cursor.close()
conn.commit()
for stu_id in del_user:
if os.path.exists('{}/stu_{}'.format(self.datasets, stu_id)):
try:
shutil.rmtree('{}/stu_{}'.format(self.datasets, stu_id))
except Exception as e:
logging.error('系统无法删除删除{}/stu_{}'.format(self.datasets, stu_id))
self.logQueue.put('Error:删除人脸数据失败,请手动删除{}/stu_{}目录'.format(self.datasets, stu_id))
text = '你已成功删除{}个用户记录。'.format(len(del_user))
informativeText = '<b>删除成功。</b>'
self.callDialog(QMessageBox.Information, text, informativeText, QMessageBox.Ok)
self.initDb()
self.deleteUserButton.setIcon(QIcon('./icons/success.png'))
self.deleteUserButton.setEnabled(False)
self.queryUserButton.setIcon(QIcon())
finally:
conn.close()
def receiveLog(self):
while True:
data = self.logQueue.get()
if data:
self.receiveLogSignal.emit(data)
def logOutput(self, log):
time = datetime.now().strftime('[%Y/%m/%d %H:%M:%S]')
log = time + ' ' + log + '\n'
self.logTextEdit.moveCursor(QTextCursor.End)
self.logTextEdit.insertPlainText(log)
self.logTextEdit.ensureCursorVisible()
@staticmethod
def callDialog(icon, text, informativeText, standardButtons, defaultButton=None):
msg = QMessageBox()
msg.setWindowIcon(QIcon('./icons/icon.png'))
msg.setWindowTitle('用户信息修改')
msg.setIcon(icon)
msg.setText(text)
msg.setInformativeText(informativeText)
msg.setStandardButtons(standardButtons)
if defaultButton:
msg.setDefaultButton(defaultButton)
return msg.exec()
class ExportExcelDialog(QDialog):
def __init__(self):
super(ExportExcelDialog, self).__init__()
loadUi('./ui/export_excel.ui', self)
self.setWindowIcon(QIcon('./icons/icon.png'))
self.show_sqlTable.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.StuCheckTable.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.show_sqlTable.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.StuCheckTable.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.select_table_pushButton.clicked.connect(self.select_table_show)
self.export_pushButton.clicked.connect(self.export_to_excel)
self.DelpushButton.clicked.connect(self.del_table)
self.fresh_table_list()
def fresh_table_list(self):
conn, cursor = DataManageUI.connect_to_sql()
table_list = self.get_sql_table(cursor)
cursor.close()
conn.commit()
conn.close()
self.print_sql_tablelist(table_list)
@staticmethod
def get_sql_table(cursor):
sql = "show tables;"
cursor.execute(sql)
tables = [cursor.fetchall()]
table_list = re.findall('(\'.*?\')', str(tables))
table_list = [re.sub("'", '', each) for each in table_list]
return table_list
def print_sql_tablelist(self, table_list):
while self.show_sqlTable.rowCount() > 0:
self.show_sqlTable.removeRow(0)
for row_index, row_data in enumerate(table_list):
self.show_sqlTable.insertRow(row_index)
self.show_sqlTable.setItem(row_index, 0, QTableWidgetItem(str(row_data)))
def select_table_show(self):
if not self.show_sqlTable.selectedItems():
return
self.select_table = self.show_sqlTable.selectedItems()[0].text()
try:
conn, cursor = DataManageUI.connect_to_sql()
if not self.table_exists(cursor, self.select_table):
raise FileNotFoundError
sql_select = 'SELECT * FROM `%s`' % self.select_table
cursor.execute(sql_select)
conn.commit()
stu_data = cursor.fetchall()
attendance_cnt = 0
if len(stu_data[0]) != self.StuCheckTable.columnCount():
text = 'Error!'
informativeText = '<b>表格格式不正确,请重新选择正确的签到表格。</b>'
self.callDialog(QMessageBox.Critical, text, informativeText, QMessageBox.Ok)
return
while self.StuCheckTable.rowCount() > 0:
self.StuCheckTable.removeRow(0)
for row_index, row_data in enumerate(stu_data):
self.StuCheckTable.insertRow(row_index)
if row_data[2] == 1:
attendance_cnt += 1
for col_index, col_data in enumerate(row_data):
self.StuCheckTable.setItem(row_index, col_index, QTableWidgetItem(str(col_data)))
self.export_pushButton.setEnabled(True)
self.DelpushButton.setEnabled(True)
except FileNotFoundError:
logging.error('系统找不到数据库表{}'.format(self.select_table))
except Exception as e:
print(e)
logging.error('读取数据库异常,无法完成数据库查询')
else:
cursor.close()
conn.close()
def export_to_excel(self):
if not os.path.isdir('./export_excel'):
os.makedirs('./export_excel')
save_path = os.path.join('./export_excel', self.select_table + '.xls')
head_list = ['学号', '姓名', '是否出勤', '出勤时间']
xls = ExcelWrite.Workbook()
sheet = xls.add_sheet("Sheet1")
style = XFStyle()
pattern = Pattern()
pattern.pattern = Pattern.SOLID_PATTERN
pattern.pattern_fore_colour = 0x16
style.pattern = pattern
for col in range(len(head_list)):
sheet.write(0, col, head_list[col], style)
sheet.col(col).width = 4240
try:
conn, cursor = DataManageUI.connect_to_sql()
sql = 'select * from `%s`' % self.select_table
cursor.execute(sql)
row = 0
stu_data = cursor.fetchall()
for stu_info in stu_data:
stu_info = list(stu_info)
if stu_info[3]:
stu_info[3] = stu_info[3].strftime('%Y/%m/%d %H:%M:%S')
row = row + 1
for col in range(len(stu_info)):
sheet.write(row, col, stu_info[col])
xls.save(save_path)
cursor.close()
conn.close()
text = 'Success!'
informativeText = '<b>课程{}签到表 导出成功! 目标路径:./export_excel</b>'.format(self.select_table)
self.callDialog(QMessageBox.Information, text, informativeText, QMessageBox.Ok)
except Exception as e:
print(e)
text = 'Error!'
informativeText = '<b>导出失败!</b>'
self.callDialog(QMessageBox.Critical, text, informativeText, QMessageBox.Ok)
def del_table(self):
self.select_table = self.show_sqlTable.selectedItems()[0].text()
text = '确定<font color=blue> 删除 </font>表格<font color=blue> {} </font> 吗?<font color=red>该操作不可逆!</font>'.format(
self.select_table)
informativeText = '<b>是否继续?</b>'
ret = self.callDialog(QMessageBox.Warning, text, informativeText, QMessageBox.Yes | QMessageBox.No,
QMessageBox.No)
if ret == QMessageBox.Yes:
sql_del_table = 'DROP TABLE `%s`' % self.select_table
try:
conn, cursor = DataManageUI.connect_to_sql()
if not self.table_exists(cursor, self.select_table):
raise FileNotFoundError
cursor.execute(sql_del_table)
conn.commit()
text = 'Success!'
informativeText = '<b>{} 签到表 已删除!</b>'.format(self.select_table)
self.callDialog(QMessageBox.Information, text, informativeText, QMessageBox.Ok)
except FileNotFoundError:
logging.error('系统找不到数据库表{}'.format(self.select_table))
except Exception as e:
print(e)
text = 'Error!'
informativeText = '<b>无法删除!</b>'
self.callDialog(QMessageBox.Critical, text, informativeText, QMessageBox.Ok)
else:
cursor.close()
conn.close()
self.fresh_table_list()
if __name__ == '__main__':
logging.config.fileConfig('./config/logging.cfg')
app = QApplication(sys.argv)
window = DataManageUI()
window.show()
sys.exit(app.exec())
|