diff --git a/credit-tracker.properties b/credit-tracker.properties index a388d11..324bcd6 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-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 22 * * *} +ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 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 919f0c2..97b975d 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -13,10 +13,8 @@ public class CreditTrackerApplication { ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class); MonitoringService monitoringService = ctx.getBean(MonitoringService.class); -// UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class); -// updateProcessingPackagesService.run(); + updateProcessingPackagesService.run(); monitoringService.run(); -// updatingFlagsService.run(); } } diff --git a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java index 55d0768..fb148dc 100644 --- a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java +++ b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java @@ -31,7 +31,6 @@ public class PersonFactoryBuilderV2 { private SubjectProfile personData; - private Long uid; public static PersonFactoryBuilderV2 builder() { return new PersonFactoryBuilderV2(); @@ -39,8 +38,6 @@ public class PersonFactoryBuilderV2 { public PersonFactoryBuilderV2 personData(SubjectProfile personData) { this.personData = personData; - //todo откуда брать этот id -// this.uid = personData.getIdentifier(); return this; } @@ -64,7 +61,7 @@ public class PersonFactoryBuilderV2 { person.setConsentOdt(ODT); } person.setIp(IP); - person.setUid(uid); + person.setUid(personData.getMapId()); person.setOksm(OKSM); person.setConsentDate(DATE_FORMAT.format(LocalDate.now())); person.setConsentPurpose(CONSENT_PURPOSE); diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java index d42f3ee..462ba1d 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java @@ -5,13 +5,14 @@ import java.util.function.Supplier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl; +import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRepeatedRestoreImpl; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRestoreImpl; import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; @@ -24,12 +25,14 @@ public class PipelineStarterConfig { public Function> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings, UserPackageMapper userPackageMapper, - PhoneService2 phoneService2) { + PhoneFidMapMapper phoneFidMapMapper) { return (setupParam) -> switch (setupParam.mode()) { case DEFAULT -> - new KickPipeLineDefaultImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + new KickPipeLineDefaultImpl(xml, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings); case RESTORE -> - new KickPipeLineRestoreImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + new KickPipeLineRestoreImpl(xml, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings); + case REPEATED_RESTORE -> + new KickPipeLineRepeatedRestoreImpl(xml, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings); }; } diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java b/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java index b703a30..9ab56ab 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java @@ -17,5 +17,7 @@ public class UserPackage { private String status; private LocalDateTime deactivatedAt; private String lastDownloadedReport; -// private LocalDateTime monitoringAt; + private LocalDateTime monitoringAt; + private LocalDateTime lastSearchedAt; + private String rejectStatus; } diff --git a/src/main/java/ru/nbch/credit_tracker/enums/RejectedStatus.java b/src/main/java/ru/nbch/credit_tracker/enums/RejectedStatus.java new file mode 100644 index 0000000..72c7f2a --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/enums/RejectedStatus.java @@ -0,0 +1,17 @@ +package ru.nbch.credit_tracker.enums; + +public enum RejectedStatus { + OK("0"), + ERROR("1"),; + + + private final String key; + + RejectedStatus(String key) { + this.key = key; + } + + public String getKey() { + return key; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java index 3c0f423..581fe10 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -46,6 +46,7 @@ import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; import ru.nbch.credit_tracker.service.task.impl.Saver; import ru.nbch.credit_tracker.utils.CTUtil; +import ru.nbch.credit_tracker.utils.CtFileUtils; import ru.nbch.credit_tracker.utils.EnumUtil; import ru.nbch.credit_tracker.utils.LogUtil; import ru.nbch.credit_tracker.utils.ZipUtils; @@ -114,6 +115,9 @@ public class SignalFacadeV3 { if (e.getCause() instanceof XMLStreamException) { throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError); } + if (e instanceof SignalClientException) { + throw e; + } throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e); } catch (IOException e) { log.warn("Error deleting tmp file", e); @@ -138,10 +142,10 @@ public class SignalFacadeV3 { } // проверка на уникальность пакета -// if (signalService.isPackageNotUnique(userId, packageName)) { -// signalService.rejectPackage(userId, packageName); -// throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError); -// } + if (userPackageService.isPackageNotUnique(userId, packageName)) { + userPackageService.rejectPackage(userId, packageName); + throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError); + } Long packageId = userPackageService.registerPackage(userId, packageName); LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId)); @@ -151,6 +155,7 @@ public class SignalFacadeV3 { .runAsync(() -> { try { saver.readAndSave(packageId, parsedFilePath); + CtFileUtils.deleteQuietly(parsedFilePath); } catch (Throwable t) { log.error(ExceptionUtils.getStackTrace(t)); pipeline.close(); 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 9fc2285..081b297 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 @@ -15,12 +15,15 @@ public interface PhoneFidMapMapper { void register(@Param("phoneFidMap") List phoneFidMap); List registerJson(@Param("json") String json); void setFidJson(@Param("json") String json); + void setErrorJson(@Param("json") String json); void registerWithId(@Param("phoneFidMap") List phoneFidMap); List findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset); List findNullFidPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset); List findNotNullFidPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset); + List pageByPackageIdAndOlderThanN(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset, @Param("daysCount") Integer daysCount); + List pageByPackageIdAndNewerThanN(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset, @Param("daysCount") Integer daysCount); void updatePhoneFidMaps(@Param("subjects") Collection subjects); void updateOne(@Param("subject") PhoneFidMap subject); void updateFid(@Param("subjects") Collection subjects); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java index e11d5be..9c28b22 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java @@ -37,4 +37,6 @@ public interface UserPackageMapper { List findAllPackagesByUserId(@Param("userId") String userId); List findByMemberCodeAndPackageName(@Param("memberCode") String memberCode, @Param("packageName") String packageName); + + void setRejectedStatus(@Param("id") Long id, @Param("status") String status); } 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 bc1064f..5d868cf 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java @@ -58,7 +58,7 @@ public class PhoneRepository { birth_dt, file_since_dt as file_since_dt_name FROM NAME - WHERE fid in (:fid) WITH UR + WHERE fid in (:fids) WITH UR """; private final NamedParameterJdbcTemplate jdbcTemplate; diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java index 63809e6..6c5348c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java @@ -1,13 +1,15 @@ package ru.nbch.credit_tracker.service; import java.nio.file.Path; +import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.aop.framework.AopContext; import org.springframework.context.annotation.EnableAspectJAutoProxy; import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc; import ru.nbch.credit_tracker.entities.app.UserPackage; +import ru.nbch.credit_tracker.enums.RejectedStatus; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.utils.CtFileUtils; @@ -19,37 +21,39 @@ public class MonitoringErrorProcesingService { private MonitoringErrorProcesingService self; private final SignalService signalService; + private final UserPackageMapper userPackageMapper; private final SignalRestService signalRestService; - public MonitoringErrorProcesingService(SignalService signalService, SignalRestService signalRestService) { + public MonitoringErrorProcesingService(SignalService signalService, + UserPackageMapper userPackageMapper, + SignalRestService signalRestService) { this.signalService = signalService; + this.userPackageMapper = userPackageMapper; this.signalRestService = signalRestService; } public void findErrors(Long packageId) { UserPackage userPackage = signalService.findPackage(packageId); - getCurrentProxy().processPackage(userPackage, packageId); + String packCode = userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName(); + log.info("Проверка на ошибки при постановке на мониторинг. packCode={}, packageId={}", packCode, packageId); + String fileName = signalRestService.getErrorFileName(packageId, packCode); + if (StringUtils.isEmpty(fileName)) { + userPackageMapper.setRejectedStatus(packageId, RejectedStatus.OK.getKey()); + return; + } + + Optional filePath = signalRestService.loadErrorFile(fileName, userPackage); + if (filePath.isPresent()) { + try { + signalService.updateSubjectsErrors(filePath.get(), packageId); + } catch (Exception e) { + log.error("Ошибка при записи сообщений об ошибках в сформированных субъектах для мониторинга. Package_id = {}", packageId, e); + } + CtFileUtils.deleteQuietly(filePath.get()); + } + userPackageMapper.setRejectedStatus(packageId, RejectedStatus.ERROR.getKey()); } - @SetAndClearMdc - public void processPackage(UserPackage userPackage, Long packageId) { - if (userPackage != null) { - String packCode = userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName(); - log.info("Проверка на ошибки при постановке на мониторинг. packCode={}, packageId={}", packCode, packageId); - String fileName = signalRestService.getErrorFileName(packageId, packCode); - if (StringUtils.isNotBlank(fileName)) { - String filePath = signalRestService.loadErrorFile(packageId, fileName); - if (StringUtils.isNotBlank(filePath)) { - try { - signalService.updateSubjectsErrors(filePath, packageId); - } catch (Exception e) { - log.error("Ошибка при записи сообщений об ошибках в сформированных субъектах для мониторинга. Package_id = {}", packageId, e); - } - CtFileUtils.deleteQuietly(Path.of(filePath)); - } - } - } - } private MonitoringErrorProcesingService getCurrentProxy() { if (self == null) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index bccf1d4..ec17c6b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -288,9 +288,9 @@ public class SignalService { return persons; } - public void updateSubjectsErrors(String filePath, Long packageId) { + public void updateSubjectsErrors(Path filePath, Long packageId) { log.info("Start updating subjects errors. PackageId: {}", packageId); - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) { + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(filePath), 8 * 1024 * 1024)) { staxXmlProcessor.read(in, MonitoringReportErrors.class, request -> { if (!request.getPersons().isEmpty()) { @@ -301,13 +301,12 @@ public class SignalService { .errorText(p.getValue()) .build(); }).toList(); - phoneFidMapService.updatePhoneFidMapsErrors(subjects); + phoneService2.updateOnError(subjects); log.trace("Updated subject's batch. Total count: {}. PackageId: {}", request.getPersons().size(), packageId); request.getPersons().clear(); } }, - request -> request.getPersons().isEmpty() - , + request -> request.getPersons().isEmpty(), batchSize, "/Errors/p"); } catch (IOException e) { throw new RuntimeException("Error while parsing MonitoringReportErrors", e); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java index 02a8754..3b0f16c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java @@ -5,16 +5,13 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import lombok.extern.slf4j.Slf4j; -import org.springframework.aop.framework.AopContext; import org.springframework.context.annotation.EnableAspectJAutoProxy; import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; 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.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.Pipeline; import ru.nbch.credit_tracker.service.task.PipelinePackageManager; import ru.nbch.credit_tracker.service.task.SetupParam; @@ -32,34 +29,23 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @EnableAspectJAutoProxy(exposeProxy = true) public class UpdateProcessingPackagesService { - private UpdateProcessingPackagesService self; - - private final SignalService signalService; private final UserPackageMapper userPackageMapper; private final PhoneFidMapMapper phoneFidMapMapper; - private final SignalRestService signalRestService; - private final MonitoringErrorProcesingService monitoringErrorProcesingService; private final PipelinePackageManager pipelinePackageManager; private final CreditTrackerSettings creditTrackerSettings; private final Saver saver; private final ExecutorService pipelineExecutorService; private final ExecutorService saverExecutorService; - public UpdateProcessingPackagesService(SignalService signalService, - UserPackageMapper userPackageMapper, + public UpdateProcessingPackagesService(UserPackageMapper userPackageMapper, PhoneFidMapMapper phoneFidMapMapper, - SignalRestService signalRestService, - MonitoringErrorProcesingService monitoringErrorProcesingService, PipelinePackageManager pipelinePackageManager, CreditTrackerSettings creditTrackerSettings, Saver saver, ExecutorService pipelineExecutorService, ExecutorService saverExecutorService) { - this.signalService = signalService; this.userPackageMapper = userPackageMapper; this.phoneFidMapMapper = phoneFidMapMapper; - this.signalRestService = signalRestService; - this.monitoringErrorProcesingService = monitoringErrorProcesingService; this.pipelinePackageManager = pipelinePackageManager; this.creditTrackerSettings = creditTrackerSettings; this.saver = saver; @@ -70,16 +56,9 @@ public class UpdateProcessingPackagesService { public void run() { acceptedRestore(); processingRestore(); -// log.info("Start updating processing packages"); -// List packages = signalService.findPackagesOnProcessing(); -// for (UserPackage userPackage : packages) { -// log.info("Updating flags for Package {}", userPackage.getPackageName()); -// getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName()); -// } -// log.info("Finish updating processing packages"); + processingRepeatedRestore(); } - private void acceptedRestore() { List packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name()); log.debug("Found {} packages in status ACCEPTED", packages.size()); @@ -89,8 +68,6 @@ public class UpdateProcessingPackagesService { log.trace("Deleted {} rows", count); Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage); log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath); - //todo ???запускает асинхронный процесс без ожидания, на старте мы должны восстановиться один раз и - // видимо дождаться выполнения, иначе может не хватить пула, памяти или чего еще Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT)); CompletableFuture .runAsync(() -> { @@ -110,7 +87,8 @@ public class UpdateProcessingPackagesService { private void processingRestore() { List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name()); - log.debug("Found {} packages in status PROCESSING", packages.size()); + packages = packages.stream().filter(userPackage -> userPackage.getRejectStatus() == null).toList(); + log.debug("Found {} packages in status PROCESSING and rejected_status = null", packages.size()); for (UserPackage userPackage : packages) { try { Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE)); @@ -121,33 +99,17 @@ public class UpdateProcessingPackagesService { } } - @SetAndClearMdc(argIndex = 1) - public void process(Long packageId, String packageName) { - try { - SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, null); - if (updateResult.isNoFidsFound()) { - log.info("No fids found for PackageId: {} . Package status ON_MONITORING. No registration in signal service.", packageId); - signalService.updatePackageStatus(packageId, Status.ON_MONITORING); - return; + private void processingRepeatedRestore() { + List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING_REPEATED.name()); + log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size()); + for (UserPackage userPackage : packages) { + try { + Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE)); + pipeline.start(pipelineExecutorService); + } catch (Exception e) { + log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); } - if (signalRestService.setupMonitoring(packageId)) { - log.info("Package was successfully registered in signal service. PackageId: {}", packageId); - signalService.updatePackageStatus(packageId, Status.ON_MONITORING); - monitoringErrorProcesingService.findErrors(packageId); - } else { - log.info("Package registration was rejected by signal service. PackageId: {}", packageId); - signalService.updatePackageStatus(packageId, Status.ERROR); - } - } catch (Exception e) { - signalService.updatePackageStatus(packageId, Status.ERROR); - log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e); } } - private UpdateProcessingPackagesService getCurrentProxy() { - if (self == null) { - self = (UpdateProcessingPackagesService) AopContext.currentProxy(); - } - return self; - } } 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 cac9915..3033210 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -12,6 +12,7 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; 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.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.signal.SignalRestService; /** @@ -28,17 +29,19 @@ public class UpdatingFlagsService { private final MonitoringErrorProcesingService monitoringErrorProcesingService; private final Long daysToUpdateFlags; private final PhoneFidMapMapper phoneFidMapMapper; + private final UserPackageMapper userPackageMapper; public UpdatingFlagsService(SignalService signalService, SignalRestService signalRestService, MonitoringErrorProcesingService monitoringErrorProcesingService, CreditTrackerSettings config, - PhoneFidMapMapper phoneFidMapMapper) { + PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper) { this.signalService = signalService; this.signalRestService = signalRestService; this.monitoringErrorProcesingService = monitoringErrorProcesingService; this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1; this.phoneFidMapMapper = phoneFidMapMapper; + this.userPackageMapper = userPackageMapper; } @Scheduled(cron = "${ct.update-flags-cron-expression}") @@ -62,7 +65,8 @@ public class UpdatingFlagsService { Long packageId = userPackage.getId(); String packageName = userPackage.getPackageName(); try { - signalService.updatePackageStatus(packageId, Status.PROCESSING); + //todo set reject_status = null + userPackageMapper.updatePackageStatus(packageId, Status.PROCESSING_REPEATED.name()); SignalService.UpdateResult updateResult = signalService.updateFlags2(userPackage); if (updateResult.isUpdated()) { if (signalRestService.setupMonitoring(packageId)) { 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 806c92e..17d22eb 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 @@ -13,6 +13,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.indic.Name; +import ru.nbch.credit_tracker.mapper.signal.FlagDetailMapper; import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.repository.PhoneRepository; @@ -24,6 +25,7 @@ public class PhoneService2 { private final PhoneRepository phoneRepository; private final PhoneFidMapMapper phoneFidMapMapper; + private final FlagDetailMapper flagDetailMapper; private final ObjectMapper entityToJsonMapper; private final Comparator byDateThenIdNullSafe = Comparator.comparing(Name::getFileSinceDate, Comparator.nullsFirst(Comparator.naturalOrder())) @@ -31,9 +33,11 @@ public class PhoneService2 { public PhoneService2(PhoneRepository phoneRepository, PhoneFidMapMapper phoneFidMapMapper, + FlagDetailMapper flagDetailMapper, ObjectMapper entityToJsonMapper) { this.phoneRepository = phoneRepository; this.phoneFidMapMapper = phoneFidMapMapper; + this.flagDetailMapper = flagDetailMapper; this.entityToJsonMapper = entityToJsonMapper; } @@ -64,6 +68,7 @@ public class PhoneService2 { * Если пакет ACCEPTED, то в базе могут находиться записи с уже рассчитанными FID * Нужен для исключения долгих запросов findNameByPhones * Смотри UpdateProcessingPackagesService:processingRestore + * * @param subjects * @return - отображение NAME к FID, отфильтрованному по логике * FID с максимальным значением FILE_SINCE_DT. Если FILE_SINCE_DT оказался одинаковый, берется максимальный FID @@ -85,7 +90,6 @@ public class PhoneService2 { public List registerSubjects(List subjects) { try { - //fixme, временное решение String json = entityToJsonMapper.writeValueAsString(subjects.stream() .map(phoneFidMap -> new HashMap() {{ put("subject_id", phoneFidMap.getSubjectId()); @@ -102,12 +106,12 @@ public class PhoneService2 { public void updateFid(List subjects) { try { - //fixme, временное решение + //регистрация в signals_phone_fid_map String json = entityToJsonMapper.writeValueAsString(subjects.stream() .map(phoneFidMap -> new HashMap() {{ put("id", phoneFidMap.getId()); put("fid", phoneFidMap.getFid()); - put("calculated_at", phoneFidMap.getCalculatedAt()); + put("searched_at", phoneFidMap.getCalculatedAt()); }}).toList()); phoneFidMapMapper.setFidJson(json); } catch (JsonProcessingException e) { @@ -115,16 +119,38 @@ public class PhoneService2 { } } + public void registerSignals(List subjects) { + try { + //регистрация в signals_flags - апдейтим при FlagsEnricher + String jsonFlagDetail = entityToJsonMapper.writeValueAsString(subjects.stream() + .filter(phoneFidMap -> phoneFidMap.getFid() != null) + .map(subject -> new HashMap() {{ + put("subject_id", subject.getSubjectId()); + put("fid", subject.getFid()); + put("phone", subject.getPhone()); + put("map_id", subject.getId()); + }}).toList()); + flagDetailMapper.updateCalcFlags(jsonFlagDetail); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + + public void updateOnError(List subjects) { + try { + String json = entityToJsonMapper.writeValueAsString(subjects.stream() + .map(phoneFidMap -> new HashMap() {{ + put("id", phoneFidMap.getId()); + put("error_text", phoneFidMap.getErrorText()); + }}).toList()); + phoneFidMapMapper.setErrorJson(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public List pageByPackageId(Long packageId, Integer limit, Integer offset) { return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset); } - public List pageByPackageIdWithNullFids(Long packageId, Integer limit, Integer offset) { - 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/signal/SignalRestService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java index 6edb685..692028b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java @@ -1,19 +1,12 @@ package ru.nbch.credit_tracker.service.signal; import jakarta.xml.bind.DatatypeConverter; -import java.io.ByteArrayInputStream; -import java.io.File; -import java.io.InputStream; -import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.security.MessageDigest; import java.util.Base64; import java.util.Optional; -import java.util.UUID; -import java.util.zip.GZIPInputStream; -import java.util.zip.GZIPOutputStream; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Qualifier; @@ -38,7 +31,6 @@ import ru.nbch.credit_tracker.entities.external_response.online_report.SgnlOnlin import ru.nbch.credit_tracker.service.SignalService; import ru.nbch.credit_tracker.service.xml.jaxb.IXmlProcessor; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.CTUtil; import ru.nbch.credit_tracker.utils.ZipUtils; @Slf4j @@ -161,7 +153,7 @@ public class SignalRestService { * * @return путь к разархивированному файлу. Пример: /tmp_folder//0001XX_MY_BATCH_223112.1.errors.xml.gz */ - public String loadErrorFile(Long packageId, String fileName) { + public Optional loadErrorFile(String fileName, UserPackage userPackage) { String url = DOWNLOAD_REPORT_URI + "?fNameRequest=" + fileName; HttpHeaders headers = new HttpHeaders(); @@ -176,7 +168,17 @@ public class SignalRestService { byte[].class ); - return unzipReport(response, packageId, fileName); + Path targetFilePathWithExt = null; + if (response.getHeaders().getContentType().equals(MEDIA_TYPE_GZIP)) { + targetFilePathWithExt = Path.of(config.getLoadDir(), + "tmp", userPackage.getUserId(), userPackage.getPackageName(), fileName.replace(".xml.gz",".xml")); + ZipUtils.unpackGzip(targetFilePathWithExt, response.getBody()); + } else if (isXmlError(response.getBody())){ + SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class); + String description = error.getError().getDescription() != null ? error.getError().getDescription() : "Неизвестная ошибка"; + log.error("Ошибка при скачивание отчета. fNameReport={}. Описание: {}", fileName, description); + } + return Optional.ofNullable(targetFilePathWithExt); } /** @@ -247,18 +249,6 @@ public class SignalRestService { return true; } - - private static boolean isError(SgnlReport response) { - return response.getError() != null - && (!StringUtils.isEmpty(response.getError().getDescription()) || !StringUtils.isEmpty(response.getError().getName())); - } - - private static String errorDescription(SgnlReport response) { - if (response.getError() == null) - return "unknown error"; - return "%s: %s".formatted(response.getError().getName(), response.getError().getDescription()); - } - /** * @return Список отчетов, по которым сработали сигналы из официального сервиса сигналов */ @@ -324,40 +314,6 @@ public class SignalRestService { return Optional.ofNullable(targetFilePathWithExt); } - private String unzipReport(ResponseEntity response, Long packageId, String reportName) { - boolean isGzipped = response.getHeaders().getContentType() != null - && response.getHeaders().getContentType().toString().equals("application/x-gzip"); - - if (!isGzipped && isXmlError(response.getBody())) { - SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class); - if (error.getError().getDescription() != null) { - log.error("Ошибка при скачивание отчета. fNameReport={}. Описание: {}", reportName, error.getError().getDescription()); - return null; - } - log.error("Неизвестная ошибка при скачивание отчета. fNameReport={}", reportName); - return null; - } else { - String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator; - String filePath = folderPath + UUID.randomUUID() + reportName + ".xml"; - CTUtil.createFolder(folderPath); - - try (InputStream in = new GZIPInputStream(new ByteArrayInputStream(response.getBody())); - OutputStream out = Files.newOutputStream(Path.of(filePath))) { - - byte[] buffer = new byte[8 * 1024 * 1024]; - int bytesRead; - while ((bytesRead = in.read(buffer)) > 0) { - out.write(buffer, 0, bytesRead); - } - - return filePath; - } catch (Exception e) { - log.error(e.getMessage(), e); - return null; - } - } - } - private String buildToken(String signalsToken, String signalsuser) throws Exception { try { MessageDigest md = MessageDigest.getInstance("MD5"); @@ -371,25 +327,20 @@ public class SignalRestService { } } - public Resource compressFile(String path) throws Exception { - try (InputStream in = Files.newInputStream(Path.of(path)); - OutputStream out = Files.newOutputStream(Path.of(path + ".gz")); - GZIPOutputStream gzipOut = new GZIPOutputStream(out)) { - - byte[] buffer = new byte[8 * 1024 * 1024]; - int bytesRead; - while ((bytesRead = in.read(buffer)) > 0) { - gzipOut.write(buffer, 0, bytesRead); - } - gzipOut.finish(); - - return new FileUrlResource(path + ".gz"); - } - } - private boolean isXmlError(byte[] data) { if (data == null || data.length < 10) return false; String start = new String(data, 0, 10).trim(); return start.startsWith(" idEnricherQueueFactory, @@ -53,7 +55,8 @@ public class PipelinePackageManager implements InitializingBean { Function> pipelineStarterFactory, FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, CreditTrackerSettings creditTrackerSettings, - PipelineBackoff pipelineBackoff, Saver saver) { + PipelineBackoff pipelineBackoff, Saver saver, + MonitoringErrorProcesingService monitoringErrorProcesingService) { this.signalRestService = signalRestService; this.idEnricherQueueFactory = idEnricherQueueFactory; this.nameEnricherQueueFactory = nameEnricherQueueFactory; @@ -67,6 +70,7 @@ public class PipelinePackageManager implements InitializingBean { this.creditTrackerSettings = creditTrackerSettings; this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); this.pipelineBackoff = pipelineBackoff; + this.monitoringErrorProcesingService = monitoringErrorProcesingService; } public Pipeline buildAndRun(SetupParam setupParam) throws Exception { @@ -111,6 +115,7 @@ public class PipelinePackageManager implements InitializingBean { boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath()); if (success) { userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); + monitoringErrorProcesingService.findErrors(packageId); } } catch (Exception ex) { log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); 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 e22a1d6..54b1029 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 @@ -73,6 +73,7 @@ public class FidNameEnricher implements Handler }).toList(); phoneService2.updateFid(subjectForSave); + phoneService2.registerSignals(subjectForSave); } private void processByFid(List subjects) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java index 6fef2ec..92ca9a1 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java @@ -59,37 +59,29 @@ public class FlagsEnricher implements Consumer { Map flag90ByPastDueArrear = flagsService.findFlag90ByPastDueArrear(fids, yearAgo); Map flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(fids, yearAgo); - List flagDetails = subjects.stream().map(subject -> { - Integer debt5000 = null; - Integer flagDays90 = null; - LocalDateTime calculatedAt = null; - - if (subject.getFid() != null) { - debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0); + List flagDetails = subjects.stream() + .filter(subjectProfile -> subjectProfile.getFid() != null) + .map(subject -> { + Integer debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0); int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0); int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0); - flagDays90 = Math.max(pastDue90, paymt90); - calculatedAt = LocalDateTime.now(); - } + Integer flagDays90 = Math.max(pastDue90, paymt90); - Integer finalDebt500 = debt5000; - Integer finalFlagDays9 = flagDays90; - LocalDateTime finalCalculatedAt = calculatedAt; + return FlagDetail.builder() + .mapId(subject.getMapId()) + .subjectId(subject.getIdentifier()) + .packageId(subjectPayload.getPackageId()) + .fid(subject.getFid()) + .phone(subject.getPhone()) + .flagDebt5000(debt5000) + .flag90days(flagDays90) + .calculatedAt(LocalDateTime.now()) + .build(); + }).toList(); - return FlagDetail.builder() - .mapId(subject.getMapId()) - .subjectId(subject.getIdentifier()) - .packageId(subjectPayload.getPackageId()) - .fid(subject.getFid()) - .phone(subject.getPhone()) - .flagDebt5000(finalDebt500) - .flag90days(finalFlagDays9) - .calculatedAt(finalCalculatedAt) - .build(); - }).toList(); + log.debug("updating calculated flags size {}", flagDetails.size()); + flagsService.updateCalculationFlags(flagDetails); if (update) { - log.debug("updating calculated flags size {}", flagDetails.size()); - flagsService.updateCalculationFlags(flagDetails); //если signals_phone_fid_map#fid == null, удалить signals_flags с таким map_id List phoneFidMapIdsWithNullFid = subjects.stream() .filter(subject -> subject.getFid() == null) @@ -100,8 +92,6 @@ public class FlagsEnricher implements Consumer { log.debug("deleting flags for signals_phone_fid_map#fid == null; amount: {}", phoneFidMapIdsWithNullFid.size()); flagsService.deleteFlags(phoneFidMapIdsWithNullFid); } - } else { - flagsService.registerCalculationFlags(flagDetails); } } 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 bd6e334..5a23bd0 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 @@ -10,10 +10,10 @@ import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.exceptions.SignalClientException; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @@ -25,7 +25,8 @@ public abstract class KickPipeLine implements IPipeLineStarter { private final Path pkgPath; private final StaxXmlProcessor xmlProcessor; protected final UserPackageMapper userPackageMapper; - protected final PhoneService2 phoneService2; + protected final PhoneFidMapMapper phoneFidMapMapper; + protected final CreditTrackerSettings creditTrackerSettings; private final int BATCH_SIZE; private boolean isInterrupted = false; protected int offset = 0; @@ -34,13 +35,14 @@ public abstract class KickPipeLine implements IPipeLineStarter { public KickPipeLine(StaxXmlProcessor xmlProcessor, SetupParam setupParam, UserPackageMapper userPackageMapper, - PhoneService2 phoneService2, + PhoneFidMapMapper phoneFidMapMapper, CreditTrackerSettings creditTrackerSettings) { this.packageId = setupParam.packageId(); this.pkgPath = setupParam.xmlReqPath(); this.xmlProcessor = xmlProcessor; this.userPackageMapper = userPackageMapper; - this.phoneService2 = phoneService2; + this.phoneFidMapMapper = phoneFidMapMapper; + this.creditTrackerSettings = creditTrackerSettings; this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); } 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 f304c09..89012b5 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 @@ -3,8 +3,8 @@ package ru.nbch.credit_tracker.service.task.impl; import java.util.List; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhoneFidMap; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @@ -13,9 +13,9 @@ public class KickPipeLineDefaultImpl extends KickPipeLine { public KickPipeLineDefaultImpl(StaxXmlProcessor xmlProcessor, SetupParam setupParam, UserPackageMapper userPackageMapper, - PhoneService2 phoneService2, + PhoneFidMapMapper phoneFidMapMapper, CreditTrackerSettings creditTrackerSettings) { - super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + super(xmlProcessor, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings); } @Override @@ -25,7 +25,7 @@ public class KickPipeLineDefaultImpl extends KickPipeLine { @Override protected List nextPage(Integer limit) { - return phoneService2.pageByPackageId(packageId, limit, offset); + return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java new file mode 100644 index 0000000..f850af6 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java @@ -0,0 +1,43 @@ +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; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; + +public class KickPipeLineRepeatedRestoreImpl extends KickPipeLine { + + private boolean oldRowsWereFound = false; + private final Integer daysToUpdateFlags; + public KickPipeLineRepeatedRestoreImpl(StaxXmlProcessor xmlProcessor, + SetupParam setupParam, + UserPackageMapper userPackageMapper, + PhoneFidMapMapper phoneFidMapMapper, + CreditTrackerSettings creditTrackerSettings) { + super(xmlProcessor, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings); + this.daysToUpdateFlags = creditTrackerSettings.getDaysToUpdateFlags().intValue(); + } + + @Override + protected PipelineMode mode() { + return PipelineMode.RESTORE; + } + + @Override + protected List nextPage(Integer limit) { + List pageWithNullFids = new ArrayList<>(); + if (!oldRowsWereFound) { + pageWithNullFids = phoneFidMapMapper.pageByPackageIdAndOlderThanN(packageId, limit, offset, daysToUpdateFlags); + } + if (pageWithNullFids.isEmpty()) { + oldRowsWereFound = true; + offset = 0; + return phoneFidMapMapper.pageByPackageIdAndNewerThanN(packageId, limit, offset, daysToUpdateFlags); + } + return pageWithNullFids; + } +} 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 eaeae10..2f504bd 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 @@ -4,8 +4,8 @@ import java.util.ArrayList; import java.util.List; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhoneFidMap; +import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @@ -15,9 +15,9 @@ public class KickPipeLineRestoreImpl extends KickPipeLine { public KickPipeLineRestoreImpl(StaxXmlProcessor xmlProcessor, SetupParam setupParam, UserPackageMapper userPackageMapper, - PhoneService2 phoneService2, + PhoneFidMapMapper phoneFidMapMapper, CreditTrackerSettings creditTrackerSettings) { - super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + super(xmlProcessor, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings); } @Override @@ -29,12 +29,12 @@ public class KickPipeLineRestoreImpl extends KickPipeLine { protected List nextPage(Integer limit) { List pageWithNullFids = new ArrayList<>(); if (!nullFidsWereFound) { - pageWithNullFids = phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset); + pageWithNullFids = phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset); } if (pageWithNullFids.isEmpty()) { nullFidsWereFound = true; offset = 0; - return phoneService2.pageByPackageIdWithNotNullFids(packageId, limit, offset); + return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset); } return pageWithNullFids; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java index 91403b6..bab15ac 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java @@ -1,6 +1,12 @@ package ru.nbch.credit_tracker.service.task.impl; +/** + * DEFAULT - обычный режим при первичной постановки пакета на мониторинг; + * RESTORE - режим, в котором предусмотренно восстановление работы, после падения при штатной обработке пакета; + * REPEATED_RESTORE - режим, в котором предусмотрено восстановление, после падения в момент перерасчета флагов; + */ public enum PipelineMode { DEFAULT, - RESTORE + RESTORE, + REPEATED_RESTORE } diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 82773e5..d9102c4 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -71,12 +71,22 @@ update signals_phone_fid_map dst - set fid = src.fid, - calculated_at = src.calculated_at + set fid = src.fid, + searched_at = now(), + error_text = null FROM json_to_recordset(#{json}::json) AS src( id bigint, - fid int, - calculated_at timestamp + fid int + ) + where dst.id = src.id; + + + + update signals_phone_fid_map dst + set error_text = src.error_text + FROM json_to_recordset(#{json}::json) AS src( + id bigint, + error_text text ) where dst.id = src.id; @@ -149,9 +159,6 @@ subject_id, phone, fid, - flag_90_12, - flag_debt5000, - calculated_at, error_text FROM signals_phone_fid_map WHERE package_id = #{packageId} @@ -164,10 +171,6 @@ 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 null @@ -181,9 +184,6 @@ 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 @@ -191,6 +191,32 @@ LIMIT #{limit} OFFSET #{offset} + + + + + @@ -88,9 +90,15 @@ WHERE id = #{id} + + UPDATE signals_user_packages + SET reject_status = #{status} + WHERE id = #{id} + + - SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at + SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status FROM signals_user_packages WHERE membercode IS NOT NULL AND LEFT(membercode, 6) = #{userId} @@ -110,7 +118,7 @@ - SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at + SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status FROM signals_user_packages WHERE status in @@ -127,7 +135,7 @@