买油

src/main/java/com/biutag/supervisiondata/mapper/dwd/GBaseJJDMapper.java

@@ -22,7 +22,7 @@ public interface GBaseJJDMapper extends BaseMapper<GBaseJJD> {
     List<GBaseJJD> selectDataToRisk(@Param("startTime")String startTime, @Param("endTime")String endTime);
 
 
-    @Select(" select bjdh, " +
+    @Select(" select bjdh as mobile, " +
             " sum(IF(SUBSTR(jqxzdm, 1, 1) != '9',1,0)) as normal, " +
             " sum(IF(SUBSTR(jqxzdm, 1, 1) = '9',1,0)) as unNormal " +
             " from dwd_asj_zhtx_jjd where bjsj >= '2024-01-01 00:00:00' and bjdh is not null " +

src/main/java/com/biutag/supervisiondata/mapper/dwd/GBaseOilBuyMapper.java

@@ -0,0 +1,12 @@
+package com.biutag.supervisiondata.mapper.dwd;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.biutag.supervisiondata.pojo.entity.dwd.GBaseOilBuy;
+
+/**
+ * @author kami on 2024-11-27 14:07:13
+ * @version 0.0.1
+ * @since 1.8
+ */
+public interface GBaseOilBuyMapper extends BaseMapper<GBaseOilBuy> {
+}

src/main/java/com/biutag/supervisiondata/mapper/mine/RiskPersonalTagMapper.java

@@ -2,6 +2,10 @@ package com.biutag.supervisiondata.mapper.mine;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.biutag.supervisiondata.pojo.entity.mine.RiskPersonalTag;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Update;
+
+import java.util.List;
 
 /**
  * @author kami on 2024-11-08 19:43:32
@@ -9,5 +13,17 @@ import com.biutag.supervisiondata.pojo.entity.mine.RiskPersonalTag;
  * @since 1.8
  */
 public interface RiskPersonalTagMapper extends BaseMapper<RiskPersonalTag> {
+    /**
+     *
+     * @return
+     */
+    @Update("<script>" +
+            " update risk_personal_tag set del = 1 where small_tag = #{smallTag} and id_code in ( " +
+            "    <foreach collection='list' separator=','  item='item'> " +
+            "          #{item}" +
+            "    </foreach>" +
+            " ) "+
+            "</script>")
+    int updateToDelete(@Param("list") List<String> list,@Param("smallTag")String smallTag);
 
 }

src/main/java/com/biutag/supervisiondata/pojo/constants/Default.java

@@ -42,7 +42,7 @@ public class Default {
 
     public static final String BASIC = "基础信息";
 
-    public static final String MUCH_CALL = "重点人员报警异常";
+    public static final String OIL_BUY = "近3月购买散装油记录";
 
     // ***********因素*************
     public static final String AGE_IG = "年龄因素";
@@ -79,4 +79,6 @@ public class Default {
 
     public static final String NO_LOVE_IG = "感情失意因素";
 
+    public static final String GOODS_IG ="购买危险物品行为因素";
+
 }

src/main/java/com/biutag/supervisiondata/pojo/entity/dwd/GBaseOilBuy.java

@@ -0,0 +1,42 @@
+package com.biutag.supervisiondata.pojo.entity.dwd;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author kami on 2024-11-27 11:53:41
+ * @version 0.0.1
+ * @since 1.8
+ */
+@Data
+@TableName("`dwd_qb_nbgl_syssyj`")
+public class GBaseOilBuy implements Serializable {
+    @TableField("xh")
+    String id;
+    @TableField("pcsdm")
+    String orgId;
+    @TableField("pcsmc")
+    String orgName;
+    /**
+     * 购买地
+     */
+    @TableField("dwmc")
+    String companyName;
+    @TableField("gmrxm")
+    String name;
+    @TableField("zjhm")
+    String idCode;
+    @TableField("gmsl")
+    String number;
+    @TableField("gmyt")
+    String useTo;
+    @TableField("lxfs")
+    String mobile;
+    @TableField("yplx")
+    String type;
+    @TableField("gmsj")
+    String time;
+}

src/main/java/com/biutag/supervisiondata/repository/GBaseOilBuyRepository.java

@@ -0,0 +1,13 @@
+package com.biutag.supervisiondata.repository;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.biutag.supervisiondata.mapper.dwd.GBaseOilBuyMapper;
+import com.biutag.supervisiondata.pojo.entity.dwd.GBaseOilBuy;
+
+/**
+ * @author kami on 2024-11-27 14:09:03
+ * @version 0.0.1
+ * @since 1.8
+ */
+public class GBaseOilBuyRepository extends ServiceImpl<GBaseOilBuyMapper, GBaseOilBuy> {
+}

src/main/java/com/biutag/supervisiondata/service/PointService.java

@@ -28,4 +28,10 @@ public interface PointService {
     void syncControlNoPower(TaskParamDomain domain);
 
     void syncPointCall(TaskParamDomain domain);
+
+    /**
+     * 散装油
+     * @param domain domain
+     */
+    void syncOilBuy(TaskParamDomain domain);
 }

src/main/java/com/biutag/supervisiondata/service/RiskModelTaskClueService.java

@@ -13,4 +13,12 @@ import java.util.List;
 public interface RiskModelTaskClueService  {
 
     void saveClues(List<RiskModelTaskClue> list, String bigTag);
+
+    /**
+     * 删除线索并删除标签（只是删除或修改直接调用repository）
+     * @param list 线索
+     * @param idCodes 要删除标签的身份证号
+     * @param smallTag 要删除的标签（小）
+     */
+    void deleteCluesAndDeleteTag(List<RiskModelTaskClue> list, List<String> idCodes, String smallTag);
 }

src/main/java/com/biutag/supervisiondata/service/impl/MarriageServiceImpl.java

@@ -2,6 +2,7 @@ package com.biutag.supervisiondata.service.impl;
 
 import com.alibaba.fastjson2.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.biutag.supervisiondata.common.interfaces.CompletableFutureUtil;
 import com.biutag.supervisiondata.pojo.ClueData;
 import com.biutag.supervisiondata.pojo.constants.Default;
 import com.biutag.supervisiondata.pojo.domain.TaskParamDomain;
@@ -23,6 +24,7 @@ import org.springframework.stereotype.Service;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -54,20 +56,55 @@ public class MarriageServiceImpl implements MarriageService {
 
         List<GBaseHY> hyList = new ArrayList<>();
         log.info("开始查询结婚数据：{}", domain.getIdCards().size());
-        int count = 0;
-        for (List<String> idCard : domain.getIdCards()) {
-            count++;
-            hyList.addAll(hyRepository.list(new LambdaQueryWrapper<GBaseHY>()
-                    .select(GBaseHY::getYwlxdmmc, GBaseHY::getYwlxdmdm,GBaseHY::getYxbzdm, GBaseHY::getNfsfhm, GBaseHY::getNvfsfhm, GBaseHY::getDjrq)
-                    .eq(GBaseHY::getYxbzdm, "1")
-                    .in(GBaseHY::getNfsfhm, idCard)));
-            hyList.addAll(hyRepository.list(new LambdaQueryWrapper<GBaseHY>()
-                    .select(GBaseHY::getYwlxdmmc, GBaseHY::getYwlxdmdm, GBaseHY::getYxbzdm, GBaseHY::getNfsfhm, GBaseHY::getNvfsfhm,GBaseHY::getDjrq)
-                    .eq(GBaseHY::getYxbzdm, "1")
-                    .in(GBaseHY::getNvfsfhm, idCard)));
-            log.info("查询进度: {}", count);
+        List<List<String>> manIdCards = new ArrayList<>();
+        List<List<String>> femanIdCards = new ArrayList<>();
+        for (RiskPersonal person : domain.getPersons()) {
+            if(manIdCards.isEmpty()) {
+                manIdCards.add(new ArrayList<>());
+                femanIdCards.add(new ArrayList<>());
+            }
+            if("2".equals(person.getGender())) {
+                femanIdCards.getLast().add(person.getIdCode());
+                if(femanIdCards.getLast().size() >= 10000) {
+                    femanIdCards.add(new ArrayList<>());
+                }
+            } else {
+                manIdCards.getLast().add(person.getIdCode());
+                if(manIdCards.getLast().size() >= 10000) {
+                    manIdCards.add(new ArrayList<>());
+                }
+            }
         }
 
+        CompletableFuture.allOf(
+                CompletableFutureUtil.runSyncObject(() -> manIdCards.forEach(item -> hyList.addAll(hyRepository.list(new LambdaQueryWrapper<GBaseHY>()
+                        .select(GBaseHY::getYwlxdmmc, GBaseHY::getYwlxdmdm,GBaseHY::getYxbzdm, GBaseHY::getNfsfhm, GBaseHY::getNvfsfhm, GBaseHY::getDjrq)
+                        .eq(GBaseHY::getYxbzdm, "1")
+                        .in(GBaseHY::getNfsfhm, item))))),
+                CompletableFutureUtil.runSyncObject(() -> femanIdCards.forEach(item -> hyList.addAll(hyRepository.list(new LambdaQueryWrapper<GBaseHY>()
+                        .select(GBaseHY::getYwlxdmmc, GBaseHY::getYwlxdmdm, GBaseHY::getYxbzdm, GBaseHY::getNfsfhm, GBaseHY::getNvfsfhm,GBaseHY::getDjrq)
+                        .eq(GBaseHY::getYxbzdm, "1")
+                        .in(GBaseHY::getNvfsfhm, item)))))
+        ).join();
+
+        LocalDateTime time = null;
+        for (GBaseHY gBaseHY : hyList) {
+            if(gBaseHY.getDjrq() == null) {
+                continue;
+            }
+            LocalDateTime tmp = gBaseHY.getDjrq().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
+            if(tmp.isAfter(LocalDateTime.now())) {
+                continue;
+            }
+            if(time == null) {
+                time = tmp;
+                continue;
+            }
+            if(tmp.isAfter(time)) {
+                time = tmp;
+            }
+        }
+        log.info("最大时间{}", time);
         // 每个人应该是最新的婚姻登记为标准
         Map<String, List<GBaseHY>> manMap = new HashMap<>();
         Map<String, List<GBaseHY>> femanMap = new HashMap<>();
@@ -96,7 +133,7 @@ public class MarriageServiceImpl implements MarriageService {
         List<ClueData> toInsertData2 = new ArrayList<>();
         List<RiskModelTaskClue> toUpdateData2 = new ArrayList<>();
 
-        long targetTime = System.currentTimeMillis() - 31536000000l;
+        LocalDateTime targetTime = LocalDateTime.now().minusMonths(12);
         for (RiskPersonal person : domain.getPersons()) {
             List<GBaseHY> tmpList = manMap.get(person.getIdCode());
             if(tmpList == null) {
@@ -116,18 +153,17 @@ public class MarriageServiceImpl implements MarriageService {
                     last = gBaseHY;
                     continue;
                 }
-                if(gBaseHY.getDjrq().getTime() > last.getDjrq().getTime()) {
+                if(gBaseHY.getDjrq().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().isAfter(last.getDjrq().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime())) {
                     last = gBaseHY;
                 }
             }
-            count++;
             // 看看是不是离婚
             if(last == null || last.getYwlxdmdm() == null) {
                 continue;
             }
             if(last.getYwlxdmdm().toUpperCase().contains("IB") || last.getYwlxdmdm().toUpperCase().contains("ICB")) {
                 // 离婚 如果旧数据不存在就要新增
-                if(last.getDjrq().getTime() > targetTime) {
+                if(last.getDjrq().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().isAfter(targetTime)) {
                     // 近365天的离婚
                     RiskModelTaskClue clue = oldClueMap.get(person.getIdCode());
                     if(clue != null) {
@@ -164,7 +200,7 @@ public class MarriageServiceImpl implements MarriageService {
             } else if(last.getYwlxdmdm().toUpperCase().contains("IA") || last.getYwlxdmdm().toUpperCase().contains("ICA")) {
                 // 结婚 如果旧数据存在就要修改为0
                 // 近365天的离婚
-                if(last.getDjrq().getTime() > targetTime) {
+                if(last.getDjrq().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().isAfter(targetTime)) {
                     RiskModelTaskClue clue = oldClueMap.get(person.getIdCode());
                     if(clue == null) {
                         continue;

src/main/java/com/biutag/supervisiondata/service/impl/PersonalServiceImpl.java

@@ -123,7 +123,7 @@ public class PersonalServiceImpl implements PersonalService {
                 syrkxx = idCodeMap.get(item.getBjrzjhm());
                 cache.put(item.getBjrzjhm(), new Byte[0]);
                 if (syrkxx != null) {
-                    personals.add(createPerson(syrkxx, item.getBjsj(), now, end));
+                    personals.add(createPerson(syrkxx, item.getBjsj(), now, end,Default.BIG_TAG_JJD));
                 }
                 continue;
             }
@@ -135,7 +135,7 @@ public class PersonalServiceImpl implements PersonalService {
                         syrkxx = idCodeMap.get(s);
                         cache.put(s, new Byte[0]);
                         if (syrkxx != null) {
-                            personals.add(createPerson(syrkxx, item.getBjsj(), now, end));
+                            personals.add(createPerson(syrkxx, item.getBjsj(), now, end, Default.BIG_TAG_JJD));
                         }
                     }
                 }
@@ -145,7 +145,7 @@ public class PersonalServiceImpl implements PersonalService {
             syrkxx = compare(xxList, item.getBjrxm());
             if (syrkxx != null) {
                 cache.put(syrkxx.getGmsfhm(), new Byte[0]);
-                personals.add(createPerson(syrkxx, item.getBjsj(), now, end));
+                personals.add(createPerson(syrkxx, item.getBjsj(), now, end, Default.BIG_TAG_JJD));
             }
         }
         riskPersonalService.saveDistinct(personals);
@@ -223,7 +223,7 @@ public class PersonalServiceImpl implements PersonalService {
             exist.put(syrkxx.getGmsfhm(), new Byte[0]);
             Contradiction contradiction = map.get(syrkxx.getGmsfhm());
             personals.add(createPerson(syrkxx, Optional.of(contradiction).map(Contradiction::getMaxTime)
-                    .orElse(null), now.toLocalDate(), now));
+                    .orElse(null), now.toLocalDate(), now, Default.BIG_TAG_SHWFR));
         }
         riskPersonalService.saveDistinct(personals);
     }
@@ -236,25 +236,52 @@ public class PersonalServiceImpl implements PersonalService {
 
     @Override
     public void pullPetition(LocalDateTime start) {
-        List<DataPetition12337> pList1 = petition12337Repository.list(new LambdaQueryWrapper<DataPetition12337>()
-                .select(DataPetition12337::getName, DataPetition12337::getIdCode, DataPetition12337::getPhone, DataPetition12337::getThirdDepartId, DataPetition12337::getThirdDepartName)
-                .ge(DataPetition12337::getDiscoverTime, start));
-
-        List<DataPetitionComplaint> pList2 = complaintRepository.list(new LambdaQueryWrapper<DataPetitionComplaint>()
-                .select(DataPetitionComplaint::getResponderName, DataPetitionComplaint::getResponderPhone, DataPetitionComplaint::getResponderName, DataPetitionComplaint::getThirdDepartId, DataPetitionComplaint::getThirdDepartName)
-                .ge(DataPetitionComplaint::getDiscoveryTime, start));
-
-        List<DataCaseVerif> pList3 = dataCaseVerifRepository.list(new LambdaQueryWrapper<DataCaseVerif>()
-                .select(DataCaseVerif::getResponderName, DataCaseVerif::getResponderPhone, DataCaseVerif::getThirdDepartId, DataCaseVerif::getThirdDepartName)
-                .ge(DataCaseVerif::getDiscoveryTime, start));
+        List<DataCaseVerif> pList3 = new ArrayList<>();
+        List<DataPetition12337> pList1 = new ArrayList<>();
+        List<DataPetitionComplaint> pList2 = new ArrayList<>();
+        LocalDateTime now = LocalDateTime.now();
+        // 捞数据
+        CompletableFuture.allOf(
+                CompletableFutureUtil.runSyncObject(() -> pList1.addAll(petition12337Repository.list(new LambdaQueryWrapper<DataPetition12337>()
+                                .select(DataPetition12337::getName, DataPetition12337::getIdCode, DataPetition12337::getPhone, DataPetition12337::getThirdDepartId, DataPetition12337::getThirdDepartName)
+                                .ge(DataPetition12337::getDiscoverTime, start)))),
+                CompletableFutureUtil.runSyncObject(() -> pList2.addAll(complaintRepository.list(new LambdaQueryWrapper<DataPetitionComplaint>()
+                        .select(DataPetitionComplaint::getResponderName, DataPetitionComplaint::getResponderPhone, DataPetitionComplaint::getResponderName, DataPetitionComplaint::getThirdDepartId, DataPetitionComplaint::getThirdDepartName)
+                        .ge(DataPetitionComplaint::getDiscoveryTime, start)))),
+                CompletableFutureUtil.runSyncObject(() -> pList3.addAll(dataCaseVerifRepository.list(new LambdaQueryWrapper<DataCaseVerif>()
+                        .select(DataCaseVerif::getResponderName, DataCaseVerif::getResponderPhone, DataCaseVerif::getThirdDepartId, DataCaseVerif::getThirdDepartName)
+                        .ge(DataCaseVerif::getDiscoveryTime, start))))
+        ).join();
 
-        List<String> ids = new ArrayList<>();
+        List<String> idCards = new ArrayList<>();
         List<String> mobiles = new ArrayList<>();
-        ids.addAll(pList1.stream().map(DataPetition12337::getIdCode).toList());
-        ids.addAll(pList2.stream().map(DataPetitionComplaint::getResponderIdCode).toList());
+        idCards.addAll(pList1.stream().map(DataPetition12337::getIdCode).toList());
+        idCards.addAll(pList2.stream().map(DataPetitionComplaint::getResponderIdCode).toList());
         mobiles.addAll(pList3.stream().map(DataCaseVerif::getResponderPhone).toList());
 
-
+        List<String> idCards2 = idCards.stream().distinct().toList();
+        List<String> mobiles2 = mobiles.stream().distinct().toList();
+
+        List<GBaseSYRKXX> xxList = new ArrayList<>();
+        // 实有人口数据整合
+        CompletableFuture.allOf(CompletableFutureUtil.runSyncObject(() -> xxList.addAll(getByIdCode(IntStream.range(0, (idCards2.size() + 9999) / 10000)
+                        .mapToObj(i -> idCards2.subList(i * 10000, Math.min((i + 1) * 10000, idCards2.size())))
+                        .toList()))),
+                CompletableFutureUtil.runSyncObject(() -> xxList.addAll(getByMobiles(IntStream.range(0, (mobiles2.size() + 9999) / 10000)
+                        .mapToObj(i -> mobiles2.subList(i * 10000, Math.min((i + 1) * 10000, mobiles2.size())))
+                        .toList())))).join();
+        // 去重
+        List<RiskPersonal> personals = new ArrayList<>();
+        Map<String, Byte[]> map = new WeakHashMap<>();
+        LocalDate today = now.toLocalDate();
+        Date nowDate = new Date();
+        for (GBaseSYRKXX xx : xxList) {
+            if(map.get(xx.getGmsfhm()) == null) {
+                personals.add(createPerson(xx, nowDate, today, now, Default.BIG_TAG_PETITION));
+                map.put(xx.getGmsfhm(), new Byte[0]);
+            }
+        }
+        riskPersonalService.saveDistinct(personals);
     }
 
     private final SupExternalDepartRepository supExternalDepartRepository;
@@ -324,7 +351,7 @@ public class PersonalServiceImpl implements PersonalService {
     }
 
 
-    RiskPersonal createPerson(GBaseSYRKXX item, Date bjsj, LocalDate now, LocalDateTime time) {
+    RiskPersonal createPerson(GBaseSYRKXX item, Date date, LocalDate now, LocalDateTime time, String tag) {
         RiskPersonal person = new RiskPersonal();
         person.setName(item.getXm());
         person.setMobileNumber(item.getLxdh());
@@ -335,9 +362,9 @@ public class PersonalServiceImpl implements PersonalService {
         person.setCreateTime(time);
         person.setAge(IdCodeUtil.idCard2Age(person.getIdCode(), now));
         person.setGender(IdCodeUtil.idCard2Gender(person.getIdCode()));
-        person.setTags(Default.BIG_TAG_JJD);
+        person.setTags(tag);
         try {
-            LocalDateTime controlTime = bjsj.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
+            LocalDateTime controlTime = date.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
             person.setControlTime(controlTime);
         } catch (Exception e) {
             log.info("报警时间处理异常");

src/main/java/com/biutag/supervisiondata/service/impl/PointServiceImpl.java

@@ -9,6 +9,7 @@ import com.biutag.supervisiondata.pojo.constants.Default;
 import com.biutag.supervisiondata.pojo.database.MuchCall;
 import com.biutag.supervisiondata.pojo.domain.*;
 import com.biutag.supervisiondata.pojo.entity.dwd.GBaseJJD;
+import com.biutag.supervisiondata.pojo.entity.dwd.GBaseOilBuy;
 import com.biutag.supervisiondata.pojo.entity.dwd.GBaseTbDZdrylb;
 import com.biutag.supervisiondata.pojo.entity.dwd.GBaseZDRY;
 import com.biutag.supervisiondata.pojo.entity.mine.*;
@@ -44,8 +45,6 @@ public class PointServiceImpl implements PointService {
 
     private final GBaseZDRYRepository zdryRepository;
 
-    private final GBaseZDXLRepository zdxlRepository;
-
     private final RiskPersonalService riskPersonalService;
 
     private final RiskModelTaskClueService modelTaskClueService;
@@ -415,6 +414,93 @@ public class PointServiceImpl implements PointService {
         taskRepository.updateById(updateTask);
     }
 
+    private final GBaseOilBuyRepository oilBuyRepository;
+
+    @Override
+    public void syncOilBuy(TaskParamDomain domain) {
+        LocalDateTime currentTime = LocalDateTime.now().minusMonths(3);
+        List<GBaseOilBuy> buys = oilBuyRepository.list(new LambdaQueryWrapper<GBaseOilBuy>()
+                .select(GBaseOilBuy::getId,GBaseOilBuy::getNumber,GBaseOilBuy::getOrgId,GBaseOilBuy::getOrgName,
+                        GBaseOilBuy::getName,GBaseOilBuy::getCompanyName,GBaseOilBuy::getType,GBaseOilBuy::getTime,
+                        GBaseOilBuy::getUseTo,GBaseOilBuy::getMobile,GBaseOilBuy::getIdCode)
+                .ge(GBaseOilBuy::getTime, currentTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));
+        RiskTask task = createTask(buys.size(), 52);
+        taskRepository.save(task);
+        // 旧数据
+        List<RiskModelTaskClue> oldList = Optional.ofNullable(domain.getMap().get(task.getModelId())).orElse(new ArrayList<>());
+
+        Map<String, RiskModelTaskClue> oldMap = oldList.stream().collect(Collectors.toMap(RiskModelTaskClue::getCaseIds, Function.identity(), (val, old)-> val));
+
+        Map<String, List<GBaseOilBuy>> buyMap = buys.stream().collect(Collectors.groupingBy(GBaseOilBuy::getIdCode));
+        // 新增记录
+        List<RiskModelTaskClue> toInsert = new ArrayList<>();
+        for (RiskPersonal person : domain.getPersons()) {
+            List<GBaseOilBuy> mineBuy = buyMap.get(person.getIdCode());
+            if(mineBuy == null) {
+                continue;
+            }
+            for (GBaseOilBuy item : mineBuy) {
+                if(oldMap.get(BhUtil.oilBh(item.getId())) != null) {
+                    continue;
+                }
+                ClueData data = ClueData.builder()
+                        .name(person.getName())
+                        .id(BhUtil.oilBh(item.getId()))
+                        .tag(Default.OIL_BUY)
+                        .idCode(person.getIdCode())
+                        .departId(item.getOrgId())
+                        .departName(item.getOrgName())
+                        .personId(person.getId())
+                        .sourceData("发现该人员于"+item.getTime()+"在"+item.getCompanyName()+"购买容量为"+item.getNumber()+"L的"+item.getType()+",用作"+item.getUseTo()+",风险分增加2分")
+                        .build();
+                try {
+                    data.setEventTime(LocalDateTime.parse(item.getTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+                } catch (Exception e) {
+                    data.setEventTime(LocalDateTime.now());
+                    log.error("散装油时间转化异常");
+                }
+                toInsert.add(createClue(task.getModelId(), 2, task.getId(), data));
+            }
+        }
+        List<RiskModelTaskClue> toDelete = new ArrayList<>();
+        Map<String, Byte[]> deletePerson = new WeakHashMap<>();
+        Map<String, Byte[]> livePerson = new WeakHashMap<>();
+        // 删除记录
+        for (RiskModelTaskClue clue : oldList) {
+            if(clue.getEventTime().isBefore(currentTime)){
+                RiskModelTaskClue delete = new RiskModelTaskClue();
+                delete.setId(clue.getId());
+                delete.setDel(1);
+                toDelete.add(delete);
+                deletePerson.put(clue.getIdCode(), new Byte[0]);
+            } else {
+                livePerson.put(clue.getIdCode(), new Byte[0]);
+            }
+        }
+        List<String> idCodes = new ArrayList<>();
+        // 删除标签 从deletePerson遍历，存活中没有，新增没有，说明该标签就没了
+        for (String s : deletePerson.keySet()) {
+            if(livePerson.get(s)==null && buyMap.get(s) == null) {
+                idCodes.add(s);
+            }
+        }
+        if (!toInsert.isEmpty()) {
+            log.info("增加散装油人员:{}", toInsert.size());
+            modelTaskClueService.saveClues(toInsert, Default.GOODS_IG);
+        }
+        if(!toDelete.isEmpty()) {
+            modelTaskClueService.deleteCluesAndDeleteTag(toDelete, idCodes, Default.OIL_BUY);
+        }
+
+        RiskTask updateTask = new RiskTask();
+        updateTask.setId(task.getId());
+        updateTask.setInsertSize(toInsert.size());
+        updateTask.setUpdateSize(toDelete.size());
+        updateTask.setState(1);
+        updateTask.setEndTime(LocalDateTime.now());
+        taskRepository.updateById(updateTask);
+    }
+
     RiskModelTaskClue createClue(Integer modelId, Integer score, Integer taskId, ClueData data) {
         RiskModelTaskClue clue = new RiskModelTaskClue();
         clue.setModelId(modelId);

src/main/java/com/biutag/supervisiondata/service/impl/RiskModelTaskClueServiceImpl.java

@@ -9,6 +9,7 @@ import com.biutag.supervisiondata.pojo.constants.Default;
 import com.biutag.supervisiondata.pojo.entity.mine.RiskModelTaskClue;
 import com.biutag.supervisiondata.pojo.entity.mine.RiskPersonalControlRecord;
 import com.biutag.supervisiondata.pojo.entity.mine.RiskPersonalTag;
+import com.biutag.supervisiondata.repository.RiskModelTaskClueRepository;
 import com.biutag.supervisiondata.repository.RiskPersonalTagRepository;
 import com.biutag.supervisiondata.service.RiskModelTaskClueService;
 import com.biutag.supervisiondata.service.RiskPersonalTagService;
@@ -42,6 +43,8 @@ public class RiskModelTaskClueServiceImpl implements RiskModelTaskClueService {
 
     private final RiskPersonalTagRepository tagRepository;
 
+    private final RiskModelTaskClueRepository clueRepository;
+
     Connection getConnection() throws SQLException {
         return dataSource.getConnection();
     }
@@ -111,6 +114,14 @@ public class RiskModelTaskClueServiceImpl implements RiskModelTaskClueService {
         }
     }
 
+    @Override
+    public void deleteCluesAndDeleteTag(List<RiskModelTaskClue> list, List<String> idCodes, String smallTag) {
+        clueRepository.updateBatchById(list);
+        if(!idCodes.isEmpty()) {
+            tagRepository.getBaseMapper().updateToDelete(idCodes, smallTag);
+        }
+    }
+
     final String sql = "INSERT INTO risk_model_task_clue(model_id, name, id_code, task_id, source_id, risk_reason, data, score, event_time, create_time, update_time, case_ids) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)";
 
     public void batchInsert(List<RiskModelTaskClue> alliances, String bigTag) {

src/main/java/com/biutag/supervisiondata/util/BhUtil.java

@@ -20,6 +20,8 @@ public class BhUtil {
 
     private static final String ZY_PIX = "zy___";
 
+    private static final String OIL_PIX = "oil___";
+
     public static String ajBh(String id) {
         return AJ_PIX+id;
     }
@@ -39,4 +41,6 @@ public class BhUtil {
     public static String zyBh(String id) {
         return ZY_PIX+id;
     }
+
+    public static String oilBh(String id) { return OIL_PIX+id; }
 }

src/main/resources/mapper/dwd/GBaseOilBuyMapper.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.biutag.supervisiondata.mapper.dwd.GBaseOilBuyMapper">
+
+</mapper>