Prepare data for writing

src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java

@@ -0,0 +1,30 @@
+package ru.nbch.credit_tracker.entities.external_request.kafka;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.util.List;
+
+public class InquiryDateDeserializer extends JsonDeserializer<LocalDateTime> {
+    @Override
+    public LocalDateTime deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+        if (p.currentToken() == JsonToken.START_ARRAY) {
+            List<Integer> dateParts = p.readValueAs(List.class);
+            if (dateParts.size() >= 6) {
+                int year = dateParts.get(0);
+                int month = dateParts.get(1);
+                int day = dateParts.get(2);
+                int hour = dateParts.get(3);
+                int minute = dateParts.get(4);
+                int second = dateParts.get(5);
+                int nano = dateParts.size() > 6 ? dateParts.get(6) : 0;
+                return LocalDateTime.of(year, month, day, hour, minute, second, nano);
+            }
+        }
+        return null;
+    }
+}

src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java

@@ -1,10 +1,14 @@
 package ru.nbch.credit_tracker.entities.external_request.kafka;
 
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 
+import java.time.LocalDateTime;
+import java.util.List;
+
 @AllArgsConstructor
 @NoArgsConstructor
 @Getter
@@ -12,4 +16,26 @@ import lombok.Setter;
 public class PackageMessage {
     private Integer fid;
     private String memberCode;
+    @JsonDeserialize(using = InquiryDateDeserializer.class)
+    private LocalDateTime inquiryDate;
+    private String inquiryPurpose;
+    private Long inquiryAmount;
+    private String billingCode;
+    private Integer inquiryType;
+    private List<Person> persons;
+
+    public boolean isNaturalPerson() {
+        return persons != null && !persons.isEmpty();
+    }
+
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Getter
+    @Setter
+    public static class Person {
+        private String lastName;
+        private String firstName;
+        private String middleName;
+        private List<Integer> birthDate;
+    }
 }

src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java

@@ -1,6 +1,7 @@
 package ru.nbch.credit_tracker.model;
 
 import lombok.*;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 import ru.nbch.credit_tracker.enums.EventType;
 
 @Getter
@@ -12,4 +13,12 @@ public class SignalKafkaEvent {
     private EventType eventType;
 
     private UserPackagePhoneData userPackagePhoneData;
+    private PackageMessage packageMessage;
+
+    private String subjectId;
+    private Integer flagDebt5000;
+    private Integer flag90days;
+    private String signalType;
+    private String busCategory;
+    private Integer segment;
 }

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

@@ -8,6 +8,7 @@ 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.app.UserPackage;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 import ru.nbch.credit_tracker.enums.EventType;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
@@ -17,6 +18,7 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.LocalDateTime;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
@@ -82,9 +84,22 @@ public class SignalKafkaEventWriterService {
 
             if (!exists) {
                 Files.createFile(targetFilePathWithExt);
+                String packageName = packagePhoneData.getPackageName();
                 // TODO начать запись
             }
 
+            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 записать сигнал
         } catch (Exception e) {
             log.error("Report was not created for PackageId = {}, PackageName = {}. {}", packagePhoneData.getPackageId(), packagePhoneData.getPackageName(), ExceptionUtils.getStackTrace(e));

src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml

@@ -62,7 +62,7 @@
                      FROM json_to_recordset(#{json}::json) AS t(
                                                                 phone bigint
                          ))
-        select sup.id AS packageId, sup.signal_types AS signalTypes, spr.phone_l AS phone, spr.subject_id AS subjectId
+        select sup.id AS packageId, sup.signal_types AS signalTypes, sup.membercode AS userId, sup.package_name AS packageName, spr.phone_l AS phone, spr.subject_id AS subjectId
         FROM signals_phone_fid_map spr
                  join signals_user_packages sup on sup.id = spr.package_id
         where spr.phone_l in (select * from src) and sup.status = 'ON_MONITORING'