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

Update MonitoringService to read nbki responses by batches

Browse source
This commit is contained in:
Mikhail Trofimov 2025-07-17 17:05:39 +03:00
parent f2a827c8f2
commit f052841954
14 changed files with 340 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/main/java/ru/nbch/credit_tracker/config/xml/XmlConfig.java
View file

@ -5,7 +5,6 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.entities.response.errors.MonitoringError;
import ru.nbch.credit_tracker.entities.response.monitoring_signal_response.MonitoringSignalResponse;
import ru.nbch.credit_tracker.entities.response.online_download.OnlineDownloadReport;
import ru.nbch.credit_tracker.entities.response.online_report.SgnlOnlineReport;
import ru.nbch.credit_tracker.entities.response.success.MonitoringResponse;
import ru.nbch.credit_tracker.service.xml.jaxb.IXmlProcessor;
@ -23,7 +22,6 @@ public class XmlConfig {
public IXmlProcessor xmlProcessor() throws JAXBException {
return new XmlProcessorImplAnyClasses(Charset.forName("windows-1251"),
// unmarshalling classes
OnlineDownloadReport.class,
SgnlOnlineReport.class,
// marshalling classes
MonitoringError.class,
@ -36,6 +34,7 @@ public class XmlConfig {
List<StaxReader<?>> staxReader = new ArrayList<>();
staxReader.add(new SgnlReportStatusStaxReader());
staxReader.add(new MonitoringRequestStaxReader());
staxReader.add(new OnlineDownloadReportPersonsStaxReader());
return new XmlProcessorStaxImpl(staxReader);
}
}

14
src/main/java/ru/nbch/credit_tracker/entities/response/online_download/Inquiry.java
View file

@ -42,4 +42,18 @@ public class Inquiry {
return this.signals;
}
private Signal currentSignal;
public void createCurrentSignal() {
currentSignal = new Signal();
}
public Signal getCurrentSignal() {
return currentSignal;
}
public void finalizeSignal() {
getSignals().add(currentSignal);
}
}

14
src/main/java/ru/nbch/credit_tracker/entities/response/online_download/Person.java
View file

@ -27,4 +27,18 @@ public class Person {
}
return this.inquiries;
}
private Inquiry currentInquiry;
public void createCurrentInquiry() {
currentInquiry = new Inquiry();
}
public Inquiry getCurrentInquiry() {
return currentInquiry;
}
public void finalizeInquiry() {
getInquiries().add(currentInquiry);
}
}

13
src/main/java/ru/nbch/credit_tracker/entities/response/online_download/Persons.java
View file

@ -24,5 +24,18 @@ public class Persons {
return this.persons;
}
private Person currentPerson;
public void createCurrentPerson() {
currentPerson = new Person();
}
public Person getCurrentPerson() {
return currentPerson;
}
public void finalizePerson() {
getPersons().add(currentPerson);
}
}

89
src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java
View file

@ -10,18 +10,19 @@ import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
import ru.nbch.credit_tracker.entities.dto.UserPackage;
import ru.nbch.credit_tracker.entities.response.monitoring_signal_response.*;
import ru.nbch.credit_tracker.entities.response.online_download.Inquiry;
import ru.nbch.credit_tracker.entities.response.online_download.OnlineDownloadReport;
import ru.nbch.credit_tracker.entities.response.online_download.Person;
import ru.nbch.credit_tracker.entities.response.online_download.Persons;
import ru.nbch.credit_tracker.entities.response.online_report.Report;
import ru.nbch.credit_tracker.entities.response.online_report.SgnlOnlineReport;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
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 java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.*;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.Executors;
@ -38,16 +39,19 @@ public class MonitoringService {
private final SignalRestService signalRestService;
private final SignalService signalService;
private final IXmlProcessor xmlProcessor;
private final StaxXmlProcessor staxXmlProcessor;
private final CreditTrackerSettings config;
private final Integer batchSize = 10000;
public MonitoringService(SignalRestService signalRestService,
SignalService signalService,
IXmlProcessor xmlProcessor,
StaxXmlProcessor staxXmlProcessor,
CreditTrackerSettings config) {
this.signalRestService = signalRestService;
this.signalService = signalService;
this.xmlProcessor = xmlProcessor;
this.staxXmlProcessor = staxXmlProcessor;
this.config = config;
}
@ -89,21 +93,23 @@ public class MonitoringService {
log.info("Last Downloaded Report Name is empty");
}
OnlineDownloadReport report = null;
try {
String reportPath = null;
if (StringUtils.isBlank(userPackage.getLastDownloadedReport()) && newReport.isPresent()) {
report = loadReport(newReport.get()); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным
reportPath = loadReport(userPackage.getId(), newReport.get()); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным
} else if (lastDownloadedReportDate != null && newReport.isPresent()) {
LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport.get());
if (newReportDate.isAfter(lastDownloadedReportDate)) {
report = loadReport(newReport.get()); // полученный отчет новее записанного в базу
reportPath = loadReport(userPackage.getId(), newReport.get()); // полученный отчет новее записанного в базу
}
}
if (report == null) {
if (reportPath == null) {
log.info("Report was not loaded for package {}", userPackage.getPackageName());
continue;
}
MonitoringSignalResponse response = generateResponse(userPackage, report.getPersons().getPersons());
MonitoringSignalResponse response = generateResponse(userPackage, reportPath);
String fileName = createResponseFile(userPackage, response);
if (fileName != null) {
log.info("New report fileName: {}", fileName);
@ -112,19 +118,48 @@ public class MonitoringService {
} else {
log.info("Report was not created for package {}", userPackage.getPackageName());
}
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
log.info("Finish monitoring");
}
private MonitoringSignalResponse generateResponse(UserPackage userPackage, List<Person> persons) {
private MonitoringSignalResponse generateResponse(UserPackage userPackage, String reportPath) {
MonitoringSignalResponse response = new MonitoringSignalResponse();
response.setPackageName(userPackage.getPackageName()); // Имя пакета, переданного при установке
Subjects subjects = new Subjects();
try (InputStream in = Files.newInputStream(Path.of(reportPath))) {
staxXmlProcessor.read(in, Persons.class,
personsData -> {
List<Subject> subjectsBatch = processPersons(userPackage, personsData.getPersons());
subjects.getSubjects().addAll(subjectsBatch);
personsData.getPersons().clear();
},
10000,
"/Report/Persons/Person");
} catch (IOException e) {
throw new RuntimeException("Error while parsing nbki downloaded report", e);
}
response.setSubjects(subjects);
try {
Files.delete(Path.of(reportPath));
} catch (IOException e) {
log.warn("File {} hasn't been deleted", reportPath);
}
return response;
}
private List<Subject> processPersons(UserPackage userPackage, List<Person> persons) {
List<Subject> subjects = new ArrayList<>();
for (Person person : persons) {
Signals signals = new Signals();
for (Inquiry inquiry : person.getInquiries()) {
// Отбираем только тех фл, у которых bus_category in (MFO, MKK, MFK) и есть сигналы с кодом 16
if (!CTUtil.checkBusCategory(inquiry.getBusCategory()) && inquiry.getSignals().stream().anyMatch(signal -> signal.getN().equals(16))) {
if (!CTUtil.checkBusCategory(inquiry.getBusCategory()) || inquiry.getSignals().stream().anyMatch(signal -> signal.getN().equals(16))) {
continue;
}
Signal signal = new Signal();
@ -144,12 +179,11 @@ public class MonitoringService {
subject.setFlagDebt5000(subjectData.getFlagDebt5000());
subject.setFlag9012(subjectData.getFlag90days());
subject.setSignals(signals);
subjects.getSubjects().add(subject);
subjects.add(subject);
}
}
}
response.setSubjects(subjects);
return response;
return subjects;
}
private String createResponseFile(UserPackage userPackage, MonitoringSignalResponse response) {
@ -202,22 +236,35 @@ public class MonitoringService {
log.info("Signal hits updated for package {}", userPackage.getPackageName());
}
private OnlineDownloadReport loadReport(String reportName) {
/**
* Loads .gz report from nbki service, unzips it to tmp_folder
*
* @return path to unzipped file. Example: <loadDir>/tmp_folder/<package_id>/0001XX_MY_BATCH_204.1.online.2025-07-08T20-00-46.xml.gz.xml
*/
public String loadReport(Integer packageId, String reportName) {
byte[] archive = signalRestService.downloadReport(reportName);
if (archive == null) {
return null;
}
OnlineDownloadReport report;
try (GZIPInputStream zipInputStream = new GZIPInputStream(new ByteArrayInputStream(archive))) {
report = xmlProcessor.read(zipInputStream, OnlineDownloadReport.class);
String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator + packageId + File.separator;
String filePath = folderPath + reportName + ".xml";
CTUtil.createFolder(folderPath);
} catch (Exception e) {
try (InputStream in = new GZIPInputStream(new ByteArrayInputStream(archive));
OutputStream out = Files.newOutputStream(Path.of(filePath))) {
byte[] buffer = new byte[65536];
int bytesRead;
while ((bytesRead = in.read(buffer)) > 0) {
out.write(buffer, 0, bytesRead);
}
return filePath;
} catch (IOException e) {
log.error(e.getMessage(), e);
return null;
}
return report;
}
}

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

@ -20,11 +20,9 @@ import ru.nbch.credit_tracker.service.signal.HitsService;
import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
import ru.nbch.credit_tracker.service.signal.UserPackageService;
import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter;
import ru.nbch.credit_tracker.utils.CTUtil;
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.*;
import java.util.function.Consumer;
@ -175,15 +173,7 @@ public class SignalService {
String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator + packageId + File.separator;
String filePath = folderPath + UUID.randomUUID() + "_output.xml";
File folder = new File(folderPath);
if (!folder.exists()) {
try {
Files.createDirectories(Path.of(folderPath));
} catch (IOException e) {
log.error("Could not create folder: {}", folderPath, e);
throw new RuntimeException(e);
}
}
CTUtil.createFolder(folderPath);
Signals signals = buildSignals(packageId);

26
src/main/java/ru/nbch/credit_tracker/service/xml/stax/BaseStaxReader.java
View file

@ -6,15 +6,22 @@ import javax.xml.stream.XMLStreamReader;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Iterator;
import java.util.function.Consumer;
public abstract class BaseStaxReader<T> implements StaxReader<T> {
private final Deque<String> stack = new ArrayDeque<>();
protected final StringBuilder buffer = new StringBuilder();
protected abstract T createObject();
@Override
public T read(XMLStreamReader reader) throws XMLStreamException {
return read(reader, null, null, null);
}
public T read(XMLStreamReader reader, Consumer<T> objProcessor, Integer batchSize, String endBatchElementPath) throws XMLStreamException {
T obj = createObject();
int objCount = 0;
try {
while (reader.hasNext()) {
boolean breakExecution = false;
@ -29,14 +36,31 @@ public abstract class BaseStaxReader<T> implements StaxReader<T> {
String text = reader.getText().trim();
if (!text.isEmpty()) buffer.append(text);
}
case XMLStreamConstants.ATTRIBUTE -> {
breakExecution = innerProcess(reader, obj);
}
case XMLStreamConstants.END_ELEMENT -> {
breakExecution = innerProcess(reader, obj);
String lastElementPath = getPath();
endElement();
if (objProcessor != null) {
if (endBatchElementPath.equals(lastElementPath)) {
objCount++;
if (objCount == batchSize) {
objProcessor.accept(obj);
objCount = 0;
}
}
}
}
default -> {
}
default -> {}
}
if (breakExecution) break;
}
if (objProcessor != null) {
objProcessor.accept(obj);
}
return obj;
} catch (Exception e) {
stack.clear();

77
src/main/java/ru/nbch/credit_tracker/service/xml/stax/OnlineDownloadReportPersonsStaxReader.java Normal file
View file

@ -0,0 +1,77 @@
package ru.nbch.credit_tracker.service.xml.stax;
import lombok.extern.slf4j.Slf4j;
import ru.nbch.credit_tracker.entities.response.online_download.Persons;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamReader;
@Slf4j
public class OnlineDownloadReportPersonsStaxReader extends BaseStaxReader<Persons> {
@Override
protected Persons createObject() {
return new Persons();
}
@Override
protected boolean innerProcess(XMLStreamReader reader, Persons obj) throws XMLStreamException {
String text = buffer.toString();
String currPath = getPath();
switch (currPath) {
case "/Report/Persons/Person" ->
obj.finalizePerson();
case "/Report/Persons/Person/Inquiry" ->
obj.getCurrentPerson().finalizeInquiry();
case "/Report/Persons/Person/Inquiry/Signal" ->
obj.getCurrentPerson().getCurrentInquiry().finalizeSignal();
}
return false;
}
@Override
protected void innerProcessStartElement(XMLStreamReader reader, Persons obj) throws XMLStreamException {
String currPath = getPath();
switch (currPath) {
case "/Report/Persons/Person" ->
obj.createCurrentPerson();
case "/Report/Persons/Person/Inquiry" ->
obj.getCurrentPerson().createCurrentInquiry();
case "/Report/Persons/Person/Inquiry/Signal" ->
obj.getCurrentPerson().getCurrentInquiry().createCurrentSignal();
}
for (int i = 0; i < reader.getAttributeCount(); i++) {
String text = reader.getAttributeValue(i);
switch (reader.getAttributeLocalName(i)) {
case "uid" ->
obj.getCurrentPerson().setUid(text);
case "date" ->
obj.getCurrentPerson().getCurrentInquiry().setDate(text);
case "bus_category" ->
obj.getCurrentPerson().getCurrentInquiry().setBusCategory(text);
case "inq_p" -> {
try {
int inqP = Integer.parseInt(text);
obj.getCurrentPerson().getCurrentInquiry().setInqP(inqP);
} catch (NumberFormatException e) {
log.warn("Couldn't parse inqP: {}", text);
}
}
case "n" -> {
try {
int n = Integer.parseInt(text);
obj.getCurrentPerson().getCurrentInquiry().getCurrentSignal().setN(n);
} catch (NumberFormatException e) {
log.warn("Couldn't parse signal n: {}", text);
}
}
}
}
}
@Override
public Class<Persons> type() {
return Persons.class;
}
}

2
src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxReader.java
View file

@ -2,8 +2,10 @@ package ru.nbch.credit_tracker.service.xml.stax;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamReader;
import java.util.function.Consumer;
public interface StaxReader<T> {
Class<T> type();
T read(XMLStreamReader xmlStreamReader) throws XMLStreamException;
T read(XMLStreamReader reader, Consumer<T> objProcessor, Integer batchSize, String endBatchElementPath) throws XMLStreamException;
}

2
src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlProcessor.java
View file

@ -1,9 +1,11 @@
package ru.nbch.credit_tracker.service.xml.stax;
import java.io.InputStream;
import java.util.function.Consumer;
public interface StaxXmlProcessor {
<Z> Z read(String xml, Class<Z> clazz);
<Z> Z read(byte[] xml, Class<Z> clazz);
<Z> Z read(InputStream inputStream, Class<Z> clazz);
<Z> Z read(InputStream inputStream, Class<Z> clazz, Consumer<Z> objProcessor, Integer batchSize, String endBatchElementPath);
}

17
src/main/java/ru/nbch/credit_tracker/service/xml/stax/XmlProcessorStaxImpl.java
View file

@ -14,6 +14,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
public class XmlProcessorStaxImpl implements StaxXmlProcessor {
private final Logger log = LoggerFactory.getLogger(getClass());
@ -83,4 +84,20 @@ public class XmlProcessorStaxImpl implements StaxXmlProcessor {
throw new RuntimeException(e);
}
}
@Override
public <Z> Z read(InputStream inputStream, Class<Z> clazz, Consumer<Z> objProcessor, Integer batchSize, String endBatchElementPath) {
try {
XMLStreamReader xmlStreamReader = xmlInputFactory.createXMLStreamReader(inputStream);
@SuppressWarnings("unchecked")
StaxReader<Z> staxProcessor = (StaxReader<Z>) staxReaders.get(clazz);
if (staxProcessor != null) {
return staxProcessor.read(xmlStreamReader, objProcessor, batchSize, endBatchElementPath);
}
throw new IllegalStateException("no stax reader registered for " + clazz);
} catch (Exception e) {
log.error(ExceptionUtils.getStackTrace(e));
throw new RuntimeException(e);
}
}
}

18
src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java
View file

@ -1,7 +1,12 @@
package ru.nbch.credit_tracker.utils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
@ -10,6 +15,7 @@ import java.util.Date;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@Slf4j
public class CTUtil {
private static DateTimeFormatter onlineReportDateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH-mm-ss");
@ -65,4 +71,16 @@ public class CTUtil {
}
return Date.from(date.atStartOfDay(ZoneId.systemDefault()).toInstant());
}
public static void createFolder(String folderPath) {
File folder = new File(folderPath);
if (!folder.exists()) {
try {
Files.createDirectories(Path.of(folderPath));
} catch (IOException e) {
log.error("Could not create folder: {}", folderPath, e);
throw new RuntimeException(e);
}
}
}
}

31
src/test/java/ru/nbch/credit_tracker/stax/StaxTest.java
View file

@ -8,10 +8,9 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.nbch.credit_tracker.entities.request.monitoring_request.MonitoringRequest;
import ru.nbch.credit_tracker.entities.request.monitoring_request.Subject;
import ru.nbch.credit_tracker.service.xml.stax.MonitoringRequestStaxReader;
import ru.nbch.credit_tracker.service.xml.stax.StaxReader;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
import ru.nbch.credit_tracker.service.xml.stax.XmlProcessorStaxImpl;
import ru.nbch.credit_tracker.entities.response.online_download.Person;
import ru.nbch.credit_tracker.entities.response.online_download.Persons;
import ru.nbch.credit_tracker.service.xml.stax.*;
import javax.xml.stream.XMLStreamException;
import java.io.IOException;
@ -34,6 +33,7 @@ public class StaxTest {
public StaxXmlProcessor xmlProcessor() {
List<StaxReader<?>> staxReader = new ArrayList<>();
staxReader.add(new MonitoringRequestStaxReader());
staxReader.add(new OnlineDownloadReportPersonsStaxReader());
return new XmlProcessorStaxImpl(staxReader);
}
}
@ -84,4 +84,27 @@ public class StaxTest {
Assertions.assertNull(monitoringRequest.getSubjects().getSubject().get(0).getId());
}
@Test
void testParseNbkiMonitoringReport() {
List<Person> firstPersonList = new ArrayList<>();
staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class,
personsData -> {
Assertions.assertTrue(personsData.getPersons().size() == 1 || personsData.getPersons().isEmpty());
firstPersonList.addAll(personsData.getPersons());
personsData.getPersons().clear();
},
1,
"/Report/Persons/Person");
List<Person> secondPersonList = new ArrayList<>();
staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class,
personsData -> {
secondPersonList.addAll(personsData.getPersons());
personsData.getPersons().clear();
},
3,
"/Report/Persons/Person");
Assertions.assertEquals(firstPersonList.size(), secondPersonList.size());
}
}

44
src/test/resources/monitoring_request/xml/nbki_monitoring_report.xml Normal file
View file

@ -0,0 +1,44 @@
<?xml version="1.0" encoding="windows-1251"?>
<Report>
<Id>0001XX_MY_BATCH_204</Id>
<Version>1</Version>
<User>0001ZZ000001</User>
<ReportCreated>08.07.2025 20:00:33</ReportCreated>
<Persons>
<Person uid="abc123">
<Inquiry own="0" inq_p="1" amount="30000" currency="RUB" serial_num="2503180011" date="01.01.2025" bus_category="BANK" inq_t="1" channel="XML">
<Signal n="16"/>
<Signal n="25"/>
<Signal n="26"/>
</Inquiry>
<Inquiry own="0" inq_p="2" amount="30000" currency="RUB" serial_num="2503180014" date="02.02.2025" bus_category="MFK" inq_t="3" channel="XML">
<Signal n="25"/>
</Inquiry>
</Person>
<Person uid="abc124">
<Inquiry own="0" inq_p="3" amount="30000" currency="RUB" serial_num="2503180011" date="03.03.2025" bus_category="MKK" inq_t="1" channel="XML">
<Signal n="16"/>
<Signal n="25"/>
<Signal n="15"/>
<Signal n="27"/>
</Inquiry>
</Person>
<Person uid="abc125">
<Inquiry own="0" inq_p="4" amount="30000" currency="RUB" serial_num="2503180011" date="04.04.2025" bus_category="CAT" inq_t="1" channel="XML">
<Signal n="16"/>
<Signal n="25"/>
</Inquiry>
</Person>
<Person uid="abc126">
<Inquiry own="0" inq_p="5" amount="30000" currency="RUB" serial_num="2503180011" date="05.05.2025" bus_category="BBB" inq_t="1" channel="XML">
<Signal n="16"/>
<Signal n="25"/>
</Inquiry>
</Person>
<Person uid="abc127">
<Inquiry own="0" inq_p="6" amount="30000" currency="RUB" serial_num="2503180011" date="06.06.2025" bus_category="CCC" inq_t="1" channel="XML">
<Signal n="16"/>
</Inquiry>
</Person>
</Persons>
</Report>