新增事件导入更新及天级定时更新

src/main/java/com/zhiwei/brandkbs2/controller/admin/EventController.java

@@ -124,7 +124,7 @@ public class EventController extends BaseController {
     @ApiImplicitParam(name = "eventIds", value = "事件ID集合", paramType = "body", dataType = "list")
     @PostMapping("/analyze")
     public ResponseResult analysisEvents(@RequestBody JSONObject info) {
-        eventService.analysisEvents(info.getJSONArray("eventIds").toJavaList(String.class));
+        eventService.analysisEventsAsync(info.getJSONArray("eventIds").toJavaList(String.class));
         behaviorService.pushBehaviorOld(OPERATION, "批量更新事件", request);
         return ResponseResult.success();
     }

src/main/java/com/zhiwei/brandkbs2/service/EventService.java

@@ -190,6 +190,12 @@ public interface EventService {
     void analysisEvents(List<String> eventIds);
 
     /**
+     * 异步批量更新事件
+     * @param eventIds 事件ID集合
+     */
+    void analysisEventsAsync(List<String> eventIds);
+
+    /**
      * 搜索舆情全部事件标签
      *
      * @param linkedGroupId 关联项目组ID
@@ -278,4 +284,11 @@ public interface EventService {
      */
     List<JSONObject> getLastEventTop(String keyword,int limit);
 
+    /**
+     * 获取未结束的事件
+     * @param projectId
+     * @return
+     */
+    List<Event> findNotEndEventByProjectId(String projectId);
+
 }

src/main/java/com/zhiwei/brandkbs2/service/TaskService.java

@@ -33,4 +33,8 @@ public interface TaskService{
      */
     void cleanAggreeTask();
 
+    /**
+     * 事件相关更新
+     */
+    void eventUpdate();
 }

src/main/java/com/zhiwei/brandkbs2/service/impl/EventDataServiceImpl.java

@@ -64,13 +64,19 @@ public class EventDataServiceImpl implements EventDataService {
     @Override
     public void analysisEvent(Event event) {
         long startTime = event.getStartTime();
-        long endTime = event.isEndStatus() ? event.getEndTime() : System.currentTimeMillis();
-        Period periodDay = new Period(startTime, endTime, PeriodType.days());
+        long now = System.currentTimeMillis();
+        // 时间异常
+        if(startTime >= now){
+            return;
+        }
+        long endTime = event.isEndStatus() ? event.getEndTime() : now;
         //时间区间大于40天 就不更新
-        if (periodDay.getDays() >= 40 || startTime > System.currentTimeMillis()) {
-            event.setEndTime(endTime);
-            event.setEndStatus(true);
-            eventDao.updateOne(event);
+        if (new Period(startTime, endTime, PeriodType.days()).getDays() >= 40) {
+            if (!event.isEndStatus()) {
+                event.setEndTime(endTime);
+                event.setEndStatus(true);
+                eventDao.updateOne(event);
+            }
             log.info("analysisEvent-eventId:{},时间区间大于40天，不做更新", event.getId());
             return;
         }

src/main/java/com/zhiwei/brandkbs2/service/impl/EventServiceImpl.java

@@ -138,6 +138,7 @@ public class EventServiceImpl implements EventService {
         AtomicInteger count = new AtomicInteger();
         //事件太多塞满队列了
         ApplicationProjectListener.getThreadPool().execute(() -> {
+            List<String> eventIds = new ArrayList<>();
             yqEventList.forEach(yqEventDTO -> {
                 Event existEvent = eventDao.getEventByUniqueIds(yqEventDTO.getYqEventId(), projectId, contendId);
                 Event event = Event.createFromYqEventDTO(yqEventDTO, collectionName, projectId, contendId);
@@ -149,12 +150,14 @@ public class EventServiceImpl implements EventService {
                     event.setCTime(new Date().getTime());
                     eventDao.insertOne(event);
                 }
+                eventIds.add(event.getId());
                 if (count.incrementAndGet() == yqEventList.size()) {
                     stringRedisTemplate.opsForValue().set(redisKey, String.valueOf(100), 1, TimeUnit.MINUTES);
                 } else {
                     stringRedisTemplate.opsForValue().set(redisKey, String.valueOf((count.get()) * 100 / yqEventList.size()));
                 }
             });
+            analysisEventsAsync(eventIds);
         });
     }
 
@@ -456,7 +459,10 @@ public class EventServiceImpl implements EventService {
 
     @Override
     public void analysisEvents(List<String> eventIds) {
-        ApplicationProjectListener.getThreadPool().execute(() -> eventIds.forEach(eventId -> {
+        if (Tools.isEmpty(eventIds)) {
+            return;
+        }
+        eventIds.forEach(eventId -> {
             String redisKey = null;
             try {
                 Event event = getEventById(eventId);
@@ -469,7 +475,12 @@ public class EventServiceImpl implements EventService {
                 log.error("eventId:{}更新失败", eventId, e);
                 redisUtil.set(redisKey, "-1");
             }
-        }));
+        });
+    }
+
+    @Override
+    public void analysisEventsAsync(List<String> eventIds) {
+        ApplicationProjectListener.getThreadPool().execute(() -> analysisEvents(eventIds));
     }
 
     @Override
@@ -666,6 +677,12 @@ public class EventServiceImpl implements EventService {
         }).sorted((o1, o2) -> Long.compare((long) o2.get("total"), (long) o1.get("total"))).collect(Collectors.toList());
     }
 
+    @Override
+    public List<Event> findNotEndEventByProjectId(String projectId) {
+        Query query = Query.query(Criteria.where("projectId").is(projectId).and("endStatus").is(false));
+        return eventDao.findList(query);
+    }
+
     /**
      * 获取时间筛选条件
      *

src/main/java/com/zhiwei/brandkbs2/service/impl/TaskServiceImpl.java

@@ -3,10 +3,7 @@ package com.zhiwei.brandkbs2.service.impl;
 import com.zhiwei.brandkbs2.auth.UserThreadLocal;
 import com.zhiwei.brandkbs2.common.GlobalPojo;
 import com.zhiwei.brandkbs2.config.Constant;
-import com.zhiwei.brandkbs2.dao.AggreeResultDao;
-import com.zhiwei.brandkbs2.dao.BrandkbsTaskDao;
-import com.zhiwei.brandkbs2.dao.ChannelDao;
-import com.zhiwei.brandkbs2.dao.ReportSettingsDao;
+import com.zhiwei.brandkbs2.dao.*;
 import com.zhiwei.brandkbs2.enmus.ReportTypeEnum;
 import com.zhiwei.brandkbs2.es.ChannelEsDao;
 import com.zhiwei.brandkbs2.es.EsClientDao;
@@ -75,6 +72,9 @@ public class TaskServiceImpl implements TaskService {
     @Resource(name = "commonServiceImpl")
     CommonService commonService;
 
+    @Resource(name = "eventServiceImpl")
+    EventService eventService;
+
     @Resource(name = "taskServiceExecutor")
     ThreadPoolTaskExecutor taskServiceExecutor;
 
@@ -277,6 +277,19 @@ public class TaskServiceImpl implements TaskService {
         log.info("本次清理过期聚合:{}条", deleteCount);
     }
 
+    @Override
+    public void eventUpdate() {
+        for (Project project : projectService.getAllProjectsWithStart()) {
+            List<Event> events = eventService.findNotEndEventByProjectId(project.getId());
+            log.info("{}项目,未结束事件{}个", project.getProjectName(), events.size());
+            if (Tools.isEmpty(events)) {
+                continue;
+            }
+            eventService.analysisEvents(events.stream().map(Event::getId).collect(Collectors.toList()));
+            log.info("{}项目,未结束事件已更新{}个", project.getProjectName(), events.size());
+        }
+    }
+
     private boolean reportSendByProject(Project project) {
         boolean flag = false;
         // 扫描setting信息

src/main/java/com/zhiwei/brandkbs2/task/ControlCenter.java

@@ -77,4 +77,17 @@ public class ControlCenter {
         }
     }
 
+    @Async("scheduledExecutor")
+    @Scheduled(cron = "0 0 2 * * ?")
+    public void eventUpdate() {
+        log.info("定时按天更新事件任务-启动");
+        try {
+            taskService.eventUpdate();
+        } catch (Exception e) {
+            log.error("定时按天更新事件任务-出错", e);
+        } finally {
+            log.info("定时按天更新事件任务-结束");
+        }
+    }
+
 }