资金预警需求-大额交易报表：

1.改用调度任务创建大额交易数据

main/java/kd/cosmic/jkjt/tmc/bei/common/DateUtils.java

@@ -7,8 +7,8 @@ import java.time.ZoneId;
 import java.util.Date;
 
 public class DateUtils {
-    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-    private static final SimpleDateFormat DATE_FORMAT_DAY = new SimpleDateFormat("yyyy-MM-dd");
+    public static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    public static final SimpleDateFormat DATE_FORMAT_DAY = new SimpleDateFormat("yyyy-MM-dd");
 
     private static final SimpleDateFormat DATE_FORMAT_MONTH= new SimpleDateFormat("yyyy-MM");
 
@@ -48,4 +48,13 @@ public class DateUtils {
     public static String formatDateMonth(Date date) {
         return DATE_FORMAT_MONTH.format(date);
     }
+
+    public static Date parseDate(String dateStr) {
+        try {
+            return DATE_FORMAT.parse(dateStr);
+        } catch (Exception e) {
+            e.printStackTrace();
+            return null;
+        }
+    }
 }

main/java/kd/cosmic/jkjt/tmc/bei/task/LargeTransReportDataTask.java

@@ -0,0 +1,178 @@
+package kd.cosmic.jkjt.tmc.bei.task;
+
+import com.google.common.collect.Lists;
+import kd.bos.algo.DataSet;
+import kd.bos.context.RequestContext;
+import kd.bos.dataentity.entity.DynamicObject;
+import kd.bos.dataentity.entity.DynamicObjectCollection;
+import kd.bos.exception.KDException;
+import kd.bos.logging.BizLog;
+import kd.bos.logging.Log;
+import kd.bos.logging.LogFactory;
+import kd.bos.orm.ORM;
+import kd.bos.orm.query.QCP;
+import kd.bos.orm.query.QFilter;
+import kd.bos.schedule.api.StopTask;
+import kd.bos.schedule.executor.AbstractTask;
+import kd.bos.servicehelper.BusinessDataServiceHelper;
+import kd.bos.servicehelper.QueryServiceHelper;
+import kd.bos.servicehelper.operation.SaveServiceHelper;
+import kd.cosmic.jkjt.tmc.bei.common.CommonUtils;
+import kd.cosmic.jkjt.tmc.bei.common.DateUtils;
+import kd.cosmic.jkjt.tmc.bei.common.constant.BeiBeTransDetailConstant;
+import kd.cosmic.jkjt.tmc.bei.common.constant.LargeTransConfigConstant;
+
+import java.math.BigDecimal;
+import java.util.*;
+import java.util.stream.Collectors;
+
+
+/**
+ * 大额交易预警任务
+ */
+public class LargeTransReportDataTask extends AbstractTask implements StopTask {
+    private static final Log logger = LogFactory.getLog(LargeTransReportDataTask.class);
+
+    private static final String LARGETRANS_RPTDATA_LASTTIME = "LARGETRANS_RPTDATA_LASTTIME";
+
+    // 获取上次更新时间
+    Date lastUpdateTime ;
+
+    Date nowDate ;
+
+    @Override
+    public void execute(RequestContext requestContext, Map<String, Object> map) throws KDException {
+
+        lastUpdateTime = getLastUpdateTime();
+
+        nowDate = new Date();
+
+        Set<Long> largeTransConfig = getLargeTransConfig();
+
+        List<DataSet> dataSetList = new ArrayList<>();
+        largeTransConfig.stream().forEach(configId -> dataSetList.add(getLargeTransDataByConfig(configId)));
+
+        DataSet totalData = CommonUtils.joinDataSetList(dataSetList);
+
+        DynamicObjectCollection allDataCol = ORM.create().toPlainDynamicObjectCollection(totalData);
+        List<DynamicObject> largeTransDataList = new ArrayList<>();
+
+        List<DynamicObject> allDataList = allDataCol.stream().map(r -> r).collect(Collectors.toList());
+        List<List<DynamicObject>> partitionList = Lists.partition(allDataList, 1000);
+        partitionList.stream().forEach(r -> largeTransDataList.addAll(createDataPatch(r)));
+
+        SaveServiceHelper.save(largeTransDataList.toArray(new DynamicObject[0]));
+        DynamicObject paramInfo = BusinessDataServiceHelper.loadSingle("nckd_message_feishu", "nckd_params",
+                new QFilter("number", QCP.equals, LARGETRANS_RPTDATA_LASTTIME).toArray());
+        if(paramInfo != null){
+            paramInfo.set("nckd_params" , DateUtils.DATE_FORMAT.format(nowDate));
+            SaveServiceHelper.save(new DynamicObject[]{ paramInfo });
+        }
+
+    }
+
+    protected List<DynamicObject> createDataPatch(List<DynamicObject> dataList){
+        // 查询是否已经创建过
+        List<DynamicObject> newDataList = new ArrayList<>();
+        List<Long> transDetailIdList = dataList.stream().map(r -> r.getLong("id")).collect(Collectors.toList());
+        List<Long> existDetailIdList = QueryServiceHelper.query("nckd_largetransrptdata", "nckd_detailid",
+                        new QFilter("nckd_detailid", QCP.in, transDetailIdList).toArray())
+                .stream().map(r -> r.getLong("nckd_detailid"))
+                .collect(Collectors.toList());
+        for(DynamicObject data : dataList){
+            DynamicObject dataObj = BusinessDataServiceHelper.newDynamicObject("nckd_largetransrptdata");
+            long transDetailId = data.getLong("id");
+            if(existDetailIdList.contains(transDetailId)){
+                continue;
+            }
+            dataObj.set("nckd_detailid", data.getLong("id"));
+            dataObj.set("nckd_company", data.getLong("nckd_company"));
+            dataObj.set("nckd_rulename", data.getString("nckd_rulename"));
+            dataObj.set("nckd_oppunit", data.getString("nckd_oppunit"));
+            dataObj.set("nckd_amount", data.getBigDecimal("nckd_amount"));
+            dataObj.set("nckd_payaccount", data.getString("nckd_payaccount"));
+            dataObj.set("nckd_payaccbanknumber", data.getString("nckd_payaccbanknumber"));
+            dataObj.set("nckd_payaccbank", data.getString("nckd_payaccbank"));
+            dataObj.set("nckd_description", data.getString("nckd_description"));
+            dataObj.set("nckd_biztime", data.getDate("nckd_biztime"));
+            newDataList.add(dataObj);
+        }
+        return newDataList;
+    }
+
+    protected DataSet getLargeTransDataByConfig(Long largeTransIdConfig) {
+        List<DataSet> dataSetList = new ArrayList<>();
+        DynamicObject configInfo = BusinessDataServiceHelper.loadSingle(largeTransIdConfig, LargeTransConfigConstant.ENTITYID);
+        DynamicObject company = configInfo.getDynamicObject(LargeTransConfigConstant.KEY_NCKD_COMPANY);
+        // 增加互斥锁，防止多个任务同时执行
+        DynamicObjectCollection ruleEntryCol = configInfo.getDynamicObjectCollection(LargeTransConfigConstant.RULEENTRY.ENTITYID);
+        ruleEntryCol.stream().forEach(ruleEntry -> dataSetList.add(getCompanyRuleEntryDataSet(company, ruleEntry)));
+        return CommonUtils.joinDataSetList(dataSetList);
+    }
+
+    protected DataSet getCompanyRuleEntryDataSet(DynamicObject company , DynamicObject ruleEntry){
+        // 最小金额
+        BigDecimal lowerAmount = ruleEntry.getBigDecimal(LargeTransConfigConstant.RULEENTRY.KEY_NCKD_AMOUNT_LOWER);
+        // 最大金额
+        BigDecimal highAmount = ruleEntry.getBigDecimal(LargeTransConfigConstant.RULEENTRY.KEY_NCKD_AMOUNT_HIGH);
+        if(!ruleEntry.getBoolean("nckd_enable")){
+            BizLog.log("规则没有启用s，不查询数据。");
+            return null;
+        }
+        String ruleName = ruleEntry.getString(LargeTransConfigConstant.RULEENTRY.KEY_NCKD_RULENAME);
+        DynamicObjectCollection receiverCol = ruleEntry.getDynamicObjectCollection(LargeTransConfigConstant.RULEENTRY.KEY_NCKD_RECEIVER);
+        if(receiverCol == null || receiverCol.isEmpty()){
+            BizLog.log("消息接收人为空，不查询数据。");
+            return null;
+        }
+        // 查询公司下，时间范围内符合规则条件的数据
+        return getTransDetailDataSet(ruleName, company, lowerAmount, highAmount);
+    }
+
+    protected DataSet getTransDetailDataSet(String ruleName, DynamicObject company , BigDecimal lowerAmount , BigDecimal highAmount){
+
+        // 获取交易日期过滤条件
+        List<QFilter> filterList = new ArrayList<>();
+        filterList.add(new QFilter(BeiBeTransDetailConstant.KEY_COMPANY, QFilter.equals, company.getPkValue()));
+        filterList.add(new QFilter(BeiBeTransDetailConstant.KEY_BIZTIME, QFilter.large_equals, lastUpdateTime));
+        filterList.add(new QFilter(BeiBeTransDetailConstant.KEY_BIZTIME, QFilter.less_than, nowDate));
+
+        if(lowerAmount != null && lowerAmount.compareTo(BigDecimal.ZERO) > 0){
+            filterList.add(new QFilter(BeiBeTransDetailConstant.KEY_DEBITAMOUNT, QFilter.large_equals, lowerAmount));
+        }
+        if (highAmount != null && highAmount.compareTo(BigDecimal.ZERO) > 0){
+            filterList.add(new QFilter(BeiBeTransDetailConstant.KEY_DEBITAMOUNT, QFilter.less_than, highAmount));
+        }
+        String selector = String.format("id , '%s' nckd_company , '%s' nckd_rulename , %s nckd_oppunit , %s nckd_amount , '%s' nckd_payaccount , %s nckd_payaccbanknumber , %s nckd_payaccbank , %s nckd_description , %s nckd_biztime",
+                company.getPkValue(), ruleName, "oppunit", "debitamount", company.getString("name"), "accountbank.bankaccountnumber", "bank.name", "description", "biztime");
+        return QueryServiceHelper.queryDataSet(this.getClass().getName() ,BeiBeTransDetailConstant.ENTITYID, selector, filterList.toArray(new QFilter[0]),"bizdate desc");
+    }
+
+    protected Date getLastUpdateTime() {
+        DynamicObject paramInfo = QueryServiceHelper.queryOne("nckd_message_feishu", "nckd_params",
+                new QFilter("number", QCP.equals, LARGETRANS_RPTDATA_LASTTIME).toArray());
+        if (paramInfo != null) {
+            String lastUpdateTimeStr = paramInfo.getString("nckd_params");
+            Date date = DateUtils.parseDate(lastUpdateTimeStr);
+            if(date != null){
+                return date ;
+            }
+        }
+        // 默认一天前的数据
+        return DateUtils.getDateByDayAndHour(-1, 0);
+    }
+
+    protected Set<Long> getLargeTransConfig() {
+
+        List<QFilter> filterList = new ArrayList<>();
+        filterList.add(new QFilter(LargeTransConfigConstant.KEY_STATUS, QFilter.equals, "C"));
+        filterList.add(new QFilter(LargeTransConfigConstant.KEY_ENABLE, QFilter.equals, "1"));
+        DynamicObjectCollection configCol = QueryServiceHelper.query(LargeTransConfigConstant.ENTITYID,
+                "id,nckd_company.id", filterList.toArray(new QFilter[0]));
+        Map<Long, Long> companyIdMap = configCol.stream()
+                .collect(Collectors.toMap(r -> r.getLong("nckd_company.id"), r -> r.getLong("id"), (a, b) -> a));
+        return companyIdMap.values().stream().collect(Collectors.toSet());
+    }
+
+
+}