merge: recover BOM changed sources and batch synthesis
This commit is contained in:
commit
a11aa15017
@ -151,7 +151,7 @@ export const competitorAnalysisResultV1Schema = z.object({
|
||||
}).optional(),
|
||||
),
|
||||
}),
|
||||
skus: z.array(competitorSkuSchema).min(1).max(200),
|
||||
skus: z.array(competitorSkuSchema).min(1).max(600),
|
||||
analyzedAt: z.iso.datetime(),
|
||||
})
|
||||
|
||||
|
||||
@ -218,6 +218,8 @@ describe('BOM 多模态分析器', () => {
|
||||
const originalMaxImages = process.env.BOM_COSTING_AI_MAX_IMAGES
|
||||
const originalFactBatchSize = process.env.BOM_COSTING_AI_FACT_BATCH_SIZE
|
||||
const originalAnalysisTimeout = process.env.BOM_COSTING_AI_TIMEOUT_MS
|
||||
const originalSynthesisSkuBatchSize =
|
||||
process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE
|
||||
|
||||
beforeEach(() => {
|
||||
process.env.BOM_COSTING_AI_FACT_BATCH_SIZE = '6'
|
||||
@ -245,6 +247,12 @@ describe('BOM 多模态分析器', () => {
|
||||
} else {
|
||||
process.env.BOM_COSTING_AI_TIMEOUT_MS = originalAnalysisTimeout
|
||||
}
|
||||
if (originalSynthesisSkuBatchSize === undefined) {
|
||||
delete process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE
|
||||
} else {
|
||||
process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE =
|
||||
originalSynthesisSkuBatchSize
|
||||
}
|
||||
jest.restoreAllMocks()
|
||||
})
|
||||
|
||||
@ -722,6 +730,56 @@ describe('BOM 多模态分析器', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('将高规格商品拆成多个汇总批次,并覆盖每一个平台 SKU', async () => {
|
||||
process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE = '100'
|
||||
const highCardinalitySnapshot: MarketDetailSnapshot = {
|
||||
...snapshot,
|
||||
skus: Array.from({ length: 201 }, (_, index) => ({
|
||||
...snapshot.skus[0],
|
||||
externalSkuId: `sku-${index + 1}`,
|
||||
name: `规格 ${index + 1}`,
|
||||
})),
|
||||
}
|
||||
const batchSizes: number[] = []
|
||||
const fetchSpy = jest.spyOn(global, 'fetch').mockImplementation(
|
||||
async (_input, init) => {
|
||||
const requestBody = JSON.parse(String((init as RequestInit).body)) as {
|
||||
messages?: Array<{ content: string }>
|
||||
}
|
||||
const prompt = JSON.parse(
|
||||
requestBody.messages?.[1]?.content ?? '{}',
|
||||
) as {
|
||||
skuTruth?: Array<{ externalSkuId: string }>
|
||||
}
|
||||
const skuTruth = prompt.skuTruth ?? []
|
||||
batchSizes.push(skuTruth.length)
|
||||
return createStreamingResponse(JSON.stringify({
|
||||
product: { category: '书桌' },
|
||||
structureGroups: [analyzedStructureGroup()],
|
||||
skus: skuTruth.map((sku) => analyzedSku(sku.externalSkuId)),
|
||||
}))
|
||||
},
|
||||
)
|
||||
|
||||
const output = await new BomCostingMultimodalAnalyzer().analyzeWithTrace({
|
||||
sourceUrl: highCardinalitySnapshot.detailUrl,
|
||||
snapshot: highCardinalitySnapshot,
|
||||
existingFactSnapshot: createCompleteFactSnapshot(highCardinalitySnapshot),
|
||||
})
|
||||
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(3)
|
||||
expect(batchSizes).toEqual([100, 100, 1])
|
||||
expect(output.result.skus).toHaveLength(201)
|
||||
expect(new Set(output.result.skus.map((sku) => sku.externalSkuId)).size).toBe(201)
|
||||
expect(output.result.skus[0]?.structureGroupKey).toBe('batch-1-desk-120')
|
||||
expect(output.result.skus[100]?.structureGroupKey).toBe('batch-2-desk-120')
|
||||
expect(output.result.skus[200]?.structureGroupKey).toBe('batch-3-desk-120')
|
||||
expect(output.trace).toMatchObject({
|
||||
inputSkuCount: 201,
|
||||
synthesisBatchCount: 3,
|
||||
})
|
||||
})
|
||||
|
||||
it('最终汇总紧凑传递逐图事实,不丢失可核验的证据内容', async () => {
|
||||
const factSnapshot = createCompleteFactSnapshot(snapshot)
|
||||
const image = factSnapshot.imageFacts[0]!
|
||||
|
||||
@ -220,7 +220,8 @@ const aiAnalysisPayloadSchema = z.object({
|
||||
),
|
||||
})
|
||||
|
||||
const MAX_MULTIMODAL_ANALYSIS_SKUS = 200
|
||||
const MAX_MULTIMODAL_ANALYSIS_SKUS = 500
|
||||
const DEFAULT_SYNTHESIS_SKU_BATCH_SIZE = 80
|
||||
const MAX_EXPLICIT_FACT_SKU_IDS = 20
|
||||
|
||||
function describeHighCardinalitySkuCatalog(skus: MarketSkuSnapshot[]) {
|
||||
@ -260,6 +261,8 @@ export interface BomCostingMultimodalTrace {
|
||||
imageCount: number
|
||||
inputSkuCount: number
|
||||
factBatchCount: number
|
||||
synthesisBatchCount: number
|
||||
synthesisCallIds: string[]
|
||||
reusedImageCount: number
|
||||
analyzedImageCount: number
|
||||
durationMs: number
|
||||
@ -272,6 +275,18 @@ export interface BomCostingMultimodalOutput {
|
||||
trace: BomCostingMultimodalTrace
|
||||
}
|
||||
|
||||
export interface BomCostingSynthesisBatchProgress {
|
||||
batchIndex: number
|
||||
totalBatches: number
|
||||
batchSkuCount: number
|
||||
completedSkuCount: number
|
||||
model: string
|
||||
callId: string | null
|
||||
durationMs: number
|
||||
promptTokens: number | null
|
||||
completionTokens: number | null
|
||||
}
|
||||
|
||||
function serializeSkuForPrompt(sku: MarketSkuSnapshot) {
|
||||
return {
|
||||
externalSkuId: sku.externalSkuId,
|
||||
@ -327,9 +342,10 @@ function buildUserText(
|
||||
snapshot: MarketDetailSnapshot,
|
||||
factSnapshot: BomCostingSourceFactSnapshot,
|
||||
analysisInstruction?: string,
|
||||
skus: MarketSkuSnapshot[] = snapshot.skus,
|
||||
) {
|
||||
return JSON.stringify({
|
||||
task: '合并全部逐图事实与 SKU 真相,生成结构组级家具 BOM,并把每个 SKU 绑定到对应结构组',
|
||||
task: '合并已保存的逐图事实与当前批次 SKU 真相,生成结构组级家具 BOM,并把当前批次每个 SKU 绑定到对应结构组',
|
||||
outputContractVersion: 'bom-costing/synthesis/v4',
|
||||
reviewInstruction: analysisInstruction?.trim() || undefined,
|
||||
reviewInstructionRule:
|
||||
@ -355,13 +371,40 @@ function buildUserText(
|
||||
shopName: snapshot.shopName,
|
||||
detailUrl: snapshot.detailUrl,
|
||||
skuCount: snapshot.skus.length,
|
||||
batchSkuCount: skus.length,
|
||||
},
|
||||
skuTruth: snapshot.skus.map(serializeSkuForPrompt),
|
||||
skuTruth: skus.map(serializeSkuForPrompt),
|
||||
installationKnowledge: buildSynthesisInstallationKnowledge(snapshot),
|
||||
extractedImageFacts: buildSynthesisFactInput(factSnapshot),
|
||||
})
|
||||
}
|
||||
|
||||
function splitSkuBatches(skus: MarketSkuSnapshot[]) {
|
||||
const batchSize = readBomCostingBoundedInteger(
|
||||
process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE,
|
||||
DEFAULT_SYNTHESIS_SKU_BATCH_SIZE,
|
||||
1,
|
||||
120,
|
||||
)
|
||||
return Array.from(
|
||||
{ length: Math.ceil(skus.length / batchSize) },
|
||||
(_, index) => skus.slice(index * batchSize, (index + 1) * batchSize),
|
||||
)
|
||||
}
|
||||
|
||||
function scopeBatchStructureGroupKeys(
|
||||
skus: z.infer<typeof competitorSkuSchema>[],
|
||||
batchIndex: number,
|
||||
totalBatches: number,
|
||||
) {
|
||||
if (totalBatches <= 1) return skus
|
||||
const prefix = `batch-${batchIndex + 1}-`
|
||||
return skus.map((sku) => ({
|
||||
...sku,
|
||||
structureGroupKey: `${prefix}${sku.structureGroupKey}`.slice(0, 160),
|
||||
}))
|
||||
}
|
||||
|
||||
function describeAnalysisPayloadError(error: unknown): string {
|
||||
if (error instanceof z.ZodError) {
|
||||
const hasSkuIssue = error.issues.some((issue) => issue.path[0] === 'skus')
|
||||
@ -495,6 +538,15 @@ export interface BomCostingMultimodalAnalyzeInput {
|
||||
snapshot: BomCostingSourceFactSnapshot
|
||||
trace: BomCostingFactExtractionTrace
|
||||
}) => Promise<void>
|
||||
onSynthesisBatchStarted?: (
|
||||
progress: Pick<
|
||||
BomCostingSynthesisBatchProgress,
|
||||
'batchIndex' | 'totalBatches' | 'batchSkuCount' | 'completedSkuCount'
|
||||
>,
|
||||
) => Promise<void>
|
||||
onSynthesisBatchCompleted?: (
|
||||
progress: BomCostingSynthesisBatchProgress,
|
||||
) => Promise<void>
|
||||
}
|
||||
|
||||
export class BomCostingMultimodalAnalyzer {
|
||||
@ -555,81 +607,144 @@ export class BomCostingMultimodalAnalyzer {
|
||||
const maxTokens = /^gpt-5(?:[.-]|$)/i.test(config.model)
|
||||
? Math.min(configuredMaxTokens, 12_000)
|
||||
: configuredMaxTokens
|
||||
const responsePayload = await callBomCostingMultimodalModel({
|
||||
config,
|
||||
operation: 'SYNTHESIS',
|
||||
systemPrompt: SYSTEM_PROMPT,
|
||||
userContent: buildUserText(
|
||||
input.snapshot,
|
||||
factOutput.snapshot,
|
||||
input.analysisInstruction,
|
||||
),
|
||||
maxTokens,
|
||||
timeoutMs,
|
||||
})
|
||||
const skuBatches = splitSkuBatches(input.snapshot.skus)
|
||||
const analyzedSkus: z.infer<typeof competitorSkuSchema>[] = []
|
||||
let product: z.infer<typeof aiAnalysisPayloadSchema>['product'] | null =
|
||||
null
|
||||
let model = config.model
|
||||
let callId: string | null = null
|
||||
const synthesisCallIds: string[] = []
|
||||
let synthesisDurationMs = 0
|
||||
let synthesisPromptTokens: number | null = null
|
||||
let synthesisCompletionTokens: number | null = null
|
||||
|
||||
try {
|
||||
const parsedPayload = parseUniqueAnalysisPayload(
|
||||
responsePayload.content,
|
||||
)
|
||||
const analyzedSkus = expandStructureGroupBom(parsedPayload)
|
||||
assertExactSkuCoverage(input.snapshot.skus, analyzedSkus)
|
||||
const imageUrls = Array.from(
|
||||
new Set([
|
||||
...input.snapshot.mainImages,
|
||||
...input.snapshot.detailImages,
|
||||
...input.snapshot.skuImages,
|
||||
]),
|
||||
).slice(0, 30)
|
||||
|
||||
const result = competitorAnalysisResultV1Schema.parse({
|
||||
schemaVersion: BOM_COSTING_ANALYSIS_SCHEMA_VERSION,
|
||||
sourceUrl: input.sourceUrl,
|
||||
product: {
|
||||
title: input.snapshot.title,
|
||||
brand: parsedPayload.product.brand,
|
||||
category: parsedPayload.product.category,
|
||||
imageUrls,
|
||||
dimensions: parsedPayload.product.dimensions,
|
||||
},
|
||||
skus: mergeSourceSkuTruth(input.snapshot.skus, analyzedSkus),
|
||||
analyzedAt: new Date().toISOString(),
|
||||
for (const [batchIndex, skuBatch] of skuBatches.entries()) {
|
||||
await input.onSynthesisBatchStarted?.({
|
||||
batchIndex: batchIndex + 1,
|
||||
totalBatches: skuBatches.length,
|
||||
batchSkuCount: skuBatch.length,
|
||||
completedSkuCount: analyzedSkus.length,
|
||||
})
|
||||
return {
|
||||
result,
|
||||
trace: {
|
||||
const responsePayload = await callBomCostingMultimodalModel({
|
||||
config,
|
||||
operation: 'SYNTHESIS',
|
||||
systemPrompt: SYSTEM_PROMPT,
|
||||
userContent: buildUserText(
|
||||
input.snapshot,
|
||||
factOutput.snapshot,
|
||||
input.analysisInstruction,
|
||||
skuBatch,
|
||||
),
|
||||
maxTokens,
|
||||
timeoutMs,
|
||||
})
|
||||
|
||||
try {
|
||||
const parsedPayload = parseUniqueAnalysisPayload(
|
||||
responsePayload.content,
|
||||
)
|
||||
const batchSkus = expandStructureGroupBom(parsedPayload)
|
||||
assertExactSkuCoverage(skuBatch, batchSkus)
|
||||
analyzedSkus.push(
|
||||
...scopeBatchStructureGroupKeys(
|
||||
batchSkus,
|
||||
batchIndex,
|
||||
skuBatches.length,
|
||||
),
|
||||
)
|
||||
product ??= parsedPayload.product
|
||||
model = responsePayload.model || model
|
||||
callId ??= responsePayload.id
|
||||
if (responsePayload.id) synthesisCallIds.push(responsePayload.id)
|
||||
synthesisDurationMs += responsePayload.durationMs
|
||||
synthesisPromptTokens = addNullableNumber(
|
||||
synthesisPromptTokens,
|
||||
responsePayload.promptTokens,
|
||||
)
|
||||
synthesisCompletionTokens = addNullableNumber(
|
||||
synthesisCompletionTokens,
|
||||
responsePayload.completionTokens,
|
||||
)
|
||||
await input.onSynthesisBatchCompleted?.({
|
||||
batchIndex: batchIndex + 1,
|
||||
totalBatches: skuBatches.length,
|
||||
batchSkuCount: skuBatch.length,
|
||||
completedSkuCount: analyzedSkus.length,
|
||||
model: responsePayload.model,
|
||||
callId: responsePayload.id,
|
||||
imageCount: factOutput.snapshot.imageFacts.length,
|
||||
inputSkuCount: input.snapshot.skus.length,
|
||||
factBatchCount: factOutput.trace.batchCount,
|
||||
reusedImageCount: factOutput.trace.reusedImageCount,
|
||||
analyzedImageCount: factOutput.trace.analyzedImageCount,
|
||||
durationMs:
|
||||
factOutput.trace.durationMs + responsePayload.durationMs,
|
||||
promptTokens: addNullableNumber(
|
||||
factOutput.trace.promptTokens,
|
||||
responsePayload.promptTokens,
|
||||
),
|
||||
completionTokens: addNullableNumber(
|
||||
factOutput.trace.completionTokens,
|
||||
responsePayload.completionTokens,
|
||||
),
|
||||
},
|
||||
durationMs: responsePayload.durationMs,
|
||||
promptTokens: responsePayload.promptTokens,
|
||||
completionTokens: responsePayload.completionTokens,
|
||||
})
|
||||
} catch (error) {
|
||||
console.warn('[BOM costing] multimodal synthesis contract rejected', {
|
||||
batchIndex: batchIndex + 1,
|
||||
totalBatches: skuBatches.length,
|
||||
batchSkuCount: skuBatch.length,
|
||||
model: responsePayload.model,
|
||||
callId: responsePayload.id,
|
||||
finishReason: responsePayload.finishReason,
|
||||
issues: describeAiOutputIssues(error),
|
||||
errorName: error instanceof Error ? error.name : 'UnknownError',
|
||||
})
|
||||
throw new AppError(
|
||||
ErrorCode.BOM_COSTING_INVALID_PROVIDER,
|
||||
describeAnalysisPayloadError(error),
|
||||
)
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn('[BOM costing] multimodal synthesis contract rejected', {
|
||||
model: responsePayload.model,
|
||||
callId: responsePayload.id,
|
||||
finishReason: responsePayload.finishReason,
|
||||
issues: describeAiOutputIssues(error),
|
||||
errorName: error instanceof Error ? error.name : 'UnknownError',
|
||||
})
|
||||
}
|
||||
|
||||
if (!product) {
|
||||
throw new AppError(
|
||||
ErrorCode.BOM_COSTING_INVALID_PROVIDER,
|
||||
describeAnalysisPayloadError(error),
|
||||
'多模态分析没有返回产品结构信息',
|
||||
)
|
||||
}
|
||||
assertExactSkuCoverage(input.snapshot.skus, analyzedSkus)
|
||||
const imageUrls = Array.from(
|
||||
new Set([
|
||||
...input.snapshot.mainImages,
|
||||
...input.snapshot.detailImages,
|
||||
...input.snapshot.skuImages,
|
||||
]),
|
||||
).slice(0, 30)
|
||||
|
||||
const result = competitorAnalysisResultV1Schema.parse({
|
||||
schemaVersion: BOM_COSTING_ANALYSIS_SCHEMA_VERSION,
|
||||
sourceUrl: input.sourceUrl,
|
||||
product: {
|
||||
title: input.snapshot.title,
|
||||
brand: product.brand,
|
||||
category: product.category,
|
||||
imageUrls,
|
||||
dimensions: product.dimensions,
|
||||
},
|
||||
skus: mergeSourceSkuTruth(input.snapshot.skus, analyzedSkus),
|
||||
analyzedAt: new Date().toISOString(),
|
||||
})
|
||||
return {
|
||||
result,
|
||||
trace: {
|
||||
model,
|
||||
callId,
|
||||
imageCount: factOutput.snapshot.imageFacts.length,
|
||||
inputSkuCount: input.snapshot.skus.length,
|
||||
factBatchCount: factOutput.trace.batchCount,
|
||||
synthesisBatchCount: skuBatches.length,
|
||||
synthesisCallIds,
|
||||
reusedImageCount: factOutput.trace.reusedImageCount,
|
||||
analyzedImageCount: factOutput.trace.analyzedImageCount,
|
||||
durationMs: factOutput.trace.durationMs + synthesisDurationMs,
|
||||
promptTokens: addNullableNumber(
|
||||
factOutput.trace.promptTokens,
|
||||
synthesisPromptTokens,
|
||||
),
|
||||
completionTokens: addNullableNumber(
|
||||
factOutput.trace.completionTokens,
|
||||
synthesisCompletionTokens,
|
||||
),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -19,6 +19,8 @@ import { marketDetailClient, type MarketDetailSnapshot } from './market-detail-c
|
||||
const mockRunAnalysis = jest.fn()
|
||||
const mockBuildStoredSnapshot = jest.fn()
|
||||
const mockRefreshSourceMetadata = jest.fn()
|
||||
const mockArchive = jest.fn()
|
||||
const mockSourceImagesChangedError = class extends Error {}
|
||||
|
||||
jest.mock('@/lib/storage', () => ({
|
||||
readObjectStorageKeyFromUrl: (value: string) => {
|
||||
@ -57,9 +59,10 @@ jest.mock('./bom-costing-source-archive.service', () => ({
|
||||
buildMarketSnapshotFromStoredSource: (...args: unknown[]) =>
|
||||
mockBuildStoredSnapshot(...args),
|
||||
bomCostingSourceArchiveService: {
|
||||
archive: jest.fn(),
|
||||
archive: (...args: unknown[]) => mockArchive(...args),
|
||||
refreshMetadata: (...args: unknown[]) => mockRefreshSourceMetadata(...args),
|
||||
},
|
||||
BomCostingSourceImagesChangedError: mockSourceImagesChangedError,
|
||||
}))
|
||||
|
||||
const snapshot: MarketDetailSnapshot = {
|
||||
@ -297,6 +300,8 @@ function caseRecord() {
|
||||
describe('BomCostingProcessingService', () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks()
|
||||
mockArchive.mockReset()
|
||||
mockRefreshSourceMetadata.mockReset()
|
||||
jest.spyOn(bomCostingRepository, 'claimCaseForProcessing').mockResolvedValue(true)
|
||||
jest.spyOn(bomCostingRepository, 'findProcessingAnalysisRun')
|
||||
.mockResolvedValue({
|
||||
@ -586,6 +591,76 @@ describe('BomCostingProcessingService', () => {
|
||||
expect(bomCostingRepository.failSourceSnapshot).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('旧版资料的图片已更新时自动重新归档,再继续核算', async () => {
|
||||
const legacySnapshot: MarketDetailSnapshot = {
|
||||
...snapshot,
|
||||
skus: Array.from({ length: 500 }, (_, index) => ({
|
||||
...snapshot.skus[0],
|
||||
externalSkuId: `legacy-sku-${index + 1}`,
|
||||
name: `旧规格 ${index + 1}`,
|
||||
})),
|
||||
requestId: 'legacy-request',
|
||||
}
|
||||
const refreshedSnapshot: MarketDetailSnapshot = {
|
||||
...snapshot,
|
||||
skuTotalRows: 1,
|
||||
skuReturnedRows: 1,
|
||||
skuTruncated: false,
|
||||
mainImages: ['https://storage.example.com/latest-main.jpg'],
|
||||
requestId: 'refreshed-request',
|
||||
}
|
||||
const rearchivedSource = {
|
||||
...sourceSnapshot,
|
||||
rawSnapshot: refreshedSnapshot,
|
||||
storedImageCount: 2,
|
||||
}
|
||||
jest.spyOn(bomCostingRepository, 'findCaseForProcessing').mockResolvedValue({
|
||||
...caseRecord(),
|
||||
sourceSnapshot: {
|
||||
...sourceSnapshot,
|
||||
skuCount: 500,
|
||||
rawSnapshot: legacySnapshot,
|
||||
},
|
||||
} as never)
|
||||
jest
|
||||
.spyOn(bomCostingRepository, 'claimSourceSnapshotMetadataRefresh')
|
||||
.mockResolvedValue(true)
|
||||
jest.spyOn(bomCostingRepository, 'updateCaseTitle').mockResolvedValue({} as never)
|
||||
jest
|
||||
.spyOn(marketDetailClient, 'readProduct')
|
||||
.mockResolvedValue(refreshedSnapshot)
|
||||
mockRefreshSourceMetadata.mockRejectedValue(
|
||||
new mockSourceImagesChangedError('竞品图片集合已经变化'),
|
||||
)
|
||||
mockArchive.mockResolvedValue(rearchivedSource)
|
||||
mockRunAnalysis.mockRejectedValue(new Error('停止在重新归档后的测试边界'))
|
||||
const { BomCostingProcessingService } = await import(
|
||||
'./bom-costing-processing.service'
|
||||
)
|
||||
|
||||
await new BomCostingProcessingService().processCase('case-1')
|
||||
|
||||
expect(mockArchive).toHaveBeenCalledWith(expect.objectContaining({
|
||||
sourceSnapshotId: 'source-1',
|
||||
snapshot: refreshedSnapshot,
|
||||
onProgress: expect.any(Function),
|
||||
}))
|
||||
expect(bomCostingRepository.updateCaseStage).toHaveBeenCalledWith(
|
||||
'case-1',
|
||||
expect.objectContaining({
|
||||
eventType: 'SOURCE_REARCHIVE_STARTED',
|
||||
message: '竞品图片已更新,正在重新归档最新资料',
|
||||
}),
|
||||
)
|
||||
expect(bomCostingRepository.updateCaseStage).toHaveBeenCalledWith(
|
||||
'case-1',
|
||||
expect.objectContaining({
|
||||
eventType: 'SOURCE_REARCHIVE_COMPLETED',
|
||||
}),
|
||||
)
|
||||
expect(bomCostingRepository.failSourceSnapshot).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('旧版元数据刷新失败时恢复归档为可复用状态', async () => {
|
||||
const legacySnapshot: MarketDetailSnapshot = {
|
||||
...snapshot,
|
||||
|
||||
@ -24,6 +24,8 @@ import { bomCostingRepository } from '@/server/repositories/bom-costing.reposito
|
||||
import {
|
||||
buildMarketSnapshotFromStoredSource,
|
||||
bomCostingSourceArchiveService,
|
||||
BomCostingSourceImagesChangedError,
|
||||
type BomCostingArchiveProgress,
|
||||
} from './bom-costing-source-archive.service'
|
||||
import { prepareBomCostingAnalysis } from './bom-costing-result-preparer'
|
||||
import { getCompetitorAnalysisProvider } from './bom-costing-provider'
|
||||
@ -241,6 +243,42 @@ function buildAnalysisImagePlan(
|
||||
})
|
||||
}
|
||||
|
||||
async function persistArchiveProgress(
|
||||
caseId: string,
|
||||
progress: BomCostingArchiveProgress,
|
||||
) {
|
||||
const settled = progress.storedCount + progress.failedCount
|
||||
const caseProgress =
|
||||
20 +
|
||||
Math.round((settled / Math.max(progress.totalCount, 1)) * 24)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType:
|
||||
progress.phase === 'STARTED'
|
||||
? 'ASSET_ARCHIVE_ITEM_STARTED'
|
||||
: progress.status === BomCostingSourceAssetStatus.STORED
|
||||
? 'ASSET_STORED'
|
||||
: 'ASSET_ARCHIVE_FAILED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: caseProgress,
|
||||
message:
|
||||
progress.phase === 'STARTED'
|
||||
? `正在归档第 ${progress.sortOrder + 1} 张图片`
|
||||
: `图片归档 ${settled}/${progress.totalCount}`,
|
||||
payload: {
|
||||
phase: progress.phase,
|
||||
assetId: progress.assetId,
|
||||
kind: progress.kind,
|
||||
assetStatus: progress.status,
|
||||
storedUrl: progress.storedUrl,
|
||||
sortOrder: progress.sortOrder,
|
||||
externalSkuIds: progress.externalSkuIds,
|
||||
storedCount: progress.storedCount,
|
||||
failedCount: progress.failedCount,
|
||||
totalCount: progress.totalCount,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
export class BomCostingProcessingService {
|
||||
async processCase(caseId: string) {
|
||||
const claimed = await bomCostingRepository.claimCaseForProcessing(caseId)
|
||||
@ -301,25 +339,74 @@ export class BomCostingProcessingService {
|
||||
},
|
||||
})
|
||||
const fetched = await marketDetailClient.readProduct(record.sourceUrl)
|
||||
source = await bomCostingSourceArchiveService.refreshMetadata({
|
||||
source,
|
||||
snapshot: fetched,
|
||||
})
|
||||
sourceMetadataRefreshOwnedByTask = false
|
||||
await bomCostingRepository.updateCaseTitle(caseId, fetched.title)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'SOURCE_METADATA_REFRESH_COMPLETED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: 18,
|
||||
message: `已补全 ${fetched.skuReturnedRows ?? fetched.skus.length}/${fetched.skuTotalRows ?? fetched.skus.length} 条平台规格元数据,沿用 ${source.storedImageCount} 张归档图片`,
|
||||
payload: {
|
||||
try {
|
||||
source = await bomCostingSourceArchiveService.refreshMetadata({
|
||||
source,
|
||||
snapshot: fetched,
|
||||
})
|
||||
sourceMetadataRefreshOwnedByTask = false
|
||||
await bomCostingRepository.updateCaseTitle(caseId, fetched.title)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'SOURCE_METADATA_REFRESH_COMPLETED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: 18,
|
||||
message: `已补全 ${fetched.skuReturnedRows ?? fetched.skus.length}/${fetched.skuTotalRows ?? fetched.skus.length} 条平台规格元数据,沿用 ${source.storedImageCount} 张归档图片`,
|
||||
payload: {
|
||||
sourceSnapshotId: source.id,
|
||||
skuTotalRows: fetched.skuTotalRows ?? fetched.skus.length,
|
||||
skuReturnedRows:
|
||||
fetched.skuReturnedRows ?? fetched.skus.length,
|
||||
archivedImageCount: source.storedImageCount,
|
||||
},
|
||||
})
|
||||
} catch (error) {
|
||||
if (!(error instanceof BomCostingSourceImagesChangedError)) {
|
||||
throw error
|
||||
}
|
||||
|
||||
sourceMetadataRefreshOwnedByTask = false
|
||||
sourceOwnedByTask = true
|
||||
await bomCostingRepository.updateCaseTitle(caseId, fetched.title)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'SOURCE_REARCHIVE_STARTED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: 18,
|
||||
message: '竞品图片已更新,正在重新归档最新资料',
|
||||
payload: {
|
||||
sourceSnapshotId: source.id,
|
||||
previousImageCount: source.storedImageCount,
|
||||
latestImageCount:
|
||||
fetched.mainImages.length +
|
||||
fetched.skuImages.length +
|
||||
fetched.detailImages.length,
|
||||
},
|
||||
})
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'ASSET_ARCHIVE_STARTED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: 20,
|
||||
message: '正在将最新竞品图片归档到对象存储',
|
||||
})
|
||||
source = await bomCostingSourceArchiveService.archive({
|
||||
sourceSnapshotId: source.id,
|
||||
skuTotalRows: fetched.skuTotalRows ?? fetched.skus.length,
|
||||
skuReturnedRows:
|
||||
fetched.skuReturnedRows ?? fetched.skus.length,
|
||||
archivedImageCount: source.storedImageCount,
|
||||
},
|
||||
})
|
||||
snapshot: fetched,
|
||||
onProgress: (progress) =>
|
||||
persistArchiveProgress(caseId, progress),
|
||||
})
|
||||
sourceOwnedByTask = false
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'SOURCE_REARCHIVE_COMPLETED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: 46,
|
||||
message: `已更新并归档 ${source.storedImageCount} 张最新竞品图片`,
|
||||
payload: {
|
||||
sourceSnapshotId: source.id,
|
||||
storedImageCount: source.storedImageCount,
|
||||
failedImageCount: source.failedImageCount,
|
||||
skuCount: source.skuCount,
|
||||
},
|
||||
})
|
||||
}
|
||||
} else {
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'SOURCE_METADATA_REFRESH_WAITING',
|
||||
@ -385,40 +472,8 @@ export class BomCostingProcessingService {
|
||||
source = await bomCostingSourceArchiveService.archive({
|
||||
sourceSnapshotId: source.id,
|
||||
snapshot: fetched,
|
||||
onProgress: async (progress) => {
|
||||
const settled = progress.storedCount + progress.failedCount
|
||||
const caseProgress =
|
||||
20 +
|
||||
Math.round(
|
||||
(settled / Math.max(progress.totalCount, 1)) * 24,
|
||||
)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType:
|
||||
progress.phase === 'STARTED'
|
||||
? 'ASSET_ARCHIVE_ITEM_STARTED'
|
||||
: progress.status === BomCostingSourceAssetStatus.STORED
|
||||
? 'ASSET_STORED'
|
||||
: 'ASSET_ARCHIVE_FAILED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: caseProgress,
|
||||
message:
|
||||
progress.phase === 'STARTED'
|
||||
? `正在归档第 ${progress.sortOrder + 1} 张图片`
|
||||
: `图片归档 ${settled}/${progress.totalCount}`,
|
||||
payload: {
|
||||
phase: progress.phase,
|
||||
assetId: progress.assetId,
|
||||
kind: progress.kind,
|
||||
assetStatus: progress.status,
|
||||
storedUrl: progress.storedUrl,
|
||||
sortOrder: progress.sortOrder,
|
||||
externalSkuIds: progress.externalSkuIds,
|
||||
storedCount: progress.storedCount,
|
||||
failedCount: progress.failedCount,
|
||||
totalCount: progress.totalCount,
|
||||
},
|
||||
})
|
||||
},
|
||||
onProgress: (progress) =>
|
||||
persistArchiveProgress(caseId, progress),
|
||||
})
|
||||
sourceOwnedByTask = false
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
@ -823,6 +878,41 @@ export class BomCostingProcessingService {
|
||||
message: '正在合并逐图事实并生成各 SKU 的结构化 BOM',
|
||||
})
|
||||
},
|
||||
onSynthesisBatchStarted: async (progress) => {
|
||||
const caseProgress =
|
||||
78 +
|
||||
Math.floor(
|
||||
(progress.completedSkuCount /
|
||||
Math.max(analysisInputStats.inputSkuCount, 1)) *
|
||||
7,
|
||||
)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'MODEL_SYNTHESIS_BATCH_STARTED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: caseProgress,
|
||||
message:
|
||||
`正在汇总第 ${progress.batchIndex}/${progress.totalBatches} 批规格结构` +
|
||||
`(${progress.batchSkuCount} 个 SKU)`,
|
||||
payload: progress,
|
||||
})
|
||||
},
|
||||
onSynthesisBatchCompleted: async (progress) => {
|
||||
const caseProgress =
|
||||
78 +
|
||||
Math.round(
|
||||
(progress.completedSkuCount /
|
||||
Math.max(analysisInputStats.inputSkuCount, 1)) *
|
||||
7,
|
||||
)
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'MODEL_SYNTHESIS_BATCH_COMPLETED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: caseProgress,
|
||||
message:
|
||||
`已完成 ${progress.completedSkuCount}/${analysisInputStats.inputSkuCount} 个 SKU 的结构汇总`,
|
||||
payload: progress,
|
||||
})
|
||||
},
|
||||
})
|
||||
const persistedResult = archivedSource
|
||||
? replaceAnalysisImageUrlsWithStableArchive(output.result, archivedSource)
|
||||
@ -830,7 +920,7 @@ export class BomCostingProcessingService {
|
||||
await bomCostingRepository.updateCaseStage(caseId, {
|
||||
eventType: 'MODEL_ANALYSIS_COMPLETED',
|
||||
status: BomCostingCaseStatus.ANALYZING,
|
||||
progress: 80,
|
||||
progress: 86,
|
||||
message: `结构分析完成,识别到 ${persistedResult.skus.length} 个 SKU`,
|
||||
payload: {
|
||||
...output.trace,
|
||||
|
||||
@ -17,6 +17,7 @@ import {
|
||||
isBomCostingMultimodalConfigured,
|
||||
type BomCostingMultimodalInputStats,
|
||||
type BomCostingMultimodalOutput,
|
||||
type BomCostingSynthesisBatchProgress,
|
||||
} from './bom-costing-multimodal-analyzer'
|
||||
import type {
|
||||
BomCostingFactBatchProgress,
|
||||
@ -46,6 +47,15 @@ export interface CompetitorAnalysisProvider {
|
||||
snapshot: BomCostingSourceFactSnapshot
|
||||
trace: BomCostingFactExtractionTrace
|
||||
}) => Promise<void>
|
||||
onSynthesisBatchStarted?: (
|
||||
progress: Pick<
|
||||
BomCostingSynthesisBatchProgress,
|
||||
'batchIndex' | 'totalBatches' | 'batchSkuCount' | 'completedSkuCount'
|
||||
>,
|
||||
) => Promise<void>
|
||||
onSynthesisBatchCompleted?: (
|
||||
progress: BomCostingSynthesisBatchProgress,
|
||||
) => Promise<void>
|
||||
}): Promise<BomCostingMultimodalOutput>
|
||||
}
|
||||
|
||||
@ -278,6 +288,8 @@ class FixtureCompetitorAnalysisProvider implements CompetitorAnalysisProvider {
|
||||
imageCount: 0,
|
||||
inputSkuCount: result.skus.length,
|
||||
factBatchCount: 0,
|
||||
synthesisBatchCount: 0,
|
||||
synthesisCallIds: [],
|
||||
reusedImageCount: 0,
|
||||
analyzedImageCount: 0,
|
||||
durationMs: 0,
|
||||
@ -324,6 +336,15 @@ class MarketDetailMultimodalProvider implements CompetitorAnalysisProvider {
|
||||
snapshot: BomCostingSourceFactSnapshot
|
||||
trace: BomCostingFactExtractionTrace
|
||||
}) => Promise<void>
|
||||
onSynthesisBatchStarted?: (
|
||||
progress: Pick<
|
||||
BomCostingSynthesisBatchProgress,
|
||||
'batchIndex' | 'totalBatches' | 'batchSkuCount' | 'completedSkuCount'
|
||||
>,
|
||||
) => Promise<void>
|
||||
onSynthesisBatchCompleted?: (
|
||||
progress: BomCostingSynthesisBatchProgress,
|
||||
) => Promise<void>
|
||||
}) {
|
||||
if (!input.snapshot) {
|
||||
throw new Error('竞品资料快照不存在,不能开始多模态分析')
|
||||
|
||||
@ -24,6 +24,13 @@ type SourceWithAssets = BomCostingSourceSnapshot & {
|
||||
assets: BomCostingSourceAsset[]
|
||||
}
|
||||
|
||||
export class BomCostingSourceImagesChangedError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message)
|
||||
this.name = 'BomCostingSourceImagesChangedError'
|
||||
}
|
||||
}
|
||||
|
||||
export interface BomCostingArchiveProgress {
|
||||
phase: 'STARTED' | 'SETTLED'
|
||||
assetId: string
|
||||
@ -254,7 +261,7 @@ export class BomCostingSourceArchiveService {
|
||||
seeds.length !== input.source.assets.length ||
|
||||
seeds.some((seed) => !existingByOriginalUrl.has(seed.originalUrl))
|
||||
) {
|
||||
throw new Error(
|
||||
throw new BomCostingSourceImagesChangedError(
|
||||
'竞品图片集合已经变化,不能只刷新 SKU 元数据;请重新归档竞品资料',
|
||||
)
|
||||
}
|
||||
@ -266,7 +273,7 @@ export class BomCostingSourceArchiveService {
|
||||
existing.status !== BomCostingSourceAssetStatus.STORED ||
|
||||
!existing.storageKey
|
||||
) {
|
||||
throw new Error(
|
||||
throw new BomCostingSourceImagesChangedError(
|
||||
'旧竞品图片没有完整归档,不能只刷新 SKU 元数据;请重新归档竞品资料',
|
||||
)
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user