You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1334 lines
54 KiB
Python

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

import sys
import os
from PyQt5.QtCore import Qt, QTimer, QAbstractTableModel, QModelIndex, QPoint, QSize, pyqtSignal, QFile,QTextStream
from PyQt5.QtGui import QBrush, QColor, QFont, QStandardItemModel, QStandardItem
from PyQt5.QtWidgets import (QApplication, QMainWindow, QTableView, QToolButton,
QPushButton, QVBoxLayout, QWidget, QHBoxLayout,
QLabel, QCheckBox, QSpinBox, QMenu, QFileDialog,
QTabWidget, QTabBar, QListWidget, QListWidgetItem, QDialog,
QLineEdit, QFormLayout, QDialogButtonBox, QMessageBox,
QHeaderView, QToolBar, QAction, QStatusBar, QSplitter, QAbstractItemView, QFrame,
QProgressDialog)
from docx import Document
from docx.shared import Pt, RGBColor
from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
import qtawesome as qta
from datetime import datetime
# 导入其他模块
from model.ProcedureModel.ProcedureProcessor import ExcelParser
from utils.DBModels.ProcedureModel import DatabaseManager
from UI.ProcedureManager.HistoryViewer import HistoryViewerWidget
from UI.ProcedureManager.StepExecutor import StepExecutor # 修改导入路径
from UI.ProcedureManager.KeywordManager import KeywordManagerWidget # 新增导入
from UI.ProcedureManager.ProcedureEditor import ProcedureEditor # 新增导入
class ExecutionDetailDialog(QDialog):
"""步骤详情对话框"""
def __init__(self, step_info, parent=None):
super().__init__(parent)
self.setWindowTitle("步骤详情")
self.setMinimumWidth(600)
layout = QVBoxLayout()
form_layout = QFormLayout()
form_layout.addRow("步骤ID", QLabel(step_info['step_id']))
form_layout.addRow("步骤描述", QLabel(step_info['step_description']))
form_layout.addRow("执行时间", QLabel(step_info['execution_time']))
form_layout.addRow("执行结果", QLabel("成功" if step_info['result'] else "失败"))
layout.addLayout(form_layout)
button_box = QDialogButtonBox(QDialogButtonBox.Ok)
button_box.accepted.connect(self.accept)
layout.addWidget(button_box)
self.setLayout(layout)
class AddCategoryDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("添加新分类")
self.setFixedSize(300, 150)
layout = QVBoxLayout()
form_layout = QFormLayout()
self.nameEdit = QLineEdit()
self.nameEdit.setPlaceholderText("输入分类名称")
form_layout.addRow("分类名称:", self.nameEdit)
layout.addLayout(form_layout)
button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
button_box.accepted.connect(self.accept)
button_box.rejected.connect(self.reject)
layout.addWidget(button_box)
self.setLayout(layout)
def getCategoryName(self):
return self.nameEdit.text().strip()
class ProcedureManager(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("规程管理系统")
self.setGeometry(100, 100, 1200, 800)
# 初始化数据库
self.initDB()
# 创建顶部按钮布局
self.createTopButtonLayout()
def setupButtonIcon(self, button, icon_name, color):
"""设置按钮图标,支持不同状态的颜色变化"""
# 正常状态图标
normal_icon = qta.icon(icon_name, color=color)
# 悬停状态图标(稍微亮一些)
hover_color = self.lightenColor(color, 0.2)
hover_icon = qta.icon(icon_name, color=hover_color)
# 按下状态图标(稍微暗一些)
pressed_color = self.darkenColor(color, 0.2)
pressed_icon = qta.icon(icon_name, color=pressed_color)
# 设置图标
button.setIcon(normal_icon)
# 存储不同状态的图标,用于状态切换
button._normal_icon = normal_icon
button._hover_icon = hover_icon
button._pressed_icon = pressed_icon
# 连接事件
button.enterEvent = lambda event: self.onButtonEnter(button, event)
button.leaveEvent = lambda event: self.onButtonLeave(button, event)
button.mousePressEvent = lambda event: self.onButtonPress(button, event)
button.mouseReleaseEvent = lambda event: self.onButtonRelease(button, event)
def lightenColor(self, color, factor):
"""使颜色变亮"""
if color.startswith('#'):
# 十六进制颜色
r = int(color[1:3], 16)
g = int(color[3:5], 16)
b = int(color[5:7], 16)
r = min(255, int(r + (255 - r) * factor))
g = min(255, int(g + (255 - g) * factor))
b = min(255, int(b + (255 - b) * factor))
return f"#{r:02x}{g:02x}{b:02x}"
return color
def darkenColor(self, color, factor):
"""使颜色变暗"""
if color.startswith('#'):
# 十六进制颜色
r = int(color[1:3], 16)
g = int(color[3:5], 16)
b = int(color[5:7], 16)
r = max(0, int(r * (1 - factor)))
g = max(0, int(g * (1 - factor)))
b = max(0, int(b * (1 - factor)))
return f"#{r:02x}{g:02x}{b:02x}"
return color
def onButtonEnter(self, button, event):
"""按钮鼠标进入事件"""
if hasattr(button, '_hover_icon'):
button.setIcon(button._hover_icon)
# 调用原始的enterEvent
QPushButton.enterEvent(button, event)
def onButtonLeave(self, button, event):
"""按钮鼠标离开事件"""
if hasattr(button, '_normal_icon'):
button.setIcon(button._normal_icon)
# 调用原始的leaveEvent
QPushButton.leaveEvent(button, event)
def onButtonPress(self, button, event):
"""按钮鼠标按下事件"""
if hasattr(button, '_pressed_icon'):
button.setIcon(button._pressed_icon)
# 调用原始的mousePressEvent
QPushButton.mousePressEvent(button, event)
def onButtonRelease(self, button, event):
"""按钮鼠标释放事件"""
# 检查鼠标是否还在按钮上
if button.rect().contains(event.pos()):
if hasattr(button, '_hover_icon'):
button.setIcon(button._hover_icon)
else:
if hasattr(button, '_normal_icon'):
button.setIcon(button._normal_icon)
# 调用原始的mouseReleaseEvent
QPushButton.mouseReleaseEvent(button, event)
# 添加状态栏
def initUI(self):
# 创建主容器
mainContainer = QWidget()
mainLayout = QVBoxLayout()
mainLayout.setSpacing(0)
mainLayout.setContentsMargins(0, 0, 0, 0)
# 添加顶部按钮布局
mainLayout.addWidget(self.topButtonWidget)
# 创建主控件 - 已经是QTabWidget
self.tabs = QTabWidget()
# 设置标签页样式,消除边距
self.tabs.setStyleSheet("""
QTabWidget::pane {
border: none;
margin: 0px;
padding: 0px;
}
QTabWidget::tab-bar {
alignment: left;
}
""")
mainLayout.addWidget(self.tabs)
# 设置主容器
mainContainer.setLayout(mainLayout)
self.setCentralWidget(mainContainer)
# 启用标签页关闭按钮
self.tabs.setTabsClosable(True)
self.tabs.tabCloseRequested.connect(self.closeTab) # 新增关闭标签页信号连接
# 添加初始的规程管理标签页必须先创建UI组件
self.initProcedureManagementTab() # 先创建UI组件
# 初始化其他组件在UI创建后调用
self.loadCategories() # 现在categoryList已存在
# 新增:确保加载第一个分类的规程
if self.categoryList.count() > 0:
self.categorySelected(self.categoryList.item(0), None)
# 初始化标签锁定状态
self.activeExecutorIndex = -1
self.loadStylesheet()
self.statusBar = QStatusBar()
self.setStatusBar(self.statusBar)
# 批量执行相关变量
self.batchExecutionQueue = [] # 批量执行队列
self.currentBatchIndex = 0 # 当前执行的规程索引
self.isBatchExecuting = False # 是否正在批量执行
self.batchExecutionTimer = QTimer() # 批量执行检查定时器
self.batchExecutionTimer.timeout.connect(self.checkBatchExecutionStatus)
self.currentBatchTabIndex = -1 # 当前批量执行的标签页索引
def initDB(self):
# 初始化数据库
self.db = DatabaseManager()
# 新增:关闭标签页的处理方法
def closeTab(self, index):
# 规程管理标签页(索引0)不能关闭
if index == 0:
return
widget = self.tabs.widget(index)
# 检查是否有正在运行的执行器
if isinstance(widget, StepExecutor) and widget.isRunning:
QMessageBox.warning(
self,
"操作被阻止",
"当前有规程正在执行中,请先停止执行后再关闭标签页。"
)
return
# 如果是当前激活的执行器标签页,解锁所有标签页
if index == self.activeExecutorIndex:
self.lockTabs(False, index)
# 移除标签页
self.tabs.removeTab(index)
widget.deleteLater()
# 修改:锁定/解锁标签页方法(增加关闭按钮控制)
def lockTabs(self, lock, currentIndex):
"""锁定或解锁标签页只用Qt原生关闭按钮方案"""
self.activeExecutorIndex = currentIndex if lock else -1
self.tabs.setTabsClosable(True)
for index in range(self.tabs.count()):
enable = not lock or index == currentIndex
self.tabs.setTabEnabled(index, enable)
if index == 0:
self.tabs.setTabEnabled(0, True)
if index > 0:
if lock and index != currentIndex:
self.tabs.tabBar().setTabButton(index, QTabBar.RightSide, None)
# 解锁时强制恢复所有关闭按钮
if not lock:
self.tabs.setTabsClosable(False)
self.tabs.setTabsClosable(True)
# 修改:打开规程执行界面(显示规程全名)
def openProcedureInExecutor(self, item=None):
# 如果通过工具栏调用item为None使用当前选中项
currentItem = item or self.procedureList.currentItem()
if not currentItem:
return
# 检查是否有正在运行的执行器
if self.hasRunningExecutor():
QMessageBox.warning(
self,
"操作被阻止",
"当前有规程正在执行中,请先停止执行后再打开其他规程。"
)
return
procId = currentItem.data(Qt.UserRole) # 修改变量名
# print(procId, 22222222222222)
procedureData = self.db.getProcedureContent(procId) # 修改变量名
if not procedureData:
QMessageBox.warning(self, "打开失败", "无法获取规程内容")
return
# 创建新的执行界面传入规程ID和数据库管理器
executor = StepExecutor(procedureData, procId, self.db)
# 获取规程名称和编号
procName = procedureData["规程信息"]["规程名称"]
procNumber = procedureData["规程信息"]["规程编号"]
# 添加为新标签页,显示规程全名(名称+编号)
tab_index = self.tabs.addTab(executor, f"{procName} ({procNumber})")
# 记录当前执行器的标签索引
executor.tabLockRequired.connect(lambda lock: self.lockTabs(lock, tab_index))
# 切换到新添加的标签页
self.tabs.setCurrentWidget(executor)
# 修改:只在多轮次执行完成时自动重置,单轮次执行完成时不重置
executor.executionFinished.connect(lambda executor_instance: self.handleExecutionFinished(executor_instance))
self.statusBar.showMessage(f"已打开规程执行界面: {procName}", 3000)
def refreshProcedureList(self):
"""刷新规程列表"""
currentCategoryItem = self.categoryList.currentItem()
categoryId = currentCategoryItem.data(Qt.UserRole) if currentCategoryItem else None
self.loadProcedures(categoryId)
def initProcedureManagementTab(self):
"""创建规程管理主标签页 - 现代化卡片式设计"""
mainWidget = QWidget()
mainWidget.setStyleSheet("background-color: #F5F7FA;")
# 主布局 - 删除边距以消除空白区域
mainLayout = QHBoxLayout()
mainLayout.setSpacing(4)
mainLayout.setContentsMargins(0, 0, 0, 0)
# 左侧分类卡片
categoryCard = QWidget()
categoryCard.setObjectName("categoryCard")
categoryLayout = QVBoxLayout()
categoryLayout.setSpacing(8)
categoryLayout.setContentsMargins(8, 8, 8, 8)
# 分类标题
categoryTitle = QLabel("规程分类")
categoryTitle.setObjectName("cardTitle")
categoryLayout.addWidget(categoryTitle)
# 分类列表
self.categoryList = QListWidget()
self.categoryList.setObjectName("categoryList")
self.categoryList.currentItemChanged.connect(self.categorySelected)
# 设置分类列表接受拖放
self.categoryList.setAcceptDrops(True)
self.categoryList.setDragDropMode(QAbstractItemView.DropOnly)
# 设置分类列表右键菜单
self.categoryList.setContextMenuPolicy(Qt.CustomContextMenu)
self.categoryList.customContextMenuRequested.connect(self.showCategoryContextMenu)
categoryLayout.addWidget(self.categoryList)
# 分类操作按钮
categoryCard.setLayout(categoryLayout)
# 右侧规程卡片
procedureCard = QWidget()
procedureCard.setObjectName("procedureCard")
procedureLayout = QVBoxLayout()
procedureLayout.setSpacing(8)
procedureLayout.setContentsMargins(8, 8, 8, 8)
# 规程标题和统计信息
procedureHeaderLayout = QHBoxLayout()
procedureTitle = QLabel("规程列表")
procedureTitle.setObjectName("cardTitle")
# 规程统计标签
self.procedureCountLabel = QLabel("共 0 个规程")
self.procedureCountLabel.setObjectName("procedureCountLabel")
procedureHeaderLayout.addWidget(procedureTitle)
procedureHeaderLayout.addStretch()
procedureHeaderLayout.addWidget(self.procedureCountLabel)
procedureLayout.addLayout(procedureHeaderLayout)
# 搜索和过滤器
searchLayout = QHBoxLayout()
searchLayout.setSpacing(12)
# 搜索框
self.searchEdit = QLineEdit()
self.searchEdit.setObjectName("searchEdit")
self.searchEdit.setPlaceholderText("搜索规程名称或编号...")
self.searchEdit.textChanged.connect(self.filterProcedures)
searchLayout.addWidget(self.searchEdit, 1)
procedureLayout.addLayout(searchLayout)
# 规程列表
self.procedureList = QListWidget()
self.procedureList.setObjectName("procedureList")
# 设置规程列表支持拖拽
self.procedureList.setDragEnabled(True)
self.procedureList.setDragDropMode(QAbstractItemView.DragOnly)
self.procedureList.itemDoubleClicked.connect(
lambda item: self.openProcedureInExecutor(item)
)
# 设置规程列表右键菜单
self.procedureList.setContextMenuPolicy(Qt.CustomContextMenu)
self.procedureList.customContextMenuRequested.connect(self.showProcedureContextMenu)
# 拖拽事件处理
self.procedureList.dragEnterEvent = self.procedureDragEnterEvent
self.categoryList.dropEvent = self.categoryDropEvent
procedureLayout.addWidget(self.procedureList)
procedureCard.setLayout(procedureLayout)
# 添加到主布局
mainLayout.addWidget(categoryCard)
mainLayout.addWidget(procedureCard, 1) # 规程卡片占据剩余空间
mainWidget.setLayout(mainLayout)
self.tabs.addTab(mainWidget, "规程管理")
def createTopButtonLayout(self):
"""创建顶部按钮布局"""
# 创建顶部按钮容器
self.topButtonWidget = QWidget()
self.topButtonWidget.setObjectName("topButtonWidget")
self.topButtonWidget.setFixedHeight(60)
self.topButtonWidget.setStyleSheet("""
QWidget#topButtonWidget {
background-color: #FFFFFF;
border-bottom: 1px solid #E5E7EB;
}
""")
# 创建横向布局
buttonLayout = QHBoxLayout()
buttonLayout.setSpacing(8)
buttonLayout.setContentsMargins(16, 8, 16, 8)
# 导入规程按钮 - 绿色主题
self.importBtn = QPushButton("导入规程")
self.importBtn.setObjectName("importToolBtn")
self.importBtn.setIconSize(QSize(18, 18))
self.importBtn.setStatusTip("导入Excel规程文件")
self.importBtn.clicked.connect(self.importProcedure)
self.setupButtonIcon(self.importBtn, 'fa5s.file-import', '#047857')
buttonLayout.addWidget(self.importBtn)
# 添加分类按钮 - 蓝色主题
self.addCategoryBtn = QPushButton("添加分类")
self.addCategoryBtn.setObjectName("addCategoryToolBtn")
self.addCategoryBtn.setIconSize(QSize(18, 18))
self.addCategoryBtn.setStatusTip("添加新的分类")
self.addCategoryBtn.clicked.connect(self.addCategory)
self.setupButtonIcon(self.addCategoryBtn, 'fa5s.folder-plus', '#1D4ED8')
buttonLayout.addWidget(self.addCategoryBtn)
# 删除分类按钮 - 红色主题
self.deleteCategoryBtn = QPushButton("删除分类")
self.deleteCategoryBtn.setObjectName("deleteCategoryToolBtn")
self.deleteCategoryBtn.setIconSize(QSize(18, 18))
self.deleteCategoryBtn.setStatusTip("删除当前分类")
self.deleteCategoryBtn.clicked.connect(self.deleteCurrentCategory)
self.setupButtonIcon(self.deleteCategoryBtn, 'fa5s.folder-minus', '#B91C1C')
buttonLayout.addWidget(self.deleteCategoryBtn)
# 添加分隔符
separator1 = QFrame()
separator1.setFrameShape(QFrame.VLine)
separator1.setFrameShadow(QFrame.Sunken)
separator1.setStyleSheet("QFrame { color: #E5E7EB; }")
buttonLayout.addWidget(separator1)
# 打开规程按钮 - 蓝色主题
self.openProcedureBtn = QPushButton("打开规程")
self.openProcedureBtn.setObjectName("openProcedureToolBtn")
self.openProcedureBtn.setIconSize(QSize(18, 18))
self.openProcedureBtn.setStatusTip("在步骤执行工具中打开选中的规程")
self.openProcedureBtn.clicked.connect(self.openProcedureInExecutor)
self.setupButtonIcon(self.openProcedureBtn, 'fa5s.folder-open', '#1D4ED8')
buttonLayout.addWidget(self.openProcedureBtn)
# 删除规程按钮 - 红色主题
self.deleteProcedureBtn = QPushButton("删除规程")
self.deleteProcedureBtn.setObjectName("deleteProcedureToolBtn")
self.deleteProcedureBtn.setIconSize(QSize(18, 18))
self.deleteProcedureBtn.setStatusTip("删除选中的规程")
self.deleteProcedureBtn.clicked.connect(self.deleteSelectedProcedure)
self.setupButtonIcon(self.deleteProcedureBtn, 'fa5s.trash', '#B91C1C')
buttonLayout.addWidget(self.deleteProcedureBtn)
# 添加分隔符
separator2 = QFrame()
separator2.setFrameShape(QFrame.VLine)
separator2.setFrameShadow(QFrame.Sunken)
separator2.setStyleSheet("QFrame { color: #E5E7EB; }")
buttonLayout.addWidget(separator2)
# 历史记录按钮 - 紫色主题
self.historyBtn = QPushButton("历史记录")
self.historyBtn.setObjectName("historyToolBtn")
self.historyBtn.setIconSize(QSize(18, 18))
self.historyBtn.setStatusTip("查看历史执行记录")
self.historyBtn.clicked.connect(self.openHistoryViewer)
self.setupButtonIcon(self.historyBtn, 'fa5s.history', '#7C3AED')
buttonLayout.addWidget(self.historyBtn)
# 批量执行按钮 - 绿色主题
self.batchExecuteBtn = QPushButton("批量执行")
self.batchExecuteBtn.setObjectName("batchExecuteToolBtn")
self.batchExecuteBtn.setIconSize(QSize(18, 18))
self.batchExecuteBtn.setStatusTip("按顺序批量执行当前分类中的所有规程")
self.batchExecuteBtn.clicked.connect(self.batchExecuteProcedures)
self.setupButtonIcon(self.batchExecuteBtn, 'fa5s.play-circle', '#047857')
buttonLayout.addWidget(self.batchExecuteBtn)
# 关键词管理按钮 - 紫色主题
self.keywordManageBtn = QPushButton("关键词管理")
self.keywordManageBtn.setObjectName("keywordManageToolBtn")
self.keywordManageBtn.setIconSize(QSize(18, 18))
self.keywordManageBtn.setStatusTip("管理执行步骤关键词字段库")
self.keywordManageBtn.clicked.connect(self.openKeywordManager)
self.setupButtonIcon(self.keywordManageBtn, 'fa5s.key', '#7C3AED')
buttonLayout.addWidget(self.keywordManageBtn)
# 导出规程按钮 - 橙色主题
self.exportProcedureBtn = QPushButton("导出规程")
self.exportProcedureBtn.setObjectName("exportProcedureToolBtn")
self.exportProcedureBtn.setIconSize(QSize(18, 18))
self.exportProcedureBtn.setStatusTip("导出规程为Excel文件")
self.exportProcedureBtn.clicked.connect(self.exportSelectedProcedure)
self.setupButtonIcon(self.exportProcedureBtn, 'fa5s.file-export', '#D97706')
buttonLayout.addWidget(self.exportProcedureBtn)
# 添加弹性空间,将按钮推到左侧
buttonLayout.addStretch()
# 设置布局
self.topButtonWidget.setLayout(buttonLayout)
def loadCategories(self):
self.categoryList.clear()
categories = self.db.getCategories()
for catId, catName in categories:
item = QListWidgetItem(catName)
item.setData(Qt.UserRole, catId)
self.categoryList.addItem(item)
if self.categoryList.count() > 0:
self.categoryList.setCurrentRow(0)
# 新增:手动触发分类选择信号
self.categoryList.itemSelectionChanged.emit()
def loadProcedures(self, categoryId=None):
self.procedureList.clear()
procedures = self.db.getProcedures(categoryId)
# 存储原始数据用于过滤
self.allProcedures = procedures
for procId, name, number, type, createdAt in procedures:
item = QListWidgetItem(f"{name} ({number})")
item.setData(Qt.UserRole, procId)
item.setToolTip(f"类型: {type}\n创建时间: {createdAt}")
self.procedureList.addItem(item)
# 更新规程统计信息
count = len(procedures)
if hasattr(self, 'procedureCountLabel'):
self.procedureCountLabel.setText(f"{count} 个规程")
def filterProcedures(self):
"""过滤规程列表"""
if not hasattr(self, 'allProcedures') or not hasattr(self, 'searchEdit'):
return
searchText = self.searchEdit.text().lower()
self.procedureList.clear()
filteredCount = 0
for procId, name, number, type, createdAt in self.allProcedures:
# 文本搜索过滤
if searchText and searchText not in name.lower() and searchText not in number.lower():
continue
# 添加到列表
item = QListWidgetItem(f"{name} ({number})")
item.setData(Qt.UserRole, procId)
item.setToolTip(f"类型: {type}\n创建时间: {createdAt}")
self.procedureList.addItem(item)
filteredCount += 1
# 更新统计信息
if hasattr(self, 'procedureCountLabel'):
totalCount = len(self.allProcedures)
if filteredCount == totalCount:
self.procedureCountLabel.setText(f"{totalCount} 个规程")
else:
self.procedureCountLabel.setText(f"显示 {filteredCount} / {totalCount} 个规程")
def categorySelected(self, currentItem, previousItem):
if currentItem:
categoryId = currentItem.data(Qt.UserRole)
self.loadProcedures(categoryId)
def importProcedure(self):
"""导入规程 - 支持多文件选择"""
filePaths, _ = QFileDialog.getOpenFileNames(
self,
"选择规程文件",
"",
"Excel文件 (*.xlsx *.xls)"
)
if not filePaths:
return
currentItem = self.categoryList.currentItem()
categoryId = currentItem.data(Qt.UserRole) if currentItem else None
# 统计导入结果
totalFiles = len(filePaths)
successCount = 0
skipCount = 0
errorCount = 0
errorDetails = []
# 批量导入确认
if totalFiles > 1:
reply = QMessageBox.question(
self, "批量导入确认",
f"您选择了 {totalFiles} 个文件进行导入。\n\n"
f"导入过程中如果遇到同名规程,将会逐个询问是否覆盖。\n"
f"是否继续批量导入?",
QMessageBox.Yes | QMessageBox.No
)
if reply != QMessageBox.Yes:
return
# 创建进度对话框
progress = None
if totalFiles > 1:
progress = QProgressDialog("正在导入规程...", "取消", 0, totalFiles, self)
progress.setWindowModality(Qt.WindowModal)
progress.setMinimumDuration(0)
progress.show()
# 逐个处理文件
for index, filePath in enumerate(filePaths):
if progress:
progress.setValue(index)
progress.setLabelText(f"正在导入: {os.path.basename(filePath)}")
# 检查是否取消
if progress.wasCanceled():
break
try:
result = self.importSingleProcedure(filePath, categoryId, totalFiles > 1)
if result == "success":
successCount += 1
elif result == "skip":
skipCount += 1
elif result == "cancel":
# 用户取消了整个导入过程
break
except Exception as e:
errorCount += 1
fileName = os.path.basename(filePath)
errorDetails.append(f"{fileName}: {str(e)}")
print(f"导入文件 {fileName} 时发生错误: {str(e)}")
# 关闭进度对话框
if progress:
progress.close()
# 刷新规程列表
if successCount > 0:
self.loadProcedures(categoryId)
# 显示导入结果
self.showImportResults(totalFiles, successCount, skipCount, errorCount, errorDetails)
def importSingleProcedure(self, filePath, categoryId, isBatchMode=False):
"""导入单个规程文件
Args:
filePath: 文件路径
categoryId: 分类ID
isBatchMode: 是否为批量模式
Returns:
str: "success", "skip", "cancel", "error"
"""
try:
parsedData = ExcelParser.parseProcedure(filePath)
procInfo = parsedData["规程信息"]
procedureName = procInfo["规程名称"]
procedureNumber = procInfo["规程编号"]
fileName = os.path.basename(filePath)
# 检查是否存在同名规程
existingProcedure = self.db.getProcedureByNameAndNumber(procedureName, procedureNumber)
if existingProcedure:
# 存在同名规程,询问是否覆盖
existingId, existingName, existingNumber, existingType, existingCategoryId = existingProcedure
if isBatchMode:
# 批量模式下的对话框包含更多选项
msgBox = QMessageBox(self)
msgBox.setWindowTitle("发现同名规程")
msgBox.setText(f"文件: {fileName}\n\n"
f"发现同名规程:\n"
f"规程名称:{existingName}\n"
f"规程编号:{existingNumber}\n"
f"规程类型:{existingType}\n\n"
f"请选择操作:")
overwriteBtn = msgBox.addButton("覆盖", QMessageBox.YesRole)
skipBtn = msgBox.addButton("跳过", QMessageBox.NoRole)
cancelBtn = msgBox.addButton("取消全部", QMessageBox.RejectRole)
msgBox.exec_()
clickedBtn = msgBox.clickedButton()
if clickedBtn == cancelBtn:
return "cancel"
elif clickedBtn == skipBtn:
return "skip"
elif clickedBtn == overwriteBtn:
# 覆盖现有规程
success = self.db.updateProcedureById(
existingId, categoryId, procedureName, procedureNumber,
procInfo["规程类型"], parsedData, ""
)
if success:
return "success"
else:
raise Exception("无法覆盖规程,请检查数据库连接")
else:
# 单文件模式的原有逻辑
reply = QMessageBox.question(
self, "发现同名规程",
f"发现同名规程:\n"
f"规程名称:{existingName}\n"
f"规程编号:{existingNumber}\n"
f"规程类型:{existingType}\n\n"
f"是否要覆盖这个规程?\n"
f"覆盖后原规程内容将丢失。",
QMessageBox.Yes | QMessageBox.No | QMessageBox.Cancel
)
if reply == QMessageBox.Cancel:
return "cancel"
elif reply == QMessageBox.Yes:
# 覆盖现有规程
success = self.db.updateProcedureById(
existingId, categoryId, procedureName, procedureNumber,
procInfo["规程类型"], parsedData, ""
)
if success:
self.statusBar.showMessage(f"成功覆盖规程: {procedureName}", 5000)
return "success"
else:
raise Exception("无法覆盖规程,请检查数据库连接")
else:
return "skip"
# 不存在同名规程,正常添加
procId = self.db.addProcedure(
categoryId,
procedureName,
procedureNumber,
procInfo["规程类型"],
parsedData,
"" # 显式传递空字符串作为report_path
)
if procId:
if not isBatchMode:
self.statusBar.showMessage(f"成功导入规程: {procedureName}", 5000)
return "success"
else:
raise Exception("无法导入规程,请检查数据库连接")
except Exception as e:
if not isBatchMode:
QMessageBox.critical(self, "导入错误", f"导入规程时发生错误:\n{str(e)}")
raise e
def showImportResults(self, totalFiles, successCount, skipCount, errorCount, errorDetails):
"""显示导入结果统计"""
if totalFiles == 1:
# 单文件导入,使用原有的消息显示方式
return
# 多文件导入结果统计
resultMsg = f"批量导入完成!\n\n"
resultMsg += f"总文件数: {totalFiles}\n"
resultMsg += f"成功导入: {successCount}\n"
resultMsg += f"跳过文件: {skipCount}\n"
resultMsg += f"导入失败: {errorCount}\n"
if errorCount > 0:
resultMsg += f"\n失败详情:\n"
for detail in errorDetails[:5]: # 最多显示5个错误
resultMsg += f"{detail}\n"
if len(errorDetails) > 5:
resultMsg += f"• ... 还有 {len(errorDetails) - 5} 个错误\n"
# 根据结果选择消息类型
if errorCount == 0:
if successCount > 0:
QMessageBox.information(self, "导入成功", resultMsg)
self.statusBar.showMessage(f"批量导入完成: 成功 {successCount}", 5000)
else:
QMessageBox.information(self, "导入完成", resultMsg)
else:
QMessageBox.warning(self, "导入完成(有错误)", resultMsg)
def addCategory(self):
dialog = AddCategoryDialog(self)
if dialog.exec_() == QDialog.Accepted:
categoryName = dialog.getCategoryName()
if categoryName:
if self.db.addCategory(categoryName):
self.loadCategories()
else:
QMessageBox.warning(self, "添加失败", "分类名称已存在,请使用其他名称")
def deleteCurrentCategory(self):
currentItem = self.categoryList.currentItem()
if not currentItem:
return
categoryId = currentItem.data(Qt.UserRole)
categoryName = currentItem.text()
# 检查是否为默认分类
if categoryName == "默认分类":
QMessageBox.warning(
self,
"操作禁止",
"默认分类不可删除!"
)
return
reply = QMessageBox.question(
self,
"确认删除",
f"确定要删除分类 '{categoryName}' 吗?\n该分类下的规程将移动到默认分类。",
QMessageBox.Yes | QMessageBox.No
)
if reply == QMessageBox.Yes:
if self.db.deleteCategory(categoryId):
self.loadCategories()
self.statusBar.showMessage(f"已删除分类: {categoryName}", 5000)
def deleteSelectedProcedure(self):
currentItem = self.procedureList.currentItem()
if not currentItem:
return
procId = currentItem.data(Qt.UserRole)
procName = currentItem.text()
reply = QMessageBox.question(
self,
"确认删除",
f"确定要删除规程 '{procName}' 吗?",
QMessageBox.Yes | QMessageBox.No
)
if reply == QMessageBox.Yes:
if self.db.deleteProcedure(procId):
currentCategoryItem = self.categoryList.currentItem()
categoryId = currentCategoryItem.data(Qt.UserRole) if currentCategoryItem else None
self.loadProcedures(categoryId)
self.statusBar.showMessage(f"已删除规程: {procName}", 5000)
def hasRunningExecutor(self):
"""检查当前是否有正在运行的执行器 - 修正实现"""
for index in range(self.tabs.count()):
widget = self.tabs.widget(index)
# 修正: 只检查isRunning状态移除对isActive的检查
if isinstance(widget, StepExecutor) and widget.isRunning:
return True
return False
# 新增:打开历史记录查看器
def openHistoryViewer(self):
"""打开历史记录查看器"""
historyViewer = HistoryViewerWidget(self.db, self)
# 创建新标签页
tabIndex = self.tabs.addTab(historyViewer, "历史记录")
self.tabs.setCurrentIndex(tabIndex)
self.statusBar.showMessage("已打开历史记录查看器", 3000)
def loadStylesheet(self):
qssPath = "Static/Procedure.qss"
# 回退到原始QSS
qssFile = QFile(qssPath)
if qssFile.exists():
qssFile.open(QFile.ReadOnly | QFile.Text)
stream = QTextStream(qssFile)
stream.setCodec("UTF-8")
self.setStyleSheet(stream.readAll())
qssFile.close()
# print(f"成功加载样式表: {qssPath}")
# else:
# print(f"警告:样式表文件不存在: {qssPath}")
def closeEvent(self, event):
self.db.close()
event.accept()
def procedureDragEnterEvent(self, event):
"""处理规程列表的拖拽进入事件"""
if event.mimeData().hasText():
event.acceptProposedAction()
def categoryDropEvent(self, event):
"""处理分类列表的拖放事件:将拖拽过来的规程移动到当前分类"""
# 获取拖拽源
source = event.source()
# 确保拖拽源是规程列表
if source != self.procedureList:
event.ignore()
return
# 获取选中的规程项(拖拽项)
items = self.procedureList.selectedItems()
if not items:
event.ignore()
return
# 获取目标分类项(鼠标释放位置对应的分类项)
pos = event.pos()
targetItem = self.categoryList.itemAt(pos)
if not targetItem:
event.ignore()
return
# 获取目标分类ID
targetCategoryId = targetItem.data(Qt.UserRole)
# 获取被拖拽的规程ID只处理第一个选中的项
procItem = items[0]
procId = procItem.data(Qt.UserRole)
procName = procItem.text()
# 更新数据库
if self.db.updateProcedureCategory(procId, targetCategoryId):
# 更新成功,重新加载目标分类的规程列表(因为规程已经移动到新分类)
self.loadProcedures(targetCategoryId)
# 新增:设置当前选中的分类为目标分类
self.categoryList.setCurrentItem(targetItem)
# 显示状态信息
self.statusBar.showMessage(f"已将规程 '{procName}' 移动到分类 '{targetItem.text()}'", 5000)
else:
QMessageBox.warning(self, "移动失败", "移动规程失败,请检查数据库连接")
event.accept()
# 新增:分类列表右键菜单
def showCategoryContextMenu(self, pos):
"""显示分类列表的右键菜单"""
item = self.categoryList.itemAt(pos)
if item:
menu = QMenu()
deleteAction = menu.addAction("删除分类")
deleteAction.triggered.connect(self.deleteCurrentCategory)
menu.exec_(self.categoryList.mapToGlobal(pos))
# 新增:规程列表右键菜单
def showProcedureContextMenu(self, pos):
"""显示规程列表的右键菜单"""
item = self.procedureList.itemAt(pos)
menu = QMenu()
if item:
# 编辑规程
editAction = menu.addAction(qta.icon('fa5s.edit', color='orange'), "编辑规程")
editAction.triggered.connect(self.editSelectedProcedure)
# 导出规程
exportAction = menu.addAction(qta.icon('fa5s.file-export', color='green'), "导出规程")
exportAction.triggered.connect(self.exportSelectedProcedure)
menu.addSeparator()
# 删除规程
deleteAction = menu.addAction(qta.icon('fa5s.trash', color='red'), "删除规程")
deleteAction.triggered.connect(self.deleteSelectedProcedure)
else:
# 在空白区域右键时显示新建规程
pass
menu.exec_(self.procedureList.mapToGlobal(pos))
# 批量执行相关方法
def batchExecuteProcedures(self):
"""开始批量执行当前分类中的所有规程"""
# 检查是否有正在运行的执行器
if self.hasRunningExecutor():
QMessageBox.warning(
self,
"操作被阻止",
"当前有规程正在执行中,请先停止执行后再开始批量执行。"
)
return
# 获取当前分类的所有规程
currentItem = self.categoryList.currentItem()
if not currentItem:
QMessageBox.warning(self, "未选择分类", "请先选择一个分类")
return
categoryId = currentItem.data(Qt.UserRole)
procedures = self.db.getProcedures(categoryId)
if not procedures:
QMessageBox.information(self, "无规程", "当前分类中没有规程")
return
# 确认批量执行
reply = QMessageBox.question(
self,
"确认批量执行",
f"确定要批量执行当前分类中的 {len(procedures)} 个规程吗?\n"
f"规程将按顺序逐个执行,每个规程执行完毕后自动开始下一个。",
QMessageBox.Yes | QMessageBox.No
)
if reply == QMessageBox.Yes:
# 初始化批量执行队列
self.batchExecutionQueue = [(procId, name, number) for procId, name, number, type, createdAt in procedures]
self.currentBatchIndex = 0
self.isBatchExecuting = True
# 更新状态
self.statusBar.showMessage(f"开始批量执行 {len(procedures)} 个规程", 3000)
# 开始执行第一个规程
self.executeNextBatchProcedure()
def executeNextBatchProcedure(self):
"""执行下一个规程"""
print(f"executeNextBatchProcedure: 批量执行={self.isBatchExecuting}, 当前索引={self.currentBatchIndex}, 队列长度={len(self.batchExecutionQueue)}")
if not self.isBatchExecuting or self.currentBatchIndex >= len(self.batchExecutionQueue):
print("批量执行结束条件满足调用finishBatchExecution")
self.finishBatchExecution()
return
# 获取当前要执行的规程
procId, name, number = self.batchExecutionQueue[self.currentBatchIndex]
print(f"准备执行规程: {name} ({number}) - {self.currentBatchIndex + 1}/{len(self.batchExecutionQueue)}")
# 获取规程数据
procedureData = self.db.getProcedureContent(procId)
if not procedureData:
QMessageBox.warning(self, "获取规程失败", f"无法获取规程 {name} 的内容")
self.currentBatchIndex += 1
self.executeNextBatchProcedure()
return
# 创建执行器
executor = StepExecutor(procedureData, procId, self.db)
# 设置规程全名
procName = procedureData["规程信息"]["规程名称"]
procNumber = procedureData["规程信息"]["规程编号"]
tab_title = f"批量执行-{procName}({procNumber})"
# 添加为新标签页
tab_index = self.tabs.addTab(executor, tab_title)
self.currentBatchTabIndex = tab_index # 记录当前标签页索引
print(f"创建标签页: {tab_title}, 索引={tab_index}")
# 连接标签锁定信号
executor.tabLockRequired.connect(lambda lock: self.lockTabs(lock, tab_index))
# 切换到新添加的标签页
self.tabs.setCurrentWidget(executor)
# 开始自动执行
executor.startAutoExecute()
print(f"开始自动执行规程: {procName}")
# 启动批量执行状态检查定时器(如果还没启动)
if not self.batchExecutionTimer.isActive():
self.batchExecutionTimer.start(1000) # 每秒检查一次
print("启动批量执行状态检查定时器")
else:
print("批量执行状态检查定时器已在运行")
# 更新状态
self.statusBar.showMessage(f"正在执行: {name} ({number}) - {self.currentBatchIndex + 1}/{len(self.batchExecutionQueue)}", 0)
def checkBatchExecutionStatus(self):
"""检查批量执行状态"""
if not self.isBatchExecuting:
return
# 检查当前是否有正在运行的执行器
has_running = self.hasRunningExecutor()
print(f"批量执行状态检查: 当前索引={self.currentBatchIndex}, 队列长度={len(self.batchExecutionQueue)}, 有运行执行器={has_running}")
if not has_running:
# 如果没有正在运行的执行器,说明当前规程执行完成
print(f"规程执行完成,准备执行下一个规程")
# 确保当前标签页存在且有效
if self.currentBatchTabIndex > 0 and self.currentBatchTabIndex < self.tabs.count():
# 关闭当前执行器标签页
self.tabs.removeTab(self.currentBatchTabIndex)
self.currentBatchTabIndex = -1 # 重置标签页索引
print(f"已关闭标签页,索引={self.currentBatchTabIndex}")
# 移动到下一个规程
self.currentBatchIndex += 1
print(f"移动到下一个规程,新索引={self.currentBatchIndex}")
if self.currentBatchIndex >= len(self.batchExecutionQueue):
# 所有规程执行完成
print("所有规程执行完成")
self.finishBatchExecution()
else:
# 延迟一下再执行下一个规程,让用户看到执行结果
print(f"延迟2秒后执行下一个规程: {self.currentBatchIndex + 1}/{len(self.batchExecutionQueue)}")
# 停止当前定时器,避免重复检查
self.batchExecutionTimer.stop()
QTimer.singleShot(2000, self.executeNextBatchProcedure)
def finishBatchExecution(self):
"""完成批量执行"""
self.isBatchExecuting = False
self.batchExecutionQueue = []
self.currentBatchIndex = 0
self.currentBatchTabIndex = -1 # 重置标签页索引
self.batchExecutionTimer.stop()
# 更新状态
self.statusBar.showMessage("批量执行完成", 5000)
# 显示完成消息
QMessageBox.information(
self,
"批量执行完成",
"所有规程已按顺序执行完毕!"
)
def handleExecutionFinished(self, executor_instance):
"""处理执行完成事件"""
# 检查是否为多轮次执行
if executor_instance.infiniteCycles or executor_instance.remainingCycles > 1:
# 多轮次执行完成,自动重置
QTimer.singleShot(0, lambda: executor_instance.resetExecution(fromAuto=True))
else:
# 单轮次执行完成,不自动重置,让用户手动控制
# 可以在这里添加其他处理逻辑,比如显示完成提示等
pass
def openKeywordManager(self):
"""打开关键词管理界面"""
keywordManager = KeywordManagerWidget(self.db, self)
# 创建新标签页
tabIndex = self.tabs.addTab(keywordManager, "关键词管理")
self.tabs.setCurrentIndex(tabIndex)
self.statusBar.showMessage("已打开关键词管理界面", 3000)
def editSelectedProcedure(self):
"""编辑选中的规程"""
currentItem = self.procedureList.currentItem()
if not currentItem:
QMessageBox.warning(self, "未选择规程", "请先选择一个规程进行编辑")
return
# 获取规程信息
procedureId = currentItem.data(Qt.UserRole)
procedureName = currentItem.text()
# 获取规程数据
procedureData = self.db.getProcedureContent(procedureId)
if not procedureData:
QMessageBox.warning(self, "获取规程失败", f"无法获取规程 {procedureName} 的内容")
return
# 创建规程编辑器
editor = ProcedureEditor(procedureData, procedureId, self.db, self)
# 创建新标签页
tabIndex = self.tabs.addTab(editor, f"编辑规程-{procedureName}")
self.tabs.setCurrentIndex(tabIndex)
self.statusBar.showMessage(f"已打开规程编辑器: {procedureName}", 3000)
def exportSelectedProcedure(self):
"""导出选中的规程"""
currentItem = self.procedureList.currentItem()
if not currentItem:
QMessageBox.warning(self, "未选择规程", "请先选择一个要导出的规程")
return
procId = currentItem.data(Qt.UserRole)
procedureData = self.db.getProcedureContent(procId)
if not procedureData:
QMessageBox.warning(self, "获取失败", "无法获取规程内容")
return
try:
# 选择保存路径
procName = procedureData["规程信息"]["规程名称"]
procNumber = procedureData["规程信息"]["规程编号"]
defaultName = f"{procName}_{procNumber}.xlsx"
filePath, _ = QFileDialog.getSaveFileName(
self, "导出规程", defaultName, "Excel文件 (*.xlsx)"
)
if not filePath:
return
# 创建Excel文件
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill
from openpyxl.utils import get_column_letter
wb = Workbook()
ws = wb.active
ws.title = "测试脚本"
# 设置标题样式
titleFont = Font(bold=True, size=12)
titleAlignment = Alignment(horizontal='center', vertical='center')
titleFill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
# 写入规程信息
procInfo = procedureData["规程信息"]
ws['A1'] = "规程名称"
ws['B1'] = procInfo["规程名称"]
ws['C1'] = "规程编号"
ws['D1'] = procInfo["规程编号"]
ws['E1'] = "规程类型"
ws['F1'] = procInfo["规程类型"]
# 写入测试用例信息
testCaseInfo = procedureData["测试用例信息"]
ws['A2'] = "测试用例"
ws['B2'] = testCaseInfo["测试用例"]
ws['C2'] = "用例编号"
ws['D2'] = testCaseInfo["用例编号"]
ws['E2'] = "工况描述"
ws['F2'] = testCaseInfo["工况描述"]
# 设置表头
headers = ['步骤', '实验步骤', '操作类型', '预期结果', '实际结果', '一致性', '测试时间', '备注']
for col, header in enumerate(headers, 1):
cell = ws.cell(row=4, column=col, value=header)
cell.font = titleFont
cell.alignment = titleAlignment
cell.fill = titleFill
# 写入测试步骤
currentRow = 5
for mainStep in procedureData["测试步骤"]:
# 写入主步骤
ws.cell(row=currentRow, column=1, value=mainStep['步骤ID'])
ws.cell(row=currentRow, column=2, value=mainStep['步骤描述'])
ws.cell(row=currentRow, column=3, value=mainStep['操作类型'])
ws.cell(row=currentRow, column=4, value=mainStep.get('预期结果', ''))
currentRow += 1
# 写入子步骤
for subStep in mainStep.get('子步骤', []):
ws.cell(row=currentRow, column=1, value=subStep['序号'])
ws.cell(row=currentRow, column=2, value=subStep['操作'])
ws.cell(row=currentRow, column=3, value=subStep['操作类型'])
ws.cell(row=currentRow, column=4, value=subStep.get('预期结果', ''))
ws.cell(row=currentRow, column=5, value='') # 实际结果导出时为空
ws.cell(row=currentRow, column=6, value='') # 一致性导出时为空
ws.cell(row=currentRow, column=7, value='') # 测试时间导出时为空
ws.cell(row=currentRow, column=8, value='') # 备注导出时为空
currentRow += 1
# 调整列宽
for col in range(1, len(headers) + 1):
ws.column_dimensions[get_column_letter(col)].width = 15
# 保存文件
wb.save(filePath)
QMessageBox.information(self, "导出成功", f"规程已成功导出到:\n{filePath}")
self.statusBar.showMessage(f"规程已导出: {procName}", 5000)
except Exception as e:
QMessageBox.critical(self, "导出错误", f"导出规程时发生错误:\n{str(e)}")