5 Commits
v1.14.1 ... dev

Author SHA1 Message Date
Misaka
cf9976f605 fix: add AT and ZONE to PostgreSQL SQL keywords for prepareSql
The prepareSql function quotes any word not in SQL_KEYWORDS as an
identifier. Since AT and ZONE were missing from the set, the expression
(NOW() AT TIME ZONE 'UTC') was mangled into (NOW() "AT" TIME "ZONE"
'UTC'), causing INSERT failures on PostgreSQL.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-28 22:00:06 +08:00
Misaka
a6c2e2ccc2 fix: use explicit UTC timestamp in PostgreSQL dialect
PostgreSQL's CURRENT_TIMESTAMP returns session-local time, unlike
SYSUTCDATETIME() (SQL Server) and UTC_TIMESTAMP() (MySQL) which
explicitly return UTC. Switch to (NOW() AT TIME ZONE 'UTC') to
keep operation history timestamps consistent across all databases.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-28 21:48:28 +08:00
Misaka_Company
21089e8b40 perf(import): bypass intermediate Excel file in extraction pipeline
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>
2026-04-28 17:11:30 +08:00
Misaka_Company
f36c88aa89 fix(extractor): display operation time in local timezone
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-28 13:53:54 +08:00
Misaka_Company
dbb8e4904e perf: optimize order resolution history writes 2026-04-28 13:39:08 +08:00
14 changed files with 868 additions and 138 deletions

View File

@@ -126,7 +126,9 @@ export function registerExtractorHandlers(): void {
sendLog(sender, 'info', '正在解析订单号...') sendLog(sender, 'info', '正在解析订单号...')
const resolver = new OrderNumberResolver(dbService) const resolver = new OrderNumberResolver(dbService)
const resolutionStart = Date.now()
const mappings = await resolver.resolve(input.orderNumbers) const mappings = await resolver.resolve(input.orderNumbers)
const resolutionDurationMs = Date.now() - resolutionStart
// Get valid order numbers and warnings // Get valid order numbers and warnings
const validOrderNumbers = resolver.getValidOrderNumbers(mappings) const validOrderNumbers = resolver.getValidOrderNumbers(mappings)
@@ -146,7 +148,16 @@ export function registerExtractorHandlers(): void {
) )
} }
log.info('Resolved order numbers', { count: validOrderNumbers.length }) log.info('Resolved order numbers', {
inputCount: input.orderNumbers.length,
count: validOrderNumbers.length,
durationMs: resolutionDurationMs
})
sendLog(
sender,
'info',
`订单号解析完成:${validOrderNumbers.length}/${input.orderNumbers.length} 个有效,耗时 ${(resolutionDurationMs / 1000).toFixed(2)}`
)
// Initialize operation history recording // Initialize operation history recording
const currentUser = SessionManager.getInstance().getUserInfo() const currentUser = SessionManager.getInstance().getUserInfo()
@@ -155,6 +166,7 @@ export function registerExtractorHandlers(): void {
// Save order records to history (preserve productionId -> orderNumber mapping) // Save order records to history (preserve productionId -> orderNumber mapping)
if (currentUser) { if (currentUser) {
const historyInsertStart = Date.now()
const orderRecords = mappings.map((m) => ({ const orderRecords = mappings.map((m) => ({
productionId: m.productionId || null, productionId: m.productionId || null,
orderNumber: m.orderNumber || m.input orderNumber: m.orderNumber || m.input
@@ -165,9 +177,11 @@ export function registerExtractorHandlers(): void {
currentUser.username, currentUser.username,
orderRecords orderRecords
) )
const historyInsertDurationMs = Date.now() - historyInsertStart
log.info('Operation history batch created', { log.info('Operation history batch created', {
batchId, batchId,
recordCount: orderRecords.length recordCount: orderRecords.length,
durationMs: historyInsertDurationMs
}) })
} }

View File

@@ -96,45 +96,12 @@ export class DataImportService {
// Step 1: Read Excel file // Step 1: Read Excel file
log.info('Reading Excel file...') log.info('Reading Excel file...')
const { records, sourceNumbers } = await this.readExcelFile(filePath) const { records, sourceNumbers } = await this.readExcelFile(filePath)
result.recordsRead = records.length
result.uniqueSourceNumbers = sourceNumbers.size
log.info('Excel read completed', { log.info('Excel read completed', {
recordsRead: result.recordsRead, recordsRead: records.length,
uniqueSourceNumbers: result.uniqueSourceNumbers uniqueSourceNumbers: sourceNumbers.size
}) })
if (records.length === 0) { await this.importRecords(records, batchSize, result)
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
} catch (error) { } catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error) const errorMsg = error instanceof Error ? error.message : String(error)
result.errors.push(`Import failed: ${errorMsg}`) result.errors.push(`Import failed: ${errorMsg}`)
@@ -167,6 +134,92 @@ export class DataImportService {
return result 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 * Read Excel file and extract records
* @param filePath - Path to the Excel file * @param filePath - Path to the Excel file

View File

@@ -26,7 +26,7 @@ export class PostgreSqlDialect implements SqlDialect {
} }
currentTimestamp(): string { currentTimestamp(): string {
return 'CURRENT_TIMESTAMP' return "(NOW() AT TIME ZONE 'UTC')"
} }
upsert(params: { upsert(params: {

View File

@@ -13,6 +13,7 @@ import { createDialect, type SqlDialect } from './dialects'
import { createLogger, getRequestId, trackDuration } from '../logger' import { createLogger, getRequestId, trackDuration } from '../logger'
const log = createLogger('DiscreteMaterialPlanDAO') const log = createLogger('DiscreteMaterialPlanDAO')
const SQLSERVER_REPLACE_SOURCE_NUMBER_BATCH_SIZE = 25
/** /**
* Material plan record interface * Material plan record interface
@@ -562,9 +563,16 @@ export class DiscreteMaterialPlanDAO {
const tableName = this.getTableName() const tableName = this.getTableName()
const dialect = this.getDialect() const dialect = this.getDialect()
// SQL Server has a limit of 2100 parameters per query if (dbService.type === 'sqlserver') {
// Each record has 28 columns, so max rows per batch = 2100 / 28 = 75 return await this.batchInsertSqlServerJson(
// Leave some margin for query overhead dbService,
tableName,
records,
batchSize,
batchId
)
}
const columnsPerRow = 28 const columnsPerRow = 28
const effectiveBatchSize = Math.min(batchSize, dialect.maxBatchRows(columnsPerRow)) const effectiveBatchSize = Math.min(batchSize, dialect.maxBatchRows(columnsPerRow))
const totalBatches = Math.ceil(records.length / effectiveBatchSize) 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 * Insert a single batch of records with tracking
*/ */
@@ -641,37 +880,7 @@ export class DiscreteMaterialPlanDAO {
return 0 return 0
} }
// Build column list (excluding id) const columns = this.getInsertColumns()
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'
]
// Build parameterized insert // Build parameterized insert
const values: any[] = [] 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 * Get the value for a specific column from the record
*/ */
@@ -777,6 +1071,10 @@ export class DiscreteMaterialPlanDAO {
return null return null
} }
if (value instanceof Date && Number.isNaN(value.getTime())) {
return null
}
// Handle empty strings for string fields // Handle empty strings for string fields
if (typeof value === 'string' && value.trim() === '') { if (typeof value === 'string' && value.trim() === '') {
return null return null

View File

@@ -126,35 +126,47 @@ export class ExtractorOperationHistoryDAO {
recordCount: records.length recordCount: records.length
}) })
for (const record of records) { const columnsPerRecord = 5
const batchSize = Math.max(1, dialect.maxBatchRows(columnsPerRecord))
for (let offset = 0; offset < records.length; offset += batchSize) {
const batch = records.slice(offset, offset + batchSize)
try { try {
const valuesSql: string[] = []
const params: (string | number | null)[] = []
batch.forEach((record, index) => {
const paramOffset = index * columnsPerRecord
valuesSql.push(
`(${dialect.param(paramOffset)}, ${dialect.param(paramOffset + 1)}, ${dialect.param(paramOffset + 2)}, ${dialect.param(paramOffset + 3)}, ${dialect.param(paramOffset + 4)}, ${dialect.currentTimestamp()}, 'pending')`
)
params.push(batchId, userId, username, record.productionId || null, record.orderNumber)
})
const sqlString = ` const sqlString = `
INSERT INTO ${tableName} INSERT INTO ${tableName}
(BatchId, UserId, Username, ProductionId, OrderNumber, OperationTime, Status) (BatchId, UserId, Username, ProductionId, OrderNumber, OperationTime, Status)
VALUES VALUES
(${dialect.param(0)}, ${dialect.param(1)}, ${dialect.param(2)}, ${dialect.param(3)}, ${dialect.param(4)}, ${dialect.currentTimestamp()}, 'pending') ${valuesSql.join(',\n ')}
` `
await trackDuration( await trackDuration(async () => await dbService.query(sqlString, params), {
async () => operationName: 'ExtractorOperationHistoryDAO.insertBatchRecords',
await dbService.query(sqlString, [ context: {
batchId, tableName,
userId, operationType: 'INSERT',
username, batchId,
record.productionId || null, batchOffset: offset,
record.orderNumber batchCount: batch.length
]),
{
operationName: 'ExtractorOperationHistoryDAO.insertBatchRecords',
context: { tableName, operationType: 'INSERT', batchId }
} }
) })
} catch (error) { } catch (error) {
log.error('Error inserting individual record', { log.error('Error inserting record batch', {
tableName, tableName,
operationType: 'INSERT', operationType: 'INSERT',
requestId, requestId,
batchId, batchId,
orderNumber: record.orderNumber, batchOffset: offset,
batchCount: batch.length,
error: error instanceof Error ? error.message : String(error) error: error instanceof Error ? error.message : String(error)
}) })
} }

View File

@@ -339,7 +339,11 @@ const SQL_KEYWORDS = new Set([
'IF', 'IF',
'CURRENT_TIMESTAMP', 'CURRENT_TIMESTAMP',
'NOW', 'NOW',
'GETDATE' 'GETDATE',
// ==================== Timezone Expression ====================
'AT',
'ZONE'
]) ])
/** /**

View File

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

View File

@@ -10,6 +10,7 @@ import type {
LogLevel LogLevel
} from '../../types/extractor.types' } from '../../types/extractor.types'
import { DataImportService } from '../database/data-importer' import { DataImportService } from '../database/data-importer'
import type { MaterialPlanRecord } from '../database/discrete-material-plan-dao'
import { createLogger, withRequestContext, getRequestId } from '../logger' import { createLogger, withRequestContext, getRequestId } from '../logger'
import { trackDuration } from '../logger/performance-monitor' import { trackDuration } from '../logger/performance-monitor'
@@ -112,16 +113,18 @@ export class ExtractorService {
// Always clean up temporary files regardless of merge success // Always clean up temporary files regardless of merge success
await this.cleanupTempFiles(result.downloadedFiles, input.orderNumbers) await this.cleanupTempFiles(result.downloadedFiles, input.orderNumbers)
// Auto-import to database if merge was successful // Auto-import parsed records directly. The merged Excel file is an archive artifact,
if (result.mergedFile) { // not the source for persistence.
if (mergeResult.records.length > 0) {
const importProgress = (1 + totalBatches + 1) * progressPerPoint const importProgress = (1 + totalBatches + 1) * progressPerPoint
input.onProgress?.('正在写入数据库...', importProgress, { input.onProgress?.('正在写入数据库...', importProgress, {
phase: 'importing', phase: 'importing',
totalBatches totalBatches
}) })
const importResult = await this.importToDatabaseWithLogging( const importResult = await this.importRecordsToDatabaseWithLogging(
result.mergedFile, mergeResult.records,
input.onLog input.onLog,
result.mergedFile
) )
result.importResult = importResult result.importResult = importResult
@@ -168,9 +171,10 @@ export class ExtractorService {
recordCount: number recordCount: number
error?: string error?: string
orderRecordCounts: Array<{ orderNumber: string; recordCount: number }> orderRecordCounts: Array<{ orderNumber: string; recordCount: number }>
records: MaterialPlanRecord[]
}> { }> {
if (filePaths.length === 0) { 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 }) 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 }) log.info('Merge summary', { orderCount: allOrders.length, recordCount })
const records = this.buildMaterialPlanRecords(allOrders)
if (recordCount === 0) { if (recordCount === 0) {
log.warn('No records found in any downloaded files', { orderNumbers }) 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 // Generate output filename with timestamp
@@ -238,7 +243,7 @@ export class ExtractorService {
log.info('Saving merged file', { outputPath }) log.info('Saving merged file', { outputPath })
await this.saveMergedOrders(allOrders, outputPath) await this.saveMergedOrders(allOrders, outputPath)
log.info('Merged file saved successfully', { recordCount }) log.info('Merged file saved successfully', { recordCount })
return { mergedFile: outputPath, recordCount, orderRecordCounts } return { mergedFile: outputPath, recordCount, orderRecordCounts, records }
} catch (error) { } catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error) const errorMsg = error instanceof Error ? error.message : String(error)
const errorStack = error instanceof Error ? error.stack : '' const errorStack = error instanceof Error ? error.stack : ''
@@ -253,6 +258,7 @@ export class ExtractorService {
mergedFile: null, mergedFile: null,
recordCount, recordCount,
orderRecordCounts, orderRecordCounts,
records,
error: `保存合并文件失败:${errorMsg}` error: `保存合并文件失败:${errorMsg}`
} }
} }
@@ -370,6 +376,78 @@ export class ExtractorService {
log.debug('File saved successfully', { outputPath }) 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 * Clean up temporary batch files after merging
* @param filePaths - Array of temporary file paths to delete * @param filePaths - Array of temporary file paths to delete
@@ -458,4 +536,70 @@ export class ExtractorService {
return trackedResult.result 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
}
} }

View File

@@ -32,6 +32,8 @@ const PRODUCTION_ID_PATTERN = /^\d{2}[A-Z]\d{1,6}$/i
*/ */
const ORDER_NUMBER_PATTERN = /^SC\d{14}$/i const ORDER_NUMBER_PATTERN = /^SC\d{14}$/i
const RESOLUTION_QUERY_BATCH_SIZE = 1000
/** /**
* Database table and field names * Database table and field names
* Loaded from config.yaml via ConfigManager * Loaded from config.yaml via ConfigManager
@@ -56,6 +58,14 @@ export class OrderNumberResolver {
this.dbService = dbService this.dbService = dbService
} }
private chunk<T>(items: T[], size: number): T[][] {
const chunks: T[][] = []
for (let index = 0; index < items.length; index += size) {
chunks.push(items.slice(index, index + size))
}
return chunks
}
/** /**
* Get table name based on database type * Get table name based on database type
* Converts schema.tablename format to database-specific quoting: * Converts schema.tablename format to database-specific quoting:
@@ -156,36 +166,36 @@ export class OrderNumberResolver {
// P1: Deduplicate input productionIds to avoid redundant queries // P1: Deduplicate input productionIds to avoid redundant queries
const uniqueProductionIds = [...new Set(productionIds)] const uniqueProductionIds = [...new Set(productionIds)]
// Use parameterized query to prevent SQL injection
const placeholders = uniqueProductionIds.map((_, i) => `@p${i}`).join(', ')
const params = uniqueProductionIds
let sql: string
if (this.dbService.type === 'sqlserver') {
// P0: Use DISTINCT to prevent duplicates from one-to-many relationships
// 使用 COLLATE 指定不区分大小写的排序规则
sql = `SELECT DISTINCT [${dbConfig.FIELD_PRODUCTION_ID}], [${dbConfig.FIELD_ORDER_NUMBER}] FROM ${tableName} WHERE [${dbConfig.FIELD_PRODUCTION_ID}] COLLATE SQL_Latin1_General_CP1_CI_AS IN (${placeholders})`
} else if (this.dbService.type === 'postgresql') {
// PostgreSQL: 使用双引号保护中文标识符UPPER 实现不区分大小写
// 注意getTableName() 已返回带双引号的表名,不应再加引号
const pgPlaceholders = uniqueProductionIds.map((_, i) => `UPPER($${i + 1})`).join(', ')
sql = `SELECT DISTINCT "${dbConfig.FIELD_PRODUCTION_ID}", "${dbConfig.FIELD_ORDER_NUMBER}" FROM ${tableName} WHERE UPPER("${dbConfig.FIELD_PRODUCTION_ID}") IN (${pgPlaceholders})`
} else {
const idPlaceholders = uniqueProductionIds.map(() => 'UPPER(?)').join(', ')
// P0: Use DISTINCT to prevent duplicates from one-to-many relationships
// MySQL: 使用 UPPER 确保不区分大小写
sql = `SELECT DISTINCT \`${dbConfig.FIELD_PRODUCTION_ID}\`, \`${dbConfig.FIELD_ORDER_NUMBER}\` FROM \`${tableName}\` WHERE UPPER(\`${dbConfig.FIELD_PRODUCTION_ID}\`) IN (${idPlaceholders})`
}
const result = await this.dbService.query(sql, params)
const mappings = new Map<string, string>() const mappings = new Map<string, string>()
for (const row of result.rows) { const batches = this.chunk(uniqueProductionIds, RESOLUTION_QUERY_BATCH_SIZE)
const keys = Object.keys(row)
const prodId = row[keys[0]] as string for (const batch of batches) {
const orderNum = row[keys[1]] as string let sql: string
if (prodId && orderNum) { const params = batch
mappings.set(prodId, orderNum)
if (this.dbService.type === 'sqlserver') {
const placeholders = batch.map((_, i) => `@p${i}`).join(', ')
// P0: Use DISTINCT to prevent duplicates from one-to-many relationships.
sql = `SELECT DISTINCT [${dbConfig.FIELD_PRODUCTION_ID}], [${dbConfig.FIELD_ORDER_NUMBER}] FROM ${tableName} WHERE [${dbConfig.FIELD_PRODUCTION_ID}] COLLATE SQL_Latin1_General_CP1_CI_AS IN (${placeholders})`
} else if (this.dbService.type === 'postgresql') {
// PostgreSQL: 使用双引号保护中文标识符UPPER 实现不区分大小写。
const pgPlaceholders = batch.map((_, i) => `UPPER($${i + 1})`).join(', ')
sql = `SELECT DISTINCT "${dbConfig.FIELD_PRODUCTION_ID}", "${dbConfig.FIELD_ORDER_NUMBER}" FROM ${tableName} WHERE UPPER("${dbConfig.FIELD_PRODUCTION_ID}") IN (${pgPlaceholders})`
} else {
const idPlaceholders = batch.map(() => 'UPPER(?)').join(', ')
// MySQL: 使用 UPPER 确保不区分大小写。
sql = `SELECT DISTINCT \`${dbConfig.FIELD_PRODUCTION_ID}\`, \`${dbConfig.FIELD_ORDER_NUMBER}\` FROM \`${tableName}\` WHERE UPPER(\`${dbConfig.FIELD_PRODUCTION_ID}\`) IN (${idPlaceholders})`
}
const result = await this.dbService.query(sql, params)
for (const row of result.rows) {
const keys = Object.keys(row)
const prodId = row[keys[0]] as string
const orderNum = row[keys[1]] as string
if (prodId && orderNum) {
mappings.set(prodId, orderNum)
}
} }
} }
@@ -235,13 +245,14 @@ export class OrderNumberResolver {
// Build results while preserving original input order // Build results while preserving original input order
// Note: Multiple productionIDs mapping to the same order number is VALID (not an error) // Note: Multiple productionIDs mapping to the same order number is VALID (not an error)
const results: OrderMapping[] = [] const results: OrderMapping[] = []
const processedInputs = new Set<string>()
for (const input of inputs) { for (const input of inputs) {
// Skip if this exact input was already processed // Skip if this exact input was already processed
const alreadyProcessed = results.some((r) => r.input === input) if (processedInputs.has(input)) {
if (alreadyProcessed) {
continue continue
} }
processedInputs.add(input)
const mapping: OrderMapping = { input, resolved: false } const mapping: OrderMapping = { input, resolved: false }

View File

@@ -62,12 +62,12 @@ const formatDateTime = (dateStr: string) => {
return dateStr // Return original if invalid return dateStr // Return original if invalid
} }
// Use UTC methods to display the time as stored in database (without timezone conversion) // Use local time for display
const year = date.getUTCFullYear() const year = date.getFullYear()
const month = String(date.getUTCMonth() + 1).padStart(2, '0') const month = String(date.getMonth() + 1).padStart(2, '0')
const day = String(date.getUTCDate()).padStart(2, '0') const day = String(date.getDate()).padStart(2, '0')
const hours = String(date.getUTCHours()).padStart(2, '0') const hours = String(date.getHours()).padStart(2, '0')
const minutes = String(date.getUTCMinutes()).padStart(2, '0') const minutes = String(date.getMinutes()).padStart(2, '0')
return `${year}-${month}-${day} ${hours}:${minutes}` return `${year}-${month}-${day} ${hours}:${minutes}`
} }

View File

@@ -46,8 +46,8 @@ describe('PostgreSqlDialect', () => {
}) })
describe('currentTimestamp', () => { describe('currentTimestamp', () => {
it('should return CURRENT_TIMESTAMP', () => { it('should return explicit UTC timestamp expression', () => {
expect(dialect.currentTimestamp()).toBe('CURRENT_TIMESTAMP') expect(dialect.currentTimestamp()).toBe("(NOW() AT TIME ZONE 'UTC')")
}) })
}) })

View File

@@ -0,0 +1,104 @@
import { describe, it, expect, beforeEach, vi } from 'vitest'
import {
DiscreteMaterialPlanDAO,
type MaterialPlanRecord
} from '../../../../src/main/services/database/discrete-material-plan-dao'
import type { IDatabaseService } from '../../../../src/main/services/database'
vi.mock('../../../../src/main/services/logger', () => ({
createLogger: vi.fn(() => ({
info: vi.fn(),
debug: vi.fn(),
warn: vi.fn(),
error: vi.fn()
})),
getRequestId: vi.fn(() => 'test-request-id'),
trackDuration: vi.fn(async (fn) => ({ result: await fn(), durationMs: 1, isSlow: false }))
}))
vi.mock('../../../../src/main/services/database', () => ({
create: vi.fn()
}))
function createRecord(sourceNumber: string, index: number): MaterialPlanRecord {
return {
factory: '工厂A',
materialStatus: '已审批',
planNumber: `PLAN-${index}`,
sourceNumber,
materialType: '标准',
productCode: 'P001',
productName: '产品A',
productUnit: 'PCS',
productPlanQuantity: 1,
useDepartment: '',
remark: '',
creator: '',
createDate: new Date('2026-04-28T00:00:00Z'),
approver: '',
approveDate: new Date('2026-04-28T00:00:00Z'),
sequenceNumber: index,
materialCode: `MAT-${index}`,
materialName: '物料A',
specification: '',
model: '',
drawingNumber: '',
materialQuality: '',
planQuantity: 1,
unit: 'PCS',
requiredDate: new Date('2026-04-28T00:00:00Z'),
warehouse: '',
unitUsage: 1,
cumulativeOutputQuantity: 0,
bomVersion: ''
}
}
describe('DiscreteMaterialPlanDAO', () => {
let mockDbService: IDatabaseService
beforeEach(async () => {
vi.clearAllMocks()
mockDbService = {
type: 'sqlserver',
connect: vi.fn(),
disconnect: vi.fn(),
isConnected: vi.fn(() => true),
query: vi.fn(async (_sql, params = []) => {
const rows = JSON.parse(params[1] || '[]')
return {
rows: [{ deletedCount: 0, insertedCount: rows.length }],
columns: ['deletedCount', 'insertedCount'],
rowCount: 1
}
}),
transaction: vi.fn()
}
const database = await import('../../../../src/main/services/database')
vi.mocked(database.create).mockResolvedValue(mockDbService)
})
it('splits SQL Server replace operations by source number batches', async () => {
const records = Array.from({ length: 151 }, (_, index) =>
createRecord(`SC-${String(index).padStart(4, '0')}`, index)
)
const dao = new DiscreteMaterialPlanDAO()
const result = await dao.replaceBySourceNumbers(records, 1000)
expect(result).toEqual({ deleted: 0, inserted: 151 })
expect(mockDbService.query).toHaveBeenCalledTimes(7)
const firstParams = vi.mocked(mockDbService.query).mock.calls[0][1] || []
const sixthParams = vi.mocked(mockDbService.query).mock.calls[5][1] || []
const seventhParams = vi.mocked(mockDbService.query).mock.calls[6][1] || []
expect(JSON.parse(firstParams[0])).toHaveLength(25)
expect(JSON.parse(sixthParams[0])).toHaveLength(25)
expect(JSON.parse(seventhParams[0])).toHaveLength(1)
expect(JSON.parse(firstParams[1])).toHaveLength(25)
expect(JSON.parse(sixthParams[1])).toHaveLength(25)
expect(JSON.parse(seventhParams[1])).toHaveLength(1)
})
})

View File

@@ -97,6 +97,14 @@ describe('ExtractorService', () => {
recordsImported: 0, recordsImported: 0,
uniqueSourceNumbers: 0, uniqueSourceNumbers: 0,
errors: [] errors: []
} as ImportResult),
importFromRecords: vi.fn().mockResolvedValue({
success: true,
recordsRead: 0,
recordsDeleted: 0,
recordsImported: 0,
uniqueSourceNumbers: 0,
errors: []
} as ImportResult) } as ImportResult)
} }
@@ -173,6 +181,65 @@ describe('ExtractorService', () => {
expect(Array.isArray(result.errors)).toBe(true) expect(Array.isArray(result.errors)).toBe(true)
}) })
it('should import parsed records directly instead of re-reading merged Excel', async () => {
mockExtractorCoreInstance.downloadAllBatches.mockResolvedValue({
downloadedFiles: ['./file1.xlsx'],
errors: []
})
mockExcelParserInstance.parse = vi.fn().mockImplementation(() => {
mockExcelParserInstance._lastOrders = [
{
orderInfo: {
factory: '工厂A',
planNumber: 'PLAN001',
productionOrder: 'ORD001',
productCode: 'P001',
productName: '产品A',
plannedQuantity: '10',
unit: 'PCS'
},
materials: [
{
sequence: 1,
materialCode: 'MAT001',
materialName: '物料A',
quantity: 2,
unit: 'PCS'
}
]
}
]
return Promise.resolve()
})
mockDataImportInstance.importFromRecords.mockResolvedValue({
success: true,
recordsRead: 1,
recordsDeleted: 0,
recordsImported: 1,
uniqueSourceNumbers: 1,
errors: []
} as ImportResult)
const service = new ExtractorService(mockAuthService, './test-downloads')
vi.spyOn(service as any, 'saveMergedOrders').mockResolvedValue(undefined)
const result = await service.extract({
orderNumbers: ['ORD001'],
onProgress: vi.fn(),
onLog: vi.fn()
})
expect(result.importResult?.success).toBe(true)
expect(mockDataImportInstance.importFromRecords).toHaveBeenCalledTimes(1)
expect(mockDataImportInstance.importFromExcel).not.toHaveBeenCalled()
expect(mockDataImportInstance.importFromRecords.mock.calls[0][0][0]).toMatchObject({
planNumber: 'PLAN001',
sourceNumber: 'ORD001',
materialCode: 'MAT001',
planQuantity: 2
})
})
}) })
describe('mergeFiles()', () => { describe('mergeFiles()', () => {
@@ -196,6 +263,7 @@ describe('ExtractorService', () => {
] ]
const service = new ExtractorService(mockAuthService, './test-downloads') const service = new ExtractorService(mockAuthService, './test-downloads')
vi.spyOn(service as any, 'saveMergedOrders').mockResolvedValue(undefined)
// @ts-ignore - accessing private method for testing // @ts-ignore - accessing private method for testing
const result = await service.mergeFiles(['./file1.xlsx'], ['ORD001']) const result = await service.mergeFiles(['./file1.xlsx'], ['ORD001'])
@@ -231,6 +299,7 @@ describe('ExtractorService', () => {
}) })
const service = new ExtractorService(mockAuthService, './test-downloads') const service = new ExtractorService(mockAuthService, './test-downloads')
vi.spyOn(service as any, 'saveMergedOrders').mockResolvedValue(undefined)
// @ts-ignore - accessing private method for testing // @ts-ignore - accessing private method for testing
const result = await service.mergeFiles( const result = await service.mergeFiles(

View File

@@ -173,6 +173,25 @@ describe('OrderNumberResolver', () => {
// Should be optimized to query unique values only // Should be optimized to query unique values only
expect(mockDbService.query).toHaveBeenCalledTimes(1) expect(mockDbService.query).toHaveBeenCalledTimes(1)
}) })
it('splits large mapping queries into bounded batches', async () => {
const largeInput = Array.from({ length: 1001 }, (_, i) => `22A${i}`)
vi.mocked(mockDbService.query).mockImplementation(async (_sql, params = []) => ({
rows: params.map((prodId, i) => ({
总排号: prodId,
: `SC7020260212${String(i).padStart(5, '0')}`
})),
columns: ['总排号', '生产订单号'],
rowCount: params.length
}))
await resolver.mapProductionIdsToOrderNumbers(largeInput)
expect(mockDbService.query).toHaveBeenCalledTimes(2)
expect(vi.mocked(mockDbService.query).mock.calls[0][1]).toHaveLength(1000)
expect(vi.mocked(mockDbService.query).mock.calls[1][1]).toHaveLength(1)
})
}) })
describe('error handling', () => { describe('error handling', () => {