diff --git a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateSerializer.java b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateSerializer.java new file mode 100644 index 0000000..710deaa --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateSerializer.java @@ -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 { + + @Override + public void serialize(LocalDateTime value, JsonGenerator gen, SerializerProvider serializers) throws IOException { + if (value == null) { + gen.writeNull(); + return; + } + + List dateParts = Arrays.asList( + value.getYear(), + value.getMonthValue(), + value.getDayOfMonth(), + value.getHour(), + value.getMinute(), + value.getSecond(), + value.getNano() + ); + + gen.writeObject(dateParts); + } +} \ No newline at end of file diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index 7f48c21..36ccbf6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -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); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java index 5eaa42e..b051522 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java @@ -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 userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); // TODO + PROCESSING_REPEATED? + List userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); for (UserPackage userPackage : userPackages) { - Path targetFilePath = Path.of(config.getLoadDir(), - "tmp", userPackage.getUserId(), userPackage.getPackageName()) + 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("\n"); + writer.write("\n"); + writer.write(" " + escapeXml(packageName) + "\n"); + writer.write(" \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(" \n"); + writer.write(" " + escapeXml(id) + "\n"); + writer.write(" " + clientId + "\n"); + writer.write(" " + flagDebt5000 + "\n"); + writer.write(" " + flag90days + "\n"); + writer.write(" \n"); + writer.write(" \n"); + writer.write(" " + escapeXml(signalType) + "\n"); + + if (busCategory != null) { + writer.write(" " + escapeXml(busCategory) + "\n"); + } + + writer.write(" " + formatter.format(inquiryDate) + "\n"); + writer.write(" " + escapeXml(inquiryPurpose) + "\n"); + writer.write(" " + segment + "\n"); + writer.write(" \n"); + writer.write(" \n"); + writer.write(" \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(" \n"); + writer.write("\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("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace("\"", """) + .replace("'", "'"); + } + private SignalKafkaEvent take() { try { return signalEventsQueue.take();