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安全错误
数据库的锁怎样保障安全
ps5生化危机8连接服务器失败
地下城的新服务器在哪
网络安全钥密不对
关于小学网络安全班会记录
泰格软件数据库怎么安装
sql数据库创建实例
网盘小型服务器
软件开发培训要多长时间
db2 数据库日志查看
天门淘宝客系统软件开发
有哪些软件开发比赛
网络安全安全教育教案幼儿园
远程服务器登录电脑卡住
使用网络安全知识
打开有关服务器的佛的图片
C语言笔记软件开发
网络安全伴我成长作品
网络安全警戒
黑莓Q10软件开发
通保三级通信网络安全防护服务
齐鲁网络安全培训
帝国时代3无法连接在线服务器
软件开发生命周期的含义
网络安全保护和安全防护的区别
网络安全在我心中班会
番茄todo软件开发资金来源
3a5000多路服务器
如何查询服务器当前时间
塔石串口服务器连接wincc
网络安全与数字签名技术开题报告