人员合并

src/main/java/com/winhc/task/PersonMergeIncremnetTask.java

@@ -104,6 +104,12 @@ public class PersonMergeIncremnetTask {
         return flag;
     }
 
+    private boolean saveMysqlStatus(String ds, String status, String topic_type) {
+        String sql = "REPLACE INTO node_relation_success_status VALUES (" + ds + ", " + status + ", " + topic_type + ", now()" + ", now())";
+        jdbcTemplate.execute(sql);
+        return true;
+    }
+
     private boolean querySussessStatus(MergeParam param) throws UnsupportedEncodingException, InterruptedException {
         boolean flag = false;
         ArrayList<Integer> hours = new ArrayList<>();
@@ -116,16 +122,29 @@ public class PersonMergeIncremnetTask {
                 flag = true;
                 break;
             }
-            Thread.sleep(60 * 1000);
-            log.info("waiting  success status !!!");
+
             int hour = DateUtil.getCurrentHours();
-            if (hour >= 8 && !hours.contains(hour)) {
-                hours.add(hour);
-                log.info("上游阻塞告警 !!!!!!");
+            if (hour > 10) {
+                log.info("start  MergePersonScheduled !!!");
                 if (!CompanyUtils.isWindows()) {
-                    dingUtils.send("上游阻塞告警 !!!!!!");
+                    dingUtils.send("MergePersonScheduled start !!!");
                 }
+                //存储任务记录
+                saveMysqlStatus(param.getLastPartition(), "1", "100");
+                Thread.sleep(5 * 60 * 1000);
+                flag = true;
+                break;
             }
+            Thread.sleep(60 * 1000);
+            log.info("waiting  success status !!!");
+
+//            if (hour >= 8 && !hours.contains(hour)) {
+//                hours.add(hour);
+//                log.info("上游阻塞告警 !!!!!!");
+//                if (!CompanyUtils.isWindows()) {
+//                    dingUtils.send("上游阻塞告警 !!!!!!");
+//                }
+//            }
 
             //超时退出
             if (DateUtil.getCurrentHours() >= 22) {
@@ -155,7 +174,7 @@ public class PersonMergeIncremnetTask {
     private void mergeAndExport(MergeParam param) {
         Session session = driver.session();
         //合并昨天逻辑
-        mergePerson(param.getLastPartition(), session);
+        //mergePerson(param.getLastPartition(), session);
         //导出昨天合并csv
         exportMergePerson2CSV(param.getLastPartition(), session, param.getMergePath());
         //导出昨天删除csv
@@ -265,7 +284,7 @@ public class PersonMergeIncremnetTask {
                 map.put("update_time", time);
                 map.put("person_id", list.get(0));
                 map.put("person_name", list.get(1));
-                if ("0".equals(flag)) {//合并
+                if ("0" .equals(flag)) {//合并
                     map.put("company_id", list.get(2));
                     map.put("deleted", list.get(3));
                     map.put("label", list.get(4));