[Python] 纯文本查看 复制代码
# -*- coding: utf-8 -*-
import os
import sys
import time
import json
import requests
import logging
from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QLabel, QHBoxLayout, QAction, \
QSizePolicy, QTextEdit, QDialog, QLineEdit, QMessageBox, QFileDialog, QComboBox, QTableWidget, QTableWidgetItem, \
QScrollArea, QTabWidget, QToolTip
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QUrl, QTimer
from PyQt5.QtGui import QFont, QColor
from PyQt5.QtGui import QDesktopServices # 导入 QDesktopServices
import pandas as pd
from collections import OrderedDict
# 定义Excel文件路径
PATHS_EXCEL_FILE = "paths.xlsx"
# 读取Excel文件并转换为有序字典
def read_paths_from_excel(file_path):
try:
df = pd.read_excel(file_path, sheet_name='Paths')
ordered_dict = OrderedDict()
for _, row in df.iterrows():
line = row['Line']
machine = row['Machine']
shift = row['Shift']
type_ = row['Type']
path_template = row['Path Template']
if line not in ordered_dict:
ordered_dict[line] = OrderedDict()
if machine not in ordered_dict[line]:
ordered_dict[line][machine] = OrderedDict()
if shift not in ordered_dict[line][machine]:
ordered_dict[line][machine][shift] = OrderedDict()
ordered_dict[line][machine][shift][type_] = path_template
return ordered_dict
except Exception as e:
log_message(f"读取Excel文件 {file_path} 时发生错误: {str(e)}")
return OrderedDict()
# 读取配置信息
def read_config_from_excel(file_path):
try:
config_df = pd.read_excel(file_path, sheet_name='Config', index_col='Key')['Value'].to_dict()
return config_df
except Exception as e:
log_message(f"读取Excel文件 {file_path} 的配置信息时发生错误: {str(e)}")
return {}
# 获取目录中文件数量
def get_directory_file_count(directory):
try:
log_message(f"正在检查路径: {directory}")
if directory is None or not os.path.exists(directory):
log_message(f"路径 {directory} 不存在或为空")
return 0
files = os.listdir(directory)
file_count = len(files)
log_message(f"路径 {directory} 中的文件数量: {file_count}")
return file_count
except PermissionError:
log_message(f"权限不足无法访问路径 {directory}")
return 0
except Exception as e:
log_message(f"读取路径 {directory} 时发生错误: {str(e)}")
return 0
# 设置日志记录
def setup_logging():
log_dir = "C:\\ZTL-log"
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, "monitor_log.txt")
logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 记录日志消息
def log_message(message):
print(message) # 添加打印输出，方便调试
logging.info(message)
class FileScannerThread(QThread):
update_signal = pyqtSignal(str, str, dict)
finished_signal = pyqtSignal(dict)
def __init__(self, lines_machines_path_templates, push_interval_minutes):
super().__init__()
self.lines_machines_path_templates = lines_machines_path_templates
self.push_interval_minutes = push_interval_minutes
self.running = False
def run(self):
while self.running:
self.run_once()
time.sleep(self.push_interval_minutes * 60)
def run_once(self):
current_time = time.localtime()
current_hour = current_time.tm_hour
current_minute = current_time.tm_min
# 根据当前时间判断白班/夜班
if (current_hour >= 8 and current_hour < 20):
current_shift = "DayShift"
shift_display = "白班"
elif (current_hour >= 20 or current_hour < 8):
current_shift = "NightShift"
shift_display = "夜班"
else:
self.update_signal.emit("不在监控时间段内", "", {})
log_message("不在监控时间段内，跳过本次监控")
return
# 日期切换逻辑
current_date = time.strftime("%Y%m%d")
if current_shift == "NightShift":
if current_hour >= 0 and current_hour < 8:
yesterday = time.localtime(time.time() - 86400)
current_date = time.strftime("%Y%m%d", yesterday)
results = {}
for line, machines in self.lines_machines_path_templates.items():
line_results = {}
total_files = 0
good_files = 0
for machine, shifts in machines.items():
ng_template = shifts[current_shift].get("NG", "")
ok_template = shifts[current_shift].get("OK", "")
ng_directory, ok_directory = self.find_existing_directory(ng_template, ok_template, current_time)
ng_count = get_directory_file_count(ng_directory) or 0
ok_count = get_directory_file_count(ok_directory) or 0
total_files += ng_count + ok_count
good_files += ok_count
folder_counts = self.get_folder_counts(ng_directory)
ip_address = ""
self.update_signal.emit(line, machine, folder_counts)
line_results[machine] = {"total_files": total_files, "good_files": good_files,
"ok_count": ok_count, "ng_count": ng_count, "folder_counts": folder_counts}
results[line] = line_results
self.finished_signal.emit(results)
def find_existing_directory(self, ng_template, ok_template, current_time):
date_formats = ["%Y%m%d", "%Y-%m-%d", "%y%m%d", "%Y_%m_%d"] # 新增 %Y_%m_%d 格式
found_ng_directory = None
found_ok_directory = None
if current_time.tm_hour >= 0 and current_time.tm_hour < 8:
current_time = time.localtime(time.time() - 86400)
for fmt in date_formats:
try:
current_date = time.strftime(fmt, current_time)
ng_directory = ng_template.format(zsz=current_date) if ng_template else ""
ok_directory = ok_template.format(zsz=current_date) if ok_template else ""
if ng_directory and os.path.exists(ng_directory):
found_ng_directory = ng_directory
if ok_directory and os.path.exists(ok_directory):
found_ok_directory = ok_directory
if found_ng_directory and found_ok_directory:
break
except ValueError:
pass
return found_ng_directory, found_ok_directory
def get_folder_counts(self, ng_directory):
folder_counts = {}
if ng_directory and os.path.exists(ng_directory):
for folder_name in os.listdir(ng_directory):
folder_path = os.path.join(ng_directory, folder_name)
if os.path.isdir(folder_path):
folder_counts[folder_name] = get_directory_file_count(folder_path)
return folder_counts
class PathManagerDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("路径管理")
self.setGeometry(100, 100, 600, 400)
layout = QVBoxLayout()
self.text_edit = QTextEdit()
self.text_edit.setReadOnly(True)
layout.addWidget(self.text_edit)
button_layout = QHBoxLayout()
self.edit_button = QPushButton("编辑路径")
self.edit_button.clicked.connect(self.edit_paths)
button_layout.addWidget(self.edit_button)
self.save_button = QPushButton("保存路径")
self.save_button.clicked.connect(self.save_paths)
self.save_button.setEnabled(False)
button_layout.addWidget(self.save_button)
layout.addLayout(button_layout)
self.setLayout(layout)
self.load_paths()
def load_paths(self):
global LINES_MACHINES_PATH_TEMPLATES
LINES_MACHINES_PATH_TEMPLATES = read_paths_from_excel(PATHS_EXCEL_FILE)
paths_json = json.dumps(LINES_MACHINES_PATH_TEMPLATES, ensure_ascii=False, indent=4)
self.text_edit.setText(paths_json)
def edit_paths(self):
self.text_edit.setReadOnly(False)
self.edit_button.setEnabled(False)
self.save_button.setEnabled(True)
def save_paths(self):
try:
new_paths = json.loads(self.text_edit.toPlainText())
df = pd.DataFrame(columns=["Line", "Machine", "Shift", "Type", "Path Template"])
for line, machines in new_paths.items():
for machine, shifts in machines.items():
for shift, types in shifts.items():
for type_, path_template in types.items():
df = df.append({"Line": line, "Machine": machine, "Shift": shift, "Type": type_,
"Path Template": path_template}, ignore_index=True)
df.sort_values(by=["Line", "Machine", "Shift", "Type"], inplace=True)
df.to_excel(PATHS_EXCEL_FILE, sheet_name='Paths', index=False)
self.text_edit.setReadOnly(True)
self.edit_button.setEnabled(True)
self.save_button.setEnabled(False)
QMessageBox.information(self, "成功", "路径已保存")
except json.JSONDecodeError:
QMessageBox.warning(self, "错误", "JSON格式不正确，请检查并重试")
class ConfigManagerDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("配置管理")
self.setGeometry(100, 100, 600, 400)
layout = QVBoxLayout()
self.text_edit = QTextEdit()
self.text_edit.setReadOnly(True)
layout.addWidget(self.text_edit)
button_layout = QHBoxLayout()
self.edit_button = QPushButton("编辑配置")
self.edit_button.clicked.connect(self.edit_config)
button_layout.addWidget(self.edit_button)
self.save_button = QPushButton("保存配置")
self.save_button.clicked.connect(self.save_config)
self.save_button.setEnabled(False)
button_layout.addWidget(self.save_button)
layout.addLayout(button_layout)
self.setLayout(layout)
self.load_config()
def load_config(self):
global CONFIG
CONFIG = read_config_from_excel(PATHS_EXCEL_FILE)
config_json = json.dumps(CONFIG, ensure_ascii=False, indent=4)
self.text_edit.setText(config_json)
def edit_config(self):
self.text_edit.setReadOnly(False)
self.edit_button.setEnabled(False)
self.save_button.setEnabled(True)
def save_config(self):
try:
new_config = json.loads(self.text_edit.toPlainText())
df = pd.DataFrame(list(new_config.items()), columns=['Key', 'Value'])
df.to_excel(PATHS_EXCEL_FILE, sheet_name='Config', index=False)
self.text_edit.setReadOnly(True)
self.edit_button.setEnabled(True)
self.save_button.setEnabled(False)
QMessageBox.information(self, "成功", "配置已保存")
except json.JSONDecodeError:
QMessageBox.warning(self, "错误", "JSON格式不正确，请检查并重试")
class MonitorApp(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("设备文件分类监控系统（产线分页） V25.04.21.14")
self.setGeometry(100, 100, 1600, 800)
# 初始化 tables 字典
self.tables = {}
# 读取配置信息
global CONFIG
CONFIG = read_config_from_excel(PATHS_EXCEL_FILE)
self.push_interval_minutes = int(CONFIG.get('PUSH_INTERVAL_MINUTES', 60))
self.webhook_url = CONFIG.get('WEBHOOK_URL', '')
# 创建主部件和布局
main_widget = QWidget()
self.setCentralWidget(main_widget)
main_layout = QVBoxLayout()
main_layout.setSpacing(1)
# 标题
title_label = QLabel("<b>设备文件分类监控系统</b>", self)
title_label.setFont(QFont('Arial', QFont.Bold, 18))
title_label.setAlignment(Qt.AlignHCenter | Qt.AlignTop)
title_label.setStyleSheet("color: #ffffff; background-color: #2c3e50; padding: 5px;")
title_label.setFixedHeight(40)
main_layout.addWidget(title_label)
# 工具栏区域
toolbar_layout = QHBoxLayout()
toolbar_layout.setSpacing(10)
# 开始监控按钮
self.start_button = QPushButton("开始监控", self)
self.start_button.setFixedSize(150, 50)
self.start_button.clicked.connect(self.start_monitoring)
self.start_button.setToolTip("开始监控")
self.start_button.setStyleSheet("""
QPushButton {
border-radius: 25px;
background-color: #2ecc71;
color: white;
font-size: 16px;
}
QPushButton:hover {
background-color: #27ae60;
}
QPushButton:pressed {
background-color: #229954;
}
""")
toolbar_layout.addWidget(self.start_button)
# 停止监控按钮
self.stop_button = QPushButton("停止监控", self)
self.stop_button.setFixedSize(150, 50)
self.stop_button.clicked.connect(self.stop_monitoring)
self.stop_button.setEnabled(False)
self.stop_button.setToolTip("停止监控")
self.stop_button.setStyleSheet("""
QPushButton {
border-radius: 25px;
background-color: #e74c3c;
color: white;
font-size: 16px;
}
QPushButton:hover {
background-color: #c0392b;
}
QPushButton:pressed {
background-color: #a93226;
}
""")
toolbar_layout.addWidget(self.stop_button)
# 管理路径按钮
self.manage_paths_button = QPushButton("管理路径", self)
self.manage_paths_button.setFixedSize(150, 50)
self.manage_paths_button.clicked.connect(self.open_path_manager)
self.manage_paths_button.setToolTip("管理路径")
self.manage_paths_button.setStyleSheet("""
QPushButton {
border-radius: 25px;
background-color: #3498db;
color: white;
font-size: 16px;
}
QPushButton:hover {
background-color: #2980b9;
}
QPushButton:pressed {
background-color: #2471a3;
}
""")
toolbar_layout.addWidget(self.manage_paths_button)
# 管理配置按钮
self.manage_config_button = QPushButton("管理配置", self)
self.manage_config_button.setFixedSize(150, 50)
self.manage_config_button.clicked.connect(self.open_config_manager)
self.manage_config_button.setToolTip("管理配置")
self.manage_config_button.setStyleSheet("""
QPushButton {
border-radius: 25px;
background-color: #f39c12;
color: white;
font-size: 16px;
}
QPushButton:hover {
background-color: #e67e22;
}
QPushButton:pressed {
background-color: #d35400;
}
""")
toolbar_layout.addWidget(self.manage_config_button)
# 刷新时间按钮
self.refresh_button = QPushButton("刷新时间", self)
self.refresh_button.setFixedSize(150, 50)
self.refresh_button.clicked.connect(self.update_date_and_shift)
self.refresh_button.setToolTip("手动刷新时间")
self.refresh_button.setStyleSheet("""
QPushButton {
border-radius: 25px;
background-color: #f39c12;
color: white;
font-size: 16px;
}
QPushButton:hover {
background-color: #e67e22;
}
QPushButton:pressed {
background-color: #d35400;
}
""")
toolbar_layout.addWidget(self.refresh_button)
# 空白间隔
spacer = QWidget()
spacer.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred)
toolbar_layout.addWidget(spacer)
# 日期和班次标签
self.date_shift_label = QLabel("", self)
self.date_shift_label.setFixedSize(200, 50)
self.date_shift_label.setAlignment(Qt.AlignCenter)
self.date_shift_label.setStyleSheet("""
QLabel {
color: white;
font-size: 16px;
border: 2px solid #34495e;
background-color: #2c3e50;
border-radius: 25px;
}
""")
toolbar_layout.addWidget(self.date_shift_label)
# 倒计时标签
self.countdown_label = QLabel(f"下次监控将在 {self.push_interval_minutes} 分钟后进行...", self)
self.countdown_label.setFixedSize(300, 50)
self.countdown_label.setAlignment(Qt.AlignCenter)
self.countdown_label.setStyleSheet("""
QLabel {
color: white;
font-size: 16px;
border: 2px solid #34495e;
background-color: #2c3e50;
border-radius: 25px;
}
""")
toolbar_layout.addWidget(self.countdown_label)
# 监控状态标签
self.status_label = QLabel("监控未启动", self)
self.status_label.setFixedSize(200, 50)
self.status_label.setAlignment(Qt.AlignCenter)
self.status_label.setStyleSheet("""
QLabel {
color: white;
font-size: 16px;
border: 2px solid #34495e;
background-color: #2c3e50;
border-radius: 25px;
}
""")
toolbar_layout.addWidget(self.status_label)
main_layout.addLayout(toolbar_layout)
# 图表区域
self.tab_widget = QTabWidget()
main_layout.addWidget(self.tab_widget)
main_widget.setLayout(main_layout)
self.monitoring = False
setup_logging()
global LINES_MACHINES_PATH_TEMPLATES
LINES_MACHINES_PATH_TEMPLATES = read_paths_from_excel(PATHS_EXCEL_FILE)
self.file_scanner_thread = FileScannerThread(LINES_MACHINES_PATH_TEMPLATES, self.push_interval_minutes)
self.file_scanner_thread.update_signal.connect(self.update_ui)
self.file_scanner_thread.finished_signal.connect(self.send_notification)
self.time_update_timer = QTimer(self)
self.time_update_timer.timeout.connect(self.update_date_and_shift)
self.time_update_timer.start(60000)
self.countdown_timer = QTimer(self)
self.countdown_timer.timeout.connect(self.update_countdown_label)
self.countdown_timer.start(1000)
self.auto_tab_switch_timer = QTimer(self)
self.auto_tab_switch_timer.timeout.connect(self.switch_to_next_tab)
self.auto_tab_switch_timer.start(10000) # 每10秒自动切换一次
self.current_tab_index = 0
self.update_date_and_shift()
self.generate_ui_elements()
self.path_manager_dialog = None
self.config_manager_dialog = None
self.setWindowFlags(Qt.WindowMinMaxButtonsHint | Qt.WindowCloseButtonHint)
self.centralWidget().setMinimumSize(800, 600)
self.resize(1600, 800)
self.showMaximized()
self.adjust_font_sizes()
def adjust_font_sizes(self):
screen_size = QApplication.desktop().screenGeometry()
screen_width = screen_size.width()
screen_height = screen_size.height()
base_font_size = 12
font_scale_factor = min(screen_width / 1920, screen_height / 1080, 1.0)
adjusted_font_size = max(base_font_size * font_scale_factor, 10)
font = QFont()
font.setPointSize(int(adjusted_font_size))
self.setFont(font)
for widget in self.findChildren(QLabel):
if isinstance(widget, QLabel):
if widget.objectName() == "ng_rate_label":
widget.setFont(QFont('Arial', int(adjusted_font_size * 2), QFont.Bold))
else:
widget.setFont(font)
for widget in self.findChildren(QPushButton):
widget.setFont(font)
for widget in self.findChildren(QTextEdit):
widget.setFont(font)
def resizeEvent(self, event):
super().resizeEvent(event)
self.adjust_font_sizes()
def generate_ui_elements(self):
self.tab_widget.clear()
colors = ["#ADD8E9", "#FFFACD", "#E6E6FA", "#F0E68C", "#FFA07A"]
all_lines = list(LINES_MACHINES_PATH_TEMPLATES.keys())
for line_name in all_lines:
tab = QWidget()
tab_layout = QVBoxLayout(tab)
tab_layout.setSpacing(10)
machines = LINES_MACHINES_PATH_TEMPLATES[line_name]
machine_names = sorted(machines.keys())
for i in range(0, len(machine_names), 8):
line_layout = QHBoxLayout()
line_layout.setSpacing(10)
for j in range(i, min(i + 8, len(machine_names))):
machine_name = machine_names[j]
container = QWidget()
container_layout = QVBoxLayout(container)
container_layout.setContentsMargins(10, 10, 10, 10)
container_layout.setSpacing(5)
info_container = QWidget()
info_layout = QHBoxLayout(info_container)
info_layout.setContentsMargins(0, 0, 0, 0)
info_layout.setSpacing(5)
line_label = QLabel(line_name, self)
line_label.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
line_label.setStyleSheet("""
QLabel {
color: black;
font-weight: bold;
}
""")
machine_label = QLabel(machine_name, self)
machine_label.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
machine_label.setStyleSheet("""
QLabel {
color: black;
font-weight: bold;
}
""")
info_layout.addWidget(line_label)
info_layout.addWidget(machine_label)
info_container.setStyleSheet(f"""
QWidget {{
border: none;
background-color: {colors[all_lines.index(line_name) % len(colors)]};
border-top-left-radius: 10px;
border-top-right-radius: 10px;
padding: 5px;
}}
""")
query_button = QPushButton("查询", self)
query_button.setFixedSize(45, 25)
query_button.setStyleSheet("""
QPushButton {
border-radius: 4px;
background-color: #3498db;
color: white;
}
""")
query_button.clicked.connect(
lambda checked, ln=line_name, mn=machine_name: self.check_and_open_path(ln, mn))
table = QTableWidget(self)
table.setColumnCount(2)
table.setHorizontalHeaderLabels(["分类", "数量"])
table.horizontalHeader().setStretchLastSection(True)
table.verticalHeader().setVisible(False)
table.setAlternatingRowColors(True)
table.setSelectionBehavior(QTableWidget.SelectRows)
table.setMinimumWidth(20)
table.setMaximumWidth(300)
table.setFixedHeight(200)
container_layout.addWidget(info_container)
container_layout.addWidget(query_button)
container_layout.addWidget(table)
container.setStyleSheet(f"""
QWidget {{
border: 2px solid #34495e;
background-color: transparent;
border-radius: 10px;
}}
""")
line_layout.addWidget(container)
self.tables[(line_name, machine_name)] = table
tab_layout.addLayout(line_layout)
self.tab_widget.addTab(tab, f"{line_name}")
def start_monitoring(self):
if not self.monitoring:
self.monitoring = True
log_message("监控开始")
self.file_scanner_thread.running = True
self.file_scanner_thread.start()
self.start_button.setEnabled(False)
self.stop_button.setEnabled(True)
self.status_label.setText("正在监控")
self.status_label.setStyleSheet("""
QLabel {
color: white;
font-size: 16px;
border: 2px solid #34495e;
background-color: #27ae60;
border-radius: 25px;
}
""")
self.update_countdown_label()
else:
log_message("监控已经启动")
def stop_monitoring(self):
if self.monitoring:
self.monitoring = False
log_message("监控停止")
self.file_scanner_thread.running = False
self.start_button.setEnabled(True)
self.stop_button.setEnabled(False)
self.status_label.setText("监控未启动")
self.status_label.setStyleSheet("""
QLabel {
color: white;
font-size: 16px;
border: 2px solid #34495e;
background-color: #e74c3c;
border-radius: 25px;
}
""")
def update_ui(self, line, machine, folder_counts):
if (line, machine) in self.tables:
table = self.tables[(line, machine)]
table.clearContents()
table.setRowCount(len(folder_counts))
max_count = max(folder_counts.values(), default=0)
max_folders = [k for k, v in folder_counts.items() if v == max_count]
rows = []
for folder_name, count in folder_counts.items():
if folder_name in max_folders:
rows.insert(0, (folder_name, count))
else:
rows.append((folder_name, count))
for row, (folder_name, count) in enumerate(rows):
item_folder = QTableWidgetItem(folder_name)
item_count = QTableWidgetItem(str(count))
item_folder.setTextAlignment(Qt.AlignCenter)
item_count.setTextAlignment(Qt.AlignCenter)
table.setItem(row, 0, item_folder)
table.setItem(row, 1, item_count)
if folder_name in max_folders:
item_folder.setBackground(QColor("#FFCCCC"))
item_count.setBackground(QColor("#FFCCCC"))
log_message(f"更新 {line}-{machine}: 文件夹及数量={folder_counts}")
# 设置列宽
table.setColumnWidth(0, 100) # 分类列宽度
table.setColumnWidth(1, 50) # 数量列宽度
def update_date_and_shift(self):
current_time = time.localtime()
current_hour = current_time.tm_hour
current_minute = current_time.tm_min
if (current_hour >= 8 and current_hour < 20):
current_shift = "DayShift"
shift_display = "白班"
elif (current_hour >= 20 or current_hour < 8):
current_shift = "NightShift"
shift_display = "夜班"
else:
shift_display = "不在监控时间段内"
current_date = time.strftime("%Y%m%d")
if current_shift == "NightShift":
if current_hour >= 0 and current_hour < 8:
yesterday = time.localtime(time.time() - 86400)
current_date = time.strftime("%Y%m%d", yesterday)
date_shift_text = f"日期: {current_date}\n班次: {shift_display}"
self.date_shift_label.setText(date_shift_text)
if self.monitoring:
self.update_countdown_label()
def update_countdown_label(self):
if self.monitoring:
current_time = time.localtime()
next_run_time = self.next_run_time(current_time)
if next_run_time > time.mktime(current_time):
remaining_seconds = int(next_run_time - time.mktime(current_time))
minutes_remaining = remaining_seconds // 60
seconds_remaining = remaining_seconds % 60
countdown_text = f"下次监控将在 {minutes_remaining} 分 {seconds_remaining} 秒后进行..."
self.countdown_label.setText(countdown_text)
else:
self.countdown_label.setText("正在执行监控...")
else:
self.countdown_label.setText(f"下次监控将在 {self.push_interval_minutes} 分钟后进行...")
def next_run_time(self, current_time):
current_hour = current_time.tm_hour
current_minute = current_time.tm_min
current_second = current_time.tm_sec
minutes_since_last_push = current_minute % self.push_interval_minutes
minutes_to_next_push = self.push_interval_minutes - minutes_since_last_push
next_run_time = time.struct_time((
current_time.tm_year,
current_time.tm_mon,
current_time.tm_mday,
current_time.tm_hour,
current_minute + minutes_to_next_push,
0,
current_time.tm_wday,
current_time.tm_yday,
current_time.tm_isdst
))
return time.mktime(next_run_time)
def open_path_manager(self):
if self.path_manager_dialog is None:
self.path_manager_dialog = PathManagerDialog(self)
self.path_manager_dialog.accepted.connect(self.on_path_manager_accepted)
self.path_manager_dialog.rejected.connect(self.on_path_manager_rejected)
self.path_manager_dialog.show()
def on_path_manager_accepted(self):
self.generate_ui_elements()
self.path_manager_dialog = None
def on_path_manager_rejected(self):
self.path_manager_dialog = None
def open_config_manager(self):
if self.config_manager_dialog is None:
self.config_manager_dialog = ConfigManagerDialog(self)
self.config_manager_dialog.accepted.connect(self.on_config_manager_accepted)
self.config_manager_dialog.rejected.connect(self.on_config_manager_rejected)
self.config_manager_dialog.show()
def on_config_manager_accepted(self):
self.generate_ui_elements()
self.config_manager_dialog = None
def on_config_manager_rejected(self):
self.config_manager_dialog = None
def check_and_open_path(self, line_name, machine_name):
current_time = time.localtime()
current_hour = current_time.tm_hour
if (current_hour >= 8 and current_hour < 20):
current_shift = "DayShift"
elif (current_hour >= 20 or current_hour < 8):
current_shift = "NightShift"
else:
current_shift = "NightShift" # 修改这里，使0点至8点使用夜班逻辑
ng_template = LINES_MACHINES_PATH_TEMPLATES.get(line_name, {}).get(machine_name, {}).get(current_shift, {}).get(
"NG", "")
if not ng_template:
QMessageBox.warning(self, "警告", "路径模板未配置或无效")
return
date_formats = ["%Y%m%d", "%Y-%m-%d", "%y%m%d", "%Y_%m_%d"] # 新增 %Y_%m_%d 格式
found_path = None
if current_shift == "NightShift":
current_time = time.localtime(time.time() - 86400)
for fmt in date_formats:
try:
current_date = time.strftime(fmt, current_time)
path = ng_template.format(zsz=current_date)
if os.path.exists(path):
found_path = path
break
except ValueError:
pass
if found_path:
url = QUrl.fromLocalFile(found_path)
QDesktopServices.openUrl(url)
else:
QMessageBox.warning(self, "警告", "未找到有效的路径")
def send_notification(self, results):
headers = {'Content-Type': 'application/json'}
for line, line_results in results.items():
message_content = f"产线: {line}\n"
for machine, machine_results in line_results.items():
ng_count = machine_results["ng_count"]
folder_counts = machine_results["folder_counts"]
message_content += f"\n机台: {machine}\n 不良有{ng_count}类\n"
for folder_name, count in folder_counts.items():
message_content += f" 分类: {folder_name}, 数量: {count}\n"
payload = {
"msgtype": "text",
"text": {
"content": message_content
}
}
response = requests.post(self.webhook_url, headers=headers, data=json.dumps(payload))
if response.status_code == 200:
log_message(f"通知发送成功 - 产线: {line}")
else:
log_message(f"通知发送失败 - 产线: {line}, 错误: {response.text}")
def switch_to_next_tab(self):
if self.tab_widget.count() > 0:
self.current_tab_index = (self.current_tab_index + 1) % self.tab_widget.count()
self.tab_widget.setCurrentIndex(self.current_tab_index)
def mousePressEvent(self, event):
super().mousePressEvent(event)
self.pause_auto_tab_switch()
def pause_auto_tab_switch(self):
self.auto_tab_switch_timer.stop()
self.resume_timer = QTimer(self)
self.resume_timer.singleShot(5000, self.resume_auto_tab_switch)
def resume_auto_tab_switch(self):
self.auto_tab_switch_timer.start(10000)
if __name__ == '__main__':
app = QApplication(sys.argv)
monitor_app = MonitorApp()
monitor_app.showMaximized()
sys.exit(app.exec_())
发表于 2025-6-9 09:34
|
发表于 2025-6-11 14:01
