【情报二期】数据域服务 - 布控预警结果数据查询接口代码

pom.xml

@@ -42,6 +42,19 @@
       <version>3.5.1</version>
     </dependency>-->
 
+    <!-- GaussDB JDBC Driver -->
+    <!--<dependency>
+      <groupId>com.huawei.gaussdb</groupId>
+      <artifactId>gaussdb-jdbc</artifactId>
+      <version>2.0.0</version>
+    </dependency>-->
+
+    <!--<dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-data-jpa</artifactId>
+      <version>2.5.15</version>
+    </dependency>-->
+
     <!-- JTS Topology Suite 处理二维几何图形的 Java 类库 -->
     <dependency>
       <groupId>org.locationtech.jts</groupId>

src/main/java/com/hhwy/qbeqsjy/controller/CtrlController.java

@@ -5,6 +5,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.hhwy.qbeqsjy.common.Constants;
 import com.hhwy.qbeqsjy.domain.AjaxResult;
 import com.hhwy.qbeqsjy.dto.CtrlDTO;
+import com.hhwy.qbeqsjy.dto.QueryDTO;
 import com.hhwy.qbeqsjy.service.CtrlService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -137,6 +138,22 @@ public class CtrlController {
     }
 
 
+    /**
+     * 条件查询布控任务预警结果数据，支持分页查询，一次查询默认最大返回1000条数据
+     * @param queryDTO
+     * @return
+     */
+    @PostMapping("/getWarningInfo")
+    public AjaxResult getWarningInfo(@RequestBody QueryDTO queryDTO){
+        try{
+            return AjaxResult.success(ctrlService.getWarningInfoList(queryDTO));
+        } catch (Exception e){
+            log.error("查询布控任务预警结果数据发生异常", e);
+        }
+        return AjaxResult.error(Constants.SERVER_ERROR_MSG);
+    }
+
+
 
 
 

src/main/java/com/hhwy/qbeqsjy/dao/CtrlDao.java

@@ -1,15 +1,23 @@
 package com.hhwy.qbeqsjy.dao;
 
+import com.hhwy.qbeqsjy.common.Constants;
+import com.hhwy.qbeqsjy.dto.QueryDTO;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Repository;
 import javax.annotation.Resource;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 @Slf4j
 @Repository
 public class CtrlDao {
 
+    @Value("${query.max.count}")
+    private Integer queryMaxCount;
     @Resource
     private JdbcTemplate gaussDBJdbcTemplate;
 
@@ -71,7 +79,7 @@ public class CtrlDao {
      * @return
      */
     public boolean batchInsertWarningInfo(List<Object[]> warningInfoList){
-        String sql = "insert into t_ctrl_warning_info (task_id, clue, clue_type, warning_msg, create_time) values (?, ?, ?, ?, ?)";
+        String sql = "insert into t_ctrl_warning_info (task_id, clue, clue_type, warning_msg, capture_time) values (?, ?, ?, ?, ?)";
         try {
             int[] insertRows = gaussDBJdbcTemplate.batchUpdate(sql, warningInfoList);
             return insertRows.length == warningInfoList.size();
@@ -80,6 +88,83 @@ public class CtrlDao {
         }
         return false;
     }
+
+
+    /**
+     * 条件查询布控任务预警结果数据
+     * @param queryDTO
+     * @return
+     */
+    public List<Map<String, Object>> queryWarningInfoList(QueryDTO queryDTO){
+        String taskId = queryDTO.getTaskId();
+        String startTime = queryDTO.getStartTime();
+        String endTime = queryDTO.getEndTime();
+        Integer pageNum = queryDTO.getPageNum() == null ? Constants.NUM_1 : queryDTO.getPageNum();
+        Integer pageSize = queryDTO.getPageSize() == null ? queryMaxCount : queryDTO.getPageSize();
+        List<Map<String, Object>> warningInfoList = new ArrayList<>();
+        StringBuilder sb = new StringBuilder("select task_id, clue, clue_type, warning_msg, capture_time from t_ctrl_warning_info where 1 = 1");
+        // 1.根据任务ID过滤
+        if(StringUtils.isNotBlank(taskId)){
+            sb.append(" and task_id = '");
+            sb.append(taskId);
+            sb.append("'");
+        }
+        // 2.根据查询时间段过滤
+        if(StringUtils.isNotBlank(startTime) && StringUtils.isNotBlank(endTime)){
+            sb.append(" and capture_time >= '");
+            sb.append(startTime);
+            sb.append("'");
+            sb.append(" and capture_time <= '");
+            sb.append(endTime);
+            sb.append("'");
+        }
+        // 3.根据截获时间降序排列
+        sb.append(" order by capture_time desc");
+        sb.append(" limit ");
+        sb.append(pageSize);
+        sb.append(" offset ");
+        sb.append(pageNum);
+        try {
+            warningInfoList = gaussDBJdbcTemplate.queryForList(sb.toString());
+        } catch (Exception e){
+            log.error("【查询布控任务预警结果数据发生异常】 异常信息：", e);
+        }
+        return warningInfoList;
+    }
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
 
 
 

src/main/java/com/hhwy/qbeqsjy/dto/QueryDTO.java

@@ -0,0 +1,28 @@
+package com.hhwy.qbeqsjy.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+/**
+ * 数据查询DTO对象
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class QueryDTO {
+
+    // 任务编号
+    private String taskId;
+    // 数据查询开始时间
+    private String startTime;
+    // 数据查询截止时间
+    private String endTime;
+    // 分页查询 - 页码
+    private Integer pageNum;
+    // 分页查询 - 页容量
+    private Integer pageSize;
+
+
+}

src/main/java/com/hhwy/qbeqsjy/kafka/service/ConsumeService.java

@@ -1,13 +1,10 @@
 package com.hhwy.qbeqsjy.kafka.service;
 
-import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
-import com.hhwy.qbeqsjy.common.CommonUtil;
 import com.hhwy.qbeqsjy.common.Constants;
 import com.hhwy.qbeqsjy.common.TimeTool;
 import com.hhwy.qbeqsjy.dao.CtrlDao;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -16,12 +13,10 @@ import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.stereotype.Service;
-
 import javax.annotation.Resource;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Date;
 import java.util.List;
 
 /**
@@ -61,19 +56,19 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                 log.info("kafka集群已开启安全模式");
                 AuthService.securityPrepare();
             } catch (IOException e) {
-                log.error("kafka集群安全认证失败！异常信息 : {}.", e);
+                log.error("kafka集群安全认证失败，异常信息 : {}.", e);
                 return;
             }
             log.info("kafka集群安全认证成功");
         }
         AuthService consumerAuthService = new AuthService(topic);
-        log.info("【QBEQSJYCTRL】 消费者服务已启动");
+        log.info("【QBEQSJYCTRL - 消费者服务】 服务已启动");
         process(consumerAuthService.consumer, topic);
     }
 
 
     /**
-     * 消费目标topic中预警结果数据
+     * 消费目标topic中的预警结果数据，处理后批量入库
      * @param consumer
      * @param topic
      */
@@ -83,29 +78,16 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
         while (true) {
             // 间隔30秒拉取一次数据
             ConsumerRecords<Integer, String> records = consumer.poll(Constants.WAIT_TIME);
-            // 业务处理逻辑（解析、处理、入库）
+            // 数据解析、入库
             if (records != null && records.count() > 0) {
-                Date createTime = new Date();
                 for (ConsumerRecord<Integer, String> record : records) {
                     JSONObject message = JSONObject.parseObject(record.value());
 //                    log.info(message.toJSONString());
-                    // 解析[ORIFIELD]节点，该节点数据格式与布控资源表有关，资源表不同，数据格式不同
+                    // 获取[ORIFIELD]节点数据，该节点数据格式与布控资源表有关，表结构不同，数据格式不同
                     JSONObject oriField = message.getJSONObject("ORIFIELD");
-                    String baseStationId = oriField.getString("BASE_STATION_ID");
-                    // 过滤基站ID为空的数据
-                    if (StringUtils.isBlank(baseStationId)) {
-                        continue;
-                    }
-
-                    // 查询当前任务对应的布控区域信息，进行范围比对
-
-
-                    String lon = oriField.getString("LONGITUDE");
-                    String lat = oriField.getString("LATITUDE");
-
-
-
-                    // 解析[UNICTRL]节点，该节点数据格式保持一致，与布控资源表无关
+                    // 考虑到[ORIFIELD]节点数据格式不统一，此处不进行解析，数据域统一入库，并以接口形式同步到用户域后，由前端业务针对不同的资源表进行定制化解析
+                    String warningInfoJSONStr = oriField.toJSONString();
+                    // 解析[UNICTRL]节点，该节点数据为统一格式，与布控资源表无关
                     JSONObject uniCtrl = message.getJSONObject("UNICTRL");
                     // 样例：clueId = 120000_59275350793392128_D201005_13502082832
                     String clueId = uniCtrl.getString("U_CLUEID");
@@ -113,17 +95,16 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                     String taskId = clueIdStrArr[Constants.NUM_1];
                     String clueType = clueIdStrArr[Constants.NUM_2];
                     String clue = clueIdStrArr[Constants.NUM_3];
-                    String warningInfoJSONStr = message.toJSONString();
-                    String resourceId = uniCtrl.getString("U_RESID");
-                    String captureTime = TimeTool.timeStampToDateString(Long.valueOf(oriField.getString("CAPTURE_TIME")));
-                    Object[] warningInfo = new Object[]{taskId, clue, clueType, warningInfoJSONStr, createTime};
+                    // 原始预警数据中的"截获时间"
+//                    String captureTime = TimeTool.timeStampToDateString(Long.valueOf(oriField.getString("CAPTURE_TIME")));
+                    // 不采用原始预警数据中的"截获时间"，而使用当前时间作为"截获时间"
+                    String captureTime = TimeTool.timeStampToDateString(TimeTool.getNowTimeStamp());
+                    Object[] warningInfo = new Object[]{taskId, clue, clueType, warningInfoJSONStr, captureTime};
                     warningInfoList.add(warningInfo);
-
-
                 }
-                //====================== 3.符合条件的预警数据批量入库 - to do ======================
+                //====================== 预警数据批量入库 ======================
                 ctrlDao.batchInsertWarningInfo(warningInfoList);
-                log.info("【QBEQSJYCTRL - 消费者】 已消费数据量：{}", warningInfoList.size());
+                log.info("【QBEQSJYCTRL - 消费者服务】 当次消费的数据量：{}", warningInfoList.size());
                 warningInfoList.clear();
             }
         }

src/main/java/com/hhwy/qbeqsjy/service/CtrlService.java

@@ -9,6 +9,7 @@ import com.hhwy.qbeqsjy.common.TimeTool;
 import com.hhwy.qbeqsjy.dao.CtrlDao;
 import com.hhwy.qbeqsjy.domain.AjaxResult;
 import com.hhwy.qbeqsjy.dto.CtrlDTO;
+import com.hhwy.qbeqsjy.dto.QueryDTO;
 import com.hhwy.qbeqsjy.task.CtrlTask;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -43,6 +44,8 @@ public class CtrlService {
     private String multiThreadEnabled;
     @Value("${thread.num}")
     private int threadNum;
+    @Value("${res.auth.code}")
+    private String resAUTHCode;
     @Value("${ctrl.txt.output.path}")
     private String ctrlTxtOutputPath;
     @Value("${stop.ctrl.txt.output.path}")
@@ -71,6 +74,11 @@ public class CtrlService {
         if(!Constants.MULTI_THREAD_ENABLED_FLAG.equals(multiThreadEnabled)){
             ExecutorService executorService = Executors.newFixedThreadPool(Constants.NUM_1);
             executorService.execute(() -> {
+                // 1.如果线索是手机号，且任务信息中包含布控区域，则下发区域布控（区域范围预警）；如果不包含布控区域，则下发账号布控（全天津市范围预警）
+                // 2.如果线索是身份证号，则不考虑布控区域的因素，只下发账号布控（全天津市登记类数据预警：如宾馆、网吧等）
+
+                // - todo
+
                 Date createTime = new Date();
                 List<Object[]> ctrlErrorList = new ArrayList<>();
                 int ctrlNum = 0;
@@ -502,6 +510,16 @@ public class CtrlService {
     }
 
 
+    /**
+     * 条件查询布控任务预警结果数据
+     * @param queryDTO
+     * @return
+     */
+    public List<Map<String, Object>> getWarningInfoList(QueryDTO queryDTO){
+        return ctrlDao.queryWarningInfoList(queryDTO);
+    }
+
+
 
 
 }

src/main/java/com/hhwy/qbeqsjy/strategy/impl/AuthStrategy.java

@@ -14,7 +14,6 @@ import java.util.Date;
 public class AuthStrategy implements Strategy {
     @Override
     public JSONObject doParse(JSONObject warningInfo) {
-        // 原始字段
         JSONObject oriField = warningInfo.getJSONObject("ORIFIELD");
         // 线索值
 //        String clue = oriField.getString("SRC_AUTH_ACCOUNT");

src/main/resources/application.properties

@@ -51,9 +51,11 @@ thread.num=5
 ctrl.txt.output.path=/home/qbeqsjy/ctrl_error/
 stop.ctrl.txt.output.path=/home/qbeqsjy/stop_ctrl_error/
 
+# \u9884\u8B66\u6570\u636E\u67E5\u8BE2\u4E00\u6B21\u6700\u5927\u8FD4\u56DE\u6570\u636E\u6761\u6570\uFF0C\u9ED8\u8BA41000\u6761
+query.max.count=1000
+
 # \u5E03\u63A7\u8D44\u6E90\u5217\u8868
-# \u6570\u636E\u8D44\u6E90\u7F16\u7801=\u539F\u59CB\u5B57\u6BB5
-134B102=
+res.auth.code=134B102