自有数据

qhtx-eta-integrator/qhtx-integrator-application/src/main/java/com/qhtx/eta/controller/ETAController.java

@@ -5,6 +5,7 @@ import com.qhtx.eta.controller.request.ETAIndexPushRequest;
 import com.qhtx.eta.domain.facade.ETAFacadeService;
 import com.qhtx.eta.domain.service.DWIndexFrameService;
 import com.qhtx.eta.infra.entity.dw.DWIndex;
+import com.qhtx.eta.vo.Result;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.Resource;
@@ -18,8 +19,8 @@ public class ETAController {
 
 
     @PostMapping("/push")
-    public String getIndex(@RequestBody ETAIndexPushRequest etaIndexPushRequest) {
+    public Result<String> getIndex(@RequestBody ETAIndexPushRequest etaIndexPushRequest) {
         etaFacadeService.pushIndex(etaIndexPushRequest.getIndexCode());
-        return "";
+        return Result.success("同步成功");
     }
 }

qhtx-eta-integrator/qhtx-integrator-common/src/main/java/com/qhtx/eta/common/eunms/ErrorEnum.java

@@ -9,6 +9,7 @@ public enum ErrorEnum {
      * SYSTEM INNER ERROR CODE
      */
     BEAN_COVERT_TO_MAP_ERROR("00001", "bean转换成map失败"),
+    PARSE_TIME_ERROR("00002","解析时间错误"),
     /**
      * SYSTEM ERROR CODE
      */

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/api/http/request/ETAPushIndexRequest.java

@@ -1,9 +1,11 @@
 package com.qhtx.eta.domain.api.http.request;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.qhtx.eta.domain.api.http.response.eta.ETAPushData;
 import lombok.Data;
 
 import java.io.Serializable;
+import java.util.List;
 
 @Data
 public class ETAPushIndexRequest implements Serializable {
@@ -19,7 +21,7 @@ public class ETAPushIndexRequest implements Serializable {
     private String frequency;
     @JsonProperty("SourceName")
     private String sourceName;
-
-    
+    @JsonProperty("DataList")
+    private List<ETAPushData> dataList;
 }
 

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/api/http/response/eta/ETAPushData.java

@@ -7,15 +7,10 @@ import java.io.Serializable;
 
 @Data
 public class ETAPushData implements Serializable {
-    private static final long serialVersionUID = 4376692558237866233L;
-    @JsonProperty("IndexCode")
-    private String indexCode;
-    @JsonProperty("IndexName")
-    private String indexName;
-    @JsonProperty("Unit")
-    private String unit;
-    @JsonProperty("Frequency")
-    private String frequency;
-    @JsonProperty("SourceName")
-    private String sourceName;
+    private static final long serialVersionUID = -2579452224351461118L;
+    @JsonProperty("Date")
+    private String date;
+    @JsonProperty("Value")
+    private String value;
+
 }

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/api/http/response/eta/ETAPushDataResponse.java

@@ -1,7 +1,5 @@
 package com.qhtx.eta.domain.api.http.response.eta;
 
-import com.qhtx.eta.domain.entity.ETAPushIndexDTO;
-
 import java.util.List;
 
 public class ETAPushDataResponse extends ETAHttpResponse<ETAPushData, ETAPushIndexDTO> {

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/api/proxy/impl/ETAPushParameterProxy.java

@@ -0,0 +1,20 @@
+package com.qhtx.eta.domain.api.proxy.impl;
+
+import com.qhtx.eta.common.eunms.ErrorEnum;
+import com.qhtx.eta.common.exception.ETAException;
+import com.qhtx.eta.domain.api.http.request.ETADataHttpRequest;
+import com.qhtx.eta.domain.api.http.request.ETAPushIndexRequest;
+import com.qhtx.eta.domain.api.proxy.ApiParameterProxy;
+
+/**
+ * @author mac
+ */
+public class ETAPushParameterProxy extends ApiParameterProxy<ETAPushIndexRequest> {
+    @Override
+    protected ETAPushIndexRequest acceptParam(Object data) {
+        if (data instanceof ETADataHttpRequest) {
+            return (ETAPushIndexRequest) data;
+        }
+        throw new ETAException(ErrorEnum.PARAM_TYPE_ERROR);
+    }
+}

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/convert/DWIndexDTOConverter.java

@@ -6,7 +6,11 @@ import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
 import org.mapstruct.Mapper;
 import org.mapstruct.factory.Mappers;
 
+import java.util.List;
+
+
 @Mapper
+
 public interface DWIndexDTOConverter {
 
     DWIndexDTOConverter INSTANCE = Mappers.getMapper(DWIndexDTOConverter.class);
@@ -14,4 +18,6 @@ public interface DWIndexDTOConverter {
     DWIndexDTO convertToDTO(DWIndex dwIndex);
 
     TEtaIndexPushTask convertToTEtaIndexPushTask(DWIndexDTO dwIndexDTO);
+
+    List<TEtaIndexPushTask> convertToTEtaIndexPushTaskList(List<DWIndexDTO> dwIndexDTO);
 }

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/convert/DWIndexDataDTOConverter.java

@@ -0,0 +1,19 @@
+package com.qhtx.eta.domain.convert;
+
+import com.qhtx.eta.domain.entity.DWIndexDTO;
+import com.qhtx.eta.domain.entity.DWIndexDataDTO;
+import com.qhtx.eta.infra.entity.dw.DWIndex;
+import com.qhtx.eta.infra.entity.dw.TDampDwIndexData;
+import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.factory.Mappers;
+
+@Mapper
+public interface DWIndexDataDTOConverter {
+
+    DWIndexDataDTOConverter INSTANCE = Mappers.getMapper(DWIndexDataDTOConverter.class);
+
+    @Mapping(source = "dataDate",target = "date")
+    DWIndexDataDTO convertToDTO(TDampDwIndexData dwIndexData);
+}

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/entity/DWIndexDataDTO.java

@@ -4,11 +4,12 @@ import lombok.Data;
 
 import java.io.Serializable;
 import java.math.BigDecimal;
+import java.util.Date;
 
 @Data
 public class DWIndexDataDTO implements Serializable {
 
     private static final long serialVersionUID = -2967752044467244518L;
-    private String date;
+    private Date date;
     private BigDecimal value;
 }

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/entity/ETAPushIndexDTO.java

@@ -1,17 +0,0 @@
-package com.qhtx.eta.domain.entity;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import lombok.Data;
-
-import java.io.Serializable;
-
-@Data
-public class ETAPushIndexDTO implements Serializable {
-    private static final long serialVersionUID = -5650101144913878139L;
-    private String indexCode;
-    private String indexName;
-    private String unit;
-    private String frequency;
-    private String sourceName;
-}
-

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/enums/ETAInterfaceEnum.java

@@ -4,6 +4,7 @@ import com.qhtx.eta.domain.api.http.response.eta.*;
 import com.qhtx.eta.domain.api.proxy.ApiParameterProxy;
 import com.qhtx.eta.domain.api.proxy.impl.ETADataParameterProxy;
 import com.qhtx.eta.common.eunms.HttpMethod;
+import com.qhtx.eta.domain.api.proxy.impl.ETAPushParameterProxy;
 import lombok.Getter;
 
 @Getter
@@ -18,7 +19,7 @@ public enum ETAInterfaceEnum {
     // GET_EDB_DETAIL("getEdbDetail", "/v1/edb/detail", HttpMethod.GET, "指标详情", EdbDetailProxy.class, false),
     GET_EDB_DATA("getEdbData", "/v1/edb/data", HttpMethod.GET, "指标数据", ETADataParameterProxy.class, true, QuotaDataResponse.class),
     GET_CLASSIFY_LIST("getClassifyList", "/v1/edb/classify/list", HttpMethod.GET, "指标分类列表", null, true, ClassifyDataResponse.class),
-    PUSH_ETA_INDEX("pushEtaIndex", "/v1/edb/push", HttpMethod.POST, "外部数据写入ETA数据源(自由数据)", null, true, ETAPushDataResponse.class),
+    PUSH_ETA_INDEX("pushEtaIndex", "/v1/edb/push", HttpMethod.POST, "外部数据写入ETA数据源(自由数据)", ETAPushParameterProxy.class, true, ETAPushDataResponse.class),
 
     ;
 

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/enums/ETATaskEnum.java

@@ -7,6 +7,7 @@ public enum ETATaskEnum {
     SYNC_ETA_INDEX_DATA("data", "syncETAData"),
     SYNC_ETA_INDEX_INFO("info", "syncETAInfo"),
     SYNC_INDEX_CLASSIFY_LIST("classify", "syncETAClassifyList"),
+    SYNC_ETA_PUSH_DATA("push", "syncETAPushData"),
     ;
 
 

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/facade/impl/ETAFacadeServiceImpl.java

@@ -1,31 +1,72 @@
 package com.qhtx.eta.domain.facade.impl;
 
+import com.google.common.collect.Lists;
 import com.qhtx.eta.common.annotation.Facade;
+import com.qhtx.eta.common.constant.ETAConstants;
 import com.qhtx.eta.common.eunms.DataSourceType;
 import com.qhtx.eta.common.eunms.ErrorEnum;
 import com.qhtx.eta.common.exception.ETAException;
 import com.qhtx.eta.domain.annotation.DomainTransDataSource;
+import com.qhtx.eta.domain.api.ApiServiceHolder;
 import com.qhtx.eta.domain.entity.DWIndexDTO;
+import com.qhtx.eta.domain.entity.DWIndexDataDTO;
+import com.qhtx.eta.domain.enums.ETAInterfaceEnum;
 import com.qhtx.eta.domain.facade.ETAFacadeService;
 import com.qhtx.eta.domain.service.DWIndexFrameService;
+import com.qhtx.eta.domain.service.ETAPushDataService;
+import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
+import com.qhtx.eta.infra.service.TEtaIndexPushTaskService;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.List;
+import java.util.stream.Collectors;
 
+/**
+ * @author mac
+ */
 @Service
 @Facade
+@Slf4j
 public class ETAFacadeServiceImpl implements ETAFacadeService {
 
     @Resource
     private DWIndexFrameService dwIndexFrameService;
 
+    @Resource
+    private ETAPushDataService etaPushDataService;
+    @Resource
+    private ApiServiceHolder apiServiceHolder;
 
     @Override
     @DomainTransDataSource(dataSourceType = DataSourceType.DW)
     public void pushIndex(String indexCode) {
         DWIndexDTO dwIndexDTO = dwIndexFrameService.pushIndexToEta(indexCode);
-        if (dwIndexDTO == null) {
-            throw new ETAException(ErrorEnum.DW_INDEX_NOT_FOUND);
+        if (dwIndexDTO.getDataList().size() > ETAConstants.ETA_PUSH_DATA_LIMIT) {
+            log.info("数据量超过限制，分批推送");
+            List<DWIndexDataDTO> sortList = dwIndexDTO.getDataList().stream().sorted(Comparator.comparing(DWIndexDataDTO::getDate)).collect(Collectors.toList());
+            Lists.partition(sortList, ETAConstants.ETA_PUSH_DATA_LIMIT).forEach(dataList -> {
+                Date latestDataDate = dataList.stream().map(item -> item.getDate()).max(Comparator.comparing(Date::getTime)).get();
+                //防止数据漏更新，先推送，再更新本地任务记录，下游做好冗余和幂等
+                DWIndexDTO dataDTO = new DWIndexDTO();
+                dataDTO.setIndexCode(dwIndexDTO.getIndexCode());
+                dataDTO.setFrequency(dwIndexDTO.getFrequency());
+                dataDTO.setIndexName(dwIndexDTO.getIndexName());
+                dataDTO.setRemark(dwIndexDTO.getRemark());
+                dataDTO.setSourceName(dwIndexDTO.getSourceName());
+                dataDTO.setDataList(dataList);
+                apiServiceHolder.runApi(ETAInterfaceEnum.PUSH_ETA_INDEX, dataDTO);
+                TEtaIndexPushTask tEtaIndexPushTask = etaPushDataService.getETAPushTaskByIndexCode(dataDTO.getIndexCode());
+                tEtaIndexPushTask.setLatestDataDate(new SimpleDateFormat(ETAConstants.DATE_PATTERN).format(latestDataDate));
+                tEtaIndexPushTask.setDataCount(dataList.size());
+                dataDTO.setDataList(dataList);
+            });
         }
+        apiServiceHolder.runApi(ETAInterfaceEnum.PUSH_ETA_INDEX, dwIndexDTO);
     }
 }

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/job/ETASyncJob.java

@@ -4,23 +4,22 @@ import com.qhtx.eta.common.eunms.DataSourceType;
 import com.qhtx.eta.domain.config.ScheduleConfig;
 import com.qhtx.eta.domain.config.ScheduleTaskConfig;
 import com.qhtx.eta.domain.convert.ClassifyMappingDTOConverter;
-import com.qhtx.eta.domain.entity.ClassifyMappingDTO;
-import com.qhtx.eta.domain.entity.QuotaDataUpdateBO;
-import com.qhtx.eta.domain.entity.QuotaInfoDTO;
+import com.qhtx.eta.domain.entity.*;
 import com.qhtx.eta.domain.enums.ETADataStatus;
 import com.qhtx.eta.domain.enums.ETAInterfaceEnum;
 import com.qhtx.eta.domain.enums.ETATaskEnum;
 import com.qhtx.eta.domain.enums.ExcludedQuotaSource;
 import com.qhtx.eta.domain.service.DWIndexFrameService;
 import com.qhtx.eta.domain.service.ETAClassifyService;
+import com.qhtx.eta.domain.service.ETAPushDataService;
 import com.qhtx.eta.domain.service.ETAQuotaService;
 import com.qhtx.eta.domain.api.ApiServiceHolder;
 import com.qhtx.eta.domain.api.http.request.ETADataHttpRequest;
 import com.qhtx.eta.common.constant.ETAConstants;
-import com.qhtx.eta.domain.entity.QuotaDataDTO;
 import com.qhtx.eta.domain.task.FixedDelayTaskRegistrar;
 import com.qhtx.eta.domain.task.SchedulingRunnable;
 import com.qhtx.eta.infra.annotation.UseDataSource;
+import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
 import com.qhtx.eta.infra.enums.Frequency;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
@@ -69,6 +68,8 @@ public class ETASyncJob implements CommandLineRunner {
     private DWIndexFrameService dwIndexFrameService;
 
 
+    @Resource
+    private ETAPushDataService etaPushDataService;
 //    @Resource(name = "ETAIndexFrameThreadPool")
 //    private ThreadPoolExecutor etaIndexFrameThreadPool;
 
@@ -78,6 +79,13 @@ public class ETASyncJob implements CommandLineRunner {
         return taskMappingMap.get(configId);
     }
 
+    /**
+     * 同步指标数据到自有数据源
+     */
+    public void syncETAPushData(){
+        //获取需要更新的数据
+      //  List<ETAPushIndexDTO> pushTaskList = etaPushDataService.getPushIndexTaskList();
+    }
     /**
      * 同步指标信息
      */
@@ -94,6 +102,9 @@ public class ETASyncJob implements CommandLineRunner {
         }
     }
 
+    /**
+     *   同步ETA的指标框架到一期
+     */
     public void syncETAClassifyList() {
         List<?> respList = apiServiceHolder.runApi(ETAInterfaceEnum.GET_CLASSIFY_LIST);
         if (respList != null) {
@@ -109,7 +120,6 @@ public class ETASyncJob implements CommandLineRunner {
                     dwIndexFrameService.processing(ClassifyMappingDTOConverter.INSTANCE.convertDTOToDWIndexFrameList(processingList));
                 }
                 //开始获取新增和需要更新的框架数据
-
             } catch (Exception e) {
                 e.printStackTrace();
                 log.warn("同步eta指标分类失败:{}", e.getMessage());

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/service/ETAPushDataService.java

@@ -0,0 +1,12 @@
+package com.qhtx.eta.domain.service;
+
+import com.qhtx.eta.domain.entity.DWIndexDTO;
+import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
+
+import java.util.List;
+
+public interface ETAPushDataService {
+    List<DWIndexDTO> getPushIndexTaskList();
+
+    TEtaIndexPushTask getETAPushTaskByIndexCode(String indexCode);
+}

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/service/impl/DWIndexFrameServiceImpl.java

@@ -6,9 +6,12 @@ import com.qhtx.eta.common.constant.ETAConstants;
 import com.qhtx.eta.common.eunms.ErrorEnum;
 import com.qhtx.eta.common.exception.ETAException;
 import com.qhtx.eta.domain.annotation.DomainTransDataSource;
+import com.qhtx.eta.domain.api.http.response.eta.ETAPushData;
 import com.qhtx.eta.domain.convert.DWIndexDTOConverter;
+import com.qhtx.eta.domain.convert.DWIndexDataDTOConverter;
 import com.qhtx.eta.domain.convert.DWIndexFrameDTOConverter;
 import com.qhtx.eta.domain.entity.DWIndexDTO;
+import com.qhtx.eta.domain.entity.DWIndexDataDTO;
 import com.qhtx.eta.domain.entity.DWIndexFrameDTO;
 import com.qhtx.eta.domain.service.DWIndexFrameService;
 import com.qhtx.eta.domain.utils.RedisUtils;
@@ -29,10 +32,8 @@ import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.UUID;
+import java.text.SimpleDateFormat;
+import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.stream.Collectors;
@@ -166,13 +167,13 @@ public class DWIndexFrameServiceImpl implements DWIndexFrameService {
                 log.error("数据推送异常", e);
             }
         } else {
-            return DWIndexDTOConverter.INSTANCE.convertToDTO(dwIndex);
+            allResults = tDampDwIndexDataService.queryAllByLimit(indexCode, 0, total);
         }
-        System.out.println("allResults size:" + allResults.size());
-//        List<TDampDwIndexData> dataList = tDampDwIndexDataService.queryAllByLimit(indexCode, 0, 500);
-//        DWIndexDTO dwIndexDTO = DWIndexDTOConverter.INSTANCE.convertToDTO(dwIndex);
-//        TEtaIndexPushTask tEtaIndexPushTask = DWIndexDTOConverter.INSTANCE.convertToTEtaIndexPushTask(dwIndexDTO);
-//        etaIndexPushTaskService.insert(tEtaIndexPushTask);
+        DWIndexDTO dwIndexDTO = DWIndexDTOConverter.INSTANCE.convertToDTO(dwIndex);
+        List<DWIndexDataDTO> dtoList = allResults.stream().map(item -> DWIndexDataDTOConverter.INSTANCE.convertToDTO(item)).collect(Collectors.toList());
+        dwIndexDTO.setDataList(dtoList);
+        TEtaIndexPushTask tEtaIndexPushTask = DWIndexDTOConverter.INSTANCE.convertToTEtaIndexPushTask(dwIndexDTO);
+        etaIndexPushTaskService.insert(tEtaIndexPushTask);
         return DWIndexDTOConverter.INSTANCE.convertToDTO(dwIndex);
     }
 

qhtx-eta-integrator/qhtx-integrator-domain/src/main/java/com/qhtx/eta/domain/service/impl/ETAPushDataServiceImpl.java

@@ -0,0 +1,28 @@
+package com.qhtx.eta.domain.service.impl;
+
+import com.qhtx.eta.domain.convert.DWIndexDTOConverter;
+import com.qhtx.eta.domain.entity.DWIndexDTO;
+import com.qhtx.eta.domain.service.ETAPushDataService;
+import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
+import com.qhtx.eta.infra.service.TEtaIndexPushTaskService;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+@Service
+public class ETAPushDataServiceImpl implements ETAPushDataService {
+
+    @Resource
+    private TEtaIndexPushTaskService tEtaIndexPushTaskService;
+
+    @Override
+    public List<DWIndexDTO> getPushIndexTaskList() {
+        return null;
+    }
+
+    @Override
+    public TEtaIndexPushTask getETAPushTaskByIndexCode(String indexCode) {
+        return tEtaIndexPushTaskService.getByIndexCode(indexCode);
+    }
+}

qhtx-eta-integrator/qhtx-integrator-infra/src/main/java/com/qhtx/eta/infra/entity/dw/TEtaIndexPushTask.java

@@ -8,7 +8,7 @@ import java.io.Serializable;
 /**
  * (TEtaIndexPushTask)实体类
  *
- * @author makejava
+ * @author ch
  * @since 2024-09-30 13:54:17
  */
 @Data
@@ -20,7 +20,7 @@ public class TEtaIndexPushTask {
 
     private Date createTime = new Date();
 
-    private Date updateTime ;
+    private Date updateTime;
 
     private Date startDate;
 
@@ -30,12 +30,16 @@ public class TEtaIndexPushTask {
 
     private String unit;
 
+    private String LatestDataDate;
+
     private String sourceName;
 
     private Date lastPushTime;
 
     private Integer dataCount;
 
+    private String frequency;
+
 
 }
 

qhtx-eta-integrator/qhtx-integrator-infra/src/main/java/com/qhtx/eta/infra/mapper/TEtaIndexPushTaskDao.java

@@ -3,6 +3,8 @@ package com.qhtx.eta.infra.mapper;
 import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+
 /**
  * (TEtaIndexPushTask)表数据库访问层
  *
@@ -30,5 +32,8 @@ public interface TEtaIndexPushTaskDao {
     void update(TEtaIndexPushTask tEtaIndexPushTask);
 
 
+    List<TEtaIndexPushTask> getAllTask();
+
+    TEtaIndexPushTask getByIndexCode(@Param("indexCode") String indexCode);
 }
 

qhtx-eta-integrator/qhtx-integrator-infra/src/main/java/com/qhtx/eta/infra/service/TEtaIndexPushTaskService.java

@@ -4,6 +4,8 @@ package com.qhtx.eta.infra.service;
 import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+
 /**
  * (TEtaIndexPushTask)表服务接口
  *
@@ -29,5 +31,7 @@ public interface TEtaIndexPushTaskService {
      */
     void update(TEtaIndexPushTask tEtaIndexPushTask);
 
+    List<TEtaIndexPushTask> getAllTask();
 
+    TEtaIndexPushTask getByIndexCode(String indexCode);
 }

qhtx-eta-integrator/qhtx-integrator-infra/src/main/java/com/qhtx/eta/infra/service/impl/TEtaIndexPushTaskServiceImpl.java

@@ -2,11 +2,18 @@ package com.qhtx.eta.infra.service.impl;
 
 
 import com.qhtx.eta.infra.entity.dw.TEtaIndexPushTask;
+import com.qhtx.eta.infra.enums.Frequency;
 import com.qhtx.eta.infra.mapper.TEtaIndexPushTaskDao;
 import com.qhtx.eta.infra.service.TEtaIndexPushTaskService;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.util.Date;
+import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * (TEtaIndexPushTask)表服务实现类
@@ -15,6 +22,7 @@ import javax.annotation.Resource;
  * @since 2024-09-30 13:43:27
  */
 @Service("tEtaIndexPushTaskService")
+@Slf4j
 public class TEtaIndexPushTaskServiceImpl implements TEtaIndexPushTaskService {
     @Resource
     private TEtaIndexPushTaskDao tEtaIndexPushTaskDao;
@@ -42,5 +50,52 @@ public class TEtaIndexPushTaskServiceImpl implements TEtaIndexPushTaskService {
         this.tEtaIndexPushTaskDao.update(tEtaIndexPushTask);
     }
 
+    @Override
+    public List<TEtaIndexPushTask> getAllTask() {
+        return tEtaIndexPushTaskDao.getAllTask().stream().filter(item -> {
+            String frequencyStr = String.format("%s度", item.getFrequency());
+            Frequency frequency = Frequency.getFrequencyByDesc(frequencyStr);
+            if (frequency == null) {
+                log.warn("指标{} 频度获取失败,不支持的频度类型{},不同步数据", item.getIndexCode(), item.getFrequency());
+                return false;
+            }
+            Date lastPushTime = item.getLastPushTime();
+            if (needPush(lastPushTime, frequency)) {
+                return true;
+            }
+            return false;
+        }).collect(Collectors.toList());
+    }
+
+    @Override
+    public TEtaIndexPushTask getByIndexCode(String indexCode) {
+        return tEtaIndexPushTaskDao.getByIndexCode(indexCode);
+    }
 
+    private boolean needPush(Date startDate, Frequency frequency) {
+        int numberOfPeriods = -1;
+        if (frequency == null) {
+            frequency = Frequency.DAILY;
+        }
+        LocalDate today = LocalDate.now();
+        LocalDate localDate = startDate.toInstant().atZone(ZoneId.of("Asia/Shanghai")).toLocalDate();
+        switch (frequency) {
+            case DAILY:
+                 return localDate.minusDays(numberOfPeriods).compareTo(today)<=0;
+            case WEEKLY:
+                return localDate.minusWeeks(numberOfPeriods).compareTo(today)<=0;
+            case MONTHLY:
+                return localDate.minusMonths(numberOfPeriods).compareTo(today)<=0;
+            case YEARLY:
+                return localDate.minusYears(numberOfPeriods).compareTo(today)<=0;
+            case SEMI_ANNUAL:
+                return localDate.minusMonths(6L * numberOfPeriods).compareTo(today)<=0;
+            case QUARTERLY:
+                return localDate.minusMonths(3L * numberOfPeriods).compareTo(today)<=0;
+            case DECADELY:
+                return localDate.minusDays(10L * numberOfPeriods).compareTo(today)<=0;
+            default:
+                throw new IllegalArgumentException("Unsupported frequency: " + frequency);
+        }
+    }
 }

qhtx-eta-integrator/qhtx-integrator-infra/src/main/resources/mapper/TEtaIndexPushTaskDao.xml

@@ -61,5 +61,8 @@
         where ID = #{id}
     </update>
 
+    <select id="getByIndexCode" >
+            SELECT * from T_ETA_INDEX_PUSH_TASK where INDEX_CODE=#{indexCode}
+    </select>
 </mapper>