From bc34f3f89c9f454c5b445be4b44629eae78fd9bf Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Sun, 12 Oct 2025 20:43:25 +0300 Subject: [PATCH] Added FlagsEnricher; Added id to SubjectProfile to use it as uid in PersonFactoryBuilderV2 --- credit-tracker.properties | 10 ++-- .../builder/PersonFactoryBuilderV2.java | 11 ++--- .../config/task/ConsumerConfigV2.java | 34 ++++++++----- .../entities/dto/SubjectProfile.java | 1 + .../mapper/signal/PhoneFidMapMapper.java | 6 ++- .../credit_tracker/model/DummyPayload.java | 17 +++++++ .../credit_tracker/model/FlagsPayload.java | 38 +++++++++++++++ .../service/indic/FlagsService2.java | 14 ++++++ .../service/task/FidNameEnricherV2.java | 21 ++++---- .../service/task/FlagsEnricher.java | 48 +++++++++++++++++++ .../mapper/signal/PhoneFidMapMapper.xml | 12 ++++- 11 files changed, 174 insertions(+), 38 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java diff --git a/credit-tracker.properties b/credit-tracker.properties index e8241c1..ca9ec6c 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -8,9 +8,9 @@ server.tomcat.accesslog.pattern=%h %l %u %t "%r" %s %b # datasources # indic ct.indic-data-source-settings.driver=com.ibm.db2.jcc.DB2Driver -ct.indic-data-source-settings.url=${DB_URL_INDIC:jdbc:db2://10.230.227.101:2668/cprosd22:currentSchema=INDIC;} -ct.indic-data-source-settings.username=${USERNAME_INDIC:tester} -ct.indic-data-source-settings.password=${PASSWORD_INDIC:tester} +ct.indic-data-source-settings.url=${DB_URL_INDIC:jdbc:db2://10.230.227.100:2668/cprosd22:currentSchema=INDIC;} +ct.indic-data-source-settings.username=${USERNAME_INDIC:ros9appl} +ct.indic-data-source-settings.password=${PASSWORD_INDIC:R0s9Appl} ct.indic-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_INDIC:10} ct.indic-data-source-settings.minimum-idle=${MINIMUM_IDLE_INDIC:10} ct.indic-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_INDIC:10000} @@ -60,8 +60,8 @@ ct.signals-root-uri=${SIGNALS_ROOT_URI:http://10.230.227.102:11094/signalRUalfa/ ct.signals-token=${SIGNALS_TOKEN:gttcHxwcGojoAvZK} ct.signals-user=${SIGNALS_USER:0001ZZ000001} -ct.load-dir=/opt/app/reports -#ct.load-dir=D:\\folder_name\\signals +#ct.load-dir=/opt/app/reports +ct.load-dir=D:\\folder_name\\signals # min ct.monitoring-interval=${MONITORING_INTERVAL:15} diff --git a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java index 393d03a..a29cf97 100644 --- a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java +++ b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java @@ -1,7 +1,5 @@ package ru.nbch.credit_tracker.builder; -import java.time.LocalDate; -import java.time.format.DateTimeFormatter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import ru.nbch.credit_tracker.entities.dto.IdProfile; @@ -10,6 +8,9 @@ import ru.nbch.credit_tracker.entities.dto.SubjectProfile; import ru.nbch.credit_tracker.entities.external_request.signals.Person; import ru.nbch.credit_tracker.utils.CTUtil; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; + @Slf4j public class PersonFactoryBuilderV2 { private final static DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("dd.MM.yyyy"); @@ -39,11 +40,7 @@ public class PersonFactoryBuilderV2 { public PersonFactoryBuilderV2 personData(SubjectProfile personData) { this.personData = personData; - return this; - } - - public PersonFactoryBuilderV2 uid(Integer uid) { - this.uid = uid; + this.uid = personData.getId(); return this; } diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java b/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java index 2b8a357..618de65 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java @@ -1,16 +1,12 @@ package ru.nbch.credit_tracker.config.task; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import ru.nbch.credit_tracker.model.*; +import ru.nbch.credit_tracker.service.task.*; + import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import ru.nbch.credit_tracker.model.PhoneFidPayloadV2; -import ru.nbch.credit_tracker.model.SendSignalsPayload; -import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2; -import ru.nbch.credit_tracker.service.task.FidIdEnricherV2; -import ru.nbch.credit_tracker.service.task.FidNameEnricherV2; -import ru.nbch.credit_tracker.service.task.TaskConsumerV2; -import ru.nbch.credit_tracker.service.task.TaskPayloadV2; @Configuration public class ConsumerConfigV2 { @@ -19,6 +15,12 @@ public class ConsumerConfigV2 { public BlockingQueue savingRequest2() { return new LinkedBlockingQueue<>(); } + + @Bean + public BlockingQueue flagsPayload() { + return new LinkedBlockingQueue<>(); + } + @Bean public BlockingQueue enricherQueue2() { return new LinkedBlockingQueue<>(); @@ -37,19 +39,25 @@ public class ConsumerConfigV2 { @SuppressWarnings("unchecked") @Bean public Integer consumerConfigurer(BlockingQueue savingRequest2, - BlockingQueue enricherQueue2, - BlockingQueue enricherIdQueue2, - FidNameEnricherV2 fidNameEnricherV2, - FidIdEnricherV2 fidIdEnricherV2) { + BlockingQueue enricherQueue2, + BlockingQueue enricherIdQueue2, + BlockingQueue flagsPayloadQueue, + FidNameEnricherV2 fidNameEnricherV2, + FidIdEnricherV2 fidIdEnricherV2, + FlagsEnricher flagsEnricher) { SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true); SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true); + DummyPayload FLAGS_POISON = new DummyPayload(true); TaskConsumerV2 fidNameEnricher = new TaskConsumerV2(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON); TaskConsumerV2 fidIdEnricher = new TaskConsumerV2(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON); + TaskConsumerV2 flagsEnricherConsumer = new TaskConsumerV2(flagsEnricher, flagsPayloadQueue, new LinkedBlockingQueue(), FLAGS_POISON); Thread thread1 = new Thread(fidNameEnricher::start); Thread thread2 = new Thread(fidIdEnricher::start); + Thread thread3 = new Thread(flagsEnricherConsumer::start); thread1.start(); thread2.start(); + thread3.start(); return 0; } } diff --git a/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java b/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java index d1ec4ad..03aaeac 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java @@ -6,6 +6,7 @@ import lombok.Setter; @Getter @Setter public class SubjectProfile { + private Integer id; private Integer fid; private IdProfile idProfile; private NameProfile nameProfile; 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 739158b..490eb83 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 @@ -1,17 +1,19 @@ package ru.nbch.credit_tracker.mapper.signal; -import java.util.Collection; -import java.util.List; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject; +import java.util.Collection; +import java.util.List; + @Mapper public interface PhoneFidMapMapper { void registerSubjects(@Param("subjects") List subjects, @Param("packageId") Integer packageId); void register(@Param("phoneFidMap") List phoneFidMap, @Param("packageId") Integer packageId); + void registerWithId(@Param("phoneFidMap") List phoneFidMap); List findFullPhoneFidMap(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit); diff --git a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java b/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java new file mode 100644 index 0000000..b8166d7 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java @@ -0,0 +1,17 @@ +package ru.nbch.credit_tracker.model; + +import ru.nbch.credit_tracker.service.task.TaskPayloadV2; + +public class DummyPayload implements TaskPayloadV2 { + + private final boolean poison; + + public DummyPayload(boolean poison) { + this.poison = poison; + } + + @Override + public boolean isPoison() { + return poison; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java b/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java new file mode 100644 index 0000000..9c4b167 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java @@ -0,0 +1,38 @@ +package ru.nbch.credit_tracker.model; + +import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; +import ru.nbch.credit_tracker.service.task.TaskPayloadV2; + +import java.util.Map; + +public class FlagsPayload implements TaskPayloadV2 { + private final Integer packageId; + private final Map fidsMapByBatch; + private final boolean poison; + + public FlagsPayload(Integer packageId, Map fidsMapByBatch) { + this.packageId = packageId; + this.fidsMapByBatch = fidsMapByBatch; + this.poison = false; + } + + public FlagsPayload(boolean poison) { + this.packageId = null; + this.fidsMapByBatch = null; + this.poison = poison; + } + + public Integer getPackageId() { + return packageId; + } + + public Map getFidsMapByBatch() { + return fidsMapByBatch; + } + + @Override + public boolean isPoison() { + return poison; + } + +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java index 629b181..045003d 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java @@ -5,12 +5,14 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import java.math.BigDecimal; import java.math.BigInteger; import java.sql.ResultSet; import java.sql.SQLException; import java.time.LocalDate; +import java.time.LocalDateTime; import java.util.Comparator; import java.util.HashMap; import java.util.List; @@ -40,6 +42,18 @@ public class FlagsService2 { this.jdbcTemplate = jdbcTemplate; } + public void fillFlagsData(Map fidsMapByBatch) { + List fids = fidsMapByBatch.keySet().stream().toList(); + Map fidsflagsDataMap = findFlags(fids); + + fidsflagsDataMap.forEach((fid, data) -> { + PhoneFidMap phoneFidMap = fidsMapByBatch.get(fid); + phoneFidMap.setFlagDebt5000(data.getFlagDebt5000()); + phoneFidMap.setFlag90days(data.getFlag90days()); + phoneFidMap.setCalculatedAt(LocalDateTime.now()); + }); + } + public Map findFlags(List fids) { if (fids.isEmpty()) { return new HashMap<>(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java index 79c3c17..69dee34 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java @@ -1,45 +1,48 @@ package ru.nbch.credit_tracker.service.task; -import java.util.List; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.dto.NameProfile; import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import ru.nbch.credit_tracker.entities.dto.SubjectProfile; import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; +import ru.nbch.credit_tracker.model.FlagsPayload; import ru.nbch.credit_tracker.model.PhoneFidPayloadV2; import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2; import ru.nbch.credit_tracker.service.indic.PhoneService2; -import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; @Service @Slf4j public class FidNameEnricherV2 implements DispatcherV2 { private final PhoneService2 phoneService2; - private final PhoneFidMapService phoneFidMapService; private final PhoneFidMapMapper mapper; + private final BlockingQueue flagsPayloadQueue; public FidNameEnricherV2(PhoneService2 phoneService2, - PhoneFidMapService phoneFidMapService, - PhoneFidMapMapper mapper) { + PhoneFidMapMapper mapper, + BlockingQueue flagsPayloadQueue) { this.phoneService2 = phoneService2; - this.phoneFidMapService = phoneFidMapService; this.mapper = mapper; + this.flagsPayloadQueue = flagsPayloadQueue; } @Override public SubjectsProfilePayloadV2 handle(PhoneFidPayloadV2 taskPayload) { List phoneFidMap = taskPayload.getPhoneFidsBatch(); log.debug("FidNameEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMap.size()); -// phoneFidMapService.register(phoneFidMap, taskPayload.getPackageId()); + mapper.registerWithId(phoneFidMap); // надо зарегистрировать phoneFidMap'ы т.к далее id участвует в формированиии запроса в оф. сервис uid = id Map fidsMapByBatch = phoneService2.defineFidsAndNameByPhone(phoneFidMap); log.debug("Found {} fids by batch", fidsMapByBatch.size()); + flagsPayloadQueue.add(new FlagsPayload(taskPayload.getPackageId(), fidsMapByBatch)); // TODO find out how to use it in other way //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать -// mapper.updateFid(fidsMapByBatch.values()); List subjectProfiles = fidsMapByBatch.values().stream().map(pf -> { SubjectProfile subjectProfile = new SubjectProfile(); + subjectProfile.setId(pf.getId()); subjectProfile.setFid(pf.getFid()); subjectProfile.setNameProfile(NameProfile.builder() .firstName(pf.getFirstName()) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java new file mode 100644 index 0000000..bcd2b26 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java @@ -0,0 +1,48 @@ +package ru.nbch.credit_tracker.service.task; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; +import ru.nbch.credit_tracker.model.DummyPayload; +import ru.nbch.credit_tracker.model.FlagsPayload; +import ru.nbch.credit_tracker.service.indic.FlagsService2; + +import java.util.List; +import java.util.Map; + +@Service +@Slf4j +public class FlagsEnricher implements DispatcherV2{ + + private final FlagsService2 flagsService; + private final PhoneFidMapMapper mapper; + + public FlagsEnricher(FlagsService2 flagsService, + PhoneFidMapMapper mapper) { + this.flagsService = flagsService; + this.mapper = mapper; + } + + @Override + public DummyPayload handle(FlagsPayload taskPayload) { + Map fidsMapByBatch = taskPayload.getFidsMapByBatch(); + log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), fidsMapByBatch.size()); + flagsService.fillFlagsData(fidsMapByBatch); + + List subjects = fidsMapByBatch.values().stream().toList(); + + log.debug("Start updating subject's batch"); + if (!subjects.isEmpty()) { + int updateBatchSize = 7000; + for (int j = 0; j < subjects.size(); j += updateBatchSize) { + int endIndex = Math.min(j + updateBatchSize, subjects.size()); + mapper.updatePhoneFidMaps(subjects.subList(j, endIndex)); + log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, subjects.size(), taskPayload.getPackageId()); + } + } + log.debug("Complete updating subject's batch"); + + return new DummyPayload(false); + } +} diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 6b97082..aacf6a6 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -26,10 +26,18 @@ - INSERT INTO signals_phone_fid_map (rec_uuid, subject_id, phone, package_id) + INSERT INTO signals_phone_fid_map (subject_id, phone, package_id) values - (#{phoneFid.recUuid}, #{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId}) + (#{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId}) + + + + + INSERT INTO signals_phone_fid_map (subject_id, phone, package_id) + values + + (#{phoneFid.subjectId}, #{phoneFid.phone}, #{phoneFid.packageId})