From d5c6e7e08fa2e90628b26e971fe2b9af0ea5abc8 Mon Sep 17 00:00:00 2001 From: ialbert Date: Thu, 27 Nov 2025 21:11:32 +0300 Subject: [PATCH] updating flags --- .../mapper/signal/FlagDetailMapper.java | 3 ++ .../credit_tracker/service/SignalService.java | 40 ++++++++++++++++++- .../service/indic/FlagsService2.java | 27 +++++++++++-- .../service/task/impl/FidNameEnricher.java | 2 + .../service/task/impl/FlagsEnricher.java | 22 +++++++++- .../mapper/signal/FlagDetailMapper.xml | 26 ++++++++++++ src/main/resources/sql/v2.21.ddl | 5 ++- 7 files changed, 117 insertions(+), 8 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java index 1b9436d..f912043 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.mapper.signal; +import java.util.List; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import ru.nbch.credit_tracker.entities.app.FlagDetail; @@ -7,5 +8,7 @@ import ru.nbch.credit_tracker.entities.app.FlagDetail; @Mapper public interface FlagDetailMapper { void registerCalcFlags(@Param("json") String json); + void updateCalcFlags(@Param("json") String json); FlagDetail getByMapId(@Param("mapId") Long mapId); + void deleteFlags(@Param("mapIds") List mapIds); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index 9b8b6c1..5c4f2f0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -29,6 +29,8 @@ import ru.nbch.credit_tracker.entities.external_request.signals.Person; import ru.nbch.credit_tracker.entities.external_request.signals.Signals; import ru.nbch.credit_tracker.entities.external_response.monitoring_report_error.MonitoringReportErrors; import ru.nbch.credit_tracker.enums.Status; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.FlagsService; import ru.nbch.credit_tracker.service.indic.PersonService; import ru.nbch.credit_tracker.service.indic.PhoneService; @@ -36,6 +38,8 @@ import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.signal.UserPackageDao; +import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; +import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; import ru.nbch.credit_tracker.utils.CTUtil; @@ -55,8 +59,10 @@ public class SignalService { private final FlagsService flagsService; private final PersonService personService; private final StaxXmlProcessor staxXmlProcessor; + private final FidNameEnricher fidNameEnricher; + private final FlagsEnricher flagsEnricher; - private final int batchSize = 10000; + private final int batchSize; public SignalService(CreditTrackerSettings config, HitsService hitsService, @@ -66,7 +72,7 @@ public class SignalService { PhoneService phoneService, FlagsService flagsService, PersonService personService, - StaxXmlProcessor staxXmlProcessor) { + StaxXmlProcessor staxXmlProcessor, FidNameEnricher fidNameEnricher, FlagsEnricher flagsEnricher) { this.config = config; this.hitsService = hitsService; this.userPackageService = userPackageService; @@ -76,6 +82,36 @@ public class SignalService { this.flagsService = flagsService; this.personService = personService; this.staxXmlProcessor = staxXmlProcessor; + this.batchSize = config.getPipeline().subjectBatchSize(); + this.fidNameEnricher = fidNameEnricher; + this.flagsEnricher = flagsEnricher; + } + + public UpdateResult updateFlags2(UserPackage userPackage) { + log.debug("updating flags for package [{} {}]", userPackage.getPackageName(), userPackage.getId()); + boolean updated = false; + boolean noFidsFound = true; + Long packageId = userPackage.getId(); + int offset = 0; + List page = phoneService2.pageByPackageId(packageId, batchSize, offset); + int counter = 1; + while (!page.isEmpty()) { + log.debug("got page number {}, size {}", counter++, page.size()); + List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() + .mapId(phoneFidMap.getId()) + .identifier(phoneFidMap.getSubjectId()) + .phone(phoneFidMap.getPhone()) + .build()).toList(); + SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles); + log.debug("page number {}: updating fids...", counter); + fidNameEnricher.handle(payload); + log.debug("page number {}: updating flags...", counter); + flagsEnricher.process(payload, true); + offset += batchSize; + log.debug("loading next page..."); + page = phoneService2.pageByPackageId(packageId, batchSize, offset); + } + return new UpdateResult(updated, noFidsFound); } /** 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 ccd1171..22f1399 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 @@ -1,12 +1,12 @@ package ru.nbch.credit_tracker.service.indic; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; import java.time.LocalDate; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.app.FlagDetail; @@ -44,6 +44,28 @@ public class FlagsService2 { .collect(Collectors.toMap(PaymentsFlag::getFid, PaymentsFlag::getFlag)); } + public void updateCalculationFlags(List flagDetails) { + try { + //fixme, временное решение + String jsonFlagDetail = entityToJsonMapper.writeValueAsString(flagDetails.stream() + .map(flagDetail -> new HashMap() {{ + put("fid", flagDetail.getFid()); + put("map_id", flagDetail.getMapId()); + put("flag_debt5000", flagDetail.getFlagDebt5000()); + put("flag_90_12", flagDetail.getFlag90days()); + put("calculated_at", flagDetail.getCalculatedAt()); + }}).toList()); + + flagDetailMapper.updateCalcFlags(jsonFlagDetail); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + + public void deleteFlags(List mapIds) { + flagDetailMapper.deleteFlags(mapIds); + } + public void registerCalculationFlags(List flagDetails) { try { //fixme, временное решение @@ -62,7 +84,6 @@ public class FlagsService2 { } catch (JsonProcessingException e) { throw new RuntimeException(e); } - } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java index 3f099a2..dda9a5f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java @@ -51,6 +51,8 @@ public class FidNameEnricher implements Handler subject.setFid(name.getFid()); subject.setNameData(nameData); + } else { + subject.setFid(null); } return PhoneFidMap.builder() 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 334f75f..6fef2ec 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 @@ -14,7 +14,6 @@ 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.FlagDetail; -import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.FlagsService2; @@ -43,6 +42,10 @@ public class FlagsEnricher implements Consumer { } public void process(SubjectPayload subjectPayload) { + process(subjectPayload, false); + } + + public void process(SubjectPayload subjectPayload, boolean update) { List subjects = subjectPayload.getSubjects(); log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), subjects.size()); @@ -84,7 +87,22 @@ public class FlagsEnricher implements Consumer { .calculatedAt(finalCalculatedAt) .build(); }).toList(); - flagsService.registerCalculationFlags(flagDetails); + if (update) { + log.debug("updating calculated flags size {}", flagDetails.size()); + flagsService.updateCalculationFlags(flagDetails); + //если signals_phone_fid_map#fid == null, удалить signals_flags с таким map_id + List phoneFidMapIdsWithNullFid = subjects.stream() + .filter(subject -> subject.getFid() == null) + .map(SubjectProfile::getMapId) + .filter(Objects::nonNull) + .toList(); + if (!phoneFidMapIdsWithNullFid.isEmpty()) { + log.debug("deleting flags for signals_phone_fid_map#fid == null; amount: {}", phoneFidMapIdsWithNullFid.size()); + flagsService.deleteFlags(phoneFidMapIdsWithNullFid); + } + } else { + flagsService.registerCalculationFlags(flagDetails); + } } @Override diff --git a/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml b/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml index 0611123..d9277a7 100644 --- a/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml @@ -28,6 +28,32 @@ calculated_at timestamp) + + INSERT INTO signals_flags + (map_id, fid, flag_debt5000, flag_90_12, calculated_at) + SELECT map_id, fid, flag_debt5000, flag_90_12, calculated_at + FROM json_to_recordset(#{json}::json) AS t( + map_id bigint, + fid int, + flag_debt5000 smallint, + flag_90_12 smallint, + calculated_at timestamp) + ON CONFLICT (map_id) DO UPDATE + SET + fid = EXCLUDED.fid, + flag_debt5000 = EXCLUDED.flag_debt5000, + flag_90_12 = EXCLUDED.flag_90_12, + calculated_at = EXCLUDED.calculated_at; + + + + DELETE FROM signals_flags + WHERE map_id IN + + #{mapId} + + +