fix 设备数采

xiaoyi

2026-10-28 8eb62ee0efe77e264cd48b3f1fd835e9b8430318

fix 设备数采

 yudao-module-hrm-api/src/main/java/cn/iocoder/yudao/module/hrm/enums/ErrorCodeConstants.java

@@ -94,6 +94,15 @@
    ErrorCode ATTENDANCE_EXCEPTION_BPM_PROCESS_DEFINITION_NOT_EXISTS = new ErrorCode(1_030_004_015, "考勤异常审批流程不存在");
    ErrorCode ATTENDANCE_CLOCK_ALREADY_CLOCKED = new ErrorCode(1_030_004_016, "已打过卡，请勿重复打卡");
    ErrorCode ATTENDANCE_HOLIDAY_DATE_RANGE_INVALID = new ErrorCode(1_030_004_017, "结束日期【{}】不能早于开始日期【{}】，无法保存。请重新选择正确的日期区间");
    /**
     * 登录账号未绑定员工档案。
     *
     * 注意：不要复用通用的 EMPLOYEE_NOT_EXISTS（1_030_000_000）去传自定义文案 ——
     * 该错误码的消息是「员工不存在」，不含 {} 占位符，ServiceExceptionUtil 会打
     * 「参数过多」警告并丢弃传入的文案，用户只能看到「员工不存在」，
     * 拿不到「请联系管理员绑定员工档案」这句可操作的提示。
     */
    ErrorCode ATTENDANCE_EMPLOYEE_NOT_BOUND = new ErrorCode(1_030_004_018, "当前登录账号未关联员工档案，无法打卡。请联系管理员在【员工管理】中将该登录账号绑定到员工");
    ErrorCode ATTENDANCE_LOCATION_RULE_INVALID = new ErrorCode(1_030_004_050, "考勤规则地点配置不完整，请联系管理员");
    ErrorCode ATTENDANCE_LOCATION_REQUIRED = new ErrorCode(1_030_004_051, "请开启定位并提供经纬度、精度和坐标系后打卡");
    ErrorCode ATTENDANCE_LOCATION_COORDINATE_INVALID = new ErrorCode(1_030_004_052, "打卡经纬度超出有效范围");

 yudao-module-hrm/src/main/java/cn/iocoder/yudao/module/hrm/controller/admin/attendance/HrmAttendanceRecordController.java

@@ -46,7 +46,7 @@
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
import static cn.iocoder.yudao.framework.common.util.date.DateUtils.FORMAT_YEAR_MONTH_DAY;
import static cn.iocoder.yudao.module.hrm.enums.ErrorCodeConstants.EMPLOYEE_NOT_EXISTS;
import static cn.iocoder.yudao.module.hrm.enums.ErrorCodeConstants.ATTENDANCE_EMPLOYEE_NOT_BOUND;
import static cn.iocoder.yudao.module.hrm.enums.LogRecordConstants.*;

@Tag(name = "管理后台 - HRM 考勤记录")
@@ -73,8 +73,7 @@
        Long loginUserId = getLoginUserId();
        HrmEmployeeDO employee = employeeService.getEmployeeByUserId(loginUserId);
        if (employee == null) {
            throw exception(EMPLOYEE_NOT_EXISTS,
                    "当前登录账号未关联员工档案，无法打卡。请联系管理员在【员工管理】中将该登录账号绑定到员工。");
            throw exception(ATTENDANCE_EMPLOYEE_NOT_BOUND);
        }
        HrmAttendanceClockResult result = attendanceRecordService.clockIn(
                employee.getId(), LocalDateTime.now(), reqVO, legacyLocation);
@@ -90,8 +89,7 @@
        Long loginUserId = getLoginUserId();
        HrmEmployeeDO employee = employeeService.getEmployeeByUserId(loginUserId);
        if (employee == null) {
            throw exception(EMPLOYEE_NOT_EXISTS,
                    "当前登录账号未关联员工档案，无法打卡。请联系管理员在【员工管理】中将该登录账号绑定到员工。");
            throw exception(ATTENDANCE_EMPLOYEE_NOT_BOUND);
        }
        HrmAttendanceClockResult result = attendanceRecordService.clockOut(
                employee.getId(), LocalDateTime.now(), reqVO, legacyLocation);

 yudao-module-hrm/src/main/java/cn/iocoder/yudao/module/hrm/service/attendance/HrmAttendanceRecordServiceImpl.java

@@ -84,8 +84,7 @@
        // 1. 登录用户 -> 员工档案
        HrmEmployeeDO employee = employeeService.getEmployeeByUserId(loginUserId);
        if (employee == null) {
            throw exception(EMPLOYEE_NOT_EXISTS,
                    "当前登录账号未关联员工档案，无法打卡。请联系管理员在【员工管理】中将该登录账号绑定到员工。");
            throw exception(ATTENDANCE_EMPLOYEE_NOT_BOUND);
        }
        HrmAttendanceMyStatusRespVO respVO = new HrmAttendanceMyStatusRespVO();
        respVO.setUserId(loginUserId);

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/controller/admin/collect/vo/MesCollectDeviceRespVO.java

@@ -52,5 +52,14 @@
    private String workOrderCode;
    @Schema(description = "未消除报警数量", example = "1")
    private Long activeAlarmCount;
    /**
     * 最近一次采集的完整数据 JSON。
     *
     * <p>设备网关上报的是中文键的瞬时值（如 {"计米":268.35,"线速度":0.0}），
     * 与注塑机协议的 STS/OPM/OT 结构不同，结构化字段（sts/opm/ot/temps）会是空的，
     * 前端展示「当前数据」需要这个原始 JSON。</p>
     */
    @Schema(description = "最近采集数据 JSON", example = "{\"计米\":268.35}")
    private String dataJson;

}

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/enums/collect/MesCollectMsgTypeEnum.java

@@ -17,6 +17,8 @@
    TECH("tech", "工艺参数"),
    OP_LOG("opLog", "参数变更日志"),
    ALARM("alarm", "报警"),
    CONNECT("connect", "设备上线"),
    DISCONNECT("disconnect", "设备离线"),
    UNKNOWN("unknown", "未知类型");

    private final String type;

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/config/MesCollectProperties.java

@@ -14,6 +14,15 @@
@Data
public class MesCollectProperties {

    /**
     * 报文格式：注塑机 /YLCY/IMM 协议（报文带 topic 字段，主题末段为报文类型）
     */
    public static final String FORMAT_YLCY = "ylcy";
    /**
     * 报文格式：设备网关协议（主题固定 gateway/xxx，报文首层 key 为设备名，值为属性字典）
     */
    public static final String FORMAT_GATEWAY = "gateway";

    private Mqtt mqtt = new Mqtt();
    private Retention retention = new Retention();

@@ -35,9 +44,17 @@
         */
        private String clientIdPrefix = "mes-collect";
        /**
         * 主题前缀，例如 /YLCY/IMM
         * 主题前缀，例如 /YLCY/IMM、gateway
         */
        private String topicPrefix = "/YLCY/IMM";
        /**
         * 报文格式，见 {@link MesCollectProperties#FORMAT_YLCY}、{@link MesCollectProperties#FORMAT_GATEWAY}
         */
        private String format = FORMAT_YLCY;
        /**
         * 显式指定订阅主题，多个用逗号分隔；为空时按 format + topicPrefix 自动推导
         */
        private String topics;
        private Integer qos = 0;
        private Integer keepAlive = 65;
        private Integer connectionTimeout = 10;
@@ -46,6 +63,13 @@
         */
        private String alarmTopicSuffix = "wm";

        /**
         * 是否为设备网关格式报文
         */
        public boolean isGatewayFormat() {
            return FORMAT_GATEWAY.equalsIgnoreCase(format == null ? "" : format.trim());
        }

    }

    @Data

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/framework/collect/mqtt/MesCollectMqttClient.java

@@ -12,6 +12,9 @@

import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -114,15 +117,34 @@

    private void subscribe() {
        MesCollectProperties.Mqtt mqtt = properties.getMqtt();
        String topic = normalizeTopicPrefix(mqtt.getTopicPrefix()) + "+/+";
        try {
            client.subscribe(topic, mqtt.getQos());
            log.info("[subscribe][数采 MQTT 订阅成功：{}]", topic);
        } catch (Exception e) {
            log.error("[subscribe][数采 MQTT 订阅失败：{}]", topic, e);
        for (String topic : resolveSubscribeTopics(mqtt)) {
            try {
                client.subscribe(topic, mqtt.getQos());
                log.info("[subscribe][数采 MQTT 订阅成功：{}]", topic);
            } catch (Exception e) {
                log.error("[subscribe][数采 MQTT 订阅失败：{}]", topic, e);
            }
        }
    }

    /**
     * 解析需要订阅的主题。
     *
     * <p>显式配置 topics 优先；网关格式订阅整个前缀（gateway/#），
     * 因为网关除 attributes 外还会发 connect / disconnect，后续也可能新增主题；
     * 其余格式沿用 /YLCY/IMM/+/+ 两级通配。</p>
     */
    private List<String> resolveSubscribeTopics(MesCollectProperties.Mqtt mqtt) {
        if (StrUtil.isNotBlank(mqtt.getTopics())) {
            return Arrays.stream(mqtt.getTopics().split(","))
                    .map(String::trim)
                    .filter(StrUtil::isNotBlank)
                    .toList();
        }
        String prefix = normalizeTopicPrefix(mqtt.getTopicPrefix());
        return Collections.singletonList(mqtt.isGatewayFormat() ? prefix + "#" : prefix + "+/+");
    }

    private String normalizeTopicPrefix(String prefix) {
        if (StrUtil.isBlank(prefix)) {
            return "";

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectDataParser.java

@@ -201,7 +201,32 @@
        return parts.length >= 2 ? parts[parts.length - 2] : null;
    }

    private String extractTopicSuffix(String mqttTopic) {
    /**
     * 网关格式报文拆包：报文首层 key 是设备名，value 是该设备的属性字典。
     *
     * <p>例如 {"630框绞机2":{"计米":268.3499}} 拆成 {630框绞机2 -> {"计米":268.3499}}；
     * 连接事件 {"device":"大拉"} 这类扁平报文没有嵌套对象，返回空 Map 由调用方另行处理。</p>
     */
    public Map<String, JsonNode> splitGatewayDevices(JsonNode root) {
        Map<String, JsonNode> devices = new LinkedHashMap<>();
        if (root == null || !root.isObject()) {
            return devices;
        }
        root.properties().forEach(entry -> {
            String deviceCode = StrUtil.trim(entry.getKey());
            JsonNode data = entry.getValue();
            if (StrUtil.isBlank(deviceCode) || data == null || !data.isObject()) {
                return;
            }
            devices.put(deviceCode, data);
        });
        return devices;
    }

    /**
     * 取主题末段作为报文类型，如 gateway/attributes -> attributes、gateway/disconnect -> disconnect
     */
    public String extractTopicSuffix(String mqttTopic) {
        if (StrUtil.isBlank(mqttTopic)) {
            return null;
        }

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectMessageDispatcher.java

@@ -13,6 +13,7 @@
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@@ -26,6 +27,11 @@
@Component
@Slf4j
public class MesCollectMessageDispatcher {

    /**
     * 网关格式里承载设备属性快照的主题末段
     */
    private static final String ATTR_TOPIC_SUFFIX = "attributes";

    @Resource
    private MesCollectProperties properties;
@@ -61,6 +67,11 @@
    }

    private void process(String topic, String payload) {
        // 网关格式报文结构完全不同（主题固定、设备以报文首层 key 区分），单独处理
        if (properties.getMqtt().isGatewayFormat()) {
            processGateway(topic, payload);
            return;
        }
        JsonNode root;
        try {
            root = parser.parsePayload(payload);
@@ -111,6 +122,70 @@
        }
    }

    /**
     * 网关格式报文处理。
     *
     * <p>attributes：报文首层 key 是设备名、value 是属性字典，按设备逐台落实时数据；
     * connect / disconnect：连接事件，只留痕不落实时数据，便于排查设备离线。</p>
     */
    private void processGateway(String topic, String payload) {
        JsonNode root;
        try {
            root = parser.parsePayload(payload);
        } catch (Exception e) {
            log.warn("[processGateway][主题({}) 报文不是合法 JSON：{}]", topic, StrUtil.maxLength(payload, 200));
            dataService.markRawMessageFailed(
                    dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN,
                            LocalDateTime.now(), payload),
                    "报文不是合法 JSON：" + e.getMessage());
            return;
        }

        // 连接事件：{"device":"大拉"}，没有属性字典
        if (!ATTR_TOPIC_SUFFIX.equalsIgnoreCase(parser.extractTopicSuffix(topic))) {
            String deviceCode = StrUtil.trim(parser.getText(root, "device"));
            MesDvMachineryDO machinery = StrUtil.isBlank(deviceCode) ? null
                    : machineryMapper.selectByCollectDeviceCode(deviceCode);
            dataService.saveRawMessage(deviceCode, machinery == null ? null : machinery.getId(), topic,
                    MesCollectMsgTypeEnum.getByType(parser.extractTopicSuffix(topic)),
                    LocalDateTime.now(), payload);
            return;
        }

        Map<String, JsonNode> devices = parser.splitGatewayDevices(root);
        if (devices.isEmpty()) {
            log.debug("[processGateway][主题({}) 报文里没有设备属性：{}]", topic, StrUtil.maxLength(payload, 200));
            dataService.saveRawMessage(null, null, topic, MesCollectMsgTypeEnum.UNKNOWN, LocalDateTime.now(), payload);
            return;
        }
        // 网关报文不带时间戳，统一按接收时间落库
        LocalDateTime eventTime = parser.parseEventTime(root);
        devices.forEach((deviceCode, data) -> {
            MesDvMachineryDO machinery = machineryMapper.selectByCollectDeviceCode(deviceCode);
            if (machinery == null) {
                warnUnboundDevice(deviceCode);
            }
            Long machineryId = machinery == null ? null : machinery.getId();
            Long taskId = null;
            Long workOrderId = null;
            if (machinery != null && machinery.getWorkstationId() != null) {
                MesProTaskDO task = taskResolver.resolve(machinery.getWorkstationId());
                if (task != null) {
                    taskId = task.getId();
                    workOrderId = task.getWorkOrderId();
                }
            }
            Long rawId = dataService.saveRawMessage(deviceCode, machineryId, topic,
                    MesCollectMsgTypeEnum.REALTIME, eventTime, payload);
            try {
                dataService.saveRealtime(deviceCode, machineryId, taskId, workOrderId, eventTime, data);
            } catch (Exception e) {
                dataService.markRawMessageFailed(rawId, e.getMessage());
                log.error("[processGateway][主题({}) 设备({}) 处理失败]", topic, deviceCode, e);
            }
        });
    }

    private void warnUnboundDevice(String deviceCode) {
        long now = System.currentTimeMillis();
        Long last = unboundWarnCache.get(deviceCode);

 yudao-module-mes/src/main/java/cn/iocoder/yudao/module/mes/service/collect/MesCollectQueryService.java

@@ -92,6 +92,8 @@
                vo.setOpmName(MesCollectOperateModeEnum.getNameByMode(realtime.getOpm()));
                vo.setOt(realtime.getOt());
                vo.setTemps(realtime.getTemps());
                // 网关格式报文的结构化字段为空，前端「当前数据」列直接展示原始 JSON
                vo.setDataJson(realtime.getDataJson());
            } else {
                vo.setOnline(false);
            }

 yudao-server/src/main/resources/application-chdl.yaml

@@ -235,3 +235,22 @@
    username: root # 备份账号，留空使用数据源账号
    password: 'changhui@123456..' # 备份密码，留空使用数据源密码
    command: # mysqldump 命令，留空自动检测部署环境（本机 mysqldump 或 Docker MySQL 容器）；可手动指定，如：docker exec mysql8 mysqldump

  mes:
    collect:
      mqtt: # 长辉线缆设备网关数采：主题固定 gateway/attributes，报文首层 key 为设备名，值为属性字典（如 {"630框绞机2":{"计米":268.3499}}）
        enabled: true # 关掉后不再连接 Broker，仅保留历史数据查询
        url: tcp://101.245.92.159:1888
        username: admin
        password: admin
        client-id-prefix: mes-collect-chdl
        format: gateway # 报文格式：ylcy=注塑机 /YLCY/IMM 协议；gateway=设备网关协议（首层 key 为设备名，值为属性字典）
        topic-prefix: gateway # 网关主题前缀，网关格式下订阅 gateway/#
        qos: 0
        keep-alive: 65
        connection-timeout: 10
        alarm-topic-suffix: wm # 报文 topic 字段或主题末段命中该后缀即按报警处理
      retention:
        raw-days: 7 # 原始报文保留天数
        realtime-days: 30 # 实时状态历史保留天数
        batch-size: 50000 # 单次清理条数