BLD-ERPAuto/BIPMaterialManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(import): bypass intermediate Excel file in extraction pipeline

Browse Source 
Replace the Extract → Write Excel → Read Excel → Import DB flow with
direct record-to-database persistence. The extractor now builds
MaterialPlanRecord[] from parsed orders and imports them without the
round-trip through a merged Excel file.

Key changes:
- Add importFromRecords() to DataImportService for record-based import
- Add SQL Server OPENJSON batch insert and atomic replace operations
  in DiscreteMaterialPlanDAO for efficient bulk writes
- Extract common import logic into private importRecords() method
- Configure explicit request/connection timeouts for SQL Server
- Add unit tests for direct record import path

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-04-28 17:11:30 +08:00
parent f36c88aa89
commit 21089e8b40
6 changed files with 748 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
src/main/services/database/data-importer.ts
View File

@@ -96,45 +96,12 @@ export class DataImportService {
// Step 1: Read Excel file
log.info('Reading Excel file...')
const { records, sourceNumbers } = await this.readExcelFile(filePath)
result.recordsRead = records.length
result.uniqueSourceNumbers = sourceNumbers.size
log.info('Excel read completed', {
recordsRead: result.recordsRead,
uniqueSourceNumbers: result.uniqueSourceNumbers
recordsRead: records.length,
uniqueSourceNumbers: sourceNumbers.size
})
if (records.length === 0) {
result.success = true
result.errors.push('Excel file contains no data records')
return result
}
// Step 2: Delete existing records by SourceNumber
log.info('Deleting existing records...', {
sourceNumberCount: sourceNumbers.size
})
const sourceNumberArray = Array.from(sourceNumbers)
result.recordsDeleted = await this.dao.deleteBySourceNumbers(sourceNumberArray)
log.info('Existing records deleted', {
recordsDeleted: result.recordsDeleted
})
// Step 3: Batch insert new records
log.info('Inserting new records...', {
recordCount: records.length,
batchSize
})
result.recordsImported = await this.dao.batchInsert(records, batchSize)
log.info('Records imported successfully', {
recordsImported: result.recordsImported
})
result.success = true
await this.importRecords(records, batchSize, result)
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
result.errors.push(`Import failed: ${errorMsg}`)
@@ -167,6 +134,92 @@ export class DataImportService {
return result
}
/**
* Import already parsed records to database.
*
* This is the preferred path for extraction: the downloader/parser already has
* structured rows, so database persistence should not require writing and
* reading an intermediate Excel file.
*/
async importFromRecords(records: MaterialPlanRecord[], batchSize = 1000): Promise<ImportResult> {
const result: ImportResult = {
success: false,
recordsRead: 0,
recordsDeleted: 0,
recordsImported: 0,
uniqueSourceNumbers: 0,
errors: []
}
try {
log.info('Starting import from parsed records', {
recordCount: records.length,
batchSize
})
return await this.importRecords(records, batchSize, result)
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
result.errors.push(`Import failed: ${errorMsg}`)
log.error('Import from records failed', { error: errorMsg })
return result
} finally {
try {
await this.dao.disconnect()
} catch (e) {
log.warn('Error disconnecting DAO', {
error: e instanceof Error ? e.message : String(e)
})
}
logAuditWithCurrentUser(
AuditAction.DATA_IMPORT,
'MATERIAL_PLAN',
result.success ? AuditStatus.SUCCESS : AuditStatus.FAILURE,
{
recordsRead: result.recordsRead,
recordsDeleted: result.recordsDeleted,
recordsImported: result.recordsImported,
uniqueSourceNumbers: result.uniqueSourceNumbers,
errorCount: result.errors.length
}
)
}
}
private async importRecords(
records: MaterialPlanRecord[],
batchSize: number,
result: ImportResult
): Promise<ImportResult> {
const sourceNumbers = new Set(records.map((record) => record.sourceNumber).filter(Boolean))
result.recordsRead = records.length
result.uniqueSourceNumbers = sourceNumbers.size
if (records.length === 0) {
result.success = true
result.errors.push('No data records to import')
return result
}
// Step 1: Replace existing records by SourceNumber
log.info('Replacing existing records...', {
sourceNumberCount: sourceNumbers.size
})
const replaceResult = await this.dao.replaceBySourceNumbers(records, batchSize)
result.recordsDeleted = replaceResult.deleted
result.recordsImported = replaceResult.inserted
log.info('Records replaced successfully', {
recordsDeleted: result.recordsDeleted,
recordsImported: result.recordsImported
})
result.success = true
return result
}
/**
* Read Excel file and extract records
* @param filePath - Path to the Excel file

366
src/main/services/database/discrete-material-plan-dao.ts
View File

@@ -13,6 +13,7 @@ import { createDialect, type SqlDialect } from './dialects'
import { createLogger, getRequestId, trackDuration } from '../logger'
const log = createLogger('DiscreteMaterialPlanDAO')
const SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE = 25
/**
* Material plan record interface
@@ -562,9 +563,16 @@ export class DiscreteMaterialPlanDAO {
const tableName = this.getTableName()
const dialect = this.getDialect()
// SQL Server has a limit of 2100 parameters per query
// Each record has 28 columns, so max rows per batch = 2100 / 28 = 75
// Leave some margin for query overhead
if (dbService.type === 'sqlserver') {
return await this.batchInsertSqlServerJson(
dbService,
tableName,
records,
batchSize,
batchId
)
}
const columnsPerRow = 28
const effectiveBatchSize = Math.min(batchSize, dialect.maxBatchRows(columnsPerRow))
const totalBatches = Math.ceil(records.length / effectiveBatchSize)
@@ -626,6 +634,237 @@ export class DiscreteMaterialPlanDAO {
}
}
async replaceBySourceNumbers(
records: MaterialPlanRecord[],
batchSize = 1000
): Promise<{ deleted: number; inserted: number }> {
if (!records || records.length === 0) {
return { deleted: 0, inserted: 0 }
}
const dbService = await this.getDatabaseService()
const sourceNumbers = [...new Set(records.map((record) => record.sourceNumber).filter(Boolean))]
if (dbService.type === 'sqlserver') {
return await this.replaceSqlServerJson(dbService, records, sourceNumbers)
}
const deleted = await this.deleteBySourceNumbers(sourceNumbers)
const inserted = await this.batchInsert(records, batchSize)
return { deleted, inserted }
}
private async replaceSqlServerJson(
dbService: IDatabaseService,
records: MaterialPlanRecord[],
sourceNumbers: string[]
): Promise<{ deleted: number; inserted: number }> {
const tableName = this.getTableName()
const columns = this.getInsertColumns()
const withColumns = this.getSqlServerJsonWithColumns(columns)
const quotedColumns = columns.map((column) => `[${column}]`).join(', ')
const recordsBySourceNumber = this.groupRecordsBySourceNumber(records)
const totalBatches = Math.ceil(
sourceNumbers.length / SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE
)
let totalDeleted = 0
let totalInserted = 0
log.info('SQL Server JSON replace started', {
tableName,
operationType: 'REPLACE',
totalSourceNumbers: sourceNumbers.length,
totalRecords: records.length,
sourceNumberBatchSize: SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE,
totalBatches
})
for (
let offset = 0;
offset < sourceNumbers.length;
offset += SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE
) {
const sourceNumberBatch = sourceNumbers.slice(
offset,
offset + SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE
)
const batchNumber = Math.floor(offset / SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE) + 1
const recordBatch = sourceNumberBatch.flatMap(
(sourceNumber) => recordsBySourceNumber.get(sourceNumber) || []
)
const jsonRows = recordBatch.map((record) => this.buildJsonRow(record, columns))
const sqlString = `
DECLARE @deleted int = 0;
DECLARE @inserted int = 0;
BEGIN TRY
BEGIN TRANSACTION;
DELETE target
FROM ${tableName} AS target
INNER JOIN OPENJSON(@p0)
WITH (SourceNumber nvarchar(100) '$') AS source
ON target.SourceNumber = source.SourceNumber;
SET @deleted = @@ROWCOUNT;
INSERT INTO ${tableName} (${quotedColumns})
SELECT ${quotedColumns}
FROM OPENJSON(@p1)
WITH (
${withColumns}
);
SET @inserted = @@ROWCOUNT;
COMMIT TRANSACTION;
END TRY
BEGIN CATCH
IF @@TRANCOUNT > 0
ROLLBACK TRANSACTION;
THROW;
END CATCH;
SELECT @deleted AS deletedCount, @inserted AS insertedCount;
`
const result = await trackDuration(
async () =>
await dbService.query(sqlString, [
JSON.stringify(sourceNumberBatch),
JSON.stringify(jsonRows)
]),
{
operationName: 'DiscreteMaterialPlanDAO.replaceSqlServerJsonBatch',
context: {
tableName,
operationType: 'REPLACE',
batchNumber,
totalBatches,
sourceNumberCount: sourceNumberBatch.length,
recordCount: recordBatch.length
}
}
)
const stats = result.result.rows[0] || {}
totalDeleted += Number(stats.deletedCount || 0)
totalInserted += Number(stats.insertedCount || recordBatch.length)
log.debug('SQL Server JSON replace batch completed', {
tableName,
batchNumber,
totalBatches,
sourceNumberCount: sourceNumberBatch.length,
recordCount: recordBatch.length
})
}
log.info('SQL Server JSON replace completed', {
tableName,
operationType: 'REPLACE',
totalDeleted,
totalInserted,
totalBatches
})
return {
deleted: totalDeleted,
inserted: totalInserted
}
}
private groupRecordsBySourceNumber(
records: MaterialPlanRecord[]
): Map<string, MaterialPlanRecord[]> {
const groups = new Map<string, MaterialPlanRecord[]>()
for (const record of records) {
if (!record.sourceNumber) {
continue
}
const existing = groups.get(record.sourceNumber) || []
existing.push(record)
groups.set(record.sourceNumber, existing)
}
return groups
}
private async batchInsertSqlServerJson(
dbService: IDatabaseService,
tableName: string,
records: MaterialPlanRecord[],
batchSize: number,
batchId: string
): Promise<number> {
const columns = this.getInsertColumns()
const effectiveBatchSize = Math.max(1, batchSize)
const totalBatches = Math.ceil(records.length / effectiveBatchSize)
let totalInserted = 0
log.info('SQL Server JSON batch insert started', {
tableName,
operationType: 'INSERT',
requestId: batchId,
totalRecords: records.length,
effectiveBatchSize,
totalBatches
})
for (let i = 0; i < records.length; i += effectiveBatchSize) {
const batch = records.slice(i, i + effectiveBatchSize)
const batchNumber = Math.floor(i / effectiveBatchSize) + 1
const jsonRows = batch.map((record) => this.buildJsonRow(record, columns))
const withColumns = this.getSqlServerJsonWithColumns(columns)
const quotedColumns = columns.map((column) => `[${column}]`).join(', ')
const sqlString = `
INSERT INTO ${tableName} (${quotedColumns})
SELECT ${quotedColumns}
FROM OPENJSON(@p0)
WITH (
${withColumns}
)
`
const result = await trackDuration(
async () => await dbService.query(sqlString, [JSON.stringify(jsonRows)]),
{
operationName: 'DiscreteMaterialPlanDAO.insertBatchSqlServerJson',
context: {
tableName,
operationType: 'INSERT',
batchId,
batchNumber,
totalBatches,
recordCount: batch.length
}
}
)
totalInserted += result.result.rowCount || batch.length
log.debug('Inserted SQL Server JSON batch', {
batch: batchNumber,
totalBatches,
count: batch.length,
batchId
})
}
log.info('SQL Server JSON batch insert completed', {
tableName,
operationType: 'INSERT',
requestId: batchId,
totalInserted,
batchSize: effectiveBatchSize,
totalBatches
})
return totalInserted
}
/**
* Insert a single batch of records with tracking
*/
@@ -641,37 +880,7 @@ export class DiscreteMaterialPlanDAO {
return 0
}
// Build column list (excluding id)
const columns = [
'Factory',
'MaterialStatus',
'PlanNumber',
'SourceNumber',
'MaterialType',
'ProductCode',
'ProductName',
'ProductUnit',
'ProductPlanQuantity',
'UseDepartment',
'Remark',
'Creator',
'CreateDate',
'Approver',
'ApproveDate',
'SequenceNumber',
'MaterialCode',
'MaterialName',
'Specification',
'Model',
'DrawingNumber',
'MaterialQuality',
'PlanQuantity',
'Unit',
'RequiredDate',
'Warehouse',
'UnitUsage',
'CumulativeOutputQuantity'
]
const columns = this.getInsertColumns()
// Build parameterized insert
const values: any[] = []
@@ -730,6 +939,91 @@ export class DiscreteMaterialPlanDAO {
})
}
private getInsertColumns(): string[] {
return [
'Factory',
'MaterialStatus',
'PlanNumber',
'SourceNumber',
'MaterialType',
'ProductCode',
'ProductName',
'ProductUnit',
'ProductPlanQuantity',
'UseDepartment',
'Remark',
'Creator',
'CreateDate',
'Approver',
'ApproveDate',
'SequenceNumber',
'MaterialCode',
'MaterialName',
'Specification',
'Model',
'DrawingNumber',
'MaterialQuality',
'PlanQuantity',
'Unit',
'RequiredDate',
'Warehouse',
'UnitUsage',
'CumulativeOutputQuantity'
]
}
private buildJsonRow(record: MaterialPlanRecord, columns: string[]): Record<string, unknown> {
const row: Record<string, unknown> = {}
for (const column of columns) {
const value = this.getColumnValue(record, column)
row[column] = value instanceof Date ? value.toISOString() : value
}
return row
}
private getSqlServerJsonWithColumns(columns: string[]): string {
return columns
.map((column) => `[${column}] ${this.getSqlServerJsonColumnType(column)} '$.${column}'`)
.join(',\n ')
}
private getSqlServerJsonColumnType(column: string): string {
const columnTypes: Record<string, string> = {
Factory: 'nvarchar(100)',
MaterialStatus: 'nvarchar(50)',
PlanNumber: 'nvarchar(100)',
SourceNumber: 'nvarchar(100)',
MaterialType: 'nvarchar(100)',
ProductCode: 'nvarchar(100)',
ProductName: 'nvarchar(255)',
ProductUnit: 'nvarchar(50)',
ProductPlanQuantity: 'decimal(18,4)',
UseDepartment: 'nvarchar(100)',
Remark: 'nvarchar(500)',
Creator: 'nvarchar(100)',
CreateDate: 'datetime2',
Approver: 'nvarchar(100)',
ApproveDate: 'datetime2',
SequenceNumber: 'int',
MaterialCode: 'nvarchar(100)',
MaterialName: 'nvarchar(255)',
Specification: 'nvarchar(255)',
Model: 'nvarchar(255)',
DrawingNumber: 'nvarchar(100)',
MaterialQuality: 'nvarchar(100)',
PlanQuantity: 'decimal(18,4)',
Unit: 'nvarchar(50)',
RequiredDate: 'datetime2',
Warehouse: 'nvarchar(100)',
UnitUsage: 'decimal(18,6)',
CumulativeOutputQuantity: 'decimal(18,4)'
}
return columnTypes[column] || 'nvarchar(max)'
}
/**
* Get the value for a specific column from the record
*/
@@ -777,6 +1071,10 @@ export class DiscreteMaterialPlanDAO {
return null
}
if (value instanceof Date && Number.isNaN(value.getTime())) {
return null
}
// Handle empty strings for string fields
if (typeof value === 'string' && value.trim() === '') {
return null

2
src/main/services/database/sql-server.ts
View File

@@ -38,6 +38,8 @@ export class SqlServerService implements IDatabaseService {
user: this.config.user,
password: this.config.password,
database: this.config.database,
requestTimeout: 60000,
connectionTimeout: 15000,
options: {
encrypt: this.config.options?.encrypt ?? false,
trustServerCertificate: this.config.options?.trustServerCertificate ?? false

160
src/main/services/erp/extractor.ts
View File

@@ -10,6 +10,7 @@ import type {
LogLevel
} from '../../types/extractor.types'
import { DataImportService } from '../database/data-importer'
import type { MaterialPlanRecord } from '../database/discrete-material-plan-dao'
import { createLogger, withRequestContext, getRequestId } from '../logger'
import { trackDuration } from '../logger/performance-monitor'
@@ -112,16 +113,18 @@ export class ExtractorService {
// Always clean up temporary files regardless of merge success
await this.cleanupTempFiles(result.downloadedFiles, input.orderNumbers)
// Auto-import to database if merge was successful
if (result.mergedFile) {
// Auto-import parsed records directly. The merged Excel file is an archive artifact,
// not the source for persistence.
if (mergeResult.records.length > 0) {
const importProgress = (1 + totalBatches + 1) * progressPerPoint
input.onProgress?.('正在写入数据库...', importProgress, {
phase: 'importing',
totalBatches
})
const importResult = await this.importToDatabaseWithLogging(
result.mergedFile,
input.onLog
const importResult = await this.importRecordsToDatabaseWithLogging(
mergeResult.records,
input.onLog,
result.mergedFile
)
result.importResult = importResult
@@ -168,9 +171,10 @@ export class ExtractorService {
recordCount: number
error?: string
orderRecordCounts: Array<{ orderNumber: string; recordCount: number }>
records: MaterialPlanRecord[]
}> {
if (filePaths.length === 0) {
return { mergedFile: null, recordCount: 0, orderRecordCounts: [] }
return { mergedFile: null, recordCount: 0, orderRecordCounts: [], records: [] }
}
log.info('Starting merge', { fileCount: filePaths.length, orderCount: orderNumbers.length })
@@ -219,10 +223,11 @@ export class ExtractorService {
}
log.info('Merge summary', { orderCount: allOrders.length, recordCount })
const records = this.buildMaterialPlanRecords(allOrders)
if (recordCount === 0) {
log.warn('No records found in any downloaded files', { orderNumbers })
return { mergedFile: null, recordCount: 0, orderRecordCounts }
return { mergedFile: null, recordCount: 0, orderRecordCounts, records }
}
// Generate output filename with timestamp
@@ -238,7 +243,7 @@ export class ExtractorService {
log.info('Saving merged file', { outputPath })
await this.saveMergedOrders(allOrders, outputPath)
log.info('Merged file saved successfully', { recordCount })
return { mergedFile: outputPath, recordCount, orderRecordCounts }
return { mergedFile: outputPath, recordCount, orderRecordCounts, records }
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
const errorStack = error instanceof Error ? error.stack : ''
@@ -253,6 +258,7 @@ export class ExtractorService {
mergedFile: null,
recordCount,
orderRecordCounts,
records,
error: `保存合并文件失败：${errorMsg}`
}
}
@@ -370,6 +376,78 @@ export class ExtractorService {
log.debug('File saved successfully', { outputPath })
}
private buildMaterialPlanRecords(
orders: Array<{ orderInfo: any; materials: any[] }>
): MaterialPlanRecord[] {
const records: MaterialPlanRecord[] = []
for (const order of orders) {
const { orderInfo, materials } = order
for (const material of materials) {
records.push({
factory: this.toText(orderInfo.factory),
materialStatus: this.toText(orderInfo.materialStatus),
planNumber: this.toText(orderInfo.planNumber),
sourceNumber: this.toText(orderInfo.productionOrder),
materialType: this.toText(orderInfo.materialType),
productCode: this.toText(orderInfo.productCode),
productName: this.toText(orderInfo.productName),
productPlanQuantity: this.toNumber(orderInfo.plannedQuantity),
productUnit: this.toText(orderInfo.unit),
useDepartment: this.toText(orderInfo.department),
remark: this.toText(orderInfo.remark),
creator: this.toText(orderInfo.creator),
createDate: this.toDate(orderInfo.createDate),
approver: this.toText(orderInfo.approver),
approveDate: this.toDate(orderInfo.approveDate),
sequenceNumber: this.toNumber(material.sequence),
materialCode: this.toText(material.materialCode),
materialName: this.toText(material.materialName),
specification: this.toText(material.specification),
model: this.toText(material.model),
drawingNumber: this.toText(material.drawingNumber),
materialQuality: this.toText(material.material),
planQuantity: this.toNumber(material.quantity),
unit: this.toText(material.unit),
requiredDate: this.toDate(material.requiredDate),
warehouse: this.toText(material.warehouse),
unitUsage: this.toNumber(material.unitUsage),
cumulativeOutputQuantity: this.toNumber(material.cumulativeOutboundQty),
bomVersion: ''
})
}
}
return records
}
private toText(value: unknown): string {
if (value === null || value === undefined) {
return ''
}
return String(value).trim()
}
private toNumber(value: unknown): number {
if (value === null || value === undefined || value === '') {
return 0
}
const parsed = Number(value)
return Number.isFinite(parsed) ? parsed : 0
}
private toDate(value: unknown): Date {
if (value instanceof Date) {
return value
}
if (value === null || value === undefined || value === '') {
return new Date(NaN)
}
const parsed = new Date(String(value))
return parsed
}
/**
* Clean up temporary batch files after merging
* @param filePaths - Array of temporary file paths to delete
@@ -458,4 +536,70 @@ export class ExtractorService {
return trackedResult.result
}
private async importRecordsToDatabaseWithLogging(
records: MaterialPlanRecord[],
onLog?: (level: LogLevel, message: string) => void,
archiveFilePath?: string | null
): Promise<ImportResult> {
log.info('Starting database import from parsed records', {
recordCount: records.length,
archiveFilePath
})
onLog?.('info', `开始导入数据到数据库...`)
const trackedResult = await trackDuration(
async () => {
const importService = new DataImportService()
try {
const result = await importService.importFromRecords(records, 1000)
log.info('Import completed', {
success: result.success,
recordsRead: result.recordsRead,
recordsDeleted: result.recordsDeleted,
recordsImported: result.recordsImported
})
if (result.success) {
onLog?.(
'success',
`导入完成：读取 ${result.recordsRead} 条，删除 ${result.recordsDeleted} 条，导入 ${result.recordsImported} 条`
)
} else if (result.errors.length > 0) {
result.errors.forEach((err) => onLog?.('error', err))
}
return result
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
log.error('Import failed', {
error: errorMsg,
archiveFilePath,
downloadDir: this.downloadDir
})
onLog?.('error', `导入失败：${errorMsg}`)
return {
success: false,
recordsRead: 0,
recordsDeleted: 0,
recordsImported: 0,
uniqueSourceNumbers: 0,
errors: [errorMsg]
}
}
},
{
operationName: 'Database Import',
context: {
recordCount: records.length,
archiveFilePath
}
}
)
return trackedResult.result
}
}