"""监控服务""" import time import threading from datetime import datetime from database.access_db import AccessDB from database.sql_server import SQLServerDB from config.settings import ACCESS_DB_PATH, ACCESS_DB_PASSWORD, MONITOR_INTERVAL from tcp.server import TCPServer class MonitoringService: def __init__(self, tcp_server): self.tcp_server = tcp_server self.task_flags = {} self.monitored_tasks = set() self.inserted_tasks = {} self.tasks_lock = threading.Lock() def monitor_access_flag_changes(self): """监控Access数据库中派发任务的Flag状态""" print("开始监控Access数据库中派发任务的Flag状态") while True: try: # 每2秒检查一次 time.sleep(MONITOR_INTERVAL) with self.tasks_lock: # 如果没有需要监控的任务，跳过 if not self.monitored_tasks: continue # 创建需要监控的任务列表副本 tasks_to_monitor = self.monitored_tasks.copy() # 检查SQL Server中任务是否还存在 sql_db = SQLServerDB() try: existing_tasks_in_sql = sql_db.check_existing_tasks(list(tasks_to_monitor)) finally: sql_db.close() # 分离已删除和未删除的任务 erp_ids_str = [str(erp_id) for erp_id in tasks_to_monitor] deleted_from_sql_tasks = set(erp_ids_str) - existing_tasks_in_sql print(f"SQL Server中仍存在的任务: {existing_tasks_in_sql}") print(f"已从SQL Server删除的任务: {deleted_from_sql_tasks}") # 只有已从SQL Server删除的任务才需要在Access中查找 if not deleted_from_sql_tasks: continue # 查询Access数据库中已删除任务的状态 access_db = AccessDB(ACCESS_DB_PATH, ACCESS_DB_PASSWORD) try: current_tasks = access_db.query_task_status(list(deleted_from_sql_tasks)) finally: access_db.close() # 处理任务状态变化 self._process_task_status_changes(deleted_from_sql_tasks, current_tasks) except Exception as e: print(f"监控Access数据库 Flag时发生错误: {e}") import traceback traceback.print_exc() continue def _process_task_status_changes(self, deleted_from_sql_tasks, current_tasks): """处理任务状态变化""" # 检查每个已删除任务的状态变化 for erp_id_str in deleted_from_sql_tasks: erp_id = int(erp_id_str) current_flag = current_tasks.get(erp_id_str, "") previous_flag = self.task_flags.get(erp_id_str, "") # 添加调试信息 print(f"检查任务 {erp_id} - 当前Flag: '{current_flag}', 之前Flag: '{previous_flag}'") # 如果状态发生变化 if current_flag != previous_flag: with self.tasks_lock: artifact_id = self.inserted_tasks.get(erp_id, "Unknown") print( f"派发任务 ErpID {erp_id} (ArtifactID: {artifact_id}) 的Flag值已更新: {previous_flag} -> {current_flag}") self.task_flags[erp_id_str] = current_flag # 根据Flag值末尾的字母执行相应操作并更新自定义数据表状态 self._handle_flag_status_change(erp_id, current_flag, artifact_id) # 检查是否有任务记录已被删除（不在查询结果中但仍在监控列表中） # 这表示任务可能已完成或从系统中移除 missing_tasks = set(deleted_from_sql_tasks) - set(current_tasks.keys()) if missing_tasks: self._handle_missing_tasks(missing_tasks) def _handle_flag_status_change(self, erp_id, current_flag, artifact_id): """处理标志状态变化""" if current_flag.endswith('d'): self._handle_status_d(erp_id, artifact_id) elif current_flag.endswith('w'): self._handle_status_w(erp_id, artifact_id) elif current_flag.endswith('n'): self._handle_status_n(erp_id, artifact_id) elif current_flag.endswith('p'): self._handle_status_p(erp_id, artifact_id) elif current_flag.endswith('x'): self._handle_status_x(erp_id, artifact_id) def _handle_status_d(self, erp_id, artifact_id): """处理状态'd' - 未进行生产""" print(f"派发任务 ErpID {erp_id}: 未进行生产") # 调用同事提供的状态更新函数 try: print(1) update_custom_table_status(erp_id, "未进行生产") except Exception as e: print(f"更新状态时出错: {e}") # 发送数据给TCP客户端（只发送erp_id和状态） try: status_data = { "erp_id": erp_id, "status": "未进行生产" } self.tcp_server.send_data(status_data) except Exception as e: print(f"发送状态数据给TCP客户端时出错: {e}") def _handle_status_w(self, erp_id, artifact_id): """处理状态'w' - 正在生产中""" print(f"派发任务 ErpID {erp_id}: 正在生产中") # 调用同事提供的状态更新函数 try: print(2) update_custom_table_status(erp_id, "正在生产中") except Exception as e: print(f"更新状态时出错: {e}") # 发送数据给TCP客户端（只发送erp_id和状态） try: status_data = { "erp_id": erp_id, "status": "正在生产中" } self.tcp_server.send_data(status_data) except Exception as e: print(f"发送状态数据给TCP客户端时出错: {e}") def _handle_status_n(self, erp_id, artifact_id): """处理状态'n' - 生产完毕""" print(f"派发任务 ErpID {erp_id}: 生产完毕") # 任务完成，可以从监控列表中移除 with self.tasks_lock: self.monitored_tasks.discard(erp_id) print(f"派发任务 ErpID {erp_id} 已完成，停止监控") # 调用同事提供的状态更新函数 try: print(3) update_custom_table_status(erp_id, "生产完毕") except Exception as e: print(f"更新状态时出错: {e}") # 发送数据给TCP客户端（只发送erp_id和状态） try: status_data = { "erp_id": erp_id, "status": "生产完毕" } self.tcp_server.send_data(status_data) except Exception as e: print(f"发送状态数据给TCP客户端时出错: {e}") def _handle_status_p(self, erp_id, artifact_id): """处理状态'p' - 生产中断""" print(f"派发任务 ErpID {erp_id}: 生产中断") # 任务中断，可以从监控列表中移除 with self.tasks_lock: self.monitored_tasks.discard(erp_id) print(f"派发任务 ErpID {erp_id} 已中断，停止监控") # 调用同事提供的状态更新函数 try: print(4) update_custom_table_status(erp_id, "生产中断") except Exception as e: print(f"更新状态时出错: {e}") # 发送数据给TCP客户端（只发送erp_id和状态） try: status_data = { "erp_id": erp_id, "status": "生产中断" } self.tcp_server.send_data(status_data) except Exception as e: print(f"发送状态数据给TCP客户端时出错: {e}") def _handle_status_x(self, erp_id, artifact_id): """处理状态'x' - 数据已接收""" print(f"派发任务 ErpID {erp_id}: 数据已接收") # 调用同事提供的状态更新函数 try: print(5) # update_custom_table_status(erp_id, "数据已接收") except Exception as e: print(f"更新状态时出错: {e}") # 发送数据给TCP客户端（只发送erp_id和状态） try: status_data = { "erp_id": erp_id, "status": "数据已接收" } self.tcp_server.send_data(status_data) except Exception as e: print(f"发送状态数据给TCP客户端时出错: {e}") def _handle_missing_tasks(self, missing_tasks): """处理缺失的任务""" for erp_id_str in missing_tasks: erp_id = int(erp_id_str) with self.tasks_lock: artifact_id = self.inserted_tasks.get(erp_id, "Unknown") self.monitored_tasks.discard(erp_id) self.inserted_tasks.pop(erp_id, None) self.task_flags.pop(erp_id_str, None) print(f"派发任务 ErpID {erp_id} (ArtifactID: {artifact_id}) 记录已从Access数据库中删除或完成")