import os import time import json import shutil import base64 from PyQt6.QtGui import QCloseEvent import win32gui import win32con import win32api from PyQt6 import * from PyQt6.QtCore import * from PyQt6.QtGui import * from PyQt6.QtWidgets import * from ui.Ui_dataSyncForm import Ui_dataSyncForm from common import common from taskModel.taskManager import taskManager from dmGroupModel.dmGroupManager import dmGroupManager from deviceModel.deviceManager import deviceManager from devGroupModel.devGroupManager import devGroupManager from devmodelModel.devModelManager import devModelManager from taskGroupModel.taskGroupManager import taskGroupManager from instructionModel.instructionManager import instructionManager from taskInstructionModel.taskInstructionManager import taskInstructionManager from config import config from models import Session from logs import log class DataSyncForm(QWidget): def __init__(self, parent=None): super().__init__(parent) self.ui = Ui_dataSyncForm() self.ui.setupUi(self) # 创建一个文本编辑器来显示 Git 命令的输出 self.outputTextEdit = QTextEdit() self.outputTextEdit.setReadOnly(True) self.gitProcess = QProcess(self) self.hwnd = None self.topWidget = None self.fileSystemModel1 = QFileSystemModel() self.fileSystemModel2 = QFileSystemModel() self.git_path = config.data["project"]["git_path"] self.platform_path1 = config.data["project"]["platform_path1"] self.platform_path2 = config.data["project"]["platform_path2"] self.fileSystemModel1.setRootPath(self.platform_path1) self.ui.treeView1.setModel(self.fileSystemModel1) self.fileSystemModel2.setRootPath(self.platform_path2) self.ui.treeView2.setModel(self.fileSystemModel2) self.ui.treeView1.setRootIndex(self.fileSystemModel1.index(self.platform_path1)) self.ui.treeView2.setModel(self.fileSystemModel2) self.ui.treeView2.setRootIndex(self.fileSystemModel2.index(self.platform_path2)) self.ui.pbGitManager1.clicked.connect(self.startGitGui1) self.ui.pbGitManager2.clicked.connect(self.startGitGui2) self.ui.pbImport.clicked.connect(self.importData) self.ui.pbGitSync.clicked.connect(self.gitSync) self.ui.pbDataSync.clicked.connect(self.dataSync) self.ui.treeView1.resizeEvent = self.onResize self.ui.treeView2.resizeEvent = self.onResize self.destroyed.connect(self.cleanup) def startGitGui1(self): # 假设 git.exe 位于系统路径中 self.gitProcess.setWorkingDirectory(self.platform_path1) self.gitProcess.start(self.git_path, ["gui"]) # if self.gitProcess.waitForStarted(): # time.sleep(2) # self.find_windows_with_title_prefix("Git Gui") def startGitGui2(self): # 假设 git.exe 位于系统路径中 self.gitProcess.setWorkingDirectory(self.platform_path2) self.gitProcess.start(self.git_path, ["gui"]) def merge_directories(self, src_dir, dst_dir): """ 将src_dir中的文件复制到dst_dir中，如果文件重名则覆盖，保留dst_dir中独有的文件。 :param src_dir: 源目录路径 :param dst_dir: 目标目录路径 """ # 确保目标目录存在 if not os.path.exists(dst_dir): os.makedirs(dst_dir) # 遍历源目录中的所有文件（包括子目录中的文件） for root, dirs, files in os.walk(src_dir): for file in files: src_path = os.path.join(root, file) rel_path = os.path.relpath(src_path, src_dir) # 获取相对于源目录的路径 dst_path = os.path.join(dst_dir, rel_path) # 确保目标路径的目录存在 dst_dir_path = os.path.dirname(dst_path) if not os.path.exists(dst_dir_path): os.makedirs(dst_dir_path) shutil.copy2(src_path, dst_path) # 使用shutil.copy2来保留文件的元数据 def importData(self): #打开文件选择对话框，选择zip文件 filePath, _ = QFileDialog.getOpenFileName(self, "选择文件", "", "Zip Files (*.zip)") if filePath: exportPath = os.path.join(os.path.dirname(__file__),"export") #清空exportPath目录 common.clearDir(exportPath) #创建exportPath目录 common.createDir(exportPath) shutil.unpack_archive(filePath, exportPath, "zip") #遍历exportPath目录子目录名 for root, dirs, files in os.walk(exportPath): for dir in dirs: if dir == "dev_model" or dir == "instruction": #将dev_model目录下的文件强制复制到self.platform_path1目录下 self.merge_directories(os.path.join(root, dir), os.path.join(self.platform_path1, dir)) else: #将device目录下的文件复制到self.platform_path2目录下 self.merge_directories(os.path.join(root, dir), os.path.join(self.platform_path2, dir)) def gitSync(self): if self.topWidget == None: self.topWidget = self.findTopLevelParent(self) if self.topWidget != None: #必须这样写，不能写else if self.topWidget.askDialog("同步数据","是否同步?"): # 删除self.platform_path目录下的所有文件 除了 .git 目录和.git目录下的所有文件 for root, dirs, files in os.walk(self.platform_path1): if ".git" not in root: for name in files: os.remove(os.path.join(root, name)) for root, dirs, files in os.walk(self.platform_path2): if ".git" not in root: for name in files: os.remove(os.path.join(root, name)) # 删除所有表 self.ui.lbState.setText("正在同步数据...") self.ui.pbImport.setEnabled(False) self.ui.pbGitSync.setEnabled(False) self.ui.pbDataSync.setEnabled(False) self.writeJson("dev_model") self.writeJson("instruction") self.writeJson("device") self.writeJson("device_group") self.writeJson("task_group") self.writeJson("task") self.writeJson("task_instruction") self.writeJson("device_model_group") self.ui.lbState.setText("数据同步完成") self.ui.pbImport.setEnabled(True) self.ui.pbGitSync.setEnabled(True) self.ui.pbDataSync.setEnabled(True) def writeJson(self, table_name): manager = None if table_name == "dev_model": manager = devModelManager elif table_name == "instruction": manager = instructionManager elif table_name == "device": manager = deviceManager elif table_name == "device_group": manager = devGroupManager elif table_name == "task_group": manager = taskGroupManager elif table_name == "task": manager = taskManager elif table_name == "task_instruction": manager = taskInstructionManager elif table_name == "device_model_group": manager = dmGroupManager if manager: infos = manager.getInfo('all') for info in infos: QCoreApplication.processEvents() Session.writeJson(table_name, common.ensure_serializable(info)) def onResize(self, event): self.setAllColumnWidth() def dataSync(self): if self.topWidget == None: self.topWidget = self.findTopLevelParent(self) if self.topWidget != None: #必须这样写，不能写else if self.topWidget.askDialog("同步数据","是否同步?"): self.ui.lbState.setText("正在同步数据...") self.ui.pbImport.setEnabled(False) self.ui.pbGitSync.setEnabled(False) self.ui.pbDataSync.setEnabled(False) Session.deleteAllTable() self.dataSync1() self.dataSync2() self.ui.lbState.setText("数据同步完成") self.ui.pbImport.setEnabled(True) self.ui.pbGitSync.setEnabled(True) self.ui.pbDataSync.setEnabled(True) # 批量设置列宽 def setAllColumnWidth(self): try: if self.fileSystemModel1.rowCount() > 0: # 对水平表头的所有行设置为自动调整大小以适应内容 self.ui.treeView1.header().setSectionResizeMode(0,QHeaderView.ResizeMode.Stretch) if self.fileSystemModel2.rowCount() > 0: self.ui.treeView2.header().setSectionResizeMode(0,QHeaderView.ResizeMode.Stretch) except Exception as e: print(e) def dataSync1(self): #遍历目录，查找所有json文件，并读取内容 for root, dirs, files in os.walk(self.platform_path1): #表名为二级目录名 table_name = os.path.basename(root) manager = None if table_name == "dev_model": manager = devModelManager elif table_name == "instruction": manager = instructionManager elif table_name == "device_model_group": manager = dmGroupManager if manager: for file in files: QCoreApplication.processEvents() try: if file.endswith(".json"): file_path = os.path.join(root, file) with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) manager.create_from_git_json(data) except Exception as e: log.error("dataSync1", e) def dataSync2(self): #遍历目录，查找所有json文件，并读取内容 for root, dirs, files in os.walk(self.platform_path2): #表名为二级目录名 table_name = os.path.basename(root) manager = None if table_name == "task": manager = taskManager elif table_name == "device": manager = deviceManager elif table_name == "device_group": manager = devGroupManager elif table_name == "task_group": manager = taskGroupManager elif table_name == "task_instruction": manager = taskInstructionManager if manager: for file in files: QCoreApplication.processEvents() try: if file.endswith(".json"): file_path = os.path.join(root, file) with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) manager.create_from_git_json(data) except Exception as e: log.error("dataSync2", e) def findTopLevelParent(self, widget): parent = widget.parent() while parent and not isinstance(parent, QMainWindow): # 假设顶层窗口是 QMainWindow widget = parent parent = widget.parent() return parent def cleanup(self): self.gitProcess.terminate() self.close_windows_with_title_prefix("Git Gui") def close_windows_with_title_prefix(self, prefix): def callback(hwnd, extra): if win32gui.IsWindowVisible(hwnd) and win32gui.GetWindowText(hwnd).startswith(prefix): # win32gui.SetParent(hwnd, self.ui.widget.winId()) # win32gui.CloseWindow(hwnd)#self.hwnd = hwnd win32gui.PostMessage(hwnd, win32con.WM_CLOSE, 0, 0) # win32gui.MoveWindow(hwnd, 0, 0, self.ui.widget.width(), self.ui.widget.height(), True) # #全屏显示并隐藏标题栏 # win32gui.SetWindowLong(hwnd, win32con.GWL_STYLE, win32con.WS_POPUP | win32con.WS_VISIBLE) # print(f"Found window: {win32gui.GetWindowText(hwnd)} with handle: {hwnd}") win32gui.EnumWindows(callback, None)