treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NP: get rid of SignalRestService and personal data loading/saving

Browse source
This commit is contained in:
Mikhail Trofimov 2026-07-02 12:50:07 +03:00
parent 2709eaeb19
commit 44512925f4
43 changed files with 170 additions and 2116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
credit-tracker.properties
View file

@ -1,3 +1,4 @@
# TODO delete unused env vars
server.port=${SERVER_PORT:8099} server.port=${SERVER_PORT:8099}
server.tomcat.accesslog.enabled=true server.tomcat.accesslog.enabled=true
server.tomcat.accesslog.directory=/opt/app/logs/access 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.monitoring-request-path=monitoring-request.xsd
ct.xsd-validation.signal-package-path=signal-package.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=/opt/app/reports
#ct.load-dir=D:\\folder_name\\signals #ct.load-dir=D:\\folder_name\\signals
ct.tmp.dir=D:\\folder_name\\signals\\tmp 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 #select NAME + FID BY PHONE
ct.pipeline.name-enricher-stage-concurrency=${PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY:20} ct.pipeline.name-enricher-stage-concurrency=${PIPELINE_NAME_ENRICHER_STAGE_CONCURRENCY:20}
ct.pipeline.name-enricher-queue-capacity=${PIPELINE_NAME_ENRICHER_QUEUE_CAPACITY: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 #Save part of signal xml file and merge; only one thread
ct.pipeline.merger-queue-capacity=${PIPELINE_MERGER_QUEUE_CAPACITY:50} 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 #select NAME + FID BY PHONE
ct.pipeline.name-updater-stage-concurrency=${PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY:20} ct.pipeline.name-updater-stage-concurrency=${PIPELINE_NAME_UPDATER_STAGE_CONCURRENCY:20}
ct.pipeline.name-updater-queue-capacity=${PIPELINE_NAME_UPDATER_QUEUE_CAPACITY: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 #select and calc flags

88
src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilder.java
View file

@ -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;
}
}

2
src/main/java/ru/nbch/credit_tracker/config/rest/ResponseErrorHandler.java
View file

@ -5,9 +5,11 @@ import org.springframework.http.client.ClientHttpResponse;
import org.springframework.web.client.DefaultResponseErrorHandler; import org.springframework.web.client.DefaultResponseErrorHandler;
import org.springframework.web.client.HttpClientErrorException; import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.HttpServerErrorException; import org.springframework.web.client.HttpServerErrorException;
import ru.nbch.credit_tracker.refactoring.Suspicios;
import java.io.IOException; import java.io.IOException;
@Suspicios("Не используется больше")
@Slf4j @Slf4j
public class ResponseErrorHandler extends DefaultResponseErrorHandler { public class ResponseErrorHandler extends DefaultResponseErrorHandler {
@Override @Override

11
src/main/java/ru/nbch/credit_tracker/config/rest/RestConfig.java
View file

@ -3,8 +3,6 @@ package ru.nbch.credit_tracker.config.rest;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory; 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.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
@ -12,16 +10,9 @@ import java.util.concurrent.Executors;
@Configuration @Configuration
public class RestConfig { 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") @Bean(name = "executorService")
public ExecutorService executorService() { public ExecutorService executorService() {
return Executors.newCachedThreadPool(new CustomizableThreadFactory("package-registration-")); return Executors.newCachedThreadPool(new CustomizableThreadFactory("package-registration-"));
} }
} }

2
src/main/java/ru/nbch/credit_tracker/config/rest/RootUriTemplateHandler.java
View file

@ -1,10 +1,12 @@
package ru.nbch.credit_tracker.config.rest; package ru.nbch.credit_tracker.config.rest;
import org.springframework.web.util.DefaultUriBuilderFactory; import org.springframework.web.util.DefaultUriBuilderFactory;
import ru.nbch.credit_tracker.refactoring.Suspicios;
import java.net.URI; import java.net.URI;
import java.util.Map; import java.util.Map;
@Suspicios("Не используется больше")
public class RootUriTemplateHandler extends DefaultUriBuilderFactory { public class RootUriTemplateHandler extends DefaultUriBuilderFactory {
private final String rootUri; private final String rootUri;

4
src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java
View file

@ -32,10 +32,6 @@ public class CreditTrackerSettings {
@NestedConfigurationProperty @NestedConfigurationProperty
private KafkaConsumerSettings kafkaConsumerSettings; private KafkaConsumerSettings kafkaConsumerSettings;
private String signalsRootUri;
private String signalsToken;
private String signalsuser;
private Long maxXmlFileSize; private Long maxXmlFileSize;
private String loadDir; private String loadDir;
private Long monitoringInterval; private Long monitoringInterval;

7
src/main/java/ru/nbch/credit_tracker/config/settings/PipelineSettings.java
View file

@ -1,8 +1,9 @@
package ru.nbch.credit_tracker.config.settings; package ru.nbch.credit_tracker.config.settings;
import org.springframework.boot.context.properties.bind.DefaultValue;
import java.time.Duration; import java.time.Duration;
import java.util.List; import java.util.List;
import org.springframework.boot.context.properties.bind.DefaultValue;
//X это количество потоков Save //X это количество потоков Save
//тогда: Name должно быть ~3x, Fid должно быть ~2x чтобы они поспевали друг за другом? //тогда: Name должно быть ~3x, Fid должно быть ~2x чтобы они поспевали друг за другом?
@ -10,12 +11,8 @@ public record PipelineSettings(
@DefaultValue("5000") Integer subjectBatchSize, @DefaultValue("5000") Integer subjectBatchSize,
@DefaultValue("15") Integer nameEnricherStageConcurrency, @DefaultValue("15") Integer nameEnricherStageConcurrency,
@DefaultValue("20") Integer nameEnricherQueueCapacity, @DefaultValue("20") Integer nameEnricherQueueCapacity,
@DefaultValue("10") Integer idEnricherStageConcurrency,
@DefaultValue("15") Integer idEnricherQueueCapacity,
@DefaultValue("15") Integer nameUpdaterStageConcurrency, @DefaultValue("15") Integer nameUpdaterStageConcurrency,
@DefaultValue("20") Integer nameUpdaterQueueCapacity, @DefaultValue("20") Integer nameUpdaterQueueCapacity,
@DefaultValue("10") Integer idUpdaterStageConcurrency,
@DefaultValue("15") Integer idUpdaterQueueCapacity,
@DefaultValue("50") Integer mergerQueueCapacity, @DefaultValue("50") Integer mergerQueueCapacity,
@DefaultValue("5") Integer errorAttemptCount, @DefaultValue("5") Integer errorAttemptCount,
@DefaultValue("2s") Duration errorAttemptDelay, @DefaultValue("2s") Duration errorAttemptDelay,

17
src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java
View file

@ -5,7 +5,10 @@ import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; 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.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.*; import ru.nbch.credit_tracker.service.task.impl.*;
@ -24,13 +27,9 @@ public class PipelineStarterConfig {
PackageRecordsService packageRecordsService, PackageRecordsService packageRecordsService,
PhonesMapMapper phonesMapMapper) { PhonesMapMapper phonesMapMapper) {
return (setupParam) -> switch (setupParam.mode()) { return (setupParam) -> switch (setupParam.mode()) {
case DEFAULT, RESEND -> case DEFAULT, RESEND, RESTORE ->
new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
case RESTORE -> case REPEATED_RESTORE, MAKE_XML_BY_REPEATED ->
new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
case REPEATED_RESTORE ->
new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
case MAKE_XML_BY_REPEATED ->
new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
case RECALC_PERSONAL_DATA -> case RECALC_PERSONAL_DATA ->
new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings); new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings);
@ -38,8 +37,8 @@ public class PipelineStarterConfig {
} }
@Bean @Bean
public Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) { public Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) {
return () -> new XmlMergeCoordinator(creditTrackerSettings); return () -> new SetupMonitoringPipelineTerminator(creditTrackerSettings);
} }
@Bean @Bean

13
src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
View file

@ -3,7 +3,6 @@ package ru.nbch.credit_tracker.config.task;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; 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.SubjectPayload;
import ru.nbch.credit_tracker.model.impl.VoidPayload; import ru.nbch.credit_tracker.model.impl.VoidPayload;
@ -20,23 +19,13 @@ public class StageConfig {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity()); return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity());
} }
@Bean("idEnricherQueueFactory")
public StageQueueFactory<SubjectPayload> idEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
}
@Bean("idUpdaterQueueFactory")
public StageQueueFactory<SubjectPayload> idUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity());
}
@Bean("noOpQueueFactory") @Bean("noOpQueueFactory")
public StageQueueFactory<VoidPayload> noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) { public StageQueueFactory<VoidPayload> noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
} }
@Bean("partFileQueueFactory") @Bean("partFileQueueFactory")
public StageQueueFactory<PartFilePayload> partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) { public StageQueueFactory<SubjectPayload> partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) {
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
} }

14
src/main/java/ru/nbch/credit_tracker/entities/app/PhonesMap.java
View file

@ -1,12 +1,8 @@
package ru.nbch.credit_tracker.entities.app; package ru.nbch.credit_tracker.entities.app;
import java.time.LocalDate; import lombok.*;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter @Getter
@Setter @Setter
@ -17,12 +13,6 @@ public class PhonesMap {
private Long id; private Long id;
private Integer fid; private Integer fid;
private Long phone; 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 flagDebt5000;
private Integer flag90days; private Integer flag90days;
private LocalDateTime searchedAt; private LocalDateTime searchedAt;

12
src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java
View file

@ -1,11 +1,8 @@
package ru.nbch.credit_tracker.entities.indic; package ru.nbch.credit_tracker.entities.indic;
import lombok.*;
import java.time.LocalDate; import java.time.LocalDate;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter @Getter
@Setter @Setter
@ -17,9 +14,4 @@ public class Name {
private String number; private String number;
private LocalDate fileSinceDate; private LocalDate fileSinceDate;
private LocalDate fileSinceDatePn; private LocalDate fileSinceDatePn;
private String first;
private String last;
private String middle;
private LocalDate birthDate;
} }

11
src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java
View file

@ -29,7 +29,6 @@ import ru.nbch.credit_tracker.model.MonitoringValidationError;
import ru.nbch.credit_tracker.model.ValidationResult; import ru.nbch.credit_tracker.model.ValidationResult;
import ru.nbch.credit_tracker.service.UserPackageService; import ru.nbch.credit_tracker.service.UserPackageService;
import ru.nbch.credit_tracker.service.scoring.AuthorizationService; 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.LpPipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
@ -67,7 +66,6 @@ public class SignalFacade {
private final AuthorizationService authorizationService; private final AuthorizationService authorizationService;
private final NpPipelinePackageFabric npPipelinePackageFabric; private final NpPipelinePackageFabric npPipelinePackageFabric;
private final LpPipelinePackageFabric lpPipelinePackageFabric; private final LpPipelinePackageFabric lpPipelinePackageFabric;
private final SignalRestService signalRestService;
private final Saver saver; private final Saver saver;
private final PipelinePackageOrchestrator pipelineOrchestrator; private final PipelinePackageOrchestrator pipelineOrchestrator;
private final Pattern DOWNLOAD_FILE_PATTERN = Pattern.compile("^(?<member>[A-Za-z0-9]{6})_(?<package>[A-Za-z0-9_]+)_(?<ts>\\d{4}\\d{2}\\d{2}_\\d{2}\\d{2}\\d{2})(?:\\.xml)?\\.zip$"); private final Pattern DOWNLOAD_FILE_PATTERN = Pattern.compile("^(?<member>[A-Za-z0-9]{6})_(?<package>[A-Za-z0-9_]+)_(?<ts>\\d{4}\\d{2}\\d{2}_\\d{2}\\d{2}\\d{2})(?:\\.xml)?\\.zip$");
@ -78,7 +76,6 @@ public class SignalFacade {
AuthorizationService authorizationService, AuthorizationService authorizationService,
NpPipelinePackageFabric npPipelinePackageManager, NpPipelinePackageFabric npPipelinePackageManager,
LpPipelinePackageFabric lpPipelinePackageFabric, LpPipelinePackageFabric lpPipelinePackageFabric,
SignalRestService signalRestService,
Saver saver, PipelinePackageOrchestrator pipelineOrchestrator) { Saver saver, PipelinePackageOrchestrator pipelineOrchestrator) {
this.monitoringRequestValidator = monitoringRequestValidator; this.monitoringRequestValidator = monitoringRequestValidator;
this.config = config; this.config = config;
@ -86,7 +83,6 @@ public class SignalFacade {
this.authorizationService = authorizationService; this.authorizationService = authorizationService;
this.npPipelinePackageFabric = npPipelinePackageManager; this.npPipelinePackageFabric = npPipelinePackageManager;
this.lpPipelinePackageFabric = lpPipelinePackageFabric; this.lpPipelinePackageFabric = lpPipelinePackageFabric;
this.signalRestService = signalRestService;
this.saver = saver; this.saver = saver;
this.pipelineOrchestrator = pipelineOrchestrator; this.pipelineOrchestrator = pipelineOrchestrator;
} }
@ -252,13 +248,6 @@ public class SignalFacade {
throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName); 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(); LocalDateTime deactivatedAt = LocalDateTime.now();
userPackageService.deletePackageById(packageToDelete.getId(), deactivatedAt); userPackageService.deletePackageById(packageToDelete.getId(), deactivatedAt);
MonitoringPackageDeleteResponse response = new MonitoringPackageDeleteResponse(); MonitoringPackageDeleteResponse response = new MonitoringPackageDeleteResponse();

2
src/main/java/ru/nbch/credit_tracker/mapper/indic/PersonMapper.java
View file

@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.mapper.indic;
import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Param;
import ru.nbch.credit_tracker.entities.indic.Id;
import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.entities.indic.Name;
import java.util.List; import java.util.List;
@ -14,5 +13,4 @@ public interface PersonMapper {
List<Name> findNameDataByFids(@Param("fids") List<Integer> fids); List<Name> findNameDataByFids(@Param("fids") List<Integer> fids);
List<Id> findPassportData(@Param("fids") List<Integer> fids);
} }

1
src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java
View file

@ -10,7 +10,6 @@ import java.util.List;
@Mapper @Mapper
public interface PackagesRecordsMapper { public interface PackagesRecordsMapper {
void setErrorJson(@Param("json") String json);
List<Long> registerJson(@Param("json") String json); List<Long> registerJson(@Param("json") String json);

2
src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java
View file

@ -24,8 +24,6 @@ public interface PhonesMapMapper {
List<PhonesMap> findEligibleForFlagsInsert(@Param("lastId") Long lastId, List<PhonesMap> findEligibleForFlagsInsert(@Param("lastId") Long lastId,
@Param("batchSize") Integer batchSize); @Param("batchSize") Integer batchSize);
void updateIdCachedData(@Param("json") String json);
void updateNameCachedData(@Param("json") String json); void updateNameCachedData(@Param("json") String json);
void updateFlagCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json);

2
src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java
View file

@ -12,7 +12,5 @@ public class SubjectProfile {
private Long mapId; private Long mapId;
private Long phone; private Long phone;
private Integer fid; private Integer fid;
private NameData nameData;
private IdData idData;
private FlagData flagData; private FlagData flagData;
} }

6
src/main/java/ru/nbch/credit_tracker/model/impl/SignalPayload.java Normal file
View file

@ -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 {
}

8
src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java
View file

@ -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 {
}

107
src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java
View file

@ -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<Path> 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<PackagesRecordsMap> 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;
}
}

43
src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java
View file

@ -1,7 +1,6 @@
package ru.nbch.credit_tracker.service.indic; package ru.nbch.credit_tracker.service.indic;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.entities.indic.Id; import ru.nbch.credit_tracker.entities.indic.Id;
import ru.nbch.credit_tracker.entities.indic.Name; 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.model.SubjectProfile;
import ru.nbch.credit_tracker.utils.CTUtil; import ru.nbch.credit_tracker.utils.CTUtil;
import java.time.LocalDate;
import java.util.Comparator; import java.util.Comparator;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.function.BinaryOperator; import java.util.function.BinaryOperator;
@ -36,14 +33,6 @@ public class PersonService {
this.mapper = mapper; this.mapper = mapper;
} }
public Map<Integer, Id> filledPassportsData(List<Integer> fids) {
if (fids.isEmpty()) {
return new HashMap<>();
}
List<Id> passports = mapper.findPassportData(fids);
return groupPassportsData(passports);
}
public Map<Integer, Id> groupPassportsData(List<Id> passports) { public Map<Integer, Id> groupPassportsData(List<Id> passports) {
return passports.stream() return passports.stream()
.filter(this::isDUL) .filter(this::isDUL)
@ -75,38 +64,6 @@ public class PersonService {
List<Name> names = mapper.findNameData(phones); List<Name> names = mapper.findNameData(phones);
return names.stream() 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<String, Name> findNameByFids(List<SubjectProfile> subjects) {
List<Integer> fids = subjects.stream()
.map(SubjectProfile::getFid)
.toList();
List<Name> 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( .collect(Collectors.toMap(
Name::getNumber, Name::getNumber,
Function.identity(), Function.identity(),

13
src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java
View file

@ -39,17 +39,4 @@ public class PhoneService {
} }
} }
public void updateOnError(List<PackagesRecordsMap> subjects) {
try {
String json = entityToJsonMapper.writeValueAsString(subjects.stream()
.map(phoneFidMap -> new HashMap<String, Object>() {{
put("id", phoneFidMap.getId());
put("error_text", phoneFidMap.getErrorText());
}}).toList());
phoneFidMapMapper.setErrorJson(json);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
} }

12
src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java
View file

@ -8,7 +8,6 @@ import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; 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.NpPipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.SetupParam;
@ -35,7 +34,6 @@ public class StartupRestoreService {
private final NpPipelinePackageFabric npPipelinePackageManager; private final NpPipelinePackageFabric npPipelinePackageManager;
private final PipelinePackageOrchestrator pipelineOrchestrator; private final PipelinePackageOrchestrator pipelineOrchestrator;
private final CreditTrackerSettings creditTrackerSettings; private final CreditTrackerSettings creditTrackerSettings;
private final MonitoringErrorProcesingService monitoringErrorProcesingService;
private final Saver saver; private final Saver saver;
public StartupRestoreService(UserPackageMapper userPackageMapper, public StartupRestoreService(UserPackageMapper userPackageMapper,
@ -43,30 +41,20 @@ public class StartupRestoreService {
NpPipelinePackageFabric npPipelinePackageManager, NpPipelinePackageFabric npPipelinePackageManager,
PipelinePackageOrchestrator pipelineOrchestrator, PipelinePackageOrchestrator pipelineOrchestrator,
CreditTrackerSettings creditTrackerSettings, CreditTrackerSettings creditTrackerSettings,
MonitoringErrorProcesingService monitoringErrorProcesingService,
Saver saver) { Saver saver) {
this.userPackageMapper = userPackageMapper; this.userPackageMapper = userPackageMapper;
this.phoneFidMapMapper = phoneFidMapMapper; this.phoneFidMapMapper = phoneFidMapMapper;
this.npPipelinePackageManager = npPipelinePackageManager; this.npPipelinePackageManager = npPipelinePackageManager;
this.pipelineOrchestrator = pipelineOrchestrator; this.pipelineOrchestrator = pipelineOrchestrator;
this.creditTrackerSettings = creditTrackerSettings; this.creditTrackerSettings = creditTrackerSettings;
this.monitoringErrorProcesingService = monitoringErrorProcesingService;
this.saver = saver; this.saver = saver;
} }
public void run() { public void run() {
onMonitoringRestore();
acceptedRestore(); acceptedRestore();
processingRestore(); processingRestore();
} }
private void onMonitoringRestore() {
List<UserPackage> packages = userPackageMapper.findOnMonitoringPackages();
packages.forEach(userPackage -> {
monitoringErrorProcesingService.findErrors(userPackage.getId());
});
}
private void acceptedRestore() { private void acceptedRestore() {
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name()); List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name());
log.debug("Found {} packages in status ACCEPTED", packages.size()); log.debug("Found {} packages in status ACCEPTED", packages.size());

21
src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java → src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java
View file

@ -21,19 +21,19 @@ import java.util.concurrent.CompletableFuture;
@Slf4j @Slf4j
@Service @Service
@EnableScheduling @EnableScheduling
public class UpdatingFidAndPdService { public class UpdatingFidService {
private final NpPipelinePackageFabric npPipelinePackageFabric; private final NpPipelinePackageFabric npPipelinePackageFabric;
private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator;
private final PipelinePackageOrchestrator pipelinePackageOrchestrator; private final PipelinePackageOrchestrator pipelinePackageOrchestrator;
private final UserPackageMapper userPackageMapper; private final UserPackageMapper userPackageMapper;
private final PhonesMapService phonesMapService; private final PhonesMapService phonesMapService;
public UpdatingFidAndPdService(CreditTrackerSettings config, public UpdatingFidService(CreditTrackerSettings config,
NpPipelinePackageFabric npPipelinePackageFabric, NpPipelinePackageFabric npPipelinePackageFabric,
PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator,
PipelinePackageOrchestrator pipelinePackageOrchestrator, PipelinePackageOrchestrator pipelinePackageOrchestrator,
UserPackageMapper userPackageMapper, UserPackageMapper userPackageMapper,
PhonesMapService phonesMapService) { PhonesMapService phonesMapService) {
this.npPipelinePackageFabric = npPipelinePackageFabric; this.npPipelinePackageFabric = npPipelinePackageFabric;
this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator;
this.pipelinePackageOrchestrator = pipelinePackageOrchestrator; this.pipelinePackageOrchestrator = pipelinePackageOrchestrator;
@ -42,7 +42,7 @@ public class UpdatingFidAndPdService {
} }
@Scheduled(cron = "${ct.update-fid-cron-expression}") @Scheduled(cron = "${ct.update-fid-cron-expression}")
public void update() { public void updateNp() {
if (!pipelineRecalcPDOrchestrator.checkPermits()) { if (!pipelineRecalcPDOrchestrator.checkPermits()) {
log.info("The previous task has not yet been completed."); log.info("The previous task has not yet been completed.");
return; return;
@ -81,4 +81,9 @@ public class UpdatingFidAndPdService {
log.error("Failed to build pipeline", e); log.error("Failed to build pipeline", e);
} }
} }
@Scheduled(cron = "${ct.update-fid-cron-expression}")
public void updateLp() {
// TODO
}
} }

54
src/main/java/ru/nbch/credit_tracker/service/mode/migration/CTMigrationApplication.java
View file

@ -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();
}
}

62
src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java
View file

@ -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);
}
}

424
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/MonitoringService.java
View file

@ -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<PackageMessage> 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<PackageMessage> messages) {
for (PackageMessage message : messages) {
Integer fid = message.getFid();
List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFid(fid);
for (PhonesMap phonesMap : phonesMaps) {
if (phonesMap != null) {
Long clientId = phonesMap.getPhone();
Integer flagDebt5000 = phonesMap.getFlagDebt5000();
Integer flag90days = phonesMap.getFlag90days();
List<PackagesRecordsMap> 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<PackageMessage> messages) {
List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
List<String> memberCodes = messages.stream()
.map(message -> message.getMemberCode().substring(0,6))
.toList();
List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
// key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров)
Map<Integer, List<PhonesMap>> phonesDataMap = phonesMaps.stream()
.collect(Collectors.groupingBy(PhonesMap::getFid));
Set<Long> uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet());
// прелоад по memberCode категорий
Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
// subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
List<UserPackagePhoneData> packagePhoneData = packageRecordsService
.selectUserPackagePhoneData(uniquePhones);
// 1 номер может быть у нескольких пакетов
Map<Long, List<UserPackagePhoneData>> 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<PhonesMap> phonesMapList = phonesDataMap.get(fid);
if (phonesMapList == null) {
continue;
}
for (PhonesMap phonesMap : phonesMapList) {
// флаги просрочек
Integer flagDebt5000 = phonesMap.getFlagDebt5000();
Integer flag90days = phonesMap.getFlag90days();
// в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
List<UserPackagePhoneData> 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<UserPackage> 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<String> 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<Path> 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<Path> 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<Subject> 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<Subject> processPersons(List<Person> persons, UserPackage userPackage) {
List<Subject> subjects = new ArrayList<>();
Map<Long, SubjectRepInfo> 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<Hit> 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;
}
}
}

24
src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java
View file

@ -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.LegalsMapMapper;
import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; 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.*;
import java.util.function.Function; import java.util.function.Function;
@ -61,10 +64,6 @@ public class PackageRecordsService {
.map(phoneMap -> new HashMap<String, Object>() {{ .map(phoneMap -> new HashMap<String, Object>() {{
put("fid", phoneMap.getFid()); put("fid", phoneMap.getFid());
put("phone", phoneMap.getPhone()); put("phone", phoneMap.getPhone());
put("name_1", phoneMap.getName1());
put("first", phoneMap.getFirst());
put("middle", phoneMap.getMiddle());
put("birth_dt", phoneMap.getBirthDt());
}}).toList()); }}).toList());
phonesMapMapper.registerJson(json); phonesMapMapper.registerJson(json);
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
@ -98,21 +97,6 @@ public class PackageRecordsService {
} }
} }
public void updateIdCache(List<IdData> idData) {
try {
String json = entityToJsonMapper.writeValueAsString(idData.stream()
.filter(Objects::nonNull)
.map(phoneMap -> new HashMap<String, Object>() {{
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> flagData) { public void updateNpFlagCache(List<FlagData> flagData) {
try { try {
String json = entityToJsonMapper.writeValueAsString(flagData.stream() String json = entityToJsonMapper.writeValueAsString(flagData.stream()

15
src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java
View file

@ -2,13 +2,14 @@ package ru.nbch.credit_tracker.service.signal;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.SubjectProfile;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
@Service @Service
public class PhonesMapService { public class PhonesMapService {
private final PhonesMapMapper phonesMapMapper; private final PhonesMapMapper phonesMapMapper;
@ -20,7 +21,7 @@ public class PhonesMapService {
this.entityToJsonMapper = entityToJsonMapper; this.entityToJsonMapper = entityToJsonMapper;
} }
public void resetFidAndPDByPhones(List<Long> phones) { public void resetFidAndFlagsByPhones(List<Long> phones) {
try { try {
String json = entityToJsonMapper.writeValueAsString(phones.stream() String json = entityToJsonMapper.writeValueAsString(phones.stream()
.filter(Objects::nonNull) .filter(Objects::nonNull)
@ -46,16 +47,12 @@ public class PhonesMapService {
} }
} }
public void updateNameCachedData(List<SubjectProfile> subjectProfiles) { public void updateFidPhoneCachedData(List<SubjectProfile> subjectProfiles) {
try { try {
String json = entityToJsonMapper.writeValueAsString(subjectProfiles.stream() String json = entityToJsonMapper.writeValueAsString(subjectProfiles.stream()
.filter(Objects::nonNull) .filter(Objects::nonNull)
.map(subjectProfile -> new HashMap<String, Object>() {{ .map(subjectProfile -> new HashMap<String, Object>() {{
put("fid", subjectProfile.getFid()); 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()); put("phone", subjectProfile.getPhone());
}}).toList()); }}).toList());
phonesMapMapper.updateNameCachedData(json); phonesMapMapper.updateNameCachedData(json);

336
src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java
View file

@ -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<String, Resource> body = new LinkedMultiValueMap<>();
body.add("file", new FileUrlResource(gzip.toString()));
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.MULTIPART_FORM_DATA);
headers.setBasicAuth(authToken);
HttpEntity<MultiValueMap<String, Resource>> requestEntity = new HttpEntity<>(body, headers);
ResponseEntity<String> 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<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(headers);
ResponseEntity<String> 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 путь к разархивированному файлу. Пример: <loadDir>/tmp_folder/<package_id>/0001XX_MY_BATCH_223112.1.errors.xml.gz
*/
public Optional<Path> loadErrorFile(String fileName, UserPackage userPackage) {
String url = DOWNLOAD_REPORT_URI + "?fNameRequest=" + fileName;
HttpHeaders headers = new HttpHeaders();
headers.setBasicAuth(authToken);
HttpEntity<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(headers);
ResponseEntity<byte[]> 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<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(headers);
ResponseEntity<String> 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<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(headers);
String packCode = username.substring(0, 6) + "_" + packageName;
ResponseEntity<String> 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<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(headers);
ResponseEntity<String> 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 путь к разархивированному файлу. Пример: <loadDir>/tmp_folder/<package_id>/0001XX_MY_BATCH_204.1.online.2025-07-08T20-00-46.xml.gz.xml
*/
public Optional<Path> downloadReport(UserPackage userPackage, String reportName) throws IOException {
String url = DOWNLOAD_ONLINE_REPORT_URI + "?fNameReport=" + reportName;
HttpHeaders headers = new HttpHeaders();
headers.setBasicAuth(authToken);
HttpEntity<MultiValueMap<String, String>> requestEntity = new HttpEntity<>(headers);
ResponseEntity<byte[]> 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("<?xml");
}
private boolean isError(SgnlReport response) {
return response.getError() != null
&& (!StringUtils.isEmpty(response.getError().getDescription()) || !StringUtils.isEmpty(response.getError().getName()));
}
private String errorDescription(SgnlReport response) {
if (response.getError() == null)
return "unknown error";
return "%s: %s".formatted(response.getError().getName(), response.getError().getDescription());
}
}

68
src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageFabric.java
View file

@ -7,7 +7,8 @@ import org.springframework.stereotype.Component;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.config.task.StageQueueFactory; import ru.nbch.credit_tracker.config.task.StageQueueFactory;
import ru.nbch.credit_tracker.model.impl.*; import ru.nbch.credit_tracker.model.impl.*;
import ru.nbch.credit_tracker.service.task.impl.*; import ru.nbch.credit_tracker.service.task.impl.NpFidEnricher;
import ru.nbch.credit_tracker.service.task.impl.NpFidUpdater;
import ru.nbch.pipeline.engine.core.Pipeline; import ru.nbch.pipeline.engine.core.Pipeline;
import ru.nbch.pipeline.engine.core.PipelineBackoff; import ru.nbch.pipeline.engine.core.PipelineBackoff;
import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
@ -20,63 +21,45 @@ import java.util.function.Supplier;
@Component @Component
@Slf4j @Slf4j
public class NpPipelinePackageFabric implements InitializingBean { public class NpPipelinePackageFabric implements InitializingBean {
private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
private final StageQueueFactory<SubjectPayload> idUpdaterQueueFactory;
private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory; private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
private final StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory; private final StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory;
private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory; private final StageQueueFactory<SubjectPayload> sendSignalQueueFactory;
private final StageQueueFactory<VoidPayload> noOpQueueFactory; private final StageQueueFactory<VoidPayload> noOpQueueFactory;
private final FidIdEnricher fidIdEnricherHandler; private final NpFidEnricher npFidEnricher;
private final FidIdUpdater fidIdUpdater; private final NpFidUpdater npFidUpdater;
private final FidNameEnricher fidNameEnricher; private final Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory;
private final FidNameUpdater fidNameUpdater;
private final FidNameUpdaterMigration fidNameUpdaterMigration;
private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory; private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory;
private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory; private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final CreditTrackerSettings creditTrackerSettings; private final CreditTrackerSettings creditTrackerSettings;
private final PipelineBackoff pipelineBackoff; private final PipelineBackoff pipelineBackoff;
public NpPipelinePackageFabric(@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory, public NpPipelinePackageFabric(@Qualifier("nameEnricherQueueFactory") StageQueueFactory<SubjectPayload> nameEnricherQueueFactory,
@Qualifier("idUpdaterQueueFactory") StageQueueFactory<SubjectPayload> idUpdaterQueueFactory,
@Qualifier("nameEnricherQueueFactory") StageQueueFactory<SubjectPayload> nameEnricherQueueFactory,
@Qualifier("nameUpdaterQueueFactory") StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory, @Qualifier("nameUpdaterQueueFactory") StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory,
@Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory, @Qualifier("partFileQueueFactory") StageQueueFactory<SubjectPayload> sendSignalQueueFactory,
@Qualifier("noOpQueueFactory") StageQueueFactory<VoidPayload> noOpQueueFactory, @Qualifier("noOpQueueFactory") StageQueueFactory<VoidPayload> noOpQueueFactory,
FidIdEnricher fidIdEnricherHandler, NpFidUpdater npFidUpdater,
FidIdUpdater fidIdUpdater, Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory,
FidNameUpdater fidNameUpdater,
FidNameUpdaterMigration fidNameUpdaterMigration,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory, Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory, Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
FidNameEnricher fidNameEnricher, NpFidEnricher npFidEnricher,
CreditTrackerSettings creditTrackerSettings, CreditTrackerSettings creditTrackerSettings,
PipelineBackoff pipelineBackoff) { PipelineBackoff pipelineBackoff) {
this.idEnricherQueueFactory = idEnricherQueueFactory;
this.idUpdaterQueueFactory = idUpdaterQueueFactory;
this.nameEnricherQueueFactory = nameEnricherQueueFactory; this.nameEnricherQueueFactory = nameEnricherQueueFactory;
this.nameUpdaterQueueFactory = nameUpdaterQueueFactory; this.nameUpdaterQueueFactory = nameUpdaterQueueFactory;
this.sendSignalQueueFactory = sendSignalQueueFactory; this.sendSignalQueueFactory = sendSignalQueueFactory;
this.noOpQueueFactory = noOpQueueFactory; this.noOpQueueFactory = noOpQueueFactory;
this.fidIdEnricherHandler = fidIdEnricherHandler; this.npFidUpdater = npFidUpdater;
this.fidIdUpdater = fidIdUpdater;
this.fidNameUpdater = fidNameUpdater;
this.fidNameUpdaterMigration = fidNameUpdaterMigration;
this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.pipelineTerminatorFactory = pipelineTerminatorFactory;
this.noOpTerminatorFactory = noOpTerminatorFactory; this.noOpTerminatorFactory = noOpTerminatorFactory;
this.pipelineStarterFactory = pipelineStarterFactory; this.pipelineStarterFactory = pipelineStarterFactory;
this.fidNameEnricher = fidNameEnricher; this.npFidEnricher = npFidEnricher;
this.creditTrackerSettings = creditTrackerSettings; this.creditTrackerSettings = creditTrackerSettings;
this.pipelineBackoff = pipelineBackoff; this.pipelineBackoff = pipelineBackoff;
} }
public Pipeline build(SetupParam setupParam) throws Exception { public Pipeline build(SetupParam setupParam) throws Exception {
return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), sendSignalQueueFactory.getObject())
.handler(fidNameEnricher) .handler(npFidEnricher)
.andThen()
.stage(sendSignalQueueFactory.getObject())
.handler(fidIdEnricherHandler)
.andThen() .andThen()
.fillInitialQueue(pipelineStarterFactory.apply(setupParam)) .fillInitialQueue(pipelineStarterFactory.apply(setupParam))
.pollFinishingQueue(pipelineTerminatorFactory.get()) .pollFinishingQueue(pipelineTerminatorFactory.get())
@ -86,26 +69,9 @@ public class NpPipelinePackageFabric implements InitializingBean {
} }
public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception { public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception {
return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), noOpQueueFactory.getObject())
.handler(fidNameUpdater) .handler(npFidUpdater)
.andThen() .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)) .fillInitialQueue(pipelineStarterFactory.apply(setupParam))
.pollFinishingQueue(noOpTerminatorFactory.get()) .pollFinishingQueue(noOpTerminatorFactory.get())
.backoff(pipelineBackoff) .backoff(pipelineBackoff)

75
src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java
View file

@ -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<SubjectPayload> idUpdaterQueueFactory;
private final StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory;
private final StageQueueFactory<VoidPayload> noOpQueueFactory;
private final FidIdUpdater fidIdUpdater;
private final FidNameUpdaterMigration fidNameUpdaterMigration;
private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory;
private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final CreditTrackerSettings creditTrackerSettings;
private final PipelineBackoff pipelineBackoff;
public PipelineMigrationPackageFabric(@Qualifier("idUpdaterQueueFactory") StageQueueFactory<SubjectPayload> idUpdaterQueueFactory,
@Qualifier("nameUpdaterQueueFactory") StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory,
@Qualifier("noOpQueueFactory") StageQueueFactory<VoidPayload> noOpQueueFactory,
FidIdUpdater fidIdUpdater,
FidNameUpdaterMigration fidNameUpdaterMigration,
Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
Function<SetupParam, IPipeLineStarter<SubjectPayload>> 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());
}
}

37
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java
View file

@ -4,14 +4,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; 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.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SignalPayload;
import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode; 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.credit_tracker.utils.error.ExceptionUtils;
import ru.nbch.pipeline.engine.core.Pipeline; import ru.nbch.pipeline.engine.core.Pipeline;
@ -27,25 +24,20 @@ public class PipelinePackageOrchestrator {
private final Logger log = LoggerFactory.getLogger(getClass()); private final Logger log = LoggerFactory.getLogger(getClass());
private final CreditTrackerSettings creditTrackerSettings; private final CreditTrackerSettings creditTrackerSettings;
private final UserPackageMapper userPackageMapper; private final UserPackageMapper userPackageMapper;
private final SignalRestService signalRestService;
private final ExecutorService pipelineExecutorService; private final ExecutorService pipelineExecutorService;
private final ExecutorService saverExecutorService; private final ExecutorService saverExecutorService;
private final Object lock = new Object(); private final Object lock = new Object();
private final java.util.Deque<Launch> pending = new java.util.ArrayDeque<>(); private final java.util.Deque<Launch> pending = new java.util.ArrayDeque<>();
private final MonitoringErrorProcesingService monitoringErrorProcesingService;
private final Semaphore semaphore; private final Semaphore semaphore;
public PipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, public PipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
UserPackageMapper userPackageMapper, UserPackageMapper userPackageMapper,
SignalRestService signalRestService,
ExecutorService pipelineExecutorService, ExecutorService pipelineExecutorService,
ExecutorService saverExecutorService, MonitoringErrorProcesingService monitoringErrorProcesingService) { ExecutorService saverExecutorService) {
this.creditTrackerSettings = creditTrackerSettings; this.creditTrackerSettings = creditTrackerSettings;
this.userPackageMapper = userPackageMapper; this.userPackageMapper = userPackageMapper;
this.signalRestService = signalRestService;
this.pipelineExecutorService = pipelineExecutorService; this.pipelineExecutorService = pipelineExecutorService;
this.saverExecutorService = saverExecutorService; this.saverExecutorService = saverExecutorService;
this.monitoringErrorProcesingService = monitoringErrorProcesingService;
this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
} }
@ -103,7 +95,7 @@ public class PipelinePackageOrchestrator {
try { try {
long finishTime = System.currentTimeMillis(); long finishTime = System.currentTimeMillis();
log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime()); log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime());
SignalXmlFilePayload res = (SignalXmlFilePayload) v; SignalPayload res = (SignalPayload) v;
var ctx = pipeline.obtainContext(PackageVariable.class); var ctx = pipeline.obtainContext(PackageVariable.class);
AtomicInteger cacheHits = ctx.get(PackageVariable.cacheHits); AtomicInteger cacheHits = ctx.get(PackageVariable.cacheHits);
AtomicInteger cacheMisses = ctx.get(PackageVariable.cacheMisses); AtomicInteger cacheMisses = ctx.get(PackageVariable.cacheMisses);
@ -112,31 +104,24 @@ public class PipelinePackageOrchestrator {
cacheHits.get(), cacheHits.get(),
cacheMisses.get(), cacheMisses.get(),
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."); log.info("when complete: terminator was interrupted.");
return; return;
} else if (e != null) { } else if (e != null) {
log.error(ExceptionUtils.getStackTrace(e)); log.error(ExceptionUtils.getStackTrace(e));
return; return;
} }
if (res.signalXmlFilePath() == null) { if (!res.hasFids()) {
log.info("No one fid was found by packageId : {}; xml file won't be sent", packageId); log.info("No one fid was found by packageId : {}.", packageId);
userPackageMapper.updateNoFidMonitoring(packageId); userPackageMapper.updateNoFidMonitoring(packageId);
return; return;
} }
log.trace("Signal file by path: {} is completed", res.signalXmlFilePath()); PipelineMode mode = ctx.get(PackageVariable.mode);
boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath()); if (mode != null && mode.equals(PipelineMode.DEFAULT)) {
if (success) { userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
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);
} else { } else {
userPackageMapper.setRejectedStatus(packageId, RejectedStatus.ERROR.getKey()); userPackageMapper.updatePackageOnMonitoringSuccess(packageId, null);
} }
} catch (Exception ex) { } catch (Exception ex) {
log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));

134
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java
View file

@ -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<SubjectPayload, PartFilePayload> {
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<SubjectProfile> subjectsProfile = taskPayload.getSubjects();
log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
List<Integer> 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<Integer, Id> idByFid;
if (!noIdDataFids.isEmpty()) {
idByFid = personService.filledPassportsData(noIdDataFids);
} else {
idByFid = new HashMap<>();
}
List<IdData> idCacheOnUpdate = new ArrayList<>();
List<Person> 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<Person> 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;
}
}

80
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java
View file

@ -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<SubjectPayload, VoidPayload> {
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<SubjectProfile> subjectsProfile = taskPayload.getSubjects();
log.debug("FidIdUpdater handle; size: {}", subjectsProfile.size());
List<Integer> noIdDataFids = subjectsProfile.stream()
.map(SubjectProfile::getFid)
.toList();
Map<Integer, Id> idByFid = personService.filledPassportsData(noIdDataFids);
List<IdData> 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;
}
}

75
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java
View file

@ -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<SubjectPayload, SubjectPayload> {
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<SubjectProfile> subjects = taskPayload.getSubjects();
var ctx = obtainContext(PackageVariable.class, taskPayload);
log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size());
Map<String, Name> nameByPhone = personService.findNameByFids(subjects);
List<SubjectProfile> changingFidPhones = new ArrayList<>();
List<Long> 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;
}
}

12
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java
View file

@ -8,8 +8,6 @@ import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.exceptions.SignalClientException;
import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; 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.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload;
@ -73,16 +71,6 @@ public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
.mapId(phoneFidMap.getId()) .mapId(phoneFidMap.getId())
.phone(phoneFidMap.getPhone()) .phone(phoneFidMap.getPhone())
.fid(phoneFidMap.getFid()) .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(); .build()).toList();
try { try {
SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles); SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles);

52
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java → src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java
View file

@ -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.app.PhonesMap;
import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.entities.indic.Name;
import ru.nbch.credit_tracker.model.FlagData; 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.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload;
@ -28,14 +26,14 @@ import java.util.stream.Stream;
*/ */
@Service @Service
@Slf4j @Slf4j
public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload> { public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
private final PersonService personService; private final PersonService personService;
private final CreditTrackerSettings creditTrackerSettings; private final CreditTrackerSettings creditTrackerSettings;
private final PackageRecordsService packageRecordsService; private final PackageRecordsService packageRecordsService;
public FidNameEnricher(PersonService personService, public NpFidEnricher(PersonService personService,
CreditTrackerSettings creditTrackerSettings, CreditTrackerSettings creditTrackerSettings,
PackageRecordsService packageRecordsService) { PackageRecordsService packageRecordsService) {
this.personService = personService; this.personService = personService;
this.creditTrackerSettings = creditTrackerSettings; this.creditTrackerSettings = creditTrackerSettings;
this.packageRecordsService = packageRecordsService; this.packageRecordsService = packageRecordsService;
@ -88,14 +86,10 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
.phone(phone) .phone(phone)
.build()); .build());
} }
fillByName(subject, name); subject.setFid(name.getFid());
return Stream.of(PhonesMap.builder() return Stream.of(PhonesMap.builder()
.fid(subject.getFid()) .fid(subject.getFid())
.phone(phone) .phone(phone)
.name1(name.getLast())
.first(name.getFirst())
.middle(name.getMiddle())
.birthDt(name.getBirthDate())
.build()); .build());
} }
return Stream.empty(); return Stream.empty();
@ -107,29 +101,8 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
} }
private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) { 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) Optional.of(cachedRecord)
.filter(cr -> cachedRecord.getSerNum() != null && cachedRecord.getIdType() != null) .ifPresent(cr -> subjectProfile.setFid(cr.getFid()));
.map(cr -> Optional.of(
(IdData.builder()
.serNum(cr.getSerNum())
.idType(cr.getIdType())
.build()
)))
.ifPresent(idData -> subjectProfile.setIdData(idData.get()));
Optional.of(cachedRecord) Optional.of(cachedRecord)
.filter(cr -> cachedRecord.getFlagDebt5000() != null && cachedRecord.getFlag90days() != null) .filter(cr -> cachedRecord.getFlagDebt5000() != null && cachedRecord.getFlag90days() != null)
@ -142,19 +115,6 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
.ifPresent(flagData -> subjectProfile.setFlagData(flagData.get())); .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 @Override
public int preferredConcurrency() { public int preferredConcurrency() {
return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency();

29
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java → src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java
View file

@ -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.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload; 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.UserPackageService;
import ru.nbch.credit_tracker.service.indic.PersonService; import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
@ -21,18 +22,18 @@ import java.util.stream.Stream;
@Service @Service
@Slf4j @Slf4j
public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> { public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
private final PersonService personService; private final PersonService personService;
private final CreditTrackerSettings creditTrackerSettings; private final CreditTrackerSettings creditTrackerSettings;
private final PackageRecordsService packageRecordsService; private final PackageRecordsService packageRecordsService;
private final PhonesMapService phonesMapService; private final PhonesMapService phonesMapService;
private final UserPackageService userPackageService; private final UserPackageService userPackageService;
public FidNameUpdater(PersonService personService, public NpFidUpdater(PersonService personService,
CreditTrackerSettings creditTrackerSettings, CreditTrackerSettings creditTrackerSettings,
PackageRecordsService packageRecordsService, PackageRecordsService packageRecordsService,
PhonesMapService phonesMapService, PhonesMapService phonesMapService,
UserPackageService userPackageService) { UserPackageService userPackageService) {
this.personService = personService; this.personService = personService;
this.creditTrackerSettings = creditTrackerSettings; this.creditTrackerSettings = creditTrackerSettings;
this.packageRecordsService = packageRecordsService; this.packageRecordsService = packageRecordsService;
@ -41,7 +42,7 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
} }
@Override @Override
public SubjectPayload handle(SubjectPayload taskPayload) { public VoidPayload handle(SubjectPayload taskPayload) {
List<SubjectProfile> subjects = taskPayload.getSubjects(); List<SubjectProfile> subjects = taskPayload.getSubjects();
var ctx = obtainContext(PackageVariable.class, taskPayload); var ctx = obtainContext(PackageVariable.class, taskPayload);
log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size()); log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size());
@ -54,10 +55,6 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
if (name != null) { if (name != null) {
if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) { if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) {
subjectProfile.setFid(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); changingFidPhones.add(subjectProfile);
return Stream.empty(); return Stream.empty();
} }
@ -69,12 +66,12 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
}).toList(); }).toList();
List<Long> changingFids = new ArrayList<>(); List<Long> changingFids = new ArrayList<>();
if (!notFoundFidPhones.isEmpty()) { if (!notFoundFidPhones.isEmpty()) {
phonesMapService.resetFidAndPDByPhones(notFoundFidPhones); phonesMapService.resetFidAndFlagsByPhones(notFoundFidPhones);
changingFids.addAll(notFoundFidPhones); changingFids.addAll(notFoundFidPhones);
} }
if (!changingFidPhones.isEmpty()) { if (!changingFidPhones.isEmpty()) {
phonesMapService.updateNameCachedData(changingFidPhones); phonesMapService.updateFidPhoneCachedData(changingFidPhones);
changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList()); changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList());
} }
@ -88,7 +85,7 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
userPackageService.setFidChanged(packageIds); userPackageService.setFidChanged(packageIds);
} }
} }
return new SubjectPayload(null, null, null, changingFidPhones); return new VoidPayload();
} }
@Override @Override
@ -97,7 +94,7 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
} }
@Override @Override
public SubjectPayload cookPoison() { public VoidPayload cookPoison() {
return SubjectPayload.POISON; return VoidPayload.POISON;
} }
} }

66
src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java Normal file
View file

@ -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<SubjectPayload, SignalPayload> {
private final CompletableFuture<SignalPayload> done = new CompletableFuture<>();
private final CreditTrackerSettings config;
private BlockingQueue<SubjectPayload> 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<SubjectPayload> mergerQueue) {
this.inQueue = mergerQueue;
}
@Override
public CompletableFuture<SignalPayload> doneFuture() {
return done;
}
}

179
src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
View file

@ -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<PartFilePayload, SignalXmlFilePayload> {
private final Charset encoding = Charset.forName("windows-1251");
private final CompletableFuture<SignalXmlFilePayload> done = new CompletableFuture<>();
private Path out;
private final CreditTrackerSettings config;
private BlockingQueue<PartFilePayload> 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(); // </Id>
w.writeStartElement("Auth");
w.writeStartElement("User");
w.writeCharacters(config.getSignalsuser());
w.writeEndElement(); // </User>
w.writeEndElement(); // </Auth>
// <Conditions><Persons>
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("</Persons>");
w.write("</Conditions>");
w.write("</Signals>");
}
}
/**
* Копирует всё содержимое внутри <Persons>...</Persons> входного файла
* в текущий 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<PartFilePayload> mergerQueue) {
this.inQueue = mergerQueue;
}
@Override
public CompletableFuture<SignalXmlFilePayload> doneFuture() {
return done;
}
}

38
src/main/resources/mybatis/mapper/indic/PersonMapper.xml
View file

@ -10,10 +10,6 @@
<result property="number" column="number"/> <result property="number" column="number"/>
<result property="fileSinceDate" column="file_since_dt_name"/> <result property="fileSinceDate" column="file_since_dt_name"/>
<result property="fileSinceDatePn" column="file_since_dt_pn"/> <result property="fileSinceDatePn" column="file_since_dt_pn"/>
<result property="first" column="first"/>
<result property="last" column="NAME_1"/>
<result property="middle" column="middle"/>
<result property="birthDate" column="birth_dt"/>
</resultMap> </resultMap>
<resultMap id="passportDataResultMap" type="ru.nbch.credit_tracker.entities.indic.Id"> <resultMap id="passportDataResultMap" type="ru.nbch.credit_tracker.entities.indic.Id">
@ -31,10 +27,6 @@
pn.number as number, pn.number as number,
pn.fid as fid, pn.fid as fid,
pn.FILE_SINCE_DT as file_since_dt_pn, 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 nf.file_since_dt as file_since_dt_name
FROM PHONE_NORM pn FROM PHONE_NORM pn
JOIN NAME nf JOIN NAME nf
@ -46,34 +38,4 @@
WITH UR WITH UR
</select> </select>
<select id="findNameDataByFids" resultMap="nameDataResultMap">
SELECT
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
ON nf.fid = pn.fid
WHERE pn.fid in
<foreach item="fid" collection="fids" open="(" separator="," close=")">
#{fid}
</foreach>
WITH UR
</select>
<select id="findPassportData" resultMap="passportDataResultMap" resultType="java.util.List">
SELECT FID, SER_NUM, ID_NUM, ISS_DTE, FILE_SINCE_DT, ID_TYPE, ID_TYPE2
FROM ID
WHERE FID IN
<foreach item="fid" collection="fids" open="(" separator="," close=")">
#{fid}
</foreach>
WITH UR
</select>
</mapper> </mapper>

10
src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml
View file

@ -104,16 +104,6 @@
RETURNING id; RETURNING id;
</select> </select>
<update id="setErrorJson" parameterType="string">
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;
</update>
<delete id="deleteByPackageId" parameterType="long"> <delete id="deleteByPackageId" parameterType="long">
DELETE DELETE
FROM signals_phone_fid_map FROM signals_phone_fid_map

107
src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml
View file

@ -9,12 +9,6 @@
<result property="id" column="id"/> <result property="id" column="id"/>
<result property="fid" column="fid"/> <result property="fid" column="fid"/>
<result property="phone" column="phone"/> <result property="phone" column="phone"/>
<result property="name1" column="name_1"/>
<result property="first" column="first"/>
<result property="middle" column="middle"/>
<result property="birthDt" column="birth_dt"/>
<result property="serNum" column="ser_num"/>
<result property="idType" column="id_type"/>
<result property="flagDebt5000" column="flag_debt5000"/> <result property="flagDebt5000" column="flag_debt5000"/>
<result property="flag90days" column="flag_90_12"/> <result property="flag90days" column="flag_90_12"/>
<result property="searchedAt" column="searched_at"/> <result property="searchedAt" column="searched_at"/>
@ -25,12 +19,6 @@
select id, select id,
fid, fid,
phone, phone,
name_1,
first,
middle,
birth_dt,
ser_num,
id_type,
flag_debt5000, flag_debt5000,
flag_90_12, flag_90_12,
searched_at, searched_at,
@ -43,12 +31,6 @@
select id, select id,
fid, fid,
phone, phone,
name_1,
first,
middle,
birth_dt,
ser_num,
id_type,
flag_debt5000, flag_debt5000,
flag_90_12, flag_90_12,
searched_at, searched_at,
@ -74,12 +56,6 @@
select id, select id,
fid, fid,
phone, phone,
name_1,
first,
middle,
birth_dt,
ser_num,
id_type,
flag_debt5000, flag_debt5000,
flag_90_12, flag_90_12,
searched_at, searched_at,
@ -93,12 +69,6 @@
<select id="getCachedData" resultMap="PhoneMap" resultType="java.util.List"> <select id="getCachedData" resultMap="PhoneMap" resultType="java.util.List">
SELECT fid, SELECT fid,
phone, phone,
name_1,
first,
middle,
birth_dt,
ser_num,
id_type,
flag_debt5000, flag_debt5000,
flag_90_12, flag_90_12,
searched_at, searched_at,
@ -114,38 +84,22 @@
<select id="registerJson" parameterType="string"> <select id="registerJson" parameterType="string">
WITH src AS (SELECT row_number() OVER () AS rn, WITH src AS (SELECT row_number() OVER () AS rn,
fid, fid,
phone, phone
name_1,
first,
middle,
birth_dt
FROM json_to_recordset(#{json}::json) AS t( FROM json_to_recordset(#{json}::json) AS t(
fid int, fid int,
phone bigint, phone bigint
name_1 varchar(2000),
first varchar(2000),
middle varchar(2000),
birth_dt timestamp
)) ))
INSERT INSERT
INTO signals_phones_map INTO signals_phones_map
(fid, phone, name_1, first, middle, birth_dt, searched_at) (fid, phone, searched_at)
SELECT fid, SELECT fid,
phone, phone,
name_1,
first,
middle,
birth_dt,
now() now()
FROM src FROM src
order by phone order by phone
ON CONFLICT (phone) DO UPDATE ON CONFLICT (phone) DO UPDATE
SET fid = EXCLUDED.fid, SET fid = EXCLUDED.fid,
phone = EXCLUDED.phone, phone = EXCLUDED.phone
name_1 = EXCLUDED.name_1,
first = EXCLUDED.first,
middle = EXCLUDED.middle,
birth_dt = EXCLUDED.birth_dt
</select> </select>
<select id="findEligibleForFlagsUpdate" resultMap="PhoneMap"> <select id="findEligibleForFlagsUpdate" resultMap="PhoneMap">
@ -169,45 +123,12 @@
limit #{batchSize} limit #{batchSize}
</select> </select>
<update id="updateIdCachedData" parameterType="string">
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);
</update>
<update id="updateNameCachedData" parameterType="string"> <update id="updateNameCachedData" parameterType="string">
WITH src AS (SELECT DISTINCT ON (phone) fid, WITH src AS (SELECT DISTINCT ON (phone) fid,
phone, phone
name_1,
first,
middle,
birth_dt
FROM json_to_recordset(#{json}::json) AS src( FROM json_to_recordset(#{json}::json) AS src(
fid int, fid int,
phone bigint, phone bigint
name_1 varchar(255),
first varchar(255),
middle varchar(255),
birth_dt timestamp
) )
order by phone), order by phone),
@ -219,18 +140,10 @@
update signals_phones_map dst update signals_phones_map dst
set fid = src.fid, set fid = src.fid,
name_1 = src.name_1,
first = src.first,
middle = src.middle,
birth_dt = src.birth_dt,
searched_at = now() searched_at = now()
FROM json_to_recordset(#{json}::json) AS src( FROM json_to_recordset(#{json}::json) AS src(
fid int, fid int,
phone bigint, phone bigint
name_1 varchar(255),
first varchar(255),
middle varchar(255),
birth_dt timestamp
) )
where dst.phone = src.phone where dst.phone = src.phone
AND dst.id in (select id from to_lock); AND dst.id in (select id from to_lock);
@ -269,12 +182,6 @@
update signals_phones_map dst update signals_phones_map dst
set fid = null, set fid = null,
name_1 = null,
first = null,
middle = null,
birth_dt = null,
ser_num = null,
id_type = null,
flag_debt5000 = null, flag_debt5000 = null,
flag_90_12 = null, flag_90_12 = null,
calculated_at = null, calculated_at = null,