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

http://jira.mfd.msk:8088/browse/BKI-3241

Browse source
This commit is contained in:
etreschenkov 2025-12-01 19:20:35 +03:00
parent 945073948f
commit a2ef593bc9
26 changed files with 298 additions and 244 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
credit-tracker.properties
View file

@ -66,7 +66,7 @@ ct.load-dir=D:\\folder_name\\signals
# min
ct.monitoring-interval=${MONITORING_INTERVAL:15}
# days
ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 22 * * *}
ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 0 22 * * *}
ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30}
ct.db-version=${DB_VERSION:1234}

4
src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java
View file

@ -13,10 +13,8 @@ public class CreditTrackerApplication {
ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args);
UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
// UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class);
// updateProcessingPackagesService.run();
updateProcessingPackagesService.run();
monitoringService.run();
// updatingFlagsService.run();
}
}

5
src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java
View file

@ -31,7 +31,6 @@ public class PersonFactoryBuilderV2 {
private SubjectProfile personData;
private Long uid;
public static PersonFactoryBuilderV2 builder() {
return new PersonFactoryBuilderV2();
@ -39,8 +38,6 @@ public class PersonFactoryBuilderV2 {
public PersonFactoryBuilderV2 personData(SubjectProfile personData) {
this.personData = personData;
//todo откуда брать этот id
// this.uid = personData.getIdentifier();
return this;
}
@ -64,7 +61,7 @@ public class PersonFactoryBuilderV2 {
person.setConsentOdt(ODT);
}
person.setIp(IP);
person.setUid(uid);
person.setUid(personData.getMapId());
person.setOksm(OKSM);
person.setConsentDate(DATE_FORMAT.format(LocalDate.now()));
person.setConsentPurpose(CONSENT_PURPOSE);

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

@ -5,13 +5,14 @@ import java.util.function.Supplier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.indic.PhoneService2;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRepeatedRestoreImpl;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRestoreImpl;
import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
@ -24,12 +25,14 @@ public class PipelineStarterConfig {
public Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml,
CreditTrackerSettings creditTrackerSettings,
UserPackageMapper userPackageMapper,
PhoneService2 phoneService2) {
PhoneFidMapMapper phoneFidMapMapper) {
return (setupParam) -> switch (setupParam.mode()) {
case DEFAULT ->
new KickPipeLineDefaultImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
new KickPipeLineDefaultImpl(xml, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings);
case RESTORE ->
new KickPipeLineRestoreImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
new KickPipeLineRestoreImpl(xml, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings);
case REPEATED_RESTORE ->
new KickPipeLineRepeatedRestoreImpl(xml, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings);
};
}

4
src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java
View file

@ -17,5 +17,7 @@ public class UserPackage {
private String status;
private LocalDateTime deactivatedAt;
private String lastDownloadedReport;
// private LocalDateTime monitoringAt;
private LocalDateTime monitoringAt;
private LocalDateTime lastSearchedAt;
private String rejectStatus;
}

17
src/main/java/ru/nbch/credit_tracker/enums/RejectedStatus.java Normal file
View file

@ -0,0 +1,17 @@
package ru.nbch.credit_tracker.enums;
public enum RejectedStatus {
OK("0"),
ERROR("1"),;
private final String key;
RejectedStatus(String key) {
this.key = key;
}
public String getKey() {
return key;
}
}

13
src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java
View file

@ -46,6 +46,7 @@ import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.credit_tracker.service.task.impl.Saver;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.credit_tracker.utils.CtFileUtils;
import ru.nbch.credit_tracker.utils.EnumUtil;
import ru.nbch.credit_tracker.utils.LogUtil;
import ru.nbch.credit_tracker.utils.ZipUtils;
@ -114,6 +115,9 @@ public class SignalFacadeV3 {
if (e.getCause() instanceof XMLStreamException) {
throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError);
}
if (e instanceof SignalClientException) {
throw e;
}
throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e);
} catch (IOException e) {
log.warn("Error deleting tmp file", e);
@ -138,10 +142,10 @@ public class SignalFacadeV3 {
}
// проверка на уникальность пакета
// if (signalService.isPackageNotUnique(userId, packageName)) {
// signalService.rejectPackage(userId, packageName);
// throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError);
// }
if (userPackageService.isPackageNotUnique(userId, packageName)) {
userPackageService.rejectPackage(userId, packageName);
throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError);
}
Long packageId = userPackageService.registerPackage(userId, packageName);
LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId));
@ -151,6 +155,7 @@ public class SignalFacadeV3 {
.runAsync(() -> {
try {
saver.readAndSave(packageId, parsedFilePath);
CtFileUtils.deleteQuietly(parsedFilePath);
} catch (Throwable t) {
log.error(ExceptionUtils.getStackTrace(t));
pipeline.close();

3
src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java
View file

@ -15,12 +15,15 @@ public interface PhoneFidMapMapper {
void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
List<Long> registerJson(@Param("json") String json);
void setFidJson(@Param("json") String json);
void setErrorJson(@Param("json") String json);
void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset);
List<PhoneFidMap> findNullFidPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset);
List<PhoneFidMap> findNotNullFidPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset);
List<PhoneFidMap> pageByPackageIdAndOlderThanN(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset, @Param("daysCount") Integer daysCount);
List<PhoneFidMap> pageByPackageIdAndNewerThanN(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset, @Param("daysCount") Integer daysCount);
void updatePhoneFidMaps(@Param("subjects") Collection<PhoneFidMap> subjects);
void updateOne(@Param("subject") PhoneFidMap subject);
void updateFid(@Param("subjects") Collection<PhoneFidMap> subjects);

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

@ -37,4 +37,6 @@ public interface UserPackageMapper {
List<UserPackage> findAllPackagesByUserId(@Param("userId") String userId);
List<UserPackage> findByMemberCodeAndPackageName(@Param("memberCode") String memberCode,
@Param("packageName") String packageName);
void setRejectedStatus(@Param("id") Long id, @Param("status") String status);
}

2
src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java
View file

@ -58,7 +58,7 @@ public class PhoneRepository {
birth_dt,
file_since_dt as file_since_dt_name
FROM NAME
WHERE fid in (:fid) WITH UR
WHERE fid in (:fids) WITH UR
""";
private final NamedParameterJdbcTemplate jdbcTemplate;

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

@ -1,13 +1,15 @@
package ru.nbch.credit_tracker.service;
import java.nio.file.Path;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.aop.framework.AopContext;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.RejectedStatus;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.utils.CtFileUtils;
@ -19,37 +21,39 @@ public class MonitoringErrorProcesingService {
private MonitoringErrorProcesingService self;
private final SignalService signalService;
private final UserPackageMapper userPackageMapper;
private final SignalRestService signalRestService;
public MonitoringErrorProcesingService(SignalService signalService, SignalRestService signalRestService) {
public MonitoringErrorProcesingService(SignalService signalService,
UserPackageMapper userPackageMapper,
SignalRestService signalRestService) {
this.signalService = signalService;
this.userPackageMapper = userPackageMapper;
this.signalRestService = signalRestService;
}
public void findErrors(Long packageId) {
UserPackage userPackage = signalService.findPackage(packageId);
getCurrentProxy().processPackage(userPackage, packageId);
String packCode = userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName();
log.info("Проверка на ошибки при постановке на мониторинг. packCode={}, packageId={}", packCode, packageId);
String fileName = signalRestService.getErrorFileName(packageId, packCode);
if (StringUtils.isEmpty(fileName)) {
userPackageMapper.setRejectedStatus(packageId, RejectedStatus.OK.getKey());
return;
}
Optional<Path> filePath = signalRestService.loadErrorFile(fileName, userPackage);
if (filePath.isPresent()) {
try {
signalService.updateSubjectsErrors(filePath.get(), packageId);
} catch (Exception e) {
log.error("Ошибка при записи сообщений об ошибках в сформированных субъектах для мониторинга. Package_id = {}", packageId, e);
}
CtFileUtils.deleteQuietly(filePath.get());
}
userPackageMapper.setRejectedStatus(packageId, RejectedStatus.ERROR.getKey());
}
@SetAndClearMdc
public void processPackage(UserPackage userPackage, Long packageId) {
if (userPackage != null) {
String packCode = userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName();
log.info("Проверка на ошибки при постановке на мониторинг. packCode={}, packageId={}", packCode, packageId);
String fileName = signalRestService.getErrorFileName(packageId, packCode);
if (StringUtils.isNotBlank(fileName)) {
String filePath = signalRestService.loadErrorFile(packageId, fileName);
if (StringUtils.isNotBlank(filePath)) {
try {
signalService.updateSubjectsErrors(filePath, packageId);
} catch (Exception e) {
log.error("Ошибка при записи сообщений об ошибках в сформированных субъектах для мониторинга. Package_id = {}", packageId, e);
}
CtFileUtils.deleteQuietly(Path.of(filePath));
}
}
}
}
private MonitoringErrorProcesingService getCurrentProxy() {
if (self == null) {

9
src/main/java/ru/nbch/credit_tracker/service/SignalService.java
View file

@ -288,9 +288,9 @@ public class SignalService {
return persons;
}
public void updateSubjectsErrors(String filePath, Long packageId) {
public void updateSubjectsErrors(Path filePath, Long packageId) {
log.info("Start updating subjects errors. PackageId: {}", packageId);
try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) {
try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(filePath), 8 * 1024 * 1024)) {
staxXmlProcessor.read(in, MonitoringReportErrors.class,
request -> {
if (!request.getPersons().isEmpty()) {
@ -301,13 +301,12 @@ public class SignalService {
.errorText(p.getValue())
.build();
}).toList();
phoneFidMapService.updatePhoneFidMapsErrors(subjects);
phoneService2.updateOnError(subjects);
log.trace("Updated subject's batch. Total count: {}. PackageId: {}", request.getPersons().size(), packageId);
request.getPersons().clear();
}
},
request -> request.getPersons().isEmpty()
,
request -> request.getPersons().isEmpty(),
batchSize, "/Errors/p");
} catch (IOException e) {
throw new RuntimeException("Error while parsing MonitoringReportErrors", e);

64
src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java
View file

@ -5,16 +5,13 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopContext;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
import ru.nbch.credit_tracker.service.task.SetupParam;
@ -32,34 +29,23 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
@EnableAspectJAutoProxy(exposeProxy = true)
public class UpdateProcessingPackagesService {
private UpdateProcessingPackagesService self;
private final SignalService signalService;
private final UserPackageMapper userPackageMapper;
private final PhoneFidMapMapper phoneFidMapMapper;
private final SignalRestService signalRestService;
private final MonitoringErrorProcesingService monitoringErrorProcesingService;
private final PipelinePackageManager pipelinePackageManager;
private final CreditTrackerSettings creditTrackerSettings;
private final Saver saver;
private final ExecutorService pipelineExecutorService;
private final ExecutorService saverExecutorService;
public UpdateProcessingPackagesService(SignalService signalService,
UserPackageMapper userPackageMapper,
public UpdateProcessingPackagesService(UserPackageMapper userPackageMapper,
PhoneFidMapMapper phoneFidMapMapper,
SignalRestService signalRestService,
MonitoringErrorProcesingService monitoringErrorProcesingService,
PipelinePackageManager pipelinePackageManager,
CreditTrackerSettings creditTrackerSettings,
Saver saver,
ExecutorService pipelineExecutorService,
ExecutorService saverExecutorService) {
this.signalService = signalService;
this.userPackageMapper = userPackageMapper;
this.phoneFidMapMapper = phoneFidMapMapper;
this.signalRestService = signalRestService;
this.monitoringErrorProcesingService = monitoringErrorProcesingService;
this.pipelinePackageManager = pipelinePackageManager;
this.creditTrackerSettings = creditTrackerSettings;
this.saver = saver;
@ -70,16 +56,9 @@ public class UpdateProcessingPackagesService {
public void run() {
acceptedRestore();
processingRestore();
// log.info("Start updating processing packages");
// List<UserPackage> packages = signalService.findPackagesOnProcessing();
// for (UserPackage userPackage : packages) {
// log.info("Updating flags for Package {}", userPackage.getPackageName());
// getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName());
// }
// log.info("Finish updating processing packages");
processingRepeatedRestore();
}
private void acceptedRestore() {
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name());
log.debug("Found {} packages in status ACCEPTED", packages.size());
@ -89,8 +68,6 @@ public class UpdateProcessingPackagesService {
log.trace("Deleted {} rows", count);
Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage);
log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
//todo ???запускает асинхронный процесс без ожидания, на старте мы должны восстановиться один раз и
// видимо дождаться выполнения, иначе может не хватить пула, памяти или чего еще
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT));
CompletableFuture
.runAsync(() -> {
@ -110,7 +87,8 @@ public class UpdateProcessingPackagesService {
private void processingRestore() {
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name());
log.debug("Found {} packages in status PROCESSING", packages.size());
packages = packages.stream().filter(userPackage -> userPackage.getRejectStatus() == null).toList();
log.debug("Found {} packages in status PROCESSING and rejected_status = null", packages.size());
for (UserPackage userPackage : packages) {
try {
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE));
@ -121,33 +99,17 @@ public class UpdateProcessingPackagesService {
}
}
@SetAndClearMdc(argIndex = 1)
public void process(Long packageId, String packageName) {
try {
SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, null);
if (updateResult.isNoFidsFound()) {
log.info("No fids found for PackageId: {} . Package status ON_MONITORING. No registration in signal service.", packageId);
signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
return;
private void processingRepeatedRestore() {
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING_REPEATED.name());
log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size());
for (UserPackage userPackage : packages) {
try {
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE));
pipeline.start(pipelineExecutorService);
} catch (Exception e) {
log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
}
if (signalRestService.setupMonitoring(packageId)) {
log.info("Package was successfully registered in signal service. PackageId: {}", packageId);
signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
monitoringErrorProcesingService.findErrors(packageId);
} else {
log.info("Package registration was rejected by signal service. PackageId: {}", packageId);
signalService.updatePackageStatus(packageId, Status.ERROR);
}
} catch (Exception e) {
signalService.updatePackageStatus(packageId, Status.ERROR);
log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e);
}
}
private UpdateProcessingPackagesService getCurrentProxy() {
if (self == null) {
self = (UpdateProcessingPackagesService) AopContext.currentProxy();
}
return self;
}
}

8
src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java
View file

@ -12,6 +12,7 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
/**
@ -28,17 +29,19 @@ public class UpdatingFlagsService {
private final MonitoringErrorProcesingService monitoringErrorProcesingService;
private final Long daysToUpdateFlags;
private final PhoneFidMapMapper phoneFidMapMapper;
private final UserPackageMapper userPackageMapper;
public UpdatingFlagsService(SignalService signalService,
SignalRestService signalRestService,
MonitoringErrorProcesingService monitoringErrorProcesingService,
CreditTrackerSettings config,
PhoneFidMapMapper phoneFidMapMapper) {
PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper) {
this.signalService = signalService;
this.signalRestService = signalRestService;
this.monitoringErrorProcesingService = monitoringErrorProcesingService;
this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1;
this.phoneFidMapMapper = phoneFidMapMapper;
this.userPackageMapper = userPackageMapper;
}
@Scheduled(cron = "${ct.update-flags-cron-expression}")
@ -62,7 +65,8 @@ public class UpdatingFlagsService {
Long packageId = userPackage.getId();
String packageName = userPackage.getPackageName();
try {
signalService.updatePackageStatus(packageId, Status.PROCESSING);
//todo set reject_status = null
userPackageMapper.updatePackageStatus(packageId, Status.PROCESSING_REPEATED.name());
SignalService.UpdateResult updateResult = signalService.updateFlags2(userPackage);
if (updateResult.isUpdated()) {
if (signalRestService.setupMonitoring(packageId)) {

48
src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java
View file

@ -13,6 +13,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import ru.nbch.credit_tracker.entities.indic.Name;
import ru.nbch.credit_tracker.mapper.signal.FlagDetailMapper;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.repository.PhoneRepository;
@ -24,6 +25,7 @@ public class PhoneService2 {
private final PhoneRepository phoneRepository;
private final PhoneFidMapMapper phoneFidMapMapper;
private final FlagDetailMapper flagDetailMapper;
private final ObjectMapper entityToJsonMapper;
private final Comparator<Name> byDateThenIdNullSafe =
Comparator.comparing(Name::getFileSinceDate, Comparator.nullsFirst(Comparator.naturalOrder()))
@ -31,9 +33,11 @@ public class PhoneService2 {
public PhoneService2(PhoneRepository phoneRepository,
PhoneFidMapMapper phoneFidMapMapper,
FlagDetailMapper flagDetailMapper,
ObjectMapper entityToJsonMapper) {
this.phoneRepository = phoneRepository;
this.phoneFidMapMapper = phoneFidMapMapper;
this.flagDetailMapper = flagDetailMapper;
this.entityToJsonMapper = entityToJsonMapper;
}
@ -64,6 +68,7 @@ public class PhoneService2 {
* Если пакет ACCEPTED, то в базе могут находиться записи с уже рассчитанными FID
* Нужен для исключения долгих запросов findNameByPhones
* Смотри UpdateProcessingPackagesService:processingRestore
*
* @param subjects
* @return - отображение NAME к FID, отфильтрованному по логике
* FID с максимальным значением FILE_SINCE_DT. Если FILE_SINCE_DT оказался одинаковый, берется максимальный FID
@ -85,7 +90,6 @@ public class PhoneService2 {
public List<Long> registerSubjects(List<PhoneFidMap> subjects) {
try {
//fixme, временное решение
String json = entityToJsonMapper.writeValueAsString(subjects.stream()
.map(phoneFidMap -> new HashMap<String, Object>() {{
put("subject_id", phoneFidMap.getSubjectId());
@ -102,12 +106,12 @@ public class PhoneService2 {
public void updateFid(List<PhoneFidMap> subjects) {
try {
//fixme, временное решение
//регистрация в signals_phone_fid_map
String json = entityToJsonMapper.writeValueAsString(subjects.stream()
.map(phoneFidMap -> new HashMap<String, Object>() {{
put("id", phoneFidMap.getId());
put("fid", phoneFidMap.getFid());
put("calculated_at", phoneFidMap.getCalculatedAt());
put("searched_at", phoneFidMap.getCalculatedAt());
}}).toList());
phoneFidMapMapper.setFidJson(json);
} catch (JsonProcessingException e) {
@ -115,16 +119,38 @@ public class PhoneService2 {
}
}
public void registerSignals(List<PhoneFidMap> subjects) {
try {
//регистрация в signals_flags - апдейтим при FlagsEnricher
String jsonFlagDetail = entityToJsonMapper.writeValueAsString(subjects.stream()
.filter(phoneFidMap -> phoneFidMap.getFid() != null)
.map(subject -> new HashMap<String, Object>() {{
put("subject_id", subject.getSubjectId());
put("fid", subject.getFid());
put("phone", subject.getPhone());
put("map_id", subject.getId());
}}).toList());
flagDetailMapper.updateCalcFlags(jsonFlagDetail);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
public void updateOnError(List<PhoneFidMap> 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);
}
}
public List<PhoneFidMap> pageByPackageId(Long packageId, Integer limit, Integer offset) {
return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset);
}
public List<PhoneFidMap> pageByPackageIdWithNullFids(Long packageId, Integer limit, Integer offset) {
return phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset);
}
public List<PhoneFidMap> pageByPackageIdWithNotNullFids(Long packageId, Integer limit, Integer offset) {
return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset);
}
}

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

@ -1,19 +1,12 @@
package ru.nbch.credit_tracker.service.signal;
import jakarta.xml.bind.DatatypeConverter;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.util.Base64;
import java.util.Optional;
import java.util.UUID;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
@ -38,7 +31,6 @@ import ru.nbch.credit_tracker.entities.external_response.online_report.SgnlOnlin
import ru.nbch.credit_tracker.service.SignalService;
import ru.nbch.credit_tracker.service.xml.jaxb.IXmlProcessor;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.credit_tracker.utils.ZipUtils;
@Slf4j
@ -161,7 +153,7 @@ public class SignalRestService {
*
* @return путь к разархивированному файлу. Пример: <loadDir>/tmp_folder/<package_id>/0001XX_MY_BATCH_223112.1.errors.xml.gz
*/
public String loadErrorFile(Long packageId, String fileName) {
public Optional<Path> loadErrorFile(String fileName, UserPackage userPackage) {
String url = DOWNLOAD_REPORT_URI + "?fNameRequest=" + fileName;
HttpHeaders headers = new HttpHeaders();
@ -176,7 +168,17 @@ public class SignalRestService {
byte[].class
);
return unzipReport(response, packageId, fileName);
Path targetFilePathWithExt = null;
if (response.getHeaders().getContentType().equals(MEDIA_TYPE_GZIP)) {
targetFilePathWithExt = Path.of(config.getLoadDir(),
"tmp", userPackage.getUserId(), userPackage.getPackageName(), fileName.replace(".xml.gz",".xml"));
ZipUtils.unpackGzip(targetFilePathWithExt, response.getBody());
} else if (isXmlError(response.getBody())){
SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class);
String description = error.getError().getDescription() != null ? error.getError().getDescription() : "Неизвестная ошибка";
log.error("Ошибка при скачивание отчета. fNameReport={}. Описание: {}", fileName, description);
}
return Optional.ofNullable(targetFilePathWithExt);
}
/**
@ -247,18 +249,6 @@ public class SignalRestService {
return true;
}
private static boolean isError(SgnlReport response) {
return response.getError() != null
&& (!StringUtils.isEmpty(response.getError().getDescription()) || !StringUtils.isEmpty(response.getError().getName()));
}
private static String errorDescription(SgnlReport response) {
if (response.getError() == null)
return "unknown error";
return "%s: %s".formatted(response.getError().getName(), response.getError().getDescription());
}
/**
* @return Список отчетов, по которым сработали сигналы из официального сервиса сигналов
*/
@ -324,40 +314,6 @@ public class SignalRestService {
return Optional.ofNullable(targetFilePathWithExt);
}
private String unzipReport(ResponseEntity<byte[]> response, Long packageId, String reportName) {
boolean isGzipped = response.getHeaders().getContentType() != null
&& response.getHeaders().getContentType().toString().equals("application/x-gzip");
if (!isGzipped && isXmlError(response.getBody())) {
SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class);
if (error.getError().getDescription() != null) {
log.error("Ошибка при скачивание отчета. fNameReport={}. Описание: {}", reportName, error.getError().getDescription());
return null;
}
log.error("Неизвестная ошибка при скачивание отчета. fNameReport={}", reportName);
return null;
} else {
String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator;
String filePath = folderPath + UUID.randomUUID() + reportName + ".xml";
CTUtil.createFolder(folderPath);
try (InputStream in = new GZIPInputStream(new ByteArrayInputStream(response.getBody()));
OutputStream out = Files.newOutputStream(Path.of(filePath))) {
byte[] buffer = new byte[8 * 1024 * 1024];
int bytesRead;
while ((bytesRead = in.read(buffer)) > 0) {
out.write(buffer, 0, bytesRead);
}
return filePath;
} catch (Exception e) {
log.error(e.getMessage(), e);
return null;
}
}
}
private String buildToken(String signalsToken, String signalsuser) throws Exception {
try {
MessageDigest md = MessageDigest.getInstance("MD5");
@ -371,25 +327,20 @@ public class SignalRestService {
}
}
public Resource compressFile(String path) throws Exception {
try (InputStream in = Files.newInputStream(Path.of(path));
OutputStream out = Files.newOutputStream(Path.of(path + ".gz"));
GZIPOutputStream gzipOut = new GZIPOutputStream(out)) {
byte[] buffer = new byte[8 * 1024 * 1024];
int bytesRead;
while ((bytesRead = in.read(buffer)) > 0) {
gzipOut.write(buffer, 0, bytesRead);
}
gzipOut.finish();
return new FileUrlResource(path + ".gz");
}
}
private boolean isXmlError(byte[] data) {
if (data == null || data.length < 10) return false;
String start = new String(data, 0, 10).trim();
return start.startsWith("<?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());
}
}

7
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -15,6 +15,7 @@ import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher;
import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher;
@ -42,6 +43,7 @@ public class PipelinePackageManager implements InitializingBean {
private final CreditTrackerSettings creditTrackerSettings;
private final Semaphore semaphore;
private final PipelineBackoff pipelineBackoff;
private final MonitoringErrorProcesingService monitoringErrorProcesingService;
public PipelinePackageManager(SignalRestService signalRestService,
@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
@ -53,7 +55,8 @@ public class PipelinePackageManager implements InitializingBean {
Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher,
CreditTrackerSettings creditTrackerSettings,
PipelineBackoff pipelineBackoff, Saver saver) {
PipelineBackoff pipelineBackoff, Saver saver,
MonitoringErrorProcesingService monitoringErrorProcesingService) {
this.signalRestService = signalRestService;
this.idEnricherQueueFactory = idEnricherQueueFactory;
this.nameEnricherQueueFactory = nameEnricherQueueFactory;
@ -67,6 +70,7 @@ public class PipelinePackageManager implements InitializingBean {
this.creditTrackerSettings = creditTrackerSettings;
this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
this.pipelineBackoff = pipelineBackoff;
this.monitoringErrorProcesingService = monitoringErrorProcesingService;
}
public Pipeline buildAndRun(SetupParam setupParam) throws Exception {
@ -111,6 +115,7 @@ public class PipelinePackageManager implements InitializingBean {
boolean success = signalRestService.setupMonitoring(res.signalXmlFilePath());
if (success) {
userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
monitoringErrorProcesingService.findErrors(packageId);
}
} catch (Exception ex) {
log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));

1
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java
View file

@ -73,6 +73,7 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
}).toList();
phoneService2.updateFid(subjectForSave);
phoneService2.registerSignals(subjectForSave);
}
private void processByFid(List<SubjectProfile> subjects) {

46
src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java
View file

@ -59,37 +59,29 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
Map<Integer, Integer> flag90ByPastDueArrear = flagsService.findFlag90ByPastDueArrear(fids, yearAgo);
Map<Integer, Integer> flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(fids, yearAgo);
List<FlagDetail> flagDetails = subjects.stream().map(subject -> {
Integer debt5000 = null;
Integer flagDays90 = null;
LocalDateTime calculatedAt = null;
if (subject.getFid() != null) {
debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0);
List<FlagDetail> flagDetails = subjects.stream()
.filter(subjectProfile -> subjectProfile.getFid() != null)
.map(subject -> {
Integer debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0);
int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0);
int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0);
flagDays90 = Math.max(pastDue90, paymt90);
calculatedAt = LocalDateTime.now();
}
Integer flagDays90 = Math.max(pastDue90, paymt90);
Integer finalDebt500 = debt5000;
Integer finalFlagDays9 = flagDays90;
LocalDateTime finalCalculatedAt = calculatedAt;
return FlagDetail.builder()
.mapId(subject.getMapId())
.subjectId(subject.getIdentifier())
.packageId(subjectPayload.getPackageId())
.fid(subject.getFid())
.phone(subject.getPhone())
.flagDebt5000(debt5000)
.flag90days(flagDays90)
.calculatedAt(LocalDateTime.now())
.build();
}).toList();
return FlagDetail.builder()
.mapId(subject.getMapId())
.subjectId(subject.getIdentifier())
.packageId(subjectPayload.getPackageId())
.fid(subject.getFid())
.phone(subject.getPhone())
.flagDebt5000(finalDebt500)
.flag90days(finalFlagDays9)
.calculatedAt(finalCalculatedAt)
.build();
}).toList();
log.debug("updating calculated flags size {}", flagDetails.size());
flagsService.updateCalculationFlags(flagDetails);
if (update) {
log.debug("updating calculated flags size {}", flagDetails.size());
flagsService.updateCalculationFlags(flagDetails);
//если signals_phone_fid_map#fid == null, удалить signals_flags с таким map_id
List<Long> phoneFidMapIdsWithNullFid = subjects.stream()
.filter(subject -> subject.getFid() == null)
@ -100,8 +92,6 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
log.debug("deleting flags for signals_phone_fid_map#fid == null; amount: {}", phoneFidMapIdsWithNullFid.size());
flagsService.deleteFlags(phoneFidMapIdsWithNullFid);
}
} else {
flagsService.registerCalculationFlags(flagDetails);
}
}

10
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java
View file

@ -10,10 +10,10 @@ import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.exceptions.SignalClientException;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.indic.PhoneService2;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@ -25,7 +25,8 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
private final Path pkgPath;
private final StaxXmlProcessor xmlProcessor;
protected final UserPackageMapper userPackageMapper;
protected final PhoneService2 phoneService2;
protected final PhoneFidMapMapper phoneFidMapMapper;
protected final CreditTrackerSettings creditTrackerSettings;
private final int BATCH_SIZE;
private boolean isInterrupted = false;
protected int offset = 0;
@ -34,13 +35,14 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
public KickPipeLine(StaxXmlProcessor xmlProcessor,
SetupParam setupParam,
UserPackageMapper userPackageMapper,
PhoneService2 phoneService2,
PhoneFidMapMapper phoneFidMapMapper,
CreditTrackerSettings creditTrackerSettings) {
this.packageId = setupParam.packageId();
this.pkgPath = setupParam.xmlReqPath();
this.xmlProcessor = xmlProcessor;
this.userPackageMapper = userPackageMapper;
this.phoneService2 = phoneService2;
this.phoneFidMapMapper = phoneFidMapMapper;
this.creditTrackerSettings = creditTrackerSettings;
this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize();
}

8
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java
View file

@ -3,8 +3,8 @@ package ru.nbch.credit_tracker.service.task.impl;
import java.util.List;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.indic.PhoneService2;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@ -13,9 +13,9 @@ public class KickPipeLineDefaultImpl extends KickPipeLine {
public KickPipeLineDefaultImpl(StaxXmlProcessor xmlProcessor,
SetupParam setupParam,
UserPackageMapper userPackageMapper,
PhoneService2 phoneService2,
PhoneFidMapMapper phoneFidMapMapper,
CreditTrackerSettings creditTrackerSettings) {
super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
super(xmlProcessor, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings);
}
@Override
@ -25,7 +25,7 @@ public class KickPipeLineDefaultImpl extends KickPipeLine {
@Override
protected List<PhoneFidMap> nextPage(Integer limit) {
return phoneService2.pageByPackageId(packageId, limit, offset);
return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset);
}
}

43
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java Normal file
View file

@ -0,0 +1,43 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.util.ArrayList;
import java.util.List;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
public class KickPipeLineRepeatedRestoreImpl extends KickPipeLine {
private boolean oldRowsWereFound = false;
private final Integer daysToUpdateFlags;
public KickPipeLineRepeatedRestoreImpl(StaxXmlProcessor xmlProcessor,
SetupParam setupParam,
UserPackageMapper userPackageMapper,
PhoneFidMapMapper phoneFidMapMapper,
CreditTrackerSettings creditTrackerSettings) {
super(xmlProcessor, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings);
this.daysToUpdateFlags = creditTrackerSettings.getDaysToUpdateFlags().intValue();
}
@Override
protected PipelineMode mode() {
return PipelineMode.RESTORE;
}
@Override
protected List<PhoneFidMap> nextPage(Integer limit) {
List<PhoneFidMap> pageWithNullFids = new ArrayList<>();
if (!oldRowsWereFound) {
pageWithNullFids = phoneFidMapMapper.pageByPackageIdAndOlderThanN(packageId, limit, offset, daysToUpdateFlags);
}
if (pageWithNullFids.isEmpty()) {
oldRowsWereFound = true;
offset = 0;
return phoneFidMapMapper.pageByPackageIdAndNewerThanN(packageId, limit, offset, daysToUpdateFlags);
}
return pageWithNullFids;
}
}

10
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java
View file

@ -4,8 +4,8 @@ import java.util.ArrayList;
import java.util.List;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.indic.PhoneService2;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@ -15,9 +15,9 @@ public class KickPipeLineRestoreImpl extends KickPipeLine {
public KickPipeLineRestoreImpl(StaxXmlProcessor xmlProcessor,
SetupParam setupParam,
UserPackageMapper userPackageMapper,
PhoneService2 phoneService2,
PhoneFidMapMapper phoneFidMapMapper,
CreditTrackerSettings creditTrackerSettings) {
super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
super(xmlProcessor, setupParam, userPackageMapper, phoneFidMapMapper, creditTrackerSettings);
}
@Override
@ -29,12 +29,12 @@ public class KickPipeLineRestoreImpl extends KickPipeLine {
protected List<PhoneFidMap> nextPage(Integer limit) {
List<PhoneFidMap> pageWithNullFids = new ArrayList<>();
if (!nullFidsWereFound) {
pageWithNullFids = phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset);
pageWithNullFids = phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset);
}
if (pageWithNullFids.isEmpty()) {
nullFidsWereFound = true;
offset = 0;
return phoneService2.pageByPackageIdWithNotNullFids(packageId, limit, offset);
return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset);
}
return pageWithNullFids;
}

8
src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java
View file

@ -1,6 +1,12 @@
package ru.nbch.credit_tracker.service.task.impl;
/**
* DEFAULT - обычный режим при первичной постановки пакета на мониторинг;
* RESTORE - режим, в котором предусмотренно восстановление работы, после падения при штатной обработке пакета;
* REPEATED_RESTORE - режим, в котором предусмотрено восстановление, после падения в момент перерасчета флагов;
*/
public enum PipelineMode {
DEFAULT,
RESTORE
RESTORE,
REPEATED_RESTORE
}

54
src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml
View file

@ -71,12 +71,22 @@
<update id="setFidJson" parameterType="string">
update signals_phone_fid_map dst
set fid = src.fid,
calculated_at = src.calculated_at
set fid = src.fid,
searched_at = now(),
error_text = null
FROM json_to_recordset(#{json}::json) AS src(
id bigint,
fid int,
calculated_at timestamp
fid int
)
where dst.id = src.id;
</update>
<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>
@ -149,9 +159,6 @@
subject_id,
phone,
fid,
flag_90_12,
flag_debt5000,
calculated_at,
error_text
FROM signals_phone_fid_map
WHERE package_id = #{packageId}
@ -164,10 +171,6 @@
package_id,
subject_id,
phone,
fid,
flag_90_12,
flag_debt5000,
calculated_at,
error_text
FROM signals_phone_fid_map
WHERE package_id = #{packageId} and fid is null
@ -181,9 +184,6 @@
subject_id,
phone,
fid,
flag_90_12,
flag_debt5000,
calculated_at,
error_text
FROM signals_phone_fid_map
WHERE package_id = #{packageId} and fid is not null
@ -191,6 +191,32 @@
LIMIT #{limit} OFFSET #{offset}
</select>
<select id="pageByPackageIdAndOlderThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">
SELECT id,
package_id,
subject_id,
phone,
error_text
FROM signals_phone_fid_map
WHERE package_id = #{packageId} and searched_at &lt; now() - make_interval(days => #{daysCount})
ORDER BY id
LIMIT #{limit} OFFSET #{offset}
</select>
<select id="pageByPackageIdAndNewerThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">
SELECT id,
package_id,
subject_id,
phone,
fid,
error_text
FROM signals_phone_fid_map
WHERE package_id = #{packageId} and searched_at &gt;= now() - make_interval(days => #{daysCount})
ORDER BY id
LIMIT #{limit} OFFSET #{offset}
</select>
<select id="findPhoneFidMapById" resultMap="PhoneFidMapResultMap"
resultType="ru.nbch.credit_tracker.entities.app.PhoneFidMap">
SELECT id,

20
src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml
View file

@ -13,7 +13,9 @@
<result property="status" column="status"/>
<result property="deactivatedAt" column="deactivated_at"/>
<result property="lastDownloadedReport" column="last_downloaded_report"/>
<!-- <result property="monitoringAt" column="monitoring_at"/>-->
<result property="monitoringAt" column="monitoring_at"/>
<result property="lastSearchedAt" column="last_searched_at"/>
<result property="rejectStatus" column="reject_status"/>
</resultMap>
<select id="isPackageNotUnique" resultType="boolean">
@ -88,9 +90,15 @@
WHERE id = #{id}
</update>
<update id="setRejectedStatus">
UPDATE signals_user_packages
SET reject_status = #{status}
WHERE id = #{id}
</update>
<select id="findPackage" resultMap="packageDataResultMap"
resultType="ru.nbch.credit_tracker.entities.app.UserPackage">
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status
FROM signals_user_packages
WHERE id = #{id}
LIMIT 1
@ -98,7 +106,7 @@
<select id="findUserPackages" resultMap="packageDataResultMap"
resultType="ru.nbch.credit_tracker.entities.app.UserPackage">
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status
FROM signals_user_packages
WHERE membercode IS NOT NULL
AND LEFT(membercode, 6) = #{userId}
@ -110,7 +118,7 @@
<select id="findAllPackagesByUserId" resultMap="packageDataResultMap"
resultType="ru.nbch.credit_tracker.entities.app.UserPackage">
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status
FROM signals_user_packages
WHERE membercode IS NOT NULL
AND LEFT(membercode, 6) = #{userId}
@ -118,7 +126,7 @@
<select id="findAllPackagesByStatus" resultMap="packageDataResultMap"
resultType="ru.nbch.credit_tracker.entities.app.UserPackage">
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status
FROM signals_user_packages
WHERE status in
<foreach collection="statuses" item="status" separator="," open="(" close=")">
@ -127,7 +135,7 @@
</select>
<select id="findByMemberCodeAndPackageName" resultMap="packageDataResultMap">
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at, last_searched_at, reject_status
FROM signals_user_packages
WHERE LEFT(membercode, 6) = #{memberCode} AND package_name = #{packageName}
</select>