dbf importer add logs
This commit is contained in:
akulikov
parent
0d607ebf52
commit
a07a31914b
4 changed files with 69 additions and 42 deletions
|
|
@ -3,6 +3,7 @@ package ru.spcex.clearing.dbf.importer.logic.stages;
|
|||
import com.linuxense.javadbf.DBFField;
|
||||
import com.linuxense.javadbf.DBFReader;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
|
@ -74,34 +75,42 @@ public class ImportToDB extends Stage {
|
|||
if (batchIdx == batchCount && batchRem == 0) continue;
|
||||
final Integer currentBatchIdx = batchIdx;
|
||||
|
||||
int[] rowsInserted = dbfJdbcTemplate.batchUpdate(
|
||||
"insert into " + currTable.name() +
|
||||
" (" + String.join(",", columnNames) + ")" +
|
||||
" values (" + String.join(",", Collections.nCopies(columnCount, "?")) + ")",
|
||||
new BatchPreparedStatementSetter() {
|
||||
@Override
|
||||
public void setValues(PreparedStatement ps, int i) throws SQLException {
|
||||
Object[] currValuesRow = values.get(currentBatchIdx * maxBatchSize + i);
|
||||
int parameterIndex = 1;
|
||||
for (Object object : currValuesRow) {
|
||||
ps.setObject(parameterIndex, object, columnTypes.get(parameterIndex - 1));
|
||||
parameterIndex++;
|
||||
try {
|
||||
int[] rowsInserted = dbfJdbcTemplate.batchUpdate(
|
||||
"insert into " + currTable.name() +
|
||||
" (" + String.join(",", columnNames) + ")" +
|
||||
" values (" + String.join(",", Collections.nCopies(columnCount, "?")) + ")",
|
||||
new BatchPreparedStatementSetter() {
|
||||
@Override
|
||||
public void setValues(PreparedStatement ps, int i) throws SQLException {
|
||||
Object[] currValuesRow = values.get(currentBatchIdx * maxBatchSize + i);
|
||||
int parameterIndex = 1;
|
||||
for (Object object : currValuesRow) {
|
||||
ps.setObject(parameterIndex, object, columnTypes.get(parameterIndex - 1));
|
||||
parameterIndex++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getBatchSize() {
|
||||
return currentBatchIdx == batchCount ? batchRem : maxBatchSize;
|
||||
}
|
||||
});
|
||||
log.info("Imported records: " + rowsInserted.length);
|
||||
@Override
|
||||
public int getBatchSize() {
|
||||
return currentBatchIdx == batchCount ? batchRem : maxBatchSize;
|
||||
}
|
||||
});
|
||||
log.info("uuid {}. Imported records: {}", resultContainer.getUuid(), rowsInserted.length);
|
||||
} catch (DataAccessException e) {
|
||||
log.error(String.format("uuid %s. Can't insert records", resultContainer.getUuid()), e);
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
}
|
||||
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
log.error(String.format("uuid %s. Can't read source file %s.",
|
||||
resultContainer.getUuid(),
|
||||
resultContainer.getDbfFile().getName()), e);
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
|
||||
|
||||
return StageResult.OK;
|
||||
return StageResult.COMPLETE;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import java.io.IOException;
|
|||
import java.nio.file.Files;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.Objects;
|
||||
import java.util.UUID;
|
||||
|
||||
@Component
|
||||
public class LoadFileFromDisk extends Stage {
|
||||
|
|
@ -23,25 +24,27 @@ public class LoadFileFromDisk extends Stage {
|
|||
@Override
|
||||
public StageResult process(ResultContainer resultContainer) {
|
||||
Objects.requireNonNull(resultContainer.getDbfFile());
|
||||
UUID taskUuid = resultContainer.getUuid();
|
||||
|
||||
File dbfFile = resultContainer.getDbfFile();
|
||||
byte[] fileBytes;
|
||||
try {
|
||||
log.debug("uuid {}. Read all bytes from source file {}", taskUuid, resultContainer.getDbfFile().getName());
|
||||
fileBytes = Files.readAllBytes(Paths.get(dbfFile.getAbsolutePath()));
|
||||
if (fileBytes.length == 0) throw new IOException("Empty file");
|
||||
if (properties.deleteSrcFiles()) {
|
||||
boolean deleteOk = dbfFile.delete();
|
||||
if (!deleteOk) {
|
||||
log.warn("Can't remove source file {}.", dbfFile.getPath());
|
||||
log.warn("uuid {}. Can't remove source file {}.", taskUuid, dbfFile.getPath());
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
}
|
||||
if (fileBytes.length == 0) throw new IOException("Empty file");
|
||||
} catch (IOException e) {
|
||||
log.error("Can't read file " + dbfFile.getPath() + ". File was skipped.", e);
|
||||
log.error(String.format("uuid %s. Can't read file %s", taskUuid, resultContainer.getDbfFile().getName()), e);
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
log.debug("uuid {}. Read all bytes from source file complete", taskUuid);
|
||||
resultContainer.setDbfSource(fileBytes);
|
||||
|
||||
return StageResult.OK;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -20,11 +20,18 @@ public class Processor {
|
|||
}
|
||||
|
||||
public void process(ResultContainer task) {
|
||||
log.info("Start work with task " + task.getUuid());
|
||||
log.info("uuid {}. Task started", task.getUuid());
|
||||
StageResult result = null;
|
||||
long startMills = System.currentTimeMillis();
|
||||
for (Stage currStage : pipeline) {
|
||||
log.info("{} stage for task {}", currStage.getClass().getSimpleName(), task.getUuid());
|
||||
StageResult result = currStage.process(task);
|
||||
log.info("uuid {}. Stage: {}", task.getUuid(), currStage.getClass().getSimpleName());
|
||||
result = currStage.process(task);
|
||||
if (Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(result)) break;
|
||||
}
|
||||
long endMills = System.currentTimeMillis();
|
||||
log.info("uuid {}. Task completed, result: {}, time working: {}",
|
||||
task.getUuid(),
|
||||
result,
|
||||
endMills - startMills);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ import java.sql.ResultSet;
|
|||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* Фильтрация источника на предмет соответствия полей
|
||||
|
|
@ -48,27 +49,29 @@ public class ValidateFields extends Stage implements InitializingBean {
|
|||
public StageResult process(ResultContainer resultContainer) {
|
||||
Objects.requireNonNull(resultContainer.getDbfTable());
|
||||
Objects.requireNonNull(resultContainer.getDbfSource());
|
||||
UUID taskUuid = resultContainer.getUuid();
|
||||
Table currTable = resultContainer.getDbfTable();
|
||||
byte[] source = resultContainer.getDbfSource();
|
||||
|
||||
TableStructureDBF currTableStructure = dbStructureDBF.getTables().get(currTable);
|
||||
if (currTableStructure == null) {
|
||||
log.warn("Unknown table {}, table was skipped.", currTable);
|
||||
return StageResult.COMPLETE;
|
||||
log.error("uuid {}. Unknown table {}.", taskUuid, currTable);
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
|
||||
log.info("Check source " + currTable);
|
||||
log.info("uuid {}. Check source {}", taskUuid, currTable);
|
||||
|
||||
try (InputStream is = new ByteArrayInputStream(source);
|
||||
DBFReader dbfReader = new DBFReader(is, dbfCharset)) {
|
||||
|
||||
int columnCount = dbfReader.getFieldCount();
|
||||
if (columnCount != currTableStructure.getColumns().size()) {
|
||||
log.warn("Source for table {} doesn't match DB columns count. Source column count: {}. DB column count: {}. Source was skipped.",
|
||||
log.error("uuid {}. Source for table {} doesn't match DB columns count. Source column count: {}. DB column count: {}.",
|
||||
taskUuid,
|
||||
currTable,
|
||||
columnCount,
|
||||
currTableStructure.getColumns().size());
|
||||
return StageResult.COMPLETE;
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
|
||||
boolean valid = true;
|
||||
|
|
@ -78,14 +81,18 @@ public class ValidateFields extends Stage implements InitializingBean {
|
|||
String currFieldName = currField.getName();
|
||||
ColumnStructureDBF currColumnStructure = currTableStructure.getIgnoreCase(currFieldName);
|
||||
if (currColumnStructure == null) {
|
||||
log.warn("Source for table {} contain unknown field {}. ", currTable, currFieldName);
|
||||
log.error("uuid {}. Source for table {} contain unknown field {}. ",
|
||||
taskUuid,
|
||||
currTable,
|
||||
currFieldName);
|
||||
valid = false;
|
||||
continue;
|
||||
}
|
||||
|
||||
DBFDataType dbfDataType = currField.getType();
|
||||
if (!currColumnStructure.getType().getDbfType().equals(dbfDataType)) {
|
||||
log.warn("Field {} from source {} has wrong data type (actual {}, expected {}).",
|
||||
log.error("uuid {}. Field {} from source {} has wrong data type (actual {}, expected {}).",
|
||||
taskUuid,
|
||||
currFieldName,
|
||||
currTable,
|
||||
dbfDataType.name(),
|
||||
|
|
@ -98,7 +105,8 @@ public class ValidateFields extends Stage implements InitializingBean {
|
|||
if (fieldLength != currColumnStructure.getLength()) {
|
||||
|
||||
if (!dbfDataType.equals(DBFDataType.DATE)) {
|
||||
log.warn("Field {} from source {} has wrong length (actual {}, expected {}).",
|
||||
log.error("uuid {}. Field {} from source {} has wrong length (actual {}, expected {}).",
|
||||
taskUuid,
|
||||
currFieldName,
|
||||
currTable,
|
||||
fieldLength,
|
||||
|
|
@ -107,14 +115,15 @@ public class ValidateFields extends Stage implements InitializingBean {
|
|||
continue;
|
||||
} else {
|
||||
if (fieldLength > currColumnStructure.getLength()) {
|
||||
log.warn("...");
|
||||
log.error("uuid {}. ...", taskUuid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int decimalCount = currField.getDecimalCount();
|
||||
if (decimalCount != currColumnStructure.getDecimalDigits()) {
|
||||
log.warn("Field {} from source {} has wrong decimal count (actual {}, expected {}).",
|
||||
log.error("uuid {}. Field {} from source {} has wrong decimal count (actual {}, expected {}).",
|
||||
taskUuid,
|
||||
currFieldName,
|
||||
currTable,
|
||||
decimalCount,
|
||||
|
|
@ -123,12 +132,11 @@ public class ValidateFields extends Stage implements InitializingBean {
|
|||
}
|
||||
|
||||
if (!valid) {
|
||||
log.warn("Source was skipped.");
|
||||
return StageResult.COMPLETE;
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
}
|
||||
} catch (IOException e) {
|
||||
log.error("Can't read source for table {}. Source was skipped.", currTable);
|
||||
log.error(String.format("uuid %s. Can't read source for table %s.", taskUuid, currTable), e);
|
||||
return StageResult.ERROR;
|
||||
}
|
||||
|
||||
|
|
@ -181,7 +189,7 @@ public class ValidateFields extends Stage implements InitializingBean {
|
|||
try {
|
||||
dbfCharset = Charset.forName(properties.getDbfEncoding());
|
||||
} catch (Exception e) {
|
||||
throw new ConfigException("В properties файле содержится неизвестная кодировка: " + properties.getDbfEncoding(), e);
|
||||
throw new ConfigException("Unknown encoding from properties: " + properties.getDbfEncoding(), e);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue