restore package in ACCEPTER;PROCCESSING statuses

credit-tracker.properties

@@ -66,7 +66,7 @@ ct.load-dir=D:\\folder_name\\signals
 # min
 ct.monitoring-interval=${MONITORING_INTERVAL:15}
 # days
-ct.update-flags-interval=${UPDATE_FLAGS_INTERVAL:0 0 0 2 * ?}
+ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 22 * * *}
 ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30}
 
 ct.db-version=${DB_VERSION:1234}

src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java

@@ -14,8 +14,8 @@ public class CreditTrackerApplication {
 		UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
 		MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
 //		UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class);
-		updateProcessingPackagesService.run();
+//		updateProcessingPackagesService.run();
-//		monitoringService.run();
+		monitoringService.run();
 //		updatingFlagsService.run();
 	}
 

src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java

@@ -31,7 +31,7 @@ public class CreditTrackerSettings {
     private Long maxXmlFileSize;
     private String loadDir;
     private Long monitoringInterval;
-    private String updateFlagsInterval;
+    private String updateFlagsCronExpression;
     private Long daysToUpdateFlags;
     private String encoding;
     private String dbVersion;

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

@@ -31,4 +31,5 @@ public interface PhoneFidMapMapper {
     boolean hasNonNullFids(@Param("packageId") Long packageId);
 
     Long deleteByPackageId(@Param("packageId") Long id);
+    boolean hasOldRow(@Param("packageId") Long packageId, @Param("daysCount") Integer daysCount);
 }

src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java

@@ -33,7 +33,7 @@ public class PhoneRepository {
         WITH UR
         """;
 
-    private static final String SELCT_FIDS_WITH_NAME = """
+    private static final String SELECT_FIDS_WITH_NAME = """
         SELECT
             pn.number,
             pn.fid,
@@ -49,6 +49,18 @@ public class PhoneRepository {
         WHERE number in (:phones) WITH UR
         """;
 
+    private static final String SELECT_NAME_BY_FIDS = """
+        SELECT
+            fid,
+            first,
+            middle,
+            name_1,
+            birth_dt,
+            file_since_dt as file_since_dt_name
+        FROM NAME 
+        WHERE fid in (:fid) WITH UR
+        """;
+
     private final NamedParameterJdbcTemplate jdbcTemplate;
 
     public PhoneRepository(@Qualifier("jdbcTemplateIndic") NamedParameterJdbcTemplate jdbcTemplate) {
@@ -70,7 +82,7 @@ public class PhoneRepository {
         MapSqlParameterSource params = new MapSqlParameterSource();
         params.addValue("phones", phones);
 
-        return jdbcTemplate.query(SELCT_FIDS_WITH_NAME, params, (rs, rowNum) ->
+        return jdbcTemplate.query(SELECT_FIDS_WITH_NAME, params, (rs, rowNum) ->
             new PhoneRepository.FidsDataWithName(
                 rs.getInt("FID"),
                 rs.getString("NUMBER"),
@@ -87,7 +99,7 @@ public class PhoneRepository {
         MapSqlParameterSource params = new MapSqlParameterSource();
         params.addValue("phones", phones);
 
-        return jdbcTemplate.query(SELCT_FIDS_WITH_NAME, params, (rs, rowNum) ->
+        return jdbcTemplate.query(SELECT_FIDS_WITH_NAME, params, (rs, rowNum) ->
             Name.builder()
                 .fid(rs.getInt("FID"))
                 .number(rs.getString("NUMBER"))
@@ -99,6 +111,21 @@ public class PhoneRepository {
                 .build());
     }
 
+    public List<Name> findNameByFids(List<Integer> fids) {
+        MapSqlParameterSource params = new MapSqlParameterSource();
+        params.addValue("fids", fids);
+
+        return jdbcTemplate.query(SELECT_NAME_BY_FIDS, params, (rs, rowNum) ->
+            Name.builder()
+                .fid(rs.getInt("FID"))
+                .fileSinceDate(getLocalDate(rs, "FILE_SINCE_DT_NAME"))
+                .first(rs.getString("FIRST"))
+                .middle(rs.getString("MIDDLE"))
+                .last(rs.getString("NAME_1"))
+                .birthDate(getLocalDate(rs, "BIRTH_DT"))
+                .build());
+    }
+
     private static LocalDate getLocalDate(ResultSet rs, String col) throws SQLException {
         java.sql.Date d = rs.getDate(col);
         return d != null ? d.toLocalDate() : null;

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

@@ -101,7 +101,7 @@ public class MonitoringService {
             log.info("Package has no fids. Package will be skipped.");
             return;
         }
-        SgnlOnlineReport onlineReports = signalRestService.getReports(userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName());
+        SgnlOnlineReport onlineReports = signalRestService.getReports(userPackage.getUserId() + "_" + userPackage.getPackageName());
 
         if (onlineReports == null) {
             log.info("No reports found for package");

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

@@ -3,16 +3,18 @@ package ru.nbch.credit_tracker.service;
 import java.time.LocalDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
-import java.util.function.Predicate;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.MDC;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.scheduling.support.CronTrigger;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.app.UserPackage;
 import ru.nbch.credit_tracker.enums.Status;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 
 /**
@@ -22,21 +24,33 @@ import ru.nbch.credit_tracker.service.signal.SignalRestService;
 @Slf4j
 @Service
 @EnableScheduling
-public class UpdatingFlagsService /*implements InitializingBean*/ {
+public class UpdatingFlagsService implements InitializingBean {
 
     private final SignalService signalService;
     private final SignalRestService signalRestService;
     private final MonitoringErrorProcesingService monitoringErrorProcesingService;
     private final Long daysToUpdateFlags;
+    private final String updateFlagsCronExpression;
+    private final PhoneFidMapMapper phoneFidMapMapper;
 
     public UpdatingFlagsService(SignalService signalService,
                                 SignalRestService signalRestService,
                                 MonitoringErrorProcesingService monitoringErrorProcesingService,
-                                CreditTrackerSettings config) {
+                                CreditTrackerSettings config,
+                                PhoneFidMapMapper phoneFidMapMapper) {
         this.signalService = signalService;
         this.signalRestService = signalRestService;
         this.monitoringErrorProcesingService = monitoringErrorProcesingService;
         this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1;
+        this.updateFlagsCronExpression = config.getUpdateFlagsCronExpression() != null ? config.getUpdateFlagsCronExpression() : "*/10 * * * * *";
+        this.phoneFidMapMapper = phoneFidMapMapper;
+    }
+
+    public void run() {
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+        scheduler.setPoolSize(1);
+        scheduler.initialize();
+        scheduler.schedule(this::update, new CronTrigger(updateFlagsCronExpression));
     }
 
     @Scheduled(cron = "${ct.update-flags-interval}")
@@ -44,9 +58,13 @@ public class UpdatingFlagsService /*implements InitializingBean*/ {
         log.info("Start updating flags");
         List<UserPackage> packages = signalService.findPackagesOnMonitoring();
         for (UserPackage userPackage : packages) {
-            log.info("Updating flags for Package {}", userPackage.getPackageName());
             MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() +"]");
+            log.info("Check on updating flags for Package {}", userPackage.getPackageName());
+            boolean hasOldRow = phoneFidMapMapper.hasOldRow(userPackage.getId(), daysToUpdateFlags.intValue());
+            log.info("There are {} old rows", hasOldRow ? "exist" : "doesn't exist");
+            if (hasOldRow) {
                 updateFlags(userPackage);
+            }
             MDC.remove("ru.nbch.credit_tracker.log.mdc.key");
         }
         log.info("Finish updating flags");
@@ -57,8 +75,7 @@ public class UpdatingFlagsService /*implements InitializingBean*/ {
         String packageName = userPackage.getPackageName();
         try {
             signalService.updatePackageStatus(packageId, Status.PROCESSING);
-            Predicate<PhoneFidMap> filter = subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt());
+            SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, null);
-            SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, filter);
             if (updateResult.isUpdated()) {
                 if (signalRestService.setupMonitoring(packageId)) {
 //                    LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now();
@@ -83,4 +100,9 @@ public class UpdatingFlagsService /*implements InitializingBean*/ {
         long daysPassed = ChronoUnit.DAYS.between(pastDateTime, now);
         return daysPassed >= daysToUpdateFlags;
     }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+
+    }
 }

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

@@ -59,6 +59,30 @@ public class PhoneService2 {
             ));
     }
 
+    /**
+     * Поиск данных по таблицам NAME по списку FID
+     * Если пакет ACCEPTED, то в базе могут находиться записи с уже рассчитанными FID
+     * Нужен для исключения долгих запросов findNameByPhones
+     * Смотри UpdateProcessingPackagesService:processingRestore
+     * @param subjects
+     * @return - отображение NAME к FID, отфильтрованному по логике
+     * FID с максимальным значением FILE_SINCE_DT. Если FILE_SINCE_DT оказался одинаковый, берется максимальный FID
+     */
+    public Map<Integer, Name> nameByFid(List<SubjectProfile> subjects) {
+        List<Integer> fids = subjects.stream()
+            .map(SubjectProfile::getFid)
+            .toList();
+
+        List<Name> names = phoneRepository.findNameByFids(fids);
+
+        return names.stream()
+            .collect(Collectors.toMap(
+                Name::getFid,
+                Function.identity(),
+                BinaryOperator.maxBy(byDateThenIdNullSafe)
+            ));
+    }
+
     public List<Long> registerSubjects(List<PhoneFidMap> subjects) {
         try {
             //fixme, временное решение
@@ -99,4 +123,8 @@ public class PhoneService2 {
         return phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset);
     }
 
+    public List<PhoneFidMap> pageByPackageIdWithNotNullFids(Long packageId, Integer limit, Integer offset) {
+        return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset);
+    }
+
 }

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

@@ -18,6 +18,7 @@ import ru.nbch.credit_tracker.utils.CTUtil;
 /**
  * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных.
  * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки.
+ * Имеет функуию при режиме RESTORE у Pipeline, когда фиды были установлены, но данные нужно подтянуть снова
  */
 @Service
 @Slf4j
@@ -36,6 +37,15 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
         List<SubjectProfile> subjects = taskPayload.getSubjects();
         log.debug("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size());
 
+        if (subjects.getFirst().getFid() == null) {
+            processByNumber(subjects);
+        } else {
+            processByFid(subjects);
+        }
+        return taskPayload;
+    }
+
+    private void processByNumber(List<SubjectProfile> subjects) {
         LocalDateTime now = LocalDateTime.now();
         Map<String, Name> nameByPhone = phoneService2.findNameByNumber(subjects);
         List<PhoneFidMap> subjectForSave = subjects.stream().map(subject -> {
@@ -63,7 +73,24 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
         }).toList();
 
         phoneService2.updateFid(subjectForSave);
-        return taskPayload;
+    }
+
+    private void processByFid(List<SubjectProfile> subjects) {
+        Map<Integer, Name> nameByPhone = phoneService2.nameByFid(subjects);
+        subjects.forEach(subject -> {
+            Name name = nameByPhone.get(subject.getFid());
+            if (name != null) {
+                NameData nameData = NameData.builder()
+                    .middleName(name.getMiddle())
+                    .firstName(name.getFirst())
+                    .lastName(name.getLast())
+                    .fileSinceDate(name.getFileSinceDate())
+                    .birthDate(name.getBirthDate())
+                    .build();
+
+                subject.setNameData(nameData);
+            }
+        });
     }
 
     @Override

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

@@ -28,6 +28,7 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
     protected final PhoneService2 phoneService2;
     private final int BATCH_SIZE;
     private boolean isInterrupted = false;
+    protected int offset = 0;
 
 
     public KickPipeLine(StaxXmlProcessor xmlProcessor,
@@ -53,10 +54,9 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
     public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
         try {
             log.info("Start registering subjects. PackageId: {};", packageId);
-            int offset = 0;
             int limit = BATCH_SIZE;
             UserPackage userPackage = userPackageMapper.findPackage(packageId);
-            List<PhoneFidMap> page = nextPage(limit, offset);
+            List<PhoneFidMap> page = nextPage(limit);
             while (!page.isEmpty()) {
                 log.debug("Get page {}", page.size());
                 List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
@@ -72,10 +72,10 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
                     throw new PipelineStarterException();
                 }
                 offset += limit;
-                page = nextPage(limit, offset);
+                page = nextPage(limit);
             }
         } catch (PipelineStarterException e) {
-            log.info("Pipeline starter shitting down...");
+            log.info("Pipeline starter shutting down...");
         } finally {
             if (isInterrupted) {
                 //noinspection ReturnInsideFinallyBlock
@@ -92,5 +92,5 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
     }
 
     protected abstract PipelineMode mode();
-    protected abstract  List<PhoneFidMap> nextPage(Integer limit, Integer offset);
+    protected abstract  List<PhoneFidMap> nextPage(Integer limit);
 }

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

@@ -24,7 +24,7 @@ public class KickPipeLineDefaultImpl extends KickPipeLine {
     }
 
     @Override
-    protected List<PhoneFidMap> nextPage(Integer limit, Integer offset) {
+    protected List<PhoneFidMap> nextPage(Integer limit) {
         return phoneService2.pageByPackageId(packageId, limit, offset);
     }
 

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.task.impl;
 
+import java.util.ArrayList;
 import java.util.List;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
@@ -10,7 +11,7 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 
 public class KickPipeLineRestoreImpl extends KickPipeLine {
 
-
+    private boolean nullFidsWereFound = false;
     public KickPipeLineRestoreImpl(StaxXmlProcessor xmlProcessor,
                                    SetupParam setupParam,
                                    UserPackageMapper userPackageMapper,
@@ -25,7 +26,16 @@ public class KickPipeLineRestoreImpl extends KickPipeLine {
     }
 
     @Override
-    protected List<PhoneFidMap> nextPage(Integer limit, Integer offset) {
+    protected List<PhoneFidMap> nextPage(Integer limit) {
-        return phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset);
+        List<PhoneFidMap> pageWithNullFids = new ArrayList<>();
+        if (!nullFidsWereFound) {
+            pageWithNullFids = phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset);
+        }
+        if (pageWithNullFids.isEmpty()) {
+            nullFidsWereFound = true;
+            offset = 0;
+            return phoneService2.pageByPackageIdWithNotNullFids(packageId, limit, offset);
+        }
+        return pageWithNullFids;
     }
 }

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

@@ -175,6 +175,22 @@
         LIMIT #{limit} OFFSET #{offset}
     </select>
 
+    <select id="findNotNullFidPhoneFidMap" resultMap="PhoneFidMapResultMap" resultType="java.util.List">
+        SELECT id,
+               package_id,
+               subject_id,
+               phone,
+               fid,
+               flag_90_12,
+               flag_debt5000,
+               calculated_at,
+               error_text
+        FROM signals_phone_fid_map
+        WHERE package_id = #{packageId} and fid is not null
+        ORDER BY id
+        LIMIT #{limit} OFFSET #{offset}
+    </select>
+
     <select id="findPhoneFidMapById" resultMap="PhoneFidMapResultMap"
             resultType="ru.nbch.credit_tracker.entities.app.PhoneFidMap">
         SELECT id,
@@ -198,5 +214,11 @@
           AND fid is not null
     </select>
 
-
+    <select id="hasOldRow" resultType="java.lang.Boolean">
+        select EXISTS(SELECT 1
+        FROM signals_phone_fid_map
+        WHERE package_id = #{packageId}
+        AND calculated_at &lt; now() - make_interval(days => #{daysCount})
+        )
+    </select>
 </mapper>