From 5d3b973919972f6fe25ba451794ca3c7c13ef006 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Wed, 26 Nov 2025 20:25:06 +0300 Subject: [PATCH] new logic [setup on monitoring; download report] --- credit-tracker.properties | 2 +- .../CreditTrackerApplication.java | 5 +- .../component/MonitoringRequestValidator.java | 2 +- .../SimpleMessageResolver.java | 4 +- .../config/rest/GlobalExceptionHandler.java | 8 +- .../{model => enums}/PoisonType.java | 2 +- .../credit_tracker/facade/SignalFacadeV3.java | 26 +- .../mapper/signal/FlagDetailMapper.java | 2 + .../mapper/signal/UserPackageMapper.java | 2 +- .../model/impl/PartFilePayload.java | 2 +- .../model/impl/SignalXmlFilePayload.java | 11 +- .../model/impl/SubjectPayload.java | 2 +- .../MonitoringErrorProcesingService.java | 5 +- .../service/MonitoringService.java | 273 ++++++++---------- .../credit_tracker/service/SignalService.java | 146 ++-------- .../service/UpdatingFlagsService.java | 15 +- ...ServiceV2.java => UserPackageService.java} | 7 +- .../{utils => service}/ValidationService.java | 2 +- .../service/signal/SignalRestService.java | 34 ++- .../service/task/PipelinePackageManager.java | 10 +- .../service/task/impl/KickPipeLineImpl.java | 2 +- .../service/xml/stax/StaxXmlWriter.java | 3 +- .../ru/nbch/credit_tracker/utils/CTUtil.java | 37 +-- .../credit_tracker/utils/CtFileUtils.java | 17 ++ .../nbch/credit_tracker/utils/ZipUtils.java | 60 +++- .../mapper/signal/FlagDetailMapper.xml | 22 ++ .../mapper/signal/UserPackageMapper.xml | 5 +- .../utils/ValidationServiceTest.java | 1 + 28 files changed, 334 insertions(+), 373 deletions(-) rename src/main/java/ru/nbch/credit_tracker/{utils/error => component}/SimpleMessageResolver.java (82%) rename src/main/java/ru/nbch/credit_tracker/{model => enums}/PoisonType.java (64%) rename src/main/java/ru/nbch/credit_tracker/service/{SignalServiceV2.java => UserPackageService.java} (91%) rename src/main/java/ru/nbch/credit_tracker/{utils => service}/ValidationService.java (98%) create mode 100644 src/main/java/ru/nbch/credit_tracker/utils/CtFileUtils.java diff --git a/credit-tracker.properties b/credit-tracker.properties index 3eaf575..daf6084 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -56,7 +56,7 @@ spring.servlet.multipart.max-request-size=${ACCEPTED_ARCHIVE_SIZE:2048MB} ct.xsd-validation.monitoring-request-path=monitoring-request.xsd ct.xsd-validation.signal-package-path=signal-package.xsd -ct.signals-root-uri=${SIGNALS_ROOT_URI:http://127.0.01:11094/signalRUalfa/} +ct.signals-root-uri=${SIGNALS_ROOT_URI:http://10.230.227.102:11094/signalRUalfa/} ct.signals-token=${SIGNALS_TOKEN:gttcHxwcGojoAvZK} ct.signals-user=${SIGNALS_USER:0001ZZ000001} diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index 891ec93..5934d6c 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ConfigurableApplicationContext; +import ru.nbch.credit_tracker.service.MonitoringService; @SpringBootApplication public class CreditTrackerApplication { @@ -10,10 +11,10 @@ public class CreditTrackerApplication { public static void main(String[] args) { ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); // UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class); -// MonitoringService monitoringService = ctx.getBean(MonitoringService.class); + MonitoringService monitoringService = ctx.getBean(MonitoringService.class); // UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class); // updateProcessingPackagesService.run(); -// monitoringService.run(); + monitoringService.run(); // updatingFlagsService.run(); } diff --git a/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java b/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java index b28b163..c4880bf 100644 --- a/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java +++ b/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java @@ -11,7 +11,7 @@ import ru.nbch.credit_tracker.exceptions.SignalException; import ru.nbch.credit_tracker.model.MonitoringValidationError; import ru.nbch.credit_tracker.model.ValidationResult; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.ValidationService; +import ru.nbch.credit_tracker.service.ValidationService; import ru.nbch.credit_tracker.utils.ZipUtils; import java.io.BufferedInputStream; diff --git a/src/main/java/ru/nbch/credit_tracker/utils/error/SimpleMessageResolver.java b/src/main/java/ru/nbch/credit_tracker/component/SimpleMessageResolver.java similarity index 82% rename from src/main/java/ru/nbch/credit_tracker/utils/error/SimpleMessageResolver.java rename to src/main/java/ru/nbch/credit_tracker/component/SimpleMessageResolver.java index 53924e8..352b8ab 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/error/SimpleMessageResolver.java +++ b/src/main/java/ru/nbch/credit_tracker/component/SimpleMessageResolver.java @@ -1,10 +1,12 @@ -package ru.nbch.credit_tracker.utils.error; +package ru.nbch.credit_tracker.component; +import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.enums.EnumKeyMessage; import java.util.Arrays; +@Component public class SimpleMessageResolver { public String resolve(EnumKeyMessage errorMessage) { if (errorMessage == null) return "null"; diff --git a/src/main/java/ru/nbch/credit_tracker/config/rest/GlobalExceptionHandler.java b/src/main/java/ru/nbch/credit_tracker/config/rest/GlobalExceptionHandler.java index 2b17835..dd687d1 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/rest/GlobalExceptionHandler.java +++ b/src/main/java/ru/nbch/credit_tracker/config/rest/GlobalExceptionHandler.java @@ -21,16 +21,18 @@ import ru.nbch.credit_tracker.enums.SignalError; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.exceptions.SignalException; import ru.nbch.credit_tracker.exceptions.SignalServerException; -import ru.nbch.credit_tracker.utils.error.SimpleMessageResolver; +import ru.nbch.credit_tracker.component.SimpleMessageResolver; @Slf4j @RestControllerAdvice public class GlobalExceptionHandler { - private final SimpleMessageResolver messageResolver = new SimpleMessageResolver(); + private final SimpleMessageResolver messageResolver; private final String encoding; - public GlobalExceptionHandler(CreditTrackerSettings settings) { + public GlobalExceptionHandler(SimpleMessageResolver messageResolver, + CreditTrackerSettings settings) { + this.messageResolver = messageResolver; this.encoding = settings.getEncoding(); } diff --git a/src/main/java/ru/nbch/credit_tracker/model/PoisonType.java b/src/main/java/ru/nbch/credit_tracker/enums/PoisonType.java similarity index 64% rename from src/main/java/ru/nbch/credit_tracker/model/PoisonType.java rename to src/main/java/ru/nbch/credit_tracker/enums/PoisonType.java index 2917ddf..f06a4b5 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/PoisonType.java +++ b/src/main/java/ru/nbch/credit_tracker/enums/PoisonType.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.model; +package ru.nbch.credit_tracker.enums; public enum PoisonType { NONE, 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 c117ca5..1f81b2b 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -35,7 +35,7 @@ import ru.nbch.credit_tracker.enums.XmlErrorFormat; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.model.MonitoringValidationError; import ru.nbch.credit_tracker.model.ValidationResult; -import ru.nbch.credit_tracker.service.SignalServiceV2; +import ru.nbch.credit_tracker.service.UserPackageService; import ru.nbch.credit_tracker.service.scoring.AuthorizationService; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.PipelinePackageManager; @@ -52,19 +52,19 @@ public class SignalFacadeV3 { private final MonitoringRequestValidator monitoringRequestValidator; private final CreditTrackerSettings config; - private final SignalServiceV2 signalService; + private final UserPackageService userPackageService; private final AuthorizationService authorizationService; private final PipelinePackageManager pipelinePackageManager; private final SignalRestService signalRestService; public SignalFacadeV3(MonitoringRequestValidator monitoringRequestValidator, CreditTrackerSettings config, - SignalServiceV2 signalService, + UserPackageService userPackageService, AuthorizationService authorizationService, PipelinePackageManager pipelinePackageManager, SignalRestService signalRestService) { this.monitoringRequestValidator = monitoringRequestValidator; this.config = config; - this.signalService = signalService; + this.userPackageService = userPackageService; this.authorizationService = authorizationService; this.pipelinePackageManager = pipelinePackageManager; this.signalRestService = signalRestService; @@ -78,7 +78,7 @@ public class SignalFacadeV3 { Path tmpReqFilePath; try { String folderPath = config.getLoadDir() + File.separator + "tmp" + File.separator; - tmpReqFilePath = ZipUtils.writeTmpFile(folderPath, file); + tmpReqFilePath = ZipUtils.unzipAndGetFirstEntry(folderPath, file); } catch (IOException e) { throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e); } @@ -89,7 +89,7 @@ public class SignalFacadeV3 { validationResult = monitoringRequestValidator.readAndValidateMonitoringRequest(tmpReqFilePath); if (!validationResult.getValidationErrors().isEmpty()) { MonitoringValidationError firstError = validationResult.getValidationErrors().getFirst(); - signalService.rejectPackage(validationResult.getUserId(), validationResult.getPackageName()); + userPackageService.rejectPackage(validationResult.getUserId(), validationResult.getPackageName()); Files.delete(tmpReqFilePath); throw firstError.getSignalException(); } @@ -116,7 +116,7 @@ public class SignalFacadeV3 { // проверка авторизации AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(userId, password); if (!authResult.isSuccess()) { - signalService.rejectPackage(userId, packageName); + userPackageService.rejectPackage(userId, packageName); throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringError); } @@ -126,7 +126,7 @@ public class SignalFacadeV3 { // throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError); // } - Long packageId = signalService.registerPackage(userId, packageName); + Long packageId = userPackageService.registerPackage(userId, packageName); LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId)); log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name()); pipelinePackageManager.buildAndRun(packageId, parsedFilePath); @@ -149,7 +149,7 @@ public class SignalFacadeV3 { throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringActivePackagesError); } - List activePackages = signalService.findActivePackages(activePackagesRequest.getUserId()); + List activePackages = userPackageService.findActivePackages(activePackagesRequest.getUserId()); if (activePackages.isEmpty()) { throw clientExcp(SignalError.ERROR_015, XmlErrorFormat.MonitoringActivePackagesError); } @@ -174,7 +174,7 @@ public class SignalFacadeV3 { throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringAllPackagesError); } - List activePackages = signalService.findAllPackages(allPackagesRequest.getUserId()); + List activePackages = userPackageService.findAllPackages(allPackagesRequest.getUserId()); if (activePackages.isEmpty()) { throw clientExcp(SignalError.ERROR_019, XmlErrorFormat.MonitoringAllPackagesError); } @@ -198,7 +198,7 @@ public class SignalFacadeV3 { } public ResponseEntity deletePackage(String username, String packageName) { - UserPackage packageToDelete = signalService.findByMemberCodeAndPackageName(username, packageName) + UserPackage packageToDelete = userPackageService.findByMemberCodeAndPackageName(username, packageName) .stream() .max(DeletePackageStatusComparator.INSTANCE) .orElseThrow(() @@ -215,7 +215,7 @@ public class SignalFacadeV3 { } LocalDateTime deactivatedAt = LocalDateTime.now(); - signalService.deletePackageById(packageToDelete.getId(), deactivatedAt); + userPackageService.deletePackageById(packageToDelete.getId(), deactivatedAt); MonitoringPackageDeleteResponse response = new MonitoringPackageDeleteResponse(); response.setStatus("Deleted"); response.setPackageName(packageName); @@ -239,7 +239,7 @@ public class SignalFacadeV3 { public ResponseEntity getReportList(String username, String packageName) { try { - UserPackage userPackage = signalService + UserPackage userPackage = userPackageService .findByMemberCodeAndPackageName(username, packageName) .stream() .max(ReportListPackageStatusComparator.INSTANCE) diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java index 8ff1adc..1b9436d 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/FlagDetailMapper.java @@ -2,8 +2,10 @@ package ru.nbch.credit_tracker.mapper.signal; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; +import ru.nbch.credit_tracker.entities.app.FlagDetail; @Mapper public interface FlagDetailMapper { void registerCalcFlags(@Param("json") String json); + FlagDetail getByMapId(@Param("mapId") Long mapId); } 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 8134408..e11d5be 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 @@ -32,7 +32,7 @@ public interface UserPackageMapper { List findUserPackages(@Param("userId") String userId, @Param("statuses") List status); - List findAllPackagesByStatus(@Param("status") String status); + List findAllPackagesByStatus(@Param("statuses") String... statuses); List findAllPackagesByUserId(@Param("userId") String userId); List findByMemberCodeAndPackageName(@Param("memberCode") String memberCode, diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java index 52015ab..f2cb147 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java @@ -2,7 +2,7 @@ package ru.nbch.credit_tracker.model.impl; import java.nio.file.Path; import lombok.Getter; -import ru.nbch.credit_tracker.model.PoisonType; +import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.model.StagePayload; @Getter diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java index 4f1694a..9955681 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java @@ -3,14 +3,5 @@ package ru.nbch.credit_tracker.model.impl; import java.nio.file.Path; import ru.nbch.credit_tracker.model.TerminatePayload; -public class SignalXmlFilePayload implements TerminatePayload { - private final Path signalXmlFilePath; - - public SignalXmlFilePayload(Path signalXmlFilePath) { - this.signalXmlFilePath = signalXmlFilePath; - } - - public Path getSignalXmlFilePath() { - return signalXmlFilePath; - } +public record SignalXmlFilePayload(Path signalXmlFilePath) implements TerminatePayload { } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java index 60938a3..bd68f1b 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java @@ -1,7 +1,7 @@ package ru.nbch.credit_tracker.model.impl; import java.util.List; -import ru.nbch.credit_tracker.model.PoisonType; +import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.model.StagePayload; import ru.nbch.credit_tracker.model.SubjectProfile; 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 a82a9c9..63809e6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.service; +import java.nio.file.Path; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.aop.framework.AopContext; @@ -8,7 +9,7 @@ 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.service.signal.SignalRestService; -import ru.nbch.credit_tracker.utils.CTUtil; +import ru.nbch.credit_tracker.utils.CtFileUtils; @Slf4j @Service @@ -44,7 +45,7 @@ public class MonitoringErrorProcesingService { } catch (Exception e) { log.error("Ошибка при записи сообщений об ошибках в сформированных субъектах для мониторинга. Package_id = {}", packageId, e); } - CTUtil.removeFile(filePath); + CtFileUtils.deleteQuietly(Path.of(filePath)); } } } 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 263b96c..17aab90 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java @@ -1,13 +1,27 @@ package ru.nbch.credit_tracker.service; +import java.io.BufferedInputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.slf4j.MDC; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.FlagDetail; import ru.nbch.credit_tracker.entities.app.Hit; -import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.external_response.online_download.Inquiry; import ru.nbch.credit_tracker.entities.external_response.online_download.Person; @@ -18,25 +32,16 @@ import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response. import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.Signal; import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.Signals; 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.UserPackageMapper; 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; import ru.nbch.credit_tracker.utils.CTUtil; - -import java.io.*; -import java.nio.file.Files; -import java.nio.file.Path; -import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.Comparator; -import java.util.List; -import java.util.UUID; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.zip.ZipEntry; -import java.util.zip.ZipOutputStream; +import ru.nbch.credit_tracker.utils.CtFileUtils; +import ru.nbch.credit_tracker.utils.ZipUtils; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; /** * Сервис периодического опроса официального сервиса сигналов на наличие новых отчетов по пакетам находящимся в статусе ON_MONITORING @@ -47,15 +52,21 @@ public class MonitoringService { private final SignalRestService signalRestService; private final SignalService signalService; + private final FlagDetailMapper flagDetailMapper; + private final UserPackageMapper userPackageMapper; private final StaxXmlProcessor staxXmlProcessor; private final CreditTrackerSettings config; public MonitoringService(SignalRestService signalRestService, SignalService signalService, + FlagDetailMapper flagDetailMapper, + UserPackageMapper userPackageMapper, StaxXmlProcessor staxXmlProcessor, CreditTrackerSettings config) { this.signalRestService = signalRestService; this.signalService = signalService; + this.flagDetailMapper = flagDetailMapper; + this.userPackageMapper = userPackageMapper; this.staxXmlProcessor = staxXmlProcessor; this.config = config; } @@ -67,14 +78,21 @@ public class MonitoringService { } private void monitor() { - log.info("Start monitoring"); - List packages = signalService.findPackagesOnMonitoring(); - for (UserPackage userPackage : packages) { - MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() +"]"); - processPackage(userPackage); - MDC.remove("ru.nbch.credit_tracker.log.mdc.key"); + try { + log.info("Start monitoring"); + List packages = userPackageMapper.findAllPackagesByStatus( + Status.ON_MONITORING.name(), + Status.PROCESSING_REPEATED.name() + ); + for (UserPackage userPackage : packages) { + MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() + "]"); + processPackage(userPackage); + MDC.remove("ru.nbch.credit_tracker.log.mdc.key"); + } + log.info("Finish monitoring"); + } catch (Exception e) { + log.error("{}", ExceptionUtils.getStackTrace(e)); } - log.info("Finish monitoring"); } private void processPackage(UserPackage userPackage) { @@ -91,8 +109,9 @@ public class MonitoringService { } // Формируем список имен файлов, по которым сработали сигналы. Сюда входят только те имена, у которых парсится дата формирования отчета - List fileNames = onlineReports.getOperationReport().getReports().getReports().stream().map(Report::getFilename) - .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName) != null).toList(); + List fileNames = onlineReports.getOperationReport().getReports().getReports().stream() + .map(Report::getFilename) + .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName) != null).toList(); if (fileNames.isEmpty()) { log.info("No new report found for package"); @@ -104,13 +123,13 @@ public class MonitoringService { if (lastDownloadedReportDate != null) { // Отбираем только те отчеты, дата которых позже даты последнего загруженного отчета fileNames = fileNames.stream() - .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName).isAfter(lastDownloadedReportDate)) // отфильтровали только те имена, которые парсятся в дату, так что null тут не придет - .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате - .toList(); + .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName).isAfter(lastDownloadedReportDate)) // отфильтровали только те имена, которые парсятся в дату, так что null тут не придет + .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате + .toList(); } else { fileNames = fileNames.stream() - .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате - .toList(); + .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате + .toList(); } if (StringUtils.isNotBlank(userPackage.getLastDownloadedReport())) { @@ -121,84 +140,81 @@ public class MonitoringService { for (String newReport : fileNames) { try { - String loadedReportPath = null; + Optional loadedReportPath = Optional.empty(); if (StringUtils.isBlank(userPackage.getLastDownloadedReport())) { - loadedReportPath = loadReport(userPackage.getId(), newReport); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным + // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным + loadedReportPath = signalRestService.downloadReport(userPackage, newReport); } else if (lastDownloadedReportDate != null) { LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport); if (newReportDate.isAfter(lastDownloadedReportDate)) { - loadedReportPath = loadReport(userPackage.getId(), newReport); // полученный отчет новее записанного в базу + // полученный отчет новее записанного в базу + loadedReportPath = signalRestService.downloadReport(userPackage, newReport); } } - if (loadedReportPath == null) { + if (loadedReportPath.isEmpty()) { log.info("Report was not loaded for package"); continue; } - String responsePath = generateResponse(userPackage, loadedReportPath); - String fileName = null; + Path responsePath = generateResponse(userPackage, newReport, loadedReportPath.get()); if (responsePath != null) { - fileName = createResponseFile(userPackage, responsePath); - } - if (fileName != null) { - log.info("New report fileName: {}", fileName); - updateSignalHits(userPackage, responsePath, fileName); - signalService.updateLastDownloadedReport(userPackage.getId(), newReport); + Path zipFile = ZipUtils.zipFile(responsePath); + log.info("New report file by path: {}", zipFile); + updateSignalHits(userPackage, responsePath.toString(), zipFile.toString()); + userPackageMapper.updatePackageLastDownloadedReport(userPackage.getId(), newReport); } else { log.info("Report was not created for package"); } } catch (Exception e) { - log.error(e.getMessage(), e); + log.error("Report was not created for package: {}", ExceptionUtils.getStackTrace(e)); } } } /** * Формирует результирующий xml файл для конечного пользователя на основе файла из официального сервиса сигналов + * * @return Путь к временному рещультирующему файлу */ - private String generateResponse(UserPackage userPackage, String loadedReportPath) { - AtomicBoolean generateResponse = new AtomicBoolean(false); // Выставляется в true, если хотя бы один субъект с мониторинга подошел под условие срабатывания сигнала. В этом случае отчет сохраняется + private Path generateResponse(UserPackage userPackage, String reportName, Path loadedReportPath) throws IOException { + // Выставляется в true, если хотя бы один субъект с мониторинга подошел под условие срабатывания сигнала. В этом случае отчет сохраняется + AtomicBoolean generateResponse = new AtomicBoolean(false); - String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator; - String responsePath = folderPath + UUID.randomUUID() + "_response_output.xml"; - CTUtil.createFolder(folderPath); + AtomicReference responsePath = new AtomicReference<>(loadedReportPath.resolveSibling( + userPackage.getUserId().substring(0, 6) + userPackage.getPackageName() + + CTUtil.getDateFromReportFileName(reportName) + ".xml")); MonitoringSignalResponse response = new MonitoringSignalResponse(); - response.setPackageName(userPackage.getPackageName()); // Имя пакета, переданного при установке + //Имя пакета, переданного при установке + response.setPackageName(userPackage.getPackageName()); // Запись в результрующий файл MonitoringSignalResponseStaxXmlWriter staxXmlWriter = new MonitoringSignalResponseStaxXmlWriter(config.getEncoding()); - staxXmlWriter.writeWithStax(responsePath, response, - writer -> { - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(loadedReportPath)), 8 * 1024 * 1024)) { - staxXmlProcessor.read(in, Persons.class, - personsData -> { // как только считали пачку Persons сразу записали ее в результирующий файл - List subjectsBatch = processPersons(personsData.getPersons()); - if (!subjectsBatch.isEmpty()) { - generateResponse.set(true); - } - writer.accept(subjectsBatch); - personsData.getPersons().clear(); // чистим список для экономии памяти - }, - pesons -> pesons.getPersons().isEmpty() - , - 10000, - "/Report/Persons/Person"); - } catch (IOException e) { - throw new RuntimeException("Error while parsing nbki downloaded report", e); + staxXmlWriter.writeWithStax(responsePath.toString(), response, + writer -> { + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(loadedReportPath), 8 * 1024 * 1024)) { + staxXmlProcessor.read(in, Persons.class, + personsData -> { // как только считали пачку Persons сразу записали ее в результирующий файл + List subjectsBatch = processPersons(personsData.getPersons()); + if (!subjectsBatch.isEmpty()) { + generateResponse.compareAndSet(false, true); + } + writer.accept(subjectsBatch); + personsData.getPersons().clear(); // чистим список для экономии памяти + }, + persons -> persons.getPersons().isEmpty(), + 10000, + "/Report/Persons/Person"); + } finally { + if (!generateResponse.get()) { // ни один субъект не подошел под условие, удаляем временный отчет + CtFileUtils.deleteQuietly(responsePath.get()); + responsePath.set(null); } - }); + } + }); - - CTUtil.removeFile(loadedReportPath); - - if (!generateResponse.get()) { // ни один субъект не подошел под условие, удаляем временный отчет - CTUtil.removeFile(responsePath); - return null; - } - - return responsePath; + CtFileUtils.deleteQuietly(loadedReportPath); + return responsePath.get(); } private List processPersons(List persons) { @@ -219,13 +235,13 @@ public class MonitoringService { // Если есть хотя бы один заполненный сигнал, значит заполняем остальные данные if (!signals.getSignals().isEmpty()) { - PhoneFidMap subjectData = signalService.findPhoneFidMapById(person.getUid()); - if (subjectData != null) { + FlagDetail flagDetail = flagDetailMapper.getByMapId(person.getUid()); + if (flagDetail != null) { Subject subject = new Subject(); - subject.setId(subjectData.getSubjectId()); // Уникальный идентификатор субъекта (равен Id из XML-пакета при установке мониторинга) - subject.setClientId(subjectData.getPhone()); - subject.setFlagDebt5000(subjectData.getFlagDebt5000()); - subject.setFlag9012(subjectData.getFlag90days()); + subject.setId(flagDetail.getSubjectId()); // Уникальный идентификатор субъекта (равен Id из XML-пакета при установке мониторинга) + subject.setClientId(flagDetail.getPhone()); + subject.setFlagDebt5000(flagDetail.getFlagDebt5000()); + subject.setFlag9012(flagDetail.getFlag90days()); subject.setSignals(signals); subjects.add(subject); } @@ -234,86 +250,39 @@ public class MonitoringService { return subjects; } - /** - * Архивирует результирующий xml файл для пользователя - * @return Путь к результрующему файлу. Пример //0001ZZ_MY_BATCH_TEST_0717_04_20250723_101929.zip - */ - private String createResponseFile(UserPackage userPackage, String responsePath) { - try { - String fileName = CTUtil.generateResponseReportFileName(userPackage.getUserId(), userPackage.getPackageName()); - Thread.sleep(1000); // Если отчеты формируются слишком быстро, нужно подождать секунду, тк в имени отчета LocalDateTime до секунд - String folderPath = config.getLoadDir() + File.separator + userPackage.getUserId().substring(0, 6) + File.separator; - - File folder = new File(folderPath); - if (!folder.exists()) { - folder.mkdir(); - } - - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 8 * 1024 * 1024); - BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(folderPath + fileName + ".zip")); - ZipOutputStream zos = new ZipOutputStream(bos)) { - - ZipEntry zipEntry = new ZipEntry(fileName + ".xml"); - zos.putNextEntry(zipEntry); - byte[] buffer = new byte[8 * 1024 * 1024]; - int bytesRead; - while ((bytesRead = in.read(buffer)) > 0) { - zos.write(buffer, 0, bytesRead); - } - zos.finish(); - zos.closeEntry(); - } - return fileName; - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } catch (Exception e) { - log.error("Ошибка при записи отчета пользователя в файл: ", e); - } - - CTUtil.removeFile(responsePath); - - return null; - } - private void updateSignalHits(UserPackage userPackage, String responsePath, String reportFileName) { log.info("Updating signal hits for {}", userPackage.getPackageName()); LocalDateTime responseSentAt = LocalDateTime.now(); // Считываем результирующий файл пачками и обновляем таблицу hits try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(responsePath)), 8 * 1024 * 1024)) { staxXmlProcessor.read(in, MonitoringSignalResponse.class, - response -> { // как только считали пачку Subjects сразу обновлям таблицу hits - List hits = new ArrayList<>(); - response.getSubjects().getSubjects().forEach(subject -> { - subject.getSignals().getSignals().forEach(signal -> { - Hit hit = new Hit(); - hit.setPackageId(userPackage.getId()); - hit.setSubjectId(subject.getId()); - hit.setSignalType(1); - hit.setResponseSentAt(responseSentAt); - hit.setReportFileName(reportFileName); - hits.add(hit); - }); - if (!hits.isEmpty()) { - signalService.insertHits(hits); - } + response -> { // как только считали пачку Subjects сразу обновлям таблицу hits + List hits = new ArrayList<>(); + response.getSubjects().getSubjects().forEach(subject -> { + subject.getSignals().getSignals().forEach(signal -> { + Hit hit = new Hit(); + hit.setPackageId(userPackage.getId()); + hit.setSubjectId(subject.getId()); + hit.setSignalType(1); + hit.setResponseSentAt(responseSentAt); + hit.setReportFileName(reportFileName); + hits.add(hit); }); - response.getSubjects().getSubjects().clear(); // чистим список для экономии памяти - }, - response -> response.getSubjects().getSubjects().isEmpty() - , - 10000, - "/MonitoringSignalResponse/Subjects/Subject"); + if (!hits.isEmpty()) { + signalService.insertHits(hits); + } + }); + response.getSubjects().getSubjects().clear(); // чистим список для экономии памяти + }, + response -> response.getSubjects().getSubjects().isEmpty() + , + 10000, + "/MonitoringSignalResponse/Subjects/Subject"); } catch (IOException e) { throw new RuntimeException("Error while updating hits", e); } - - CTUtil.removeFile(responsePath); // удалям результирующий xml файл с диска (Он уже заархивирован к этому моменту) - + CtFileUtils.deleteQuietly(Path.of(responsePath)); // удалям результирующий xml файл с диска (Он уже заархивирован к этому моменту) log.info("Signal hits updated for package {}", userPackage.getPackageName()); } - public String loadReport(Long packageId, String reportName) { - return signalRestService.downloadReport(packageId, reportName); - } - } 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 9b3c013..9ecf0e0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -1,5 +1,20 @@ package ru.nbch.credit_tracker.service; +import java.io.BufferedInputStream; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.function.Consumer; +import java.util.function.Predicate; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -13,9 +28,11 @@ import ru.nbch.credit_tracker.entities.external_request.signals.Auth; import ru.nbch.credit_tracker.entities.external_request.signals.Person; import ru.nbch.credit_tracker.entities.external_request.signals.Signals; import ru.nbch.credit_tracker.entities.external_response.monitoring_report_error.MonitoringReportErrors; -import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.enums.Status; -import ru.nbch.credit_tracker.service.indic.*; +import ru.nbch.credit_tracker.service.indic.FlagsService; +import ru.nbch.credit_tracker.service.indic.PersonService; +import ru.nbch.credit_tracker.service.indic.PhoneService; +import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.signal.UserPackageService; @@ -23,16 +40,6 @@ import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; import ru.nbch.credit_tracker.utils.CTUtil; -import java.io.BufferedInputStream; -import java.io.File; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.time.LocalDateTime; -import java.util.*; -import java.util.function.Consumer; -import java.util.function.Predicate; - @Slf4j @Service @Deprecated @@ -71,73 +78,6 @@ public class SignalService { this.staxXmlProcessor = staxXmlProcessor; } - public boolean isPackageNotUnique(MonitoringRequest request) { - return userPackageService.isPackageNotUnique(request.getUserId(), request.getPackageName()); - } - - public boolean isPackageNotUnique(String userId, String packageName) { - return userPackageService.isPackageNotUnique(userId, packageName); - } - - public Long registerPackage(MonitoringRequest request) { - UserPackage userPackage = UserPackage.builder() - .userId(request.getUserId()) - .packageName(request.getPackageName()) - .createdAt(LocalDateTime.now()) - .status(Status.PROCESSING.name()) - .build(); - return userPackageService.insertPackage(userPackage); - } - - public Long registerPackage(String userId, String packageName) { - UserPackage userPackage = UserPackage.builder() - .userId(userId) - .packageName(packageName) - .createdAt(LocalDateTime.now()) - .status(Status.PROCESSING.name()) - .build(); - return userPackageService.insertPackage(userPackage); - } - - public void registerSubjects(String filePath, Long packageId) { - log.info("Start registering subjects. PackageId: {}", packageId); - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) { - staxXmlProcessor.read(in, MonitoringRequest.class, - request -> { - if (!request.getSubjects().getSubject().isEmpty()) { - phoneFidMapService.registerSubjects(request.getSubjects().getSubject(), packageId); - log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId); - request.getSubjects().getSubject().clear(); - } - }, - request -> request.getSubjects().getSubject().isEmpty() - , - batchSize, "/MonitoringRequest/Subjects/Subject"); - } catch (IOException e) { - throw new RuntimeException("Error while parsing monitoring request", e); - } - log.info("Subjects were registered. PackageId: {}", packageId); - } - - /** - * Выгружает {@link PhoneFidMap} батчами. Передает обогащенные фидами и данными из Name {@link PhoneFidMap} - * на поиск флагов и формирование отчета в оф. сервис сигналов - * @param packageId id пакета, по которому идет выборка {@link PhoneFidMap} - */ - public void processSubjects(Long packageId) { - int offset = 0; - List batch; - do { - batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize); - if (!batch.isEmpty()) { - Map fidsMapByBatch = phoneService.defineFidsAndNameByPhone(batch); - // TODO process fidsMapByBatch here - } - offset += batchSize; - } while (!batch.isEmpty()); - } - - /** * Обрабатывает все PhoneFidMap по packageId, выгружая их пачками и фильтруя по переданному фильтру */ @@ -228,25 +168,6 @@ public class SignalService { return !fidsMap.isEmpty(); } - public void rejectPackage(MonitoringRequest request) { - UserPackage userPackage = UserPackage.builder() - .userId(request.getUserId()) - .packageName(request.getPackageName()) - .createdAt(LocalDateTime.now()) - .status(Status.REJECTED.name()) - .build(); - userPackageService.insertPackage(userPackage); - } - - public void rejectPackage(String userId, String packageName) { - UserPackage userPackage = UserPackage.builder() - .userId(userId) - .packageName(packageName) - .createdAt(LocalDateTime.now()) - .status(Status.REJECTED.name()) - .build(); - userPackageService.insertPackage(userPackage); - } public void updatePackageStatus(Long packageId, Status status) { userPackageService.updatePackageStatus(packageId, status); @@ -360,26 +281,6 @@ public class SignalService { log.info("Subject's errors were updated. PackageId: {}", packageId); } - public List findActivePackages(String userId) { - return userPackageService.findUserPackages(userId, Status.ON_MONITORING); - } - - public List findAllPackages(String userId) { - return userPackageService.findAllPackagesByUserId(userId); - } - - public boolean isPackageExists(String username, String packageName) { - return userPackageService.isPackageExists(username, packageName, Status.ON_MONITORING); - } - - public boolean isPackagesByStatusExist(String username, String packageName, Status status) { - return userPackageService.hasActivePackages(username, packageName, status); - } - - public void deletePackage(String username, String packageName, LocalDateTime deactivatedAt) { - userPackageService.deletePackage(username, packageName, Status.DELETED, Status.ON_MONITORING, deactivatedAt); - } - public List findPackagesOnMonitoring() { return userPackageService.findAllPackagesByStatus(Status.ON_MONITORING); } @@ -391,15 +292,6 @@ public class SignalService { public void insertHits(List hits) { hitsService.insertHits(hits); } - - public void updateLastDownloadedReport(Long id, String fileName) { - userPackageService.updatePackageLastDownloadedReport(id, fileName); - } - - public PhoneFidMap findPhoneFidMapById(Long id) { - return phoneFidMapService.findPhoneFidMapById(id); - } - public boolean hasNonNullfids(Long packageId) { return phoneFidMapService.hasNonNullfids(packageId); } 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 37776b3..72b31b2 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -1,5 +1,12 @@ package ru.nbch.credit_tracker.service; +import java.time.LocalDateTime; +import java.time.temporal.ChronoUnit; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; import lombok.extern.slf4j.Slf4j; import org.slf4j.MDC; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; @@ -10,14 +17,6 @@ import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.service.signal.SignalRestService; -import java.time.LocalDateTime; -import java.time.temporal.ChronoUnit; -import java.util.List; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.function.Predicate; - /** * Сервис периодического обновления фидов и флагов просрочек по пакетам находящимся в статусе ON_MONITORING * По обновленным пакетам формируется новый запрос постановки на мониторинг в официальный сервис сигналов diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java b/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java similarity index 91% rename from src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java rename to src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java index 62d31ac..1681fe0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java @@ -7,14 +7,13 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.enums.Status; -import ru.nbch.credit_tracker.service.signal.UserPackageService; @Slf4j @Service -public class SignalServiceV2 { - private final UserPackageService userPackageService; +public class UserPackageService { + private final ru.nbch.credit_tracker.service.signal.UserPackageService userPackageService; - public SignalServiceV2(UserPackageService userPackageService) { + public UserPackageService(ru.nbch.credit_tracker.service.signal.UserPackageService userPackageService) { this.userPackageService = userPackageService; } diff --git a/src/main/java/ru/nbch/credit_tracker/utils/ValidationService.java b/src/main/java/ru/nbch/credit_tracker/service/ValidationService.java similarity index 98% rename from src/main/java/ru/nbch/credit_tracker/utils/ValidationService.java rename to src/main/java/ru/nbch/credit_tracker/service/ValidationService.java index 9e7875d..edeb6c4 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/ValidationService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/ValidationService.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.utils; +package ru.nbch.credit_tracker.service; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; 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 fc418cc..6edb685 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 @@ -30,6 +30,7 @@ import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.client.RestTemplate; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.external_response.errors.SgnlReport; import ru.nbch.credit_tracker.entities.external_response.list_package.SgnlReportDeleted; import ru.nbch.credit_tracker.entities.external_response.monitoring_report_file.SgnlReportMonitorigSetupError; @@ -38,6 +39,7 @@ 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 @Service @@ -55,9 +57,10 @@ public class SignalRestService { private final StaxXmlProcessor staxProcessor; private final SignalService signalService; private final String authToken; - private final CreditTrackerSettings config; + private final static MediaType MEDIA_TYPE_GZIP = MediaType.parseMediaType("application/x-gzip"); + public SignalRestService(@Qualifier(value = "restTemplate") RestTemplate restTemplate, IXmlProcessor xmlProcessor, StaxXmlProcessor staxProcessor, @@ -77,14 +80,14 @@ public class SignalRestService { */ public boolean setupMonitoring(Long packageId) throws Exception { String path = signalService.writeSignals(packageId); - return setupMonitoring(path); + return setupMonitoring(Path.of(path)); } - public boolean setupMonitoring(String path) throws Exception { - Resource request = compressFile(path); + public boolean setupMonitoring(Path signalFilePath) throws Exception { + Path gzip = ZipUtils.packGzip(signalFilePath); MultiValueMap body = new LinkedMultiValueMap<>(); - body.add("file", request); + body.add("file", new FileUrlResource(gzip.toString())); HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.MULTIPART_FORM_DATA); @@ -93,11 +96,11 @@ public class SignalRestService { HttpEntity> requestEntity = new HttpEntity<>(body, headers); ResponseEntity response = restTemplate.postForEntity(UPDATE_URI, requestEntity, String.class); - + Files.delete(signalFilePath); + Files.delete(gzip); + log.debug("Received response: {}", response); if (response.getStatusCode().value() == 302) { // good response - CTUtil.removeFile(path); - CTUtil.removeFile(path + ".gz"); return true; } else if (response.getStatusCode().value() == 200 && response.getBody() != null) { SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class); @@ -294,7 +297,7 @@ public class SignalRestService { * * @return путь к разархивированному файлу. Пример: /tmp_folder//0001XX_MY_BATCH_204.1.online.2025-07-08T20-00-46.xml.gz.xml */ - public String downloadReport(Long packageId, String reportName) { + public Optional downloadReport(UserPackage userPackage, String reportName) { String url = DOWNLOAD_ONLINE_REPORT_URI + "?fNameReport=" + reportName; HttpHeaders headers = new HttpHeaders(); @@ -308,8 +311,17 @@ public class SignalRestService { requestEntity, byte[].class ); - - return unzipReport(response,packageId, reportName); + Path targetFilePathWithExt = null; + if (response.getHeaders().getContentType().equals(MEDIA_TYPE_GZIP)) { + targetFilePathWithExt = Path.of(config.getLoadDir(), + "tmp", userPackage.getUserId().substring(0, 6), userPackage.getPackageName(), reportName + ".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={}. Описание: {}", reportName, description); + } + return Optional.ofNullable(targetFilePathWithExt); } private String unzipReport(ResponseEntity response, Long packageId, String reportName) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java index 0dd6e0e..d7aed44 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java @@ -1,6 +1,7 @@ package ru.nbch.credit_tracker.service.task; import java.nio.file.Path; +import java.time.LocalDateTime; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -95,7 +96,7 @@ public class PipelinePackageManager implements InitializingBean { try { Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(fidNameEnricher) -// .handlerResult(flagsEnricherV2) + .handlerResult(flagsEnricherV2) .andThen() .stage(sendSignalQueueFactory.getObject()) .handler(fidIdEnricherHandler) @@ -122,8 +123,11 @@ public class PipelinePackageManager implements InitializingBean { } userPackageMapper.setLastSearchedDt(packageId); SignalXmlFilePayload res = (SignalXmlFilePayload) v; - log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath()); -// signalRestService.setupMonitoring(res.getSignalXmlFilePath().toString()); + log.trace("Signal file by path: {} is completed", res.signalXmlFilePath()); + boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath()); + if (success) { + userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); + } } catch (Exception ex) { log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); } finally { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java index f890a9e..4b2f6e0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java @@ -16,7 +16,7 @@ import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.model.PoisonType; +import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.PhoneService2; diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlWriter.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlWriter.java index b3d5d0a..5d89e63 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlWriter.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlWriter.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.service.xml.stax; +import java.io.IOException; import java.util.List; import java.util.function.Consumer; @@ -15,6 +16,6 @@ public abstract class StaxXmlWriter { public abstract void writeWithStax(String filePath, T obj, LoaderWriter loader); public interface LoaderWriter { - void accept(Consumer> writer); + void accept(Consumer> writer) throws IOException; } } diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index 46c8bb9..31b35c2 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java @@ -1,19 +1,15 @@ package ru.nbch.credit_tracker.utils; -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; - import java.io.File; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; -import java.time.LocalDate; import java.time.LocalDateTime; -import java.time.ZoneId; import java.time.format.DateTimeFormatter; -import java.util.Date; import java.util.regex.Matcher; import java.util.regex.Pattern; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; @Slf4j public class CTUtil { @@ -23,14 +19,21 @@ public class CTUtil { public static LocalDateTime parseDateFromReportFileName(String reportFileName) { + if (StringUtils.isBlank(reportFileName)) { + return null; + } + String dateTimeStr = getDateFromReportFileName(reportFileName); + return dateTimeStr != null ? LocalDateTime.parse(dateTimeStr, onlineReportDateTimeFormatter) : null; + } + + public static String getDateFromReportFileName(String reportFileName) { if (StringUtils.isBlank(reportFileName)) { return null; } Pattern pattern = Pattern.compile("\\d{4}-\\d{2}-\\d{2}T\\d{2}-\\d{2}-\\d{2}"); Matcher matcher = pattern.matcher(reportFileName); if (matcher.find()) { - String dateTimeStr = matcher.group(); - return LocalDateTime.parse(dateTimeStr, onlineReportDateTimeFormatter); + return matcher.group(); } return null; } @@ -72,13 +75,6 @@ public class CTUtil { return value.replaceAll("[^А-Яа-яA-Za-z0-9]", ""); } - public static Date localDatetoDate(LocalDate date) { - if (date == null) { - return null; - } - return Date.from(date.atStartOfDay(ZoneId.systemDefault()).toInstant()); - } - public static void createFolder(String folderPath) { File folder = new File(folderPath); if (!folder.exists()) { @@ -91,17 +87,6 @@ public class CTUtil { } } - public static void removeFile(String filePath) { - try { - Path path = Path.of(filePath); - if (Files.exists(path)) { - Files.delete(path); - } - } catch (IOException e) { - log.warn("File {} hasn't been deleted", filePath); - } - } - public static String normalizePhoneNumber(String phoneNumber) { return "9" + phoneNumber; } diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CtFileUtils.java b/src/main/java/ru/nbch/credit_tracker/utils/CtFileUtils.java new file mode 100644 index 0000000..2212169 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/utils/CtFileUtils.java @@ -0,0 +1,17 @@ +package ru.nbch.credit_tracker.utils; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class CtFileUtils { + public static void deleteQuietly(Path filePath) { + try { + Files.delete(filePath); + } catch (IOException e) { + log.warn("File {} hasn't been deleted", filePath); + } + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java b/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java index 296cfde..f17089e 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java @@ -1,13 +1,21 @@ package ru.nbch.credit_tracker.utils; +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.ByteArrayInputStream; +import java.io.FileOutputStream; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; import java.nio.file.Files; import java.nio.file.Path; import java.util.Optional; import java.util.UUID; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; +import java.util.zip.ZipOutputStream; import lombok.extern.slf4j.Slf4j; import org.springframework.web.multipart.MultipartFile; @@ -37,7 +45,7 @@ public class ZipUtils { file.getOriginalFilename() != null && file.getOriginalFilename().toLowerCase().endsWith(".zip"); } - public static Path writeTmpFile(String path, MultipartFile file) throws IOException { + public static Path unzipAndGetFirstEntry(String path, MultipartFile file) throws IOException { CTUtil.createFolder(path); Path filePath = Path.of(path, UUID.randomUUID() + "_monitoring_request.xml"); @@ -53,4 +61,54 @@ public class ZipUtils { return filePath; } } + + public static void unpackGzip(Path targetPath, byte[] content) { + try (InputStream in = new GZIPInputStream(new ByteArrayInputStream(content)); + OutputStream out = Files.newOutputStream(targetPath)) { + byte[] buffer = new byte[8 * 1024 * 1024]; + int bytesRead; + while ((bytesRead = in.read(buffer)) > 0) { + out.write(buffer, 0, bytesRead); + } + } catch (Exception e) { + log.error(e.getMessage(), e); + } + } + + public static Path packGzip(Path pathToFile) throws IOException { + Path targetPath = pathToFile.resolveSibling(pathToFile.getFileName().toString() + ".gz"); + try (InputStream in = Files.newInputStream(pathToFile); + OutputStream out = Files.newOutputStream(targetPath); + 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 targetPath; + } + } + + public static Path zipFile(Path fileForZip) throws IOException { + String fileName = fileForZip.getFileName().toString(); + Path zipFilePath = fileForZip.getParent().resolve(fileName + ".zip"); + try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(fileForZip), 8 * 1024 * 1024); + BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(zipFilePath.toString())); + ZipOutputStream zos = new ZipOutputStream(bos)) { + + ZipEntry zipEntry = new ZipEntry(fileName); + zos.putNextEntry(zipEntry); + byte[] buffer = new byte[8 * 1024 * 1024]; + int bytesRead; + while ((bytesRead = in.read(buffer)) > 0) { + zos.write(buffer, 0, bytesRead); + } + zos.finish(); + zos.closeEntry(); + } + return zipFilePath; + } } diff --git a/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml b/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml index 2a4013c..0611123 100644 --- a/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/FlagDetailMapper.xml @@ -4,6 +4,15 @@ "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> + + + + + + + + + INSERT INTO signals_flags @@ -18,4 +27,17 @@ flag_90_12 smallint, calculated_at timestamp) + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml index e4f74d5..1ecc25b 100644 --- a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml @@ -120,7 +120,10 @@ resultType="ru.nbch.credit_tracker.entities.app.UserPackage"> SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at FROM signals_user_packages - WHERE status = #{status} + WHERE status in + + #{status} +