установка пакета с учетом кэш таблицы phones_map

src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java

@@ -8,6 +8,7 @@ import ru.nbch.credit_tracker.entities.app.PhonesMap;
 @Mapper
 public interface PhonesMapMapper {
     List<PhonesMap> getCachedData(@Param("phones") List<Long> phones);
+    PhonesMap findByPhone(@Param("phone") Long phone);
     void updateIdCachedData(@Param("json") String json);
     void updateFlagCachedData(@Param("json") String json);
     void registerJson(@Param("json") String json);

src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java

@@ -34,9 +34,9 @@ import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.
 import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.Subject;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.mapper.signal.FlagDetailMapper;
-import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 import ru.nbch.credit_tracker.service.signal.HitsService;
+import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 import ru.nbch.credit_tracker.service.xml.stax.MonitoringSignalResponseStaxXmlWriter;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@@ -57,7 +57,7 @@ public class MonitoringService {
     private final UserPackageMapper userPackageMapper;
     private final StaxXmlProcessor staxXmlProcessor;
     private final HitsService hitsService;
-    private final PackagesRecordsMapper phoneFidMapMapper;
+    private final PackageRecordsService packageRecordsService;
     private final CreditTrackerSettings config;
 
     public MonitoringService(SignalRestService signalRestService,
@@ -65,14 +65,14 @@ public class MonitoringService {
                              UserPackageMapper userPackageMapper,
                              StaxXmlProcessor staxXmlProcessor,
                              HitsService hitsService,
-                             PackagesRecordsMapper phoneFidMapMapper,
+                             PackageRecordsService packageRecordsService,
                              CreditTrackerSettings config) {
         this.signalRestService = signalRestService;
         this.flagDetailMapper = flagDetailMapper;
         this.userPackageMapper = userPackageMapper;
         this.staxXmlProcessor = staxXmlProcessor;
         this.hitsService = hitsService;
-        this.phoneFidMapMapper = phoneFidMapMapper;
+        this.packageRecordsService = packageRecordsService;
         this.config = config;
     }
 

src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java

@@ -1,15 +1,12 @@
 package ru.nbch.credit_tracker.service.indic;
 
+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;
 import ru.nbch.credit_tracker.entities.indic.PaymentsFlag;
 import ru.nbch.credit_tracker.mapper.indic.IndicFlagsMapper;
 import ru.nbch.credit_tracker.mapper.signal.FlagDetailMapper;
@@ -44,46 +41,4 @@ public class FlagsService2 {
             .collect(Collectors.toMap(PaymentsFlag::getFid, PaymentsFlag::getFlag));
     }
 
-    public void updateCalculationFlags(List<FlagDetail> flagDetails) {
-        try {
-            //fixme, временное решение
-            String jsonFlagDetail = entityToJsonMapper.writeValueAsString(flagDetails.stream()
-                .map(flagDetail -> new HashMap<String, Object>() {{
-                    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<Long> mapIds) {
-        flagDetailMapper.deleteFlags(mapIds);
-    }
-
-    public void registerCalculationFlags(List<FlagDetail> flagDetails) {
-        try {
-            //fixme, временное решение
-            String jsonFlagDetail = entityToJsonMapper.writeValueAsString(flagDetails.stream()
-                .map(flagDetail -> new HashMap<String, Object>() {{
-                    put("subject_id", flagDetail.getSubjectId());
-                    put("fid", flagDetail.getFid());
-                    put("phone", flagDetail.getPhone());
-                    put("map_id", flagDetail.getMapId());
-                    put("package_id", flagDetail.getPackageId());
-                    put("flag_debt5000", flagDetail.getFlagDebt5000());
-                    put("flag_90_12", flagDetail.getFlag90days());
-                    put("calculated_at", flagDetail.getCalculatedAt());
-                }}).toList());
-            flagDetailMapper.registerCalcFlags(jsonFlagDetail);
-        } catch (JsonProcessingException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
 }

src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java

@@ -79,6 +79,25 @@ public class PackageRecordsService {
         }
     }
 
+    public void updateNameCache(List<PhonesMap> phonesMaps) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(phonesMaps.stream()
+                .filter(Objects::nonNull)
+                .map(phoneMap -> new HashMap<String, Object>() {{
+                    put("fid", phoneMap.getFid());
+                    put("phone", phoneMap.getPhone());
+                    put("name_1", phoneMap.getName1());
+                    put("first", phoneMap.getFirst());
+                    put("middle", phoneMap.getMiddle());
+                    put("birth_dt", phoneMap.getBirthDt());
+                }}).toList());
+            //todo update json
+//            phonesMapMapper.registerJson(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     public void updateIdCache(List<IdData> idData) {
         try {
             String json = entityToJsonMapper.writeValueAsString(idData.stream()
@@ -108,4 +127,8 @@ public class PackageRecordsService {
             throw new RuntimeException(e);
         }
     }
+
+    public PhonesMap findByPhone(Long phone) {
+        return phonesMapMapper.findByPhone(phone);
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/task/impl/FidUpdateNameEnricher.java

@@ -0,0 +1,136 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Stream;
+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.PhonesMap;
+import ru.nbch.credit_tracker.entities.indic.Name;
+import ru.nbch.credit_tracker.model.FlagData;
+import ru.nbch.credit_tracker.model.IdData;
+import ru.nbch.credit_tracker.model.NameData;
+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.PersonServiceV2;
+import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
+import ru.nbch.credit_tracker.service.task.Handler;
+import ru.nbch.credit_tracker.utils.CTUtil;
+
+/**
+ * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных.
+ * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки.
+ * Имеет функуию при режиме RESTORE у Pipeline, когда фиды были установлены, но данные нужно подтянуть снова
+ */
+@Service
+@Slf4j
+public class FidUpdateNameEnricher implements Handler<SubjectPayload, SubjectPayload> {
+    private final PersonServiceV2 personService;
+    private final CreditTrackerSettings creditTrackerSettings;
+    private final PackageRecordsService packageRecordsService;
+
+    public FidUpdateNameEnricher(PersonServiceV2 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("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
+        //найдем телефоны в табличном кэше
+        processByNumber(subjects);
+        return taskPayload;
+    }
+
+    private void processByNumber(List<SubjectProfile> subjects) {
+
+        Map<String, Name> nameByPhone = personService.findNameByNumber(subjects);
+
+        List<PhonesMap> subjectForSave = subjects.stream().flatMap(subject -> {
+                Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone().toString()));
+                if (name == null) {
+                    //todo set fid to null
+                    return Stream.of(PhonesMap.builder()
+                        .fid(null)
+                        .phone(subject.getPhone())
+                        .name1(null)
+                        .first(null)
+                        .middle(null)
+                        .birthDt(null)
+                        .build());
+                }
+                return Stream.of(PhonesMap.builder()
+                    .fid(subject.getFid())
+                    .phone(subject.getPhone())
+                    .name1(name.getLast())
+                    .first(name.getFirst())
+                    .middle(name.getMiddle())
+                    .birthDt(name.getBirthDate())
+                    .build());
+        }).toList();
+
+        packageRecordsService.updateNameCache(subjectForSave);
+    }
+
+    private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) {
+        NameData nameData = NameData.builder()
+            .middleName(cachedRecord.getMiddle())
+            .firstName(cachedRecord.getFirst())
+            .lastName(cachedRecord.getName1())
+            .birthDate(cachedRecord.getBirthDt())
+            .build();
+        subjectProfile.setNameData(nameData);
+
+        Optional.of(cachedRecord)
+            .filter(cr -> cachedRecord.getSerNum() != null && cachedRecord.getIdType() != null)
+            .map(cr -> Optional.of(
+                (IdData.builder()
+                    .idNum(cr.getSerNum())
+                    .idType(cr.getIdType())
+                    .build()
+            )))
+            .ifPresent(idData -> subjectProfile.setIdData(idData.get()));
+
+        Optional.of(cachedRecord)
+            .filter(cr -> cachedRecord.getFlagDebt5000() != null && cachedRecord.getFlag90days() != null)
+            .map(cr -> Optional.of(
+                (FlagData.builder()
+                    .flag_debt5000(cr.getFlagDebt5000())
+                    .flag_90_12(cr.getFlag90days())
+                    .build()
+                )))
+            .ifPresent(flagData -> subjectProfile.setFlagData(flagData.get()));
+    }
+
+    private void fillByName(SubjectProfile subjectProfile, Name name) {
+        NameData nameData = NameData.builder()
+            .middleName(name.getMiddle())
+            .firstName(name.getFirst())
+            .lastName(name.getLast())
+            .fileSinceDate(name.getFileSinceDate())
+            .birthDate(name.getBirthDate())
+            .build();
+
+        subjectProfile.setFid(name.getFid());
+        subjectProfile.setNameData(nameData);
+    }
+
+    @Override
+    public int preferredConcurrency() {
+        return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency();
+    }
+
+    @Override
+    public SubjectPayload cookPoison() {
+        return SubjectPayload.POISON;
+    }
+}

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

@@ -20,6 +20,23 @@
         <result property="calculatedAt" column="calculated_at"/>
     </resultMap>
 
+    <select id="findByPhone" resultMap="PhoneMap">
+        select fid,
+               phone,
+               name_1,
+               first,
+               middle,
+               birth_dt,
+               ser_num,
+               id_type,
+               flag_debt5000,
+               flag_90_12,
+               searched_at,
+               calculated_at
+        FROM signals_phones_map
+        WHERE phone = #{phone}
+    </select>
+
     <select id="getCachedData" resultMap="PhoneMap" resultType="java.util.List">
         SELECT fid,
         phone,
@@ -111,29 +128,28 @@
     </update>
 
 
+    <!--    <select id="pageByPackageIdAndOlderThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">-->
+    <!--        SELECT id,-->
+    <!--               package_id,-->
+    <!--               subject_id,-->
+    <!--               phone,-->
+    <!--               error_text-->
+    <!--        FROM signals_phone_fid_map-->
+    <!--        WHERE package_id = #{packageId} and searched_at &lt; now() - make_interval(days => #{daysCount})-->
+    <!--        ORDER BY id-->
+    <!--        LIMIT #{limit} OFFSET #{offset}-->
+    <!--    </select>-->
 
-<!--    <select id="pageByPackageIdAndOlderThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">-->
+    <!--    <select id="pageByPackageIdAndNewerThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">-->
-<!--        SELECT id,-->
+    <!--        SELECT id,-->
-<!--               package_id,-->
+    <!--               package_id,-->
-<!--               subject_id,-->
+    <!--               subject_id,-->
-<!--               phone,-->
+    <!--               phone,-->
-<!--               error_text-->
+    <!--               fid,-->
-<!--        FROM signals_phone_fid_map-->
+    <!--               error_text-->
-<!--        WHERE package_id = #{packageId} and searched_at &lt; now() - make_interval(days => #{daysCount})-->
+    <!--        FROM signals_phone_fid_map-->
-<!--        ORDER BY id-->
+    <!--        WHERE package_id = #{packageId} and searched_at &gt;= now() - make_interval(days => #{daysCount})-->
-<!--        LIMIT #{limit} OFFSET #{offset}-->
+    <!--        ORDER BY id-->
-<!--    </select>-->
+    <!--        LIMIT #{limit} OFFSET #{offset}-->
-
+    <!--    </select>-->
-<!--    <select id="pageByPackageIdAndNewerThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">-->
-<!--        SELECT id,-->
-<!--               package_id,-->
-<!--               subject_id,-->
-<!--               phone,-->
-<!--               fid,-->
-<!--               error_text-->
-<!--        FROM signals_phone_fid_map-->
-<!--        WHERE package_id = #{packageId} and searched_at &gt;= now() - make_interval(days => #{daysCount})-->
-<!--        ORDER BY id-->
-<!--        LIMIT #{limit} OFFSET #{offset}-->
-<!--    </select>-->
 </mapper>