From e94df35c76d0f128b43f05a580cc97ca25fc550f Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 2 Jul 2026 15:37:08 +0300 Subject: [PATCH] LP: added LpFidEnricher --- .../mapper/indic/PersonMapper.java | 4 +- .../mapper/signal/LegalsMapMapper.java | 4 + .../nbch/credit_tracker/model/FidIdData.java | 11 ++ .../credit_tracker/model/SubjectProfile.java | 9 +- .../service/indic/PersonService.java | 29 ++++ .../service/signal/PackageRecordsService.java | 33 ++++ .../service/task/LpFidEnricher.java | 154 ++++++++++++++++++ .../service/task/impl/NpFidEnricher.java | 5 +- .../ru/nbch/credit_tracker/utils/CTUtil.java | 8 + .../mybatis/mapper/indic/PersonMapper.xml | 23 +++ .../mapper/signal/LegalsFidMapMapper.xml | 2 +- .../mybatis/mapper/signal/LegalsMapMapper.xml | 44 +++++ 12 files changed, 320 insertions(+), 6 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/model/FidIdData.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java index e90c225..9ed2274 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.mapper.indic; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import ru.nbch.credit_tracker.entities.indic.Name; +import ru.nbch.credit_tracker.model.FidIdData; import java.util.List; @@ -11,6 +12,7 @@ public interface PersonMapper { List findNameData(@Param("phones") List phones); - List findNameDataByFids(@Param("fids") List fids); + List findFidsByInn(@Param("inns") List inns); + List findFidsByRegNum(@Param("regNums") List regNums); } 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 c4f51ac..4b44176 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 @@ -9,6 +9,8 @@ import java.util.List; @Mapper public interface LegalsMapMapper { + void registerJson(@Param("json") String json); + List findByFids(@Param("fids") List fids); void updateFlagCachedData(@Param("json") String json); @@ -19,4 +21,6 @@ public interface LegalsMapMapper { List findEligibleForFlagsInsert(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); + + List findCachedByInnAndRegNUm(@Param("json") String json); } diff --git a/src/main/java/ru/nbch/credit_tracker/model/FidIdData.java b/src/main/java/ru/nbch/credit_tracker/model/FidIdData.java new file mode 100644 index 0000000..801c6a1 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/FidIdData.java @@ -0,0 +1,11 @@ +package ru.nbch.credit_tracker.model; + +import lombok.Getter; +import lombok.Setter; + +@Getter +@Setter +public class FidIdData { + private Integer fid; + private String idNum; +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java b/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java index 8a2a04d..e9cedf6 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java +++ b/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java @@ -8,9 +8,16 @@ import lombok.Setter; @Setter @Builder public class SubjectProfile { + // common field private String identifier; private Long mapId; - private Long phone; private Integer fid; private FlagData flagData; + + // np fields + private Long phone; + + // lp fields + private Long inn; + private Long regNum; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java index 5fe9e31..b5229b9 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java @@ -5,9 +5,11 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.indic.Id; import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.mapper.indic.PersonMapper; +import ru.nbch.credit_tracker.model.FidIdData; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.utils.CTUtil; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -71,6 +73,33 @@ public class PersonService { )); } + public Map> findFidsByInn(List subjects) { + List inns = subjects.stream() + .map(s -> CTUtil.normalizeInnZeroesOnly(s.getInn())) + .collect(Collectors.toList()); + + if (inns.isEmpty()) { + return Collections.emptyMap(); + } + return mapper.findFidsByInn(inns) + .stream() + .collect(Collectors.groupingBy(FidIdData::getIdNum)); + } + + public Map> findFidsByRegNum(List subjects) { + List regNums = subjects.stream() + .map(s -> CTUtil.normalizeRegNumZeroesOnly(s.getRegNum())) + .collect(Collectors.toList()); + + if (regNums.isEmpty()) { + return Collections.emptyMap(); + } + return mapper.findFidsByRegNum(regNums) + .stream() + .collect(Collectors.groupingBy(FidIdData::getIdNum)); + } + + private boolean isDUL(Id p) { Integer idType; Double idType2; 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 e4cf0c3..3126937 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 @@ -53,6 +53,24 @@ public class PackageRecordsService { return phonesMaps.stream().collect(Collectors.toMap(PhonesMap::getPhone, Function.identity())); } + public Map, LegalsMap> findCachedLegalsByInnAndRegNums(List subjects) { + try { + List innAndRegNums = subjects.stream() + .map(s -> new InnRegnumPair(s.getInn(), s.getRegNum())) + .toList(); + String json = entityToJsonMapper.writeValueAsString(innAndRegNums); + + return legalsMapMapper.findCachedByInnAndRegNUm(json) + .stream() + .collect(Collectors.toMap( + lm -> Pair.of(lm.getInn(), lm.getRegNum()), + Function.identity()) + ); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public List findFullPhoneFidMap(Long packageId, Integer limit, Integer offset) { return packagesRecordsMapper.findFullPhoneFidMap(packageId, limit, offset); } @@ -71,6 +89,21 @@ public class PackageRecordsService { } } + public void saveLegalsToCache(List legalsMaps) { + try { + String json = entityToJsonMapper.writeValueAsString(legalsMaps.stream() + .filter(Objects::nonNull) + .map(legalsMap -> new HashMap() {{ + put("fid", legalsMap.getFid()); + put("inn", legalsMap.getInn()); + put("reg_num", legalsMap.getRegNum()); + }}).toList()); + legalsMapMapper.registerJson(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public List selectUniquePackageIds(List phones) { try { String json = entityToJsonMapper.writeValueAsString(phones.stream() diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java new file mode 100644 index 0000000..cf01db8 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java @@ -0,0 +1,154 @@ +package ru.nbch.credit_tracker.service.task; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.tuple.Pair; +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.model.FidIdData; +import ru.nbch.credit_tracker.model.FlagData; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PackageVariable; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.service.indic.PersonService; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.utils.CTUtil; +import ru.nbch.pipeline.engine.core.Handler; +import ru.nbch.pipeline.engine.core.PipelineContext; + +import java.util.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; + +/** + * Стадия для сопоставления FID к inn, reg_num и регистрации пакета в базе данных. + * Имеет функцию при режиме RESTORE у Pipeline, когда фиды были установлены, но данные нужно подтянуть снова + */ +@Service +@Slf4j +public class LpFidEnricher implements Handler { + private final PersonService personService; + private final CreditTrackerSettings creditTrackerSettings; + private final PackageRecordsService packageRecordsService; + + public LpFidEnricher(PersonService personService, + CreditTrackerSettings creditTrackerSettings, + PackageRecordsService packageRecordsService) { + this.personService = personService; + this.creditTrackerSettings = creditTrackerSettings; + this.packageRecordsService = packageRecordsService; + } + + @Override + public SubjectPayload handle(SubjectPayload taskPayload) { + List subjects = taskPayload.getSubjects(); + var ctx = obtainContext(PackageVariable.class, taskPayload); + Long packageId = ctx.get(PackageVariable.packageId); + log.debug("LpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size()); + //найдем inn + regNum в табличном кэше + Map, LegalsMap> cachedRecords = packageRecordsService.findCachedLegalsByInnAndRegNums(subjects); + processByNumber(subjects, cachedRecords, ctx); + return taskPayload; + } + + private void processByNumber(List subjects, + Map, LegalsMap> cachedRecords, + PipelineContext context) { + List noFidsData = subjects.stream() + .filter(sp -> !cachedRecords.containsKey(Pair.of(sp.getInn(), sp.getRegNum()))) + .toList(); + + Map> fidsByInnMap; + Map> fidsByRegNumMap; + if (!noFidsData.isEmpty()) { + fidsByInnMap = personService.findFidsByInn(noFidsData); + fidsByRegNumMap = personService.findFidsByRegNum(noFidsData); + } else { + fidsByInnMap = new HashMap<>(); + fidsByRegNumMap = new HashMap<>(); + } + AtomicInteger cacheHits = context.get(PackageVariable.cacheHits); + AtomicInteger cacheMisses = context.get(PackageVariable.cacheMisses); + + Set> seen = java.util.concurrent.ConcurrentHashMap.newKeySet(); + + List subjectForSave = subjects.stream() + .flatMap(subject -> { + Long inn = subject.getInn(); + Long regNum = subject.getRegNum(); + if (cachedRecords.containsKey(Pair.of(inn, regNum))) { + cacheHits.incrementAndGet(); + fillByCache(subject, cachedRecords.get(Pair.of(inn, regNum))); + } else { + cacheMisses.incrementAndGet(); + List fidByInn = fidsByInnMap.get(CTUtil.normalizeInnZeroesOnly(inn)); + List fidByRegNum = fidsByRegNumMap.get(CTUtil.normalizeRegNumZeroesOnly(regNum)); + + if (fidByInn == null || fidByInn.isEmpty()) { + if (noFidsData.isEmpty()) { + log.warn("Unexpected scenario...all recs are hit to cache, but nameData was trying to found in indic"); + } + return Stream.of(LegalsMap.builder() + .inn(inn) + .regNum(regNum) + .build()); + } + + List fidsByInn = fidByInn.stream().map(FidIdData::getFid).toList(); + + if (fidByRegNum != null && !fidByRegNum.isEmpty()) { + List fidsByRegNum = fidByRegNum.stream().map(FidIdData::getFid).toList(); + + // находим все пересечения по фидам + List intersection = new ArrayList<>(fidsByInn); + intersection.retainAll(fidsByRegNum); + + if (!intersection.isEmpty()) { + // берем максимальное пересечение + Integer maxFid = Collections.max(intersection); + + return Stream.of(LegalsMap.builder() + .inn(inn) + .regNum(regNum) + .fid(maxFid) + .build()); + } + } + + return Stream.of(LegalsMap.builder() + .inn(inn) + .regNum(regNum) + .build()); + } + return Stream.empty(); + }) + .filter(legalsMap -> seen.add(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()))) + .toList(); + + packageRecordsService.saveLegalsToCache(subjectForSave); + } + + private void fillByCache(SubjectProfile subjectProfile, LegalsMap cachedRecord) { + Optional.of(cachedRecord) + .ifPresent(cr -> subjectProfile.setFid(cr.getFid())); + + Optional.of(cachedRecord) + .filter(cr -> cachedRecord.getFlag90days() != null) + .map(cr -> Optional.of( + (FlagData.builder() + .flag_90_12(cr.getFlag90days()) + .build() + ))) + .ifPresent(flagData -> subjectProfile.setFlagData(flagData.get())); + } + + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); + } + + @Override + public SubjectPayload cookPoison() { + return SubjectPayload.POISON; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java index 9aa6fe3..e38d320 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java @@ -21,8 +21,7 @@ import java.util.stream.Stream; /** * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных. - * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки. - * Имеет функуию при режиме RESTORE у Pipeline, когда фиды были установлены, но данные нужно подтянуть снова + * Имеет функцию при режиме RESTORE у Pipeline, когда фиды были установлены, но данные нужно подтянуть снова */ @Service @Slf4j @@ -44,7 +43,7 @@ public class NpFidEnricher implements Handler { List subjects = taskPayload.getSubjects(); var ctx = obtainContext(PackageVariable.class, taskPayload); Long packageId = ctx.get(PackageVariable.packageId); - log.debug("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size()); + log.debug("NpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size()); //найдем телефоны в табличном кэше Map cachedRecords = packageRecordsService.findCachedPhones(subjects); processByNumber(subjects, cachedRecords, ctx); diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index 1f03fa6..8d5a23d 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java @@ -108,6 +108,14 @@ public class CTUtil { return String.format("%09d", phoneNumber); } + public static String normalizeInnZeroesOnly(Long inn) { + return String.format("%010d", inn); + } + + public static String normalizeRegNumZeroesOnly(Long regNum) { + return String.format("%013d", regNum); + } + public static String defineIdType(String idType, String idType2) { if (StringUtils.isNotBlank(idType2)) { return idType2; diff --git a/src/main/resources/mybatis/mapper/indic/PersonMapper.xml b/src/main/resources/mybatis/mapper/indic/PersonMapper.xml index eae91d3..b9bb5a3 100644 --- a/src/main/resources/mybatis/mapper/indic/PersonMapper.xml +++ b/src/main/resources/mybatis/mapper/indic/PersonMapper.xml @@ -38,4 +38,27 @@ WITH UR + + + + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml index b0b09b6..9bca8c9 100644 --- a/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml @@ -3,7 +3,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> - + + WITH src AS (SELECT row_number() OVER () AS rn, + fid, + inn, + reg_num + FROM json_to_recordset(#{json}::json) AS t( + fid int, + inn bigint, + reg_num bigint + )) + INSERT + INTO signals_legals_map + (fid, inn, reg_num, searched_at) + SELECT fid, + inn, + reg_num, + now() + FROM src + order by fid, inn + ON CONFLICT (inn, reg_num) DO UPDATE + SET + fid = EXCLUDED.fid, + inn = EXCLUDED.inn, + reg_num = EXCLUDED.reg_num + + + +