From 521e3e4e243f4840f532051d3c6ddb70ea49e35f Mon Sep 17 00:00:00 2001 From: lixiaolong <702156524@qq.com> Date: Fri, 24 Jul 2026 13:43:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(inspection):=20=E6=96=B0=E5=A2=9E=E5=B7=A1?= =?UTF-8?q?=E6=A3=80=E4=BB=AA=E8=A1=A8=E8=AF=86=E5=88=AB=E5=92=8C=E4=BA=BA?= =?UTF-8?q?=E5=B7=A5=E5=88=A4=E6=96=AD=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加巡检仪表读数识别动作枚举和处理器 - 实现人工巡检判断任务创建功能 - 扩展巡检报警监听服务支持工作流上下文 - 新增统一巡检结果表和媒体证据表 - 更新巡检告警实体和数据传输对象 - 扩展工作流执行事件传递更多节点信息 - 优化巡检报警入库服务支持订阅模式 --- .../InspectionResultController.java | 126 +++ .../web/controller/test/TeFlowController.java | 9 +- .../cmvr/web/core/config/SwaggerConfig.java | 15 + .../service/InspectionAlertListenService.java | 264 +++++-- .../inspection/domain/InspectionAlarm.java | 12 + .../inspection/domain/InspectionResult.java | 130 +++ .../domain/InspectionResultMedia.java | 43 + .../domain/dto/InspectionResultQuery.java | 32 + .../dto/InspectionResultReviewRequest.java | 32 + .../domain/vo/InspectionAlarmVo.java | 12 + .../domain/vo/InspectionReportSummaryVo.java | 33 + .../domain/vo/InspectionResultVo.java | 27 + .../InspectionFlowExecutionListener.java | 77 +- .../mapper/InspectionResultMapper.java | 9 + .../mapper/InspectionResultMediaMapper.java | 9 + .../service/IInspectionResultService.java | 72 ++ .../InspectionDetectionAlertServiceImpl.java | 32 +- .../impl/InspectionResultServiceImpl.java | 739 ++++++++++++++++++ .../inspection/InspectionAlarmMapper.xml | 29 +- .../java/com/cmvr/test/enums/ActionEnum.java | 4 +- .../runtime/event/FlowExecutionEvent.java | 24 + .../interceptor/FlowAfterInterceptor.java | 6 + .../EdgeInspectionAlertOperateService.java | 8 +- .../EdgeManualInspectionOperateService.java | 127 +++ .../edge/ti/TiTouchOperateService.java | 41 +- ...nspectionMeterRecognizeOperateService.java | 310 ++++++++ .../test/model/vo/FlowActionRequestVO.java | 7 + .../vo/inspection/InspectionAlarmRuleVO.java | 39 + .../InspectionManualReviewConfigVO.java | 33 + .../vo/inspection/InspectionMediaVO.java | 18 + .../InspectionMeterRecognizeConfigVO.java | 27 + .../service/FlowActionExecutorService.java | 19 + .../test/service/ShangdaoFileService.java | 121 +++ sql/inspection_module.sql | 66 ++ sql/inspection_result.sql | 91 +++ 35 files changed, 2507 insertions(+), 136 deletions(-) create mode 100644 cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionResultController.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResult.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResultMedia.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultQuery.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultReviewRequest.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionReportSummaryVo.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionResultVo.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMapper.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMediaMapper.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionResultService.java create mode 100644 cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionResultServiceImpl.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeManualInspectionOperateService.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/InspectionMeterRecognizeOperateService.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionAlarmRuleVO.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionManualReviewConfigVO.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMediaVO.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMeterRecognizeConfigVO.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/service/ShangdaoFileService.java create mode 100644 sql/inspection_result.sql diff --git a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionResultController.java b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionResultController.java new file mode 100644 index 0000000..b4c96bf --- /dev/null +++ b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/inspection/InspectionResultController.java @@ -0,0 +1,126 @@ +package com.cmvr.web.controller.inspection; + +import com.cmvr.common.annotation.Log; +import com.cmvr.common.core.controller.BaseController; +import com.cmvr.common.core.domain.AjaxResult; +import com.cmvr.common.core.page.TableDataInfo; +import com.cmvr.common.enums.BusinessType; +import com.cmvr.common.utils.poi.ExcelUtil; +import com.cmvr.inspection.domain.dto.InspectionResultQuery; +import com.cmvr.inspection.domain.dto.InspectionResultReviewRequest; +import com.cmvr.inspection.domain.vo.InspectionResultVo; +import com.cmvr.inspection.service.IInspectionResultService; +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiImplicitParam; +import io.swagger.annotations.ApiImplicitParams; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import lombok.RequiredArgsConstructor; +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.validation.annotation.Validated; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.PutMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import javax.servlet.http.HttpServletResponse; +import java.util.List; + +/** + * 巡检结果、人工复核和任务报表接口。 + */ +@RestController +@RequestMapping("/inspection/result") +@Api(tags = "智能巡检--巡检结果", description = "巡检结果查询、人工复核、三级告警统计和报表导出") +@ApiResponses({ + @ApiResponse(code = 200, message = "请求处理成功"), + @ApiResponse(code = 400, message = "请求参数或状态不合法"), + @ApiResponse(code = 401, message = "未登录或登录已失效"), + @ApiResponse(code = 403, message = "无接口权限"), + @ApiResponse(code = 500, message = "系统内部异常") +}) +@RequiredArgsConstructor +public class InspectionResultController extends BaseController +{ + private final IInspectionResultService inspectionResultService; + + @ApiOperation(value = "查询巡检结果列表", + notes = "分页查询PPE、仪表读数和人工判断结果；查询条件均为可选，结果按检测时间倒序。" + + "返回rows元素为InspectionResultVo，包含taskName任务名称和evidenceType主要证据类型；" + + "完整mediaList仅在详情接口返回") + @ApiImplicitParams({ + @ApiImplicitParam(name = "pageNum", value = "页码，从1开始", dataType = "int", paramType = "query", example = "1"), + @ApiImplicitParam(name = "pageSize", value = "每页数量", dataType = "int", paramType = "query", example = "10"), + @ApiImplicitParam(name = "taskInstanceId", value = "巡检任务实例数据库ID", dataType = "string", paramType = "query"), + @ApiImplicitParam(name = "taskId", value = "巡检任务ID", dataType = "string", paramType = "query"), + @ApiImplicitParam(name = "itemId", value = "检测项ID", dataType = "string", paramType = "query"), + @ApiImplicitParam(name = "resultType", value = "结果类型：PPE、METER、MANUAL", dataType = "string", paramType = "query"), + @ApiImplicitParam(name = "resultStatus", value = "状态：PENDING、NORMAL、ABNORMAL、RECOGNIZE_FAILED", dataType = "string", paramType = "query"), + @ApiImplicitParam(name = "alarmLevel", value = "告警级别：1提示、2警告、3严重", dataType = "int", paramType = "query"), + @ApiImplicitParam(name = "resultName", value = "检查名称，支持模糊查询", dataType = "string", paramType = "query") + }) + @PreAuthorize("@ss.hasPermi('inspection:result:list')") + @GetMapping("/list") + public TableDataInfo list(@ApiParam("巡检结果查询条件") InspectionResultQuery query) + { + startPage(); + return getDataTable(inspectionResultService.selectResultList(query)); + } + + @ApiOperation(value = "查询巡检结果详情", + notes = "返回结构化巡检结果及按顺序排列的全部图片、视频证据", + response = InspectionResultVo.class) + @PreAuthorize("@ss.hasPermi('inspection:result:query')") + @GetMapping("/{id}") + public AjaxResult getInfo(@ApiParam(value = "巡检结果ID", required = true) + @PathVariable String id) + { + return success(inspectionResultService.selectResultById(id)); + } + + @ApiOperation(value = "提交人工复核结果", + notes = "仅PENDING或RECOGNIZE_FAILED状态可复核。ABNORMAL必须传alarmLevel；reviewVersion用于并发控制") + @PreAuthorize("@ss.hasPermi('inspection:result:review')") + @Log(title = "巡检结果人工复核", businessType = BusinessType.UPDATE) + @PutMapping("/{id}/review") + public AjaxResult review(@ApiParam(value = "巡检结果ID", required = true) + @PathVariable String id, + @ApiParam(value = "人工复核内容", required = true) + @Validated @RequestBody InspectionResultReviewRequest request) + { + return success(inspectionResultService.review(id, request)); + } + + @ApiOperation(value = "查询巡检任务报表汇总", + notes = "返回summary统计和results明细。reviewCompleted=false表示仍有待复核或识别失败结果") + @PreAuthorize("@ss.hasPermi('inspection:result:report')") + @GetMapping("/report/{taskInstanceId}") + public AjaxResult report(@ApiParam(value = "巡检任务实例数据库ID", required = true) + @PathVariable String taskInstanceId) + { + AjaxResult result = AjaxResult.success(); + result.put("summary", inspectionResultService.buildReportSummary(taskInstanceId)); + InspectionResultQuery query = new InspectionResultQuery(); + query.setTaskInstanceId(taskInstanceId); + result.put("results", inspectionResultService.selectResultList(query)); + return result; + } + + @ApiOperation(value = "导出巡检结果报表", + notes = "按查询条件导出Excel；不传条件时导出全部巡检结果") + @PreAuthorize("@ss.hasPermi('inspection:result:export')") + @Log(title = "巡检结果", businessType = BusinessType.EXPORT) + @PostMapping("/export") + public void export(HttpServletResponse response, + @ApiParam("导出筛选条件") InspectionResultQuery query) + { + List list = inspectionResultService.selectResultList(query); + ExcelUtil util = new ExcelUtil<>(InspectionResultVo.class); + util.exportExcel(response, list, "巡检结果数据"); + } +} diff --git a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/test/TeFlowController.java b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/test/TeFlowController.java index 3966951..c43afe5 100644 --- a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/test/TeFlowController.java +++ b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/test/TeFlowController.java @@ -156,8 +156,13 @@ public class TeFlowController extends BaseController { } @PostMapping("/action") - @ApiOperation("单个节点执行") - public AjaxResult actionExecute(@RequestBody FlowActionRequestVO flowActionRequestVO) { + @ApiOperation(value = "单个节点执行", + notes = "巡检仪表节点action=INSPECTION_METER_RECOGNIZE，payload使用InspectionMeterRecognizeConfigVO；" + + "人工判断节点action=INSPECTION_MANUAL_REVIEW_CREATE，payload使用InspectionManualReviewConfigVO。" + + "该接口用于试调节点，不会生成正式巡检任务结果") + public AjaxResult actionExecute( + @io.swagger.annotations.ApiParam(value = "单节点动作和参数", required = true) + @RequestBody FlowActionRequestVO flowActionRequestVO) { return AjaxResult.ok(flowActionExecutorService.actionExecute(flowActionRequestVO)); } diff --git a/cmvr-iot-admin/src/main/java/com/cmvr/web/core/config/SwaggerConfig.java b/cmvr-iot-admin/src/main/java/com/cmvr/web/core/config/SwaggerConfig.java index 529a5ec..2ffc7f0 100644 --- a/cmvr-iot-admin/src/main/java/com/cmvr/web/core/config/SwaggerConfig.java +++ b/cmvr-iot-admin/src/main/java/com/cmvr/web/core/config/SwaggerConfig.java @@ -7,6 +7,11 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import com.cmvr.common.config.CmvrIotConfig; +import com.cmvr.test.model.vo.inspection.InspectionAlarmRuleVO; +import com.cmvr.test.model.vo.inspection.InspectionManualReviewConfigVO; +import com.cmvr.test.model.vo.inspection.InspectionMediaVO; +import com.cmvr.test.model.vo.inspection.InspectionMeterRecognizeConfigVO; +import com.fasterxml.classmate.TypeResolver; import io.swagger.annotations.ApiOperation; import io.swagger.models.auth.In; import springfox.documentation.builders.ApiInfoBuilder; @@ -34,6 +39,10 @@ public class SwaggerConfig @Autowired private CmvrIotConfig cmvrIotConfig; + /** 用于将动态工作流payload对应的参数模型显式加入Swagger文档。 */ + @Autowired + private TypeResolver typeResolver; + /** 是否开启swagger */ @Value("${swagger.enabled}") private boolean enabled; @@ -62,6 +71,12 @@ public class SwaggerConfig // 扫描所有 .apis(RequestHandlerSelectors.any()) .paths(PathSelectors.any()) .build() + // 工作流单节点接口使用JSONObject作为payload，需显式注册巡检节点参数模型。 + .additionalModels( + typeResolver.resolve(InspectionMeterRecognizeConfigVO.class), + typeResolver.resolve(InspectionAlarmRuleVO.class), + typeResolver.resolve(InspectionManualReviewConfigVO.class), + typeResolver.resolve(InspectionMediaVO.class)) /* 设置安全模式，swagger可以设置访问token */ .securitySchemes(securitySchemes()) .securityContexts(securityContexts()) diff --git a/cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java b/cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java index 1643068..df6f72a 100644 --- a/cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java +++ b/cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java @@ -13,7 +13,6 @@ import java.util.Collection; import java.util.Collections; import java.util.LinkedHashSet; import java.util.List; -import java.util.Locale; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -21,16 +20,14 @@ import java.util.concurrent.ConcurrentMap; /** * 巡检报警监听状态服务。 * - *

工作流开始监听时，根据终端ID解析终端IP，并将 IP + 报警类型保存到内存。 - * cmvr_edge_ai 推送报警后，巡检报警入库服务通过该服务判断是否需要保存。 - * 当前状态只在本 JVM 内有效，应用重启后会清空。

+ *

监听条件仍然按边缘端IP和报警类型匹配，同时保存工作流执行上下文， + * 使异步上报的PPE事件能够归属到具体巡检任务、检测项和节点。

*/ @Slf4j @Service @RequiredArgsConstructor public class InspectionAlertListenService { - /** 当前支持的 PPE 违规类型。 */ private static final Set SUPPORTED_EVENT_TYPES; static @@ -43,47 +40,59 @@ public class InspectionAlertListenService private final IDeDeviceTerminalConfigService terminalConfigService; - /** - * key 为边缘端 gRPC IP，value 为当前需要保存的报警类型集合。 - */ - private final ConcurrentMap> listenMap = new ConcurrentHashMap<>(); + /** key=grpcIp，value=当前IP下的工作流监听订阅。 */ + private final ConcurrentMap> subscriptions = + new ConcurrentHashMap<>(); - /** - * 开始监听指定终端的指定报警类型。 - * - * @param terminalId 终端ID - * @param eventTypes 报警类型，支持 No-Glove、No-Helmet，可多选 - * @return 当前终端IP下正在监听的报警类型 - */ + /** 兼容原有调用方式，创建一个不绑定工作流的监听。 */ public ListenState startListen(String terminalId, Collection eventTypes) { - String grpcIp = resolveTerminalIp(terminalId); - Set normalizedTypes = normalizeEventTypes(eventTypes, true); - - Set target = listenMap.computeIfAbsent(grpcIp, key -> ConcurrentHashMap.newKeySet()); - target.addAll(normalizedTypes); - - log.info("开始监听巡检报警，terminalId={}，grpcIp={}，eventTypes={}，currentTypes={}", - terminalId, grpcIp, normalizedTypes, target); - return new ListenState(terminalId, grpcIp, snapshot(target)); + return startListen(terminalId, eventTypes, null, null, null, null, null); } - /** - * 结束监听指定终端的指定报警类型。 - * - *

移除不存在的 IP + 类型不抛异常，只记录警告日志，避免工作流重复清理时失败。

- * - * @param terminalId 终端ID - * @param eventTypes 报警类型，支持 No-Glove、No-Helmet，可多选 - * @return 当前终端IP下剩余正在监听的报警类型 - */ - public ListenState stopListen(String terminalId, Collection eventTypes) + public ListenState startListen(String terminalId, Collection eventTypes, + String flowInstanceId, String taskId, String itemId, + String nodeId, String listenerKey) { String grpcIp = resolveTerminalIp(terminalId); Set normalizedTypes = normalizeEventTypes(eventTypes, true); - Set current = listenMap.get(grpcIp); + ConcurrentMap target = + subscriptions.computeIfAbsent(grpcIp, key -> new ConcurrentHashMap<>()); - if (current == null) + for (String eventType : normalizedTypes) + { + ListenSubscription subscription = new ListenSubscription(); + subscription.setSubscriptionId(buildSubscriptionId(flowInstanceId, itemId, listenerKey, eventType)); + subscription.setTerminalId(terminalId); + subscription.setGrpcIp(grpcIp); + subscription.setEventType(eventType); + subscription.setFlowInstanceId(flowInstanceId); + subscription.setTaskId(taskId); + subscription.setItemId(itemId); + subscription.setNodeId(nodeId); + subscription.setListenerKey(StringUtils.trimToNull(listenerKey)); + subscription.setStartedAt(System.currentTimeMillis()); + target.put(subscription.getSubscriptionId(), subscription); + } + + log.info("开始监听巡检报警，terminalId={}，grpcIp={}，flowInstanceId={}，itemId={}，eventTypes={}", + terminalId, grpcIp, flowInstanceId, itemId, normalizedTypes); + return new ListenState(terminalId, grpcIp, currentTypes(target)); + } + + /** 兼容原有调用方式，移除该IP下指定类型的全部监听。 */ + public ListenState stopListen(String terminalId, Collection eventTypes) + { + return stopListen(terminalId, eventTypes, null, null, null); + } + + public ListenState stopListen(String terminalId, Collection eventTypes, + String flowInstanceId, String itemId, String listenerKey) + { + String grpcIp = resolveTerminalIp(terminalId); + Set normalizedTypes = normalizeEventTypes(eventTypes, true); + ConcurrentMap current = subscriptions.get(grpcIp); + if (current == null || current.isEmpty()) { log.warn("结束监听巡检报警时未找到IP监听状态，terminalId={}，grpcIp={}，eventTypes={}", terminalId, grpcIp, normalizedTypes); @@ -92,67 +101,124 @@ public class InspectionAlertListenService for (String eventType : normalizedTypes) { - if (!current.remove(eventType)) + int removed = removeMatching(current, eventType, flowInstanceId, itemId, listenerKey); + if (removed == 0) { - log.warn("结束监听巡检报警时事件类型不存在，terminalId={}，grpcIp={}，eventType={}", - terminalId, grpcIp, eventType); + log.warn("结束监听巡检报警时未找到匹配订阅，terminalId={}，grpcIp={}，flowInstanceId={}，itemId={}，eventType={}", + terminalId, grpcIp, flowInstanceId, itemId, eventType); } } - if (current.isEmpty()) { - listenMap.remove(grpcIp, current); + subscriptions.remove(grpcIp, current); } - log.info("结束监听巡检报警，terminalId={}，grpcIp={}，eventTypes={}，remainingTypes={}", - terminalId, grpcIp, normalizedTypes, current); - return new ListenState(terminalId, grpcIp, snapshot(current)); + List remainingTypes = currentTypes(current); + log.info("结束监听巡检报警，terminalId={}，grpcIp={}，flowInstanceId={}，itemId={}，remainingTypes={}", + terminalId, grpcIp, flowInstanceId, itemId, remainingTypes); + return new ListenState(terminalId, grpcIp, remainingTypes); } - /** - * 判断报警是否命中当前监听条件。 - * - * @param grpcIp 边缘端 gRPC IP - * @param labels 报警标签列表，来自 payload.labels - * @return true 表示允许入库，false 表示忽略 - */ - public boolean shouldStore(String grpcIp, Collection labels) + /** 返回命中IP和任一标签的全部活动订阅。 */ + public List matchSubscriptions(String grpcIp, Collection labels) { String normalizedIp = StringUtils.trimToNull(grpcIp); if (normalizedIp == null) { log.warn("忽略巡检报警：grpc_ip为空，labels={}", labels); - return false; + return Collections.emptyList(); } - - Set listeningTypes = listenMap.get(normalizedIp); - if (listeningTypes == null || listeningTypes.isEmpty()) + ConcurrentMap current = subscriptions.get(normalizedIp); + if (current == null || current.isEmpty()) { log.info("忽略巡检报警：未开启该IP的报警监听，grpcIp={}，labels={}", normalizedIp, labels); - return false; + return Collections.emptyList(); } Set alertTypes = normalizeEventTypes(labels, false); - for (String alertType : alertTypes) + List matched = new ArrayList<>(); + for (ListenSubscription subscription : current.values()) { - if (listeningTypes.contains(alertType)) + if (alertTypes.contains(subscription.getEventType())) { - return true; + matched.add(subscription.copy()); } } - - log.info("忽略巡检报警：事件类型未命中监听条件，grpcIp={}，labels={}，listeningTypes={}", - normalizedIp, labels, listeningTypes); - return false; + if (matched.isEmpty()) + { + log.info("忽略巡检报警：事件类型未命中监听条件，grpcIp={}，labels={}，listeningTypes={}", + normalizedIp, labels, currentTypes(current)); + } + return matched; + } + + public boolean shouldStore(String grpcIp, Collection labels) + { + return !matchSubscriptions(grpcIp, labels).isEmpty(); + } + + /** 任务完成、失败或终止时清理该流程的全部监听。 */ + public void clearByFlowInstanceId(String flowInstanceId) + { + String normalized = StringUtils.trimToNull(flowInstanceId); + if (normalized == null) + { + return; + } + int removed = 0; + for (ConcurrentMap.Entry> ipEntry : subscriptions.entrySet()) + { + ConcurrentMap current = ipEntry.getValue(); + for (ConcurrentMap.Entry entry : current.entrySet()) + { + if (normalized.equals(entry.getValue().getFlowInstanceId()) && current.remove(entry.getKey(), entry.getValue())) + { + removed++; + } + } + if (current.isEmpty()) + { + subscriptions.remove(ipEntry.getKey(), current); + } + } + if (removed > 0) + { + log.info("巡检任务结束，清理报警监听，flowInstanceId={}，removed={}", normalized, removed); + } } - /** - * 查询指定终端当前监听状态，便于调试或前端展示。 - */ public ListenState getListenState(String terminalId) { String grpcIp = resolveTerminalIp(terminalId); - return new ListenState(terminalId, grpcIp, snapshot(listenMap.get(grpcIp))); + return new ListenState(terminalId, grpcIp, currentTypes(subscriptions.get(grpcIp))); + } + + private int removeMatching(ConcurrentMap current, String eventType, + String flowInstanceId, String itemId, String listenerKey) + { + int removed = 0; + for (ConcurrentMap.Entry entry : current.entrySet()) + { + ListenSubscription subscription = entry.getValue(); + boolean legacyRemove = StringUtils.isBlank(flowInstanceId); + boolean contextMatches = StringUtils.equals(flowInstanceId, subscription.getFlowInstanceId()) + && (StringUtils.isBlank(itemId) || StringUtils.equals(itemId, subscription.getItemId())) + && (StringUtils.isBlank(listenerKey) || StringUtils.equals(listenerKey, subscription.getListenerKey())); + if (eventType.equals(subscription.getEventType()) && (legacyRemove || contextMatches) + && current.remove(entry.getKey(), subscription)) + { + removed++; + } + } + return removed; + } + + private String buildSubscriptionId(String flowInstanceId, String itemId, String listenerKey, String eventType) + { + String flow = StringUtils.defaultIfBlank(flowInstanceId, "legacy"); + String item = StringUtils.defaultIfBlank(itemId, "all"); + String key = StringUtils.defaultIfBlank(listenerKey, "default"); + return flow + "|" + item + "|" + key + "|" + eventType; } private String resolveTerminalIp(String terminalId) @@ -162,14 +228,12 @@ public class InspectionAlertListenService { throw new GlobalException("terminalId不能为空"); } - DeDeviceTerminalConfig terminalConfig = terminalConfigService.selectDeDeviceTerminalConfigById(normalizedTerminalId); if (terminalConfig == null) { throw new GlobalException("未找到终端配置，terminalId=" + normalizedTerminalId); } - String grpcIp = StringUtils.trimToNull(terminalConfig.getHost()); if (grpcIp == null) { @@ -192,7 +256,6 @@ public class InspectionAlertListenService } } } - if (failOnEmpty && result.isEmpty()) { throw new GlobalException("eventTypes不能为空，当前支持No-Glove、No-Helmet"); @@ -207,7 +270,6 @@ public class InspectionAlertListenService { return null; } - for (String supported : SUPPORTED_EVENT_TYPES) { if (supported.equalsIgnoreCase(value)) @@ -215,23 +277,24 @@ public class InspectionAlertListenService return supported; } } - log.warn("忽略不支持的巡检报警类型，eventType={}，supported={}", value, SUPPORTED_EVENT_TYPES); return null; } - private List snapshot(Set eventTypes) + private List currentTypes(ConcurrentMap current) { - if (eventTypes == null || eventTypes.isEmpty()) + if (current == null || current.isEmpty()) { return Collections.emptyList(); } - return new ArrayList<>(eventTypes); + Set values = new LinkedHashSet<>(); + for (ListenSubscription subscription : current.values()) + { + values.add(subscription.getEventType()); + } + return new ArrayList<>(values); } - /** - * 当前监听状态。 - */ @Data public static class ListenState { @@ -239,4 +302,45 @@ public class InspectionAlertListenService private final String grpcIp; private final List eventTypes; } + + @Data + public static class ListenSubscription + { + /** flowInstanceId、itemId、listenerKey、eventType组成的订阅唯一标识。 */ + private String subscriptionId; + /** 工作流节点配置的终端ID。 */ + private String terminalId; + /** 由终端配置解析出的gRPC IP，也是PPE事件匹配IP。 */ + private String grpcIp; + /** 当前订阅监听的PPE事件类型。 */ + private String eventType; + /** 工作流运行实例ID，用于关联巡检任务和结束时清理。 */ + private String flowInstanceId; + /** 工作流编排任务ID。 */ + private String taskId; + /** 当前巡检检测项ID。 */ + private String itemId; + /** 启动监听的工作流节点ID。 */ + private String nodeId; + /** 同一检测项存在多个监听器时使用的可选业务标识。 */ + private String listenerKey; + /** 开始监听时间戳，单位毫秒。 */ + private long startedAt; + + private ListenSubscription copy() + { + ListenSubscription copy = new ListenSubscription(); + copy.subscriptionId = subscriptionId; + copy.terminalId = terminalId; + copy.grpcIp = grpcIp; + copy.eventType = eventType; + copy.flowInstanceId = flowInstanceId; + copy.taskId = taskId; + copy.itemId = itemId; + copy.nodeId = nodeId; + copy.listenerKey = listenerKey; + copy.startedAt = startedAt; + return copy; + } + } } diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionAlarm.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionAlarm.java index 6674bcc..d434634 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionAlarm.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionAlarm.java @@ -40,6 +40,18 @@ public class InspectionAlarm extends BaseEntity @ApiModelProperty("任务执行实例ID") private String taskInstanceId; + @ApiModelProperty("关联巡检结果ID") + private String resultId; + + @ApiModelProperty("告警来源：PPE、METER、MANUAL") + private String alarmSource; + + @ApiModelProperty("检测项ID") + private String itemId; + + @ApiModelProperty("工作流节点ID") + private String nodeId; + @Excel(name = "任务ID") @ApiModelProperty("任务ID") private String taskId; diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResult.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResult.java new file mode 100644 index 0000000..cd9a3ea --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResult.java @@ -0,0 +1,130 @@ +package com.cmvr.inspection.domain; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import com.cmvr.common.annotation.Excel; +import com.cmvr.common.core.domain.BaseEntity; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; +import lombok.EqualsAndHashCode; + +import java.math.BigDecimal; +import java.util.Date; + +/** + * 统一巡检结果，承载PPE、仪表读数和人工判断结果。 + */ +@Data +@EqualsAndHashCode(callSuper = true) +@TableName("inspection_result") +@ApiModel(value = "InspectionResult", description = "PPE识别、仪表读数识别和人工复核共用的巡检结果") +public class InspectionResult extends BaseEntity +{ + private static final long serialVersionUID = 1L; + + @TableId(value = "id", type = IdType.INPUT) + @ApiModelProperty(value = "巡检结果主键", example = "8f3e1f3a2db84de2a84eebdf8951b86a") + private String id; + + @Excel(name = "结果编码") + @ApiModelProperty(value = "巡检结果业务编码", example = "IR20260723143000123ABCDE") + private String resultCode; + + @ApiModelProperty(value = "结果幂等键，平台内部使用", hidden = true) + private String dedupeKey; + + @ApiModelProperty(value = "巡检任务实例数据库ID", example = "task-instance-id") + private String taskInstanceId; + + @ApiModelProperty(value = "工作流运行实例ID", example = "flow-instance-id") + private String flowInstanceId; + + @ApiModelProperty(value = "巡检任务ID", example = "inspection-task-id") + private String taskId; + + @ApiModelProperty(value = "检测项ID", example = "detection-item-id") + private String itemId; + + @ApiModelProperty(value = "工作流节点ID", example = "meter-node-1") + private String nodeId; + + @ApiModelProperty(value = "工作流节点名称", example = "1号压力表读数识别") + private String nodeName; + + @ApiModelProperty(value = "循环节点迭代路径JSON；非循环节点通常为[]", example = "[1,2]") + private String iterationPath; + + @Excel(name = "结果类型") + @ApiModelProperty(value = "结果类型", allowableValues = "PPE,METER,MANUAL", example = "METER") + private String resultType; + + @Excel(name = "结果状态") + @ApiModelProperty(value = "结果状态", allowableValues = "PENDING,NORMAL,ABNORMAL,RECOGNIZE_FAILED", example = "ABNORMAL") + private String resultStatus; + + @Excel(name = "检查名称") + @ApiModelProperty(value = "检查项显示名称", example = "1号压力表") + private String resultName; + + @Excel(name = "数值") + @ApiModelProperty(value = "结构化仪表读数；PPE和人工结果通常为空", example = "1.72") + private BigDecimal valueNumber; + + @Excel(name = "结果内容") + @ApiModelProperty(value = "文本结果或原始识别值", example = "1.72") + private String valueText; + + @Excel(name = "单位") + @ApiModelProperty(value = "仪表读数单位", example = "MPa") + private String unit; + + @Excel(name = "告警级别") + @ApiModelProperty(value = "最终命中的最高告警级别：1提示、2警告、3严重", allowableValues = "1,2,3", example = "2") + private Integer alarmLevel; + + @ApiModelProperty(value = "告警描述", example = "压力过高") + private String alarmMessage; + + @ApiModelProperty(value = "模型置信度，范围0到1", example = "0.96") + private BigDecimal confidence; + + @ApiModelProperty(value = "PPE边缘事件event_id", example = "event-abc") + private String sourceEventId; + + @ApiModelProperty(value = "主要证据的永久MinIO地址") + private String evidenceUrl; + + @ApiModelProperty(value = "主要证据类型，与evidenceUrl同时写入", + allowableValues = "IMAGE,VIDEO", example = "IMAGE") + private String evidenceType; + + @ApiModelProperty(value = "模型或边缘端原始返回内容") + private String rawResult; + + @ApiModelProperty(value = "最终命中的告警规则JSON快照") + private String matchedRuleJson; + + @ApiModelProperty(value = "本次执行使用的全部告警规则JSON快照") + private String ruleSnapshotJson; + + @ApiModelProperty(value = "人工复核账号", example = "admin") + private String reviewer; + + @ApiModelProperty(value = "人工复核时间") + private Date reviewTime; + + @ApiModelProperty(value = "人工复核说明") + private String reviewRemark; + + @ApiModelProperty(value = "人工复核乐观锁版本，首次复核前为0", example = "0") + private Integer reviewVersion; + + @ApiModelProperty(value = "异常结果关联的巡检告警ID") + private String alarmId; + + @Excel(name = "检测时间", dateFormat = "yyyy-MM-dd HH:mm:ss") + @ApiModelProperty(value = "结果发生时间") + private Date occurredTime; +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResultMedia.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResultMedia.java new file mode 100644 index 0000000..93d85ae --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/InspectionResultMedia.java @@ -0,0 +1,43 @@ +package com.cmvr.inspection.domain; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.io.Serializable; +import java.util.Date; + +/** + * 巡检结果关联的图片或视频证据。 + * + *

媒体内容保存在MinIO，本表只保存永久访问地址，不保存文件字节。

+ */ +@Data +@TableName("inspection_result_media") +@ApiModel(value = "InspectionResultMedia", description = "巡检结果关联的图片或视频证据") +public class InspectionResultMedia implements Serializable +{ + private static final long serialVersionUID = 1L; + + @TableId(value = "id", type = IdType.INPUT) + @ApiModelProperty("媒体记录主键") + private String id; + + @ApiModelProperty("关联巡检结果ID") + private String resultId; + + @ApiModelProperty(value = "媒体类型", allowableValues = "IMAGE,VIDEO", example = "IMAGE") + private String mediaType; + + @ApiModelProperty("MinIO永久媒体地址") + private String mediaUrl; + + @ApiModelProperty(value = "同一巡检结果内的展示顺序", example = "0") + private Integer sortOrder; + + @ApiModelProperty("记录创建时间") + private Date createTime; +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultQuery.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultQuery.java new file mode 100644 index 0000000..f8d5ce9 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultQuery.java @@ -0,0 +1,32 @@ +package com.cmvr.inspection.domain.dto; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +/** 巡检结果列表和报表导出的查询条件。 */ +@Data +@ApiModel(value = "InspectionResultQuery", description = "巡检结果查询条件，所有字段均为可选条件") +public class InspectionResultQuery +{ + @ApiModelProperty("巡检任务实例数据库ID") + private String taskInstanceId; + + @ApiModelProperty("巡检任务ID") + private String taskId; + + @ApiModelProperty("检测项ID") + private String itemId; + + @ApiModelProperty(value = "结果类型", allowableValues = "PPE,METER,MANUAL") + private String resultType; + + @ApiModelProperty(value = "结果状态", allowableValues = "PENDING,NORMAL,ABNORMAL,RECOGNIZE_FAILED") + private String resultStatus; + + @ApiModelProperty(value = "告警级别", allowableValues = "1,2,3") + private Integer alarmLevel; + + @ApiModelProperty("检查名称，支持模糊查询") + private String resultName; +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultReviewRequest.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultReviewRequest.java new file mode 100644 index 0000000..ef8b9f9 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/dto/InspectionResultReviewRequest.java @@ -0,0 +1,32 @@ +package com.cmvr.inspection.domain.dto; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import javax.validation.constraints.NotBlank; + +/** + * 人工复核提交参数。 + * + *

只能复核PENDING或RECOGNIZE_FAILED状态的结果。提交ABNORMAL时必须指定alarmLevel。

+ */ +@Data +@ApiModel("巡检结果人工复核参数") +public class InspectionResultReviewRequest +{ + @NotBlank(message = "resultStatus不能为空") + @ApiModelProperty(value = "复核结果", required = true, + allowableValues = "NORMAL,ABNORMAL", example = "ABNORMAL") + private String resultStatus; + + @ApiModelProperty(value = "异常时必填：1提示、2警告、3严重", + allowableValues = "1,2,3", example = "2") + private Integer alarmLevel; + + @ApiModelProperty(value = "人工复核说明", example = "设备右侧存在漏油") + private String reviewRemark; + + @ApiModelProperty(value = "当前reviewVersion，用于防止多人重复复核；首次通常为0", example = "0") + private Integer reviewVersion; +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionAlarmVo.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionAlarmVo.java index d7debd3..5d4edd8 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionAlarmVo.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionAlarmVo.java @@ -30,6 +30,18 @@ public class InspectionAlarmVo extends BaseEntity @ApiModelProperty("id") private String id; + @ApiModelProperty("关联巡检结果ID") + private String resultId; + + @ApiModelProperty("告警来源：PPE、METER、MANUAL") + private String alarmSource; + + @ApiModelProperty("检测项ID") + private String itemId; + + @ApiModelProperty("工作流节点ID") + private String nodeId; + /** 告警编码 */ @Excel(name = "告警编码") @ApiModelProperty("告警编码") diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionReportSummaryVo.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionReportSummaryVo.java new file mode 100644 index 0000000..a4fc392 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionReportSummaryVo.java @@ -0,0 +1,33 @@ +package com.cmvr.inspection.domain.vo; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** 单次巡检任务的结果汇总。 */ +@Data +@ApiModel(value = "InspectionReportSummaryVo", description = "单次巡检任务的结果统计汇总") +public class InspectionReportSummaryVo +{ + @ApiModelProperty("巡检任务实例数据库ID") + private String taskInstanceId; + @ApiModelProperty("结果总数") + private long totalCount; + @ApiModelProperty("正常结果数") + private long normalCount; + @ApiModelProperty("异常结果数") + private long abnormalCount; + @ApiModelProperty("待人工复核数") + private long pendingCount; + @ApiModelProperty("识别失败数") + private long recognizeFailedCount; + @ApiModelProperty("按告警级别统计，key为1、2、3") + private Map alarmLevelCounts = new LinkedHashMap<>(); + @ApiModelProperty("按结果类型统计，key为PPE、METER、MANUAL") + private Map resultTypeCounts = new LinkedHashMap<>(); + @ApiModelProperty("是否不存在待复核和识别失败结果") + private boolean reviewCompleted; +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionResultVo.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionResultVo.java new file mode 100644 index 0000000..354d282 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/domain/vo/InspectionResultVo.java @@ -0,0 +1,27 @@ +package com.cmvr.inspection.domain.vo; + +import com.cmvr.common.annotation.Excel; +import com.cmvr.inspection.domain.InspectionResult; +import com.cmvr.inspection.domain.InspectionResultMedia; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; +import lombok.EqualsAndHashCode; + +import java.util.List; + +/** 巡检结果详情，包含全部媒体证据。 */ +@Data +@EqualsAndHashCode(callSuper = true) +@ApiModel(value = "InspectionResultVo", description = "巡检结果详情及其全部图片、视频证据") +public class InspectionResultVo extends InspectionResult +{ + private static final long serialVersionUID = 1L; + + @Excel(name = "任务名称") + @ApiModelProperty(value = "巡检任务名称", example = "锅炉房日常巡检") + private String taskName; + + @ApiModelProperty("巡检结果关联的全部媒体证据，按sortOrder升序") + private List mediaList; +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java index 1850e74..3cf2d7b 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/listener/InspectionFlowExecutionListener.java @@ -4,12 +4,14 @@ package com.cmvr.inspection.listener; import cn.hutool.core.util.StrUtil; import com.cmvr.common.utils.DateUtils; import com.cmvr.framework.websocket.service.MessagePushService; +import com.cmvr.device.service.InspectionAlertListenService; import com.cmvr.inspection.domain.InspectionTaskInstance; import com.cmvr.inspection.domain.InspectionTaskLog; import com.cmvr.inspection.enums.InspectionLogTypeEnum; import com.cmvr.inspection.enums.TaskStatusEnum; import com.cmvr.inspection.service.IInspectionTaskInstanceService; import com.cmvr.inspection.service.IInspectionTaskLogService; +import com.cmvr.inspection.service.IInspectionResultService; import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; import com.cmvr.test.flow.runtime.event.FlowExecutionListener; import com.cmvr.test.flow.runtime.event.FlowExecutionModuleCodes; @@ -24,18 +26,24 @@ public class InspectionFlowExecutionListener implements FlowExecutionListener { private final IInspectionTaskInstanceService inspectionTaskInstanceService; private final IInspectionTaskLogService inspectionTaskLogService; private final MessagePushService messagePushService; + private final IInspectionResultService inspectionResultService; + private final InspectionAlertListenService inspectionAlertListenService; - /** key:itemId, value:已完成节点集合 */ - private final Map> itemNodeExecutMap = new HashMap<>(); + /** key:flowInstId|itemId, value:已完成节点集合 */ + private final Map> itemNodeExecutMap = new java.util.concurrent.ConcurrentHashMap<>(); /** key:taskInsId(流程实例id=taskInsId), value:已执行item集合 */ - private final Map> taskItemExecutMap = new HashMap<>(); + private final Map> taskItemExecutMap = new java.util.concurrent.ConcurrentHashMap<>(); public InspectionFlowExecutionListener(IInspectionTaskInstanceService inspectionTaskInstanceService, IInspectionTaskLogService inspectionTaskLogService, - MessagePushService messagePushService) { + MessagePushService messagePushService, + IInspectionResultService inspectionResultService, + InspectionAlertListenService inspectionAlertListenService) { this.inspectionTaskInstanceService = inspectionTaskInstanceService; this.inspectionTaskLogService = inspectionTaskLogService; this.messagePushService = messagePushService; + this.inspectionResultService = inspectionResultService; + this.inspectionAlertListenService = inspectionAlertListenService; } @Override @@ -52,19 +60,27 @@ public class InspectionFlowExecutionListener implements FlowExecutionListener { return; } + // 无论数据库是否可用，任务结束事件都必须先清理内存监听和进度状态。 + if (FlowExecutionEvent.EventType.TASK_COMPLETED == event.getEventType() + || FlowExecutionEvent.EventType.TASK_FAILED == event.getEventType() + || FlowExecutionEvent.EventType.TASK_STOPPED == event.getEventType()) { + clearCache(instId); + inspectionAlertListenService.clearByFlowInstanceId(instId); + } + // 1.节点完成才记录当前item下node if (StrUtil.isNotBlank(itemId) && StrUtil.isNotBlank(nodeId) && FlowExecutionEvent.EventType.NODE_COMPLETED == event.getEventType()) { - itemNodeExecutMap.computeIfAbsent(itemId, k -> new HashSet<>()).add(nodeId); + itemNodeExecutMap.computeIfAbsent(itemExecutionKey(instId, itemId), + k -> java.util.concurrent.ConcurrentHashMap.newKeySet()).add(nodeId); } // 任意事件：当前任务绑定item（item一启动就入Map，所以统计已完成item要-1） if (StrUtil.isNotBlank(instId) && StrUtil.isNotBlank(itemId)) { - taskItemExecutMap.computeIfAbsent(instId, k -> new HashSet<>()).add(itemId); + taskItemExecutMap.computeIfAbsent(instId, + k -> java.util.concurrent.ConcurrentHashMap.newKeySet()).add(itemId); } - InspectionTaskInstance taskInstance = inspectionTaskInstanceService.lambdaQuery() - .eq(InspectionTaskInstance::getTaskInsId, instId) - .one(); + InspectionTaskInstance taskInstance = findTaskInstance(instId); if (taskInstance == null) { log.warn("未查询到巡检实例,taskInsId:{}", instId); return; @@ -77,13 +93,13 @@ public class InspectionFlowExecutionListener implements FlowExecutionListener { case TASK_COMPLETED: targetStatus = TaskStatusEnum.SUCCESS.getCode(); needUpdateDb = true; - clearCache(instId); pushCompleteMsg(taskInstance.getId(), now, targetStatus); break; case TASK_FAILED: targetStatus = TaskStatusEnum.FAILED.getCode(); needUpdateDb = true; - clearCache(instId); + break; + case TASK_STOPPED: break; case NODE_COMPLETED: default: @@ -96,6 +112,15 @@ public class InspectionFlowExecutionListener implements FlowExecutionListener { inspectionTaskInstanceService.updateById(taskInstance); } + if (FlowExecutionEvent.EventType.NODE_COMPLETED == event.getEventType()) { + try { + inspectionResultService.recordWorkflowResult(event, taskInstance); + } catch (Exception e) { + log.error("保存工作流巡检结果失败,taskInstanceId:{},nodeId:{}", + taskInstance.getId(), nodeId, e); + } + } + // 节点推送 if (StrUtil.isNotBlank(nodeId)) { int waypointCount = taskInstance.getWaypointCount() == null ? 0 : taskInstance.getWaypointCount(); @@ -107,7 +132,8 @@ public class InspectionFlowExecutionListener implements FlowExecutionListener { int finishedItemNum = itemTotal - 1; // 当前item已完成节点数 - Set finishedNodeSet = itemNodeExecutMap.getOrDefault(itemId, new HashSet<>()); + Set finishedNodeSet = itemNodeExecutMap.getOrDefault( + itemExecutionKey(instId, itemId), new HashSet<>()); int finishedNode = finishedNodeSet.size(); int totalNode = event.getNodeCount() == 0 ? 1 : event.getNodeCount(); @@ -159,10 +185,35 @@ public class InspectionFlowExecutionListener implements FlowExecutionListener { /** 任务结束统一清理缓存 */ private void clearCache(String taskInsId) { Set itemList = taskItemExecutMap.getOrDefault(taskInsId, new HashSet<>()); - itemList.forEach(itemNodeExecutMap::remove); + itemList.forEach(itemId -> itemNodeExecutMap.remove(itemExecutionKey(taskInsId, itemId))); taskItemExecutMap.remove(taskInsId); } + private String itemExecutionKey(String taskInsId, String itemId) { + return taskInsId + "|" + itemId; + } + + /** + * 工作流是异步启动的，首个节点事件可能略早于巡检实例写入task_ins_id，短暂重试消除该竞态。 + */ + private InspectionTaskInstance findTaskInstance(String taskInsId) { + for (int attempt = 0; attempt < 5; attempt++) { + InspectionTaskInstance instance = inspectionTaskInstanceService.lambdaQuery() + .eq(InspectionTaskInstance::getTaskInsId, taskInsId) + .one(); + if (instance != null) { + return instance; + } + try { + Thread.sleep(50L); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + return null; + } + /** * 推送完成任务信息（参数dbInsId为数据库主键ID） */ diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMapper.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMapper.java new file mode 100644 index 0000000..bc46545 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMapper.java @@ -0,0 +1,9 @@ +package com.cmvr.inspection.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.cmvr.inspection.domain.InspectionResult; + +/** 巡检结果Mapper。 */ +public interface InspectionResultMapper extends BaseMapper +{ +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMediaMapper.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMediaMapper.java new file mode 100644 index 0000000..c0fa3d9 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/mapper/InspectionResultMediaMapper.java @@ -0,0 +1,9 @@ +package com.cmvr.inspection.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.cmvr.inspection.domain.InspectionResultMedia; + +/** 巡检结果媒体Mapper。 */ +public interface InspectionResultMediaMapper extends BaseMapper +{ +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionResultService.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionResultService.java new file mode 100644 index 0000000..0d7cc87 --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/IInspectionResultService.java @@ -0,0 +1,72 @@ +package com.cmvr.inspection.service; + +import com.baomidou.mybatisplus.extension.service.IService; +import com.cmvr.device.service.InspectionAlertListenService; +import com.cmvr.inspection.domain.InspectionDetectionAlert; +import com.cmvr.inspection.domain.InspectionResult; +import com.cmvr.inspection.domain.InspectionTaskInstance; +import com.cmvr.inspection.domain.dto.InspectionResultQuery; +import com.cmvr.inspection.domain.dto.InspectionResultReviewRequest; +import com.cmvr.inspection.domain.vo.InspectionReportSummaryVo; +import com.cmvr.inspection.domain.vo.InspectionResultVo; +import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; + +import java.util.List; + +/** 统一巡检结果业务接口。 */ +public interface IInspectionResultService extends IService +{ + /** + * 按任务、类型、状态、级别等条件查询巡检结果。 + * + * @param query 查询条件，允许为空 + * @return 按发生时间倒序排列的巡检结果 + */ + List selectResultList(InspectionResultQuery query); + + /** + * 查询巡检结果和全部媒体证据。 + * + * @param id 巡检结果ID + * @return 结果详情，不存在时返回null + */ + InspectionResultVo selectResultById(String id); + + /** + * 对待复核或识别失败结果进行人工判定。 + * + * @param id 巡检结果ID + * @param request 复核结论、告警级别、说明和版本 + * @return 更新后的巡检结果 + */ + InspectionResult review(String id, InspectionResultReviewRequest request); + + /** + * 汇总单个巡检任务实例的结果数量、告警级别和复核完成状态。 + * + * @param taskInstanceId 巡检任务实例数据库ID + * @return 报表汇总 + */ + InspectionReportSummaryVo buildReportSummary(String taskInstanceId); + + /** + * 从工作流节点标准输出中提取并保存巡检结果。 + * + * @param event 节点完成事件 + * @param taskInstance 巡检任务实例 + * @return 保存后的结果；节点无巡检结果输出时返回null + */ + InspectionResult recordWorkflowResult(FlowExecutionEvent event, InspectionTaskInstance taskInstance); + + /** + * 将边缘端PPE事件转换为统一巡检结果，并按异常级别生成告警。 + * + * @param alert 已保存的PPE原始事件 + * @param subscription 命中的工作流监听订阅 + * @param taskInstance 巡检任务实例，极端竞态下允许为空 + * @return 保存后的PPE巡检结果 + */ + InspectionResult recordPpeResult(InspectionDetectionAlert alert, + InspectionAlertListenService.ListenSubscription subscription, + InspectionTaskInstance taskInstance); +} diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionDetectionAlertServiceImpl.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionDetectionAlertServiceImpl.java index a6cce88..ef433ef 100644 --- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionDetectionAlertServiceImpl.java +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionDetectionAlertServiceImpl.java @@ -4,13 +4,16 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.cmvr.common.config.properties.MinioProperties; import com.cmvr.common.core.minio.MinioService; import com.cmvr.device.service.InspectionAlertListenService; +import com.cmvr.inspection.domain.InspectionTaskInstance; import com.cmvr.inspection.domain.InspectionDetectionAlert; import com.cmvr.inspection.domain.dto.alert.AlertEnvelope; import com.cmvr.inspection.domain.dto.alert.AlertImage; import com.cmvr.inspection.exception.DetectionAlertBadRequestException; import com.cmvr.inspection.exception.DetectionAlertTemporaryException; import com.cmvr.inspection.mapper.InspectionDetectionAlertMapper; +import com.cmvr.inspection.mapper.InspectionTaskInstanceMapper; import com.cmvr.inspection.service.IInspectionDetectionAlertService; +import com.cmvr.inspection.service.IInspectionResultService; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -26,6 +29,7 @@ import java.util.Base64; import java.util.Date; import java.util.Locale; import java.util.UUID; +import java.util.List; /** * PPE 违规报警接收业务实现。 @@ -49,6 +53,8 @@ public class InspectionDetectionAlertServiceImpl implements IInspectionDetection private final MinioProperties minioProperties; private final ObjectMapper objectMapper; private final InspectionAlertListenService inspectionAlertListenService; + private final InspectionTaskInstanceMapper taskInstanceMapper; + private final IInspectionResultService inspectionResultService; @Override @Transactional(rollbackFor = Exception.class) @@ -58,7 +64,10 @@ public class InspectionDetectionAlertServiceImpl implements IInspectionDetection String idempotencyKey = trimToNull(requestIdempotencyKey); // 工作流未开启对应 IP + 事件类型监听时，报警只确认接收，不做持久化。 - if (!inspectionAlertListenService.shouldStore(envelope.getGrpcIp(), envelope.getPayload().getLabels())) + List matchedSubscriptions = + inspectionAlertListenService.matchSubscriptions( + envelope.getGrpcIp(), envelope.getPayload().getLabels()); + if (matchedSubscriptions.isEmpty()) { log.info("忽略未监听的PPE报警，eventId={}，grpcIp={}，labels={}", eventId, envelope.getGrpcIp(), envelope.getPayload().getLabels()); @@ -88,6 +97,17 @@ public class InspectionDetectionAlertServiceImpl implements IInspectionDetection } } + for (InspectionAlertListenService.ListenSubscription subscription : matchedSubscriptions) + { + InspectionTaskInstance taskInstance = findTaskInstance(subscription.getFlowInstanceId()); + if (taskInstance == null) + { + log.warn("PPE报警未找到巡检任务实例，eventId={}，flowInstanceId={}", + eventId, subscription.getFlowInstanceId()); + } + inspectionResultService.recordPpeResult(alert, subscription, taskInstance); + } + log.info("PPE报警接收成功，alertId={}，eventId={}，sourceId={}", alert.getId(), eventId, envelope.getSourceId()); return true; @@ -103,6 +123,16 @@ public class InspectionDetectionAlertServiceImpl implements IInspectionDetection } } + private InspectionTaskInstance findTaskInstance(String flowInstanceId) + { + if (StringUtils.isBlank(flowInstanceId)) + { + return null; + } + return taskInstanceMapper.selectOne(new LambdaQueryWrapper() + .eq(InspectionTaskInstance::getTaskInsId, flowInstanceId).last("limit 1")); + } + /** * 校验协议关键字段，并返回用于重复过滤的 event_id。 */ diff --git a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionResultServiceImpl.java b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionResultServiceImpl.java new file mode 100644 index 0000000..c19b79d --- /dev/null +++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionResultServiceImpl.java @@ -0,0 +1,739 @@ +package com.cmvr.inspection.service.impl; + +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import com.cmvr.common.exception.ServiceException; +import com.cmvr.common.utils.DateUtils; +import com.cmvr.common.utils.SecurityUtils; +import com.cmvr.device.service.InspectionAlertListenService; +import com.cmvr.framework.websocket.service.MessagePushService; +import com.cmvr.inspection.domain.InspectionAlarm; +import com.cmvr.inspection.domain.InspectionDetectionAlert; +import com.cmvr.inspection.domain.InspectionResult; +import com.cmvr.inspection.domain.InspectionResultMedia; +import com.cmvr.inspection.domain.InspectionRobot; +import com.cmvr.inspection.domain.InspectionTask; +import com.cmvr.inspection.domain.InspectionTaskInstance; +import com.cmvr.inspection.domain.InspectionTaskLog; +import com.cmvr.inspection.domain.dto.InspectionResultQuery; +import com.cmvr.inspection.domain.dto.InspectionResultReviewRequest; +import com.cmvr.inspection.domain.vo.InspectionReportSummaryVo; +import com.cmvr.inspection.domain.vo.InspectionResultVo; +import com.cmvr.inspection.enums.InspectionLogTypeEnum; +import com.cmvr.inspection.enums.TaskStatusEnum; +import com.cmvr.inspection.mapper.InspectionAlarmMapper; +import com.cmvr.inspection.mapper.InspectionResultMapper; +import com.cmvr.inspection.mapper.InspectionResultMediaMapper; +import com.cmvr.inspection.mapper.InspectionRobotMapper; +import com.cmvr.inspection.mapper.InspectionTaskMapper; +import com.cmvr.inspection.mapper.InspectionTaskInstanceMapper; +import com.cmvr.inspection.service.IInspectionResultService; +import com.cmvr.inspection.service.IInspectionTaskLogService; +import com.cmvr.test.flow.runtime.event.FlowExecutionEvent; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.BeanUtils; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.math.BigDecimal; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.HashSet; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * 统一巡检结果业务实现。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class InspectionResultServiceImpl extends ServiceImpl + implements IInspectionResultService +{ + private static final String RESULT_MARKER = "_inspectionResult"; + private static final String STATUS_PENDING = "PENDING"; + private static final String STATUS_NORMAL = "NORMAL"; + private static final String STATUS_ABNORMAL = "ABNORMAL"; + private static final String STATUS_RECOGNIZE_FAILED = "RECOGNIZE_FAILED"; + + private final InspectionResultMediaMapper mediaMapper; + private final InspectionAlarmMapper alarmMapper; + private final InspectionTaskMapper taskMapper; + private final InspectionTaskInstanceMapper taskInstanceMapper; + private final InspectionRobotMapper robotMapper; + private final IInspectionTaskLogService taskLogService; + private final MessagePushService messagePushService; + + @Override + public List selectResultList(InspectionResultQuery query) + { + InspectionResultQuery condition = query == null ? new InspectionResultQuery() : query; + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(StringUtils.isNotBlank(condition.getTaskInstanceId()), + InspectionResult::getTaskInstanceId, condition.getTaskInstanceId()) + .eq(StringUtils.isNotBlank(condition.getTaskId()), + InspectionResult::getTaskId, condition.getTaskId()) + .eq(StringUtils.isNotBlank(condition.getItemId()), + InspectionResult::getItemId, condition.getItemId()) + .eq(StringUtils.isNotBlank(condition.getResultType()), + InspectionResult::getResultType, condition.getResultType()) + .eq(StringUtils.isNotBlank(condition.getResultStatus()), + InspectionResult::getResultStatus, condition.getResultStatus()) + .eq(condition.getAlarmLevel() != null, + InspectionResult::getAlarmLevel, condition.getAlarmLevel()) + .like(StringUtils.isNotBlank(condition.getResultName()), + InspectionResult::getResultName, condition.getResultName()) + .orderByDesc(InspectionResult::getOccurredTime, InspectionResult::getCreateTime); + return toResultVoList(list(wrapper)); + } + + @Override + public InspectionResultVo selectResultById(String id) + { + InspectionResult result = getById(id); + if (result == null) + { + return null; + } + InspectionResultVo vo = new InspectionResultVo(); + BeanUtils.copyProperties(result, vo); + if (StringUtils.isNotBlank(result.getTaskId())) + { + InspectionTask task = taskMapper.selectById(result.getTaskId()); + vo.setTaskName(task == null ? null : task.getTaskName()); + } + List mediaList = mediaMapper.selectList(new LambdaQueryWrapper() + .eq(InspectionResultMedia::getResultId, id) + .orderByAsc(InspectionResultMedia::getSortOrder)); + vo.setMediaList(mediaList); + return vo; + } + + /** + * 批量补充任务名称并转换为列表视图，避免按结果逐条查询任务表。 + */ + private List toResultVoList(List results) + { + if (results == null || results.isEmpty()) + { + return Collections.emptyList(); + } + + Set taskIds = results.stream() + .map(InspectionResult::getTaskId) + .filter(StringUtils::isNotBlank) + .collect(Collectors.toCollection(HashSet::new)); + Map taskMap = taskIds.isEmpty() + ? Collections.emptyMap() + : taskMapper.selectBatchIds(taskIds).stream() + .collect(Collectors.toMap(InspectionTask::getId, Function.identity(), (left, right) -> left)); + + List resultVos = new ArrayList<>(results.size()); + for (InspectionResult result : results) + { + InspectionResultVo vo = new InspectionResultVo(); + BeanUtils.copyProperties(result, vo); + InspectionTask task = taskMap.get(result.getTaskId()); + vo.setTaskName(task == null ? null : task.getTaskName()); + resultVos.add(vo); + } + return resultVos; + } + + /** + * 提交人工复核。使用reviewVersion做条件更新，保证并发审核只有一个请求成功。 + */ + @Override + @Transactional(rollbackFor = Exception.class) + public InspectionResult review(String id, InspectionResultReviewRequest request) + { + if (request == null) + { + throw new ServiceException("复核参数不能为空"); + } + String targetStatus = StringUtils.upperCase(StringUtils.trimToEmpty(request.getResultStatus())); + if (!STATUS_NORMAL.equals(targetStatus) && !STATUS_ABNORMAL.equals(targetStatus)) + { + throw new ServiceException("人工复核结果只能为NORMAL或ABNORMAL"); + } + if (STATUS_ABNORMAL.equals(targetStatus) + && (request.getAlarmLevel() == null || request.getAlarmLevel() < 1 || request.getAlarmLevel() > 3)) + { + throw new ServiceException("异常结果必须指定1到3级告警级别"); + } + + InspectionResult current = getById(id); + if (current == null) + { + throw new ServiceException("巡检结果不存在"); + } + if (!STATUS_PENDING.equals(current.getResultStatus()) + && !STATUS_RECOGNIZE_FAILED.equals(current.getResultStatus())) + { + throw new ServiceException("该巡检结果已完成复核，不能重复提交"); + } + + int expectedVersion = request.getReviewVersion() == null + ? valueOrZero(current.getReviewVersion()) : request.getReviewVersion(); + String username = SecurityUtils.getUsername(); + Date now = DateUtils.getNowDate(); + LambdaUpdateWrapper update = new LambdaUpdateWrapper<>(); + update.eq(InspectionResult::getId, id) + .eq(InspectionResult::getReviewVersion, expectedVersion) + .set(InspectionResult::getResultStatus, targetStatus) + .set(InspectionResult::getAlarmLevel, + STATUS_ABNORMAL.equals(targetStatus) ? request.getAlarmLevel() : null) + .set(InspectionResult::getAlarmMessage, + STATUS_ABNORMAL.equals(targetStatus) ? "人工复核判定异常" : null) + .set(InspectionResult::getReviewer, username) + .set(InspectionResult::getReviewTime, now) + .set(InspectionResult::getReviewRemark, request.getReviewRemark()) + .set(InspectionResult::getReviewVersion, expectedVersion + 1) + .set(InspectionResult::getUpdateBy, username) + .set(InspectionResult::getUpdateTime, now); + if (!update(update)) + { + throw new ServiceException("巡检结果已被其他人员修改，请刷新后重试"); + } + + InspectionResult reviewed = getById(id); + if (STATUS_ABNORMAL.equals(reviewed.getResultStatus())) + { + createAlarmIfNecessary(reviewed); + } + appendResultLog(reviewed, "人工复核完成"); + pushResult(reviewed); + return reviewed; + } + + @Override + public InspectionReportSummaryVo buildReportSummary(String taskInstanceId) + { + List results = list(new LambdaQueryWrapper() + .eq(InspectionResult::getTaskInstanceId, taskInstanceId)); + InspectionReportSummaryVo summary = new InspectionReportSummaryVo(); + summary.setTaskInstanceId(taskInstanceId); + summary.setTotalCount(results.size()); + + Map levelCounts = new LinkedHashMap<>(); + levelCounts.put(1, 0L); + levelCounts.put(2, 0L); + levelCounts.put(3, 0L); + Map typeCounts = new LinkedHashMap<>(); + for (InspectionResult result : results) + { + switch (StringUtils.defaultString(result.getResultStatus())) + { + case STATUS_NORMAL: + summary.setNormalCount(summary.getNormalCount() + 1); + break; + case STATUS_ABNORMAL: + summary.setAbnormalCount(summary.getAbnormalCount() + 1); + break; + case STATUS_PENDING: + summary.setPendingCount(summary.getPendingCount() + 1); + break; + case STATUS_RECOGNIZE_FAILED: + summary.setRecognizeFailedCount(summary.getRecognizeFailedCount() + 1); + break; + default: + break; + } + if (result.getAlarmLevel() != null) + { + levelCounts.put(result.getAlarmLevel(), levelCounts.getOrDefault(result.getAlarmLevel(), 0L) + 1); + } + String type = StringUtils.defaultIfBlank(result.getResultType(), "UNKNOWN"); + typeCounts.put(type, typeCounts.getOrDefault(type, 0L) + 1); + } + summary.setAlarmLevelCounts(levelCounts); + summary.setResultTypeCounts(typeCounts); + summary.setReviewCompleted(summary.getPendingCount() == 0 && summary.getRecognizeFailedCount() == 0); + return summary; + } + + /** + * 消费工作流节点的标准巡检输出。普通节点没有_inspectionResult时直接忽略。 + */ + @Override + @Transactional(rollbackFor = Exception.class) + public InspectionResult recordWorkflowResult(FlowExecutionEvent event, InspectionTaskInstance taskInstance) + { + if (event == null || event.getOutputParams() == null) + { + return null; + } + JSONObject source = event.getOutputParams().getJSONObject(RESULT_MARKER); + if (source == null) + { + return null; + } + + String iterationPath = JSON.toJSONString(event.getIterations() == null + ? Collections.emptyList() : event.getIterations()); + String dedupeKey = limit("WF|" + event.getInstId() + "|" + event.getItemId() + "|" + + event.getNodeId() + "|" + iterationPath, 255); + InspectionResult existing = findByDedupeKey(dedupeKey); + if (existing != null) + { + return existing; + } + + InspectionResult result = baseResult(taskInstance, event.getInstId(), event.getItemId(), event.getNodeId()); + result.setDedupeKey(dedupeKey); + result.setNodeName(event.getNodeName()); + result.setIterationPath(iterationPath); + result.setResultType(upper(source.getString("resultType"))); + result.setResultStatus(upper(source.getString("resultStatus"))); + result.setResultName(StringUtils.defaultIfBlank(source.getString("resultName"), event.getNodeName())); + result.setValueNumber(source.getBigDecimal("valueNumber")); + result.setValueText(source.getString("valueText")); + result.setUnit(source.getString("unit")); + result.setAlarmLevel(source.getInteger("alarmLevel")); + result.setAlarmMessage(source.getString("alarmMessage")); + result.setConfidence(source.getBigDecimal("confidence")); + result.setEvidenceUrl(source.getString("evidenceUrl")); + result.setEvidenceType(resolveSourceEvidenceType(source)); + result.setRawResult(toJsonText(source.get("rawResult"))); + result.setMatchedRuleJson(toJsonText(source.get("matchedRule"))); + result.setRuleSnapshotJson(toJsonText(source.get("ruleSnapshot"))); + validateStandardResult(result); + + try + { + insertResult(result); + } + catch (DuplicateKeyException ex) + { + return findByDedupeKey(dedupeKey); + } + saveMedia(result.getId(), source); + postCreate(result, "巡检节点产生结果"); + return result; + } + + /** + * 将一条PPE原始报警按命中的工作流订阅转换为巡检结果。 + */ + @Override + @Transactional(rollbackFor = Exception.class) + public InspectionResult recordPpeResult(InspectionDetectionAlert alert, + InspectionAlertListenService.ListenSubscription subscription, + InspectionTaskInstance taskInstance) + { + String dedupeKey = limit("PPE|" + alert.getEventId() + "|" + subscription.getSubscriptionId(), 255); + InspectionResult existing = findByDedupeKey(dedupeKey); + if (existing != null) + { + return existing; + } + + InspectionResult result = baseResult(taskInstance, subscription.getFlowInstanceId(), + subscription.getItemId(), subscription.getNodeId()); + result.setDedupeKey(dedupeKey); + result.setResultType("PPE"); + result.setResultStatus(STATUS_ABNORMAL); + result.setResultName(ppeName(subscription.getEventType())); + result.setValueText(subscription.getEventType()); + result.setAlarmLevel("No-Helmet".equals(subscription.getEventType()) ? 3 : 2); + result.setAlarmMessage(ppeMessage(subscription.getEventType())); + result.setConfidence(alert.getMaxConfidence() == null + ? null : BigDecimal.valueOf(alert.getMaxConfidence())); + result.setSourceEventId(alert.getEventId()); + result.setEvidenceUrl(alert.getImagePath()); + result.setEvidenceType(StringUtils.isBlank(alert.getImagePath()) ? null : "IMAGE"); + result.setRawResult(alert.getRawJson()); + result.setOccurredTime(fromUnixNs(alert.getTriggeredAtNs())); + + try + { + insertResult(result); + } + catch (DuplicateKeyException ex) + { + return findByDedupeKey(dedupeKey); + } + if (StringUtils.isNotBlank(result.getEvidenceUrl())) + { + insertMedia(result.getId(), "IMAGE", result.getEvidenceUrl(), 0); + } + postCreate(result, "接收到PPE违规事件"); + return result; + } + + private InspectionResult baseResult(InspectionTaskInstance taskInstance, String flowInstanceId, + String itemId, String nodeId) + { + InspectionResult result = new InspectionResult(); + result.setId(newId()); + result.setResultCode("IR" + new SimpleDateFormat("yyyyMMddHHmmssSSS").format(new Date()) + + result.getId().substring(0, 5).toUpperCase(Locale.ROOT)); + result.setFlowInstanceId(flowInstanceId); + result.setItemId(itemId); + result.setNodeId(nodeId); + if (taskInstance != null) + { + result.setTaskInstanceId(taskInstance.getId()); + result.setTaskId(taskInstance.getTaskId()); + } + result.setReviewVersion(0); + result.setOccurredTime(DateUtils.getNowDate()); + result.setCreateBy("system"); + result.setCreateTime(DateUtils.getNowDate()); + return result; + } + + private void insertResult(InspectionResult result) + { + if (baseMapper.insert(result) != 1) + { + throw new ServiceException("保存巡检结果失败"); + } + } + + private void postCreate(InspectionResult result, String logPrefix) + { + if (STATUS_ABNORMAL.equals(result.getResultStatus())) + { + createAlarmIfNecessary(result); + } + appendResultLog(result, logPrefix); + pushResult(result); + } + + /** + * 为异常结果幂等创建巡检告警。数据库result_id唯一索引是并发场景的最终保障。 + */ + private void createAlarmIfNecessary(InspectionResult result) + { + InspectionAlarm existing = alarmMapper.selectOne(new LambdaQueryWrapper() + .eq(InspectionAlarm::getResultId, result.getId()).last("limit 1")); + if (existing != null) + { + if (!StringUtils.equals(existing.getId(), result.getAlarmId())) + { + result.setAlarmId(existing.getId()); + baseMapper.updateById(result); + } + return; + } + + InspectionTask task = StringUtils.isBlank(result.getTaskId()) + ? null : taskMapper.selectById(result.getTaskId()); + InspectionTaskInstance taskInstance = null; + if (StringUtils.isNotBlank(result.getTaskInstanceId())) + { + taskInstance = new InspectionTaskInstance(); + taskInstance.setId(result.getTaskInstanceId()); + } + InspectionRobot robot = null; + if (taskInstance != null) + { + // 仅在需要构建告警时查询完整任务实例。 + taskInstance = taskInstanceMapper.selectById(result.getTaskInstanceId()); + if (taskInstance != null && StringUtils.isNotBlank(taskInstance.getRobotId())) + { + robot = robotMapper.selectById(taskInstance.getRobotId()); + } + } + + InspectionAlarm alarm = new InspectionAlarm(); + alarm.setId(newId()); + alarm.setAlarmCode("IA" + new SimpleDateFormat("yyyyMMddHHmmssSSS").format(new Date()) + + alarm.getId().substring(0, 4).toUpperCase(Locale.ROOT)); + alarm.setResultId(result.getId()); + alarm.setAlarmSource(result.getResultType()); + alarm.setItemId(result.getItemId()); + alarm.setNodeId(result.getNodeId()); + alarm.setTaskInstanceId(result.getTaskInstanceId()); + alarm.setTaskId(result.getTaskId()); + alarm.setTaskName(task == null ? null : task.getTaskName()); + alarm.setRobotId(taskInstance == null ? null : taskInstance.getRobotId()); + alarm.setRobotName(robot == null ? null : robot.getRobotName()); + alarm.setAlarmLevel(String.valueOf(result.getAlarmLevel() == null ? 2 : result.getAlarmLevel())); + alarm.setAlarmType("METER".equals(result.getResultType()) ? "1" : "2"); + alarm.setAlarmTitle(result.getResultName() + "异常"); + alarm.setAlarmContent(buildAlarmContent(result)); + alarm.setAlarmTime(result.getOccurredTime()); + alarm.setHandleStatus("0"); + alarm.setEvidenceImage(result.getEvidenceUrl()); + alarm.setCreateBy("system"); + alarm.setCreateTime(DateUtils.getNowDate()); + alarmMapper.insert(alarm); + + result.setAlarmId(alarm.getId()); + baseMapper.updateById(result); + try + { + messagePushService.pushToChannel("InspectionAlarm", alarm); + } + catch (Exception ex) + { + log.info("巡检告警无订阅者，alarmId={}", alarm.getId()); + } + } + + private String buildAlarmContent(InspectionResult result) + { + if ("METER".equals(result.getResultType())) + { + return result.getResultName() + "读数为" + StringUtils.defaultString(result.getValueText()) + + StringUtils.defaultString(result.getUnit()) + "，触发" + result.getAlarmLevel() + + "级告警：" + StringUtils.defaultString(result.getAlarmMessage()); + } + return StringUtils.defaultIfBlank(result.getAlarmMessage(), result.getResultName() + "检测异常"); + } + + /** 将业务结果追加到现有巡检任务时间线，并通知任务详情订阅者。 */ + private void appendResultLog(InspectionResult result, String prefix) + { + if (StringUtils.isBlank(result.getTaskInstanceId())) + { + log.warn("巡检结果未绑定任务实例，跳过任务日志，resultId={}，flowInstanceId={}", + result.getId(), result.getFlowInstanceId()); + return; + } + InspectionTaskLog taskLog = InspectionTaskLog.builder() + .id(newId()) + .taskInstanceId(result.getTaskInstanceId()) + .taskId(result.getTaskId()) + .itemId(result.getItemId()) + .nodeId(result.getNodeId()) + .nodeName(result.getNodeName()) + .logType(resolveLogType(result.getEvidenceUrl())) + .logContent(prefix + "：" + resultDescription(result)) + .mediaUrl(result.getEvidenceUrl()) + .status(TaskStatusEnum.SUCCESS.getCode()) + .logTime(DateUtils.getNowDate()) + .extraInfo(JSON.toJSONString(Collections.singletonMap("resultId", result.getId()))) + .build(); + taskLogService.insertInspectionTaskLog(taskLog); + try + { + messagePushService.pushToChannel("InspectionTaskInstance", taskLog); + } + catch (Exception ex) + { + log.info("巡检结果日志无订阅者，resultId={}", result.getId()); + } + } + + private String resultDescription(InspectionResult result) + { + if ("METER".equals(result.getResultType())) + { + return result.getResultName() + "=" + StringUtils.defaultString(result.getValueText()) + + StringUtils.defaultString(result.getUnit()) + "，结果=" + result.getResultStatus() + + (result.getAlarmLevel() == null ? "" : "，告警级别=" + result.getAlarmLevel()); + } + return result.getResultName() + "，结果=" + result.getResultStatus(); + } + + private void pushResult(InspectionResult result) + { + try + { + messagePushService.pushToChannel("InspectionResult", result); + } + catch (Exception ex) + { + log.info("巡检结果无订阅者，resultId={}", result.getId()); + } + } + + /** 保存人工巡检的多媒体证据；只有单一证据时也统一落媒体表。 */ + private void saveMedia(String resultId, JSONObject source) + { + JSONArray mediaList = source.getJSONArray("mediaList"); + int order = 0; + if (mediaList != null) + { + for (Object item : mediaList) + { + if (item instanceof JSONObject) + { + JSONObject media = (JSONObject) item; + insertMedia(resultId, upper(media.getString("mediaType")), media.getString("mediaUrl"), order++); + } + else if (item != null) + { + String url = String.valueOf(item); + insertMedia(resultId, detectMediaType(url), url, order++); + } + } + } + if (order == 0 && StringUtils.isNotBlank(source.getString("evidenceUrl"))) + { + insertMedia(resultId, detectMediaType(source.getString("evidenceUrl")), + source.getString("evidenceUrl"), 0); + } + } + + private void insertMedia(String resultId, String mediaType, String mediaUrl, int order) + { + if (StringUtils.isBlank(mediaUrl)) + { + return; + } + InspectionResultMedia media = new InspectionResultMedia(); + media.setId(newId()); + media.setResultId(resultId); + media.setMediaType(normalizeEvidenceType( + StringUtils.defaultIfBlank(mediaType, detectMediaType(mediaUrl)))); + media.setMediaUrl(mediaUrl); + media.setSortOrder(order); + media.setCreateTime(DateUtils.getNowDate()); + mediaMapper.insert(media); + } + + private InspectionResult findByDedupeKey(String dedupeKey) + { + return baseMapper.selectOne(new LambdaQueryWrapper() + .eq(InspectionResult::getDedupeKey, dedupeKey).last("limit 1")); + } + + private void validateStandardResult(InspectionResult result) + { + if (!("PPE".equals(result.getResultType()) || "METER".equals(result.getResultType()) + || "MANUAL".equals(result.getResultType()))) + { + throw new ServiceException("工作流巡检结果类型不支持: " + result.getResultType()); + } + if (!(STATUS_PENDING.equals(result.getResultStatus()) || STATUS_NORMAL.equals(result.getResultStatus()) + || STATUS_ABNORMAL.equals(result.getResultStatus()) + || STATUS_RECOGNIZE_FAILED.equals(result.getResultStatus()))) + { + throw new ServiceException("工作流巡检结果状态不支持: " + result.getResultStatus()); + } + if (STATUS_ABNORMAL.equals(result.getResultStatus()) + && (result.getAlarmLevel() == null || result.getAlarmLevel() < 1 || result.getAlarmLevel() > 3)) + { + throw new ServiceException("异常巡检结果必须指定1到3级告警级别"); + } + } + + private int resolveLogType(String url) + { + String type = detectMediaType(url); + if ("VIDEO".equals(type)) + { + return InspectionLogTypeEnum.VIDEO.getCode(); + } + if ("IMAGE".equals(type)) + { + return InspectionLogTypeEnum.IMAGE.getCode(); + } + return InspectionLogTypeEnum.TEXT.getCode(); + } + + private String detectMediaType(String url) + { + String normalized = StringUtils.lowerCase(StringUtils.defaultString(url)); + if (normalized.matches(".*\\.(mp4|avi|mov|mkv|webm)(\\?.*)?$")) + { + return "VIDEO"; + } + return "IMAGE"; + } + + /** + * 在结果首次入库时确定主要证据类型，列表查询无需再访问媒体明细表。 + */ + private String resolveSourceEvidenceType(JSONObject source) + { + String explicitType = normalizeEvidenceType(source.getString("evidenceType")); + if (StringUtils.isNotBlank(explicitType)) + { + return explicitType; + } + + // 人工巡检以mediaList第一条作为主要证据，和evidenceUrl的赋值规则保持一致。 + JSONArray mediaList = source.getJSONArray("mediaList"); + if (mediaList != null && !mediaList.isEmpty() && mediaList.get(0) instanceof JSONObject) + { + String firstType = normalizeEvidenceType(mediaList.getJSONObject(0).getString("mediaType")); + if (StringUtils.isNotBlank(firstType)) + { + return firstType; + } + } + + String evidenceUrl = source.getString("evidenceUrl"); + return StringUtils.isBlank(evidenceUrl) ? null : detectMediaType(evidenceUrl); + } + + /** 媒体类型只允许IMAGE或VIDEO，避免主表和媒体明细出现不一致的自由文本。 */ + private String normalizeEvidenceType(String value) + { + String normalized = upper(value); + if (StringUtils.isBlank(normalized)) + { + return null; + } + if (!"IMAGE".equals(normalized) && !"VIDEO".equals(normalized)) + { + throw new ServiceException("不支持的巡检证据类型: " + value); + } + return normalized; + } + + private String ppeName(String eventType) + { + return "No-Helmet".equals(eventType) ? "未佩戴安全帽" : "未佩戴手套"; + } + + private String ppeMessage(String eventType) + { + return "No-Helmet".equals(eventType) ? "检测到人员未佩戴安全帽" : "检测到人员未佩戴手套"; + } + + private Date fromUnixNs(Long timestampNs) + { + return timestampNs == null ? DateUtils.getNowDate() : new Date(timestampNs / 1_000_000L); + } + + private String toJsonText(Object value) + { + if (value == null) + { + return null; + } + return value instanceof String ? String.valueOf(value) : JSON.toJSONString(value); + } + + private String upper(String value) + { + return StringUtils.upperCase(StringUtils.trimToEmpty(value)); + } + + private int valueOrZero(Integer value) + { + return value == null ? 0 : value; + } + + private String limit(String value, int maxLength) + { + return value.length() <= maxLength ? value : value.substring(0, maxLength); + } + + private String newId() + { + return UUID.randomUUID().toString().replace("-", ""); + } +} diff --git a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionAlarmMapper.xml b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionAlarmMapper.xml index 02922f9..c0750c2 100644 --- a/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionAlarmMapper.xml +++ b/cmvr-iot-inspection/src/main/resources/mapper/inspection/InspectionAlarmMapper.xml @@ -13,6 +13,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + + + @@ -39,6 +43,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + + + + @@ -59,7 +67,11 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" - select id, create_by, create_time, update_by, update_time, remark, alarm_code, task_instance_id, task_id, task_name, robot_id, robot_name, alarm_level, alarm_type, alarm_title, alarm_content, alarm_location, alarm_time, handle_status, handler, handle_time, handle_remark, evidence_image from inspection_alarm + select id, create_by, create_time, update_by, update_time, remark, alarm_code, + task_instance_id, result_id, alarm_source, item_id, node_id, + task_id, task_name, robot_id, robot_name, alarm_level, alarm_type, + alarm_title, alarm_content, alarm_location, alarm_time, handle_status, + handler, handle_time, handle_remark, evidence_image from inspection_alarm select a.id, a.create_by, a.create_time, a.update_by, a.update_time, a.remark, - a.alarm_code, a.task_instance_id, a.task_id, t.task_name, a.robot_id, r.robot_name, + a.alarm_code, a.task_instance_id, a.result_id, a.alarm_source, a.item_id, a.node_id, + a.task_id, t.task_name, a.robot_id, r.robot_name, a.alarm_level, a.alarm_type, a.alarm_title, a.alarm_content, a.alarm_location, a.alarm_time, a.handle_status, a.handler, a.handle_time, a.handle_remark, a.evidence_image, u1.nick_name as create_by_name, u2.nick_name as update_by_name diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java index d0f76d2..3c00279 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java @@ -74,6 +74,7 @@ public enum ActionEnum { INTENT_RECOGNITION("LLM", "INTENT_RECOGNITION", "意图识别"), GENERATE_ADVANCED_AUDIO("LLM", "GENERATE_ADVANCED_AUDIO", "tts语音合成"), AI_AGENT_PLATFORM("LLM", "AI_AGENT_PLATFORM", "商道智能体"), + INSPECTION_METER_RECOGNIZE("LLM", "INSPECTION_METER_RECOGNIZE", "巡检仪表读数识别"), AI_TTS("LLM", "AI_TTS", "tts语音播放"), GET_CURRENT_PAGE("LLM", "GET_CURRENT_PAGE", "获取当前页面名称"), @@ -83,7 +84,8 @@ public enum ActionEnum { // 巡检报警 INSPECTION_ALERT_LISTEN_START("EDGE", "INSPECTION_ALERT_LISTEN_START", "开始监听报警事件"), - INSPECTION_ALERT_LISTEN_STOP("EDGE", "INSPECTION_ALERT_LISTEN_STOP", "结束监听报警事件") + INSPECTION_ALERT_LISTEN_STOP("EDGE", "INSPECTION_ALERT_LISTEN_STOP", "结束监听报警事件"), + INSPECTION_MANUAL_REVIEW_CREATE("EDGE", "INSPECTION_MANUAL_REVIEW_CREATE", "创建人工巡检判断任务") ; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java index 5e47fd1..bf0aec9 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/event/FlowExecutionEvent.java @@ -1,11 +1,15 @@ package com.cmvr.test.flow.runtime.event; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.test.enums.ActionEnum; import com.cmvr.test.enums.TaskStatusEnum; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; +import java.util.List; + /** * 流程执行事件。 *

@@ -62,6 +66,26 @@ public class FlowExecutionEvent { */ private String nodeName; + /** + * 当前节点动作，用于业务监听器识别标准化节点输出。 + */ + private ActionEnum action; + + /** + * 节点成功执行后的输出参数。 + */ + private JSONObject outputParams; + + /** + * 当前工作流绑定的终端ID。 + */ + private String terminalId; + + /** + * 循环节点迭代路径，用于区分同一节点的多次执行结果。 + */ + private List iterations; + /** * 当前任务状态。 */ diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java index 8ca89b5..f972f07 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowAfterInterceptor.java @@ -10,6 +10,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; +import java.util.ArrayList; import java.util.function.Function; @Slf4j @@ -55,6 +56,11 @@ public class FlowAfterInterceptor extends AbstractFlowMsgPreInterceptor{ .nodeId(message.getNodeId()) .nodeCount(message.getGraph().allNodeIds().size()) .nodeType(message.getNodeType()) + .action(message.getAction()) + .outputParams(result != null ? result.getOutputParams() : null) + .terminalId(message.getTerminalId()) + .iterations(message.getIterations() == null + ? new ArrayList<>() : new ArrayList<>(message.getIterations())) .errorMessage(errorMessage) .build(); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java index 39cf150..64a7ed1 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java @@ -37,15 +37,19 @@ public class EdgeInspectionAlertOperateService implements EdgeOperateService JSONObject inputParams = message.getInputParams() == null ? new JSONObject() : message.getInputParams(); String terminalId = StringUtils.defaultIfBlank(inputParams.getString("terminalId"), message.getTerminalId()); Collection eventTypes = resolveEventTypes(inputParams); + String listenerKey = inputParams.getString("listenerKey"); InspectionAlertListenService.ListenState state; switch (message.getAction()) { case INSPECTION_ALERT_LISTEN_START: - state = inspectionAlertListenService.startListen(terminalId, eventTypes); + state = inspectionAlertListenService.startListen(terminalId, eventTypes, + message.getInstId(), message.getTaskId(), message.getItemId(), + message.getNodeId(), listenerKey); break; case INSPECTION_ALERT_LISTEN_STOP: - state = inspectionAlertListenService.stopListen(terminalId, eventTypes); + state = inspectionAlertListenService.stopListen(terminalId, eventTypes, + message.getInstId(), message.getItemId(), listenerKey); break; default: throw new GlobalException("不支持的巡检报警监听动作: " + message.getAction()); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeManualInspectionOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeManualInspectionOperateService.java new file mode 100644 index 0000000..bb00ac4 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeManualInspectionOperateService.java @@ -0,0 +1,127 @@ +package com.cmvr.test.flow.runtime.operator.edge; + +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; +import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.model.vo.inspection.InspectionManualReviewConfigVO; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Service; + +import java.util.Collection; + +/** + * 创建待人工复核巡检结果，工作流不等待人工处理。 + * + *

输入参数见{@link InspectionManualReviewConfigVO}。节点只登记已由前置组件保存到MinIO的 + * 图片或视频地址，不复制媒体文件。节点执行后立即返回PENDING，人工通过巡检结果接口复核。

+ */ +@Service +public class EdgeManualInspectionOperateService implements EdgeOperateService +{ + private static final String RESULT_MARKER = "_inspectionResult"; + + @Override + public boolean supports(ActionEnum action) + { + return ActionEnum.INSPECTION_MANUAL_REVIEW_CREATE.equals(action); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) + { + JSONObject input = message.getInputParams() == null ? new JSONObject() : message.getInputParams(); + InspectionManualReviewConfigVO config = input.to(InspectionManualReviewConfigVO.class); + String resultName = StringUtils.defaultIfBlank(config.getResultName(), "人工巡检"); + JSONArray mediaList = resolveMediaList(input); + if (mediaList.isEmpty()) + { + throw new GlobalException("人工巡检至少需要一张图片或一个视频"); + } + + JSONObject inspectionResult = new JSONObject(); + inspectionResult.put("resultType", "MANUAL"); + inspectionResult.put("resultStatus", "PENDING"); + inspectionResult.put("resultName", resultName); + inspectionResult.put("evidenceUrl", mediaList.getJSONObject(0).getString("mediaUrl")); + inspectionResult.put("evidenceType", mediaList.getJSONObject(0).getString("mediaType")); + inspectionResult.put("mediaList", mediaList); + inspectionResult.put("rawResult", new JSONObject() + .fluentPut("reviewCriteria", config.getReviewCriteria()) + .fluentPut("defaultAlarmLevel", config.getDefaultAlarmLevel())); + + JSONObject output = new JSONObject(); + output.put("resultStatus", "PENDING"); + output.put("mediaList", mediaList); + output.put(RESULT_MARKER, inspectionResult); + return TaskNodeExecuteResult.success(output); + } + + /** + * 合并mediaList、imageUrl和videoUrl，并统一为标准媒体对象数组。 + */ + private JSONArray resolveMediaList(JSONObject input) + { + JSONArray result = new JSONArray(); + Object raw = input.get("mediaList"); + if (raw instanceof JSONArray) + { + for (Object item : (JSONArray) raw) + { + addMedia(result, item, null); + } + } + else if (raw instanceof Collection) + { + for (Object item : (Collection) raw) + { + addMedia(result, item, null); + } + } + else + { + addMedia(result, raw, null); + } + addMedia(result, input.get("imageUrl"), "IMAGE"); + addMedia(result, input.get("videoUrl"), "VIDEO"); + return result; + } + + /** 支持标准媒体对象，也兼容历史工作流直接传入URL字符串。 */ + private void addMedia(JSONArray target, Object value, String defaultType) + { + if (value == null) + { + return; + } + if (value instanceof JSONObject) + { + JSONObject source = (JSONObject) value; + String url = StringUtils.trimToNull(source.getString("mediaUrl")); + if (url != null) + { + target.add(new JSONObject() + .fluentPut("mediaUrl", url) + .fluentPut("mediaType", StringUtils.defaultIfBlank( + source.getString("mediaType"), detectType(url)))); + } + return; + } + String url = StringUtils.trimToNull(String.valueOf(value)); + if (url != null) + { + target.add(new JSONObject() + .fluentPut("mediaUrl", url) + .fluentPut("mediaType", StringUtils.defaultIfBlank(defaultType, detectType(url)))); + } + } + + /** 未显式指定类型时，根据常见视频后缀推断，其余按图片处理。 */ + private String detectType(String url) + { + String normalized = StringUtils.lowerCase(url); + return normalized.matches(".*\\.(mp4|avi|mov|mkv|webm)(\\?.*)?$") ? "VIDEO" : "IMAGE"; + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/ti/TiTouchOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/ti/TiTouchOperateService.java index 35f2813..ee25d61 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/ti/TiTouchOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/ti/TiTouchOperateService.java @@ -1,28 +1,23 @@ package com.cmvr.test.flow.runtime.operator.edge.ti; -import cn.hutool.core.io.FileUtil; -import cn.hutool.core.text.StrPool; import cn.hutool.core.util.StrUtil; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; -import com.cmvr.common.config.properties.MinioProperties; -import com.cmvr.common.core.minio.MinioService; import com.cmvr.common.exception.GlobalException; import com.cmvr.common.utils.http.CallAPIUtil; import com.cmvr.llm.config.APIProperties; import com.cmvr.llm.config.AgentConfig; -import com.cmvr.llm.util.LargeModelFileUploadUtil; import com.cmvr.test.enums.ActionEnum; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; import com.cmvr.test.flow.runtime.operator.edge.EdgeOperateService; import com.cmvr.test.service.ex.ExTiVehicleFunctionService; +import com.cmvr.test.service.ShangdaoFileService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; -import java.io.File; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -34,9 +29,8 @@ import java.util.Map; public class TiTouchOperateService implements EdgeOperateService { private final ExTiVehicleFunctionService exTiVehicleFunctionService; - private final MinioService minioService; - private final MinioProperties minioProps; private final APIProperties apiProperties; + private final ShangdaoFileService shangdaoFileService; // 设置运动模式 public static final List DRIVE_MODE_PATH = Arrays.asList("主页", "设置", "驾驶模式", "运动模式"); @@ -81,11 +75,11 @@ public class TiTouchOperateService implements EdgeOperateService { case TI_TOUCH_COORDINATES: { // 页面图片(相机) String imageUrl = inputParams.getString("imageUrl"); - String pageImageUrl = uploadImageFromMinio(imageUrl); + String pageImageUrl = shangdaoFileService.uploadFromMinio(imageUrl); // icon图片 String iconUrl = inputParams.getString("iconUrl"); - String iconImageUrl = uploadImageFromMinio(iconUrl); + String iconImageUrl = shangdaoFileService.uploadFromMinio(iconUrl); // 获取触控坐标 String touchCoordinates = getTouchCoordinates(pageImageUrl, iconImageUrl); @@ -109,10 +103,10 @@ public class TiTouchOperateService implements EdgeOperateService { } public String test(String imageUrl,String iconUrl) { - String pageImageUrl = uploadImageFromMinio(imageUrl); + String pageImageUrl = shangdaoFileService.uploadFromMinio(imageUrl); // icon图片 - String iconImageUrl = uploadImageFromMinio(iconUrl); + String iconImageUrl = shangdaoFileService.uploadFromMinio(iconUrl); // 获取触控坐标 return getTouchCoordinates(pageImageUrl, iconImageUrl); @@ -142,29 +136,6 @@ public class TiTouchOperateService implements EdgeOperateService { } } - /** - * 根据 MinIO 文件访问地址上传到商道，返回商道文件地址 - */ - private String uploadImageFromMinio(String url) { - try { - String bucketName = minioProps.getBucketName(); - String prefix = StrPool.SLASH + bucketName + StrPool.SLASH; - - String objectName = StrUtil.removePrefix( - url, - StrUtil.subBefore(url, prefix, true) + prefix - ); - - File imageFile = minioService.getFile(bucketName, objectName); - byte[] imageBytes = FileUtil.readBytes(imageFile); - - return LargeModelFileUploadUtil.uploadFile(imageBytes); - } catch (Exception e) { - log.error("上传 MinIO 文件到商道失败，url={}", url, e); - throw new GlobalException("图片上传失败"); - } - } - private String queryResultByRunId(String processId, AgentConfig agentConfig) { Map headers = new HashMap<>(); headers.put("Apikey", agentConfig.getAppKey()); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/InspectionMeterRecognizeOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/InspectionMeterRecognizeOperateService.java new file mode 100644 index 0000000..9d63e40 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/InspectionMeterRecognizeOperateService.java @@ -0,0 +1,310 @@ +package com.cmvr.test.flow.runtime.operator.llm; + +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; +import com.cmvr.llm.service.LLMAiAgentPlatformService; +import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.model.vo.inspection.InspectionAlarmRuleVO; +import com.cmvr.test.model.vo.inspection.InspectionMeterRecognizeConfigVO; +import com.cmvr.test.service.ShangdaoFileService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Service; + +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * 巡检仪表读数识别节点。 + * + *

输入参数见{@link InspectionMeterRecognizeConfigVO}。节点先将MinIO图片上传商道， + * 再调用固定智能体识别读数；识别成功后执行全部告警规则，命中多条规则时取最高级别。

+ * + *

节点输出包含value、unit、resultStatus和alarmLevel，同时通过内部字段 + * {@code _inspectionResult}交给巡检事件监听器持久化。该内部字段不需要前端配置。

+ */ +@Slf4j +@Service +@RequiredArgsConstructor +public class InspectionMeterRecognizeOperateService implements LLMOperateService +{ + private static final String API_KEY = "d9gqfkd4shheenomcfcg"; + private static final String RESULT_MARKER = "_inspectionResult"; + private static final Pattern NUMBER_PATTERN = + Pattern.compile("[-+]?(?:\\d+(?:\\.\\d+)?|\\.\\d+)"); + + private final ShangdaoFileService shangdaoFileService; + private final LLMAiAgentPlatformService llmAiAgentPlatformService; + + @Override + public boolean supports(ActionEnum action) + { + return ActionEnum.INSPECTION_METER_RECOGNIZE.equals(action); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) + { + JSONObject input = message.getInputParams() == null ? new JSONObject() : message.getInputParams(); + InspectionMeterRecognizeConfigVO config = input.to(InspectionMeterRecognizeConfigVO.class); + String imageUrl = StringUtils.trimToNull(config.getImageUrl()); + String resultName = StringUtils.defaultIfBlank(config.getResultName(), "仪表读数"); + String configuredUnit = StringUtils.trimToNull(config.getUnit()); + JSONArray ruleArray = input.getJSONArray("alarmRules"); + List rules = parseRules(config.getAlarmRules()); + + // 商道文件地址有效期较短，仅用于本次识别；长期证据仍保存原始MinIO地址。 + String shangdaoUrl = shangdaoFileService.uploadFromMinio(imageUrl); + JSONObject response = llmAiAgentPlatformService.query( + ActionEnum.INSPECTION_METER_RECOGNIZE.getAction(), shangdaoUrl, + API_KEY, Boolean.FALSE, new JSONObject()); + String rawResult = response == null ? null : response.getString("result"); + + Reading reading = parseReading(rawResult, configuredUnit); + JSONObject output = new JSONObject(); + output.put("imageUrl", imageUrl); + output.put("rawResult", rawResult); + output.put("resultName", resultName); + + JSONObject inspectionResult = new JSONObject(); + inspectionResult.put("resultType", "METER"); + inspectionResult.put("resultName", resultName); + inspectionResult.put("evidenceUrl", imageUrl); + inspectionResult.put("evidenceType", "IMAGE"); + inspectionResult.put("rawResult", rawResult); + inspectionResult.put("ruleSnapshot", ruleArray == null ? new JSONArray() : ruleArray); + + // 无法提取数值时保留原始响应并进入人工复核，不执行阈值规则，避免误报警。 + if (reading == null) + { + output.put("resultStatus", "RECOGNIZE_FAILED"); + inspectionResult.put("resultStatus", "RECOGNIZE_FAILED"); + inspectionResult.put("valueText", rawResult); + inspectionResult.put("alarmMessage", "商道返回结果无法解析为仪表读数"); + output.put(RESULT_MARKER, inspectionResult); + return TaskNodeExecuteResult.success(output); + } + + AlarmRule matchedRule = matchHighestRule(reading.value, rules); + String resultStatus = matchedRule == null ? "NORMAL" : "ABNORMAL"; + String unit = StringUtils.defaultIfBlank(configuredUnit, reading.unit); + output.put("value", reading.value); + output.put("unit", unit); + output.put("resultStatus", resultStatus); + output.put("alarmLevel", matchedRule == null ? null : matchedRule.level); + + inspectionResult.put("resultStatus", resultStatus); + inspectionResult.put("valueNumber", reading.value); + inspectionResult.put("valueText", reading.value.toPlainString()); + inspectionResult.put("unit", unit); + if (matchedRule != null) + { + inspectionResult.put("alarmLevel", matchedRule.level); + inspectionResult.put("alarmMessage", matchedRule.message); + inspectionResult.put("matchedRule", matchedRule.source); + } + output.put(RESULT_MARKER, inspectionResult); + return TaskNodeExecuteResult.success(output); + } + + /** + * 校验并转换前端配置的多级告警规则。 + */ + private List parseRules(List sourceRules) + { + List result = new ArrayList<>(); + if (sourceRules == null) + { + return result; + } + for (int i = 0; i < sourceRules.size(); i++) + { + InspectionAlarmRuleVO source = sourceRules.get(i); + if (source == null) + { + throw new GlobalException("alarmRules[" + i + "]格式错误"); + } + Integer level = source.getLevel(); + if (level == null || level < 1 || level > 3) + { + throw new GlobalException("alarmRules[" + i + "].level只能为1、2、3"); + } + String operator = StringUtils.upperCase(StringUtils.trimToEmpty(source.getOperator())); + validateOperator(operator, i); + + AlarmRule rule = new AlarmRule(); + rule.level = level; + rule.operator = operator; + rule.threshold = source.getThreshold(); + rule.minValue = source.getMinValue(); + rule.maxValue = source.getMaxValue(); + rule.includeMin = source.getIncludeMin() == null || source.getIncludeMin(); + rule.includeMax = source.getIncludeMax() == null || source.getIncludeMax(); + rule.message = StringUtils.defaultIfBlank(source.getMessage(), "仪表读数触发告警"); + rule.source = JSON.parseObject(JSON.toJSONString(source)); + validateThreshold(rule, i); + result.add(rule); + } + return result; + } + + private void validateOperator(String operator, int index) + { + if (!("GT".equals(operator) || "GTE".equals(operator) || "LT".equals(operator) + || "LTE".equals(operator) || "EQ".equals(operator) || "BETWEEN".equals(operator) + || "NOT_BETWEEN".equals(operator))) + { + throw new GlobalException("alarmRules[" + index + "].operator不支持: " + operator); + } + } + + private void validateThreshold(AlarmRule rule, int index) + { + boolean between = "BETWEEN".equals(rule.operator) || "NOT_BETWEEN".equals(rule.operator); + if (between) + { + if (rule.minValue == null || rule.maxValue == null || rule.minValue.compareTo(rule.maxValue) > 0) + { + throw new GlobalException("alarmRules[" + index + "]区间上下界配置错误"); + } + } + else if (rule.threshold == null) + { + throw new GlobalException("alarmRules[" + index + "].threshold不能为空"); + } + } + + /** 命中多条重叠规则时按level取最高严重级别，与前端数组顺序无关。 */ + private AlarmRule matchHighestRule(BigDecimal value, List rules) + { + return rules.stream() + .filter(rule -> rule.matches(value)) + .max(Comparator.comparingInt(rule -> rule.level)) + .orElse(null); + } + + /** + * 解析商道结果：优先读取标准JSON字段，兼容纯文本数字返回。 + */ + private Reading parseReading(String rawResult, String configuredUnit) + { + String text = StringUtils.trimToNull(rawResult); + if (text == null) + { + return null; + } + String cleaned = text.replace("```json", "").replace("```", "").trim(); + try + { + Object parsed = JSON.parse(cleaned); + if (parsed instanceof JSONObject) + { + JSONObject object = (JSONObject) parsed; + BigDecimal value = firstDecimal(object, "value", "reading", "meterValue", "result"); + if (value != null) + { + return new Reading(value, + StringUtils.defaultIfBlank(object.getString("unit"), configuredUnit)); + } + } + } + catch (Exception ignored) + { + log.debug("商道仪表结果不是标准JSON，尝试提取数字，rawResult={}", text); + } + + Matcher matcher = NUMBER_PATTERN.matcher(cleaned); + if (!matcher.find()) + { + return null; + } + try + { + return new Reading(new BigDecimal(matcher.group()), configuredUnit); + } + catch (NumberFormatException ex) + { + return null; + } + } + + private BigDecimal firstDecimal(JSONObject object, String... keys) + { + for (String key : keys) + { + Object value = object.get(key); + if (value instanceof Number || value instanceof String) + { + try + { + return new BigDecimal(String.valueOf(value)); + } + catch (NumberFormatException ignored) + { + // 继续尝试下一个标准字段。 + } + } + } + return null; + } + + private static final class Reading + { + private final BigDecimal value; + private final String unit; + + private Reading(BigDecimal value, String unit) + { + this.value = value; + this.unit = unit; + } + } + + private static final class AlarmRule + { + private int level; + private String operator; + private BigDecimal threshold; + private BigDecimal minValue; + private BigDecimal maxValue; + private boolean includeMin; + private boolean includeMax; + private String message; + private JSONObject source; + + private boolean matches(BigDecimal value) + { + int thresholdCompare = threshold == null ? 0 : value.compareTo(threshold); + switch (operator.toUpperCase(Locale.ROOT)) + { + case "GT": return thresholdCompare > 0; + case "GTE": return thresholdCompare >= 0; + case "LT": return thresholdCompare < 0; + case "LTE": return thresholdCompare <= 0; + case "EQ": return thresholdCompare == 0; + case "BETWEEN": return within(value); + case "NOT_BETWEEN": return !within(value); + default: return false; + } + } + + private boolean within(BigDecimal value) + { + int minCompare = value.compareTo(minValue); + int maxCompare = value.compareTo(maxValue); + boolean lowerMatched = includeMin ? minCompare >= 0 : minCompare > 0; + boolean upperMatched = includeMax ? maxCompare <= 0 : maxCompare < 0; + return lowerMatched && upperMatched; + } + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/FlowActionRequestVO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/FlowActionRequestVO.java index d9f9147..75bdbfb 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/FlowActionRequestVO.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/FlowActionRequestVO.java @@ -1,21 +1,28 @@ package com.cmvr.test.model.vo; import com.alibaba.fastjson2.JSONObject; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; import lombok.Data; /** * 工作流节点执行请求 */ @Data +@ApiModel(value = "FlowActionRequestVO", description = "工作流单节点试执行请求") public class FlowActionRequestVO { /** * 动作 */ + @ApiModelProperty(value = "动作编码。巡检新增动作：INSPECTION_METER_RECOGNIZE、INSPECTION_MANUAL_REVIEW_CREATE", + required = true, example = "INSPECTION_METER_RECOGNIZE") private String action; /** * 执行参数 */ + @ApiModelProperty(value = "节点参数。仪表识别参见InspectionMeterRecognizeConfigVO，人工判断参见InspectionManualReviewConfigVO", + required = true) private JSONObject payload; } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionAlarmRuleVO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionAlarmRuleVO.java new file mode 100644 index 0000000..c5840d5 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionAlarmRuleVO.java @@ -0,0 +1,39 @@ +package com.cmvr.test.model.vo.inspection; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.math.BigDecimal; + +/** 仪表读数的单条告警规则配置。 */ +@Data +@ApiModel(value = "InspectionAlarmRuleVO", description = "仪表读数多级告警规则") +public class InspectionAlarmRuleVO +{ + @ApiModelProperty(value = "告警级别：1提示、2警告、3严重", required = true, + allowableValues = "1,2,3", example = "2") + private Integer level; + + @ApiModelProperty(value = "比较运算符", required = true, + allowableValues = "GT,GTE,LT,LTE,EQ,BETWEEN,NOT_BETWEEN", example = "GTE") + private String operator; + + @ApiModelProperty(value = "GT/GTE/LT/LTE/EQ使用的比较阈值", example = "1.6") + private BigDecimal threshold; + + @ApiModelProperty(value = "BETWEEN/NOT_BETWEEN使用的区间下界", example = "1.6") + private BigDecimal minValue; + + @ApiModelProperty(value = "BETWEEN/NOT_BETWEEN使用的区间上界", example = "2.0") + private BigDecimal maxValue; + + @ApiModelProperty(value = "区间规则是否包含下界，默认true", example = "true") + private Boolean includeMin; + + @ApiModelProperty(value = "区间规则是否包含上界，默认true", example = "false") + private Boolean includeMax; + + @ApiModelProperty(value = "命中规则后的告警描述", example = "压力过高") + private String message; +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionManualReviewConfigVO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionManualReviewConfigVO.java new file mode 100644 index 0000000..fcb3382 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionManualReviewConfigVO.java @@ -0,0 +1,33 @@ +package com.cmvr.test.model.vo.inspection; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.util.List; + +/** 创建人工巡检复核任务的工作流节点参数。 */ +@Data +@ApiModel(value = "InspectionManualReviewConfigVO", + description = "INSPECTION_MANUAL_REVIEW_CREATE节点参数；imageUrl、videoUrl、mediaList至少提供一个") +public class InspectionManualReviewConfigVO +{ + @ApiModelProperty(value = "人工检查项名称，默认人工巡检", example = "设备外观检查") + private String resultName; + + @ApiModelProperty("单张MinIO图片地址，可引用上游拍照节点输出") + private String imageUrl; + + @ApiModelProperty("单个MinIO视频地址，可引用上游录像节点输出") + private String videoUrl; + + @ApiModelProperty("多张图片或视频；可与imageUrl、videoUrl同时使用") + private List mediaList; + + @ApiModelProperty(value = "供审核人员参考的判断标准", example = "检查设备是否破损、漏油") + private String reviewCriteria; + + @ApiModelProperty(value = "建议告警级别，仅作为审核参考；提交异常复核时仍需明确选择", + allowableValues = "1,2,3", example = "2") + private Integer defaultAlarmLevel; +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMediaVO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMediaVO.java new file mode 100644 index 0000000..923b4a8 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMediaVO.java @@ -0,0 +1,18 @@ +package com.cmvr.test.model.vo.inspection; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +/** 人工巡检节点的一条媒体证据。 */ +@Data +@ApiModel(value = "InspectionMediaVO", description = "人工巡检图片或视频参数") +public class InspectionMediaVO +{ + @ApiModelProperty(value = "媒体类型", required = true, + allowableValues = "IMAGE,VIDEO", example = "IMAGE") + private String mediaType; + + @ApiModelProperty(value = "MinIO永久媒体地址", required = true) + private String mediaUrl; +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMeterRecognizeConfigVO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMeterRecognizeConfigVO.java new file mode 100644 index 0000000..4a7a928 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/inspection/InspectionMeterRecognizeConfigVO.java @@ -0,0 +1,27 @@ +package com.cmvr.test.model.vo.inspection; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.util.List; + +/** 仪表读数识别工作流节点参数。 */ +@Data +@ApiModel(value = "InspectionMeterRecognizeConfigVO", + description = "INSPECTION_METER_RECOGNIZE节点参数") +public class InspectionMeterRecognizeConfigVO +{ + @ApiModelProperty(value = "前置拍照节点输出的MinIO图片地址", required = true, + example = "http://127.0.0.1:9000/cmvr/inspection/pressure.jpg") + private String imageUrl; + + @ApiModelProperty(value = "仪表或检查项名称，默认仪表读数", example = "1号压力表") + private String resultName; + + @ApiModelProperty(value = "仪表单位", example = "MPa") + private String unit; + + @ApiModelProperty("多级告警规则；为空时只识别和保存，不产生读数告警") + private List alarmRules; +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java index a7a61e8..05cfff8 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java @@ -18,6 +18,9 @@ import com.cmvr.edge.client.service.EdgeArmService; import com.cmvr.device.service.InspectionAlertListenService; import com.cmvr.llm.service.LLMAiAgentPlatformService; import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.operator.edge.EdgeManualInspectionOperateService; +import com.cmvr.test.flow.runtime.operator.llm.InspectionMeterRecognizeOperateService; import com.cmvr.test.model.vo.FlowActionRequestVO; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -36,6 +39,8 @@ public class FlowActionExecutorService { private final EdgeAgvService edgeAgvService; private final InspectionAlertListenService inspectionAlertListenService; private final EdgeArmService edgeArmService; + private final EdgeManualInspectionOperateService edgeManualInspectionOperateService; + private final InspectionMeterRecognizeOperateService inspectionMeterRecognizeOperateService; public String actionExecute(FlowActionRequestVO req) { @@ -67,6 +72,10 @@ public class FlowActionExecutorService { || ActionEnum.INSPECTION_ALERT_LISTEN_STOP.equals(action)) { return executeInspectionAlertAction(action, payload); } + if (ActionEnum.INSPECTION_MANUAL_REVIEW_CREATE.equals(action)) { + TaskNodeExecuteMessage message = buildSingleNodeMessage(action, payload); + return edgeManualInspectionOperateService.execute(message).getOutputParams().toJSONString(); + } if (StrUtil.isEmpty(terminalId) || StrUtil.isEmpty(deviceId)) { throw new GlobalException("EDGE 类型动作必须提供 terminalId 和 deviceId"); @@ -248,9 +257,19 @@ public class FlowActionExecutorService { return "LLM 高级音频生成 OK"; case AI_AGENT_PLATFORM: return llmAiAgentPlatformService.query(action.getAction(), req.getPayload().getJSONObject("config").getString("text"), req.getPayload().getJSONObject("config").getString("apiKey"), req.getPayload().getBoolean("invokeTts"), req.getPayload().getJSONObject("tts")).toString(); + case INSPECTION_METER_RECOGNIZE: + return inspectionMeterRecognizeOperateService.execute( + buildSingleNodeMessage(action, req.getPayload())).getOutputParams().toJSONString(); default: throw new UnsupportedOperationException("未实现的 LLM Action: " + action); } } + + private TaskNodeExecuteMessage buildSingleNodeMessage(ActionEnum action, JSONObject payload) { + TaskNodeExecuteMessage message = new TaskNodeExecuteMessage(); + message.setAction(action); + message.setInputParams(payload == null ? new JSONObject() : payload); + return message; + } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/ShangdaoFileService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ShangdaoFileService.java new file mode 100644 index 0000000..facf6f1 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ShangdaoFileService.java @@ -0,0 +1,121 @@ +package com.cmvr.test.service; + +import com.cmvr.common.config.properties.MinioProperties; +import com.cmvr.common.core.minio.MinioService; +import com.cmvr.common.exception.GlobalException; +import com.cmvr.llm.util.LargeModelFileUploadUtil; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Service; + +import java.io.File; +import java.nio.file.Files; +import java.net.URI; +import java.net.URLDecoder; + +/** + * MinIO文件转存商道服务，统一处理对象地址解析和大小限制。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class ShangdaoFileService +{ + private static final int MAX_FILE_BYTES = 20 * 1024 * 1024; + + private final MinioService minioService; + private final MinioProperties minioProperties; + + /** + * 下载指定MinIO对象并上传商道。 + * + * @param fileUrl 包含bucket的MinIO访问地址，或bucket内对象路径 + * @return 商道临时下载地址，仅用于随后调用大模型 + */ + public String uploadFromMinio(String fileUrl) + { + String objectName = resolveObjectName(fileUrl); + File temporaryFile = null; + try + { + temporaryFile = minioService.getFile(minioProperties.getBucketName(), objectName); + if (temporaryFile.length() > MAX_FILE_BYTES) + { + throw new GlobalException("巡检媒体文件不能超过20MB"); + } + byte[] bytes = Files.readAllBytes(temporaryFile.toPath()); + String shangdaoUrl = LargeModelFileUploadUtil.uploadFile(bytes); + if (StringUtils.isBlank(shangdaoUrl)) + { + throw new GlobalException("商道文件上传未返回文件地址"); + } + return shangdaoUrl; + } + catch (GlobalException ex) + { + throw ex; + } + catch (Exception ex) + { + log.error("MinIO文件上传商道失败，fileUrl={}，objectName={}", fileUrl, objectName, ex); + throw new GlobalException("巡检图片上传商道失败"); + } + finally + { + if (temporaryFile != null && temporaryFile.exists() && !temporaryFile.delete()) + { + log.warn("删除MinIO临时文件失败，path={}", temporaryFile.getAbsolutePath()); + } + } + } + + /** + * 从完整URL或对象路径中提取当前bucket内的objectName，并拒绝目录穿越路径。 + */ + private String resolveObjectName(String fileUrl) + { + String value = StringUtils.trimToNull(fileUrl); + if (value == null) + { + throw new GlobalException("imageUrl不能为空"); + } + + String bucketName = StringUtils.trimToEmpty(minioProperties.getBucketName()); + try + { + String path = value; + if (value.contains("://")) + { + path = new URI(value).getRawPath(); + } + path = URLDecoder.decode(path, "UTF-8").replace('\\', '/'); + String bucketPrefix = "/" + bucketName + "/"; + int bucketIndex = path.indexOf(bucketPrefix); + if (bucketIndex >= 0) + { + path = path.substring(bucketIndex + bucketPrefix.length()); + } + else + { + path = StringUtils.removeStart(path, "/"); + path = StringUtils.removeStart(path, bucketName + "/"); + } + String objectName = StringUtils.trimToNull(path); + if (objectName == null || objectName.contains("..")) + { + throw new GlobalException("imageUrl不是有效的MinIO对象地址"); + } + return objectName; + } + catch (GlobalException ex) + { + throw ex; + } + catch (Exception ex) + { + throw new GlobalException("无法解析MinIO文件地址: " + value); + } + } + +} diff --git a/sql/inspection_module.sql b/sql/inspection_module.sql index 4a6a268..e30f534 100644 --- a/sql/inspection_module.sql +++ b/sql/inspection_module.sql @@ -124,6 +124,10 @@ CREATE TABLE `inspection_alarm` ( `remark` varchar(500) DEFAULT NULL COMMENT '备注', `alarm_code` varchar(64) DEFAULT NULL COMMENT '告警编码', `task_instance_id` varchar(64) DEFAULT NULL COMMENT '任务执行实例ID', + `result_id` varchar(64) DEFAULT NULL COMMENT '关联巡检结果ID', + `alarm_source` varchar(20) DEFAULT NULL COMMENT '告警来源(PPE/METER/MANUAL)', + `item_id` varchar(64) DEFAULT NULL COMMENT '检测项ID', + `node_id` varchar(64) DEFAULT NULL COMMENT '工作流节点ID', `task_id` varchar(64) DEFAULT NULL COMMENT '任务ID', `task_name` varchar(100) DEFAULT NULL COMMENT '任务名称', `robot_id` varchar(64) DEFAULT NULL COMMENT '机器人ID', @@ -141,6 +145,7 @@ CREATE TABLE `inspection_alarm` ( `evidence_image` varchar(500) DEFAULT NULL COMMENT '图片证据', PRIMARY KEY (`id`), UNIQUE KEY `uk_alarm_code` (`alarm_code`), + UNIQUE KEY `uk_alarm_result_id` (`result_id`), KEY `idx_task_instance_id` (`task_instance_id`), KEY `idx_task_id` (`task_id`), KEY `idx_robot_id` (`robot_id`), @@ -177,3 +182,64 @@ CREATE TABLE `inspection_task_log` ( KEY `idx_log_time` (`log_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='巡检任务执行日志表'; +-- 6. 统一巡检结果表 +DROP TABLE IF EXISTS `inspection_result_media`; +DROP TABLE IF EXISTS `inspection_result`; +CREATE TABLE `inspection_result` ( + `id` varchar(32) NOT NULL COMMENT '主键ID', + `result_code` varchar(64) NOT NULL COMMENT '结果编码', + `dedupe_key` varchar(255) NOT NULL COMMENT '结果幂等键', + `task_instance_id` varchar(64) DEFAULT NULL COMMENT '巡检任务实例数据库ID', + `flow_instance_id` varchar(64) DEFAULT NULL COMMENT '工作流实例ID', + `task_id` varchar(64) DEFAULT NULL COMMENT '巡检任务ID', + `item_id` varchar(64) DEFAULT NULL COMMENT '检测项ID', + `node_id` varchar(64) DEFAULT NULL COMMENT '工作流节点ID', + `node_name` varchar(255) DEFAULT NULL COMMENT '工作流节点名称', + `iteration_path` varchar(255) DEFAULT NULL COMMENT '循环迭代路径JSON', + `result_type` varchar(20) NOT NULL COMMENT '结果类型(PPE/METER/MANUAL)', + `result_status` varchar(32) NOT NULL COMMENT 'PENDING/NORMAL/ABNORMAL/RECOGNIZE_FAILED', + `result_name` varchar(255) NOT NULL COMMENT '检查名称', + `value_number` decimal(20,8) DEFAULT NULL COMMENT '结构化数值', + `value_text` varchar(1000) DEFAULT NULL COMMENT '文本结果', + `unit` varchar(32) DEFAULT NULL COMMENT '单位', + `alarm_level` tinyint DEFAULT NULL COMMENT '告警级别(1提示 2警告 3严重)', + `alarm_message` varchar(1000) DEFAULT NULL COMMENT '告警描述', + `confidence` decimal(10,8) DEFAULT NULL COMMENT '模型置信度', + `source_event_id` varchar(128) DEFAULT NULL COMMENT '外部PPE事件ID', + `evidence_url` varchar(1000) DEFAULT NULL COMMENT '主要MinIO证据地址', + `evidence_type` varchar(20) DEFAULT NULL COMMENT '主要证据类型(IMAGE/VIDEO)', + `raw_result` mediumtext COMMENT '模型或边缘端原始返回', + `matched_rule_json` text COMMENT '最终命中的报警规则快照', + `rule_snapshot_json` mediumtext COMMENT '本次执行使用的完整规则快照', + `reviewer` varchar(64) DEFAULT NULL COMMENT '人工复核人', + `review_time` datetime DEFAULT NULL COMMENT '人工复核时间', + `review_remark` varchar(1000) DEFAULT NULL COMMENT '人工复核说明', + `review_version` int NOT NULL DEFAULT 0 COMMENT '人工复核乐观锁版本', + `alarm_id` varchar(64) DEFAULT NULL COMMENT '关联告警ID', + `occurred_time` datetime NOT NULL COMMENT '结果发生时间', + `create_by` varchar(64) DEFAULT '' COMMENT '创建者', + `create_time` datetime NOT NULL COMMENT '创建时间', + `update_by` varchar(64) DEFAULT '' COMMENT '更新者', + `update_time` datetime DEFAULT NULL COMMENT '更新时间', + `remark` varchar(500) DEFAULT NULL COMMENT '备注', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_inspection_result_code` (`result_code`), + UNIQUE KEY `uk_inspection_result_dedupe` (`dedupe_key`), + KEY `idx_result_task_instance` (`task_instance_id`), + KEY `idx_result_flow_instance` (`flow_instance_id`), + KEY `idx_result_status` (`result_status`), + KEY `idx_result_type` (`result_type`), + KEY `idx_result_occurred_time` (`occurred_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='统一巡检结果表'; + +-- 7. 巡检结果媒体证据表 +CREATE TABLE `inspection_result_media` ( + `id` varchar(32) NOT NULL COMMENT '主键ID', + `result_id` varchar(32) NOT NULL COMMENT '巡检结果ID', + `media_type` varchar(20) NOT NULL COMMENT 'IMAGE或VIDEO', + `media_url` varchar(1000) NOT NULL COMMENT 'MinIO媒体地址', + `sort_order` int NOT NULL DEFAULT 0 COMMENT '排序号', + `create_time` datetime NOT NULL COMMENT '创建时间', + PRIMARY KEY (`id`), + KEY `idx_result_media_result_id` (`result_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='巡检结果媒体证据表'; diff --git a/sql/inspection_result.sql b/sql/inspection_result.sql new file mode 100644 index 0000000..952a81f --- /dev/null +++ b/sql/inspection_result.sql @@ -0,0 +1,91 @@ +-- 统一巡检结果和三级告警增量脚本。 +-- 执行前请先备份数据库；本脚本不删除现有巡检数据。 + +CREATE TABLE IF NOT EXISTS `inspection_result` ( + `id` varchar(32) NOT NULL COMMENT '主键ID', + `result_code` varchar(64) NOT NULL COMMENT '结果编码', + `dedupe_key` varchar(255) NOT NULL COMMENT '结果幂等键', + `task_instance_id` varchar(64) DEFAULT NULL COMMENT '巡检任务实例数据库ID', + `flow_instance_id` varchar(64) DEFAULT NULL COMMENT '工作流实例ID', + `task_id` varchar(64) DEFAULT NULL COMMENT '巡检任务ID', + `item_id` varchar(64) DEFAULT NULL COMMENT '检测项ID', + `node_id` varchar(64) DEFAULT NULL COMMENT '工作流节点ID', + `node_name` varchar(255) DEFAULT NULL COMMENT '工作流节点名称', + `iteration_path` varchar(255) DEFAULT NULL COMMENT '循环迭代路径JSON', + `result_type` varchar(20) NOT NULL COMMENT 'PPE/METER/MANUAL', + `result_status` varchar(32) NOT NULL COMMENT 'PENDING/NORMAL/ABNORMAL/RECOGNIZE_FAILED', + `result_name` varchar(255) NOT NULL COMMENT '检查名称', + `value_number` decimal(20,8) DEFAULT NULL COMMENT '结构化数值', + `value_text` varchar(1000) DEFAULT NULL COMMENT '文本结果', + `unit` varchar(32) DEFAULT NULL COMMENT '单位', + `alarm_level` tinyint DEFAULT NULL COMMENT '告警级别(1提示 2警告 3严重)', + `alarm_message` varchar(1000) DEFAULT NULL COMMENT '告警描述', + `confidence` decimal(10,8) DEFAULT NULL COMMENT '模型置信度', + `source_event_id` varchar(128) DEFAULT NULL COMMENT '外部PPE事件ID', + `evidence_url` varchar(1000) DEFAULT NULL COMMENT '主要MinIO证据地址', + `evidence_type` varchar(20) DEFAULT NULL COMMENT '主要证据类型(IMAGE/VIDEO)', + `raw_result` mediumtext COMMENT '模型或边缘端原始返回', + `matched_rule_json` text COMMENT '最终命中的报警规则', + `rule_snapshot_json` mediumtext COMMENT '完整报警规则快照', + `reviewer` varchar(64) DEFAULT NULL COMMENT '人工复核人', + `review_time` datetime DEFAULT NULL COMMENT '人工复核时间', + `review_remark` varchar(1000) DEFAULT NULL COMMENT '人工复核说明', + `review_version` int NOT NULL DEFAULT 0 COMMENT '人工复核版本', + `alarm_id` varchar(64) DEFAULT NULL COMMENT '关联告警ID', + `occurred_time` datetime NOT NULL COMMENT '结果发生时间', + `create_by` varchar(64) DEFAULT '' COMMENT '创建者', + `create_time` datetime NOT NULL COMMENT '创建时间', + `update_by` varchar(64) DEFAULT '' COMMENT '更新者', + `update_time` datetime DEFAULT NULL COMMENT '更新时间', + `remark` varchar(500) DEFAULT NULL COMMENT '备注', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_inspection_result_code` (`result_code`), + UNIQUE KEY `uk_inspection_result_dedupe` (`dedupe_key`), + KEY `idx_result_task_instance` (`task_instance_id`), + KEY `idx_result_flow_instance` (`flow_instance_id`), + KEY `idx_result_status` (`result_status`), + KEY `idx_result_type` (`result_type`), + KEY `idx_result_occurred_time` (`occurred_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='统一巡检结果表'; + +CREATE TABLE IF NOT EXISTS `inspection_result_media` ( + `id` varchar(32) NOT NULL COMMENT '主键ID', + `result_id` varchar(32) NOT NULL COMMENT '巡检结果ID', + `media_type` varchar(20) NOT NULL COMMENT 'IMAGE或VIDEO', + `media_url` varchar(1000) NOT NULL COMMENT 'MinIO媒体地址', + `sort_order` int NOT NULL DEFAULT 0 COMMENT '排序号', + `create_time` datetime NOT NULL COMMENT '创建时间', + PRIMARY KEY (`id`), + KEY `idx_result_media_result_id` (`result_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='巡检结果媒体证据表'; + +-- 已存在inspection_result表时补充主要证据类型字段。 +SET @sql := IF((SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='inspection_result' AND column_name='evidence_type')=0, + 'ALTER TABLE inspection_result ADD COLUMN evidence_type varchar(20) DEFAULT NULL COMMENT ''主要证据类型(IMAGE/VIDEO)'' AFTER evidence_url', 'SELECT 1'); +PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +-- 回填历史结果，使用与主表evidence_url完全一致的媒体记录，避免取错非主要证据。 +UPDATE inspection_result r +JOIN inspection_result_media m ON m.result_id = r.id AND m.media_url = r.evidence_url +SET r.evidence_type = m.media_type +WHERE r.evidence_type IS NULL; + +SET @sql := IF((SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='inspection_alarm' AND column_name='result_id')=0, + 'ALTER TABLE inspection_alarm ADD COLUMN result_id varchar(64) DEFAULT NULL COMMENT ''关联巡检结果ID'' AFTER task_instance_id', 'SELECT 1'); +PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @sql := IF((SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='inspection_alarm' AND column_name='alarm_source')=0, + 'ALTER TABLE inspection_alarm ADD COLUMN alarm_source varchar(20) DEFAULT NULL COMMENT ''告警来源'' AFTER result_id', 'SELECT 1'); +PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @sql := IF((SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='inspection_alarm' AND column_name='item_id')=0, + 'ALTER TABLE inspection_alarm ADD COLUMN item_id varchar(64) DEFAULT NULL COMMENT ''检测项ID'' AFTER alarm_source', 'SELECT 1'); +PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @sql := IF((SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='inspection_alarm' AND column_name='node_id')=0, + 'ALTER TABLE inspection_alarm ADD COLUMN node_id varchar(64) DEFAULT NULL COMMENT ''工作流节点ID'' AFTER item_id', 'SELECT 1'); +PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @sql := IF((SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='inspection_alarm' AND index_name='uk_alarm_result_id')=0, + 'ALTER TABLE inspection_alarm ADD UNIQUE KEY uk_alarm_result_id (result_id)', 'SELECT 1'); +PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;