From a24a3b59759066cba8f06b07e6db21c3aee3b943 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Fri, 28 Nov 2025 14:24:54 +0300 Subject: [PATCH] restore package in ACCEPTER;PROCCESSING statuses --- credit-tracker.properties | 2 +- .../CreditTrackerApplication.java | 4 +- .../settings/CreditTrackerSettings.java | 2 +- .../mapper/signal/PhoneFidMapMapper.java | 1 + .../repository/PhoneRepository.java | 33 ++++++++++++++-- .../service/MonitoringService.java | 2 +- .../service/UpdatingFlagsService.java | 38 +++++++++++++++---- .../service/indic/PhoneService2.java | 28 ++++++++++++++ .../service/task/impl/FidNameEnricher.java | 29 +++++++++++++- .../service/task/impl/KickPipeLine.java | 10 ++--- .../task/impl/KickPipeLineDefaultImpl.java | 2 +- .../task/impl/KickPipeLineRestoreImpl.java | 16 ++++++-- .../mapper/signal/PhoneFidMapMapper.xml | 24 +++++++++++- 13 files changed, 164 insertions(+), 27 deletions(-) diff --git a/credit-tracker.properties b/credit-tracker.properties index 11f84f3..a388d11 100644 --- a/credit-tracker.properties +++ b/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} diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index 3b68321..919f0c2 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/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(); -// monitoringService.run(); +// updateProcessingPackagesService.run(); + monitoringService.run(); // updatingFlagsService.run(); } diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index ebe4062..5dcd2f8 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/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; diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java index 18bcff8..4760e64 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java +++ b/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); } diff --git a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java index d1b00b1..bc1064f 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/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 findNameByFids(List 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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java index 17aab90..49774a5 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/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"); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index a3be7a5..47e7a70 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/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 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() +"]"); - updateFlags(userPackage); + 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 filter = subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt()); - SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, filter); + SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, null); 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 { + + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java index 4771996..806c92e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java +++ b/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 nameByFid(List subjects) { + List fids = subjects.stream() + .map(SubjectProfile::getFid) + .toList(); + + List names = phoneRepository.findNameByFids(fids); + + return names.stream() + .collect(Collectors.toMap( + Name::getFid, + Function.identity(), + BinaryOperator.maxBy(byDateThenIdNullSafe) + )); + } + public List registerSubjects(List subjects) { try { //fixme, временное решение @@ -99,4 +123,8 @@ public class PhoneService2 { return phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset); } + public List pageByPackageIdWithNotNullFids(Long packageId, Integer limit, Integer offset) { + return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset); + } + } 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 dda9a5f..e22a1d6 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 @@ -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 List 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 subjects) { LocalDateTime now = LocalDateTime.now(); Map nameByPhone = phoneService2.findNameByNumber(subjects); List subjectForSave = subjects.stream().map(subject -> { @@ -63,7 +73,24 @@ public class FidNameEnricher implements Handler }).toList(); phoneService2.updateFid(subjectForSave); - return taskPayload; + } + + private void processByFid(List subjects) { + Map 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 diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java index 4939565..bd6e334 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java @@ -28,6 +28,7 @@ public abstract class KickPipeLine implements IPipeLineStarter { 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 { public void startFillingQueue(BlockingQueue firstQueue) { try { log.info("Start registering subjects. PackageId: {};", packageId); - int offset = 0; int limit = BATCH_SIZE; UserPackage userPackage = userPackageMapper.findPackage(packageId); - List page = nextPage(limit, offset); + List page = nextPage(limit); while (!page.isEmpty()) { log.debug("Get page {}", page.size()); List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() @@ -72,10 +72,10 @@ public abstract class KickPipeLine implements IPipeLineStarter { 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 { } protected abstract PipelineMode mode(); - protected abstract List nextPage(Integer limit, Integer offset); + protected abstract List nextPage(Integer limit); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java index 2d1c80c..f304c09 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java @@ -24,7 +24,7 @@ public class KickPipeLineDefaultImpl extends KickPipeLine { } @Override - protected List nextPage(Integer limit, Integer offset) { + protected List nextPage(Integer limit) { return phoneService2.pageByPackageId(packageId, limit, offset); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java index 453c65f..eaeae10 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java +++ b/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 nextPage(Integer limit, Integer offset) { - return phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset); + protected List nextPage(Integer limit) { + List 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; } } diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 275886c..82773e5 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -175,6 +175,22 @@ LIMIT #{limit} OFFSET #{offset} + + - + \ No newline at end of file