From 2709eaeb19556a3d1f531d6e7d852c3e102c38d7 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Tue, 30 Jun 2026 16:51:28 +0300 Subject: [PATCH] LP: added flags update by cron --- .../entities/app/LegalsMap.java | 2 +- .../mapper/signal/LegalsMapMapper.java | 8 +++ .../nbch/credit_tracker/model/FlagData.java | 2 + .../mode/common/UpdatingFlagsService.java | 46 ++++++++++------- .../monitoring/KafkaMonitoringService.java | 4 +- .../service/signal/PackageRecordsService.java | 17 ++++++- .../service/task/impl/FlagsEnricher.java | 35 +++++++++++-- .../mybatis/mapper/signal/LegalsMapMapper.xml | 49 +++++++++++++++++++ .../service/task/impl/FlagsEnricherTest.java | 20 ++++---- 9 files changed, 148 insertions(+), 35 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java b/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java index 9f14760..fb2d9c7 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java @@ -12,7 +12,7 @@ import java.time.LocalDateTime; public class LegalsMap { private Long id; private Long inn; - private Long regNUm; + private Long regNum; private Integer fid; private LocalDateTime searchedAt; private Integer flag90days; diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java index 7c31c1b..c4f51ac 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java @@ -11,4 +11,12 @@ public interface LegalsMapMapper { List findByFids(@Param("fids") List fids); + void updateFlagCachedData(@Param("json") String json); + + List findEligibleForFlagsUpdate(@Param("lastId") Long lastId, + @Param("interval") Integer interval, + @Param("batchSize") Integer batchSize); + + List findEligibleForFlagsInsert(@Param("lastId") Long lastId, + @Param("batchSize") Integer batchSize); } diff --git a/src/main/java/ru/nbch/credit_tracker/model/FlagData.java b/src/main/java/ru/nbch/credit_tracker/model/FlagData.java index c2e7a96..d8f7a5a 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/FlagData.java +++ b/src/main/java/ru/nbch/credit_tracker/model/FlagData.java @@ -10,6 +10,8 @@ import lombok.Setter; @Builder public class FlagData { private Long phone; + private Long inn; + private Long regNum; private Integer flag_debt5000; private Integer flag_90_12; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java index d7a16c0..bfe9f7a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFlagsService.java @@ -5,7 +5,9 @@ import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; @@ -22,16 +24,18 @@ import java.util.concurrent.atomic.AtomicBoolean; public class UpdatingFlagsService { private final static int BATCH_SIZE = 1000; private final PhonesMapMapper phonesMapMapper; - private final FlagsEnricher flagsEnricherV2; + private final LegalsMapMapper legalsMapMapper; + private final FlagsEnricher flagsEnricher; private final Integer updateDaysInterval; private final AtomicBoolean insertIsRunning = new AtomicBoolean(false); private final AtomicBoolean updateIsRunning = new AtomicBoolean(false); public UpdatingFlagsService(CreditTrackerSettings settings, - PhonesMapMapper phonesMapMapper, - FlagsEnricher flagsEnricherV2) { + PhonesMapMapper phonesMapMapper, LegalsMapMapper legalsMapMapper, + FlagsEnricher flagsEnricher) { this.phonesMapMapper = phonesMapMapper; - this.flagsEnricherV2 = flagsEnricherV2; + this.legalsMapMapper = legalsMapMapper; + this.flagsEnricher = flagsEnricher; this.updateDaysInterval = settings.getDaysToUpdateFlags(); } @@ -43,19 +47,23 @@ public class UpdatingFlagsService { } try { log.info("start update flags processor"); - long lastId = 0L; + long lastPhoneId = 0L; + long lastLegalId = 0L; int batchNumber = 1; List phonesMaps; + List legalsMaps; while (true) { - phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, updateDaysInterval, BATCH_SIZE); - log.trace("update batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size()); - if (phonesMaps.isEmpty()) { + phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastPhoneId, updateDaysInterval, BATCH_SIZE); + legalsMaps = legalsMapMapper.findEligibleForFlagsUpdate(lastLegalId, updateDaysInterval, BATCH_SIZE); + log.trace("update batch {}, signals_phones_map records loaded: {}. signals_legals_map records loaded: {}.", batchNumber, phonesMaps.size(), legalsMaps.size()); + if (phonesMaps.isEmpty() && legalsMaps.isEmpty()) { break; } - flagsEnricherV2.accept(phonesMaps); - lastId = phonesMaps.getLast().getId(); - log.trace("update batch {}, lastId {}", batchNumber++, lastId); + flagsEnricher.accept(phonesMaps, legalsMaps); + lastPhoneId = phonesMaps.getLast().getId(); + lastLegalId = legalsMaps.getLast().getId(); + log.trace("update batch {}, lastPhoneId {}, lastLegalId {}", batchNumber++, lastPhoneId, lastLegalId); } log.info("end update flags processor"); } finally { @@ -71,19 +79,23 @@ public class UpdatingFlagsService { } try { log.info("start processing new flags"); - long lastId = 0L; + long lastPhoneId = 0L; + long lastLegalId = 0L; int batchNumber = 1; List phonesMaps; + List legalsMaps; while (true) { - phonesMaps = phonesMapMapper.findEligibleForFlagsInsert(lastId, BATCH_SIZE); - log.trace("new batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size()); + phonesMaps = phonesMapMapper.findEligibleForFlagsInsert(lastPhoneId, BATCH_SIZE); + legalsMaps = legalsMapMapper.findEligibleForFlagsInsert(lastLegalId, BATCH_SIZE); + log.trace("new batch {}, signals_phones_map records loaded: {}. signals_legals_map records loaded: {}.", batchNumber, phonesMaps.size(), legalsMaps.size()); if (phonesMaps.isEmpty()) { break; } - flagsEnricherV2.accept(phonesMaps); - lastId = phonesMaps.getLast().getId(); - log.trace("new batch {}, lastId {}", batchNumber++, lastId); + flagsEnricher.accept(phonesMaps, legalsMaps); + lastPhoneId = phonesMaps.getLast().getId(); + lastLegalId = legalsMaps.getLast().getId(); + log.trace("new batch {}, lastPhoneId {}, lastLegalId {}", batchNumber++, lastPhoneId, lastLegalId); } log.info("end new flags processor"); } finally { diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index ae8567c..c2fe615 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -206,7 +206,7 @@ public class KafkaMonitoringService { .collect(Collectors.groupingBy(LegalsMap::getFid)); Set> uniqueInnRegnumPairs = legalsMaps.stream() - .map(l -> Pair.of(l.getInn(), l.getRegNUm())) + .map(l -> Pair.of(l.getInn(), l.getRegNum())) .collect(Collectors.toSet()); // прелоад по memberCode категорий @@ -243,7 +243,7 @@ public class KafkaMonitoringService { LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate(); // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются - List userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNUm())); + List userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum())); if (userPackageLegalDataList == null) { continue; diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java index 9a0f77b..ca4c436 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java @@ -113,7 +113,7 @@ public class PackageRecordsService { } } - public void updateFlagCache(List flagData) { + public void updateNpFlagCache(List flagData) { try { String json = entityToJsonMapper.writeValueAsString(flagData.stream() .filter(Objects::nonNull) @@ -128,6 +128,21 @@ public class PackageRecordsService { } } + public void updateLpFlagCache(List flagData) { + try { + String json = entityToJsonMapper.writeValueAsString(flagData.stream() + .filter(Objects::nonNull) + .map(phoneMap -> new HashMap() {{ + put("inn", phoneMap.getInn()); + put("reg_num", phoneMap.getRegNum()); + put("flag_90_12", phoneMap.getFlag_90_12()); + }}).toList()); + legalsMapMapper.updateFlagCachedData(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public PhonesMap findCachedByPhone(Long phone) { return phonesMapMapper.findByPhone(phone); } 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 188de32..0bbf704 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 @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.service.task.impl; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.model.FlagData; import ru.nbch.credit_tracker.service.indic.FlagsService; @@ -15,6 +16,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import java.util.stream.Stream; @Slf4j @@ -43,12 +45,18 @@ public class FlagsEnricher { ); } - public void process(List phonesMaps) { + public void process(List phonesMaps, List legalsMaps) { log.debug("FlagsEnricher handle: size: {}", phonesMaps.size()); List noFlagDataFids = phonesMaps.stream() .map(PhonesMap::getFid) - .toList(); + .collect(Collectors.toList()); + + List ligalsNoFlagDataFids = legalsMaps.stream() + .map(LegalsMap::getFid) + .toList(); + + noFlagDataFids.addAll(ligalsNoFlagDataFids); LocalDate yearAgo = LocalDate.now().minusYears(1); @@ -72,13 +80,30 @@ public class FlagsEnricher { .build(); return Stream.of(flagData); }).toList(); - packageRecordsService.updateFlagCache(flagCacheOnUpdate); + + + List lpFlagCacheOnUpdate = legalsMaps.stream() + .flatMap(legalMap -> { + Integer fid = legalMap.getFid(); + int pastDue90 = flag90ByPastDueArrear.getOrDefault(fid, 0); + int paymt90 = flag90ByRuPaymt.getOrDefault(fid, 0); + int allArrears90 = flag90ByRuAllArrear.getOrDefault(fid, 0); + Integer flagDays90 = Math.max(allArrears90, Math.max(pastDue90, paymt90)); + FlagData flagData = FlagData.builder() + .flag_90_12(flagDays90) + .inn(legalMap.getInn()) + .regNum(legalMap.getRegNum()) + .build(); + return Stream.of(flagData); + }).toList(); + + packageRecordsService.updateNpFlagCache(flagCacheOnUpdate); } - public void accept(List phonesMaps) { + public void accept(List phonesMaps, List legalsMaps) { ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool; log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size()); - this.ioPool.submit(() -> process(phonesMaps)); + this.ioPool.submit(() -> process(phonesMaps, legalsMaps)); } } diff --git a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml index fdcf6e5..062c5b7 100644 --- a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml @@ -32,5 +32,54 @@ + + + + + + WITH src AS (SELECT DISTINCT ON (inn, reg_num) inn, + reg_num, + flag_90_12 + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num text, + flag_90_12 smallint + ) + ORDER BY inn, reg_num), + to_lock AS (SELECT dst.id + FROM signals_legals_map dst + JOIN src ON src.inn = dst.inn AND src.reg_num = dst.reg_num + ORDER BY dst.inn, dst.reg_num + FOR UPDATE) + + UPDATE signals_legals_map dst + SET flag_90_12 = src.flag_90_12, + calculated_at = now() + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num text, + flag_90_12 smallint + ) + WHERE dst.inn = src.inn + AND dst.reg_num = src.reg_num + AND dst.id IN (SELECT id FROM to_lock); + \ No newline at end of file diff --git a/src/test/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherTest.java b/src/test/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherTest.java index 1005732..72a2283 100644 --- a/src/test/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherTest.java +++ b/src/test/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherTest.java @@ -9,6 +9,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.settings.FlagUpdaterSettings; +import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.model.FlagData; import ru.nbch.credit_tracker.service.indic.FlagsService; @@ -125,9 +126,9 @@ public class FlagsEnricherTest { when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))) .thenReturn(flag90ByRuAllArrears); - flagsEnricher.process(phonesMaps); + flagsEnricher.process(phonesMaps, List.of()); - verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture()); + verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture()); List actualFlagCache = flagCacheCaptor.getValue(); assertNotNull(actualFlagCache); @@ -148,15 +149,16 @@ public class FlagsEnricherTest { @Test void process_withEmptyList_shouldCallUpdateFlagCacheWithEmptyList() { List emptyList = List.of(); + List legalsEmptyList = List.of(); when(flagsService.findFlagsDebt5000(anyList())).thenReturn(Map.of()); when(flagsService.findFlag90ByPastDueArrear(anyList(), any(LocalDate.class))).thenReturn(Map.of()); when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(Map.of()); when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(Map.of()); - flagsEnricher.process(emptyList); + flagsEnricher.process(emptyList, legalsEmptyList); - verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture()); + verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture()); List actualFlagCache = flagCacheCaptor.getValue(); assertTrue(actualFlagCache.isEmpty()); } @@ -178,9 +180,9 @@ public class FlagsEnricherTest { when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(zeroMap); when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(zeroMap); - flagsEnricher.process(phonesMaps); + flagsEnricher.process(phonesMaps, List.of()); - verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture()); + verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture()); List actualFlagCache = flagCacheCaptor.getValue(); assertEquals(2, actualFlagCache.size()); @@ -210,9 +212,9 @@ public class FlagsEnricherTest { when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(flag90ByRuPaymt); when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(flag90ByRuAllArrears); - flagsEnricher.process(phonesMaps); + flagsEnricher.process(phonesMaps,List.of()); - verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture()); + verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture()); List actualFlagCache = flagCacheCaptor.getValue(); assertFlagData(actualFlagCache, 79123456781L, 1, 1); @@ -228,7 +230,7 @@ public class FlagsEnricherTest { when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(Map.of()); when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(Map.of()); - flagsEnricher.process(phonesMaps); + flagsEnricher.process(phonesMaps, List.of()); verify(flagsService).findFlagsDebt5000(expectedFids); verify(flagsService).findFlag90ByPastDueArrear(eq(expectedFids), any(LocalDate.class));