248 lines
8.9 KiB
TypeScript
248 lines
8.9 KiB
TypeScript
import { BadRequestException, ConflictException, Injectable } from '@nestjs/common';
|
||
import { InjectRepository } from '@nestjs/typeorm';
|
||
import { DataSource, In, Repository } from 'typeorm';
|
||
import { ImportRun } from './entities/import-run.entity';
|
||
import { ImportStep } from './entities/import-step.entity';
|
||
import { ImportRow } from './entities/import-row.entity';
|
||
import { IMPORT_ACTION_LABELS, IMPORT_STEP_LABELS, IMPORT_STEP_ORDER } from './imports.types';
|
||
import type {
|
||
CellValue,
|
||
ImportRowAction,
|
||
ImportRowDecision,
|
||
ImportStepKey,
|
||
StepCommitReceipt,
|
||
StepPreviewSummary,
|
||
} from './imports.types';
|
||
import { csvCell, parseJson, safeError } from './imports.helpers';
|
||
import { writeRow } from './imports.rows';
|
||
import { assertStepPermission, findOwnedRun, findStep } from './imports.access';
|
||
import type { ImportPrincipal } from './imports.access';
|
||
|
||
@Injectable()
|
||
export class ImportCommitService {
|
||
constructor(
|
||
@InjectRepository(ImportRun)
|
||
private readonly runs: Repository<ImportRun>,
|
||
@InjectRepository(ImportStep)
|
||
private readonly steps: Repository<ImportStep>,
|
||
@InjectRepository(ImportRow)
|
||
private readonly rows: Repository<ImportRow>,
|
||
private readonly dataSource: DataSource,
|
||
) {}
|
||
|
||
async commitStep(
|
||
principal: ImportPrincipal,
|
||
runId: string,
|
||
stepKey: ImportStepKey,
|
||
decisions: ImportRowDecision[],
|
||
): Promise<StepCommitReceipt> {
|
||
const run = await findOwnedRun(this.runs, principal.id, runId);
|
||
if (run.status === 'committed') {
|
||
const receipt = await this.existingReceipt(run, stepKey);
|
||
return { ...receipt, status: 'already_committed' };
|
||
}
|
||
if (run.currentStepKey !== stepKey) {
|
||
return {
|
||
runId,
|
||
stepKey,
|
||
status: 'conflict',
|
||
created: 0,
|
||
updated: 0,
|
||
skipped: 0,
|
||
failed: 0,
|
||
total: 0,
|
||
nextStepKey: run.currentStepKey,
|
||
runStatus: run.status,
|
||
message: `请先完成「${run.currentStepKey ? IMPORT_STEP_LABELS[run.currentStepKey] : ''}」阶段`,
|
||
};
|
||
}
|
||
const step = await findStep(this.steps, runId, stepKey);
|
||
if (!step || step.status === 'skipped') {
|
||
throw new BadRequestException(`阶段「${IMPORT_STEP_LABELS[stepKey]}」没有分配工作表`);
|
||
}
|
||
if (step.status === 'committed') {
|
||
const receipt = await this.existingReceipt(run, stepKey);
|
||
return { ...receipt, status: 'already_committed' };
|
||
}
|
||
if (step.status !== 'ready') {
|
||
throw new BadRequestException(`阶段「${IMPORT_STEP_LABELS[stepKey]}」尚未预览，请先预览确认`);
|
||
}
|
||
assertStepPermission(principal, stepKey);
|
||
|
||
const pendingRows = await this.rows.find({ where: { stepId: step.id, status: 'valid' } });
|
||
const decisionMap = new Map<number, ImportRowAction>();
|
||
for (const decision of decisions ?? []) {
|
||
if (
|
||
Number.isInteger(decision.rowId) &&
|
||
(decision.action === 'create' || decision.action === 'update' || decision.action === 'skip')
|
||
) {
|
||
decisionMap.set(decision.rowId, decision.action);
|
||
}
|
||
}
|
||
if (pendingRows.length === 0) {
|
||
throw new BadRequestException('没有可提交的有效行，请检查预览结果');
|
||
}
|
||
const rowById = new Map(pendingRows.map((row) => [row.id, row]));
|
||
for (const [rowId, action] of decisionMap) {
|
||
const row = rowById.get(rowId);
|
||
if (!row) continue;
|
||
if (action === 'skip') continue;
|
||
if (!row.action) {
|
||
throw new BadRequestException(
|
||
`第 ${row.rowNumber} 行（${row.sheetName}）没有预览判定，只能选择「跳过」`,
|
||
);
|
||
}
|
||
if (action !== row.action) {
|
||
throw new BadRequestException(
|
||
`第 ${row.rowNumber} 行（${row.sheetName}）预览判定为「${IMPORT_ACTION_LABELS[row.action]}」，不能改为「${IMPORT_ACTION_LABELS[action]}」`,
|
||
);
|
||
}
|
||
}
|
||
|
||
run.status = 'committing';
|
||
step.status = 'committing';
|
||
await this.runs.save(run);
|
||
await this.steps.save(step);
|
||
|
||
const counts = { created: 0, updated: 0, skipped: 0, failed: 0 };
|
||
try {
|
||
await this.dataSource.transaction(async (manager) => {
|
||
for (const row of pendingRows) {
|
||
const action = decisionMap.get(row.id) ?? row.action ?? 'create';
|
||
if (action === 'skip') {
|
||
row.status = 'skipped';
|
||
row.action = 'skip';
|
||
counts.skipped += 1;
|
||
await manager.save(ImportRow, row);
|
||
continue;
|
||
}
|
||
try {
|
||
const fields = parseJson<Record<string, CellValue>>(row.normalizedJson) ?? {};
|
||
const targetId = await writeRow(manager, stepKey, action, fields, row.targetId);
|
||
row.status = 'committed';
|
||
row.action = action;
|
||
row.targetId = targetId ?? row.targetId;
|
||
if (action === 'create') counts.created += 1;
|
||
else counts.updated += 1;
|
||
} catch (error) {
|
||
row.status = 'error';
|
||
row.errorsJson = JSON.stringify([`写入失败：${safeError(error)}`]);
|
||
counts.failed += 1;
|
||
}
|
||
await manager.save(ImportRow, row);
|
||
}
|
||
});
|
||
} catch (error) {
|
||
run.status = 'failed';
|
||
run.error = safeError(error).slice(0, 500);
|
||
await this.runs.save(run);
|
||
throw new ConflictException(`提交失败：${safeError(error)}`);
|
||
}
|
||
|
||
const summary: StepPreviewSummary = {
|
||
total: pendingRows.length,
|
||
valid: counts.created + counts.updated + counts.skipped,
|
||
error: counts.failed,
|
||
create: counts.created,
|
||
update: counts.updated,
|
||
skip: counts.skipped,
|
||
};
|
||
step.status = 'committed';
|
||
step.committedAt = new Date();
|
||
step.summaryJson = JSON.stringify(summary);
|
||
await this.steps.save(step);
|
||
|
||
const nextStepKey = await this.nextStepKey(runId, stepKey);
|
||
run.currentStepKey = nextStepKey;
|
||
run.status = nextStepKey ? 'ready' : 'committed';
|
||
await this.runs.save(run);
|
||
|
||
const message =
|
||
`阶段「${IMPORT_STEP_LABELS[stepKey]}」提交完成：新建 ${counts.created}、更新 ${counts.updated}、跳过 ${counts.skipped}、失败 ${counts.failed}；` +
|
||
(nextStepKey ? `下一步：${IMPORT_STEP_LABELS[nextStepKey]}` : '全部阶段已完成');
|
||
return {
|
||
runId,
|
||
stepKey,
|
||
status: 'committed',
|
||
created: counts.created,
|
||
updated: counts.updated,
|
||
skipped: counts.skipped,
|
||
failed: counts.failed,
|
||
total: pendingRows.length,
|
||
nextStepKey,
|
||
runStatus: run.status,
|
||
message,
|
||
};
|
||
}
|
||
|
||
async errorReport(
|
||
userId: number,
|
||
runId: string,
|
||
stepKey?: ImportStepKey,
|
||
): Promise<{ filename: string; buffer: Buffer }> {
|
||
const run = await findOwnedRun(this.runs, userId, runId);
|
||
const stepRecords = await this.steps.find({ where: { runId }, order: { id: 'ASC' } });
|
||
const stepIds = stepKey
|
||
? stepRecords.filter((s) => s.stepKey === stepKey).map((s) => s.id)
|
||
: stepRecords.map((s) => s.id);
|
||
if (stepIds.length === 0) return { filename: '', buffer: Buffer.from('') };
|
||
const rows = await this.rows.find({
|
||
where: { stepId: In(stepIds), status: 'error' },
|
||
order: { id: 'ASC' },
|
||
});
|
||
const lines: string[] = ['工作表,行号,原始数据,错误信息'];
|
||
for (const row of rows) {
|
||
const raw = parseJson<Record<string, CellValue>>(row.rawJson) ?? {};
|
||
const errors = parseJson<string[]>(row.errorsJson) ?? [];
|
||
lines.push(
|
||
[
|
||
csvCell(row.sheetName),
|
||
String(row.rowNumber),
|
||
csvCell(JSON.stringify(raw)),
|
||
csvCell(errors.join('；')),
|
||
].join(','),
|
||
);
|
||
}
|
||
return {
|
||
filename: `导入错误报告-${run.fileName.replace(/\.(xlsx|csv)$/i, '')}.csv`,
|
||
buffer: Buffer.from(`\uFEFF${lines.join('\n')}`, 'utf8'),
|
||
};
|
||
}
|
||
|
||
private async nextStepKey(
|
||
runId: string,
|
||
currentKey: ImportStepKey,
|
||
): Promise<ImportStepKey | null> {
|
||
const stepRecords = await this.steps.find({ where: { runId } });
|
||
const currentIndex = IMPORT_STEP_ORDER.indexOf(currentKey);
|
||
for (let i = currentIndex + 1; i < IMPORT_STEP_ORDER.length; i += 1) {
|
||
const candidate = IMPORT_STEP_ORDER[i];
|
||
const step = stepRecords.find((s) => s.stepKey === candidate);
|
||
if (step && step.status !== 'skipped' && step.status !== 'committed') {
|
||
return candidate;
|
||
}
|
||
}
|
||
return null;
|
||
}
|
||
|
||
private async existingReceipt(
|
||
run: ImportRun,
|
||
stepKey: ImportStepKey,
|
||
): Promise<Omit<StepCommitReceipt, 'status'>> {
|
||
const step = await findStep(this.steps, run.id, stepKey);
|
||
const summary = parseJson<StepPreviewSummary>(step?.summaryJson);
|
||
return {
|
||
runId: run.id,
|
||
stepKey,
|
||
created: summary?.create ?? 0,
|
||
updated: summary?.update ?? 0,
|
||
skipped: summary?.skip ?? 0,
|
||
failed: summary?.error ?? 0,
|
||
total: summary?.total ?? 0,
|
||
nextStepKey: run.currentStepKey,
|
||
runStatus: run.status,
|
||
message: `阶段「${IMPORT_STEP_LABELS[stepKey]}」此前已提交`,
|
||
};
|
||
}
|
||
}
|