pyqt5_站点管理_WaterClient_1
发表于:2025-02-19 作者:千家信息网编辑
千家信息网最后更新 2025年02月19日,主界面:*程序中有错误,过段时间再回头修改.from PyQt5.QtWidgets import QApplication,QWidget,QStatusBar,QTabWidget,QVBoxLa
千家信息网最后更新 2025年02月19日pyqt5_站点管理_WaterClient_1' i = 0 while i < len(info_list): n = 0 m = 0 tr = '
' try: msg = MIMEText(msg_txt, 'plain', 'utf-8') from_addr = self.email_window.dir_email['user'] password = self.email_window.dir_email['password'] to_addrs = '' for line in self.email_window.list_users: to_addrs = to_addrs + line + ',' stmp_server = self.email_window.dir_email['stmp'] msg['From'] = _format_addr('%s<%s>' % (self.email_window.dir_email['user'],from_addr)) msg['To'] = ','.join(to_addrs) msg['Subject'] = Header('%s%s邮件' % (self.email_window.dir_email['name'],self.get_time(), 'utf-8').encode()) server = smtplib.SMTP(stmp_server, 25) server.set_debuglevel(1) server.login(from_addr, password) server.sendmail(from_addr, [to_addrs], msg.as_string()) os.remove('.//DATA//info.pik') server.quit() return 0 except Exception as error: self.write_log('邮件发送失败!\n' + repr(error)) return 1 def auto_start(self): if time.localtime().tm_min != 3: return self.write_log('------------------程序开始------------------') try: path_list = ['.//log', './/DATA'] for dir_path in path_list: dir_path_num = self.path_exists(dir_path) if dir_path_num == 1: pass elif dir_path_num == 0: os.remove(dir_path) os.mkdir(dir_path) else: os.mkdir(dir_path) config_file = './/DATA//DBconfig.pik' config_file_num = self.path_exists(config_file) first_runsql_num = 9 if config_file_num == 0: first_runsql_num = 2 elif config_file_num == 1: os.remove(config_file) first_runsql_num = self.frist_runsql(config_file) else: first_runsql_num = self.frist_runsql(config_file) if first_runsql_num == 2 or first_runsql_num == 4: pass elif first_runsql_num > 4: os.remove('.//DATA//DBconfig.pik') self.write_log('DBconfig.pik有问题.程序终止运行.') return elif first_runsql_num == 0: self.write_log('数据库中不存在想要的数据库.') return else: self.write_log('DBconfig.pik有问题.程序终止运行.') return except Exception as error: self.write_log('执行错误') return try: run_sql_flag = self.run_sql() if run_sql_flag == 1: return if time.localtime().tm_hour == 8: if time.localtime().tm_min == 3: i = 0 while i < 4: flag = self.send_email() if flag == 0: i = 4 break i = i + 1 time.sleep(300) except Exception as error: self.write_log('最后一段时间错误:\n' + repr(error))if __name__ == '__main__': if not os.path.exists('.//DATA'): os.mkdir('.//DATA') app = QApplication(sys.argv) mainwin = mainwindows() sys.exit(app.exec_())
主界面:
*程序中有错误,过段时间再回头修改.
from PyQt5.QtWidgets import QApplication,QWidget,QStatusBar,QTabWidget,QVBoxLayoutfrom PyQt5.QtGui import QIcon,QPalette,QBrush,QPixmapfrom PyQt5.QtCore import QTimerimport sys,os,time,datetime,smtplib,pickle,pymssqlimport index_windows,stcd_windows,db_windows,email_windows,imgfrom email import encodersfrom email.header import Headerfrom email.utils import parseaddr,formataddrfrom email.mime.text import MIMEText#定义主程序界面:class mainwindows(QWidget): def __init__(self): super().__init__() self.initUI() def initUI(self): #设置标题和大小 self.qicon1 = QIcon(':/IMG/1.png') self.qicon2 = QIcon(':/IMG/2.png') self.qicon3 = QIcon(':/IMG/3.png') self.qicon4 = QIcon(':/IMG/4.png') self.windows_icon_timer_num = 1 self.windows_icon_timer = QTimer(self) self.windows_icon_timer.timeout.connect(self.windows_icon_timer_changed) self.windows_icon_timer.start(500) #背景图片 palette = QPalette() palette.setBrush(QPalette.Background,QBrush(QPixmap(':/IMG/window_bg.png'))) self.setPalette(palette) self.resize(800,600) self.setWindowTitle('检测软件') #定义tabwidget:1,首页 2,站点配置,3,数据库配置,4,邮件配置,5,其他配置 #1,首页 self.index_window = index_windows.index_window() #2,站点配置 self.stcd_window = stcd_windows.stcd_window() #3,数据库配置 self.db_window = db_windows.db_window() #4,邮件配置 self.email_window = email_windows.email_window() #5,其他配置 self.ather_window = QWidget() #定义tabwidget self.tabwidget = QTabWidget() self.tabwidget.insertTab(0,self.index_window,QIcon(':/IMG/index.png'),'首页') self.tabwidget.insertTab(1,self.stcd_window,QIcon(':/IMG/stcd.png'),'站点') self.tabwidget.insertTab(2,self.db_window,QIcon(':/IMG/database.png'),'数据库') self.tabwidget.insertTab(3,self.email_window,QIcon(':/IMG/mail.png'),'邮件') self.tabwidget.insertTab(4,self.ather_window,QIcon(':/IMG/information.png'),'杂项') # 设置状态栏 self.statusBar = QStatusBar(self) #布局管理 vbox = QVBoxLayout() vbox.addWidget(self.tabwidget) vbox.addWidget(self.statusBar) self.setLayout(vbox) self.show() self.auto_start_timer = QTimer(self) self.auto_start_timer.timeout.connect(self.auto_start) self.auto_start_timer.start(60000) def windows_icon_timer_changed(self): if self.windows_icon_timer_num == 1: self.setWindowIcon(self.qicon1) self.windows_icon_timer_num = 2 elif self.windows_icon_timer_num == 2: self.setWindowIcon(self.qicon2) self.windows_icon_timer_num = 3 elif self.windows_icon_timer_num == 3: self.setWindowIcon(self.qicon3) self.windows_icon_timer_num = 4 elif self.windows_icon_timer_num == 4: self.setWindowIcon(self.qicon4) self.windows_icon_timer_num = 1 else: self.windows_icon_timer_num = 1 def get_time(time_flag='', d=0, H=0, M=0): time_now = datetime.datetime.today() try: if time_flag == 'YMD': time_d = time_now + datetime.timedelta(days=d) return str(time_d).split()[0] # 2017-03-10 elif time_flag == 'DHM': time_M = time_now + datetime.timedelta(minutes=M) time_H = time_M + datetime.timedelta(hours=H) time_D = time_H + datetime.timedelta(days=d) return str(time_D).split('.')[0] # 2017-03-10 14:00:39 elif time_flag == 'NOW': time_1 = str(time_now).split()[0] hour = time.localtime().tm_hour return time_1 + ' ' + str(hour) + ':03:00' elif time_flag == 'HOUR': time_H = time_now + datetime.timedelta(hours=-1) time_1 = str(time_H).split()[0] hour = str(time_H).split()[1].split(':')[0] return time_1 + ' ' + hour + ':03:00' else: return str(time_now).split('.')[0] # 2017-03-10 14:00:39 except Exception as error: return '2017-1-1 00:00:00' # 2017-03-10 14:00:39 def write_log(self,message): self.index_window.index_textedit.append(self.get_time() + ': %s\n'%message) def path_exists(self,file_path): # return 2:文件不存在 # return 0:不是文件夹,且有内容 # return 1:是文件夹 try: if os.path.exists(file_path): if os.path.isfile(file_path): if os.path.getsize(file_path) > 0: return 0 else: os.remove(file_path) return 2 else: return 1 else: return 2 except Exception as error: self.write_log(repr(error)) def frist_runsql(self,DBconfig_path): # return 1:链接数据库错误 # 说明:第一次在服务器上运行时候,初始化服务器数据信息 self.write_log('--------------------第一次初始化数据库信息--------------------') # 用于链接数据库,查询数据库内容,return 0:success 1,2,3,4...error try: conn = pymssql.connect(**self.db_window.dir_db) except Exception as error: self.write_log('第一次初始化过程中,连接数据库错误:\n' + repr(error)) self.statusBar.showMessage('数据库链接错误.') return 1 try: cur = conn.cursor() # 得到数据库游标 cur.execute('SELECT Name FROM Master..SysDatabases;') # 得到所有数据库名称 row = cur.fetchone() error_status = 0 db_list = [] # 标记错误次数 while row: if row[0] == 'water' or row[0] == 'STDDB_V380': self.write_log('数据库中存在%s数据库.' % row[0]) db_list.append(row[0]) error_status = error_status + 2 else: self.write_log('扫描到的其他数据库名称:' + row[0]) row = cur.fetchone() if len(db_list) == 0: return 0 try: file_water = open(DBconfig_path, 'wb') pickle.dump(db_list, file_water) except Exception as error: self.write_log('初始化过程中,%s数据库处理部分,错误信息:\n' % row[0] + repr(error)) error_status = error_status + 5 finally: try: file_water.close() except Exception as error: self.write_log('关闭%s文件出错:\n' % row[0] + repr(error)) except Exception as error: self.write_log('初始化数据过程中,存在的问题:\n' + repr(error)) finally: try: cur.close() conn.close() self.write_log('----------------------初始化数据信息完成----------------------') return error_status except Exception as error: self.write_log('初始化数据库过程中,关闭数据库出错:' + repr(error)) return error_status + 7 def runsql(self,sql, DBname, TABLE_NAME): # return 1:数据库链接错误 # return 2: try: conn = pymssql.connect(**self.db_window.dir_db, database=DBname) except pymssql.InterfaceError as intererror: # 链接不了数据库 # print('链接不上数据库') # 记录连接日志 log = '程序无法连接数据库!' self.write_log(DBname + log + repr(intererror)) with open('.//DATA//error.log', 'a') as error_rite: error_rite.write('执行数据库查询,链接数据库失败.') return (1, []) except Exception as error: self.write_log(DBname + repr(error)) with open('.//DATA//error.log', 'a') as error_rite: error_rite.write('数据库执行错误%s' % repr(error), ) return (1, []) try: self.statusBar.showMessage('数据库链接正常.') # print('================连接数据库==========================') cur = conn.cursor() # 得到数据库游标 cur.execute(sql) # 得到所有数据库名称 row = cur.fetchone() try: row_list = [] while row: try: row_list.append((str(row[0]).strip(), str(row[1]), row[2])) except Exception as error_1: self.write_log('%s行处理数据结果过程中,错误:\n' % TABLE_NAME + repr(error_1)) finally: row = cur.fetchone() except Exception as error: self.write_log('数据库列表过程中,错误:\n' + repr(error)) except Exception as error: # print('error--------1--------' + repr(error)) self.write_log('%s执行错误' % TABLE_NAME + repr(error)) finally: try: cur.close() conn.close() return (0, row_list) except Exception as error: self.write_log(TABLE_NAME + ':关闭数据库出错!' + repr(error)) finally: self.write_log(TABLE_NAME + '-->数据脚本执行完成!') def run_sql(self): # 说明:执行数据库查询脚本 try: # return 0 :ok # return 1 :error info_list_path = './/DATA//info.pik' info_list_path_num = self.path_exists(info_list_path) pickle_infolist_num = 0 if info_list_path_num == 1: os.remove(info_list_path) pickle_infolist_num = self.pickle_infolist() elif info_list_path_num == 2: pickle_infolist_num = self.pickle_infolist() if pickle_infolist_num == 1: return 1 except Exception as error: self.write_log('执行数据库查询,最初阶段出错!\n' + repr(error)) try: config_path = './/DATA//DBconfig.pik' config_file = open(config_path, 'rb') pickle_load = pickle.load(config_file) i = 0 row_list = [] while i < len(pickle_load): if pickle_load[i] == 'water': sql = 'SELECT [meterId], [meterDt],[rainAmount] FROM [water].[dbo].[rain] ' \ 'WHERE meterDt<\'' + self.get_time('NOW') + '\' AND meterDt>=\'' + self.get_time('HOUR') + '\';' status, row_list_rain = self.runsql(sql, 'water', 'rain') if status == 0: row_list = row_list + row_list_rain sql = 'SELECT [meterId],[meterDt],[waterLv] FROM [water].[dbo].[waterLv] ' \ 'WHERE meterDt<\'' + self.get_time('NOW') + '\' AND meterDt>=\'' + self.get_time('HOUR') + '\';' status, row_list_waterlv = self.runsql(sql, 'water', 'waterLv') if status == 0: row_list = row_list + row_list_waterlv elif pickle_load[i] == 'STDDB_V380': sql = 'SELECT [STCD],[TM],[DRP] FROM [STDDB_V380].[dbo].[ST_PPTN_R] ' \ 'WHERE TM<\'' + self.get_time('NOW') + '\' AND TM>=\'' + self.get_time('HOUR') + '\';' status, row_list_pptn = self.runsql(sql, 'STDDB_V380', 'ST_PPTN_R') if status == 0: row_list = row_list + row_list_pptn sql = 'SELECT [STCD],[TM],[Z] FROM [STDDB_V380].[dbo].[ST_RIVER_R]' \ 'WHERE TM<\'' + self.get_time('NOW') + '\' AND TM>=\'' + self.get_time('HOUR') + '\';' status, row_list_river = self.runsql(sql, 'STDDB_V380', 'ST_RIVER_R') if status == 0: row_list = row_list + row_list_river sql = 'SELECT [STCD],[TM],[RZ] FROM [STDDB_V380].[dbo].[ST_RSVR_R]' \ 'WHERE TM<\'' + self.get_time('NOW') + '\' AND TM>=\'' + self.get_time('HOUR') + '\';' status, row_list_rsvr = self.runsql(sql, 'STDDB_V380', 'ST_RSVR_R') if status == 0: row_list = row_list + row_list_rsvr i = i + 1 except Exception as error: self.write_log('数据库查询总体上报错:\n' + repr(error)) try: info_file = open('.//DATA//info.pik', 'rb') info_list = pickle.load(info_file) except Exception as error: self.write_log('得站点信息列表错误,info.pik已经存在:\n' + repr(error)) try: now_hour = time.localtime().tm_hour info_list[0].append(str(now_hour - 1) + '时 至 ' + str(now_hour) + '时') i = 1 while i < len(info_list): info_list[i][0] m = 0 flag = 0 while m < len(row_list): if str(info_list[i][0][0]) == str(row_list[m][0]): flag = 1 del row_list[m] m = m - 1 m = m + 1 if flag == 1: info_list[i].append(1) else: info_list[i].append(0) i = i + 1 except Exception as error: print(repr(error)) return 1 try: info_list_file = open('.//DATA//info.pik', 'wb') pickle.dump(info_list, info_list_file) return 0 except Exception as error: self.write_log('重新压入数据错误:' + repr(error)) return 1 finally: info_list_file.close() def send_email(self): # 发送邮件 def _format_addr(s): name, addr = parseaddr(s) return formataddr((Header(name, 'utf-8').encode(), addr)) self.write_log('准备发送邮件') file = open('.//DATA//info.pik', 'rb') info_list = pickle.load(file) msg_txt = '
%d | ' % i + str( info_list[i][n]) + ' | ' m = m + 1 else: tr = tr + '' + str(info_list[i][n]) + ' | ' else: if info_list[i][n] == 0: tr = tr + '无数据 | ' else: tr = tr + '有数据 | ' n = n + 1 tr = tr + '
数据
数据库
错误
链接
配置
邮件
过程
信息
程序
查询
站点
文件
名称
第一次
问题
首页
运行
内容
文件夹
时间
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全
英灵殿服务器mod
军营网络安全周活动心得体会
分布式数据库有哪些模块
博库数据库怎么检索中华期刊
视频动态服务器未连接
怎么用笔记本直接连接服务器
数据库安装在固态硬盘上
电厂调度网络安全
销售串口终端服务器公司
重庆商贸管理系统软件开发
石化网络安全培训
数据库集群与redis
黎明杀机怎么去韩服务器
深圳车管所网络技术
经济技术开发区庆富网络技术
网络安全的主要问题是什么
心魔sp家庭系列中文数据库
剑网3指尖江湖服务器掉线
禄劝方便软件开发市场报价
中国象棋软件开发代码
数据库yk
网络安全实验室题库
涪陵区工商软件开发流程报价表
全国网络安全2022
闽盾杯网络安全大赛是干嘛的
福建约客互联网科技
数据库 匹配
数据库交互的方式
软件开发一起销售的
网络安全法规知识常识黑板报