diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/FileService.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/FileService.java new file mode 100644 index 000000000..7d76f4ed8 --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/FileService.java @@ -0,0 +1,65 @@ +package ru.spcex.clearing.csv.importer.component; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import static java.nio.file.StandardCopyOption.REPLACE_EXISTING; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.csv.importer.config.settings.CsvImporterSettings; +import ru.spcex.platform.utils.log.ExceptionUtils; + + +@Component +public class FileService { + private static final Logger log = LoggerFactory.getLogger(FileService.class); + private final Path outDir; + private final Path errDir; + + public FileService(CsvImporterSettings settings) { + this.outDir = Path.of(settings.getOutDir()); + this.errDir = Path.of(settings.getErrDir()); + checkDir(outDir); + checkDir(errDir); + } + + public void moveSuccess(Path file) { + try { + Files.move(file, outDir.resolve(file.getFileName()), REPLACE_EXISTING); + } catch (IOException e) { + log.error("failed to move to {}, trying to move to {}: {}", outDir, errDir, ExceptionUtils.getStackTrace(e)); + moveError(file); + //notification? + } + } + + public void moveError(Path file) { + try { + Files.move(file, errDir.resolve(file.getFileName()), REPLACE_EXISTING); + } catch (IOException e) { + log.error("failed to move to {}: {}", errDir, ExceptionUtils.getStackTrace(e)); + //notification? + } + } + + public static boolean isReady(Path file) { + try { + var modified = Files.getLastModifiedTime(file).toInstant(); + return modified.isBefore(java.time.Instant.now().minusSeconds(10)); + } catch (IOException e) { + log.error("failed to read last modified time {}: {}", + file, ExceptionUtils.getStackTrace(e)); + return false; + } + } + + public static void checkDir(Path dir) { + if (!dir.toFile().exists() || !dir.toFile().isDirectory()) { + throw new IllegalStateException("directory " + dir + " is not a directory"); + } + } +} + + + diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/SOrdersScheduler.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/SOrdersScheduler.java new file mode 100644 index 000000000..20e05321f --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/SOrdersScheduler.java @@ -0,0 +1,77 @@ +package ru.spcex.clearing.csv.importer.component; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import ru.clearing.classes.statics.data.misc.SOrders; +import ru.spcex.clearing.csv.importer.component.tro.data.SOrder; +import ru.spcex.clearing.csv.importer.component.tro.parser.TroParserType; +import ru.spcex.clearing.csv.importer.component.tro.processor.TroProcessor; +import ru.spcex.clearing.csv.importer.component.tro.processor.TroProcessorProvider; +import ru.spcex.clearing.csv.importer.config.settings.CsvImporterSettings; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.utils.log.ExceptionUtils; + +@Component +public class SOrdersScheduler { + private final static Logger log = LoggerFactory.getLogger(SOrdersScheduler.class); + + private final Path importDir; + private final FileService fileService; + private final TroProcessor processor; + private final Imdg sOrderImdg; + + @Autowired + public SOrdersScheduler(CsvImporterSettings settings, FileService fileService, TroProcessorProvider provider, ImdgProvider imdgProvider) { + this.importDir = Path.of(settings.getSrcDir()); + this.fileService = fileService; + processor = provider.createProcessor(TroParserType.SOrder, SOrder.class); + this.sOrderImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SOrders, SOrders.class); + } + + @Scheduled(fixedDelayString = "${csv-importer.poll-period:5000}") + public void pollDirectory() { + try (var paths = Files.list(importDir)) { + paths.filter(Files::isRegularFile) + .filter(FileService::isReady) + .sorted() + .forEach(this::processFile); + } catch (Throwable e) { + log.error("Failed to scan import directory {}: {}", + importDir, ExceptionUtils.getStackTrace(e)); + } + } + + private void processFile(Path path) { + try { + List sOrders = processor.readFile(path); + for (SOrder orderFromFile : sOrders) { + Long transId = orderFromFile.getTransId(); + SOrders order = sOrderImdg.getSingleObjectByID(transId); + boolean wasFound = order != null; + if (order == null) { + order = new SOrders(); + order.setId(transId); + } + order.setState(orderFromFile.getStatus()); + order.setOrderNum(orderFromFile.getOrderNumber()); + if (wasFound) { + sOrderImdg.update(order); + } else { + sOrderImdg.insert(order); + } + } + fileService.moveSuccess(path); + } catch (Throwable e) { + log.error("error for {}: {}", path, ExceptionUtils.getStackTrace(e)); + fileService.moveError(path); + } + } +} diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/data/SOrder.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/data/SOrder.java new file mode 100644 index 000000000..48d30e144 --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/data/SOrder.java @@ -0,0 +1,49 @@ +package ru.spcex.clearing.csv.importer.component.tro.data; + +public class SOrder { + private Long transId; + private String status; + private Long orderNumber; + private String transName; + private String description; + + public Long getTransId() { + return transId; + } + + public void setTransId(Long transId) { + this.transId = transId; + } + + public String getStatus() { + return status; + } + + public Long getOrderNumber() { + return orderNumber; + } + + public void setOrderNumber(Long orderNumber) { + this.orderNumber = orderNumber; + } + + public void setStatus(String status) { + this.status = status; + } + + public String getTransName() { + return transName; + } + + public void setTransName(String transName) { + this.transName = transName; + } + + public String getDescription() { + return description; + } + + public void setDescription(String description) { + this.description = description; + } +} diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/AbstractTroParser.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/AbstractTroParser.java new file mode 100644 index 000000000..9178f4d3c --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/AbstractTroParser.java @@ -0,0 +1,121 @@ +package ru.spcex.clearing.csv.importer.component.tro.parser; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.HashMap; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.spcex.platform.utils.enumeration.IEnumKey; + +public abstract class AbstractTroParser implements TroParser { + private final Logger log = LoggerFactory.getLogger(getClass()); + + private static final DateTimeFormatter DEFAULT_DATE_TIME_FORMAT = + DateTimeFormatter.ofPattern("dd.MM.yyyy H:mm:ss"); + + private String splitChar = ";"; + + + protected void setSplitChar(String splitChar) { + this.splitChar = splitChar; + } + + protected Map parseKeyValueLine(String line) { + Map res = new HashMap<>(); + String[] split = line.split(splitChar); + String key; + String value; + for (String part : split) { + part = part.trim(); + int firstQuote = part.indexOf("\""); + if (firstQuote != -1) { //string in quotes + int lastQuote = part.lastIndexOf("\""); + if (lastQuote == firstQuote) { + log.error("only one quote found in line: skipping `{}`...", part); + continue; + } + key = part.substring(0, firstQuote).toUpperCase().trim(); + value = part.substring(firstQuote + 1, lastQuote); + res.put(key, value); + } + int equalSign = part.indexOf('='); + if (equalSign == -1) { + log.error("no equal sign found in line: skipping `{}`...", part); + continue; + } + key = part.substring(0, equalSign).toUpperCase().trim(); + value = part.substring(equalSign + 1).trim(); + res.put(key, value); + } + return res; + } + + protected String getRequired(Map values, IEnumKey key) { + String value = values.get(key.getKey()); + if (value == null || value.isBlank()) { + throw new IllegalArgumentException("Required field is missing: " + key.getKey()); + } + return value.trim(); + } + + protected String getOptional(Map values, IEnumKey key) { + String value = values.get(key.getKey()); + return value == null ? null : value.trim(); + } + + protected Long extractLong(Map values, IEnumKey key) { + String value = getOptional(values, key); + if (value == null || value.isBlank()) { + return null; + } + try { + return Long.valueOf(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Field '%s' has invalid Long value: '%s'".formatted(key.getKey(), value), e); + } + } + + protected Integer extractInteger(Map values, IEnumKey key) { + String value = getOptional(values, key); + if (value == null || value.isBlank()) { + return null; + } + try { + return Integer.valueOf(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Field '%s' has invalid Integer value: '%s'".formatted(key.getKey(), value), e); + } + } + + protected BigDecimal extractBigDecimal(Map values, IEnumKey key) { + String value = getOptional(values, key); + if (value == null || value.isBlank()) { + return null; + } + try { + return new BigDecimal(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Field '%s' has invalid BigDecimal value: '%s'".formatted(key.getKey(), value), e); + } + } + + protected String extractString(Map values, IEnumKey key) { + return getOptional(values, key); + } + + protected LocalDateTime extractDateTime(Map values, IEnumKey key) { + String value = getOptional(values, key); + if (value == null || value.isBlank()) { + return null; + } + try { + return LocalDateTime.parse(value, DEFAULT_DATE_TIME_FORMAT); + } catch (Exception e) { + throw new IllegalArgumentException( + "Field '%s' has invalid LocalDateTime value: '%s'".formatted(key.getKey(), value), e + ); + } + } +} \ No newline at end of file diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParser.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParser.java new file mode 100644 index 000000000..a6a81b78e --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParser.java @@ -0,0 +1,6 @@ +package ru.spcex.clearing.csv.importer.component.tro.parser; + +public interface TroParser { + T parse(String line); + TroParserType getType(); +} \ No newline at end of file diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParserType.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParserType.java new file mode 100644 index 000000000..a1d936eac --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParserType.java @@ -0,0 +1,7 @@ +package ru.spcex.clearing.csv.importer.component.tro.parser; + +public enum TroParserType { + SOrder + ; + +} diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/impl/SOrderParser.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/impl/SOrderParser.java new file mode 100644 index 000000000..23cfd53f6 --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/impl/SOrderParser.java @@ -0,0 +1,39 @@ +package ru.spcex.clearing.csv.importer.component.tro.parser.impl; + +import java.util.Map; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.csv.importer.component.tro.data.SOrder; +import ru.spcex.clearing.csv.importer.component.tro.parser.AbstractTroParser; +import ru.spcex.clearing.csv.importer.component.tro.parser.TroParserType; +import ru.spcex.platform.utils.enumeration.IEnumKey; + +@Component +public class SOrderParser extends AbstractTroParser { + + @Override + public TroParserType getType() { + return TroParserType.SOrder; + } + + @Override + public SOrder parse(String line) { + Map map = parseKeyValueLine(line); + SOrder twk = new SOrder(); + twk.setStatus(extractString(map, SOrdersFileField.STATUS)); + twk.setDescription(extractString(map, SOrdersFileField.DESCRIPTION)); + twk.setTransId(extractLong(map, SOrdersFileField.TRANS_ID)); + twk.setTransName(extractString(map, SOrdersFileField.TRANS_NAME)); + twk.setOrderNumber(extractLong(map, SOrdersFileField.ORDER_NUMBER)); + return twk; + } + + public enum SOrdersFileField implements IEnumKey { + TRANS_ID, STATUS, TRANS_NAME, DESCRIPTION, ORDER_NUMBER + ; + + @Override + public String getKey() { + return name(); + } + } +} diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessor.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessor.java new file mode 100644 index 000000000..d87252109 --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessor.java @@ -0,0 +1,49 @@ +package ru.spcex.clearing.csv.importer.component.tro.processor; + +import java.io.BufferedReader; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import ru.spcex.clearing.csv.importer.component.tro.parser.TroParser; + +public class TroProcessor { + private final TroParser parser; + + public TroProcessor(TroParser parser) { + this.parser = parser; + } + + public List readFile(Path path) { + List result = new ArrayList<>(); + + try (BufferedReader reader = Files.newBufferedReader(path)) { + String line; + int lineNumber = 0; + + while ((line = reader.readLine()) != null) { + lineNumber++; + + // быстрый skip пустых строк + if (line.isBlank()) { + continue; + } + + try { + T parsed = parser.parse(line); + result.add(parsed); + } catch (Exception e) { + throw new RuntimeException( + "Failed to parse TRO line " + lineNumber + ": " + line, + e + ); + } + } + } catch (IOException e) { + throw new RuntimeException("Failed to read file: " + path, e); + } + + return result; + } +} \ No newline at end of file diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessorProvider.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessorProvider.java new file mode 100644 index 000000000..b939b7c7a --- /dev/null +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessorProvider.java @@ -0,0 +1,35 @@ +package ru.spcex.clearing.csv.importer.component.tro.processor; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import ru.spcex.clearing.csv.importer.component.tro.parser.TroParser; +import ru.spcex.clearing.csv.importer.component.tro.parser.TroParserType; + +@Configuration +public class TroProcessorProvider { + private final Map> m; + + @Autowired + public TroProcessorProvider(List> parsers) { + this.m = new HashMap<>(); + parsers.forEach(p -> { + TroParser old = m.put(p.getType(), p); + if (old != null) { + throw new IllegalArgumentException("Duplicate parser type: " + p.getType()); + } + }); + } + + //@SuppressWarnings("unchecked") + public TroProcessor createProcessor(TroParserType type, Class clazz) { + TroParser parser = m.get(type); + if (parser == null) { + throw new IllegalArgumentException("Unknown parser type: " + type); + } + return new TroProcessor<>((TroParser) parser); + + } +} diff --git a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/config/settings/CsvImporterSettings.java b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/config/settings/CsvImporterSettings.java index d36f1ce8a..32fd85b9f 100644 --- a/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/config/settings/CsvImporterSettings.java +++ b/clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/config/settings/CsvImporterSettings.java @@ -14,6 +14,33 @@ public class CsvImporterSettings { private HazelcastClientParams hazelcast; private KafkaConsumerSettings kafkaConsumer; private KafkaProducerSettings kafkaProducer; + private String srcDir; + private String outDir; + private String errDir; + + public String getSrcDir() { + return srcDir; + } + + public void setSrcDir(String srcDir) { + this.srcDir = srcDir; + } + + public String getOutDir() { + return outDir; + } + + public void setOutDir(String outDir) { + this.outDir = outDir; + } + + public String getErrDir() { + return errDir; + } + + public void setErrDir(String errDir) { + this.errDir = errDir; + } public HazelcastClientParams getHazelcast() { return hazelcast; diff --git a/clearing-parent/csv-importer/src/main/resources/application.properties b/clearing-parent/csv-importer/src/main/resources/application.properties index 0e658e601..e96651314 100644 --- a/clearing-parent/csv-importer/src/main/resources/application.properties +++ b/clearing-parent/csv-importer/src/main/resources/application.properties @@ -16,4 +16,5 @@ csv-importer.kafka-producer.acks=all csv-importer.kafka-producer.retries=0 csv-importer.kafka-producer.batch-size=16384 csv-importer.kafka-producer.linger-ms=1 -csv-importer.kafka-producer.buffer-memory=33554432 \ No newline at end of file +csv-importer.kafka-producer.buffer-memory=33554432 +csv-importer.poll-period=5000 \ No newline at end of file