Merge remote-tracking branch 'origin/jtdev' into jtdev

guoshilong

2023-07-07 c9f380fbb0076adb15c63814171950e6dc39d8ce

Merge remote-tracking branch 'origin/jtdev' into jtdev

 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/notice/constant/ZtApiUrlConstant.java

@@ -38,7 +38,7 @@
    public static final String over_z_res_api = "/services/1234567890ABCDEFGHIJKLMN/over_z_res";

    /**
     *
     * 水位超设计洪水位告警处理
     */
    public static final String over_des_rz_warn_api = "/services/1234567890ABCDEFGHIJKLMN/over_des_rz/warn";


 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/notice/service/impl/AsyncNoticeHandle.java

@@ -25,15 +25,52 @@
    @Autowired
    private SmsService smsService;


    /**
     * 异步发送告警信息
     * 水位初次超汛限告警 处理
     * @param templateId
     * @param uuid
     * @param alarmRule
     * @param adCode
     * @return
     */
    @Async
    public Map<String, Object> waterFirstOverHandle(String templateId, String uuid, AlarmRule alarmRule,String adCode) {
        Map<String, Object> map = new HashMap<>();
        // 调用中台服务接口查询数据
        JSONArray ztData = noticeHandle.getZtData("", ZtApiUrlConstant.over_z_res_api);
        // 只发短信，将发短信信息组装好
        SmsRequestTemplate smsRequestTemplate
            = noticeHandle.getSendSmsTemplate(ztData, templateId,alarmRule, ZtApiDataColumnConstant.waterFirstOverList,true,adCode);
        // 拼接后面内容
        if (smsRequestTemplate.getPersonVOList().size()>0) {
            for (List<String> list : smsRequestTemplate.getTemplateContent()) {
                // 依次拼接巡查责任人，技术责任人姓名
                noticeHandle.addPatrolTechnologyInfo(list);
                // 拼接市县水利部人员姓名及联系方式
                noticeHandle.addCityCountyBusinessPersonInfo(list);
                // 保存告警记录信息
                AlarmRecord alarmRecord = noticeHandle.saveAlarmRecord(list,alarmRule,uuid);
                if (null != alarmRecord) {
                    list.add(alarmRecord.getId().toString());
                }
            }
            // 调用短信服务
            map = smsService.sendSignMsg(smsRequestTemplate);
            return map;
        }
        return map;
    }

    /**
     * 水位超设计洪水位告警处理
     * @param templateId
     * @param uuid
     * @param alarmRule
     * @return
     */
    @Async
    public Map<String, Object> handleAlarmInfo(String templateId, String uuid, AlarmRule alarmRule) {
    public Map<String, Object> waterMoreDesignHandle(String templateId, String uuid, AlarmRule alarmRule) {
        Map<String, Object> map = new HashMap<>();
        // 调用中台服务接口查询数据
        JSONArray ztData = noticeHandle.getZtData("", ZtApiUrlConstant.over_des_rz_warn_api);

 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/notice/service/impl/NoticeHandle.java

@@ -131,6 +131,8 @@
    private void jointData(JSONArray jsonArray, List<String> columnsList, AlarmRule alarmRule, Boolean flag, List<PersonVO> personLists, List<List<String>> lists, int i) {
        // 根据水库编号，所在区域查询需要发送的人员手机号及设置
        String res_cd = jsonArray.getJSONObject(i).getString("res_cd");
        // 取出测站编码
        String stcd = jsonArray.getJSONObject(i).getString("stcd");
        // 判断时间是否存在
        String tm = jsonArray.getJSONObject(i).getString("tm");
        String day = "";
@@ -146,7 +148,7 @@
            flag = false;
        }
        // 获取告警人员信息
        getAlarmPersonList(jsonArray, columnsList, alarmRule, flag, personLists, lists, i, res_cd, day, hour, minute);
        getAlarmPersonList(jsonArray, columnsList, alarmRule, flag, personLists, lists, i, res_cd, day, hour, minute,stcd);
    }

    /**
@@ -158,10 +160,11 @@
     * @param personLists
     * @param lists
     * @param i
     * @param res_cd
     * @param day
     * @param hour
     * @param minute
     * @param res_cd 水库编码
     * @param day 日期
     * @param hour 小时
     * @param minute 分钟
     * @param stcd 测站编码
     */
    private void getAlarmPersonList(JSONArray jsonArray,
                                    List<String> columnsList,
@@ -173,12 +176,14 @@
                                    String res_cd,
                                    String day,
                                    String hour,
                                    String minute) {
                                    String minute,
                                    String stcd) {
        // 查询告警人员信息
        List<PersonVO> personVOList = getPersonListByResGuid(res_cd, alarmRule);
        if (personVOList.size() > 0) {
            for (PersonVO personVO : personVOList) {
                List<String> list = new ArrayList<>();
                list.add(stcd);
                list.add(res_cd);
                list.add(personVO.getPhone());
                if (flag) {
@@ -220,12 +225,15 @@
            AlarmRecord alarm = new AlarmRecord();
            // 设置相关信息
            alarm.setAlarmRuleId(alarmRule.getId().toString());

            alarm.setReservoirNumber(list.get(0));
            // 分别设置测站编号和水库编码
            alarm.setAlarmSurveyStationCode(list.get(0));
            alarm.setReservoirNumber(list.get(1));
            alarm.setAlarmCondition(alarmRule.getRuleRemark());
            alarm.setTempCode(uuid);
            // 新增
            boolean save = alarmRecordService.save(alarm);
            // 删除测站编号
            list.remove(0);
            if (save) {
                return alarm;
            }
@@ -276,8 +284,8 @@
        // 查询巡查责任人姓名，技术责任人姓名信息
        List<PersonVO> personVOList = attResManagePersonClient.getPersonListByResGuid(list.get(0),"4,5");
        if (personVOList.size()>0) {
            String xczrr = "";
            String jszrr = "";
            String xczrr = "-";
            String jszrr = "-";
            for (PersonVO personVO : personVOList) {
                if (personVO.getType().equals("5")) {
                    xczrr = personVO.getName();

 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/notice/service/impl/NoticeStrategyImpl.java

@@ -241,30 +241,9 @@
        // 查询当前策略对应的告警规则信息
        AlarmRule alarmRule = alarmRuleService.getOne(new QueryWrapper<AlarmRule>().eq("rule_name", arg1));
        if (null!=alarmRule) {
            // 调用中台服务接口查询数据
            JSONArray ztData = noticeHandle.getZtData("", ZtApiUrlConstant.over_z_res_api);
            // 只发短信，将发短信信息组装好
            SmsRequestTemplate smsRequestTemplate
                = noticeHandle.getSendSmsTemplate(ztData, templateId,alarmRule, ZtApiDataColumnConstant.waterFirstOverList,true,adCode);
            // 拼接后面内容
            if (smsRequestTemplate.getPersonVOList().size()>0) {
                for (List<String> list : smsRequestTemplate.getTemplateContent()) {
                    // 依次拼接巡查责任人，技术责任人姓名
                    noticeHandle.addPatrolTechnologyInfo(list);
                    // 拼接市县水利部人员姓名及联系方式
                    noticeHandle.addCityCountyBusinessPersonInfo(list);
                    // 保存告警记录信息
                    AlarmRecord alarmRecord = noticeHandle.saveAlarmRecord(list,alarmRule,uuid);
                    if (null != alarmRecord) {
                        list.add(alarmRecord.getId().toString());
//                        list.add("999");
                    }
                }
                // 调用短信服务
                Map<String, Object> map = smsService.sendSignMsg(smsRequestTemplate);
                // 返回
                return "本次-水位初次超汛限告警短信成功通知数量：" + map.get("smsRecord");
            }
            asyncNoticeHandle.waterFirstOverHandle(templateId, uuid, alarmRule,adCode);
            // 返回
            return "水位初次超汛限告警";
        }
        // 返回
        return "水位初次超汛限告警";
@@ -321,7 +300,7 @@
        AlarmRule alarmRule = alarmRuleService.getOne(new QueryWrapper<AlarmRule>().eq("rule_name", arg1));
        if (null!=alarmRule) {
            // 异步处理，暂时解决超时重置问题
            Map<String, Object> map = asyncNoticeHandle.handleAlarmInfo(templateId, uuid, alarmRule);
            Map<String, Object> map = asyncNoticeHandle.waterMoreDesignHandle(templateId, uuid, alarmRule);
            // 返回
            return "水位超设计洪水位告警";
        }

 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/sse/controller/SSEController.java

New file
@@ -0,0 +1,41 @@
package cn.gistack.alerts.sse.controller;

import cn.gistack.alerts.sse.server.SSEServer;
import cn.gistack.alerts.sse.vo.SseVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@Slf4j
@RestController
@CrossOrigin
@RequestMapping("/sse/sse")
public class SSEController {

    /**
     * 建立连接
     * @param sse
     * @return
     */
    @GetMapping("/connect")
    public SseEmitter connect(SseVO sse){
        return SSEServer.connect(sse.getType() + ":" + sse.getUserId());
    }

    /**
     * 发送消息
     * @throws InterruptedException
     */
    @GetMapping("/process")
    public void sendMessage() throws InterruptedException {
        SSEServer.sendMessage("web:123","hello!");
//        for(int i=0; i<=100; i++){
//            if(i>50&&i<70){
//                Thread.sleep(500L);
//            }else{
//                Thread.sleep(100L);
//            }
//            SSEServer.batchSendMessage(String.valueOf(i));
//        }
    }
}

 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/sse/server/SSEServer.java

New file
@@ -0,0 +1,148 @@
package cn.gistack.alerts.sse.server;

import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

@Slf4j
public class SSEServer {

    /**
     * 当前连接数
     */
    private static AtomicInteger count = new AtomicInteger(0);

    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    public static SseEmitter connect(String userId){
        //设置超时时间，0表示不过期，默认是30秒，超过时间未完成会抛出异常
        SseEmitter sseEmitter = new SseEmitter(0L);
        //注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeOutCallBack(userId));
        sseEmitterMap.put(userId,sseEmitter);
        //数量+1
        count.getAndIncrement();
        log.info("create new sse connect ,current user:{}",userId);
        return sseEmitter;
    }
    /**
     * 给指定用户发消息
     */
    public static void sendMessage(String userId, String message){
        if(sseEmitterMap.containsKey(userId)){
            try{
                sseEmitterMap.get(userId).send(message);
            }catch (IOException e){
                log.error("user id:{}, send message error:{}",userId,e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * 想多人发送消息，组播
     */
    public static void groupSendMessage(String groupId, String message){
        if(sseEmitterMap!=null&&!sseEmitterMap.isEmpty()){
            sseEmitterMap.forEach((k,v) -> {
                try{
                    if(k.startsWith(groupId)){
                        v.send(message, MediaType.APPLICATION_JSON);
                    }
                }catch (IOException e){
                    log.error("user id:{}, send message error:{}",groupId,message);
                    removeUser(k);
                }
            });
        }
    }

    /**
     * 批量发送消息
     * @param message
     */
    public static void batchSendMessage(String message) {
        sseEmitterMap.forEach((k,v)->{
            try{
                v.send(message,MediaType.APPLICATION_JSON);
            }catch (IOException e){
                log.error("user id:{}, send message error:{}",k,e.getMessage());
                removeUser(k);
            }
        });
    }

    /**
     * 群发消息
     */
    public static void batchSendMessage(String message, Set<String> userIds){
        userIds.forEach(userId->sendMessage(userId,message));
    }

    /**
     * 用户离线删除用户
     * @param userId
     */
    public static void removeUser(String userId){
        sseEmitterMap.remove(userId);
        //数量-1
        count.getAndDecrement();
        log.info("remove user id:{}",userId);
    }


    public static List<String> getIds(){
        return new ArrayList<>(sseEmitterMap.keySet());
    }

    public static int getUserCount(){
        return count.intValue();
    }

    /**
     * 结束回调
     * @param userId
     * @return
     */
    private static Runnable completionCallBack(String userId) {
        return () -> {
            log.info("结束连接,{}",userId);
            removeUser(userId);
        };
    }

    /**
     * 超时回调
     * @param userId
     * @return
     */
    private static Runnable timeOutCallBack(String userId){
        return ()->{
            log.info("连接超时,{}",userId);
            removeUser(userId);
        };
    }

    /**
     * 错误回调
     * @param userId
     * @return
     */
    private static Consumer<Throwable> errorCallBack(String userId){
        return throwable -> {
            log.error("连接异常，{}",userId);
            removeUser(userId);
        };
    }
}

 skjcmanager/skjcmanager-service/skjcmanager-alerts/src/main/java/cn/gistack/alerts/sse/vo/SseVO.java

New file
@@ -0,0 +1,17 @@
package cn.gistack.alerts.sse.vo;

import lombok.Data;

@Data
public class SseVO {

    /**
     * 类型 web,app,小程序
     */
    private String type;

    /**
     * 用户唯一值
     */
    private String userId;
}

 skjcmanager/skjcmanager-service/skjcmanager-sm/src/main/java/cn/gistack/sm/patrol/feign/PatrolTaskImplClient.java

@@ -71,6 +71,7 @@
                // 创建任务
                patrolTaskService.startProcess(patrolTask);
            } catch (Exception e) {
                log.error("创建任务失败："  + e.getMessage());
                log.error("创建任务失败对象："  + JSON.toJSONString(item));
                failureCount.getAndIncrement();
            }

 skjcmanager/skjcmanager-service/skjcmanager-sm/src/main/java/cn/gistack/sm/resDataCheck/feign/RulesClient.java

@@ -5,10 +5,13 @@
import cn.gistack.sm.resDataCheck.service.IQualityManageService;
import cn.gistack.sm.resDataCheck.service.IRulesConditionService;
import cn.gistack.sm.resDataCheck.service.IRulesService;
import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.mp.support.Condition;
import org.springblade.core.tenant.annotation.NonDS;
import org.springblade.core.tool.api.R;
import org.springblade.core.tool.utils.StringUtil;
import org.springframework.web.bind.annotation.RestController;
import springfox.documentation.annotations.ApiIgnore;

@@ -24,6 +27,7 @@
@ApiIgnore
@RestController
@AllArgsConstructor
@Slf4j
public class RulesClient implements IRulesClient {

    private IRulesService rulesService;
@@ -202,6 +206,7 @@

    @Override
    public R createDataRules(Long id) {
        log.info("规则匹配中......");
        // 获取规则对象  返回值有关联水库，只匹配关联水库，为空表示匹配所有水库
        Rules rules = rulesService.getById(id);

@@ -237,12 +242,26 @@

        }

        if (StringUtil.isNotBlank(rules.getIsGate())) {
            sqlStr += "a.\"is_gate\" = " + rules.getIsGate();
        }

        // 拼接条件有问题，不操作
        if (sqlStr.indexOf("-1") > 0) {
            log.info("查询条件错误，请检查规则设置......：",sqlStr);
            return R.fail(-1,sqlStr + "\\n 查询条件错误，请检查规则设置");
        }

        try {
            qualityManageService.addQualityManage(id,sqlStr);
        } catch (Exception e) {
            e.printStackTrace();
            log.info("查询条件错误，请检查规则设置......：",sqlStr);
        } finally {
            log.info("执行完成......：");
        }

        // 执行sql
        return R.data(qualityManageService.addQualityManage(id,sqlStr));
        return R.success("");
    }
}