diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index 4d8cabd..6412a7a 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -24,7 +24,6 @@ import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.enums.XmlErrorFormat; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.service.SignalService; -import ru.nbch.credit_tracker.service.indic.PhoneService; import ru.nbch.credit_tracker.service.scoring.AuthorizationService; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @@ -162,22 +161,20 @@ public class SignalFacade { executor.execute(() -> { try { signalService.registerSubjects(subjects, packageId); - List phoneFidsData = signalService.registerSubjectsFidAndDebts(packageId); - if (!phoneFidsData.isEmpty()) { - Signals signals = signalService.buildSignals(packageId, phoneFidsData); - // TODO validate xsd? - if (signalRestService.update(signals)) { - log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName); - signalService.updatePackageStatus(packageId, Status.ON_MONITORING); - } else { - log.info("Package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName); - signalService.updatePackageStatus(packageId, Status.ERROR); - } + subjects.clear(); + signalService.updateSubjects(packageId, null); + Signals signals = signalService.buildSignals(packageId); + signals.getConditions().getPersons().getPerson().addAll(signalService.buildPersons(packageId, null)); + if (signalRestService.update(signals)) { + log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName); + signalService.updatePackageStatus(packageId, Status.ON_MONITORING); + } else { + log.info("Package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName); + signalService.updatePackageStatus(packageId, Status.ERROR); } } catch (Exception e) { signalService.updatePackageStatus(packageId, Status.ERROR); log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e); - // TODO очистить таблицу signals_phone_fid_map ? } }); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java index 5e6051a..9060229 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java @@ -12,10 +12,11 @@ public interface PhoneFidMapMapper { void registerSubjects(@Param("subjects") List subjects, @Param("packageId") Integer packageId); - List findPhoneFidMaps(@Param("packageId") Integer packageId); + List findPhoneFidMaps(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit); - List findFullPhoneFidMap(@Param("packageId") Integer packageId); + List findFullPhoneFidMap(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit); void updatePhoneFidMaps(@Param("subjects") List subjects); + PhoneFidMap findPhoneFidMap(@Param("packageId") Integer packageId, @Param("subjectId") String subjectId); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java index 39ff8eb..6e6ae20 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java @@ -119,7 +119,6 @@ public class MonitoringService { MonitoringSignalResponse response = new MonitoringSignalResponse(); response.setPackageName(userPackage.getPackageName()); // Имя пакета, переданного при установке Subjects subjects = new Subjects(); - List phoneFidMapList = signalService.findFullSubjectsData(userPackage.getId()); for (Person person : persons) { Signals signals = new Signals(); for (Inquiry inquiry : person.getInquiries()) { @@ -136,7 +135,8 @@ public class MonitoringService { // Если есть хотя бы один заполненный сигнал, значит заполняем остальные данные if (!signals.getSignals().isEmpty()) { - phoneFidMapList.stream().filter(p -> p.getSubjectId().equals(person.getUid())).findFirst().ifPresent(subjectData -> { + PhoneFidMap subjectData = signalService.findPhoneFidMapBySubjectId(userPackage.getId(), person.getUid()); + if(subjectData != null) { Subject subject = new Subject(); subject.setId(subjectData.getSubjectId()); // Уникальный идентификатор субъекта (равен Id из XML-пакета при установке мониторинга) subject.setClientId(subjectData.getPhone()); @@ -144,7 +144,7 @@ public class MonitoringService { subject.setFlag9012(subjectData.getFlag90days()); subject.setSignals(signals); subjects.getSubjects().add(subject); - }); + } } } response.setSubjects(subjects); 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 595c662..21c56cd 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -21,10 +21,8 @@ import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.signal.UserPackageService; import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Objects; +import java.util.*; +import java.util.function.Predicate; @Slf4j @Service @@ -80,56 +78,78 @@ public class SignalService { log.info("Subjects were registered. PackageId: {}", packageId); } - public List registerSubjectsFidAndDebts(Integer packageId) { - List subjects = phoneFidMapService.findPhoneFidMaps(packageId); - return updateFlags(subjects, packageId); + public boolean updateSubjects(Integer packageId, Predicate filter) { + boolean updated = false; + int offset = 0; + List batch; + do { + batch = phoneFidMapService.findPhoneFidMaps(packageId, offset, batchSize); + + List filteredItems; + if (filter != null) { + filteredItems = batch.stream().filter(filter).toList(); + updateFlags(batch.stream().filter(filter).toList(), packageId); + } else { + filteredItems = batch; + } + + if (!updated && !filteredItems.isEmpty()) { + updated = true; + } + + updateFlags(filteredItems, packageId); + + offset += batchSize; + } while (!batch.isEmpty()); + + return updated; } - public List updateFlags(List subjects, Integer packageId) { - log.info("Start calculating fids and flags for subjects. PackageId: {}", packageId); + public void updateFlags(List subjects, Integer packageId) { + if (subjects.isEmpty()) { + return; + } + log.info("Start calculating fids and flags for subjects batch. PackageId: {}", packageId); List fidsPhoneDataList = phoneService.findFidsPhoneData(subjects); LocalDateTime calculatedAt = LocalDateTime.now(); - for (int i = 0; i < fidsPhoneDataList.size(); i += batchSize) { - int end = Math.min(i + batchSize, fidsPhoneDataList.size()); - List currentList = fidsPhoneDataList.subList(i, end); - Map flagsDataMap = flagsService.findFlags(currentList.stream().map(PhoneService.FidsPhoneData::getFid).filter(Objects::nonNull).toList()); - List subjectToUpdate = new ArrayList<>(); - currentList.forEach(data -> { - PhoneFidMap subject; - if (data.getFid() != null) { - FlagsService.FlagsData flagsData = flagsDataMap.get(data.getFid()); - subject = PhoneFidMap.builder() - .id(data.getId()) - .packageId(packageId) - .subjectId(data.getSubjectId()) - .phone(data.getPhone()) - .fid(data.getFid()) - .flagDebt5000(flagsData.getFlagDebt5000()) - .flag90days(flagsData.getFlag90days()) - .calculatedAt(calculatedAt) - .build(); - } else { - subject = PhoneFidMap.builder() - .id(data.getId()) - .packageId(packageId) - .subjectId(data.getSubjectId()) - .phone(data.getPhone()) - .calculatedAt(calculatedAt) - .build(); - } - subjectToUpdate.add(subject); - }); + Map flagsDataMap = flagsService.findFlags(fidsPhoneDataList.stream().map(PhoneService.FidsPhoneData::getFid).filter(Objects::nonNull).toList()); + List subjectToUpdate = new ArrayList<>(); - int updateBatchSize = batchSize / 8; // 8 параметров на одну запись для обновления. Максимальное количество параметров ~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: {}", i, end, subjectToUpdate.size(), packageId); + fidsPhoneDataList.forEach(data -> { + PhoneFidMap subject; + if (data.getFid() != null) { + FlagsService.FlagsData flagsData = flagsDataMap.get(data.getFid()); + subject = PhoneFidMap.builder() + .id(data.getId()) + .packageId(packageId) + .subjectId(data.getSubjectId()) + .phone(data.getPhone()) + .fid(data.getFid()) + .flagDebt5000(flagsData.getFlagDebt5000()) + .flag90days(flagsData.getFlag90days()) + .calculatedAt(calculatedAt) + .build(); + } else { + subject = PhoneFidMap.builder() + .id(data.getId()) + .packageId(packageId) + .subjectId(data.getSubjectId()) + .phone(data.getPhone()) + .calculatedAt(calculatedAt) + .build(); } + subjectToUpdate.add(subject); + }); + + int updateBatchSize = batchSize / 8; // 8 параметров на одну запись для обновления. Максимальное количество параметров ~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.info("Finish calculating fids and flags for subjects. PackageId: {}", packageId); - return fidsPhoneDataList; + + log.info("Finish calculating fids and flags for subjects. PackageId: {}. Batch size: {}", packageId, subjects.size()); } public void rejectPackage(MonitoringRequest request) { @@ -146,7 +166,7 @@ public class SignalService { userPackageService.updatePackageStatus(packageId, status); } - public Signals buildSignals(Integer packageId, List fids) { + public Signals buildSignals(Integer packageId) { UserPackage userPackage = userPackageService.findPackage(packageId); Signals signals = new Signals(); signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы) @@ -154,23 +174,40 @@ public class SignalService { Auth auth = new Auth(); auth.setUser(config.getSignalsuser()); signals.setAuth(auth); - for (int i = 0; i < fids.size(); i += batchSize) { - int end = Math.min(i + batchSize, fids.size()); - List persons = findPersons(fids.subList(i, end)); - if (!persons.isEmpty()) { - signals.getConditions().getPersons().getPerson().addAll(persons); - } - } return signals; } - private List findPersons(List fidsPhoneData) { + public List buildPersons(Integer packageId, Predicate filter) { + List result = new ArrayList<>(); + int offset = 0; + List batch; + do { + batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize); + List persons; + if (filter != null) { + persons = findPersons(batch.stream().filter(filter).toList()); + } else { + persons = findPersons(batch); + } + if (!persons.isEmpty()) { + result.addAll(persons); + } + offset += batchSize; + } while (!batch.isEmpty()); + + return result; + } + + private List findPersons(List subjects) { + if (subjects.isEmpty()) { + return Collections.emptyList(); + } List persons = new ArrayList<>(); - Map personFidMap = personService.findPersonData(fidsPhoneData.stream() - .map(PhoneService.FidsPhoneData::getFid) + Map personFidMap = personService.findPersonData(subjects.stream() + .map(PhoneFidMap::getFid) .filter(Objects::nonNull).toList()); - fidsPhoneData.forEach(data -> { + subjects.forEach(data -> { if (data.getFid() != null) { PersonService.PersonData personData = personFidMap.get(data.getFid()); Person person = PersonFactoryBuilder.builder() @@ -201,10 +238,6 @@ public class SignalService { return userPackageService.findAllPackagesByStatus(Status.ON_MONITORING); } - public List findFullSubjectsData(Integer packageId) { - return phoneFidMapService.findFullPhoneFidMap(packageId); - } - public void insertHits(List hits) { hitsService.insertHits(hits); } @@ -212,4 +245,8 @@ public class SignalService { public void updateLastDownloadedReport(Integer id, String fileName) { userPackageService.updatePackageLastDownloadedReport(id, fileName); } + + public PhoneFidMap findPhoneFidMapBySubjectId(Integer packageId, String subjectId) { + return phoneFidMapService.findPhoneFidMap(packageId, subjectId); + } } 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 8539cbf..fef48fe 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -4,11 +4,9 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import ru.nbch.credit_tracker.entities.dto.UserPackage; import ru.nbch.credit_tracker.entities.signal_package.Signals; import ru.nbch.credit_tracker.enums.Status; -import ru.nbch.credit_tracker.service.indic.PhoneService; import ru.nbch.credit_tracker.service.signal.SignalRestService; import java.time.LocalDateTime; @@ -46,19 +44,18 @@ public class UpdatingFlagsService { List packages = signalService.findPackagesOnMonitoring(); for (UserPackage userPackage : packages) { log.info("Updating flags for Package {}", userPackage.getPackageName()); - List subjects = signalService.findFullSubjectsData(userPackage.getId()) - .stream().filter(subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt())).toList(); - updateFlags(userPackage.getId(), subjects, userPackage.getPackageName()); + updateFlags(userPackage.getId(), userPackage.getPackageName()); } log.info("Finish updating flags"); } - private void updateFlags(Integer packageId, List subjects, String packageName) { + private void updateFlags(Integer packageId, String packageName) { try { signalService.updatePackageStatus(packageId, Status.PROCESSING); - List phoneFidsData = signalService.updateFlags(subjects, packageId); - if (!phoneFidsData.isEmpty()) { - Signals signals = signalService.buildSignals(packageId, phoneFidsData); + boolean updated = signalService.updateSubjects(packageId, subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt())); + if (updated) { + Signals signals = signalService.buildSignals(packageId); + signals.getConditions().getPersons().getPerson().addAll(signalService.buildPersons(packageId, subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt()))); if (signalRestService.update(signals)) { log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName); signalService.updatePackageStatus(packageId, Status.ON_MONITORING); diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java index 18c09ae..605a4d5 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService.java @@ -22,25 +22,21 @@ public class PhoneService { public List findFidsPhoneData(List subjects) { List fidsPhoneDataList = new ArrayList<>(); - int batchSize = 10000; - for (int i = 0; i < subjects.size(); i += batchSize) { - int end = Math.min(i + batchSize, subjects.size()); - List currentSubjects = subjects.subList(i, end); - List phones = currentSubjects.stream().map(subject -> "9" + subject.getPhone()).toList(); - List fidsData = mapper.findFids(phones); - Map fidsMap = convertFidsDataToMap(fidsData); + List phones = subjects.stream().map(subject -> "9" + subject.getPhone()).toList(); + List fidsData = mapper.findFids(phones); + Map fidsMap = convertFidsDataToMap(fidsData); + + subjects.forEach(subject -> { + String phone = "9" + subject.getPhone(); + Integer fid = fidsMap.get(phone); + if (fid != null) { + fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), fid, subject.getPhone(), subject.getSubjectId())); + } else { + fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), null, subject.getPhone(), subject.getSubjectId())); + } + }); - currentSubjects.forEach(subject -> { - String phone = "9" + subject.getPhone(); - Integer fid = fidsMap.get(phone); - if (fid != null) { - fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), fid, subject.getPhone(), subject.getSubjectId())); - } else { - fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), null, subject.getPhone(), subject.getSubjectId())); - } - }); - } return fidsPhoneDataList; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java index 97fc8c0..836a73d 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java @@ -20,15 +20,19 @@ public class PhoneFidMapService { mapper.registerSubjects(subjects, packageId); } - public List findPhoneFidMaps(Integer packageId) { - return mapper.findPhoneFidMaps(packageId); + public List findPhoneFidMaps(Integer packageId, int offset, int limit) { + return mapper.findPhoneFidMaps(packageId, offset, limit); } - public List findFullPhoneFidMap(Integer packageId) { - return mapper.findFullPhoneFidMap(packageId); + public List findFullPhoneFidMap(Integer packageId, int offset, int limit) { + return mapper.findFullPhoneFidMap(packageId, offset, limit); } public void updatePhoneFidMaps(List subjects) { mapper.updatePhoneFidMaps(subjects); } + + public PhoneFidMap findPhoneFidMap(Integer packageId, String subjectId) { + return mapper.findPhoneFidMap(packageId, subjectId); + } } diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 74aa773..1d80a1c 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -44,6 +44,8 @@ SELECT id, subject_id, phone FROM signals_phone_fid_map WHERE package_id = #{packageId} + ORDER BY id + LIMIT #{limit} OFFSET #{offset} + +