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

Added writing to xml for np

Browse source
This commit is contained in:
Mikhail Trofimov 2026-06-25 11:31:03 +03:00
parent 1f142cf2a8
commit 5982819965
3 changed files with 140 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateSerializer.java Normal file
View file

@ -0,0 +1,36 @@
package ru.nbch.credit_tracker.entities.external_request.kafka;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonSerializer;
import com.fasterxml.jackson.databind.SerializerProvider;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
/**
* Needed only for test purposes (kafka message generator)
*/
public class InquiryDateSerializer extends JsonSerializer<LocalDateTime> {
@Override
public void serialize(LocalDateTime value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
if (value == null) {
gen.writeNull();
return;
}
List<Integer> dateParts = Arrays.asList(
value.getYear(),
value.getMonthValue(),
value.getDayOfMonth(),
value.getHour(),
value.getMinute(),
value.getSecond(),
value.getNano()
);
gen.writeObject(dateParts);
}
}

18
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java
View file

@ -1,12 +1,6 @@
package ru.nbch.credit_tracker.service.mode.monitoring;
import jakarta.annotation.PreDestroy;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
@ -23,6 +17,13 @@ import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.signal.SegmentMapService;
import ru.nbch.credit_tracker.utils.CTUtil;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;
@Slf4j
@Service
public class KafkaMonitoringService {
@ -136,12 +137,15 @@ public class KafkaMonitoringService {
SignalKafkaEvent event = new SignalKafkaEvent();
event.setEventType(EventType.WRITE);
event.setUserPackagePhoneData(userPackagePhoneData);
event.setPackageMessage(message);
event.setSubjectId(subjectId);
event.setFlagDebt5000(flagDebt5000);
event.setFlag90days(flag90days);
event.setSignalType(Integer.toString(signalType));
event.setBusCategory(signalType == 3 ? busCategory : null);
event.setSegment(segmentMap.get(memberCode.substring(0, 6)));
Integer segment = segmentMap.get(memberCode.substring(0, 6));
event.setSegment(segment != null ? segment : -1);
putToQueue(event);
}

113
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java
View file

@ -16,9 +16,13 @@ import ru.nbch.credit_tracker.model.SignalKafkaEvent;
import ru.nbch.credit_tracker.model.UserPackagePhoneData;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import java.io.BufferedWriter;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
@ -28,6 +32,9 @@ import java.util.concurrent.TimeUnit;
@Slf4j
@Service
public class SignalKafkaEventWriterService {
private final static Charset encoding = Charset.forName("windows-1251");
private final static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy");
private final ExecutorService executorService = Executors.newSingleThreadExecutor(
new CustomizableThreadFactory("signal-kafka-event-writer-")
);
@ -78,29 +85,16 @@ public class SignalKafkaEventWriterService {
Path folder = Path.of(config.getLoadDir(),
"tmp", packagePhoneData.getUserId(), packagePhoneData.getPackageName());
Files.createDirectories(folder);
Path targetFilePathWithExt = folder.resolve(packagePhoneData.getUserId() + "_" + packagePhoneData.getPackageName() + ".xml");
Path tmpFilePathWithExt = folder.resolve(packagePhoneData.getUserId() + "_" + packagePhoneData.getPackageName() + ".xml");
boolean exists = targetFilePathWithExt.toFile().exists();
boolean exists = tmpFilePathWithExt.toFile().exists();
if (!exists) {
Files.createFile(targetFilePathWithExt);
String packageName = packagePhoneData.getPackageName();
// TODO начать запись
Files.createFile(tmpFilePathWithExt);
writeStart(tmpFilePathWithExt, event);
}
String id = packagePhoneData.getSubjectId();
Long clientId = packagePhoneData.getPhone();
Integer flagDebt5000 = event.getFlagDebt5000();
Integer flag90day = event.getFlag90days();
PackageMessage message = event.getPackageMessage();
String signalType = event.getSignalType();
String busCategory = event.getBusCategory();
LocalDateTime inquiryDate = message.getInquiryDate();
String inquiryPurpose = message.getInquiryPurpose();
Integer segment = event.getSegment();
// TODO записать сигнал
writeNpPart(tmpFilePathWithExt, event);
} catch (Exception e) {
log.error("Report was not created for PackageId = {}, PackageName = {}. {}", packagePhoneData.getPackageId(), packagePhoneData.getPackageName(), ExceptionUtils.getStackTrace(e));
}
@ -110,19 +104,96 @@ public class SignalKafkaEventWriterService {
* Упаковывает все результирующие xml в архив
*/
private void rotate() {
List<UserPackage> userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); // TODO + PROCESSING_REPEATED?
List<UserPackage> userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
for (UserPackage userPackage : userPackages) {
Path targetFilePath = Path.of(config.getLoadDir(),
Path tmpFilePath = Path.of(config.getLoadDir(),
"tmp", userPackage.getUserId(), userPackage.getPackageName())
.resolve(userPackage.getUserId() + "_" + userPackage.getPackageName() + ".xml");
if (targetFilePath.toFile().exists()) {
if (tmpFilePath.toFile().exists()) {
writeEnd(tmpFilePath);
// TODO дописать, переименовать с timestamp, зазиповать и выложить в папку
}
}
// TODO update signals_user_packages
}
private void writeStart(Path filePath, SignalKafkaEvent event) {
UserPackagePhoneData packagePhoneData = event.getUserPackagePhoneData();
String packageName = packagePhoneData.getPackageName();
try (BufferedWriter writer = Files.newBufferedWriter(filePath, encoding,
StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING)) {
writer.write("<?xml version=\"1.0\" encoding=\"" + encoding.name() + "\"?>\n");
writer.write("<MonitoringSignalResponse>\n");
writer.write(" <PackageName>" + escapeXml(packageName) + "</PackageName>\n");
writer.write(" <Subjects>\n");
} catch (Exception e) {
log.error("Unable to write start part of file {}", filePath, e);
}
}
private void writeNpPart(Path filePath, SignalKafkaEvent event) {
UserPackagePhoneData packagePhoneData = event.getUserPackagePhoneData();
String id = packagePhoneData.getSubjectId();
Long clientId = packagePhoneData.getPhone();
Integer flagDebt5000 = event.getFlagDebt5000();
Integer flag90days = event.getFlag90days();
PackageMessage message = event.getPackageMessage();
String signalType = event.getSignalType();
String busCategory = event.getBusCategory();
LocalDateTime inquiryDate = message.getInquiryDate();
String inquiryPurpose = message.getInquiryPurpose();
Integer segment = event.getSegment();
try (BufferedWriter writer = Files.newBufferedWriter(filePath, encoding, StandardOpenOption.APPEND)) {
writer.write(" <Subject>\n");
writer.write(" <Id>" + escapeXml(id) + "</Id>\n");
writer.write(" <ClientId>" + clientId + "</ClientId>\n");
writer.write(" <FlagDebt5000>" + flagDebt5000 + "</FlagDebt5000>\n");
writer.write(" <Flag90_12>" + flag90days + "</Flag90_12>\n");
writer.write(" <Signals>\n");
writer.write(" <Signal>\n");
writer.write(" <SignalType>" + escapeXml(signalType) + "</SignalType>\n");
if (busCategory != null) {
writer.write(" <BusCategory>" + escapeXml(busCategory) + "</BusCategory>\n");
}
writer.write(" <InquiryDate>" + formatter.format(inquiryDate) + "</InquiryDate>\n");
writer.write(" <InquiryPurpose>" + escapeXml(inquiryPurpose) + "</InquiryPurpose>\n");
writer.write(" <Segment>" + segment + "</Segment>\n");
writer.write(" </Signal>\n");
writer.write(" </Signals>\n");
writer.write(" </Subject>\n");
} catch (Exception e) {
log.error("Unable to write np part of file {}", filePath, e);
}
}
private void writeEnd(Path filePath) {
try (BufferedWriter writer = Files.newBufferedWriter(filePath, encoding, StandardOpenOption.APPEND)) {
writer.write(" </Subjects>\n");
writer.write("</MonitoringSignalResponse>\n");
} catch (Exception e) {
log.error("Unable to write end part of file {}", filePath, e);
}
}
// Вспомогательный метод для защиты от спецсимволов (&, <, >, ", ')
private String escapeXml(String input) {
if (input == null) return "";
return input.replace("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;")
.replace("\"", "&quot;")
.replace("'", "&apos;");
}
private SignalKafkaEvent take() {
try {
return signalEventsQueue.take();