From d78af08405d219524eb89ff64d13c3fee0715785 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Tue, 2 Dec 2025 13:27:21 +0300 Subject: [PATCH] fix recalc flags --- .../credit_tracker/service/SignalService.java | 243 +----------------- .../service/UpdatingFlagsService.java | 59 ++--- .../service/indic/PersonService.java | 2 + .../service/signal/SignalRestService.java | 9 - .../service/task/impl/KickPipeLine.java | 4 +- .../task/impl/KickPipeLineRestoreImpl.java | 2 +- .../mapper/signal/PhoneFidMapMapper.xml | 1 - 7 files changed, 28 insertions(+), 292 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index ec17c6b..60eb9b0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -1,48 +1,25 @@ package ru.nbch.credit_tracker.service; import java.io.BufferedInputStream; -import java.io.File; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; -import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; -import java.util.Objects; -import java.util.UUID; -import java.util.function.Consumer; -import java.util.function.Predicate; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.extern.slf4j.Slf4j; 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.app.Hit; import ru.nbch.credit_tracker.entities.app.PhoneFidMap; import ru.nbch.credit_tracker.entities.app.UserPackage; -import ru.nbch.credit_tracker.entities.external_request.signals.Auth; -import ru.nbch.credit_tracker.entities.external_request.signals.Person; -import ru.nbch.credit_tracker.entities.external_request.signals.Signals; import ru.nbch.credit_tracker.entities.external_response.monitoring_report_error.MonitoringReportErrors; import ru.nbch.credit_tracker.enums.Status; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.FlagsService; -import ru.nbch.credit_tracker.service.indic.PersonService; -import ru.nbch.credit_tracker.service.indic.PhoneService; import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.signal.UserPackageDao; -import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; -import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; -import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.CTUtil; @Slf4j @Service @@ -55,13 +32,7 @@ public class SignalService { private final UserPackageDao userPackageService; private final PhoneFidMapService phoneFidMapService; private final PhoneService2 phoneService2; - private final PhoneService phoneService; - private final FlagsService flagsService; - private final PersonService personService; private final StaxXmlProcessor staxXmlProcessor; - private final FidNameEnricher fidNameEnricher; - private final FlagsEnricher flagsEnricher; - private final int batchSize; public SignalService(CreditTrackerSettings config, @@ -69,137 +40,14 @@ public class SignalService { UserPackageDao userPackageService, PhoneFidMapService phoneFidMapService, PhoneService2 phoneService2, - PhoneService phoneService, - FlagsService flagsService, - PersonService personService, - StaxXmlProcessor staxXmlProcessor, FidNameEnricher fidNameEnricher, FlagsEnricher flagsEnricher) { + StaxXmlProcessor staxXmlProcessor) { this.config = config; this.hitsService = hitsService; this.userPackageService = userPackageService; this.phoneFidMapService = phoneFidMapService; this.phoneService2 = phoneService2; - this.phoneService = phoneService; - this.flagsService = flagsService; - this.personService = personService; this.staxXmlProcessor = staxXmlProcessor; this.batchSize = config.getPipeline().subjectBatchSize(); - this.fidNameEnricher = fidNameEnricher; - this.flagsEnricher = flagsEnricher; - } - - public UpdateResult updateFlags2(UserPackage userPackage) { - log.debug("updating flags for package [{} {}]", userPackage.getPackageName(), userPackage.getId()); - Long packageId = userPackage.getId(); - int offset = 0; - List page = phoneService2.pageByPackageId(packageId, batchSize, offset); - int counter = 1; - while (!page.isEmpty()) { - log.debug("got page number {}, size {}", counter++, page.size()); - List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() - .mapId(phoneFidMap.getId()) - .identifier(phoneFidMap.getSubjectId()) - .phone(phoneFidMap.getPhone()) - .build()).toList(); - SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles); - log.debug("page number {}: updating fids...", counter); - fidNameEnricher.handle(payload); - log.debug("page number {}: updating flags...", counter); - flagsEnricher.process(payload, true); - offset += batchSize; - log.debug("loading next page..."); - page = phoneService2.pageByPackageId(packageId, batchSize, offset); - } - return new UpdateResult(true, false); - } - - /** - * Обрабатывает все PhoneFidMap по packageId, выгружая их пачками и фильтруя по переданному фильтру - */ - public UpdateResult updateSubjects(Long packageId, Predicate filter) { - boolean updated = false; - boolean noFidsFound = true; - int offset = 0; - List batch; - Map phonesByFidToProcessing = new HashMap<>(); - do { - batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize); - - List filteredItems; - if (filter != null) { - filteredItems = batch.stream().filter(filter).toList(); - } else { - filteredItems = batch; - } - - if (!updated && !filteredItems.isEmpty()) { - updated = true; - } - - if (!filteredItems.isEmpty()) { - Map fidsMapByBatch = phoneService.defineFidByPhone(filteredItems); - phoneService.defineFidByPhone(filteredItems); - log.debug("Actual fids size by phone's batch: {}", fidsMapByBatch.size()); - - if (fidsMapByBatch.size() + phonesByFidToProcessing.size() <= batchSize) { - phonesByFidToProcessing.putAll(fidsMapByBatch); - } else { - //todo execute find name and id -> forming xml - boolean fidsFound = updateFlags(phonesByFidToProcessing, packageId); - if (noFidsFound) { - noFidsFound = !fidsFound; - } - phonesByFidToProcessing = new HashMap<>(fidsMapByBatch); - } - } - - offset += batchSize; - } while (!batch.isEmpty()); - - if (!phonesByFidToProcessing.isEmpty()) { - boolean fidsFound = updateFlags(phonesByFidToProcessing, packageId); - if (noFidsFound) { - noFidsFound = !fidsFound; - } - } - - return new UpdateResult(updated, noFidsFound); - } - - /** - * Обрабатывает пачку PhoneFidMap: обновляет фиды и флаги просрочек - * @return fidsFound true - если хотя бы по одному номеру был найден фид, false - в противном случае - */ - // TODO Добавить отдельный метод. Если не нашелся фид, надо обновить в базе fid = null и флаги = null calculated_at=now() - public boolean updateFlags(Map fidsMap, Long packageId) { - log.info("Start calculating flags for subjects batch. PackageId: {}", packageId); - Map flagsDataMap = flagsService.findFlags(fidsMap.keySet().stream().toList()); - List subjectToUpdate = new ArrayList<>(); - for (Map.Entry entry : flagsDataMap.entrySet()) { - Integer fid = entry.getKey(); - FlagsService.FlagsData flagsData = entry.getValue(); - PhoneFidMap phoneFidMap = fidsMap.get(fid); - - phoneFidMap.setFid(fid); - phoneFidMap.setFlagDebt5000(flagsData.getFlagDebt5000()); - phoneFidMap.setFlag90days(flagsData.getFlag90days()); - phoneFidMap.setCalculatedAt(LocalDateTime.now()); - - subjectToUpdate.add(phoneFidMap); - } - - log.debug("Start updating subject's batch"); - if (!subjectToUpdate.isEmpty()) { - int updateBatchSize = 7000; // 9 параметров на одну запись для обновления. Максимальное количество параметров ~65000 - for (int j = 0; j < subjectToUpdate.size(); j += updateBatchSize) { - int endIndex = Math.min(j + updateBatchSize, subjectToUpdate.size()); - phoneFidMapService.updatePhoneFidMaps(subjectToUpdate.subList(j, endIndex)); - log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, subjectToUpdate.size(), packageId); - } - } - log.debug("Complete updating subject's batch"); - - log.info("Finish calculating flags for subjects. PackageId: {}. Batch size: {}", packageId, fidsMap.size()); - return !fidsMap.isEmpty(); } @@ -207,87 +55,6 @@ public class SignalService { userPackageService.updatePackageStatus(packageId, status); } - public void updatePackageOnMonitoringSuccess(Long packageId, LocalDateTime monitoringAt) { - userPackageService.updatePackageOnMonitoringSuccess(packageId, monitoringAt); - } - - /** - * Формирует файл запроса в официальный сервис сигналов, выгружая паспортные данные пачками и сразу записывая их в файл - * - * @return Путь к файлу запроса к официальному сервису сигналов на постановку на мониторинг - */ - public String writeSignals(Long packageId) { - String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator; - String filePath = folderPath + UUID.randomUUID() + "_output.xml"; - CTUtil.createFolder(folderPath); - - Signals signals = buildSignals(packageId); - - SignalsStaxXmlWriter signalsStaxXmlWriter = new SignalsStaxXmlWriter(config.getEncoding()); - - signalsStaxXmlWriter.writeWithStax(filePath, - signals, - writer -> buildPersons(packageId, writer)); - - return filePath; - } - - public Signals buildSignals(Long packageId) { - UserPackage userPackage = userPackageService.findPackage(packageId); - Signals signals = new Signals(); - signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы) - signals.setProcessWithErrors("1"); // Всегда=1 (обработать пакет полностью, по обнаруженным ошибкам (при наличии) сформировать файл отказа, записи без ошибок установить на мониторинг) - signals.setId(userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName()); // _ - Auth auth = new Auth(); - auth.setUser(config.getSignalsuser()); - signals.setAuth(auth); - return signals; - } - - /** - * Обрабатывает все PhoneFidMap по packageId, выгружая их пачками и фильтруя по переданному фильтру. - * writer записывает пачку сразу в файл - */ - private void buildPersons(Long packageId, Consumer> writer) { - int offset = 0; - List batch; - do { - batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize); - List persons = findPersons(batch); - if (!persons.isEmpty()) { - writer.accept(persons); // записывает пачку Persons в файл - } - offset += batchSize; - } while (!batch.isEmpty()); - - } - - /** - * Формирует список Person для постановки на мониторинг в официальный сервис сигналов на основе списка PhoneFidMap - */ - private List findPersons(List subjects) { - if (subjects.isEmpty()) { - return Collections.emptyList(); - } - List persons = new ArrayList<>(); - Map personFidMap = personService.findPersonData(subjects.stream() - .map(PhoneFidMap::getFid) - .filter(Objects::nonNull).toList()); - subjects.forEach(data -> { - if (data.getFid() != null) { - PersonService.PersonData personData = personFidMap.get(data.getFid()); - Person person = PersonFactoryBuilder.builder() - .personData(personData) - .uid(data.getId()) - .build(); - if (person != null) { - persons.add(person); - } - } - }); - return persons; - } - public 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)) { @@ -314,14 +81,6 @@ public class SignalService { log.info("Subject's errors were updated. PackageId: {}", packageId); } - public List findPackagesOnMonitoring() { - return userPackageService.findAllPackagesByStatus(Status.ON_MONITORING); - } - - public List findPackagesOnProcessing() { - return userPackageService.findAllPackagesByStatus(Status.PROCESSING); - } - public void insertHits(List hits) { hitsService.insertHits(hits); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index 3033210..d126b61 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -1,8 +1,7 @@ package ru.nbch.credit_tracker.service; -import java.time.LocalDateTime; -import java.time.temporal.ChronoUnit; import java.util.List; +import java.util.concurrent.ExecutorService; import lombok.extern.slf4j.Slf4j; import org.slf4j.MDC; import org.springframework.scheduling.annotation.EnableScheduling; @@ -14,6 +13,10 @@ 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; +import ru.nbch.credit_tracker.service.task.impl.PipelineMode; /** * Сервис периодического обновления фидов и флагов просрочек по пакетам находящимся в статусе ON_MONITORING @@ -30,66 +33,46 @@ public class UpdatingFlagsService { private final Long daysToUpdateFlags; private final PhoneFidMapMapper phoneFidMapMapper; private final UserPackageMapper userPackageMapper; + private final PipelinePackageManager pipelinePackageManager; + private final ExecutorService pipelineExecutorService; public UpdatingFlagsService(SignalService signalService, SignalRestService signalRestService, MonitoringErrorProcesingService monitoringErrorProcesingService, CreditTrackerSettings config, - PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper) { + PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper, + PipelinePackageManager pipelinePackageManager, + ExecutorService pipelineExecutorService) { this.signalService = signalService; this.signalRestService = signalRestService; this.monitoringErrorProcesingService = monitoringErrorProcesingService; this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1; this.phoneFidMapMapper = phoneFidMapMapper; this.userPackageMapper = userPackageMapper; + this.pipelinePackageManager = pipelinePackageManager; + this.pipelineExecutorService = pipelineExecutorService; } @Scheduled(cron = "${ct.update-flags-cron-expression}") private void update() { log.info("Start updating flags"); - List packages = signalService.findPackagesOnMonitoring(); + List packages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); for (UserPackage userPackage : packages) { - MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() +"]"); + MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() + "]"); log.info("Check on updating flags for Package {}", userPackage.getPackageName()); boolean hasOldRow = phoneFidMapMapper.hasOldRow(userPackage.getId(), daysToUpdateFlags.intValue()); log.info("There are {} old rows", hasOldRow ? "exist" : "doesn't exist"); if (hasOldRow) { - updateFlags(userPackage); + try { + userPackageMapper.updatePackageStatus(userPackage.getId(), Status.PROCESSING_REPEATED.name()); + Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT)); + pipeline.start(pipelineExecutorService); + } catch (Exception e) { + log.error("Failed to build pipeline", e); + } } MDC.remove("ru.nbch.credit_tracker.log.mdc.key"); } log.info("Finish updating flags"); } - - private void updateFlags(UserPackage userPackage) { - Long packageId = userPackage.getId(); - String packageName = userPackage.getPackageName(); - try { - //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)) { -// LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now(); - log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName); - signalService.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); - monitoringErrorProcesingService.findErrors(packageId); - } else { - log.info("Updated package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName); - signalService.updatePackageStatus(packageId, Status.ERROR); - } - } else { - log.info("No flags were updated for package id {}. Package status changed to ON_MONITORING", packageId); - signalService.updatePackageStatus(packageId, Status.ON_MONITORING); - } - } catch (Exception e) { - log.error("Ошибка при поиске фидов или перерасчете флагов просрочек. Package_id {}", packageId, e); - } - } - - private boolean hasDaysPassed(LocalDateTime pastDateTime) { - LocalDateTime now = LocalDateTime.now(); - long daysPassed = ChronoUnit.DAYS.between(pastDateTime, now); - return daysPassed >= daysToUpdateFlags; - } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java index b18f4bf..aa91e9b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java @@ -31,6 +31,8 @@ public class PersonService { return new HashMap<>(); } Map personDataMap = new HashMap<>(); + + //делаем тоже самое что сделали signalService.updateFlags2(userPackage); Map nameDataMap = findNameData(fids); Map passortDataMap = findPassportData(fids); diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java index 692028b..79c298e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java @@ -66,15 +66,6 @@ public class SignalRestService { this.authToken = buildToken(config.getSignalsToken(), config.getSignalsuser()); } - /** - * Ставит пакет на мониторинг в официальный сервис сигналов - * @return true - если пакет успешно поставился на мониторинг. false - в противном случае - */ - public boolean setupMonitoring(Long packageId) throws Exception { - String path = signalService.writeSignals(packageId); - return setupMonitoring(Path.of(path)); - } - public boolean setupMonitoring(Path signalFilePath) throws Exception { Path gzip = ZipUtils.packGzip(signalFilePath); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java index 5a23bd0..cba791f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java @@ -55,7 +55,7 @@ public abstract class KickPipeLine implements IPipeLineStarter { @Override public void startFillingQueue(BlockingQueue firstQueue) { try { - log.info("Start registering subjects. PackageId: {};", packageId); + log.info("Start kicker: mode:{}, packageId: {};", mode(), packageId); int limit = BATCH_SIZE; UserPackage userPackage = userPackageMapper.findPackage(packageId); List page = nextPage(limit); @@ -64,6 +64,8 @@ public abstract class KickPipeLine implements IPipeLineStarter { List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() .mapId(phoneFidMap.getId()) .identifier(phoneFidMap.getSubjectId()) + //наличие этого поля зависит от выборки, в определенных select намеренно опущено это поле + .fid(phoneFidMap.getFid()) .phone(phoneFidMap.getPhone()) .build()).toList(); try { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java index 2f504bd..0324ff6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java @@ -22,7 +22,7 @@ public class KickPipeLineRestoreImpl extends KickPipeLine { @Override protected PipelineMode mode() { - return PipelineMode.RESTORE; + return PipelineMode.REPEATED_RESTORE; } @Override diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index d9102c4..48e7207 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -158,7 +158,6 @@ package_id, subject_id, phone, - fid, error_text FROM signals_phone_fid_map WHERE package_id = #{packageId}