diff --git a/credit-tracker.properties b/credit-tracker.properties index 57bae8a..1bb5e9e 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -1,3 +1,4 @@ +# TODO delete unused env vars server.port=${SERVER_PORT:8099} server.tomcat.accesslog.enabled=true server.tomcat.accesslog.directory=/opt/app/logs/access @@ -70,10 +71,6 @@ 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://10.230.227.25:9888/signalRUalfa/} -ct.signals-token=${SIGNALS_TOKEN:gttcHxwcGojoAvZK} -ct.signals-user=${SIGNALS_USER:0001ZZ000001} - ct.load-dir=/opt/app/reports #ct.load-dir=D:\\folder_name\\signals ct.tmp.dir=D:\\folder_name\\signals\\tmp @@ -112,9 +109,6 @@ ct.pipeline.subject-batch-size=${PIPELINE_SUBJECT_BATCH_SIZE:10000} #select NAME + FID BY PHONE ct.pipeline.name-enricher-stage-concurrency=${PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY:20} ct.pipeline.name-enricher-queue-capacity=${PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY:20} -#select ID BY FID -ct.pipeline.id-enricher-stage-concurrency=${PIPELINE_ID_ENRICHER_STAGE_CONCURRENCY:10} -ct.pipeline.id-enricher-queue-capacity=${PIPELINE_ID_ENRICHER_QUEUE_CAPACITY:20} #Save part of signal xml file and merge; only one thread ct.pipeline.merger-queue-capacity=${PIPELINE_MERGER_QUEUE_CAPACITY:50} @@ -122,9 +116,6 @@ ct.pipeline.merger-queue-capacity=${PIPELINE_MERGER_QUEUE_CAPACITY:50} #select NAME + FID BY PHONE ct.pipeline.name-updater-stage-concurrency=${PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY:20} ct.pipeline.name-updater-queue-capacity=${PIPELINE_NAME_UPDATER_QUEUE_CAPACITY:20} -#select ID BY FID -ct.pipeline.id-updater-stage-concurrency=${PIPELINE_ID_UPDATER_STAGE_CONCURRENCY:10} -ct.pipeline.id-updater-queue-capacity=${PIPELINE_ID_UPDATER_QUEUE_CAPACITY:20} #select and calc flags diff --git a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilder.java b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilder.java deleted file mode 100644 index de55e52..0000000 --- a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilder.java +++ /dev/null @@ -1,88 +0,0 @@ -package ru.nbch.credit_tracker.builder; - -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import ru.nbch.credit_tracker.entities.external_request.signals.Person; -import ru.nbch.credit_tracker.model.IdData; -import ru.nbch.credit_tracker.model.NameData; -import ru.nbch.credit_tracker.model.SubjectProfile; - -import java.time.LocalDate; -import java.time.format.DateTimeFormatter; - -@Slf4j -public class PersonFactoryBuilder { - private final static DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("dd.MM.yyyy"); - - private final static String DEFAULT_DATE = "01.01.2000"; - private final static String IP = "0"; - private final static String OKSM = "999"; - private final static String CONSENT_PURPOSE = "16"; - private final static String REPORT_USER = "АО «НБКИ»"; - private final static String LIABILITY = "1"; - private final static String CONSENT_PERIOD = "3"; - private final static String REPORT_USER_REG_NUM = "1057746710713"; - private final static String REPORT_USER_TAX_ID = "7703548386"; - private final static String CONSENT_HASH = "hash"; - private final static String CONSENT_DIV_CODE = "000-000"; - private final static String INQ_PURPOSE = "16"; - private final static String ODT = "Иной документ"; - private final static String CONSENT_ISSUE_AUTHORITY = "Кем выдан"; - - - private SubjectProfile personData; - - public static PersonFactoryBuilder builder() { - return new PersonFactoryBuilder(); - } - - public PersonFactoryBuilder personData(SubjectProfile personData) { - this.personData = personData; - return this; - } - - public Person build() { - if (personData == null) { - return null; - } - NameData nameData = personData.getNameData(); - IdData passportData = personData.getIdData(); - Person person = new Person(); - person.setS(nameData.getLastName()); - person.setN(nameData.getFirstName()); - person.setM(StringUtils.isNotBlank(nameData.getMiddleName()) ? nameData.getMiddleName() : null); - person.setD(DATE_FORMAT.format(nameData.getBirthDate())); - - person.setPn(passportData.getSerNum()); - person.setPd(DATE_FORMAT.format(LocalDate.now())); // Do not use passport iss_date. Signal service could reject the package - person.setDt(passportData.getIdType()); - if (person.getDt() != null && person.getDt().equals("999")) { - person.setOdt(ODT); - person.setConsentOdt(ODT); - } - person.setIp(IP); - person.setUid(personData.getMapId()); - person.setOksm(OKSM); - person.setConsentDate(DATE_FORMAT.format(LocalDate.now())); - person.setConsentPurpose(CONSENT_PURPOSE); - person.setReportUser(REPORT_USER); - person.setLiability(LIABILITY); - person.setConsentPeriod(CONSENT_PERIOD); - person.setAgrDate(DATE_FORMAT.format(LocalDate.now())); - person.setReportUserRegNum(REPORT_USER_REG_NUM); - person.setReportUserTaxID(REPORT_USER_TAX_ID); - person.setConsentHash(CONSENT_HASH); - - person.setCs(person.getS()); - person.setConsentN(person.getN()); - person.setConsentM(person.getM()); - person.setConsentBd(person.getD()); - person.setConsentPn(person.getPn()); - person.setConsentPd(person.getPd()); - person.setConsentDivCode(CONSENT_DIV_CODE); - person.setConsentDt(person.getDt()); - person.setConsentIssueAuthority(CONSENT_ISSUE_AUTHORITY); - person.setInqPurpose(INQ_PURPOSE); - return person; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/config/rest/ResponseErrorHandler.java b/src/main/java/ru/nbch/credit_tracker/config/rest/ResponseErrorHandler.java index d322448..fcf7bb5 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/rest/ResponseErrorHandler.java +++ b/src/main/java/ru/nbch/credit_tracker/config/rest/ResponseErrorHandler.java @@ -5,9 +5,11 @@ import org.springframework.http.client.ClientHttpResponse; import org.springframework.web.client.DefaultResponseErrorHandler; import org.springframework.web.client.HttpClientErrorException; import org.springframework.web.client.HttpServerErrorException; +import ru.nbch.credit_tracker.refactoring.Suspicios; import java.io.IOException; +@Suspicios("Не используется больше") @Slf4j public class ResponseErrorHandler extends DefaultResponseErrorHandler { @Override diff --git a/src/main/java/ru/nbch/credit_tracker/config/rest/RestConfig.java b/src/main/java/ru/nbch/credit_tracker/config/rest/RestConfig.java index c69bf21..a3df4dc 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/rest/RestConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/rest/RestConfig.java @@ -3,8 +3,6 @@ package ru.nbch.credit_tracker.config.rest; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; -import org.springframework.web.client.RestTemplate; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -12,16 +10,9 @@ import java.util.concurrent.Executors; @Configuration public class RestConfig { - @Bean(name = "restTemplate") - public RestTemplate restTemplate(CreditTrackerSettings config) { - RestTemplate restTemplate = new RestTemplate(); - restTemplate.setUriTemplateHandler(new RootUriTemplateHandler(config.getSignalsRootUri())); - restTemplate.setErrorHandler(new ResponseErrorHandler()); - return restTemplate; - } - @Bean(name = "executorService") public ExecutorService executorService() { return Executors.newCachedThreadPool(new CustomizableThreadFactory("package-registration-")); } + } diff --git a/src/main/java/ru/nbch/credit_tracker/config/rest/RootUriTemplateHandler.java b/src/main/java/ru/nbch/credit_tracker/config/rest/RootUriTemplateHandler.java index 49aed43..c78f68a 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/rest/RootUriTemplateHandler.java +++ b/src/main/java/ru/nbch/credit_tracker/config/rest/RootUriTemplateHandler.java @@ -1,10 +1,12 @@ package ru.nbch.credit_tracker.config.rest; import org.springframework.web.util.DefaultUriBuilderFactory; +import ru.nbch.credit_tracker.refactoring.Suspicios; import java.net.URI; import java.util.Map; +@Suspicios("Не используется больше") public class RootUriTemplateHandler extends DefaultUriBuilderFactory { private final String rootUri; diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index 9045599..0ef27f5 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java @@ -32,10 +32,6 @@ public class CreditTrackerSettings { @NestedConfigurationProperty private KafkaConsumerSettings kafkaConsumerSettings; - private String signalsRootUri; - private String signalsToken; - private String signalsuser; - private Long maxXmlFileSize; private String loadDir; private Long monitoringInterval; diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java index f6421b6..85e1967 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java @@ -1,8 +1,9 @@ package ru.nbch.credit_tracker.config.settings; +import org.springframework.boot.context.properties.bind.DefaultValue; + import java.time.Duration; import java.util.List; -import org.springframework.boot.context.properties.bind.DefaultValue; //X это количество потоков Save //тогда: Name должно быть ~3x, Fid должно быть ~2x чтобы они поспевали друг за другом? @@ -10,12 +11,8 @@ public record PipelineSettings( @DefaultValue("5000") Integer subjectBatchSize, @DefaultValue("15") Integer nameEnricherStageConcurrency, @DefaultValue("20") Integer nameEnricherQueueCapacity, - @DefaultValue("10") Integer idEnricherStageConcurrency, - @DefaultValue("15") Integer idEnricherQueueCapacity, @DefaultValue("15") Integer nameUpdaterStageConcurrency, @DefaultValue("20") Integer nameUpdaterQueueCapacity, - @DefaultValue("10") Integer idUpdaterStageConcurrency, - @DefaultValue("15") Integer idUpdaterQueueCapacity, @DefaultValue("50") Integer mergerQueueCapacity, @DefaultValue("5") Integer errorAttemptCount, @DefaultValue("2s") Duration errorAttemptDelay, 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 8d00b9d..a96126b 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,7 +5,10 @@ import org.springframework.context.annotation.Configuration; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.model.impl.*; +import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload; +import ru.nbch.credit_tracker.model.impl.SignalPayload; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.*; @@ -24,13 +27,9 @@ public class PipelineStarterConfig { PackageRecordsService packageRecordsService, PhonesMapMapper phonesMapMapper) { return (setupParam) -> switch (setupParam.mode()) { - case DEFAULT, RESEND -> + case DEFAULT, RESEND, RESTORE -> new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - case RESTORE -> - new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - case REPEATED_RESTORE -> - new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - case MAKE_XML_BY_REPEATED -> + case REPEATED_RESTORE, MAKE_XML_BY_REPEATED -> new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); case RECALC_PERSONAL_DATA -> new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings); @@ -38,8 +37,8 @@ public class PipelineStarterConfig { } @Bean - public Supplier> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) { - return () -> new XmlMergeCoordinator(creditTrackerSettings); + public Supplier> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) { + return () -> new SetupMonitoringPipelineTerminator(creditTrackerSettings); } @Bean diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java index cadf8ae..b8844f6 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -3,7 +3,6 @@ package ru.nbch.credit_tracker.config.task; 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.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.model.impl.VoidPayload; @@ -20,23 +19,13 @@ public class StageConfig { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity()); } - @Bean("idEnricherQueueFactory") - public StageQueueFactory idEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { - return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); - } - - @Bean("idUpdaterQueueFactory") - public StageQueueFactory idUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { - return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity()); - } - @Bean("noOpQueueFactory") public StageQueueFactory noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); } @Bean("partFileQueueFactory") - public StageQueueFactory partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) { + public StageQueueFactory partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); } diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/PhonesMap.java b/src/main/java/ru/nbch/credit_tracker/entities/app/PhonesMap.java index 0d77123..1d21288 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/app/PhonesMap.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/PhonesMap.java @@ -1,12 +1,8 @@ package ru.nbch.credit_tracker.entities.app; -import java.time.LocalDate; +import lombok.*; + import java.time.LocalDateTime; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Getter; -import lombok.NoArgsConstructor; -import lombok.Setter; @Getter @Setter @@ -17,12 +13,6 @@ public class PhonesMap { private Long id; private Integer fid; private Long phone; - private String name1; - private String first; - private String middle; - private LocalDate birthDt; - private String serNum; - private String idType; private Integer flagDebt5000; private Integer flag90days; private LocalDateTime searchedAt; diff --git a/src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java b/src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java index 3237baa..d68dbdc 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java @@ -1,11 +1,8 @@ package ru.nbch.credit_tracker.entities.indic; +import lombok.*; + import java.time.LocalDate; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Getter; -import lombok.NoArgsConstructor; -import lombok.Setter; @Getter @Setter @@ -17,9 +14,4 @@ public class Name { private String number; private LocalDate fileSinceDate; private LocalDate fileSinceDatePn; - private String first; - private String last; - private String middle; - private LocalDate birthDate; - } diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index bbd26dc..aeea8f9 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -29,7 +29,6 @@ import ru.nbch.credit_tracker.model.MonitoringValidationError; import ru.nbch.credit_tracker.model.ValidationResult; 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.LpPipelinePackageFabric; import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; @@ -67,7 +66,6 @@ public class SignalFacade { private final AuthorizationService authorizationService; private final NpPipelinePackageFabric npPipelinePackageFabric; private final LpPipelinePackageFabric lpPipelinePackageFabric; - private final SignalRestService signalRestService; private final Saver saver; private final PipelinePackageOrchestrator pipelineOrchestrator; private final Pattern DOWNLOAD_FILE_PATTERN = Pattern.compile("^(?[A-Za-z0-9]{6})_(?[A-Za-z0-9_]+)_(?\\d{4}\\d{2}\\d{2}_\\d{2}\\d{2}\\d{2})(?:\\.xml)?\\.zip$"); @@ -78,7 +76,6 @@ public class SignalFacade { AuthorizationService authorizationService, NpPipelinePackageFabric npPipelinePackageManager, LpPipelinePackageFabric lpPipelinePackageFabric, - SignalRestService signalRestService, Saver saver, PipelinePackageOrchestrator pipelineOrchestrator) { this.monitoringRequestValidator = monitoringRequestValidator; this.config = config; @@ -86,7 +83,6 @@ public class SignalFacade { this.authorizationService = authorizationService; this.npPipelinePackageFabric = npPipelinePackageManager; this.lpPipelinePackageFabric = lpPipelinePackageFabric; - this.signalRestService = signalRestService; this.saver = saver; this.pipelineOrchestrator = pipelineOrchestrator; } @@ -252,13 +248,6 @@ public class SignalFacade { throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName); } - if (status == Status.ON_MONITORING) { - boolean successfullyDeleted = signalRestService.deletePackage(username, packageName); - if (!successfullyDeleted) { - throw clientExcp(SignalError.ERROR_017, XmlErrorFormat.MonitoringPackageDeleteError, packageName); - } - } - LocalDateTime deactivatedAt = LocalDateTime.now(); userPackageService.deletePackageById(packageToDelete.getId(), deactivatedAt); MonitoringPackageDeleteResponse response = new MonitoringPackageDeleteResponse(); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java index a4389f8..e90c225 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java @@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.mapper.indic; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; -import ru.nbch.credit_tracker.entities.indic.Id; import ru.nbch.credit_tracker.entities.indic.Name; import java.util.List; @@ -14,5 +13,4 @@ public interface PersonMapper { List findNameDataByFids(@Param("fids") List fids); - List findPassportData(@Param("fids") List fids); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java index d8cae7e..76959a4 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java @@ -10,7 +10,6 @@ import java.util.List; @Mapper public interface PackagesRecordsMapper { - void setErrorJson(@Param("json") String json); List registerJson(@Param("json") String json); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index 29da604..28fd710 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -24,8 +24,6 @@ public interface PhonesMapMapper { List findEligibleForFlagsInsert(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); - void updateIdCachedData(@Param("json") String json); - void updateNameCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json); diff --git a/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java b/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java index ea74f2f..8a2a04d 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java +++ b/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java @@ -12,7 +12,5 @@ public class SubjectProfile { private Long mapId; private Long phone; private Integer fid; - private NameData nameData; - private IdData idData; private FlagData flagData; } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalPayload.java new file mode 100644 index 0000000..c34f7ad --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalPayload.java @@ -0,0 +1,6 @@ +package ru.nbch.credit_tracker.model.impl; + +import ru.nbch.pipeline.engine.core.TerminatePayload; + +public record SignalPayload(boolean hasFids) implements TerminatePayload { +} 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 deleted file mode 100644 index a69591f..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java +++ /dev/null @@ -1,8 +0,0 @@ -package ru.nbch.credit_tracker.model.impl; - -import ru.nbch.pipeline.engine.core.TerminatePayload; - -import java.nio.file.Path; - -public record SignalXmlFilePayload(Path signalXmlFilePath) implements TerminatePayload { -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java deleted file mode 100644 index 3a52f28..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java +++ /dev/null @@ -1,107 +0,0 @@ -package ru.nbch.credit_tracker.service; - -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.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; -import ru.nbch.credit_tracker.entities.app.UserPackage; -import ru.nbch.credit_tracker.entities.external_response.monitoring_report_error.MonitoringReportErrors; -import ru.nbch.credit_tracker.enums.RejectedStatus; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.refactoring.Suspicios; -import ru.nbch.credit_tracker.service.indic.PhoneService; -import ru.nbch.credit_tracker.service.signal.SignalRestService; -import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.CtFileUtils; - -import java.io.BufferedInputStream; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.List; -import java.util.Optional; - -@Slf4j -@Service -@EnableAspectJAutoProxy(exposeProxy = true) -public class MonitoringErrorProcesingService { - - private MonitoringErrorProcesingService self; - private final UserPackageMapper userPackageMapper; - private final SignalRestService signalRestService; - private final StaxXmlProcessor staxXmlProcessor; - private final PhoneService phoneService; - private final Integer batchSize; - - public MonitoringErrorProcesingService(UserPackageMapper userPackageMapper, - SignalRestService signalRestService, - StaxXmlProcessor staxXmlProcessor, - PhoneService phoneService, - CreditTrackerSettings config) { - this.userPackageMapper = userPackageMapper; - this.signalRestService = signalRestService; - this.staxXmlProcessor = staxXmlProcessor; - this.phoneService = phoneService; - this.batchSize = config.getPipeline().subjectBatchSize(); - } - - public void findErrors(Long packageId) { - UserPackage userPackage = userPackageMapper.findPackage(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 { - updateSubjectsErrors(filePath.get(), packageId); - } catch (Exception e) { - log.error("Ошибка при записи сообщений об ошибках в сформированных субъектах для мониторинга. Package_id = {}", packageId, e); - } - CtFileUtils.deleteQuietly(filePath.get()); - } - userPackageMapper.setRejectedStatus(packageId, RejectedStatus.ERROR.getKey()); - } - - - private void updateSubjectsErrors(Path filePath, Long packageId) { - log.info("Start updating subjects errors. PackageId: {}", packageId); - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(filePath), 8 * 1024 * 1024)) { - staxXmlProcessor.read(in, MonitoringReportErrors.class, - request -> { - if (!request.getPersons().isEmpty()) { - List subjects = request.getPersons().stream().map(p -> { - log.warn("Ошибка при формировании субъекта. package_id = {}, id = {}. Описание: {}", packageId, p.getUid(), p.getValue()); - return PackagesRecordsMap.builder() - .id(p.getUid()) - .errorText(p.getValue()) - .build(); - }).toList(); - phoneService.updateOnError(subjects); - log.trace("Updated subject's batch. Total count: {}. PackageId: {}", request.getPersons().size(), packageId); - request.getPersons().clear(); - } - }, - request -> request.getPersons().isEmpty(), - batchSize, "/Errors/p"); - } catch (IOException e) { - throw new RuntimeException("Error while parsing MonitoringReportErrors", e); - } - log.info("Subject's errors were updated. PackageId: {}", packageId); - } - - @Suspicios("Нужен для аоп? А где используется") - private MonitoringErrorProcesingService getCurrentProxy() { - if (self == null) { - self = (MonitoringErrorProcesingService) AopContext.currentProxy(); - } - return self; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java index 30de6be..5fe9e31 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java @@ -1,7 +1,6 @@ package ru.nbch.credit_tracker.service.indic; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.indic.Id; import ru.nbch.credit_tracker.entities.indic.Name; @@ -9,9 +8,7 @@ import ru.nbch.credit_tracker.mapper.indic.PersonMapper; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.utils.CTUtil; -import java.time.LocalDate; import java.util.Comparator; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.function.BinaryOperator; @@ -36,14 +33,6 @@ public class PersonService { this.mapper = mapper; } - public Map filledPassportsData(List fids) { - if (fids.isEmpty()) { - return new HashMap<>(); - } - List passports = mapper.findPassportData(fids); - return groupPassportsData(passports); - } - public Map groupPassportsData(List passports) { return passports.stream() .filter(this::isDUL) @@ -75,38 +64,6 @@ public class PersonService { List names = mapper.findNameData(phones); return names.stream() - .filter(name -> - StringUtils.isNotEmpty(name.getFirst()) && - StringUtils.isNotEmpty(name.getLast()) && - name.getBirthDate() != null && !LocalDate.of(1, 1, 1).equals(name.getBirthDate()) - ) - .collect(Collectors.toMap( - Name::getNumber, - Function.identity(), - BinaryOperator.maxBy(byDateThenIdNullSafe) - )); - } - - /** - * Поиск данных по таблицам PHONE_NORM и NAME по списку номеров телефонов - * - * @param subjects - * @return - отображение NAME к номеру телефону, отфильтрованному по логике - * FID с максимальным значением FILE_SINCE_DT. Если FILE_SINCE_DT оказался одинаковый, берется максимальный FID - */ - public Map findNameByFids(List subjects) { - List fids = subjects.stream() - .map(SubjectProfile::getFid) - .toList(); - - List names = mapper.findNameDataByFids(fids); - - return names.stream() - .filter(name -> - StringUtils.isNotEmpty(name.getFirst()) && - StringUtils.isNotEmpty(name.getLast()) && - name.getBirthDate() != null && !LocalDate.of(1, 1, 1).equals(name.getBirthDate()) - ) .collect(Collectors.toMap( Name::getNumber, Function.identity(), diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java index 57cc2b8..89c86b3 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java @@ -39,17 +39,4 @@ public class PhoneService { } } - 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); - } - } - } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java index 009ba14..fd4ebf5 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java @@ -8,7 +8,6 @@ import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService; import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.SetupParam; @@ -35,7 +34,6 @@ public class StartupRestoreService { private final NpPipelinePackageFabric npPipelinePackageManager; private final PipelinePackageOrchestrator pipelineOrchestrator; private final CreditTrackerSettings creditTrackerSettings; - private final MonitoringErrorProcesingService monitoringErrorProcesingService; private final Saver saver; public StartupRestoreService(UserPackageMapper userPackageMapper, @@ -43,30 +41,20 @@ public class StartupRestoreService { NpPipelinePackageFabric npPipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator, CreditTrackerSettings creditTrackerSettings, - MonitoringErrorProcesingService monitoringErrorProcesingService, Saver saver) { this.userPackageMapper = userPackageMapper; this.phoneFidMapMapper = phoneFidMapMapper; this.npPipelinePackageManager = npPipelinePackageManager; this.pipelineOrchestrator = pipelineOrchestrator; this.creditTrackerSettings = creditTrackerSettings; - this.monitoringErrorProcesingService = monitoringErrorProcesingService; this.saver = saver; } public void run() { - onMonitoringRestore(); acceptedRestore(); processingRestore(); } - private void onMonitoringRestore() { - List packages = userPackageMapper.findOnMonitoringPackages(); - packages.forEach(userPackage -> { - monitoringErrorProcesingService.findErrors(userPackage.getId()); - }); - } - private void acceptedRestore() { List packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name()); log.debug("Found {} packages in status ACCEPTED", packages.size()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java similarity index 85% rename from src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java rename to src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java index 97d88d2..0bcbf9b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java @@ -21,19 +21,19 @@ import java.util.concurrent.CompletableFuture; @Slf4j @Service @EnableScheduling -public class UpdatingFidAndPdService { +public class UpdatingFidService { private final NpPipelinePackageFabric npPipelinePackageFabric; private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; private final PipelinePackageOrchestrator pipelinePackageOrchestrator; private final UserPackageMapper userPackageMapper; private final PhonesMapService phonesMapService; - public UpdatingFidAndPdService(CreditTrackerSettings config, - NpPipelinePackageFabric npPipelinePackageFabric, - PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, - PipelinePackageOrchestrator pipelinePackageOrchestrator, - UserPackageMapper userPackageMapper, - PhonesMapService phonesMapService) { + public UpdatingFidService(CreditTrackerSettings config, + NpPipelinePackageFabric npPipelinePackageFabric, + PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, + PipelinePackageOrchestrator pipelinePackageOrchestrator, + UserPackageMapper userPackageMapper, + PhonesMapService phonesMapService) { this.npPipelinePackageFabric = npPipelinePackageFabric; this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; this.pipelinePackageOrchestrator = pipelinePackageOrchestrator; @@ -42,7 +42,7 @@ public class UpdatingFidAndPdService { } @Scheduled(cron = "${ct.update-fid-cron-expression}") - public void update() { + public void updateNp() { if (!pipelineRecalcPDOrchestrator.checkPermits()) { log.info("The previous task has not yet been completed."); return; @@ -81,4 +81,9 @@ public class UpdatingFidAndPdService { log.error("Failed to build pipeline", e); } } + + @Scheduled(cron = "${ct.update-fid-cron-expression}") + public void updateLp() { + // TODO + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java deleted file mode 100644 index 5611300..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java +++ /dev/null @@ -1,54 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.migration; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.ComponentScan; -import org.springframework.context.annotation.FilterType; -import ru.nbch.credit_tracker.component.MonitoringRequestValidator; -import ru.nbch.credit_tracker.config.datasource.ScoringDataSourceConfig; -import ru.nbch.credit_tracker.config.rest.RestConfig; -import ru.nbch.credit_tracker.config.xml.XmlConfig; -import ru.nbch.credit_tracker.controller.SignalController; -import ru.nbch.credit_tracker.facade.SignalFacade; -import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService; -import ru.nbch.credit_tracker.service.mode.common.CTMainApplication; -import ru.nbch.credit_tracker.service.mode.common.StartupRestoreService; -import ru.nbch.credit_tracker.service.mode.common.UpdatingFidAndPdService; -import ru.nbch.credit_tracker.service.mode.common.UpdatingFlagsService; -import ru.nbch.credit_tracker.service.mode.monitoring.CTMonitoringApplication; -import ru.nbch.credit_tracker.service.mode.monitoring.MonitoringService; -import ru.nbch.credit_tracker.service.scoring.AuthorizationService; -import ru.nbch.credit_tracker.service.signal.SignalRestService; -import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; -import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; -import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; -import ru.nbch.credit_tracker.service.task.impl.Saver; - -@SpringBootApplication -@ComponentScan( - basePackages = {"ru.nbch.credit_tracker"}, - excludeFilters = { - @ComponentScan.Filter( - type = FilterType.ASSIGNABLE_TYPE, - classes = {CTMonitoringApplication.class, CTMainApplication.class, - ScoringDataSourceConfig.class,StartupRestoreService.class, UpdatingFidAndPdService.class, - UpdatingFlagsService.class, MonitoringService.class, AuthorizationService.class, - SignalFacade.class, SignalController.class, FlagsEnricher.class, NpPipelinePackageFabric.class, - RestConfig.class, SignalRestService.class, MonitoringErrorProcesingService.class, Saver.class, - PipelinePackageOrchestrator.class, XmlConfig.class, MonitoringRequestValidator.class - } - ) - } -) -public class CTMigrationApplication { - public static void main(String[] args) { - SpringApplication application = new SpringApplication(CTMigrationApplication.class); - application.setWebApplicationType(WebApplicationType.NONE); - ConfigurableApplicationContext ctx = application.run(args); - MigrationService migrationService = ctx.getBean(MigrationService.class); - migrationService.update(); - } - -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java deleted file mode 100644 index d4d59ba..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java +++ /dev/null @@ -1,62 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.migration; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.boot.SpringApplication; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.service.task.PipelineMigrationPackageFabric; -import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator; -import ru.nbch.credit_tracker.service.task.SetupParam; -import ru.nbch.credit_tracker.service.task.impl.PipelineMode; -import ru.nbch.pipeline.engine.core.Pipeline; - -import java.util.concurrent.CompletableFuture; - -@Slf4j -@Service -@EnableScheduling -public class MigrationService { - private final PipelineMigrationPackageFabric pipelinePackageManager; - private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; - private final ConfigurableApplicationContext context; - - public MigrationService(PipelineMigrationPackageFabric pipelinePackageManager, - PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, - ConfigurableApplicationContext context) { - this.pipelinePackageManager = pipelinePackageManager; - this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; - this.context = context; - } - - public void update() { - log.info("Start updating fids and PD"); - try { - Pipeline pipeline = pipelinePackageManager.buildRecalculationPDMigration(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); - CompletableFuture future = pipelineRecalcPDOrchestrator.startOrReject(pipeline); - future.whenComplete((o, throwable) -> { - try { - if (throwable != null) { - log.error(throwable.getMessage(), throwable); - } - } finally { - log.info("pipeline finished"); - pipeline.close(); - pipelineRecalcPDOrchestrator.release(); - new Thread(this::shutdown, "migration-app-exit").start(); - } - }); - } catch (Exception e) { - log.error("Failed to build pipeline", e); - } - } - - private void shutdown() { - int exitCode = SpringApplication.exit(context, () -> { - // no errors - return 0; - }); - System.exit(exitCode); - } - -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java deleted file mode 100644 index 7e55e61..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java +++ /dev/null @@ -1,424 +0,0 @@ -package ru.nbch.credit_tracker.service.mode.monitoring; - -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import org.slf4j.MDC; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.app.Hit; -import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; -import ru.nbch.credit_tracker.entities.app.PhonesMap; -import ru.nbch.credit_tracker.entities.app.UserPackage; -import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage; -import ru.nbch.credit_tracker.entities.external_response.online_download.Inquiry; -import ru.nbch.credit_tracker.entities.external_response.online_download.Person; -import ru.nbch.credit_tracker.entities.external_response.online_download.Persons; -import ru.nbch.credit_tracker.entities.external_response.online_report.Report; -import ru.nbch.credit_tracker.entities.external_response.online_report.SgnlOnlineReport; -import ru.nbch.credit_tracker.entities.user_response.monitoring_signal_response.MonitoringSignalResponse; -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.UserPackageMapper; -import ru.nbch.credit_tracker.model.UserPackagePhoneData; -import ru.nbch.credit_tracker.service.nbch.SourceService; -import ru.nbch.credit_tracker.service.signal.HitsService; -import ru.nbch.credit_tracker.service.signal.PackageRecordsService; -import ru.nbch.credit_tracker.service.signal.SignalRestService; -import ru.nbch.credit_tracker.service.xml.stax.MonitoringSignalResponseStaxXmlWriter; -import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.CTUtil; -import ru.nbch.credit_tracker.utils.CtFileUtils; -import ru.nbch.credit_tracker.utils.ZipUtils; -import ru.nbch.credit_tracker.utils.error.ExceptionUtils; - -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.*; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Collectors; - -/** - * Сервис периодического опроса официального сервиса сигналов на наличие новых отчетов по пакетам находящимся в статусе ON_MONITORING - */ -@Slf4j -@Service -public class MonitoringService { - - private final SignalRestService signalRestService; - private final UserPackageMapper userPackageMapper; - private final StaxXmlProcessor staxXmlProcessor; - private final HitsService hitsService; - private final PackageRecordsService packageRecordsService; - private final SourceService sourceService; - private final CreditTrackerSettings config; - - public MonitoringService(SignalRestService signalRestService, - UserPackageMapper userPackageMapper, - StaxXmlProcessor staxXmlProcessor, - HitsService hitsService, - PackageRecordsService packageRecordsService, SourceService sourceService, - CreditTrackerSettings config) { - this.signalRestService = signalRestService; - this.userPackageMapper = userPackageMapper; - this.staxXmlProcessor = staxXmlProcessor; - this.hitsService = hitsService; - this.packageRecordsService = packageRecordsService; - this.sourceService = sourceService; - this.config = config; - } - - public void run() { - List messages = new ArrayList<>(); - - for (int i = 1000; i < 2000; i++) { - PackageMessage messageData = new PackageMessage(); - messageData.setFid(i); - messageData.setMemberCode("ZX01LL"); - messages.add(messageData); - } - - - long start = System.currentTimeMillis(); - log.info("Start selecting by one"); - monitorByOne(messages); - log.info("Finish selecting by one. Time: {} ms", System.currentTimeMillis() - start); - - start = System.currentTimeMillis(); - log.info("Start selecting by batch"); - monitorbyBatch(messages); - log.info("Finish selecting by bacth. Time: {} ms", System.currentTimeMillis() - start); - } - - private void monitorByOne(List messages) { - for (PackageMessage message : messages) { - Integer fid = message.getFid(); - List phonesMaps = packageRecordsService.findCachedByFid(fid); - for (PhonesMap phonesMap : phonesMaps) { - if (phonesMap != null) { - Long clientId = phonesMap.getPhone(); - Integer flagDebt5000 = phonesMap.getFlagDebt5000(); - Integer flag90days = phonesMap.getFlag90days(); - List packageRecords = packageRecordsService.findByPhone(clientId); - for (PackagesRecordsMap packageRecord : packageRecords) { - Long packageId = packageRecord.getPackageId(); - String subjectId = packageRecord.getSubjectId(); - UserPackage userPackage = userPackageMapper.findPackage(packageId); - String status = userPackage.getStatus(); - String signalTypes = userPackage.getSignalTypes(); - - String busCategory = sourceService.findBusCategory(message.getMemberCode()); - // проверить статус ON_MPNITORING + разрешенные signalTypes - } - } - } - } - } - - private void monitorbyBatch(List messages) { - List fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList()); - List memberCodes = messages.stream() - .map(message -> message.getMemberCode().substring(0,6)) - .toList(); - - List phonesMaps = packageRecordsService.findCachedByFids(fids); - - // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров) - Map> phonesDataMap = phonesMaps.stream() - .collect(Collectors.groupingBy(PhonesMap::getFid)); - - Set uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet()); - - // прелоад по memberCode категорий - Map busCategories = sourceService.findBusCategories(memberCodes); - - // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING - List packagePhoneData = packageRecordsService - .selectUserPackagePhoneData(uniquePhones); - - // 1 номер может быть у нескольких пакетов - Map> userPackagePhoneDataMap = packagePhoneData.stream() - .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone)); - - for (PackageMessage message : messages) { - Integer fid = message.getFid(); - String memberCode = message.getMemberCode(); - - String busCategory = busCategories.get(memberCode.substring(0, 6)); - List phonesMapList = phonesDataMap.get(fid); - - if (phonesMapList == null) { - continue; - } - - for (PhonesMap phonesMap : phonesMapList) { - // флаги просрочек - Integer flagDebt5000 = phonesMap.getFlagDebt5000(); - Integer flag90days = phonesMap.getFlag90days(); - - // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются - List userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone()); - - if (userPackagePhoneDataList == null) { - continue; - } - - for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) { - String subjectId = userPackagePhoneData.getSubjectId(); - String signalTypes = userPackagePhoneData.getSignalTypes(); - - // проверить разрешенные signalTypes - } - } - } - } - - private void monitor() { - 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)); - } - } - - private void processPackage(UserPackage userPackage) { - log.info("Processing package"); -// if (!phoneFidMapMapper.hasNonNullFids(userPackage.getId())) { -// log.info("Package has no fids. Package will be skipped."); -// return; -// } - SgnlOnlineReport onlineReports = signalRestService.getReports(userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName()); - - if (onlineReports == null) { - log.info("No reports found for package"); - return; - } - - // Формируем список имен файлов, по которым сработали сигналы. Сюда входят только те имена, у которых парсится дата формирования отчета - 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"); - return; - } - - LocalDateTime lastDownloadedReportDate = CTUtil.parseDateFromReportFileName(userPackage.getLastDownloadedReport()); - - if (lastDownloadedReportDate != null) { - // Отбираем только те отчеты, дата которых позже даты последнего загруженного отчета - fileNames = fileNames.stream() - .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName).isAfter(lastDownloadedReportDate)) // отфильтровали только те имена, которые парсятся в дату, так что null тут не придет - .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате - .toList(); - } else { - fileNames = fileNames.stream() - .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате - .toList(); - } - - if (StringUtils.isNotBlank(userPackage.getLastDownloadedReport())) { - log.info("Last downloaded report found for package. Last Downloaded Report Name: {}", userPackage.getLastDownloadedReport()); - } else { - log.info("Last Downloaded Report Name is empty"); - } - - for (String newReport : fileNames) { - try { - Optional loadedReportPath = Optional.empty(); - if (StringUtils.isBlank(userPackage.getLastDownloadedReport())) { - // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным - loadedReportPath = signalRestService.downloadReport(userPackage, newReport); - } else if (lastDownloadedReportDate != null) { - LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport); - if (newReportDate.isAfter(lastDownloadedReportDate)) { - // полученный отчет новее записанного в базу - loadedReportPath = signalRestService.downloadReport(userPackage, newReport); - } - } - if (loadedReportPath.isEmpty()) { - log.info("Report was not loaded for package"); - continue; - } - - Path responsePath = generateResponse(userPackage, newReport, loadedReportPath.get()); - if (responsePath != null) { - Path zipFile = ZipUtils.zipFile(responsePath); - log.info("New report file by path: {}", zipFile); - updateSignalHits(userPackage, responsePath, zipFile); - } else { - log.info("Report was not created for package"); - } - userPackageMapper.updatePackageLastDownloadedReport(userPackage.getId(), newReport); - } catch (Exception e) { - log.error("Report was not created for package: {}", ExceptionUtils.getStackTrace(e)); - } - } - } - - /** - * Формирует результирующий xml файл для конечного пользователя на основе файла из официального сервиса сигналов - * - * @return Путь к временному рещультирующему файлу - */ - private Path generateResponse(UserPackage userPackage, String reportName, Path loadedReportPath) throws IOException { - // Выставляется в true, если хотя бы один субъект с мониторинга подошел под условие срабатывания сигнала. В этом случае отчет сохраняется - AtomicBoolean generateResponse = new AtomicBoolean(false); - - String commonMembercode = userPackage.getUserId().substring(0, 6); - Path rootReportPath = Path.of(config.getLoadDir(), commonMembercode); - if (!Files.exists(rootReportPath)) { - Files.createDirectories(rootReportPath); - } - LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(reportName); - newReportDate = newReportDate == null ? LocalDateTime.now() : newReportDate; - AtomicReference responsePath = new AtomicReference<>( - rootReportPath.resolve( - String.format("%s_%s_%s.xml", - commonMembercode, - userPackage.getPackageName(), - newReportDate.format(CTUtil.responseReportDateTimeFormatter)) - )); - - MonitoringSignalResponse response = new MonitoringSignalResponse(); - //Имя пакета, переданного при установке - response.setPackageName(userPackage.getPackageName()); - - // Запись в результрующий файл - MonitoringSignalResponseStaxXmlWriter staxXmlWriter = new MonitoringSignalResponseStaxXmlWriter(config.getEncoding()); - 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(), userPackage); - 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); - } - } - }); - - CtFileUtils.deleteQuietly(loadedReportPath); - return responsePath.get(); - } - - private List processPersons(List persons, UserPackage userPackage) { - List subjects = new ArrayList<>(); - Map subjectRepInfoMap = new HashMap<>(); - for (Person person : persons) { - Signals signals = new Signals(); - for (Inquiry inquiry : person.getInquiries()) { - // Отбираем только тех фл, у которых bus_category in (MFO, MKK, MFK) и есть сигналы с кодом 16 - Integer signalType = CTUtil.checkBusCategory(inquiry.getBusCategory()); - if (signalType == -1 || inquiry.getSignals().stream().noneMatch(signal -> signal.getN().equals(16)) || - !userPackage.getSignalTypes().contains(signalType.toString())) { - continue; - } - Signal signal = new Signal(); - signal.setSignalType(signalType); // Код типа сигнала по внутреннему справочнику. В текущей версии: 1 = сигнал 16 от MFO/MKK/MFK - signal.setInquiryDate(inquiry.getDate()); // Дата и время запроса КИ - signal.setInquiryPurpose(inquiry.getInqP()); // Цель запроса КИ - signals.getSignals().add(signal); - } - - // Если есть хотя бы один заполненный сигнал, значит заполняем остальные данные - if (!signals.getSignals().isEmpty()) { - PackagesRecordsMap packagesRecord; - PhonesMap cacheByPhone; - if (subjectRepInfoMap.containsKey(person.getUid())) { - SubjectRepInfo subjectRepInfo = subjectRepInfoMap.get(person.getUid()); - packagesRecord = subjectRepInfo.packagesRecord; - cacheByPhone = subjectRepInfo.cacheByPhone; - } else { - packagesRecord = packageRecordsService.findPackageRecordsById(person.getUid()); - cacheByPhone = packageRecordsService.findCachedByPhone(packagesRecord.getPhone_l()); - subjectRepInfoMap.put(person.getUid(), new SubjectRepInfo(packagesRecord, cacheByPhone)); - } - if (cacheByPhone != null) { - Subject subject = new Subject(); - subject.setId(packagesRecord.getSubjectId()); // Уникальный идентификатор субъекта (равен Id из XML-пакета при установке мониторинга) - subject.setClientId(CTUtil.normalizePhoneNumberZeroesOnly(packagesRecord.getPhone_l())); - subject.setFlagDebt5000(cacheByPhone.getFlagDebt5000()); - subject.setFlag9012(cacheByPhone.getFlag90days()); - subject.setSignals(signals); - subjects.add(subject); - } - } - } - return subjects; - } - - private void updateSignalHits(UserPackage userPackage, Path responsePath, Path reportFileName) { - log.info("Updating signal hits for {}", userPackage.getPackageName()); - LocalDateTime responseSentAt = LocalDateTime.now(); - // Считываем результирующий файл пачками и обновляем таблицу hits - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(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 -> { - if (userPackage.getSignalTypes().contains(signal.getSignalType().toString())) { - Hit hit = new Hit(); - hit.setPackageId(userPackage.getId()); - hit.setSubjectId(subject.getId()); - hit.setPhone(Long.valueOf(subject.getClientId())); - hit.setSignalType(signal.getSignalType()); - hit.setResponseSentAt(responseSentAt); - hit.setReportFileName(reportFileName.getFileName().toString()); - hits.add(hit); - } - }); - if (!hits.isEmpty()) { - hitsService.insertHits(hits); - hits.clear(); - } - }); - response.getSubjects().getSubjects().clear(); // чистим список для экономии памяти - }, - response -> response.getSubjects().getSubjects().isEmpty() - , - 10000, - "/MonitoringSignalResponse/Subjects/Subject"); - } catch (IOException e) { - throw new RuntimeException("Error while updating hits", e); - } - CtFileUtils.deleteQuietly(responsePath); // удалям результирующий xml файл с диска (Он уже заархивирован к этому моменту) - log.info("Signal hits updated for package {}", userPackage.getPackageName()); - } - - private class SubjectRepInfo { - private final PackagesRecordsMap packagesRecord; - private final PhonesMap cacheByPhone; - - private SubjectRepInfo(PackagesRecordsMap packagesRecord, PhonesMap cacheByPhone) { - this.packagesRecord = packagesRecord; - this.cacheByPhone = cacheByPhone; - } - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java index ca4c436..e4cf0c3 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java @@ -11,7 +11,10 @@ import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; -import ru.nbch.credit_tracker.model.*; +import ru.nbch.credit_tracker.model.FlagData; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.UserPackageLegalData; +import ru.nbch.credit_tracker.model.UserPackagePhoneData; import java.util.*; import java.util.function.Function; @@ -61,10 +64,6 @@ public class PackageRecordsService { .map(phoneMap -> new HashMap() {{ put("fid", phoneMap.getFid()); put("phone", phoneMap.getPhone()); - put("name_1", phoneMap.getName1()); - put("first", phoneMap.getFirst()); - put("middle", phoneMap.getMiddle()); - put("birth_dt", phoneMap.getBirthDt()); }}).toList()); phonesMapMapper.registerJson(json); } catch (JsonProcessingException e) { @@ -98,21 +97,6 @@ public class PackageRecordsService { } } - public void updateIdCache(List idData) { - try { - String json = entityToJsonMapper.writeValueAsString(idData.stream() - .filter(Objects::nonNull) - .map(phoneMap -> new HashMap() {{ - put("phone", phoneMap.getPhone()); - put("ser_num", phoneMap.getSerNum()); - put("id_type", phoneMap.getIdType()); - }}).toList()); - phonesMapMapper.updateIdCachedData(json); - } catch (JsonProcessingException e) { - throw new RuntimeException(e); - } - } - public void updateNpFlagCache(List flagData) { try { String json = entityToJsonMapper.writeValueAsString(flagData.stream() diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java index 8f80fa7..b90b1a4 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java @@ -2,13 +2,14 @@ package ru.nbch.credit_tracker.service.signal; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import java.util.HashMap; -import java.util.List; -import java.util.Objects; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.model.SubjectProfile; +import java.util.HashMap; +import java.util.List; +import java.util.Objects; + @Service public class PhonesMapService { private final PhonesMapMapper phonesMapMapper; @@ -20,7 +21,7 @@ public class PhonesMapService { this.entityToJsonMapper = entityToJsonMapper; } - public void resetFidAndPDByPhones(List phones) { + public void resetFidAndFlagsByPhones(List phones) { try { String json = entityToJsonMapper.writeValueAsString(phones.stream() .filter(Objects::nonNull) @@ -46,16 +47,12 @@ public class PhonesMapService { } } - public void updateNameCachedData(List subjectProfiles) { + public void updateFidPhoneCachedData(List subjectProfiles) { try { String json = entityToJsonMapper.writeValueAsString(subjectProfiles.stream() .filter(Objects::nonNull) .map(subjectProfile -> new HashMap() {{ put("fid", subjectProfile.getFid()); - put("name_1", subjectProfile.getNameData().getLastName()); - put("first", subjectProfile.getNameData().getFirstName()); - put("middle", subjectProfile.getNameData().getMiddleName()); - put("birth_dt", subjectProfile.getNameData().getBirthDate()); put("phone", subjectProfile.getPhone()); }}).toList()); phonesMapMapper.updateNameCachedData(json); 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 deleted file mode 100644 index d48fa76..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java +++ /dev/null @@ -1,336 +0,0 @@ -package ru.nbch.credit_tracker.service.signal; - -import jakarta.xml.bind.DatatypeConverter; -import java.io.IOException; -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 lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.core.io.FileUrlResource; -import org.springframework.core.io.Resource; -import org.springframework.http.HttpEntity; -import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpMethod; -import org.springframework.http.HttpStatusCode; -import org.springframework.http.MediaType; -import org.springframework.http.ResponseEntity; -import org.springframework.stereotype.Service; -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; -import ru.nbch.credit_tracker.entities.external_response.online_report.SgnlOnlineReport; -import ru.nbch.credit_tracker.service.xml.jaxb.IXmlProcessor; -import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.ZipUtils; - -@Slf4j -@Service -public class SignalRestService { - - private static String UPDATE_URI = "update"; - private static String DELETE_URI = "delete"; - private static String GET_REPORTS_URI = "onlinereport"; - private static String GET_REPORT_URI = "report"; - private static String DOWNLOAD_REPORT_URI = "download"; - private static String DOWNLOAD_ONLINE_REPORT_URI = "onlinedownload"; - - private final RestTemplate restTemplate; - private final IXmlProcessor xmlProcessor; - private final StaxXmlProcessor staxProcessor; - 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, - CreditTrackerSettings config) throws Exception { - this.restTemplate = restTemplate; - this.xmlProcessor = xmlProcessor; - this.staxProcessor = staxProcessor; - this.config = config; - this.authToken = buildToken(config.getSignalsToken(), config.getSignalsuser()); - } - - public boolean setupMonitoring(Path signalFilePath) throws Exception { - Path gzip = ZipUtils.packGzip(signalFilePath); - - MultiValueMap body = new LinkedMultiValueMap<>(); - body.add("file", new FileUrlResource(gzip.toString())); - - HttpHeaders headers = new HttpHeaders(); - headers.setContentType(MediaType.MULTIPART_FORM_DATA); - headers.setBasicAuth(authToken); - - 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 - return true; - } else if (response.getStatusCode().value() == 200 && response.getBody() != null) { - SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class); - if (error.getError().getDescription() != null) { - log.error("Ошибка при постановке пакета на мониторинг в официальный сервис сигналов: {}", error.getError().getDescription()); - } else { - log.error("неизвестная ошибка при постановке пакета на мониторинг в официальный сервис сигналов: {}", response.getBody()); - } - return false; - } - return false; - } - - /** - * @return имя файла с ошибками по поставленному на мониторинг отчету - */ - public String getErrorFileName(Long packageId, String packCode) { - String url = GET_REPORT_URI + "?packCode=" + packCode; - - HttpHeaders headers = new HttpHeaders(); - headers.setBasicAuth(authToken); - - HttpEntity> requestEntity = new HttpEntity<>(headers); - - ResponseEntity response = restTemplate.exchange( - url, - HttpMethod.GET, - requestEntity, - String.class - ); - - String answer = response.getBody(); - - SgnlReport error = staxProcessor.read(answer, SgnlReport.class); - if (error.getError().getDescription() != null) { - log.error("Ошибка при получении имени файла ошибки из сервиса НБКИ. packCode={}. Описание: {}", packCode, error.getError().getDescription()); - return null; - } - try { - SgnlReportMonitorigSetupError report = xmlProcessor.read(answer, SgnlReportMonitorigSetupError.class); - if (report.getOperationReport().getErrorsForNextReport() != null) { - //todo set reject_status = 0 - String fileName = report.getOperationReport().getErrorsForNextReport().getFilename(); - log.warn("Найдены ошибки по пакету. packCode={}, packageId={}, fileName = {}", packCode, packageId, fileName); - return fileName; - } else { - log.info("Пакет был поставлен на мониторинг без ошибок. packCode={}, packageId={}", packCode, packageId); - } - } catch (Exception e) { - log.error("Ошибка при получении имени файла ошибки из сервиса НБКИ. Код ошибки: {} packCode={}", response.getStatusCode().value(), packCode); - } - - return null; - } - - /** - * Загружает .gz отчет ошибок из официального сервиса сигналов, разархивирует в tmp_folder - * - * @return путь к разархивированному файлу. Пример: /tmp_folder//0001XX_MY_BATCH_223112.1.errors.xml.gz - */ - public Optional loadErrorFile(String fileName, UserPackage userPackage) { - String url = DOWNLOAD_REPORT_URI + "?fNameRequest=" + fileName; - - HttpHeaders headers = new HttpHeaders(); - headers.setBasicAuth(authToken); - - HttpEntity> requestEntity = new HttpEntity<>(headers); - - ResponseEntity response = restTemplate.exchange( - url, - HttpMethod.GET, - requestEntity, - byte[].class - ); - - 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); - } - - /** - * Удаляет пакет с мониторинга - * @return Успешность удаления пакета - */ - public boolean deletePackage(String username, String packageName) { - String packCode = username.substring(0, 6) + "_" + packageName; - String url = DELETE_URI + "?packCode=" + packCode; - HttpHeaders headers = new HttpHeaders(); - headers.setBasicAuth(authToken); - - HttpEntity> requestEntity = new HttpEntity<>(headers); - - ResponseEntity responseEntity = restTemplate.exchange( - url, - HttpMethod.GET, - requestEntity, - String.class - ); - HttpStatusCode statusCode = responseEntity.getStatusCode(); - SgnlReport response = staxProcessor.read(responseEntity.getBody(), SgnlReport.class); - - if (statusCode.value() == 200 && !isError(response)) { //все ок - return true; - } - - log.warn("Ошибка при удалении пакета из сервиса НБКИ. packCode={}. status={}. Описание: {}", - packCode, - statusCode.value(), - errorDescription(response)); - - if (statusCode.value() != 200) { //была фатальная ошибка - return false; - } - - return isRemotePackageStatedDeleted(username, packageName); - } - - private boolean isRemotePackageStatedDeleted(String username, String packageName) { - HttpHeaders headers = new HttpHeaders(); - headers.setBasicAuth(authToken); - HttpEntity> requestEntity = new HttpEntity<>(headers); - String packCode = username.substring(0, 6) + "_" + packageName; - ResponseEntity responseEntity = restTemplate.exchange( - GET_REPORT_URI + "?packCode=" + packCode, - HttpMethod.GET, - requestEntity, - String.class - ); - HttpStatusCode statusCode = responseEntity.getStatusCode(); - SgnlReportDeleted response = staxProcessor.read(responseEntity.getBody(), SgnlReportDeleted.class); - Integer deletedFlag = Optional.ofNullable(response) - .map(SgnlReportDeleted::getDeleteFlag) - .orElse(null); - - if ((deletedFlag == null) || statusCode.value() != 200) { - log.error("couldn't perform report for {}: HTTP code: {}; delete flag {}", - packCode, statusCode.value(), deletedFlag); - return false; - } - if (deletedFlag != 1) { - log.info("{} was not deleted", packCode); - return false; - } - - log.info("{} was deleted", packCode); - return true; - } - - /** - * @return Список отчетов, по которым сработали сигналы из официального сервиса сигналов - */ - public SgnlOnlineReport getReports(String packCode) { - String url = GET_REPORTS_URI + "?packCode=" + packCode; - - HttpHeaders headers = new HttpHeaders(); - headers.setBasicAuth(authToken); - - HttpEntity> requestEntity = new HttpEntity<>(headers); - - ResponseEntity response = restTemplate.exchange( - url, - HttpMethod.GET, - requestEntity, - String.class - ); - - String answer = response.getBody(); - - SgnlReport error = staxProcessor.read(answer, SgnlReport.class); - if (error.getError().getDescription() != null) { - log.error("Ошибка при получении пакета из сервиса НБКИ. packCode={}. Описание: {}", packCode, error.getError().getDescription()); - return null; - } - try { - return xmlProcessor.read(answer, SgnlOnlineReport.class); - } catch (Exception e) { - log.error("Ошибка при получении пакета из сервиса НБКИ. Код ошибки: {} packCode={}", response.getStatusCode().value(), packCode); - return null; - } - } - - /** - * Загружает .gz отчет из официального сервиса сигналов, разархивирует в tmp_folder - * - * @return путь к разархивированному файлу. Пример: /tmp_folder//0001XX_MY_BATCH_204.1.online.2025-07-08T20-00-46.xml.gz.xml - */ - public Optional downloadReport(UserPackage userPackage, String reportName) throws IOException { - String url = DOWNLOAD_ONLINE_REPORT_URI + "?fNameReport=" + reportName; - - HttpHeaders headers = new HttpHeaders(); - headers.setBasicAuth(authToken); - - HttpEntity> requestEntity = new HttpEntity<>(headers); - - ResponseEntity response = restTemplate.exchange( - url, - HttpMethod.GET, - requestEntity, - byte[].class - ); - Path targetFilePathWithExt = null; - if (response.getHeaders().getContentType().equals(MEDIA_TYPE_GZIP)) { - Path folder = Path.of(config.getLoadDir(), - "tmp", userPackage.getUserId(), userPackage.getPackageName()); - Files.createDirectories(folder); - targetFilePathWithExt = folder.resolve(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 buildToken(String signalsToken, String signalsuser) throws Exception { - try { - MessageDigest md = MessageDigest.getInstance("MD5"); - byte[] messageDigest = md.digest((signalsuser + "|" + signalsToken).getBytes(StandardCharsets.UTF_8)); - String hash = DatatypeConverter.printHexBinary(messageDigest).toLowerCase(); - Base64.Encoder encoder = Base64.getEncoder(); - return encoder.encodeToString((signalsuser + ":" + hash).getBytes(StandardCharsets.UTF_8)); - } catch (Exception e) { - log.error("Error building token", e); - throw new Exception("Error building token", e); - } - } - - 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; - private final StageQueueFactory idUpdaterQueueFactory; private final StageQueueFactory nameEnricherQueueFactory; private final StageQueueFactory nameUpdaterQueueFactory; - private final StageQueueFactory sendSignalQueueFactory; + private final StageQueueFactory sendSignalQueueFactory; private final StageQueueFactory noOpQueueFactory; - private final FidIdEnricher fidIdEnricherHandler; - private final FidIdUpdater fidIdUpdater; - private final FidNameEnricher fidNameEnricher; - private final FidNameUpdater fidNameUpdater; - private final FidNameUpdaterMigration fidNameUpdaterMigration; - private final Supplier> pipelineTerminatorFactory; + private final NpFidEnricher npFidEnricher; + private final NpFidUpdater npFidUpdater; + private final Supplier> pipelineTerminatorFactory; private final Supplier> noOpTerminatorFactory; private final Function> pipelineStarterFactory; private final CreditTrackerSettings creditTrackerSettings; private final PipelineBackoff pipelineBackoff; - public NpPipelinePackageFabric(@Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, - @Qualifier("idUpdaterQueueFactory") StageQueueFactory idUpdaterQueueFactory, - @Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, + public NpPipelinePackageFabric(@Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, @Qualifier("nameUpdaterQueueFactory") StageQueueFactory nameUpdaterQueueFactory, - @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, + @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, @Qualifier("noOpQueueFactory") StageQueueFactory noOpQueueFactory, - FidIdEnricher fidIdEnricherHandler, - FidIdUpdater fidIdUpdater, - FidNameUpdater fidNameUpdater, - FidNameUpdaterMigration fidNameUpdaterMigration, - Supplier> pipelineTerminatorFactory, + NpFidUpdater npFidUpdater, + Supplier> pipelineTerminatorFactory, Supplier> noOpTerminatorFactory, Function> pipelineStarterFactory, - FidNameEnricher fidNameEnricher, + NpFidEnricher npFidEnricher, CreditTrackerSettings creditTrackerSettings, PipelineBackoff pipelineBackoff) { - this.idEnricherQueueFactory = idEnricherQueueFactory; - this.idUpdaterQueueFactory = idUpdaterQueueFactory; this.nameEnricherQueueFactory = nameEnricherQueueFactory; this.nameUpdaterQueueFactory = nameUpdaterQueueFactory; this.sendSignalQueueFactory = sendSignalQueueFactory; this.noOpQueueFactory = noOpQueueFactory; - this.fidIdEnricherHandler = fidIdEnricherHandler; - this.fidIdUpdater = fidIdUpdater; - this.fidNameUpdater = fidNameUpdater; - this.fidNameUpdaterMigration = fidNameUpdaterMigration; + this.npFidUpdater = npFidUpdater; this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.noOpTerminatorFactory = noOpTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; - this.fidNameEnricher = fidNameEnricher; + this.npFidEnricher = npFidEnricher; this.creditTrackerSettings = creditTrackerSettings; this.pipelineBackoff = pipelineBackoff; } public Pipeline build(SetupParam setupParam) throws Exception { - return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) - .handler(fidNameEnricher) - .andThen() - .stage(sendSignalQueueFactory.getObject()) - .handler(fidIdEnricherHandler) + return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), sendSignalQueueFactory.getObject()) + .handler(npFidEnricher) .andThen() .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) .pollFinishingQueue(pipelineTerminatorFactory.get()) @@ -86,26 +69,9 @@ public class NpPipelinePackageFabric implements InitializingBean { } public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception { - return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) - .handler(fidNameUpdater) + return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), noOpQueueFactory.getObject()) + .handler(npFidUpdater) .andThen() - .stage(noOpQueueFactory.getObject()) - .handler(fidIdUpdater) - .andThen() - .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) - .pollFinishingQueue(noOpTerminatorFactory.get()) - .backoff(pipelineBackoff) - .context(PackageVariable.class) - .build(); - } - - public Pipeline buildRecalculationPDMigration(SetupParam setupParam) throws Exception { - return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) - .handler(fidNameUpdaterMigration) - .andThen() - .stage(noOpQueueFactory.getObject()) - .handler(fidIdUpdater) - .andThen() .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) .pollFinishingQueue(noOpTerminatorFactory.get()) .backoff(pipelineBackoff) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java deleted file mode 100644 index cbbc4cf..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java +++ /dev/null @@ -1,75 +0,0 @@ -package ru.nbch.credit_tracker.service.task; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.stereotype.Component; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.config.task.StageQueueFactory; -import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload; -import ru.nbch.credit_tracker.model.impl.PackageVariable; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.model.impl.VoidPayload; -import ru.nbch.credit_tracker.service.task.impl.FidIdUpdater; -import ru.nbch.credit_tracker.service.task.impl.FidNameUpdaterMigration; -import ru.nbch.pipeline.engine.core.Pipeline; -import ru.nbch.pipeline.engine.core.PipelineBackoff; -import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; -import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator; -import ru.nbch.pipeline.engine.core.setup.PipeLineBuilderV2; - -import java.util.function.Function; -import java.util.function.Supplier; - -@Component -@Slf4j -public class PipelineMigrationPackageFabric implements InitializingBean { - private final StageQueueFactory idUpdaterQueueFactory; - private final StageQueueFactory nameUpdaterQueueFactory; - private final StageQueueFactory noOpQueueFactory; - private final FidIdUpdater fidIdUpdater; - private final FidNameUpdaterMigration fidNameUpdaterMigration; - private final Supplier> noOpTerminatorFactory; - private final Function> pipelineStarterFactory; - private final CreditTrackerSettings creditTrackerSettings; - private final PipelineBackoff pipelineBackoff; - - public PipelineMigrationPackageFabric(@Qualifier("idUpdaterQueueFactory") StageQueueFactory idUpdaterQueueFactory, - @Qualifier("nameUpdaterQueueFactory") StageQueueFactory nameUpdaterQueueFactory, - @Qualifier("noOpQueueFactory") StageQueueFactory noOpQueueFactory, - FidIdUpdater fidIdUpdater, - FidNameUpdaterMigration fidNameUpdaterMigration, - Supplier> noOpTerminatorFactory, - Function> pipelineStarterFactory, - CreditTrackerSettings creditTrackerSettings, - PipelineBackoff pipelineBackoff) { - this.idUpdaterQueueFactory = idUpdaterQueueFactory; - this.nameUpdaterQueueFactory = nameUpdaterQueueFactory; - this.noOpQueueFactory = noOpQueueFactory; - this.fidIdUpdater = fidIdUpdater; - this.fidNameUpdaterMigration = fidNameUpdaterMigration; - this.noOpTerminatorFactory = noOpTerminatorFactory; - this.pipelineStarterFactory = pipelineStarterFactory; - this.creditTrackerSettings = creditTrackerSettings; - this.pipelineBackoff = pipelineBackoff; - } - - public Pipeline buildRecalculationPDMigration(SetupParam setupParam) throws Exception { - return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) - .handler(fidNameUpdaterMigration) - .andThen() - .stage(noOpQueueFactory.getObject()) - .handler(fidIdUpdater) - .andThen() - .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) - .pollFinishingQueue(noOpTerminatorFactory.get()) - .backoff(pipelineBackoff) - .context(PackageVariable.class) - .build(); - } - - @Override - public void afterPropertiesSet() throws Exception { - log.info("pipeline settings: {}", creditTrackerSettings.getPipeline()); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java index dd4c632..61a9d31 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java @@ -4,14 +4,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.enums.RejectedStatus; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.PackageVariable; -import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; -import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService; -import ru.nbch.credit_tracker.service.signal.SignalRestService; +import ru.nbch.credit_tracker.model.impl.SignalPayload; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; -import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; +import ru.nbch.credit_tracker.service.task.impl.SetupMonitoringPipelineTerminator; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; import ru.nbch.pipeline.engine.core.Pipeline; @@ -27,25 +24,20 @@ public class PipelinePackageOrchestrator { private final Logger log = LoggerFactory.getLogger(getClass()); private final CreditTrackerSettings creditTrackerSettings; private final UserPackageMapper userPackageMapper; - private final SignalRestService signalRestService; private final ExecutorService pipelineExecutorService; private final ExecutorService saverExecutorService; private final Object lock = new Object(); private final java.util.Deque pending = new java.util.ArrayDeque<>(); - private final MonitoringErrorProcesingService monitoringErrorProcesingService; private final Semaphore semaphore; public PipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, UserPackageMapper userPackageMapper, - SignalRestService signalRestService, ExecutorService pipelineExecutorService, - ExecutorService saverExecutorService, MonitoringErrorProcesingService monitoringErrorProcesingService) { + ExecutorService saverExecutorService) { this.creditTrackerSettings = creditTrackerSettings; this.userPackageMapper = userPackageMapper; - this.signalRestService = signalRestService; this.pipelineExecutorService = pipelineExecutorService; this.saverExecutorService = saverExecutorService; - this.monitoringErrorProcesingService = monitoringErrorProcesingService; this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); } @@ -103,7 +95,7 @@ public class PipelinePackageOrchestrator { try { long finishTime = System.currentTimeMillis(); log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime()); - SignalXmlFilePayload res = (SignalXmlFilePayload) v; + SignalPayload res = (SignalPayload) v; var ctx = pipeline.obtainContext(PackageVariable.class); AtomicInteger cacheHits = ctx.get(PackageVariable.cacheHits); AtomicInteger cacheMisses = ctx.get(PackageVariable.cacheMisses); @@ -112,31 +104,24 @@ public class PipelinePackageOrchestrator { cacheHits.get(), cacheMisses.get(), cacheHits.get() + cacheMisses.get()); - if (e instanceof XmlMergeCoordinator.PipelineTerminatorException) { + if (e instanceof SetupMonitoringPipelineTerminator.PipelineTerminatorException) { log.info("when complete: terminator was interrupted."); return; } else if (e != null) { log.error(ExceptionUtils.getStackTrace(e)); return; } - if (res.signalXmlFilePath() == null) { - log.info("No one fid was found by packageId : {}; xml file won't be sent", packageId); + if (!res.hasFids()) { + log.info("No one fid was found by packageId : {}.", packageId); userPackageMapper.updateNoFidMonitoring(packageId); return; } - log.trace("Signal file by path: {} is completed", res.signalXmlFilePath()); - boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath()); - if (success) { - PipelineMode mode = ctx.get(PackageVariable.mode); - if (mode != null && mode.equals(PipelineMode.DEFAULT)) { - userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); - } else { - userPackageMapper.updatePackageOnMonitoringSuccess(packageId, null); - } - monitoringErrorProcesingService.findErrors(packageId); + PipelineMode mode = ctx.get(PackageVariable.mode); + if (mode != null && mode.equals(PipelineMode.DEFAULT)) { + userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); } else { - userPackageMapper.setRejectedStatus(packageId, RejectedStatus.ERROR.getKey()); + userPackageMapper.updatePackageOnMonitoringSuccess(packageId, null); } } 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/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java deleted file mode 100644 index 7d64305..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java +++ /dev/null @@ -1,134 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.builder.PersonFactoryBuilder; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.external_request.signals.Person; -import ru.nbch.credit_tracker.entities.indic.Id; -import ru.nbch.credit_tracker.model.IdData; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.PartFilePayload; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.PersonService; -import ru.nbch.credit_tracker.service.signal.PackageRecordsService; -import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2; -import ru.nbch.credit_tracker.utils.CTUtil; -import ru.nbch.pipeline.engine.core.Handler; - -import java.io.IOException; -import java.io.OutputStream; -import java.nio.file.*; -import java.util.*; -import java.util.stream.Stream; - -import static ru.nbch.credit_tracker.utils.CTUtil.defineIdType; - -@Service -@Slf4j -public class FidIdEnricher implements Handler { - private final PersonService personService; - private final PackageRecordsService packageRecordsService; - private final CreditTrackerSettings creditTrackerSettings; - - public FidIdEnricher(PersonService personService, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { - this.personService = personService; - this.packageRecordsService = packageRecordsService; - this.creditTrackerSettings = creditTrackerSettings; - } - - @Override - public PartFilePayload handle(SubjectPayload taskPayload) { - List subjectsProfile = taskPayload.getSubjects(); - log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); - - List noIdDataFids = subjectsProfile.stream() - .filter(subjectProfile -> subjectProfile.getIdData() == null && subjectProfile.getFid() != null) - .map(SubjectProfile::getFid) - .toList(); - - long countEmptyFid = subjectsProfile.stream().filter(subjectProfile -> subjectProfile.getFid() == null).count(); - log.info("Count with empty fid: {}", countEmptyFid); - Map idByFid; - if (!noIdDataFids.isEmpty()) { - idByFid = personService.filledPassportsData(noIdDataFids); - } else { - idByFid = new HashMap<>(); - } - - List idCacheOnUpdate = new ArrayList<>(); - List personOnProcessing = subjectsProfile.stream() - .filter(subjectProfile -> subjectProfile.getFid() != null) - .flatMap(subjectProfile -> { - if (subjectProfile.getIdData() == null) { - Id id = idByFid.get(subjectProfile.getFid()); - if (id == null) { - log.warn("Id not found for fid: {}", subjectProfile.getFid()); - return Stream.empty(); - } - String serNum = CTUtil.normalizePassport(id); - IdData idData = IdData.builder() - .serNum(serNum) - .idType(defineIdType(id.getIdType(), id.getIdType2())) - .fid(subjectProfile.getFid()) - .phone(subjectProfile.getPhone()) - .build(); - idCacheOnUpdate.add(idData); - subjectProfile.setIdData(idData); - } - return Stream.of(PersonFactoryBuilder.builder() - .personData(subjectProfile) - .build()); - }) - .toList(); - - packageRecordsService.updateIdCache(idCacheOnUpdate); - log.debug("personOnProcessing size: {}", personOnProcessing.size()); - Long packageId = taskPayload.getPackageId(); - if (personOnProcessing.isEmpty()) { - return new PartFilePayload(null, packageId, taskPayload.getMemberCode(), taskPayload.getPackageName()); - } - Path partPath = writeSignals(personOnProcessing, taskPayload); - return new PartFilePayload(partPath, packageId, taskPayload.getMemberCode(), taskPayload.getPackageName()); - } - - private Path writeSignals(List persons, SubjectPayload taskPayload) { - Path dir = Paths.get(creditTrackerSettings.getLoadDir(), - "tmp", taskPayload.getMemberCode(), taskPayload.getPackageName()); - try { - Files.createDirectories(dir); - } catch (IOException e) { - throw new RuntimeException(e); - } - - String base = UUID.randomUUID() + "_output.xml"; - log.debug("Write tmp xml by packageId: {}", taskPayload.getPackageId()); - Path tmp = dir.resolve(base + ".part"); - Path done = dir.resolve(base); - - try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) { - SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, null, creditTrackerSettings.getEncoding()); - writer.writeWithStax(os); // перегрузи на OutputStream, если нужно - } catch (IOException e) { - throw new RuntimeException(e); - } - try { - Files.move(tmp, done, StandardCopyOption.ATOMIC_MOVE); - } catch (IOException e) { - throw new RuntimeException(e); - } - return done; - } - - @Override - public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().idEnricherStageConcurrency(); - } - - @Override - public PartFilePayload cookPoison() { - return PartFilePayload.POISON; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java deleted file mode 100644 index f82c35e..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java +++ /dev/null @@ -1,80 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.indic.Id; -import ru.nbch.credit_tracker.model.IdData; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.model.impl.VoidPayload; -import ru.nbch.credit_tracker.service.indic.PersonService; -import ru.nbch.credit_tracker.service.signal.PackageRecordsService; -import ru.nbch.credit_tracker.utils.CTUtil; -import ru.nbch.pipeline.engine.core.Handler; - -import java.util.List; -import java.util.Map; -import java.util.stream.Stream; - -import static ru.nbch.credit_tracker.utils.CTUtil.defineIdType; - -@Service -@Slf4j -public class FidIdUpdater implements Handler { - private final PersonService personService; - private final PackageRecordsService packageRecordsService; - private final CreditTrackerSettings creditTrackerSettings; - - public FidIdUpdater(PersonService personService, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { - this.personService = personService; - this.packageRecordsService = packageRecordsService; - this.creditTrackerSettings = creditTrackerSettings; - } - - @Override - public VoidPayload handle(SubjectPayload taskPayload) { - List subjectsProfile = taskPayload.getSubjects(); - log.debug("FidIdUpdater handle; size: {}", subjectsProfile.size()); - - List noIdDataFids = subjectsProfile.stream() - .map(SubjectProfile::getFid) - .toList(); - Map idByFid = personService.filledPassportsData(noIdDataFids); - - List personOnProcessing = subjectsProfile.stream() - .flatMap(subjectProfile -> { - Id id = idByFid.get(subjectProfile.getFid()); - if (id == null) { - log.warn("Id not found for fid: {}", subjectProfile.getFid()); - return Stream.empty(); - } - String serNum = CTUtil.normalizePassport(id); - IdData idData = IdData.builder() - .serNum(serNum) - .idType(defineIdType(id.getIdType(), id.getIdType2())) - .fid(subjectProfile.getFid()) - .phone(subjectProfile.getPhone()) - .build(); - return Stream.of(idData); - }) - .toList(); - - packageRecordsService.updateIdCache(personOnProcessing); - log.debug("personOnProcessing size: {}", personOnProcessing.size()); - - return new VoidPayload(); - } - - @Override - public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().idUpdaterStageConcurrency(); - } - - @Override - public VoidPayload cookPoison() { - return VoidPayload.POISON; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java deleted file mode 100644 index 543965e..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java +++ /dev/null @@ -1,75 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.indic.Name; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.PackageVariable; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.PersonService; -import ru.nbch.credit_tracker.service.signal.PhonesMapService; -import ru.nbch.credit_tracker.utils.CTUtil; -import ru.nbch.pipeline.engine.core.Handler; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; - -@Service -@Slf4j -public class FidNameUpdaterMigration implements Handler { - private final PersonService personService; - private final CreditTrackerSettings creditTrackerSettings; - private final PhonesMapService phonesMapService; - - public FidNameUpdaterMigration(PersonService personService, - CreditTrackerSettings creditTrackerSettings, - PhonesMapService phonesMapService) { - this.personService = personService; - this.creditTrackerSettings = creditTrackerSettings; - this.phonesMapService = phonesMapService; - } - - @Override - public SubjectPayload handle(SubjectPayload taskPayload) { - List subjects = taskPayload.getSubjects(); - var ctx = obtainContext(PackageVariable.class, taskPayload); - log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size()); - Map nameByPhone = personService.findNameByFids(subjects); - - List changingFidPhones = new ArrayList<>(); - List notFoundFidPhones = new ArrayList<>(); - subjects.forEach(subjectProfile -> { - Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone())); - if (name != null) { - subjectProfile.getNameData().setFirstName(name.getFirst()); - subjectProfile.getNameData().setLastName(name.getLast()); - subjectProfile.getNameData().setMiddleName(name.getMiddle()); - subjectProfile.getNameData().setBirthDate(name.getBirthDate()); - changingFidPhones.add(subjectProfile); - } else { - notFoundFidPhones.add(subjectProfile.getPhone()); - } - }); - - if (!notFoundFidPhones.isEmpty()) { - phonesMapService.resetFidAndPDByPhones(notFoundFidPhones); - } - - if (!changingFidPhones.isEmpty()) { - phonesMapService.updateNameCachedData(changingFidPhones); - } - return new SubjectPayload(null, null, null, changingFidPhones); - } - - @Override - public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); - } - - @Override - public SubjectPayload cookPoison() { - return SubjectPayload.POISON; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java index f062e2e..5737040 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java @@ -8,8 +8,6 @@ import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.model.IdData; -import ru.nbch.credit_tracker.model.NameData; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.SubjectPayload; @@ -73,16 +71,6 @@ public class KickRecalcPipeLine implements IPipeLineStarter { .mapId(phoneFidMap.getId()) .phone(phoneFidMap.getPhone()) .fid(phoneFidMap.getFid()) - .nameData(NameData.builder() - .middleName(phoneFidMap.getMiddle()) - .firstName(phoneFidMap.getFirst()) - .lastName(phoneFidMap.getName1()) - .birthDate(phoneFidMap.getBirthDt()) - .build()) - .idData(IdData.builder() - .idNum(phoneFidMap.getSerNum()) - .idType(phoneFidMap.getIdType()) - .build()) .build()).toList(); try { SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles); 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/NpFidEnricher.java similarity index 71% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java index 2feda07..9aa6fe3 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/NpFidEnricher.java @@ -6,8 +6,6 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhonesMap; import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.model.FlagData; -import ru.nbch.credit_tracker.model.IdData; -import ru.nbch.credit_tracker.model.NameData; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.SubjectPayload; @@ -28,14 +26,14 @@ import java.util.stream.Stream; */ @Service @Slf4j -public class FidNameEnricher implements Handler { +public class NpFidEnricher implements Handler { private final PersonService personService; private final CreditTrackerSettings creditTrackerSettings; private final PackageRecordsService packageRecordsService; - public FidNameEnricher(PersonService personService, - CreditTrackerSettings creditTrackerSettings, - PackageRecordsService packageRecordsService) { + public NpFidEnricher(PersonService personService, + CreditTrackerSettings creditTrackerSettings, + PackageRecordsService packageRecordsService) { this.personService = personService; this.creditTrackerSettings = creditTrackerSettings; this.packageRecordsService = packageRecordsService; @@ -88,14 +86,10 @@ public class FidNameEnricher implements Handler .phone(phone) .build()); } - fillByName(subject, name); + subject.setFid(name.getFid()); return Stream.of(PhonesMap.builder() .fid(subject.getFid()) .phone(phone) - .name1(name.getLast()) - .first(name.getFirst()) - .middle(name.getMiddle()) - .birthDt(name.getBirthDate()) .build()); } return Stream.empty(); @@ -107,29 +101,8 @@ public class FidNameEnricher implements Handler } private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) { - if (cachedRecord.getName1() == null || cachedRecord.getFirst() == null || - cachedRecord.getMiddle() == null || cachedRecord.getBirthDt() == null) { - log.warn("By phone : {} found empty cache rec with id: {}", subjectProfile.getPhone(), cachedRecord.getId()); - return; - } - NameData nameData = NameData.builder() - .middleName(cachedRecord.getMiddle()) - .firstName(cachedRecord.getFirst()) - .lastName(cachedRecord.getName1()) - .birthDate(cachedRecord.getBirthDt()) - .build(); - subjectProfile.setFid(cachedRecord.getFid()); - subjectProfile.setNameData(nameData); - Optional.of(cachedRecord) - .filter(cr -> cachedRecord.getSerNum() != null && cachedRecord.getIdType() != null) - .map(cr -> Optional.of( - (IdData.builder() - .serNum(cr.getSerNum()) - .idType(cr.getIdType()) - .build() - ))) - .ifPresent(idData -> subjectProfile.setIdData(idData.get())); + .ifPresent(cr -> subjectProfile.setFid(cr.getFid())); Optional.of(cachedRecord) .filter(cr -> cachedRecord.getFlagDebt5000() != null && cachedRecord.getFlag90days() != null) @@ -142,19 +115,6 @@ public class FidNameEnricher implements Handler .ifPresent(flagData -> subjectProfile.setFlagData(flagData.get())); } - private void fillByName(SubjectProfile subjectProfile, Name name) { - NameData nameData = NameData.builder() - .middleName(name.getMiddle()) - .firstName(name.getFirst()) - .lastName(name.getLast()) - .fileSinceDate(name.getFileSinceDate()) - .birthDate(name.getBirthDate()) - .build(); - - subjectProfile.setFid(name.getFid()); - subjectProfile.setNameData(nameData); - } - @Override public int preferredConcurrency() { return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java similarity index 76% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java index 16918be..81b61e0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java @@ -7,6 +7,7 @@ import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; import ru.nbch.credit_tracker.service.UserPackageService; import ru.nbch.credit_tracker.service.indic.PersonService; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; @@ -21,18 +22,18 @@ import java.util.stream.Stream; @Service @Slf4j -public class FidNameUpdater implements Handler { +public class NpFidUpdater implements Handler { private final PersonService personService; private final CreditTrackerSettings creditTrackerSettings; private final PackageRecordsService packageRecordsService; private final PhonesMapService phonesMapService; private final UserPackageService userPackageService; - public FidNameUpdater(PersonService personService, - CreditTrackerSettings creditTrackerSettings, - PackageRecordsService packageRecordsService, - PhonesMapService phonesMapService, - UserPackageService userPackageService) { + public NpFidUpdater(PersonService personService, + CreditTrackerSettings creditTrackerSettings, + PackageRecordsService packageRecordsService, + PhonesMapService phonesMapService, + UserPackageService userPackageService) { this.personService = personService; this.creditTrackerSettings = creditTrackerSettings; this.packageRecordsService = packageRecordsService; @@ -41,7 +42,7 @@ public class FidNameUpdater implements Handler { } @Override - public SubjectPayload handle(SubjectPayload taskPayload) { + public VoidPayload handle(SubjectPayload taskPayload) { List subjects = taskPayload.getSubjects(); var ctx = obtainContext(PackageVariable.class, taskPayload); log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size()); @@ -54,10 +55,6 @@ public class FidNameUpdater implements Handler { if (name != null) { if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) { subjectProfile.setFid(name.getFid()); - subjectProfile.getNameData().setFirstName(name.getFirst()); - subjectProfile.getNameData().setLastName(name.getLast()); - subjectProfile.getNameData().setMiddleName(name.getMiddle()); - subjectProfile.getNameData().setBirthDate(name.getBirthDate()); changingFidPhones.add(subjectProfile); return Stream.empty(); } @@ -69,12 +66,12 @@ public class FidNameUpdater implements Handler { }).toList(); List changingFids = new ArrayList<>(); if (!notFoundFidPhones.isEmpty()) { - phonesMapService.resetFidAndPDByPhones(notFoundFidPhones); + phonesMapService.resetFidAndFlagsByPhones(notFoundFidPhones); changingFids.addAll(notFoundFidPhones); } if (!changingFidPhones.isEmpty()) { - phonesMapService.updateNameCachedData(changingFidPhones); + phonesMapService.updateFidPhoneCachedData(changingFidPhones); changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList()); } @@ -88,7 +85,7 @@ public class FidNameUpdater implements Handler { userPackageService.setFidChanged(packageIds); } } - return new SubjectPayload(null, null, null, changingFidPhones); + return new VoidPayload(); } @Override @@ -97,7 +94,7 @@ public class FidNameUpdater implements Handler { } @Override - public SubjectPayload cookPoison() { - return SubjectPayload.POISON; + public VoidPayload cookPoison() { + return VoidPayload.POISON; } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java new file mode 100644 index 0000000..46a57cd --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java @@ -0,0 +1,66 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.exception.ExceptionUtils; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.model.impl.SignalPayload; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator; +import ru.nbch.pipeline.engine.utils.TaskTimer; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +@Slf4j +public final class SetupMonitoringPipelineTerminator implements IPipeLineTerminator { + private final CompletableFuture done = new CompletableFuture<>(); + private final CreditTrackerSettings config; + private BlockingQueue inQueue; + private final TaskTimer stopWatch = new TaskTimer(); + + public SetupMonitoringPipelineTerminator(CreditTrackerSettings config) { + this.config = config; + } + + @Override + public void run() { + AtomicBoolean hasFids = new AtomicBoolean(false); + while (!Thread.currentThread().isInterrupted()) { + try { + SubjectPayload part = inQueue.poll(5, TimeUnit.SECONDS); + if (part == null) { + continue; + } + if (!part.isPoison() && !hasFids.get()) { + hasFids.set(part.getSubjects().stream().anyMatch(s -> s.getFid() != null)); + } + if (part.isPoison()) { + log.debug("Got poison"); + done.complete(new SignalPayload(hasFids.get())); + return; + } + } catch (InterruptedException ie) { + log.info("pipeline terminator was interrupted!"); + done.completeExceptionally(new PipelineTerminatorException()); + } catch (Exception e) { + log.error("Error: {}", ExceptionUtils.getStackTrace(e)); + } + } + done.completeExceptionally(new PipelineTerminatorException()); + } + + public static class PipelineTerminatorException extends RuntimeException { + } + + @Override + public void setPollingQueue(BlockingQueue mergerQueue) { + this.inQueue = mergerQueue; + } + + @Override + public CompletableFuture doneFuture() { + return done; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java deleted file mode 100644 index b0e08a0..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java +++ /dev/null @@ -1,179 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.exception.ExceptionUtils; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.model.impl.PartFilePayload; -import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; -import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator; -import ru.nbch.pipeline.engine.utils.TaskTimer; - -import javax.xml.stream.*; -import java.io.*; -import java.nio.charset.Charset; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.StandardOpenOption; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; - -@Slf4j -public final class XmlMergeCoordinator implements IPipeLineTerminator { - private final Charset encoding = Charset.forName("windows-1251"); - private final CompletableFuture done = new CompletableFuture<>(); - private Path out; - private final CreditTrackerSettings config; - private BlockingQueue inQueue; - private final TaskTimer stopWatch = new TaskTimer(); - - public XmlMergeCoordinator(CreditTrackerSettings config) { - this.config = config; - } - - @Override - public void run() { - while (!Thread.currentThread().isInterrupted()) { - try { - PartFilePayload part = inQueue.poll(5, TimeUnit.SECONDS); - if (part == null) { - continue; - } - if (part.isPoison()) { - log.debug("Got poison"); - finishMerge(); - done.complete(new SignalXmlFilePayload(out)); - log.info("Stage time stats: {}", stopWatch.info()); - return; - } - if (part.getPartPath() == null) { - continue; - } - try (var m = stopWatch.start("XmlMergeCoordinator")) { - mergeBatch(part); - } - } catch (InterruptedException ie) { - log.info("pipeline terminator was interrupted!"); - done.completeExceptionally(new PipelineTerminatorException()); - } catch (Exception e) { - log.error("Error: {}", ExceptionUtils.getStackTrace(e)); - } - } - done.completeExceptionally(new PipelineTerminatorException()); - } - - public static class PipelineTerminatorException extends RuntimeException { - } - - private void mergeBatch(PartFilePayload input) throws Exception { - log.debug("Starting merge"); - Long packageId = input.getPackageId(); - Path partFilePath = input.getPartPath(); - boolean first = false; - if (out == null) { - first = true; - out = partFilePath.resolveSibling("merged_" + packageId + ".xml"); - } - - XMLOutputFactory xof = XMLOutputFactory.newFactory(); - try (OutputStream os = Files.newOutputStream(out, StandardOpenOption.CREATE, StandardOpenOption.APPEND)) { - XMLStreamWriter w = xof.createXMLStreamWriter(os, encoding.name()); - - if (first) { - w.writeStartDocument(encoding.name(), "1.0"); - - w.writeStartElement("Signals"); - w.writeAttribute("online", "2"); - w.writeAttribute("processWithErrors", "1"); - - w.writeStartElement("Id"); - w.writeCharacters("%s_%s".formatted(input.getMembercode().substring(0, 6), input.getPackageName())); - w.writeEndElement(); // - - w.writeStartElement("Auth"); - w.writeStartElement("User"); - w.writeCharacters(config.getSignalsuser()); - w.writeEndElement(); // - w.writeEndElement(); // - - // - w.writeStartElement("Conditions"); - w.writeStartElement("Persons"); - } - log.debug("Copy person from file {}", input.getPartPath()); - copyPersonsChildren(input.getPartPath(), w); - w.flush(); - w.close(); - } - Files.deleteIfExists(input.getPartPath()); - } - - private void finishMerge() throws Exception { - if (out == null) { - return; - } - - try (Writer w = new BufferedWriter(new OutputStreamWriter( - Files.newOutputStream(out, StandardOpenOption.APPEND), - encoding))) { - - w.write(""); - w.write(""); - w.write(""); - } - } - /** - * Копирует всё содержимое внутри ... входного файла - * в текущий writer, но без повторного открытия/закрытия корневого тега. - */ - private void copyPersonsChildren(Path input, XMLStreamWriter w) throws Exception { - XMLInputFactory xif = XMLInputFactory.newFactory(); - try (InputStream is = Files.newInputStream(input)) { - XMLStreamReader r = xif.createXMLStreamReader(is, encoding.name()); - - while (r.hasNext()) { - int t = r.next(); - if (t == XMLStreamConstants.START_ELEMENT && "Persons".equals(r.getLocalName())) { - break; - } - } - - int depth = 0; // 0 — внутри корневого контейнера - while (r.hasNext()) { - int t = r.next(); - if (t == XMLStreamConstants.START_ELEMENT) { - depth++; - writeStart(w, r); - } else if (t == XMLStreamConstants.CHARACTERS) { - if (!r.isWhiteSpace()) w.writeCharacters(r.getText()); - } else if (t == XMLStreamConstants.CDATA) { - w.writeCData(r.getText()); - } else if (t == XMLStreamConstants.END_ELEMENT) { - if (depth > 0) { - w.writeEndElement(); - depth--; - } else if ("Persons".equals(r.getLocalName())) { - break; - } - } - } - r.close(); - } - } - private void writeStart(XMLStreamWriter w, XMLStreamReader r) throws XMLStreamException { - w.writeStartElement(r.getLocalName()); - for (int i = 0; i < r.getAttributeCount(); i++) { - w.writeAttribute(r.getAttributeLocalName(i), r.getAttributeValue(i)); - } - } - - @Override - public void setPollingQueue(BlockingQueue mergerQueue) { - this.inQueue = mergerQueue; - } - - @Override - public CompletableFuture doneFuture() { - return done; - } -} diff --git a/src/main/resources/mybatis/mapper/indic/PersonMapper.xml b/src/main/resources/mybatis/mapper/indic/PersonMapper.xml index 56f615d..eae91d3 100644 --- a/src/main/resources/mybatis/mapper/indic/PersonMapper.xml +++ b/src/main/resources/mybatis/mapper/indic/PersonMapper.xml @@ -10,10 +10,6 @@ - - - - @@ -31,10 +27,6 @@ pn.number as number, pn.fid as fid, pn.FILE_SINCE_DT as file_since_dt_pn, - nf.first as first, - nf.middle as middle, - nf.name_1 as name_1, - nf.birth_dt as birth_dt, nf.file_since_dt as file_since_dt_name FROM PHONE_NORM pn JOIN NAME nf @@ -46,34 +38,4 @@ WITH UR - - - - \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml index 4b0d7a4..0adbf9f 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml @@ -104,16 +104,6 @@ RETURNING 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; - - DELETE FROM signals_phone_fid_map diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index a53c95b..ca4d1f5 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -9,12 +9,6 @@ - - - - - - @@ -25,12 +19,6 @@ select id, fid, phone, - name_1, - first, - middle, - birth_dt, - ser_num, - id_type, flag_debt5000, flag_90_12, searched_at, @@ -43,12 +31,6 @@ select id, fid, phone, - name_1, - first, - middle, - birth_dt, - ser_num, - id_type, flag_debt5000, flag_90_12, searched_at, @@ -74,12 +56,6 @@ select id, fid, phone, - name_1, - first, - middle, - birth_dt, - ser_num, - id_type, flag_debt5000, flag_90_12, searched_at, @@ -93,12 +69,6 @@ WITH src AS (SELECT row_number() OVER () AS rn, fid, - phone, - name_1, - first, - middle, - birth_dt + phone FROM json_to_recordset(#{json}::json) AS t( fid int, - phone bigint, - name_1 varchar(2000), - first varchar(2000), - middle varchar(2000), - birth_dt timestamp + phone bigint )) INSERT INTO signals_phones_map - (fid, phone, name_1, first, middle, birth_dt, searched_at) + (fid, phone, searched_at) SELECT fid, phone, - name_1, - first, - middle, - birth_dt, now() FROM src order by phone ON CONFLICT (phone) DO UPDATE SET fid = EXCLUDED.fid, - phone = EXCLUDED.phone, - name_1 = EXCLUDED.name_1, - first = EXCLUDED.first, - middle = EXCLUDED.middle, - birth_dt = EXCLUDED.birth_dt + phone = EXCLUDED.phone - - WITH src AS (SELECT DISTINCT ON (phone) phone, - ser_num, - id_type - FROM json_to_recordset(#{json}::json) AS src( - phone bigint, - ser_num varchar(255), - id_type varchar(255) - ) - order by phone), - - to_lock AS (select dst.id - from signals_phones_map dst - join src on src.phone = dst.phone - order by dst.phone - for update) - - update signals_phones_map dst - set ser_num = src.ser_num, - id_type = src.id_type - FROM src - where dst.phone = src.phone - AND dst.id in (select id from to_lock); - - WITH src AS (SELECT DISTINCT ON (phone) fid, - phone, - name_1, - first, - middle, - birth_dt + phone FROM json_to_recordset(#{json}::json) AS src( fid int, - phone bigint, - name_1 varchar(255), - first varchar(255), - middle varchar(255), - birth_dt timestamp + phone bigint ) order by phone), @@ -219,18 +140,10 @@ update signals_phones_map dst set fid = src.fid, - name_1 = src.name_1, - first = src.first, - middle = src.middle, - birth_dt = src.birth_dt, searched_at = now() FROM json_to_recordset(#{json}::json) AS src( fid int, - phone bigint, - name_1 varchar(255), - first varchar(255), - middle varchar(255), - birth_dt timestamp + phone bigint ) where dst.phone = src.phone AND dst.id in (select id from to_lock); @@ -269,12 +182,6 @@ update signals_phones_map dst set fid = null, - name_1 = null, - first = null, - middle = null, - birth_dt = null, - ser_num = null, - id_type = null, flag_debt5000 = null, flag_90_12 = null, calculated_at = null,