diff --git a/pom.xml b/pom.xml index 6cf15bc..c784c5c 100644 --- a/pom.xml +++ b/pom.xml @@ -41,6 +41,10 @@ postgresql runtime + + com.fasterxml.jackson.datatype + jackson-datatype-jsr310 + com.ibm.db2 jcc diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java index f87c3d6..d36faac 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -25,7 +25,7 @@ public class StageConfig { @Bean("flagQueueFactory") public StageQueueFactory flagQueueFactory() { - return new StageQueueFactory<>(15); + return new StageQueueFactory<>(100); } } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/indic/FlagsMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/indic/FlagsMapper.java index df988b7..bd7a5fc 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/indic/FlagsMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/indic/FlagsMapper.java @@ -5,9 +5,13 @@ import java.util.List; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import ru.nbch.credit_tracker.entities.indic.PaymentsFlag; +import ru.nbch.credit_tracker.service.indic.FlagsService; @Mapper public interface FlagsMapper { + List findRuPastDuarearData(@Param("fids") List fids); + + List findRuPaymtData(@Param("fids") List fids, @Param("reportedDate") LocalDate reportedDate); List findFlagsDebt5000(@Param("fids") List fids); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java index 3c2f8a9..6449955 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java @@ -13,6 +13,7 @@ public interface PhoneFidMapMapper { void registerSubjects(@Param("subjects") List subjects, @Param("packageId") Long packageId); void register(@Param("phoneFidMap") List phoneFidMap); + void registerJson(@Param("json") String json); void registerWithId(@Param("phoneFidMap") List phoneFidMap); List findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java index f7239ef..e0b6a4f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java @@ -1,7 +1,13 @@ package ru.nbch.credit_tracker.service.task.impl; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import java.time.LocalDate; +import java.time.LocalDateTime; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -28,6 +34,10 @@ public class FlagsEnricher implements Runnable { private BlockingQueue in; private final CompletableFuture done = new CompletableFuture<>(); private final ExecutorService ioPool; + private final ObjectMapper mapper = new ObjectMapper() + .registerModule(new JavaTimeModule()) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS) + .setSerializationInclusion(JsonInclude.Include.NON_NULL); public FlagsEnricher(FlagsService2 flagsService, PhoneFidMapMapper phoneFidMapMapper) { @@ -62,8 +72,7 @@ public class FlagsEnricher implements Runnable { done.complete(null); return; } - Future future = ioPool.submit(() -> process(subjectPayload)); - inFlight.add(future); + inFlight.add(ioPool.submit(() -> process(subjectPayload))); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } catch (Exception e) { @@ -87,14 +96,16 @@ public class FlagsEnricher implements Runnable { Map flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(fids, yearAgo); List phoneFidMaps = subjects.stream().map(subject -> { - Integer debt500 = null; - Integer pastDue90 = null; - Integer paymt90 = null; + Integer debt5000 = null; + Integer flagDays90 = null; + LocalDateTime calculatedAt = null; if (subject.getFid() != null) { - debt500 = flagsDebt5000.getOrDefault(subject.getFid(), 0); - pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0); - paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0); + debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0); + int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0); + int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0); + flagDays90 = Math.max(pastDue90, paymt90); + calculatedAt = LocalDateTime.now(); } return PhoneFidMap.builder() @@ -102,11 +113,37 @@ public class FlagsEnricher implements Runnable { .subjectId(subject.getIdentifier()) .packageId(subjectPayload.getPackageId()) .fid(subject.getFid()) - .flagDebt5000(debt500) - .flag90days(paymt90) + .flagDebt5000(debt5000) + .flag90days(flagDays90) + .calculatedAt(calculatedAt) .build(); }).toList(); - phoneFidMapMapper.register(phoneFidMaps); + + List nullObjs = phoneFidMaps.stream().filter(Objects::isNull).toList(); + log.debug("null objs: {}", nullObjs.size()); + if (!nullObjs.isEmpty()) { + int i = 0; + i++; + } + String json = null; + try { + json = mapper.writeValueAsString( + phoneFidMaps.stream() + .filter(Objects::nonNull) + .map(p -> new HashMap() {{ + put("subject_id", p.getSubjectId()); + put("fid", p.getFid()); + put("phone", p.getPhone()); + put("package_id", p.getPackageId()); + put("flag_debt5000", p.getFlagDebt5000()); + put("flag_90_12", p.getFlag90days()); + put("calculated_at", p.getCalculatedAt()); + }} + ).toList()); + } catch (Exception e) { + throw new RuntimeException(e); + } + phoneFidMapMapper.registerJson(json); } public void setPollingQueue(BlockingQueue queue) { diff --git a/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml b/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml index e4c401e..3e1a923 100644 --- a/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml +++ b/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml @@ -45,7 +45,7 @@ with UR - SELECT FID as fid, (case when (sum(case when AMT_PAST_DUE > 5000 then 1 else 0 end) = 0) then 0 else 1 end) as flag diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 4b7d386..4f763e9 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -26,13 +26,27 @@ - INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id) + INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at) values - (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}) + (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}, #{phoneFid.flagDebt5000}, #{phoneFid.flag90days}, #{phoneFid.calculatedAt}) + + INSERT INTO signals_phone_fid_map + (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at) + SELECT subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at + FROM json_to_recordset(#{json}::json) AS t( + subject_id varchar(255), + fid int, + phone text, + package_id bigint, + flag_debt5000 smallint, + flag_90_12 smallint, + calculated_at timestamp) + + INSERT INTO signals_phone_fid_map (subject_id, phone, package_id) values