feat:邮件解析-提供解析指定文件接口

service-base/src/main/java/com/simuwang/base/config/ThreadPoolConfig.java

@@ -15,7 +15,7 @@ public class ThreadPoolConfig {
     public ThreadPoolTaskExecutor valuationExecutor() {
         ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
         int cpuSize = Runtime.getRuntime().availableProcessors();
-        cpuSize = Math.max(cpuSize, 1);
+        cpuSize = Math.max(cpuSize/2, 1);
 
         taskExecutor.setCorePoolSize(cpuSize);
         taskExecutor.setMaxPoolSize(50);
@@ -32,4 +32,21 @@ public class ThreadPoolConfig {
         return taskExecutor;
     }
 
+    @Bean("asyncExecutor")
+    public ThreadPoolTaskExecutor asyncExecutor() {
+        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
+        taskExecutor.setCorePoolSize(2);
+        taskExecutor.setMaxPoolSize(50);
+        taskExecutor.setQueueCapacity(50);
+        taskExecutor.setKeepAliveSeconds(60);
+        taskExecutor.setThreadNamePrefix("asyncExecutor--");
+        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
+        taskExecutor.setAwaitTerminationSeconds(60);
+
+        // 修改拒绝策略为使用当前线程执行
+        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        // 初始化线程池
+        taskExecutor.initialize();
+        return taskExecutor;
+    }
 }

service-base/src/main/java/com/simuwang/base/mapper/EmailParseInfoMapper.java

@@ -1,6 +1,7 @@
 package com.simuwang.base.mapper;
 
 import com.simuwang.base.pojo.dos.EmailParseInfoDO;
+import com.simuwang.base.pojo.dto.EmailInfoDTO;
 import com.simuwang.base.pojo.dto.query.EmailParseQuery;
 import com.simuwang.base.pojo.vo.EmailParseInfoVO;
 import org.apache.ibatis.annotations.Mapper;
@@ -27,4 +28,6 @@ public interface EmailParseInfoMapper {
     long countEmailList(EmailParseQuery emailParseQuery);
 
     void updateParseTime(@Param("id") Integer id, @Param("parseDate") Date parseDate);
+
+    List<EmailInfoDTO> queryValuationEmailByFileId(@Param("fileIdList") List<Integer> fileIdList);
 }

service-base/src/main/java/com/simuwang/base/pojo/dto/EmailInfoDTO.java

@@ -0,0 +1,46 @@
+package com.simuwang.base.pojo.dto;
+
+import lombok.Data;
+
+import java.util.Date;
+
+@Data
+public class EmailInfoDTO {
+    /**
+     * 邮件id
+     */
+    private Integer id;
+    /**
+     * 文件id
+     */
+    private Integer fileId;
+    /**
+     * 邮箱地址
+     */
+    private String email;
+    /**
+     * 邮箱日期
+     */
+    private Date emailDate;
+    /**
+     * 邮件主题
+     */
+    private String emailTitle;
+    /**
+     * 邮件类型，1-净值，2-估值表，3-定期报告
+     */
+    private Integer emailType;
+    /**
+     * 解析状态
+     */
+    private Integer parseStatus;
+    /**
+     * 文件名称
+     */
+    private String fileName;
+
+    /**
+     * 文件地址
+     */
+    private String filePath;
+}

service-base/src/main/resources/mapper/EmailParseInfoMapper.xml

@@ -124,5 +124,23 @@
           and id = #{id}
     </update>
 
+    <select id="queryValuationEmailByFileId" resultType="com.simuwang.base.pojo.dto.EmailInfoDTO">
+        select t1.id as id,
+               t1.email as email,
+               t1.email_date as emailDate,
+               t1.email_title as emailTitle,
+               t1.email_type as emailType,
+               t2.id as fileId,
+               t2.file_name as fileName,
+               t2.file_path as filePath
+        from PPW_EMAIL.email_parse_info t1
+        join PPW_EMAIL.email_file_info t2 on t1.id = t2.email_id and t2.isvalid = 1
+        where t1.isvalid = 1
+          and t1.email_type = 2
+          and t2.id in
+        <foreach collection="fileIdList" item="fileId" index="index" open="(" separator="," close=")">
+        #{fileId}
+        </foreach>
+    </select>
 
 </mapper>

service-daq/src/main/java/com/simuwang/daq/service/EmailParseApiService.java

@@ -3,6 +3,7 @@ package com.simuwang.daq.service;
 import com.simuwang.base.pojo.dto.MailboxInfoDTO;
 
 import java.util.Date;
+import java.util.List;
 
 /**
  * @author mozuwen
@@ -27,4 +28,12 @@ public interface EmailParseApiService {
      */
     void reparseEmail(Integer emailId);
 
+
+    /**
+     * 重新解析指定估值表文件
+     *
+     * @param fileIdList 文件id列表
+     */
+    void reparseFile(List<Integer> fileIdList);
+
 }

service-daq/src/main/java/com/simuwang/daq/service/EmailParseApiServiceImpl.java

@@ -13,9 +13,12 @@ import com.simuwang.base.pojo.dos.EmailFileInfoDO;
 import com.simuwang.base.pojo.dos.EmailParseInfoDO;
 import com.simuwang.base.pojo.dto.EmailContentInfoDTO;
 import com.simuwang.base.pojo.dto.EmailFundNavDTO;
+import com.simuwang.base.pojo.dto.EmailInfoDTO;
 import com.simuwang.base.pojo.dto.MailboxInfoDTO;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 
 import java.io.BufferedReader;
@@ -24,6 +27,7 @@ import java.io.IOException;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * @author mozuwen
@@ -38,12 +42,14 @@ public class EmailParseApiServiceImpl implements EmailParseApiService {
     private final EmailParseService emailParseService;
     private final EmailParseInfoMapper emailParseInfoMapper;
     private final EmailFileInfoMapper emailFileInfoMapper;
+    private final ThreadPoolTaskExecutor asyncExecutor;
 
     public EmailParseApiServiceImpl(EmailParseService emailParseService, EmailParseInfoMapper emailParseInfoMapper,
-                                    EmailFileInfoMapper emailFileInfoMapper) {
+                                    EmailFileInfoMapper emailFileInfoMapper, @Qualifier("asyncExecutor") ThreadPoolTaskExecutor asyncExecutor) {
         this.emailParseService = emailParseService;
         this.emailParseInfoMapper = emailParseInfoMapper;
         this.emailFileInfoMapper = emailFileInfoMapper;
+        this.asyncExecutor = asyncExecutor;
     }
 
     @Override
@@ -74,24 +80,78 @@ public class EmailParseApiServiceImpl implements EmailParseApiService {
 
         List<EmailFundNavDTO> emailFundNavDTOList = CollUtil.newArrayList();
         Map<EmailContentInfoDTO, List<EmailFundNavDTO>> fileNameNavMap = MapUtil.newHashMap();
-        for (EmailContentInfoDTO emailContentInfoDTO : emailContentInfoDTOList) {
-            try {
-                List<EmailFundNavDTO> fundNavDTOList = emailParseService.parseEmail(emailContentInfoDTO, emailFieldMap);
-                fileNameNavMap.put(emailContentInfoDTO, fundNavDTOList);
-                emailFundNavDTOList.addAll(fundNavDTOList);
-            } catch (Exception e) {
-                log.error("重新解析邮件失败,邮件id:{},堆栈信息:{}", emailId, ExceptionUtil.stacktraceToString(e));
+        asyncExecutor.execute(() -> {
+            for (EmailContentInfoDTO emailContentInfoDTO : emailContentInfoDTOList) {
+                try {
+                    List<EmailFundNavDTO> fundNavDTOList = emailParseService.parseEmail(emailContentInfoDTO, emailFieldMap);
+                    fileNameNavMap.put(emailContentInfoDTO, fundNavDTOList);
+                    emailFundNavDTOList.addAll(fundNavDTOList);
+                } catch (Exception e) {
+                    log.error("重新解析邮件失败,邮件id:{},堆栈信息:{}", emailId, ExceptionUtil.stacktraceToString(e));
+                }
             }
+            // 保存相关信息 -> 邮件信息表,邮件文件表,邮件净值表,邮件规模表,基金净值表
+            emailParseService.saveRelatedTable(emailParseInfoDO.getEmail(), emailContentInfoDTOList, fileNameNavMap);
+        });
+    }
+
+    @Override
+    public void reparseFile(List<Integer> fileIdList) {
+        if (CollUtil.isEmpty(fileIdList)) {
+            return;
+        }
+        List<EmailInfoDTO> emailParseInfoDOList = emailParseInfoMapper.queryValuationEmailByFileId(fileIdList);
+        if (CollUtil.isEmpty(emailParseInfoDOList)) {
+            return;
         }
-        // 保存相关信息 -> 邮件信息表,邮件文件表,邮件净值表,邮件规模表,基金净值表
-        emailParseService.saveRelatedTable(emailParseInfoDO.getEmail(), emailContentInfoDTOList, fileNameNavMap);
+        asyncExecutor.execute(() -> {
+            Map<Integer, List<EmailInfoDTO>> emailIdFileMap = emailParseInfoDOList.stream().collect(Collectors.groupingBy(EmailInfoDTO::getId));
+            for (Map.Entry<Integer, List<EmailInfoDTO>> entry : emailIdFileMap.entrySet()) {
+                Integer emailId = entry.getKey();
+                List<EmailInfoDTO> emailInfoDTOList = entry.getValue();
+                String emailAddress = emailInfoDTOList.get(0).getEmail();
+                List<EmailContentInfoDTO> emailContentInfoDTOList = emailInfoDTOList.stream().map(this::buildEmailContentInfoDTO).collect(Collectors.toList());
+
+                List<EmailFundNavDTO> emailFundNavDTOList = CollUtil.newArrayList();
+                Map<EmailContentInfoDTO, List<EmailFundNavDTO>> fileNameNavMap = MapUtil.newHashMap();
+                for (EmailContentInfoDTO emailContentInfoDTO : emailContentInfoDTOList) {
+                    try {
+                        log.info("开始重新解析文件 -> 文件id:{}", emailContentInfoDTO.getFielId());
+                        List<EmailFundNavDTO> fundNavDTOList = emailParseService.parseEmail(emailContentInfoDTO, MapUtil.newHashMap());
+                        fileNameNavMap.put(emailContentInfoDTO, fundNavDTOList);
+                        emailFundNavDTOList.addAll(fundNavDTOList);
+                    } catch (Exception e) {
+                        log.error("重新解析文件失败,邮件id:{},文件id:{},堆栈信息:{}", emailId, emailContentInfoDTO.getFielId(), ExceptionUtil.stacktraceToString(e));
+                    }
+                }
+                // 保存相关信息 -> 邮件信息表,邮件文件表,邮件净值表,邮件规模表,基金净值表
+                emailParseService.saveRelatedTable(emailAddress, emailContentInfoDTOList, fileNameNavMap);
+            }
+            log.info("重新解析文件结束... -> 文件id:{}", fileIdList);
+        });
+    }
+
+    private EmailContentInfoDTO buildEmailContentInfoDTO(EmailInfoDTO emailInfoDTO) {
+        String emailDate = DateUtil.format(emailInfoDTO.getEmailDate(), DateConst.YYYY_MM_DD_HH_MM_SS);
+        String parseDate = DateUtil.format(new Date(), DateConst.YYYY_MM_DD_HH_MM_SS);
+        EmailContentInfoDTO contentInfoDTO = new EmailContentInfoDTO();
+        contentInfoDTO.setEmailId(emailInfoDTO.getId());
+        contentInfoDTO.setFielId(emailInfoDTO.getFileId());
+        contentInfoDTO.setEmailAddress(emailInfoDTO.getEmail());
+        contentInfoDTO.setEmailDate(emailDate);
+        contentInfoDTO.setEmailTitle(emailInfoDTO.getEmailTitle());
+        contentInfoDTO.setParseDate(parseDate);
+        contentInfoDTO.setFileName(emailInfoDTO.getFileName());
+        contentInfoDTO.setFilePath(emailInfoDTO.getFilePath());
+        contentInfoDTO.setEmailType(emailInfoDTO.getEmailType());
+        return contentInfoDTO;
     }
 
-    private List<EmailContentInfoDTO> buildEmailContentInfoDTO(Integer emailId, EmailParseInfoDO emailParseInfoDO, List<EmailFileInfoDO> emailParseInfoDOList, Map<Integer, List<String>> emailTypeMap) {
+    private List<EmailContentInfoDTO> buildEmailContentInfoDTO(Integer emailId, EmailParseInfoDO emailParseInfoDO, List<EmailFileInfoDO> emailFileInfoDOList, Map<Integer, List<String>> emailTypeMap) {
         List<EmailContentInfoDTO> emailContentInfoDTOList = CollUtil.newArrayList();
         String emailDate = DateUtil.format(emailParseInfoDO.getEmailDate(), DateConst.YYYY_MM_DD_HH_MM_SS);
         String parseDate = DateUtil.format(new Date(), DateConst.YYYY_MM_DD_HH_MM_SS);
-        for (EmailFileInfoDO fileInfoDO : emailParseInfoDOList) {
+        for (EmailFileInfoDO fileInfoDO : emailFileInfoDOList) {
             EmailContentInfoDTO contentInfoDTO = new EmailContentInfoDTO();
             contentInfoDTO.setEmailId(emailId);
             contentInfoDTO.setFielId(fileInfoDO.getId());
@@ -103,14 +163,14 @@ public class EmailParseApiServiceImpl implements EmailParseApiService {
             contentInfoDTO.setFilePath(fileInfoDO.getFilePath());
             Integer emailType = EmailUtil.getEmailTypeBySubject(emailParseInfoDO.getEmailTitle(), emailTypeMap);
             contentInfoDTO.setEmailType(emailType);
-            String emailContent = getEmailContent(fileInfoDO.getFilePath());
+            String emailContent = readHtmlFileContent(fileInfoDO.getFilePath());
             contentInfoDTO.setEmailContent(emailContent);
             emailContentInfoDTOList.add(contentInfoDTO);
         }
         return emailContentInfoDTOList;
     }
 
-    public static String getEmailContent(String filePath) {
+    public static String readHtmlFileContent(String filePath) {
         if (StrUtil.isNotBlank(filePath) && filePath.endsWith("html")) {
             StringBuilder content = new StringBuilder();
             try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {

service-deploy/src/main/test/java/com/simuwang/datadaq/DataTrusteeApplicationTests.java

@@ -1,5 +1,6 @@
 package com.simuwang.datadaq;
 
+import cn.hutool.core.collection.ListUtil;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.map.MapUtil;
 import com.simuwang.base.common.conts.DateConst;
@@ -49,4 +50,10 @@ class DataTrusteeApplicationTests {
     public void testReparseEmail() {
         emailParseApiService.reparseEmail(7);
     }
+
+
+    @Test
+    public void testReparseFile() {
+        emailParseApiService.reparseFile(ListUtil.toList(40,43));
+    }
 }