事件增删改查逻辑修改

src/main/java/com/zzsn/event/constant/Constants.java

@@ -65,12 +65,16 @@ public class Constants {
     }};
 
     //事件默认绑定搜索引擎id集合
-    public final static List<String> DEFAULT_SEARCH_ENGINE = Arrays.asList("1539145903551361026","1539146006513135617","1539146115049140225");
+    public static final Map<String, String> DEFAULT_SEARCH_ENGINE = new HashMap<String, String>() {{
+        put("1539145903551361026","3");
+        put("1539146006513135617","4");
+        put("1539146115049140225","5");
+    }};
 
     /**
      * 关键词数据入缓存 key前缀
      */
-    public static final String KEY_WORDS_TO_REDIS_PREFIX = "KEY_WORDS_TO_REDIS_1::";
+    public static final String KEY_WORDS_TO_REDIS_PREFIX = "KEY_WORDS_TO_REDIS::";
     //关键词采集通道
-    public static final String KEY_WORDS_COLLECT_TOPIC = "eventKeyWordsCrawl_1";
+    public static final String KEY_WORDS_COLLECT_TOPIC = "eventKeyWordsInfo";
 }

src/main/java/com/zzsn/event/controller/EventDataController.java

@@ -117,7 +117,7 @@ public class EventDataController {
         BeanUtils.copyProperties(keyWords, redisKeywordDTO);
         redisKeywordDTO.setStartTime(event.getStartTime());
         redisKeywordDTO.setEndTime(event.getEndTime());
-        redisKeywordDTO.setSearchEngines(Constants.DEFAULT_SEARCH_ENGINE);
+        redisKeywordDTO.setSearchEngines(new ArrayList<>(Constants.DEFAULT_SEARCH_ENGINE.values()));
         redisUtil.set(Constants.KEY_WORDS_TO_REDIS_PREFIX + keyWords.getWordsCode(), redisKeywordDTO);
         //插入xxljob
         iXxlJobInfoService.keyWordsInsert(redisKeywordDTO);

src/main/java/com/zzsn/event/controller/EventManageController.java

@@ -25,7 +25,6 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.Resource;
@@ -141,6 +140,7 @@ public class EventManageController {
             eventParam.setUpdateTime(new Date());
             eventParam.setCreateBy(currentUser.getUsername());
             Event event = eventService.saveMain(eventParam);
+            CompletableFuture.runAsync(()->{
                 //新建的专题授权给自己和客户
                 if (currentUser.getCategory().equals(Constants.ADMIN_USER)) {
                     customerDataPermissionMapService.save(new CustomerDataPermissionMap().setCustomerId(currentUser.getCustomerId()).setPermissionId(event.getId()).setCategory(Constants.PERMISSION_SUBJECT));
@@ -157,12 +157,13 @@ public class EventManageController {
                 BeanUtils.copyProperties(keyWords, redisKeywordDTO);
                 redisKeywordDTO.setStartTime(event.getStartTime());
                 redisKeywordDTO.setEndTime(event.getEndTime());
-            redisKeywordDTO.setSearchEngines(Constants.DEFAULT_SEARCH_ENGINE);
+                redisKeywordDTO.setSearchEngines(new ArrayList<>(Constants.DEFAULT_SEARCH_ENGINE.values()));
                 redisUtil.set(Constants.KEY_WORDS_TO_REDIS_PREFIX + keyWords.getWordsCode(), redisKeywordDTO);
                 //插入xxljob
                 iXxlJobInfoService.keyWordsInsert(redisKeywordDTO);
                 //为了立即响应，关键词新增时放入消息队列
                 kafkaTemplate.send(Constants.KEY_WORDS_COLLECT_TOPIC, JSON.toJSONString(redisKeywordDTO));
+            });
             return Result.OK();
         } else {
             return Result.FAIL(500,"关键词不能为空");
@@ -187,12 +188,14 @@ public class EventManageController {
         keyWordsService.update(Wrappers.<KeyWords>lambdaUpdate().eq(KeyWords::getId,keywordsVO.getId())
                 .set(KeyWords::getKeyWord,keywordsVO.getKeyword())
                 .set(KeyWords::getExclusionWord,keywordsVO.getExclusionWord()));
+        CompletableFuture.runAsync(()->{
             KeyWordsDTO redisKeywordDTO = new KeyWordsDTO();
             BeanUtils.copyProperties(keywordsVO, redisKeywordDTO);
-        redisKeywordDTO.setSearchEngines(Constants.DEFAULT_SEARCH_ENGINE);
+            redisKeywordDTO.setSearchEngines(new ArrayList<>(Constants.DEFAULT_SEARCH_ENGINE.values()));
             redisUtil.set(Constants.KEY_WORDS_TO_REDIS_PREFIX + keywordsVO.getWordsCode(), redisKeywordDTO);
             //为了立即响应，关键词编辑时放入消息队列
             kafkaTemplate.send(Constants.KEY_WORDS_COLLECT_TOPIC, JSON.toJSONString(redisKeywordDTO));
+        });
         return Result.OK();
     }
 
@@ -205,15 +208,7 @@ public class EventManageController {
      */
     @GetMapping("/enable")
     public Result<?> enable(@RequestParam String eventId){
-        int status = 1;
-        Event event = eventService.getById(eventId);
-        eventService.update(Wrappers.<Event>lambdaUpdate().eq(Event::getId,eventId).set(Event::getStatus,status));
-        iXxlJobInfoService.update(Wrappers.<XxlJobInfo>lambdaUpdate().eq(XxlJobInfo::getInfoSourceCode,event.getEventCode())
-                .set(XxlJobInfo::getTriggerStatus,status));
-        //关键词
-        KeywordsVO keywordsVO = keyWordsService.keywordInfoByEventId(eventId);
-        iXxlJobInfoService.update(Wrappers.<XxlJobInfo>lambdaUpdate().eq(XxlJobInfo::getInfoSourceCode,keywordsVO.getWordsCode())
-                .set(XxlJobInfo::getTriggerStatus,status));
+        updateStatus(eventId,1);
         return Result.OK();
     }
 
@@ -226,16 +221,21 @@ public class EventManageController {
      */
     @GetMapping("/disable")
     public Result<?> disable(@RequestParam String eventId){
-        int status = 0;
-        Event event = eventService.getById(eventId);
+        updateStatus(eventId,0);
+        return Result.OK();
+    }
+
+    private void updateStatus(String eventId,Integer status){
         eventService.update(Wrappers.<Event>lambdaUpdate().eq(Event::getId,eventId).set(Event::getStatus,status));
+        CompletableFuture.runAsync(()->{
+            Event event = eventService.getById(eventId);
             iXxlJobInfoService.update(Wrappers.<XxlJobInfo>lambdaUpdate().eq(XxlJobInfo::getInfoSourceCode,event.getEventCode())
                     .set(XxlJobInfo::getTriggerStatus,status));
             //关键词
             KeywordsVO keywordsVO = keyWordsService.keywordInfoByEventId(eventId);
             iXxlJobInfoService.update(Wrappers.<XxlJobInfo>lambdaUpdate().eq(XxlJobInfo::getInfoSourceCode,keywordsVO.getWordsCode())
                     .set(XxlJobInfo::getTriggerStatus,status));
-        return Result.OK();
+        });
     }
 
     /**

src/main/java/com/zzsn/event/kafka/KafkaConsumer.java

 package com.zzsn.event.kafka;
 
 import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.zzsn.event.constant.Constants;
-import com.zzsn.event.entity.Event;
-import com.zzsn.event.entity.EventAnalysisReport;
-import com.zzsn.event.entity.SubjectAnalysis;
-import com.zzsn.event.service.EventAnalysisReportService;
-import com.zzsn.event.service.IEventService;
-import com.zzsn.event.service.SubjectAnalysisService;
+import com.zzsn.event.entity.*;
+import com.zzsn.event.es.EsService;
+import com.zzsn.event.service.*;
+import com.zzsn.event.util.DateUtil;
+import com.zzsn.event.util.EsDateUtil;
+import com.zzsn.event.util.EsOpUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -20,6 +21,7 @@ import org.springframework.stereotype.Component;
 
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 /**
  * @author lkg
@@ -35,7 +37,15 @@ public class KafkaConsumer {
     @Autowired
     private IEventService eventService;
     @Autowired
+    private EventExtractService eventExtractService;
+    @Autowired
     private EventAnalysisReportService eventAnalysisReportService;
+    @Autowired
+    private EsService esService;
+    @Autowired
+    private IEventCategoryService eventCategoryService;
+    @Autowired
+    private EsOpUtil esOpUtil;
 
     /**
      * 获取-观点分析-分析结果数据，并入库
@@ -112,9 +122,9 @@ public class KafkaConsumer {
                     e.setTitle(removeNonBmpUniCodes(e.getTitle()));
                 });
                 LambdaQueryWrapper<SubjectAnalysis> queryWrapper = Wrappers.lambdaQuery();
-                queryWrapper.eq(SubjectAnalysis::getSubjectId,subjectId).eq(SubjectAnalysis::getCategory,category);
+                queryWrapper.eq(SubjectAnalysis::getSubjectId, subjectId).eq(SubjectAnalysis::getCategory, category);
                 int count = subjectAnalysisService.count(queryWrapper);
-                if (count>0) {
+                if (count > 0) {
                     subjectAnalysisService.remove(queryWrapper);
                 }
                 subjectAnalysisService.saveBatch(subjectAnalyses);
@@ -136,14 +146,85 @@ public class KafkaConsumer {
     public void eventReport(ConsumerRecord<String, String> record) {
         String value = record.value();
         EventAnalysisReport eventAnalysisReport = JSONObject.parseObject(value, EventAnalysisReport.class);
-        eventAnalysisReportService.modify(eventAnalysisReport.getEventId(),eventAnalysisReport.getFilePath());
-        log.info("id为-{}-的事件,分析报告更新完成",eventAnalysisReport.getEventId());
+        eventAnalysisReportService.modify(eventAnalysisReport.getEventId(), eventAnalysisReport.getFilePath());
+        log.info("id为-{}-的事件,分析报告更新完成", eventAnalysisReport.getEventId());
+    }
+
+
+    /**
+     * 获取事件挖掘的结果
+     *
+     * @param record
+     * @author lkg
+     * @date 2024/8/30
+     */
+//    @KafkaListener(topics = {Constants.EVENT_DIG_RECEIVE_TOPIC})
+    public void eventDig(ConsumerRecord<String, String> record) {
+        String value = record.value();
+        try {
+            JSONObject jsonObject = JSONObject.parseObject(value);
+            JSONArray resultData = jsonObject.getJSONArray("resultData");
+            if (resultData != null && resultData.size() > 0) {
+                for (Object item : resultData) {
+                    JSONObject eventItem = JSONObject.parseObject(item.toString());
+                    String eventId = eventItem.get("event_id").toString();
+                    String eventTitle = eventItem.get("event_title").toString();
+                    String eventType = eventItem.get("event_type").toString();
+                    String eventDesc = eventItem.get("event_name").toString();
+                    JSONArray relatedIds = eventItem.getJSONArray("related_id");
+                    List<String> idList = JSONArray.parseArray(relatedIds.toString(), String.class);
+                    List<Map<String, Object>> maps = esService.listByIds(idList);
+                    String publishDate = maps.get(0).get("publishDate").toString();
+                    //todo 保存抽取到的事件， 若抽到python判断一样的事件，如何处理
+                    saveEventExtract(eventId, eventTitle, eventType, eventDesc, EsDateUtil.esFieldDateMapping(publishDate));
+                    log.info("挖掘到事件，id-{},名称-{}，并保存成功", eventId, eventTitle);
+                }
             }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    //保存关联资讯
+    private void saveData(String eventId,List<String> idList){
+        List<Map<String, Object>> maps = esService.listByIds(idList);
+        for (Map<String, Object> map : maps) {
+            String index = map.get("index").toString();
+            String id = map.get("id").toString();
+            String subjectId = map.get("subjectId").toString();
+            String newId = id.replace(subjectId, eventId);
+            map.put("id", newId);
+            map.put("subjectId", eventId);
+            map.remove("index");
+            esOpUtil.docSaveByMap(index, newId, map);
+        }
+    }
+
+    private void saveEventExtract(String eventId, String eventTitle, String eventType, String eventDesc, String publishDate) {
+        EventExtract byId = eventExtractService.getById(eventId);
+        if (byId == null) {
+            EventExtract eventExtract = new EventExtract();
+            eventExtract.setId(eventId);
+            eventExtract.setEventName(eventTitle);
+            String[] split = eventType.split("-");
+            LambdaQueryWrapper<EventCategory> queryWrapper = Wrappers.lambdaQuery();
+            queryWrapper.eq(EventCategory::getTypeName, split[split.length - 1]);
+            EventCategory eventCategory = eventCategoryService.getOne(queryWrapper);
+            eventExtract.setEventType(eventCategory.getId());
+            eventExtract.setEventDescribe(eventDesc);
+            Date startTime = DateUtil.stringToDate(publishDate, "yyyy-MM-dd HH:mm:ss");
+            eventExtract.setStartTime(DateUtil.addDate(startTime,-30));
+            eventExtract.setCreateTime(new Date());
+            eventExtract.setCreateBy("event-dig");
+            eventExtractService.save(eventExtract);
+        }
+    }
+
 
-    private boolean exist(SubjectAnalysis subjectAnalyse){
+    private boolean exist(SubjectAnalysis subjectAnalyse) {
         LambdaQueryWrapper<SubjectAnalysis> queryWrapper = Wrappers.lambdaQuery();
-        queryWrapper.eq(SubjectAnalysis::getCategory,2).eq(SubjectAnalysis::getDataId,subjectAnalyse.getDataId())
-                .eq(SubjectAnalysis::getSubjectId,subjectAnalyse.getSubjectId());
+        queryWrapper.eq(SubjectAnalysis::getCategory, 2).eq(SubjectAnalysis::getDataId, subjectAnalyse.getDataId())
+                .eq(SubjectAnalysis::getSubjectId, subjectAnalyse.getSubjectId());
         SubjectAnalysis one = subjectAnalysisService.getOne(queryWrapper);
         return one == null;
     }

src/main/java/com/zzsn/event/mapper/xml/EventMapper.xml

@@ -47,7 +47,7 @@
     </select>
 
     <select id="pageList" resultType="com.zzsn.event.vo.EventManageVO">
-        select t2.type_name,t1.id,t1.event_icon,t1.event_name,t1.event_label,
+        select t2.type_name,t1.id,t1.event_icon,t1.event_name,t1.event_label,t1.status,
         t1.face_public,t1.publish_status,t1.create_time,t1.publish_date,t1.start_time,t1.end_time
         from event t1
         inner join event_category t2 on t1.event_type =t2.id

src/main/java/com/zzsn/event/service/EventSimpleService.java

@@ -18,6 +18,7 @@ import org.springframework.transaction.annotation.Transactional;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * 简化版事件新增逻辑
@@ -59,6 +60,7 @@ public class EventSimpleService {
     public Event simpleSave(AddEventVO addEventVO) {
         Event event = new Event();
         BeanUtils.copyProperties(addEventVO, event);
+        event.setStatus(1);
         Date endTime = event.getEndTime();
         if (endTime == null) {
             endTime = DateUtil.addDate(new Date(), 15);
@@ -81,6 +83,7 @@ public class EventSimpleService {
         event.setCreateBy(currentUser.getUsername());
         eventService.save(event);
         String eventId = event.getId();
+        CompletableFuture.runAsync(()->{
             saveMapMain(eventId);
             //事件标签
             eventTagService.save(EventTag.builder().eventId(eventId).build());
@@ -89,6 +92,7 @@ public class EventSimpleService {
             addRegionMap(eventId, regionList);
             //默认绑定搜索引擎、模型
             eventService.saveDefaultMap(eventId);
+        });
         return event;
     }
 

src/main/java/com/zzsn/event/service/impl/EventServiceImpl.java

@@ -189,7 +189,9 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
         subjectCondition.setTypeIds(typeIds);
         //在根据所有明细节点查出专题列表
         List<EventNewPlatVO> pageList = baseMapper.newPlatPageList(subjectCondition, offset, pageSize);
+        long t1 = System.currentTimeMillis();
         count(pageList);
+        System.out.println("数量统计耗时====" + (System.currentTimeMillis() - t1));
         //获取总条数
         Integer count = baseMapper.newPlatCount(subjectCondition);
         IPage<EventNewPlatVO> pageData = new Page<>(pageNo, pageSize, count);
@@ -263,6 +265,7 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
         Event event = new Event();
         //事件专题的默认分析规则参数-必填
         BeanUtils.copyProperties(addEventParam, event);
+        event.setStatus(1);
         Date endTime = event.getEndTime();
         if (endTime == null) {
             endTime = DateUtil.addDate(new Date(), 15);
@@ -273,10 +276,13 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
         event.setTimeAnaRule(5);
         String subjectCode = codeGenerateUtil.geneCodeNo(CodePrefixEnum.SUBJECT_DEFAULT.getValue());
         event.setEventCode(subjectCode);
-        String cron = CronUtil.generateCron(event.getUnit(), event.getSpace());
+        event.setUnit("2");
+        event.setSpace(1);
+        String cron = CronUtil.generateCron("2", 1);
         event.setCron(cron);
         baseMapper.insert(event);
         String eventId = event.getId();
+        CompletableFuture.runAsync(()->{
             //插入专题-类别、项目的绑定关系
             saveMapMain(event, addEventParam);
             //事件标签
@@ -286,6 +292,7 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
             addRegionMap(eventId, regionList);
             //默认绑定搜索引擎、模型
             saveDefaultMap(eventId);
+        });
         return event;
     }
 
@@ -296,6 +303,7 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
         BeanUtils.copyProperties(addEventParam, subject);
         String eventId = subject.getId();
         baseMapper.updateById(subject);
+        CompletableFuture.runAsync(()->{
             //专题-类别、项目的绑定关系处理
             //删除专题-类别绑定关系
             iSubjectTypeMapService.deleteBySubjectId(subject.getId());
@@ -308,6 +316,7 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
             eventRegionMapService.remove(queryWrapper);
             List<RegionVO> regionList = addEventParam.getRegionList();
             addRegionMap(eventId, regionList);
+        });
     }
 
     @Override
@@ -559,12 +568,12 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
     @Override
     public void saveDefaultMap(String eventId){
         //默认引擎
-        List<String> defaultSearchEngines = Constants.DEFAULT_SEARCH_ENGINE;
+        Map<String, String> defaultSearchEngines = Constants.DEFAULT_SEARCH_ENGINE;
         List<SubjectSearchEnginesMap> enginesMaps = new ArrayList<>();
-        for (String defaultSearchEngine : defaultSearchEngines) {
+        for (Map.Entry<String, String> entry : defaultSearchEngines.entrySet()) {
             SubjectSearchEnginesMap enginesMap = new SubjectSearchEnginesMap();
             enginesMap.setSubjectId(eventId);
-            enginesMap.setSearchEngineId(defaultSearchEngine);
+            enginesMap.setSearchEngineId(entry.getKey());
             enginesMaps.add(enginesMap);
         }
         subjectSearchEnginesMapService.saveBatch(enginesMaps);
@@ -646,18 +655,12 @@ public class EventServiceImpl extends ServiceImpl<EventMapper, Event> implements
                 List<SubjectSourceVO> value = entry.getValue();
                 infoSourceNumMap.put(subjectId, value.size());
             }
-            List<Integer> checkStatus = new ArrayList<>();
-            checkStatus.add(0);
-            Map<String, Integer> unCheckCountMap = subjectInfoCountMap(idList, checkStatus);
             Map<String, Integer> subjectInfoCountMap = subjectInfoCountMap(idList, null);
-
             for (EventNewPlatVO newPlatVO : pageList) {
                 int keyWordsNum = null == keyWordsNumMap.get(newPlatVO.getId()) ? 0 : keyWordsNumMap.get(newPlatVO.getId());
                 int infoSourceNum = null == infoSourceNumMap.get(newPlatVO.getId()) ? 0 : infoSourceNumMap.get(newPlatVO.getId());
                 //查询每个专题现有的信息数量
                 int subjectInfoCount = null == subjectInfoCountMap.get(newPlatVO.getId()) ? 0 : subjectInfoCountMap.get(newPlatVO.getId());
-                int unCheckCount = null == unCheckCountMap.get(newPlatVO.getId()) ? 0 : unCheckCountMap.get(newPlatVO.getId());
-                newPlatVO.setUnCheckNum(unCheckCount);
                 newPlatVO.setSubjectInfoNum(subjectInfoCount);
                 newPlatVO.setInfoSourceNum(infoSourceNum);
                 newPlatVO.setKeyWordsNum(keyWordsNum);

src/main/java/com/zzsn/event/vo/EventManageVO.java

@@ -23,4 +23,5 @@ public class EventManageVO {
     private String publishDate;
     private String startTime;
     private String endTime;
+    private Integer status;
 }