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

http://jira.mfd.msk:8088/browse/CLS-961

Browse source
This commit is contained in:
ialbert 2026-04-20 15:34:41 +03:00
parent b328c137f6
commit 9dc9e06a78
11 changed files with 477 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/FileService.java Normal file
View file

@ -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");
}
}
}

77
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/SOrdersScheduler.java Normal file
View file

@ -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<SOrder> processor;
private final Imdg<SOrders> 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<SOrder> 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);
}
}
}

49
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/data/SOrder.java Normal file
View file

@ -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;
}
}

121
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/AbstractTroParser.java Normal file
View file

@ -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<T> implements TroParser<T> {
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<String, String> parseKeyValueLine(String line) {
Map<String, String> 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<String, String> 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<String, String> values, IEnumKey key) {
String value = values.get(key.getKey());
return value == null ? null : value.trim();
}
protected Long extractLong(Map<String, String> 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<String, String> 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<String, String> 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<String, String> values, IEnumKey key) {
return getOptional(values, key);
}
protected LocalDateTime extractDateTime(Map<String, String> 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
);
}
}
}

6
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParser.java Normal file
View file

@ -0,0 +1,6 @@
package ru.spcex.clearing.csv.importer.component.tro.parser;
public interface TroParser<T> {
T parse(String line);
TroParserType getType();
}

7
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/TroParserType.java Normal file
View file

@ -0,0 +1,7 @@
package ru.spcex.clearing.csv.importer.component.tro.parser;
public enum TroParserType {
SOrder
;
}

39
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/parser/impl/SOrderParser.java Normal file
View file

@ -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<SOrder> {
@Override
public TroParserType getType() {
return TroParserType.SOrder;
}
@Override
public SOrder parse(String line) {
Map<String, String> 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();
}
}
}

49
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessor.java Normal file
View file

@ -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<T> {
private final TroParser<T> parser;
public TroProcessor(TroParser<T> parser) {
this.parser = parser;
}
public List<T> readFile(Path path) {
List<T> 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;
}
}

35
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/component/tro/processor/TroProcessorProvider.java Normal file
View file

@ -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<TroParserType, TroParser<?>> m;
@Autowired
public TroProcessorProvider(List<TroParser<?>> 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 <T> TroProcessor<T> createProcessor(TroParserType type, Class<T> clazz) {
TroParser<?> parser = m.get(type);
if (parser == null) {
throw new IllegalArgumentException("Unknown parser type: " + type);
}
return new TroProcessor<>((TroParser<T>) parser);
}
}

27
clearing-parent/csv-importer/src/main/java/ru/spcex/clearing/csv/importer/config/settings/CsvImporterSettings.java
View file

@ -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;

3
clearing-parent/csv-importer/src/main/resources/application.properties
View file

@ -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
csv-importer.kafka-producer.buffer-memory=33554432
csv-importer.poll-period=5000