app功能合并

cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/controller/task/DataTaskController.java

@@ -0,0 +1,51 @@
+package com.njcn.cssystem.controller.task;
+
+
+import com.njcn.common.pojo.annotation.OperateInfo;
+import com.njcn.common.pojo.enums.common.LogEnum;
+import com.njcn.common.pojo.enums.response.CommonResponseEnum;
+import com.njcn.common.pojo.response.HttpResult;
+import com.njcn.common.utils.HttpResultUtil;
+import com.njcn.cssystem.service.IDataTaskService;
+import com.njcn.web.controller.BaseController;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xy
+ */
+@Slf4j
+@RestController
+@RequestMapping("/dataTask")
+@Api(tags = "数据预处理")
+@AllArgsConstructor
+public class DataTaskController extends BaseController {
+
+    private final IDataTaskService dataTaskService;
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/channelHarmonic")
+    @ApiOperation("稳态数据")
+    public HttpResult<Boolean> channelHarmonic(@RequestParam("time") String time){
+        String methodDescribe = getMethodDescribe("channelHarmonic");
+        dataTaskService.channelHarmonic(time);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/runAlarm")
+    @ApiOperation("运行告警")
+    public HttpResult<Boolean> channelRunAlarm(@RequestParam("time") String time){
+        String methodDescribe = getMethodDescribe("channelRunAlarm");
+        dataTaskService.channelRunAlarm(time);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
+
+
+
+}

cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/IDataTaskService.java

@@ -0,0 +1,26 @@
+package com.njcn.cssystem.service;
+
+/**
+ * @author xy
+ */
+public interface IDataTaskService {
+
+    /**
+     * 稳态事件算法
+     * 1、计算所有监测点越限的次数 查询limit_rate表，按监测点、日期生成单独的一条数据；
+     * 2、根据监测点获取到设备，根据设备再获取哪些用户有这个设备的权限，然后将用户id收集起来。(主用户、子用户、管理员id集合)
+     * 3、将cs_harmonic生成的记录，存储到cs_event_user表中，标记为未读
+     * @param time
+     */
+    void channelHarmonic(String time);
+
+    /**
+     运行告警算法
+     1、每天生成一条记录，查询所有的设备，获取设备中断数据、告警数据；
+     2、根据设备获取哪些用户有这个设备的权限，然后将用户id收集起来。(主用户、子用户、管理员id集合)；
+     3、将cs_alarm生成的记录，存储到cs_event_user表中，标记为未读
+     * @param time
+     */
+    void channelRunAlarm(String time);
+
+}

cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/impl/DataTaskServiceImpl.java

@@ -0,0 +1,515 @@
+package com.njcn.cssystem.service.impl;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.date.DatePattern;
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.IdUtil;
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.core.util.ReflectUtil;
+import cn.hutool.core.util.StrUtil;
+import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.njcn.csdevice.api.CsDeviceUserFeignClient;
+import com.njcn.csdevice.api.CsLineFeignClient;
+import com.njcn.csdevice.pojo.dto.CsLineDTO;
+import com.njcn.csdevice.pojo.param.UserDevParam;
+import com.njcn.csdevice.pojo.po.CsDeviceUserPO;
+import com.njcn.csdevice.pojo.po.CsLinePO;
+import com.njcn.csharmonic.api.*;
+import com.njcn.csharmonic.param.CsEventUserQueryParam;
+import com.njcn.csharmonic.pojo.param.RStatLimitQueryParam;
+import com.njcn.csharmonic.pojo.po.*;
+import com.njcn.cssystem.service.IDataTaskService;
+import com.njcn.influx.imapper.PqsCommunicateMapper;
+import com.njcn.influx.pojo.po.PqsCommunicate;
+import com.njcn.influx.query.InfluxQueryWrapper;
+import com.njcn.system.api.DictTreeFeignClient;
+import com.njcn.system.enums.DicDataEnum;
+import com.njcn.system.enums.DicTreeEnum;
+import com.njcn.system.pojo.vo.DictTreeVO;
+import com.njcn.user.api.AppUserFeignClient;
+import com.njcn.user.pojo.po.User;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.Resource;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+
+/**
+ * @author xy
+ */
+@Service
+@AllArgsConstructor
+@Slf4j
+public class DataTaskServiceImpl implements IDataTaskService {
+
+    private final CsLineFeignClient csLineFeignClient;
+    private final CsDeviceUserFeignClient csDeviceUserFeignClient;
+    private final AppUserFeignClient appUserFeignClient;
+    private final RStatLimitRateDFeignClient rStatLimitRateDClient;
+    private final EventUserFeignClient eventUserFeignClient;
+    private final CsHarmonicFeignClient csHarmonicFeignClient;
+    @Resource
+    private PqsCommunicateMapper pqsCommunicateMapper;
+    private final EventFeignClient eventFeignClient;
+    private final CsAlarmFeignClient csAlarmFeignClient;
+    private final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault());
+    private final DictTreeFeignClient dictTreeFeignClient;
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    //fixme 这边事务没生效，可能后面需要研究下分布式事务
+    public void channelHarmonic(String time) {
+        List<CsHarmonic> list1 = new ArrayList<>();
+        List<CsEventUserPO> list2 = new ArrayList<>();
+        //获取所有监测点
+        List<CsLineDTO> allLine = csLineFeignClient.getAllLineDetail().getData();
+        if (CollUtil.isNotEmpty(allLine)) {
+            //过滤物联、在线监测的点
+            DictTreeVO portable = dictTreeFeignClient.queryByCode(DicDataEnum.PORTABLE.getCode()).getData();
+            if (ObjectUtil.isNotNull(portable)) {
+                allLine = allLine.stream().filter(item -> !item.getDeviceType().equals(portable.getId())).collect(Collectors.toList());
+            }
+            //监测点id集合
+            List<String> lineList = allLine.stream().map(CsLineDTO::getLineId).distinct().collect(Collectors.toList());
+            //设备id集合
+            List<String> devList = allLine.stream().map(CsLineDTO::getDeviceId).distinct().collect(Collectors.toList());
+            //获取监测点和设备关系
+            Map<String,String> lineDevMap = allLine.stream().collect(Collectors.toMap(CsLineDTO::getLineId,CsLineDTO::getDeviceId));
+
+            //获取设备和用户关系
+            UserDevParam param = new UserDevParam();
+            param.setList(devList);
+            List<CsDeviceUserPO> userList = csDeviceUserFeignClient.getList(param).getData();
+            //根据设备id分组 组是subUserId的集合
+            Map<String, List<String>> userDevMap = userList.stream().collect(Collectors.groupingBy(
+                    CsDeviceUserPO::getDeviceId,
+                    Collectors.mapping(
+                            CsDeviceUserPO::getSubUserId,
+                            Collectors.toList()
+                    )
+            ));
+            //获取管理员信息
+            List<User> adminList = appUserFeignClient.getAdminInfo().getData();
+            //获取用户id集合
+            List<String> userIdList = adminList.stream().map(User::getId).collect(Collectors.toList());
+
+            //获取越限数据
+            String date;
+            if (StringUtils.isNotBlank(time)) {
+                date = time;
+            } else {
+                date = DateUtil.yesterday().toString(DatePattern.NORM_DATE_PATTERN);
+            }
+            RStatLimitQueryParam rStatLimitQueryParam = RStatLimitQueryParam.builder().ids(lineList).date(date).endDate(date).build();
+            List<RStatLimitRateDPO> limitRates =  rStatLimitRateDClient.monitorIdsGetLimitRateInfo(rStatLimitQueryParam).getData();
+
+            if (CollUtil.isNotEmpty(limitRates)) {
+                limitRates.forEach(item->{
+                    //新增cs_harmonic数据
+                    String id = IdUtil.fastSimpleUUID();
+                    CsHarmonic csharmonic = new CsHarmonic();
+                    csharmonic.setId(id);
+                    csharmonic.setLineId(item.getLineId());
+                    csharmonic.setTime(LocalDate.parse(date));
+                    String tag = buildOverlimitTag(item);
+                    if (StringUtils.isNotBlank(tag)) {
+                        csharmonic.setTag(buildOverlimitTag(item));
+                        list1.add(csharmonic);
+
+                        //根据监测点id获取设备
+                        String deviceId = lineDevMap.get(item.getLineId());
+                        //根据设备获取用户
+                        List<String> userIds = userDevMap.get(deviceId);
+                        //添加管理员用户
+                        List<String> result = Stream.concat(userIdList.stream(), userIds.stream())
+                                .distinct()
+                                .collect(Collectors.toList());
+                        result.forEach(userId->{
+                            //新增cs_event_user数据
+                            CsEventUserPO po = new CsEventUserPO();
+                            po.setUserId(userId);
+                            po.setEventId(id);
+                            po.setStatus(0);
+                            list2.add(po);
+                        });
+                    }
+                });
+            }
+        }
+        if (CollUtil.isNotEmpty(list1)) {
+            //先删除 再录入
+            List<CsHarmonic> oldList = csHarmonicFeignClient.queryListByTime(time).getData();
+            if (CollUtil.isNotEmpty(oldList)) {
+                List<String> ids = oldList.stream().map(CsHarmonic::getId).collect(Collectors.toList());
+                csHarmonicFeignClient.deleteListByTime(time);
+                eventUserFeignClient.deleteByIds(ids);
+            }
+            csHarmonicFeignClient.addList(list1);
+            eventUserFeignClient.addUserEventList(list2);
+        }
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    //fixme 这边事务没生效，可能后面需要研究下分布式事务
+    public void channelRunAlarm(String time) {
+        CsAlarm alarm = new CsAlarm();
+        List<CsEventUserPO> list2 = new ArrayList<>();
+        //获取所有监测点
+        List<CsLineDTO> allLine = csLineFeignClient.getAllLineDetail().getData();
+        if (CollUtil.isNotEmpty(allLine)) {
+            //过滤物联、在线监测的点
+            DictTreeVO portable = dictTreeFeignClient.queryByCode(DicDataEnum.PORTABLE.getCode()).getData();
+            if (ObjectUtil.isNotNull(portable)) {
+                allLine = allLine.stream().filter(item -> !item.getDeviceType().equals(portable.getId())).collect(Collectors.toList());
+            }
+            //设备id集合
+            List<String> devList = allLine.stream().map(CsLineDTO::getDeviceId).distinct().collect(Collectors.toList());
+
+            //获取设备和用户关系
+            UserDevParam param = new UserDevParam();
+            param.setList(devList);
+            List<CsDeviceUserPO> userList = csDeviceUserFeignClient.getList(param).getData();
+            //根据设备id分组 组是subUserId的集合
+            Map<String, List<String>> userDevMap = userList.stream().collect(Collectors.groupingBy(
+                    CsDeviceUserPO::getDeviceId,
+                    Collectors.mapping(
+                            CsDeviceUserPO::getSubUserId,
+                            Collectors.toList()
+                    )
+            ));
+            //获取管理员信息
+            List<User> adminList = appUserFeignClient.getAdminInfo().getData();
+            //获取用户id集合
+            List<String> userIdList = adminList.stream().map(User::getId).collect(Collectors.toList());
+
+            //获取通讯数据
+            String date;
+            if (StringUtils.isNotBlank(time)) {
+                date = time;
+            } else {
+                date = DateUtil.yesterday().toString(DatePattern.NORM_DATE_PATTERN);
+            }
+            Map<String, List<String>> map = getDevCommunicate(devList, date + " 00:00:00", date+ " 23:59:59");
+
+            //获取告警数据
+            CsEventUserQueryParam queryParam = new CsEventUserQueryParam();
+            queryParam.setTarget(devList);
+            queryParam.setStartTime(date + " 00:00:00");
+            queryParam.setEndTime(date + " 23:59:59");
+            List<CsEventPO> alarmList = eventFeignClient.getDevAlarmList(queryParam).getData();
+            Map<String, List<CsEventPO>> alarmMap = alarmList.stream().collect(Collectors.groupingBy(CsEventPO::getDeviceId));
+
+            List<String> l0 = new ArrayList<>();
+            List<List<String>> l1 = new ArrayList<>();
+            List<List<String>> l2 = new ArrayList<>();
+            //step1:构造告警数据
+            String id = IdUtil.fastSimpleUUID();
+            alarm.setId(id);
+            alarm.setTime(LocalDate.parse(date));
+
+            devList.forEach(item->{
+                //写入中断数据
+                List<String> data1 = map.get(item);
+                if (CollUtil.isEmpty(data1)) {
+                    //获取最新的一条数据
+                    List<PqsCommunicate> latest = getRawDataLatest(item);
+                    if (CollUtil.isNotEmpty(latest)) {
+                        PqsCommunicate one = latest.get(0);
+                        if (one.getType() == 0) {
+                            data1 = Collections.singletonList(date + " 00:00:00" + " ~ " + date + " 23:59:59");
+                        }
+                    }
+                }
+                //写入告警数据
+                List<CsEventPO> data2 = alarmMap.get(item);
+                List<String> alarmIds = null;
+                if (CollUtil.isNotEmpty(data2)) {
+                    alarmIds = data2.stream().map(CsEventPO::getId).collect(Collectors.toList());
+                }
+
+                //只要有一个不为空，就需要统计
+                if (CollUtil.isNotEmpty(data1) || CollUtil.isNotEmpty(alarmIds)) {
+                    l0.add(item);
+                    l1.add(data1 != null ? Collections.singletonList(String.join("，", data1)) : null);
+                    l2.add(alarmIds);
+                }
+            });
+            alarm.setDevList(String.join(",", l0));
+            alarm.setInterruptEvent(new Gson().toJson(l1));
+            alarm.setAlarmEvent(new Gson().toJson(l2));
+            //step2:用户告警通知
+            if (CollUtil.isNotEmpty(l0)) {
+                Set<String> userIds = new HashSet<>();
+                l0.forEach(item->{
+                    List<String> ids = userDevMap.get(item);
+                    if (CollUtil.isNotEmpty(ids)) {
+                        userIds.addAll(ids);
+                    }
+                });
+                //添加管理员用户
+                List<String> result = Stream.concat(userIdList.stream(), userIds.stream())
+                        .distinct()
+                        .collect(Collectors.toList());
+                result.forEach(userId->{
+                    //新增cs_event_user数据
+                    CsEventUserPO po = new CsEventUserPO();
+                    po.setUserId(userId);
+                    po.setEventId(id);
+                    po.setStatus(0);
+                    list2.add(po);
+                });
+            }
+        }
+        if (ObjectUtil.isNotNull(alarm)) {
+            //先删除 再录入
+            List<CsAlarm> oldList = csAlarmFeignClient.queryListByTime(time).getData();
+            if (CollUtil.isNotEmpty(oldList)) {
+                List<String> ids = oldList.stream().map(CsAlarm::getId).collect(Collectors.toList());
+                csAlarmFeignClient.deleteListByTime(time);
+                eventUserFeignClient.deleteByIds(ids);
+            }
+            csAlarmFeignClient.addList(Collections.singletonList(alarm));
+            eventUserFeignClient.addUserEventList(list2);
+        }
+    }
+
+    //获取设备最新的一条数据
+    public List<PqsCommunicate> getRawDataLatest(String devId) {
+        InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);
+        influxQueryWrapper.regular(PqsCommunicate::getDevId, devId)
+                .select(PqsCommunicate::getTime)
+                .select(PqsCommunicate::getDevId)
+                .select(PqsCommunicate::getDescription)
+                .select(PqsCommunicate::getType)
+                .timeDesc()
+                .limit(1);
+        return pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper);
+    }
+
+
+    //获取设备的通讯记录
+    public Map<String, List<String>> getDevCommunicate(List<String> devList, String startTime, String endTime) {
+        InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);
+        influxQueryWrapper.regular(PqsCommunicate::getDevId, devList)
+                .select(PqsCommunicate::getTime)
+                .select(PqsCommunicate::getDevId)
+                .select(PqsCommunicate::getType)
+                .between(PqsCommunicate::getTime, startTime, endTime);
+        List<PqsCommunicate> pqsCommunicates = pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper);
+        Map<String, List<String>> result = new HashMap<>();
+
+        if (CollUtil.isNotEmpty(pqsCommunicates)) {
+            //这边需求是根据 pqsCommunicates 的数据 按设备来统计设备的掉线时间段，type = 1 是上线 type = 0 是掉线。因为这是按日来统计的，所以需要考虑 0 点 和 24 点的时间段
+            //还有个情况就是 如果只有 1 条数据 需要根据是掉线还是上线，来统计掉线时间段；如果是多条数据需要考虑 0 点 和 24 点的时间段
+
+            // 按设备 ID 分组
+            Map<String, List<PqsCommunicate>> deviceCommMap = pqsCommunicates.stream()
+                    .collect(Collectors.groupingBy(PqsCommunicate::getDevId));
+
+            deviceCommMap.forEach((devId, commList) -> {
+                // 按时间排序
+                commList.sort(Comparator.comparing(PqsCommunicate::getTime));
+
+                List<String> offlinePeriods = new ArrayList<>();
+                String offlineStartTime = null;
+                String offlineEndTime = null;
+
+                if (commList.size() == 1) {
+                    // 只有一条数据的情况
+                    PqsCommunicate singleRecord = commList.get(0);
+                    if (singleRecord.getType() == 0) {
+                        // 掉线：从掉线时间到 24 点
+                        offlineStartTime = DATE_TIME_FORMATTER.format(singleRecord.getTime());
+                        offlineEndTime = DATE_TIME_FORMATTER.format(singleRecord.getTime()).substring(0, 10) + " 23:59:59";
+                    } else {
+                        // 上线：从 0 点到上线时间
+                        offlineStartTime = DATE_TIME_FORMATTER.format(singleRecord.getTime()).substring(0, 10) + " 00:00:00";
+                        offlineEndTime = DATE_TIME_FORMATTER.format(singleRecord.getTime());
+                    }
+                    offlinePeriods.add(offlineStartTime + " ~ " + offlineEndTime);
+                } else {
+                    // 多条数据的情况
+                    for (int i = 0; i < commList.size(); i++) {
+                        PqsCommunicate current = commList.get(i);
+
+                        if (current.getType() == 0) {
+                            // 掉线记录
+                            if (offlineStartTime == null) {
+                                offlineStartTime = DATE_TIME_FORMATTER.format(current.getTime());
+                            }
+                            // 如果是最后一条记录，结束时间为 24 点
+                            if (i == commList.size() - 1) {
+                                offlineEndTime = DATE_TIME_FORMATTER.format(current.getTime()).substring(0, 10) + " 23:59:59";
+                                offlinePeriods.add(offlineStartTime + " ~ " + offlineEndTime);
+                            }
+                        } else {
+                            // 上线记录
+                            if (offlineStartTime != null) {
+                                // 有对应的掉线记录，形成完整时间段
+                                offlineEndTime = DATE_TIME_FORMATTER.format(current.getTime());
+                                offlinePeriods.add(offlineStartTime + " ~ " + offlineEndTime);
+                                offlineStartTime = null;
+                                offlineEndTime = null;
+                            } else {
+                                // 没有对应的掉线记录，说明是从 0 点开始掉线
+                                offlineStartTime = DATE_TIME_FORMATTER.format(current.getTime()).substring(0, 10) + " 00:00:00";
+                                offlineEndTime = DATE_TIME_FORMATTER.format(current.getTime());;
+                                offlinePeriods.add(offlineStartTime + " ~ " + offlineEndTime);
+                            }
+                        }
+                    }
+                }
+                if (CollUtil.isNotEmpty(offlinePeriods)) {
+                    result.put(devId, offlinePeriods);
+                }
+            });
+        }
+        return result;
+    }
+
+
+    /**
+     * 保存设备掉线时间段
+     *
+     * @param devId 设备 ID
+     * @param startTime 掉线开始时间
+     * @param endTime 掉线结束时间
+     */
+    private void saveOfflinePeriod(String devId, String startTime, String endTime) {
+        // TODO: 将掉线时间段保存到数据库或其他存储介质
+        log.info("设备 {} 掉线时间段：{} - {}", devId, startTime, endTime);
+    }
+
+
+    /**
+     * 构建越限标签字符串
+     *
+     * @param item 越限统计数据
+     * @return 越限标签
+     */
+    private String buildOverlimitTag(RStatLimitRateDPO item) {
+        String tag = "";
+
+        // 基础越限项
+        Integer freqDevOvertime = item.getFreqDevOvertime();
+        if (freqDevOvertime > 0) {
+            tag = tag + "频率偏差越限" + freqDevOvertime + "次，";
+        }
+
+        Integer voltageDevOvertime = item.getVoltageDevOvertime();
+        if (voltageDevOvertime > 0) {
+            tag = tag + "电压偏差越限" + voltageDevOvertime + "次，";
+        }
+
+        Integer ubalanceOvertime = item.getUbalanceOvertime();
+        if (ubalanceOvertime > 0) {
+            tag = tag + "三相电压不平衡度越限" + ubalanceOvertime + "次，";
+        }
+
+        Integer flickerOvertime = item.getFlickerOvertime();
+        if (flickerOvertime > 0) {
+            tag = tag + "闪变越限" + flickerOvertime + "次，";
+        }
+
+        Integer uaberranceOvertime = item.getUaberranceOvertime();
+        if (uaberranceOvertime > 0) {
+            tag = tag + "电压总谐波畸变率越限" + uaberranceOvertime + "次，";
+        }
+
+        Integer iNegOvertime = item.getINegOvertime();
+        if (iNegOvertime > 0) {
+            tag = tag + "负序电流越限" + iNegOvertime + "次，";
+        }
+
+        // 谐波电压含有率（2-25 次）
+        tag = buildHarmonicVoltageTags(item, tag);
+
+        // 谐波电流有效值（2-25 次）
+        tag = buildHarmonicCurrentTags(item, tag);
+
+        // 间谐波电压含有率（0.5-15.5 次）
+        tag = buildInterharmonicVoltageTags(item, tag);
+
+        // 去除末尾逗号
+        return trimTrailingComma(tag);
+    }
+
+    /**
+     * 批量构建谐波电压含有率 tag（2-25 次）
+     *
+     * @param item        数据对象
+     * @param originalTag 原始 tag
+     * @return 组装后的 tag
+     */
+    public static String buildHarmonicVoltageTags(Object item, String originalTag) {
+        String tag = originalTag;
+        for (int i = 2; i <= 25; i++) {
+            String fieldName = "uharm" + i + "Overtime";
+            Integer value = (Integer) ReflectUtil.getFieldValue(item, fieldName);
+            if (value != null && value > 0) {
+                tag = tag + i + "次谐波电压含有率越限" + value + "次，";
+            }
+        }
+        return tag;
+    }
+
+    /**
+     * 批量构建谐波电流有效值 tag（2-25 次）
+     *
+     * @param item        数据对象
+     * @param originalTag 原始 tag
+     * @return 组装后的 tag
+     */
+    public static String buildHarmonicCurrentTags(Object item, String originalTag) {
+        String tag = originalTag;
+        for (int i = 2; i <= 25; i++) {
+            String fieldName = "iharm" + i + "Overtime";
+            Integer value = (Integer) ReflectUtil.getFieldValue(item, fieldName);
+            if (value != null && value > 0) {
+                tag = tag + i + "次谐波电流有效值越限" + value + "次，";
+            }
+        }
+        return tag;
+    }
+
+    /**
+     * 批量构建间谐波电压含有率 tag（0.5-15.5 次）
+     *
+     * @param item        数据对象
+     * @param originalTag 原始 tag
+     * @return 组装后的 tag
+     */
+    public static String buildInterharmonicVoltageTags(Object item, String originalTag) {
+        String tag = originalTag;
+        for (int i = 1; i <= 16; i++) {
+            String fieldName = "inuharm" + i + "Overtime";
+            Integer value = (Integer) ReflectUtil.getFieldValue(item, fieldName);
+            if (value != null && value > 0) {
+                double harmonicOrder = i * 1.0 - 0.5;
+                tag = tag + harmonicOrder + "次间谐波电压含有率越限" + value + "次，";
+            }
+        }
+        return tag;
+    }
+
+    /**
+     * 去除字符串末尾的逗号
+     *
+     * @param str 原字符串
+     * @return 处理后的字符串
+     */
+    public static String trimTrailingComma(String str) {
+        return StrUtil.endWith(str, "，") ? StrUtil.subPre(str, str.length() - 1) : str;
+    }
+
+}

cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/impl/WlUserServiceImpl.java

@@ -6,6 +6,7 @@ import com.njcn.csdevice.api.CsLedgerFeignClient;
 import com.njcn.csdevice.pojo.dto.DevDetailDTO;
 import com.njcn.csdevice.pojo.param.UserDevParam;
 import com.njcn.csdevice.pojo.po.CsDeviceUserPO;
+import com.njcn.csdevice.pojo.vo.DevUserVO;
 import com.njcn.cssystem.pojo.param.WlUserParam;
 import com.njcn.cssystem.pojo.vo.WlUserVo;
 import com.njcn.cssystem.service.IWlUserService;
@@ -74,8 +75,14 @@ public class WlUserServiceImpl implements IWlUserService {
         if (CollectionUtil.isNotEmpty(devList)) {
             List<CsDeviceUserPO> list = new ArrayList<>();
             devList.forEach(item->{
+                //现根据设备获取主用户
+                DevUserVO vo = csDeviceUserFeignClient.queryUserById(item).getData();
                 CsDeviceUserPO po = new CsDeviceUserPO();
-                po.setPrimaryUserId(param.getUserId());
+                if (!Objects.isNull(vo)) {
+                    po.setPrimaryUserId(vo.getMasterUser().getId());
+                } else {
+                    po.setPrimaryUserId(param.getUserId());
+                }
                 po.setSubUserId(param.getUserId());
                 po.setDeviceId(item);
                 po.setStatus("1");

cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/task/DataTask.java

@@ -0,0 +1,35 @@
+package com.njcn.cssystem.task;
+
+import cn.hutool.core.date.DatePattern;
+import cn.hutool.core.date.DateUtil;
+import com.njcn.cssystem.service.IDataTaskService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xy
+ */
+@Component
+@EnableScheduling
+@RequiredArgsConstructor
+@Slf4j
+public class DataTask {
+
+    private final IDataTaskService taskService;
+
+    @Scheduled(cron = "0 0 2 * * ?")
+    public void csHarmonicJob() {
+        String date = DateUtil.yesterday().toString(DatePattern.NORM_DATE_PATTERN);
+        taskService.channelHarmonic(date);
+    }
+
+    @Scheduled(cron = "0 0 3 * * ?")
+    public void csAlarmJob() {
+        String date = DateUtil.yesterday().toString(DatePattern.NORM_DATE_PATTERN);
+        taskService.channelRunAlarm(date);
+    }
+
+}