LP: added LpFidEnricher

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<Name> findNameData(@Param("phones") List<String> phones);
 
-    List<Name> findNameDataByFids(@Param("fids") List<Integer> fids);
+    List<FidIdData> findFidsByInn(@Param("inns") List<String> inns);
 
+    List<FidIdData> findFidsByRegNum(@Param("regNums") List<String> regNums);
 }

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<LegalsMap> findByFids(@Param("fids") List<Integer> fids);
 
     void updateFlagCachedData(@Param("json") String json);
@@ -19,4 +21,6 @@ public interface LegalsMapMapper {
 
     List<LegalsMap> findEligibleForFlagsInsert(@Param("lastId") Long lastId,
                                                @Param("batchSize") Integer batchSize);
+
+    List<LegalsMap> findCachedByInnAndRegNUm(@Param("json") String json);
 }

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;
+}

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;
 }

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<String, List<FidIdData>> findFidsByInn(List<SubjectProfile> subjects) {
+        List<String> 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<String, List<FidIdData>> findFidsByRegNum(List<SubjectProfile> subjects) {
+        List<String> 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;

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<Pair<Long, Long>, LegalsMap> findCachedLegalsByInnAndRegNums(List<SubjectProfile> subjects) {
+        try {
+            List<InnRegnumPair> 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<PackagesRecordsMap> findFullPhoneFidMap(Long packageId, Integer limit, Integer offset) {
         return packagesRecordsMapper.findFullPhoneFidMap(packageId, limit, offset);
     }
@@ -71,6 +89,21 @@ public class PackageRecordsService {
         }
     }
 
+    public void saveLegalsToCache(List<LegalsMap> legalsMaps) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(legalsMaps.stream()
+                    .filter(Objects::nonNull)
+                    .map(legalsMap -> new HashMap<String, Object>() {{
+                        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<Long> selectUniquePackageIds(List<Long> phones) {
         try {
             String json = entityToJsonMapper.writeValueAsString(phones.stream()

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<SubjectPayload, SubjectPayload> {
+    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<SubjectProfile> 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<Pair<Long, Long>, LegalsMap> cachedRecords = packageRecordsService.findCachedLegalsByInnAndRegNums(subjects);
+        processByNumber(subjects, cachedRecords, ctx);
+        return taskPayload;
+    }
+
+    private void processByNumber(List<SubjectProfile> subjects,
+                                 Map<Pair<Long, Long>, LegalsMap> cachedRecords,
+                                 PipelineContext<PackageVariable> context) {
+        List<SubjectProfile> noFidsData = subjects.stream()
+            .filter(sp -> !cachedRecords.containsKey(Pair.of(sp.getInn(), sp.getRegNum())))
+            .toList();
+
+        Map<String, List<FidIdData>> fidsByInnMap;
+        Map<String, List<FidIdData>> 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<Pair<Long, Long>> seen = java.util.concurrent.ConcurrentHashMap.newKeySet();
+
+        List<LegalsMap> 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<FidIdData> fidByInn = fidsByInnMap.get(CTUtil.normalizeInnZeroesOnly(inn));
+                    List<FidIdData> 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<Integer> fidsByInn = fidByInn.stream().map(FidIdData::getFid).toList();
+
+                    if (fidByRegNum != null && !fidByRegNum.isEmpty()) {
+                        List<Integer> fidsByRegNum = fidByRegNum.stream().map(FidIdData::getFid).toList();
+
+                        // находим все пересечения по фидам
+                        List<Integer> 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;
+    }
+}

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<SubjectPayload, SubjectPayload> {
         List<SubjectProfile> 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<Long, PhonesMap> cachedRecords = packageRecordsService.findCachedPhones(subjects);
         processByNumber(subjects, cachedRecords, ctx);

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;

src/main/resources/mybatis/mapper/indic/PersonMapper.xml

@@ -38,4 +38,27 @@
         WITH UR
     </select>
 
+
+    <select id="findFidsByInn" resultType="ru.nbch.credit_tracker.model.FidIdData">
+        SELECT FID as fid, ID_NUM as idNum
+        FROM ID
+        WHERE ID_NUM IN
+        <foreach item="inn" collection="inns" open="(" separator="," close=")">
+            #{inn}
+        </foreach>
+        AND (ID_TYPE2 = 'innULRus' OR ID_TYPE = '81')
+        WITH UR
+    </select>
+
+    <select id="findFidsByRegNum" resultType="ru.nbch.credit_tracker.model.FidIdData">
+        SELECT FID as fid, ID_NUM as idNum
+        FROM ID
+        WHERE ID_NUM IN
+        <foreach item="regNum" collection="regNums" open="(" separator="," close=")">
+            #{regNum}
+        </foreach>
+        AND (ID_TYPE2 = 'regnumUL' OR ID_TYPE = '34')
+        WITH UR
+    </select>
+
 </mapper>

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">
 
-<mapper namespace="ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper">
+<mapper namespace="ru.nbch.credit_tracker.mapper.signal.LegalsFidMapMapper">
 
     <select id="registerJson" parameterType="string" resultType="long">
         WITH src AS (SELECT row_number() OVER () AS rn,

src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml

@@ -16,6 +16,32 @@
         <result property="lastRejectedDate" column="last_rejected_date"/>
     </resultMap>
 
+    <select id="registerJson" parameterType="string">
+        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
+    </select>
+
     <select id="findByFids" resultMap="LegalsMap">
         select id,
                inn,
@@ -32,6 +58,24 @@
         </foreach>
     </select>
 
+    <select id="findCachedByInnAndRegNUm" resultMap="LegalsMap">
+        WITH src AS (SELECT inn, regNum
+                     FROM json_to_recordset(#{json}::json) AS t(
+                                                                inn bigint,
+                                                                regNum bigint
+                         ))
+        SELECT slm.id AS id,
+               slm.fid as fid,
+               slm.inn AS inn,
+               slm.reg_num AS reg_num,
+               slm.searched_at AS searched_at,
+               slm.flag_90_12 AS flag_90_12,
+               slm.calculated_at AS calculated_at,
+               slm.last_rejected_date AS last_rejected_date
+        FROM signals_legals_map slm
+        WHERE (slm.inn, slm.reg_num) IN (SELECT inn, regNum FROM src)
+    </select>
+
     <select id="findEligibleForFlagsUpdate" resultMap="LegalsMap">
         select id, fid, inn, reg_num
         from signals_legals_map