From 1cc1695b65bb385648b2a51ee6fd2f555b02a346 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Mon, 19 Jan 2026 12:07:09 +0300 Subject: [PATCH] http://jira.mfd.msk:8088/browse/BKI-3241 --- .../CreditTrackerApplication.java | 3 -- .../mapper/signal/PhonesMapMapper.java | 1 + .../mapper/signal/UserPackageMapper.java | 1 + .../service/UpdatingFidAndPdService.java | 40 ++++++++++++---- .../service/indic/PersonServiceV2.java | 3 +- .../service/signal/PhonesMapService.java | 4 ++ .../service/task/PipelinePackageFabric.java | 1 - .../task/PipelineRecalcPDOrchestrator.java | 32 ++++--------- .../service/task/impl/FidNameEnricher.java | 2 +- .../service/task/impl/FidNameUpdater.java | 4 +- .../mapper/signal/PackageRecordsMapper.xml | 3 +- .../mybatis/mapper/signal/PhonesMapMapper.xml | 46 ++++++++----------- .../mapper/signal/UserPackageMapper.xml | 19 +++++++- 13 files changed, 92 insertions(+), 67 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index d1ef4d8..97b975d 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -5,7 +5,6 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ConfigurableApplicationContext; import ru.nbch.credit_tracker.service.MonitoringService; import ru.nbch.credit_tracker.service.UpdateProcessingPackagesService; -import ru.nbch.credit_tracker.service.UpdatingFlagsService; @SpringBootApplication public class CreditTrackerApplication { @@ -14,8 +13,6 @@ 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); -// updatingFlagsService.update(); updateProcessingPackagesService.run(); monitoringService.run(); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index 338fb7a..0829ec8 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -18,4 +18,5 @@ public interface PhonesMapMapper { void resetFidAndPD(@Param("json") String json); void registerJson(@Param("json") String json); void setFidJson(@Param("json") String json); + void deleteUnusedPhones(); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java index 38ae703..9ab791e 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java @@ -30,6 +30,7 @@ public interface UserPackageMapper { void setFidChanged(@Param("packageIds") List id); UserPackage findPackage(@Param("id") Long id); + List findForRepeatedPackages(); List findUserPackages(@Param("userId") String userId, @Param("statuses") List status); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java index dc024ef..5193514 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java @@ -1,10 +1,15 @@ package ru.nbch.credit_tracker.service; +import java.util.List; +import java.util.concurrent.CompletableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.UserPackage; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.signal.PhonesMapService; import ru.nbch.credit_tracker.service.task.Pipeline; import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; @@ -19,29 +24,48 @@ public class UpdatingFidAndPdService { private final PipelinePackageFabric pipelinePackageManager; private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; private final PipelinePackageOrchestrator pipelinePackageOrchestrator; + private final UserPackageMapper userPackageMapper; + private final PhonesMapService phonesMapService; public UpdatingFidAndPdService(CreditTrackerSettings config, PipelinePackageFabric pipelinePackageManager, PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, - PipelinePackageOrchestrator pipelinePackageOrchestrator) { + PipelinePackageOrchestrator pipelinePackageOrchestrator, + UserPackageMapper userPackageMapper, + PhonesMapService phonesMapService) { this.pipelinePackageManager = pipelinePackageManager; this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; this.pipelinePackageOrchestrator = pipelinePackageOrchestrator; + this.userPackageMapper = userPackageMapper; + this.phonesMapService = phonesMapService; } @Scheduled(cron = "${ct.update-fid-cron-expression}") public void update() { + log.info("Delete delete unused phones"); + phonesMapService.deleteUnusedPhones(); log.info("Start updating fids and PD"); try { Pipeline pipeline = pipelinePackageManager.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); - pipelineRecalcPDOrchestrator.startOrReject(pipeline, () -> { - SetupParam setupParam = new SetupParam(null, null, PipelineMode.MAKE_XML_BY_REPEATED); - try { - Pipeline makeXmlByCache = pipelinePackageManager.build(setupParam); - pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, null); - } catch (Exception e) { - log.error("Failed to build pipeline", e); + CompletableFuture future = pipelineRecalcPDOrchestrator.startOrReject(pipeline); + future.whenComplete((o, throwable) -> { + if (throwable != null) { + log.error(throwable.getMessage(), throwable); + return; } + + List packagesRepeated = userPackageMapper.findForRepeatedPackages(); + packagesRepeated.forEach(userPackage -> { + Long packageId = userPackage.getId(); + SetupParam setupParam = new SetupParam(packageId, null, PipelineMode.DEFAULT); + try { + Pipeline makeXmlByCache = pipelinePackageManager.build(setupParam); + pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId); + } catch (Exception e) { + log.error("Failed to build pipeline", e); + } + }); + }); } catch (Exception e) { log.error("Failed to build pipeline", e); diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java index b12b7e8..9f62433 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.service.indic; +import java.time.LocalDate; import java.util.Comparator; import java.util.HashMap; import java.util.List; @@ -73,7 +74,7 @@ public class PersonServiceV2 { .filter(name -> StringUtils.isNotEmpty(name.getFirst()) && StringUtils.isNotEmpty(name.getLast()) && - name.getBirthDate() != null + name.getBirthDate() != null && !LocalDate.of(1, 1, 1).equals(name.getBirthDate()) ) .collect(Collectors.toMap( Name::getNumber, diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java index 3d85393..065784e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java @@ -50,4 +50,8 @@ public class PhonesMapService { throw new RuntimeException(e); } } + + public void deleteUnusedPhones() { + phonesMapMapper.deleteUnusedPhones(); + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index b5755a6..9abc35f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -80,7 +80,6 @@ public class PipelinePackageFabric implements InitializingBean { public Pipeline build(SetupParam setupParam) throws Exception { return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(fidNameEnricher) - .handlerResult(flagsEnricherV2) .andThen() .stage(sendSignalQueueFactory.getObject()) .handler(fidIdEnricherHandler) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java index da0d0ca..0a92801 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java @@ -1,5 +1,6 @@ package ru.nbch.credit_tracker.service.task; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -24,46 +25,33 @@ public class PipelineRecalcPDOrchestrator { this.semaphore = new Semaphore(1); } - public void startOrReject(Pipeline pipeline, Runnable after) { + public CompletableFuture startOrReject(Pipeline pipeline) { synchronized (lock) { if (!semaphore.tryAcquire()) { log.debug("couldn't acquire permit from semaphore. task rejected..."); - return; + return CompletableFuture.failedFuture(new IllegalStateException("couldn't acquire permit from semaphore.")); } log.info("throttler permits after acquire: {}", semaphore.availablePermits()); } - innerStart(pipeline, after); + return innerStart(pipeline); } - private void innerStart(Pipeline pipeline, Runnable after) { + private CompletableFuture innerStart(Pipeline pipeline) { try { pipeline.start(pipelineExecutorService); - addCallback(pipeline, after); + return addCallback(pipeline); } catch (Exception e) { log.error(ExceptionUtils.getStackTrace(e)); pipeline.close(); release(); + return CompletableFuture.failedFuture(e); } } - private void addCallback(Pipeline pipeline, Runnable after) { - pipeline.getTerminator() + private CompletableFuture addCallback(Pipeline pipeline) { + return pipeline.getTerminator() .doneFuture() - .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS) - .whenComplete((v, e) -> { - try { - long finishTime = System.currentTimeMillis(); - log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime()); - after.run(); - } catch (Exception ex) { - log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); - } finally { - log.info("pipeline finished"); - pipeline.close(); - release(); - log.info("semaphore amount: {}", semaphore.availablePermits()); - } - }); + .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS); } private void release() { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java index 53cd7ab..be3e23d 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java @@ -116,7 +116,7 @@ public class FidNameEnricher implements Handler .filter(cr -> cachedRecord.getSerNum() != null && cachedRecord.getIdType() != null) .map(cr -> Optional.of( (IdData.builder() - .idNum(cr.getSerNum()) + .serNum(cr.getSerNum()) .idType(cr.getIdType()) .build() ))) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java index edd4546..f778c0c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -49,9 +49,9 @@ public class FidNameUpdater implements Handler { List notFoundFidPhones = new ArrayList<>(); subjects.forEach(subjectProfile -> { Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone())); - if (name == null) { + if (subjectProfile.getFid() == null && name == null) { notFoundFidPhones.add(subjectProfile.getPhone()); - } else if (!subjectProfile.getFid().equals(name.getFid())) { + } else if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) { subjectProfile.setFid(name.getFid()); subjectProfile.getNameData().setFirstName(name.getFirst()); subjectProfile.getNameData().setLastName(name.getLast()); diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml index dceab51..e113bbd 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml @@ -43,9 +43,10 @@ )) INSERT INTO signals_packages_records - (subject_id, phone, package_id) + (subject_id, phone, phone_l, package_id) SELECT subject_id, phone, + phone::BIGINT, package_id FROM src ORDER BY rn diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index e4beb3a..44d1e87 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -145,7 +145,8 @@ name_1 = src.name_1, first = src.first, middle = src.middle, - birth_dt = src.birth_dt + birth_dt = src.birth_dt, + searched_at = now() FROM json_to_recordset(#{json}::json) AS src( fid int, phone bigint, @@ -192,35 +193,26 @@ ser_num = null, id_type = null, flag_debt5000 = null, - flag_90_12 = null + flag_90_12 = null, + calculated_at = null FROM json_to_recordset(#{json}::json) AS src( phone bigint) where dst.phone = src.phone - - - - - - - - - - - - - - - - - - - - - - - - - + + DELETE + FROM signals_phones_map spm + WHERE NOT EXISTS (SELECT 1 + FROM signals_packages_records spr + JOIN signals_user_packages sup + ON sup.id = spr.package_id + WHERE spr.phone_l = spm.phone + AND sup.status IN + ( + 'ON_MONITORING', + 'PROCESSING', + 'PROCESSING_REPEATED' + )) + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml index 3ca7b58..0c18db0 100644 --- a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml @@ -76,7 +76,8 @@ UPDATE signals_user_packages - SET fid_changed = 1 + SET fid_changed = 1, + status = 'PROCESSING_REPEATED' WHERE id in #{packageId} @@ -124,6 +125,22 @@ LIMIT 1 + +