人物服务2.0开发

src/main/java/com/zzsn/leaderbase/config/JobExecutor.java

@@ -16,8 +16,8 @@ public class JobExecutor {
     @Autowired
     DealLeaderDataService dealLeaderDataService;
 
-    @Async
-    @Scheduled(cron = "0 43 17 * * ? ")
+    //@Async
+    //@Scheduled(cron = "0 43 17 * * ? ")
     public void task(){
         log.info("------数据抽取定时任务启动------");
 

src/main/java/com/zzsn/leaderbase/constant/CommonConstant.java

@@ -2,19 +2,6 @@ package com.zzsn.leaderbase.constant;
 
 public interface CommonConstant {
 
-	/**
-	 * 正常状态:0  禁用状态:1
-	 */
-	public static final Integer STATUS_NORMAL = 0;
-	public static final Integer STATUS_DISABLE = -1;
-
-	/**
-	 * 删除标志  已删除：1  未删除：0
-	 */
-	public static final Integer DEL_FLAG_1 = 1;
-	public static final Integer DEL_FLAG_0 = 0;
-
-
 	/** {@code 500 Server Error} (HTTP/1.0 - RFC 1945) */
     public static final Integer SC_INTERNAL_SERVER_ERROR_500 = 500;
     /** {@code 200 OK} (HTTP/1.0 - RFC 1945) */
@@ -23,31 +10,5 @@ public interface CommonConstant {
     /**访问权限认证未通过 510*/
     public static final Integer SC_JEECG_NO_AUTHZ=510;
 
-    /** 登录用户Shiro权限缓存KEY前缀 */
-    public static String PREFIX_USER_SHIRO_CACHE  = "shiro:cache:org.jeecg.config.shiro.ShiroRealm.authorizationCache:";
-    /** 登录用户Token令牌缓存KEY前缀 */
-    public static final String PREFIX_USER_TOKEN  = "prefix_user_token_";
-    /** Token缓存时间：3600秒即一小时 */
-    public static final int  TOKEN_EXPIRE_TIME  = 3600;
-
-    /**发布状态（0未发布，1已发布，2已撤销）*/
-    public static final String NO_SEND  = "0";
-    public static final String HAS_SEND  = "1";
-    public static final String HAS_CANCLE  = "2";
-
-    /**阅读状态（0未读，1已读）*/
-    public static final String HAS_READ_FLAG  = "1";
-    public static final String NO_READ_FLAG  = "0";
-
-    /**
-     * 状态(0无效1有效)
-     */
-    public static final String STATUS_0 = "0";
-    public static final String STATUS_1 = "1";
-
-
-    //处理后的专题资讯信息存储索引。
-    public final static String ES_DATA_FOR_SUBJECT = "subjectdatabase";
-
     String LAST_TIME ="lastTime" ;
 }

src/main/java/com/zzsn/leaderbase/kafka/PersonOfJobChangeKafkaConsumer.java

+package com.zzsn.leaderbase.kafka;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.zzsn.leaderbase.entity.CharacterExtractInfo;
+import com.zzsn.leaderbase.service.CharacterExtractInfoService;
+import com.zzsn.leaderbase.service.IGeneratorIdService;
+import com.zzsn.leaderbase.util.EsUtil;
+import com.zzsn.leaderbase.util.HttpUtil;
+import com.zzsn.leaderbase.vo.BaseData;
+import com.zzsn.leaderbase.vo.InfoExtractionParam;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * 人物职务变更数据接收
+ */
+@Component
+@Slf4j
+public class PersonOfJobChangeKafkaConsumer {
+
+    @Value("${python.getInfoUrl}")
+    private String relationEntityUrl;
+
+    @Autowired
+    CharacterExtractInfoService characterExtractInfoService;
+
+    @Autowired
+    IGeneratorIdService generatorIdService;
+
+    @Autowired
+    private EsUtil esUtil;
+
+    private String USERSERVER_BASEDATA = "userserver_basedata";
+
+
+    @KafkaListener(topics = "personOfJobChangeTopic", groupId = "group02")
+    public void listen(String message) {
+        log.info("kafka pull personOfJobChangeTopic: {} ......", message.substring(0,300));
+        getData(message);
+    }
+
+    public void getData(String message) {
+        //log.info(message);
+        InfoExtractionParam record = JSON.parseObject(message, InfoExtractionParam.class);
+        BaseData baseData = JSON.parseObject(message, BaseData.class);
+        if(StringUtils.isEmpty(baseData.getId())){
+            String id = String.valueOf(generatorIdService.getId());
+            baseData.setId(id);
+        }
+
+        List<CharacterExtractInfo> list = new ArrayList<>();
+        Map<String, List<InfoExtractionParam>> map = new HashMap<>();
+        map.put("data_list", Collections.singletonList(record));
+        JSONObject jsonObjectParam = new JSONObject();
+        jsonObjectParam.put("data", map);
+        JSONArray dataResultList = null;
+        try {
+            log.info("request");
+            String result = HttpUtil.doPost(relationEntityUrl, jsonObjectParam, 40000);
+            log.info("response:{}", result);
+            Object objectResult = parseResult(result);
+            if (null != objectResult) {
+                dataResultList = (JSONArray) JSONArray.parse(objectResult.toString());
+            }
+        }catch (Exception e){
+            log.error("异常：{}",e.getMessage());
+        }
+        if (dataResultList != null && dataResultList.size() > 0) {
+            esUtil.docSaveByEntity(USERSERVER_BASEDATA, baseData.getId(), baseData);
+            for (Object dataResultObject : dataResultList) {
+                JSONObject dataResult = JSONObject.parseObject(dataResultObject.toString());
+                if (dataResult.get("infos") != null) {
+                    JSONArray jsonArray = (JSONArray) JSONArray.parse(String.valueOf(dataResult.get("infos")));
+                    for (Object data : jsonArray) {
+                        JSONObject temp = JSONObject.parseObject(data.toString());
+
+                        //判重
+                        String curStation = temp.get("curStation") != null ? temp.get("curStation").toString() : "";
+                        String curWorkPlace = temp.get("curWorkPlace") != null ? temp.get("curWorkPlace").toString() : "";
+                        String person = temp.get("person") != null ? temp.get("person").toString() : "";
+                        boolean b = characterExtractInfoService.deduplication(curStation,curWorkPlace,person);
+                        if(b){
+                            log.info("数据重复：{}",data.toString());
+                        } else {
+                            String id = Long.toString(generatorIdService.getOrderId());
+                            list.add(CharacterExtractInfo.builder()
+                                    .id(id)
+                                    .name(temp.get("person") != null ? temp.get("person").toString() : "")
+                                    .department(temp.get("curWorkPlace") != null ? temp.get("curWorkPlace").toString() : "")
+                                    .duty(temp.get("curStation") != null ? temp.get("curStation").toString() : "")
+                                    .departmentOriginal(temp.get("oriWorkPlace") != null ? temp.get("oriWorkPlace").toString() : "")
+                                    .dutyOriginal(temp.get("oriStation") != null ? temp.get("oriStation").toString() : "")
+                                    .updateDesc(temp.get("updateDesc") != null ? temp.get("updateDesc").toString() : "")
+                                    .articleIndex(USERSERVER_BASEDATA)
+                                    .articleId(baseData.getId())
+                                    .articleTitle(baseData.getTitle())
+                                    .articleOrigin(baseData.getOrigin())
+                                    .checkStatus(0)
+                                    .createTime(new Date())
+                                    .build()
+                            );
+                        }
+                    }
+                }
+            }
+        }
+        if (!list.isEmpty()) {
+            characterExtractInfoService.addBulk(list);
+        }
+    }
+
+    private Object parseResult(String result) {
+        JSONObject jsonObject = JSONObject.parseObject(result);
+        if (null != jsonObject && "true".equals(jsonObject.getString("isHandleSuccess"))) {
+            return jsonObject.get("resultData");
+
+        }
+        log.error("python 服务结果异常，响应信息{}", result);
+        return null;
+    }
+
+}
\ No newline at end of file

src/main/java/com/zzsn/leaderbase/service/DealLeaderDataService.java

@@ -55,7 +55,7 @@ public class DealLeaderDataService {
         } else {
             startTime = "2023-01-01 12:12:12";
         }
-        startTime = "2023-01-01 12:12:12";
+        //startTime = "2023-01-01 12:12:12";
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
         BoolQueryBuilder boolQueryBuilder1 = QueryBuilders.boolQuery()
                 .must(QueryBuilders.termsQuery("subjectId", Arrays.asList("1750716233967157250", "1750716067187048450")))

src/main/java/com/zzsn/leaderbase/service/EsServer.java

@@ -42,10 +42,15 @@ public class EsServer {
 
 
     public Result<?> getInformation(String index, String id){
+        if(index == null){
+            return Result.OK(null);
+        } else {
             SubjectInfoVo subjectInfo = esUtil.getInfoByid(index,id, SubjectInfoVo.class);
             return Result.OK(subjectInfo);
         }
 
+    }
+
     public Result<?> getCharacterBaike(String uid, String name){
 
         //参数校验

src/main/java/com/zzsn/leaderbase/service/impl/CharacterBasicInfoServiceImpl.java

@@ -54,26 +54,9 @@ public class CharacterBasicInfoServiceImpl extends ServiceImpl<CharacterBasicInf
         Integer total = baseMapper.getListCount(basicInfoListVo);
 
         //判断是否有关联数据
-        List<String> uidList = new ArrayList<>();
-        list.forEach(info -> { uidList.add(info.getUid()); });
-        QueryWrapper<CharacterBasicInfo> queryWrapper = new QueryWrapper<>();
-        queryWrapper.in("uid", uidList)
-                .ne("main_entry",1).select("id","uid");
-        List<CharacterBasicInfo> hasChildList = this.list(queryWrapper);
-        if(hasChildList.size() > 0){
-            Map<String, List<CharacterBasicInfo>> groupList = hasChildList.stream().collect(Collectors.groupingBy(e -> e.getUid()));
-            list.forEach(info -> {
-                String uid = info.getUid();
-                List<CharacterBasicInfo> groupUid = groupList.get(uid);
-                if(groupUid != null && groupUid.size() != 0){
-                    info.setHasChild(1);
-                }
-            });
-        }
-
+        hasChild(list);
 
         //记录热门标签
-        //记录热门标签
         String tagId = basicInfoListVo.getTagId();
         if(tagId != null && tagId.length() >0){
             String[] tagIdArr = tagId.split(",");
@@ -92,6 +75,28 @@ public class CharacterBasicInfoServiceImpl extends ServiceImpl<CharacterBasicInf
         return page;
 
     }
+    //判断是否有关联数据
+    private List<CharacterBasicInfo> hasChild(List<CharacterBasicInfo> list){
+        if(list != null && list.size() > 0 ){
+            List<String> uidList = new ArrayList<>();
+            list.forEach(info -> { uidList.add(info.getUid()); });
+            QueryWrapper<CharacterBasicInfo> queryWrapper = new QueryWrapper<>();
+            queryWrapper.in("uid", uidList)
+                    .ne("main_entry",1).select("id","uid");
+            List<CharacterBasicInfo> hasChildList = this.list(queryWrapper);
+            if(hasChildList.size() > 0){
+                Map<String, List<CharacterBasicInfo>> groupList = hasChildList.stream().collect(Collectors.groupingBy(e -> e.getUid()));
+                list.forEach(info -> {
+                    String uid = info.getUid();
+                    List<CharacterBasicInfo> groupUid = groupList.get(uid);
+                    if(groupUid != null && groupUid.size() != 0){
+                        info.setHasChild(1);
+                    }
+                });
+            }
+        }
+        return list;
+    }
 
 
     @Override

src/main/java/com/zzsn/leaderbase/vo/BaseData.java

+package com.zzsn.leaderbase.vo;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * kafka接收数据实体
+ */
+@Data
+public class BaseData implements Serializable {
+    //说明：...Raw 表示原文，即原语言
+    //作者
+    private String author;
+    private String authorRaw;
+    //正文
+    private String content;
+    private String contentRaw;
+    //带标签正文
+    private String contentWithTag;
+    private String contentWithTagRaw;
+    //入库时间
+    private String createDate;
+    //信息来源id(信息源或者关键词)
+    private String sid;
+    //语言
+    private String lang;
+    private String langRaw;
+    //来源（信息来源）
+    private String origin;
+    private String originRaw;
+    //发布时间
+    private String publishDate;
+    //发布地址
+    private String sourceAddress;
+    //摘要
+    private String summary;
+    private String summaryRaw;
+    //关键词
+    private String keyWords;
+    //标题
+    private String title;
+    private String titleRaw;
+    //采集来源（如通用、定制、微信公众号等）
+    private String source;
+    //附加字段
+    private String type;
+
+    //视频下载链接
+    private String downLoadUrl;
+    //视频链接（原链接 网页版）
+    private String videoUrl;
+    //视频链接（原链接 手机版）
+    private String videoPhoneUrl;
+    //视频时长单位秒
+    private Long videoTime;
+    //视频第一帧图片
+    private String videoImg;
+    //重复id
+    private String repeatId;
+    private String subjectId;
+    //快照地址
+    private String screenShotImg;
+    //id
+    private String id;
+    //信息类别（1：报刊  2：博客  3：客户端  4：论坛  5：视频  6：外媒  7：网站  8：微博  9：微信  10：新闻  11：政务  12：其它）
+    private String infoSourceType;
+
+    //资讯关联的附件id
+    private List<String> attachmentIds;
+
+}

src/main/resources/logback-spring.xml

@@ -20,7 +20,7 @@
 			<!--日志文件输出的文件名 -->
 			<FileNamePattern>${LOG_HOME}/leader-%d{yyyy-MM-dd}.%i.log</FileNamePattern>
 			<!--日志文件保留天数 -->
-			<MaxHistory>30</MaxHistory>
+			<MaxHistory>15</MaxHistory>
 			<maxFileSize>10MB</maxFileSize>
 		</rollingPolicy>
 		<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">