diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java index b07e70c6..4a50336c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java @@ -23,6 +23,14 @@ public class BrandCheckProperties { * 10 次尝试合计约 45s，既能覆盖限流窗口又不会让单个品牌长时间占住查询线程。 */ private int retryMaxIntervalMillis = 10000; + /** + * 一次批量检查的总耗时上限（毫秒）：整批用尽后不再重试，剩余品牌按「查询失败」收尾。 + * 上限的意义不是省时间，而是给「分片回传」这类同步调用方一个时延上界——上游卡死时 + * 单品牌 10 次重试曾把一次回传拖到 103.5 秒（taskId 28599），客户端重试预算耗尽后 + * 中止了整个采集。首轮请求始终执行，故上游只是略慢时不会误降级。 + * 设为 0 或负数表示不限制。 + */ + private int totalTimeoutMillis = 90000; private int connectTimeoutMillis = 10000; private int readTimeoutMillis = 60000; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java index d6c05fdb..3bb9cafa 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java @@ -95,10 +95,14 @@ public class BrandCheckClient { public BrandCheckBatchResult checkAll(List brands, String strategy) { List distinctBrands = distinctNonBlank(brands); + // 整批共用一个耗时预算：上游 16890 卡死时，单品牌 10 次重试曾把一次分片回传拖到 + // 103.5 秒（taskId 28599），客户端重试预算耗尽后中止了整个采集。预算用尽即停止重试。 + long budgetMillis = properties.getTotalTimeoutMillis(); + long deadlineNanos = budgetMillis > 0L ? System.nanoTime() + budgetMillis * 1_000_000L : Long.MAX_VALUE; List> futures = new ArrayList<>(distinctBrands.size()); for (String brand : distinctBrands) { futures.add(CompletableFuture.supplyAsync( - () -> checkOneBrand(brand, strategy), checkExecutor)); + () -> checkOneBrand(brand, strategy, deadlineNanos), checkExecutor)); } List failedData = new ArrayList<>(); List queryFailedData = new ArrayList<>(); @@ -110,11 +114,19 @@ public class BrandCheckClient { return new BrandCheckBatchResult(distinctBrands, failedData, queryFailedData); } - private BrandCheckOutcome checkOneBrand(String brand, String strategy) { + private BrandCheckOutcome checkOneBrand(String brand, String strategy, long deadlineNanos) { int attempts = Math.max(1, properties.getRetryTimes()); BrandCheckResponse response = null; Exception lastFailure = null; for (int attempt = 1; attempt <= attempts; attempt++) { + // 预算用尽就不再重试，按查询失败收尾。只掐「重试」不打断已发出的请求， + // 故最坏耗时 ≈ 预算 + 一次请求的读超时；首轮始终执行，避免上游只是慢一点时被误降级。 + if (attempt > 1 && System.nanoTime() >= deadlineNanos) { + log.warn("[brand-check] 整批耗时预算用尽，停止重试 brand={} attempt={}/{} lastErr={}", + brand, attempt, attempts, + lastFailure == null ? "query_faild_data 持续非空" : lastFailure.getMessage()); + break; + } try { response = check(brand, strategy); } catch (Exception ex) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 63753b68..7bfe6cf9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -782,6 +782,39 @@ public class CollectDataService { throw new BusinessException("request is empty"); } ensureRustfsPayloadStorageEnabled(); + + // 锁外预检：任务不存在/已结束时立即失败，不为终态任务白跑去重查询与品牌检测。 + // 只做快速失败，并发正确性仍由锁内的重读复核保证。 + FileTaskEntity probe = fileTaskMapper.selectById(taskId); + if (probe == null || !MODULE_TYPE.equals(probe.getModuleType())) { + throw new BusinessException("任务不存在"); + } + if (STATUS_SUCCESS.equals(probe.getStatus()) || STATUS_FAILED.equals(probe.getStatus())) { + log.warn("[collect-data] 任务已结束，拒绝重复提交 taskId={} status={}", taskId, probe.getStatus()); + throw new BusinessException(40901, "任务已结束，拒绝重复提交"); + } + + // 归一化 / 去重过滤 / 品牌检测放在锁外：品牌检测是同步远程调用，上游 16890 抖动时 + // 单品牌 10 次重试合计上百秒（taskId 28599 实测：chunk 回传在锁内等品牌检测 103.5 秒， + // 期间心跳与客户端重试全部撞 40902 拿不到锁，客户端 5 次重试预算耗尽后中止整个采集）。 + // 这几步只依赖本批入参、不写任务状态，放锁外不改变 chunk 落库的串行语义。 + List rows = normalizeSubmitRows(request.getItems()); + buildParseLimits().validateChunkRowCount(rows.size()); + // 「结果文件」sheet 需要 Python 回传的全量数据（不经任何后端过滤）， + // 因此 chunk payload 写入未筛除空 ASIN 的全量 rows； + // 仅当 ASIN 非空时才进入去重 / 无效品牌 / 品牌检测链路并落 biz_task_result_item。 + List rowsForFiltering = new ArrayList<>(rows.size()); + for (CollectDataResultRowVo row : rows) { + if (row.getAsin() != null && !row.getAsin().isBlank()) { + rowsForFiltering.add(row); + } + } + long prepareStartAt = System.currentTimeMillis(); + CollectDataBatchQuery.FilterResult filtered = collectDataBatchQuery.filter(rowsForFiltering); + CollectDataBrandBatchFilter.BrandBatchOutcome brandOutcome = brandBatchFilter.filter(filtered.kept()); + log.info("[collect-data] 锁外预处理完成 taskId={} rows={} 去重后={} 品牌检测耗时={}ms", + taskId, rows.size(), filtered.kept().size(), System.currentTimeMillis() - prepareStartAt); + try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) { FileTaskEntity task = fileTaskMapper.selectById(taskId); if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { @@ -819,25 +852,23 @@ public class CollectDataService { return buildSubmitVo(task, result, chunkIndex, chunkTotal, request, 0); } - List rows = normalizeSubmitRows(request.getItems()); - buildParseLimits().validateChunkRowCount(rows.size()); CollectDataStats stats = loadStats(task); stats.receivedRows += rows.size(); stats.currentChunkRows = rows.size(); - - // 「结果文件」sheet 需要 Python 回传的全量数据（不经任何后端过滤）， - // 因此 chunk payload 写入未筛除空 ASIN 的全量 rows； - // 仅当 ASIN 非空时才进入去重 / 无效品牌 / 品牌检测链路并落 biz_task_result_item。 - List rowsForFiltering = new ArrayList<>(rows.size()); - for (CollectDataResultRowVo row : rows) { - if (row.getAsin() != null && !row.getAsin().isBlank()) { - rowsForFiltering.add(row); - } - } - CollectDataBatchQuery.FilterResult filtered = collectDataBatchQuery.filter(rowsForFiltering); stats.dedupeFilteredCount += filtered.dedupeFilteredCount(); stats.invalidFilteredCount += filtered.invalidFilteredCount(); - List accepted = filterByBrandCheck(filtered.kept(), stats); + stats.brandRejectedCount += brandOutcome.rejected().size(); + stats.brandQueryFailedCount += brandOutcome.queryFailed().size(); + // 只有被服务端判定为无效品牌（rejected）的行才写入无效品牌表； + // queryFailed 是品牌检测服务端到端失败（如缺 X-Token 422/超时/限流）， + // 一并写入会把故障期间被误伤的品牌永久拉黑，后续同品牌商品全部被 + // invalidFiltered 过滤（task-27265 实测：通过 Python 过滤的 21 行中 + // 8 行查询失败被写表后误杀，最终结果 Excel 0 行）。rejected 为空时 + // 不触发任何写入，避免空批次无意义调用。 + if (!brandOutcome.rejected().isEmpty()) { + invalidAsinBatchWriter.writeBatch(brandOutcome.rejected()); + } + List accepted = brandOutcome.accepted(); // 结果明细改为 chunk 级存储：整个 chunk 的 accepted 行共享一个 // RustFS 对象（deterministic key，同 chunk 重提覆盖同一对象）， // biz_task_result_item.payload_json 只存 {chunk, offset, payload} 引用。 @@ -872,6 +903,14 @@ public class CollectDataService { if (request.getError() != null && !request.getError().isBlank()) { markTaskFailed(task, result, request.getError(), stats); + // 失败但已收到分片：照常组装结果文件，让用户能下载已采集的数据。 + // 此前失败分支只标失败不组装，已落库的数据也没有任何结果文件可下载 + // （taskId 28599：55 个分片全部收到、187 行明细已落库，用户却拿不到文件）。 + if (hasReceivedChunks(taskId)) { + enqueueFinalWorkbook(task, result, stats); + log.warn("[collect-data] 任务失败仍组装部分结果 taskId={} error={} finalRows={}", + taskId, request.getError(), stats.finalRowCount); + } } else if (Boolean.TRUE.equals(request.getDone())) { enqueueFinalWorkbook(task, result, stats); } else { @@ -933,22 +972,6 @@ public class CollectDataService { return rows; } - private List filterByBrandCheck(List rows, CollectDataStats stats) { - CollectDataBrandBatchFilter.BrandBatchOutcome outcome = brandBatchFilter.filter(rows); - stats.brandRejectedCount += outcome.rejected().size(); - stats.brandQueryFailedCount += outcome.queryFailed().size(); - // 只有被服务端判定为无效品牌（rejected）的行才写入无效品牌表； - // queryFailed 是品牌检测服务端到端失败（如缺 X-Token 422/超时/限流）， - // 一并写入会把故障期间被误伤的品牌永久拉黑，后续同品牌商品全部被 - // invalidFiltered 过滤（task-27265 实测：通过 Python 过滤的 21 行中 - // 8 行查询失败被写表后误杀，最终结果 Excel 0 行）。rejected 为空时 - // 不触发任何写入，避免空批次无意义调用。 - if (!outcome.rejected().isEmpty()) { - invalidAsinBatchWriter.writeBatch(outcome.rejected()); - } - return outcome.accepted(); - } - private void persistChunk(Long taskId, String scopeKey, String scopeHash, @@ -1011,7 +1034,8 @@ public class CollectDataService { result.setResultFileSize(0L); result.setResultContentType(CONTENT_TYPE_XLSX); result.setRowCount(stats.finalRowCount); - result.setErrorMessage(null); + // 不清 errorMessage：失败任务的部分结果组装也走这里，清掉会让用户看不到真实失败原因 + // （成功路径的 errorMessage 本来就为 null，无需清理）。 fileResultMapper.updateById(result); task.setUpdatedAt(LocalDateTime.now()); fileTaskMapper.updateById(task); @@ -1060,6 +1084,22 @@ public class CollectDataService { stats.summaries, batch -> streamRawRows(task.getId(), batch)); String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE); + + // 复用同一份 stats 更新 finalRowCount 后再持久化，避免重复 loadStats 丢失 summaries。 + stats.finalRowCount = (int) finalRowCount; + persistStats(task, stats); + + // 失败原因先留存：下面的乐观写入会清空 result.errorMessage，任务已被判失败时要用它恢复。 + String failureReason = result.getErrorMessage(); + if (failureReason == null || failureReason.isBlank()) { + failureReason = task.getErrorMessage(); + } + if (failureReason == null || failureReason.isBlank()) { + failureReason = "任务失败，结果文件为已采集的部分数据"; + } + + // 结果行先按成功乐观写入：保持「结果文件先于任务成功落库」的时序， + // 万一进程在这两步之间退出，任务仍是 RUNNING，会被陈旧巡检重新组装（可自愈）。 result.setResultFilename(filename); result.setResultFileUrl(objectKey); result.setResultFileSize(xlsx.length()); @@ -1069,10 +1109,7 @@ public class CollectDataService { result.setErrorMessage(null); fileResultMapper.updateById(result); - // 复用同一份 stats 更新 finalRowCount 后再持久化，避免重复 loadStats 丢失 summaries。 - stats.finalRowCount = (int) finalRowCount; - persistStats(task, stats); - // 条件更新：任务可能已被 /fail 标为 FAILED（客户端报错与结果文件组装并发）—— + // 条件更新：任务可能已被判失败（客户端上报失败 / 陈旧判死与结果文件组装并发）—— // 无条件 updateById 会把 FAILED 覆盖成 SUCCESS（错误信息被清、用户看到「假成功」）。 // 结果文件已生成，故失败态下仍保留文件，只是不覆盖状态。 int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() @@ -1085,7 +1122,13 @@ public class CollectDataService { .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()) .set(FileTaskEntity::getFinishedAt, LocalDateTime.now())); if (updated == 0) { - log.warn("[collect-data] 结果文件已生成但任务已是 FAILED，保留失败态不覆盖 taskId={}", task.getId()); + // 任务已是 FAILED：结果记录改回失败语义并保留真实原因，但文件 URL 照常保留， + // 用户看到「失败 + 原因」的同时仍能下载已采集的部分结果。 + result.setSuccess(0); + result.setErrorMessage(failureReason); + fileResultMapper.updateById(result); + log.warn("[collect-data] 结果文件已生成但任务已是 FAILED，保留失败态与原因 taskId={} rows={} reason={}", + task.getId(), finalRowCount, failureReason); } } finally { FileUtil.del(xlsx); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java index 68d7a3c2..1fd1aea6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java @@ -652,9 +652,8 @@ public class ProductRiskTaskService { } matchedShopCount++; if (payload.getError() != null && !payload.getError().isBlank()) { - markResultFailed(fr, payload.getError()); - batchErrors.add(shopKey + ": " + payload.getError()); - productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + finalizeFailedShop(fr, shopKey, mergeShopPayload(taskId, shopKey, payload), + payload.getError(), batchErrors); continue; } @@ -759,9 +758,8 @@ public class ProductRiskTaskService { } if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) { - markResultFailed(fr, cachedPayload.getError()); - batchErrors.add(shopKey + ": " + cachedPayload.getError()); - productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + // 陈旧收尾同样保留已跑出来的行：失败原因照常回显，文件顺手组装 + finalizeFailedShop(fr, shopKey, cachedPayload, cachedPayload.getError(), batchErrors); changed = true; continue; } @@ -852,6 +850,9 @@ public class ProductRiskTaskService { String stem = safeFileStem(displayName); File xlsx = FileUtil.file(workRoot, stem + ".xlsx"); File zip = FileUtil.file(workRoot, stem + ".zip"); + // 失败店铺的部分结果（errorMessage 已写明原因）：组装完成时保留失败态，只挂文件。 + // 否则组装一落地就把行"洗成成功"，用户再也看不到「这个店铺其实没跑完」。 + boolean partialFailure = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank(); try { excelAssemblyService.writeWorkbook(xlsx, displayName, countries); ZipUtil.zip(zip, false, xlsx); @@ -861,8 +862,12 @@ public class ProductRiskTaskService { fr.setResultFileSize(zip.length()); fr.setResultContentType(CONTENT_TYPE_ZIP); fr.setRowCount(excelAssemblyService.countRows(countries)); - fr.setSuccess(1); - fr.setErrorMessage(null); + if (partialFailure) { + fr.setSuccess(0); + } else { + fr.setSuccess(1); + fr.setErrorMessage(null); + } fileResultMapper.updateById(fr); } finally { FileUtil.del(xlsx); @@ -899,13 +904,56 @@ public class ProductRiskTaskService { } private void enqueueResultFileAssembly(FileResultEntity result, String shopKey, ProductRiskShopPayloadDto payload) { + enqueueResultFileAssembly(result, shopKey, payload, false); + } + + /** + * @param preserveFailure 失败店铺的部分结果：保留 success=0 + 失败原因，只把文件挂上去， + * 任务状态不变（仍 FAILED），用户仍能下载已跑出来的行 + */ + private void enqueueResultFileAssembly(FileResultEntity result, String shopKey, + ProductRiskShopPayloadDto payload, boolean preserveFailure) { taskResultPayloadService.saveLatest(result.getTaskId(), MODULE_TYPE, shopKey, payload); taskResultItemService.replaceResultSnapshot(result.getTaskId(), MODULE_TYPE, result.getId(), shopKey, payload); - markResultFilePending(result, shopKey, payload); + markResultFilePending(result, shopKey, payload, preserveFailure); taskFileJobService.enqueueAssembleResult(result.getTaskId(), MODULE_TYPE, result.getId(), shopKey); } - private void markResultFilePending(FileResultEntity result, String shopKey, ProductRiskShopPayloadDto payload) { + /** + * 失败店铺的收尾：任务照旧标 FAILED（原因回显给用户），但已经把跑出来的行组装成 + * 部分结果文件随任务交付——否则用户只看到「失败」，却拿不到「哪些 ASIN 实际已被处理」 + * 的记录（会话掉线这类"跑了几个国家才断"的场景尤其需要）。一行可用数据都没有时才退化成纯失败。 + */ + private void finalizeFailedShop(FileResultEntity fr, String shopKey, + ProductRiskShopPayloadDto mergedPayload, String errorMessage, + List batchErrors) { + int rows = mergedPayload == null ? 0 : countPayloadRows(mergedPayload); + String message = errorMessage; + if (rows > 0) { + markResultFailed(fr, errorMessage); + try { + enqueueResultFileAssembly(fr, shopKey, mergedPayload, true); + batchErrors.add(shopKey + ": " + errorMessage); + productRiskTaskCacheService.removeShopMergedPayload(fr.getTaskId(), shopKey); + log.info("[product-risk] 失败店铺仍产出部分结果 taskId={} shop={} rows={} error={}", + fr.getTaskId(), shopKey, rows, errorMessage); + return; + } catch (Exception ex) { + message = errorMessage + "（部分结果组装排队失败：" + (ex.getMessage() == null ? "未知错误" : ex.getMessage()) + "）"; + log.warn("[product-risk] 失败店铺部分结果排队失败，仅标记失败 taskId={} shop={} msg={}", + fr.getTaskId(), shopKey, ex.getMessage(), ex); + } + } else { + log.info("[product-risk] 失败店铺无可用行，不产出结果文件 taskId={} shop={} error={}", + fr.getTaskId(), shopKey, errorMessage); + } + markResultFailed(fr, message); + batchErrors.add(shopKey + ": " + message); + productRiskTaskCacheService.removeShopMergedPayload(fr.getTaskId(), shopKey); + } + + private void markResultFilePending(FileResultEntity result, String shopKey, + ProductRiskShopPayloadDto payload, boolean preserveFailure) { Map> countries = excelAssemblyService.normalizeCountriesMap(payload.getCountries()); String displayName = payload.getShopName() != null && !payload.getShopName().isBlank() ? payload.getShopName().trim() @@ -916,8 +964,16 @@ public class ProductRiskTaskService { result.setResultFileSize(0L); result.setResultContentType(CONTENT_TYPE_ZIP); result.setRowCount(excelAssemblyService.countRows(countries)); - result.setSuccess(1); - result.setErrorMessage(null); + if (preserveFailure) { + // 失败店铺的部分结果：成功态与失败原因都不能动，只标「文件名已定、文件待组装」 + result.setSuccess(0); + if (result.getErrorMessage() == null || result.getErrorMessage().isBlank()) { + result.setErrorMessage("店铺未跑完，仅产出部分结果"); + } + } else { + result.setSuccess(1); + result.setErrorMessage(null); + } fileResultMapper.updateById(result); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java index 5c9c2b34..d210d1bc 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java @@ -676,8 +676,7 @@ public class ShopMatchTaskService { changed = true; ShopMatchShopPayloadDto merged = mergeShopPayload(taskId, shopKey, incoming); if (incoming.getError() != null && !incoming.getError().isBlank()) { - markResultFailed(result, incoming.getError().trim()); - shopMatchTaskCacheService.removeShopMergedPayload(taskId, shopKey); + finalizeFailedShop(result, shopKey, merged, incoming.getError().trim()); continue; } if (!isShopPayloadCompleted(merged)) { @@ -756,9 +755,8 @@ public class ShopMatchTaskService { } if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) { - markResultFailed(result, cachedPayload.getError()); - batchErrors.add(shopKey + ": " + cachedPayload.getError()); - shopMatchTaskCacheService.removeShopMergedPayload(taskId, shopKey); + // 陈旧收尾同样保留已跑出来的行：失败原因照常回显，文件顺手组装 + batchErrors.add(shopKey + ": " + finalizeFailedShop(result, shopKey, cachedPayload, cachedPayload.getError())); changed = true; continue; } @@ -819,6 +817,9 @@ public class ShopMatchTaskService { String displayName = payload.getShopName() != null && !payload.getShopName().isBlank() ? payload.getShopName().trim() : shopKey; String stem = safeFileStem(displayName); File xlsx = FileUtil.file(workRoot, stem + ".xlsx"); + // 失败店铺的部分结果（errorMessage 已写明原因）：组装完成时保留失败态，只挂文件。 + // 否则组装一落地就把行"洗成成功"，用户再也看不到「这个店铺其实没跑完」。 + boolean partialFailure = result.getErrorMessage() != null && !result.getErrorMessage().isBlank(); try { excelAssemblyService.writeWorkbook(xlsx, countries); String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE); @@ -827,8 +828,12 @@ public class ShopMatchTaskService { result.setResultFileSize(xlsx.length()); result.setResultContentType(CONTENT_TYPE_XLSX); result.setRowCount(excelAssemblyService.countRows(countries)); - result.setSuccess(1); - result.setErrorMessage(null); + if (partialFailure) { + result.setSuccess(0); + } else { + result.setSuccess(1); + result.setErrorMessage(null); + } fileResultMapper.updateById(result); } finally { FileUtil.del(xlsx); @@ -861,12 +866,55 @@ public class ShopMatchTaskService { } private void enqueueResultFileAssembly(FileResultEntity result, String shopKey, ShopMatchShopPayloadDto payload) { + enqueueResultFileAssembly(result, shopKey, payload, false); + } + + /** + * @param preserveFailure 失败店铺的部分结果：保留 success=0 + 失败原因，只把文件挂上去， + * 任务状态不变（仍 FAILED），用户仍能下载已跑出来的行 + */ + private void enqueueResultFileAssembly(FileResultEntity result, String shopKey, + ShopMatchShopPayloadDto payload, boolean preserveFailure) { taskResultPayloadService.saveLatest(result.getTaskId(), MODULE_TYPE, shopKey, payload); - markResultFilePending(result, shopKey, payload); + markResultFilePending(result, shopKey, payload, preserveFailure); taskFileJobService.enqueueAssembleResult(result.getTaskId(), MODULE_TYPE, result.getId(), shopKey); } - private void markResultFilePending(FileResultEntity result, String shopKey, ShopMatchShopPayloadDto payload) { + /** + * 失败店铺的收尾：任务照旧标 FAILED（原因回显给用户），但已经把跑出来的行组装成 + * 部分结果文件随任务交付——否则用户只看到「失败」，却拿不到「哪些 ASIN 实际已被匹配」 + * 的记录（会话掉线这类"跑了几个国家才断"的场景尤其需要）。一行可用数据都没有时才退化成纯失败。 + * + * @return 最终写入结果记录的失败原因（组装排队失败时会附上原因） + */ + private String finalizeFailedShop(FileResultEntity result, String shopKey, + ShopMatchShopPayloadDto mergedPayload, String errorMessage) { + int rows = mergedPayload == null ? 0 : countPayloadRows(mergedPayload); + String message = errorMessage; + if (rows > 0) { + markResultFailed(result, errorMessage); + try { + enqueueResultFileAssembly(result, shopKey, mergedPayload, true); + shopMatchTaskCacheService.removeShopMergedPayload(result.getTaskId(), shopKey); + log.info("[shop-match] 失败店铺仍产出部分结果 taskId={} shop={} rows={} error={}", + result.getTaskId(), shopKey, rows, errorMessage); + return errorMessage; + } catch (Exception ex) { + message = errorMessage + "（部分结果组装排队失败：" + (ex.getMessage() == null ? "未知错误" : ex.getMessage()) + "）"; + log.warn("[shop-match] 失败店铺部分结果排队失败，仅标记失败 taskId={} shop={} msg={}", + result.getTaskId(), shopKey, ex.getMessage(), ex); + } + } else { + log.info("[shop-match] 失败店铺无可用行，不产出结果文件 taskId={} shop={} error={}", + result.getTaskId(), shopKey, errorMessage); + } + markResultFailed(result, message); + shopMatchTaskCacheService.removeShopMergedPayload(result.getTaskId(), shopKey); + return message; + } + + private void markResultFilePending(FileResultEntity result, String shopKey, + ShopMatchShopPayloadDto payload, boolean preserveFailure) { Map> countries = excelAssemblyService.normalizeCountriesMap(payload.getCountries()); String displayName = payload.getShopName() != null && !payload.getShopName().isBlank() ? payload.getShopName().trim() : shopKey; String stem = safeFileStem(displayName); @@ -875,8 +923,16 @@ public class ShopMatchTaskService { result.setResultFileSize(0L); result.setResultContentType(CONTENT_TYPE_XLSX); result.setRowCount(excelAssemblyService.countRows(countries)); - result.setSuccess(1); - result.setErrorMessage(null); + if (preserveFailure) { + // 失败店铺的部分结果：成功态与失败原因都不能动，只标「文件名已定、文件待组装」 + result.setSuccess(0); + if (result.getErrorMessage() == null || result.getErrorMessage().isBlank()) { + result.setErrorMessage("店铺未跑完，仅产出部分结果"); + } + } else { + result.setSuccess(1); + result.setErrorMessage(null); + } fileResultMapper.updateById(result); } diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index e6e0315b..ef9a2737 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -300,6 +300,7 @@ aiimage: retry-times: ${AIIMAGE_BRAND_CHECK_RETRY_TIMES:10} retry-interval-millis: ${AIIMAGE_BRAND_CHECK_RETRY_INTERVAL_MILLIS:1000} retry-max-interval-millis: ${AIIMAGE_BRAND_CHECK_RETRY_MAX_INTERVAL_MILLIS:10000} + total-timeout-millis: ${AIIMAGE_BRAND_CHECK_TOTAL_TIMEOUT_MILLIS:90000} connect-timeout-millis: ${AIIMAGE_BRAND_CHECK_CONNECT_TIMEOUT_MILLIS:10000} read-timeout-millis: ${AIIMAGE_BRAND_CHECK_READ_TIMEOUT_MILLIS:60000} appearance-patent: diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataServiceTxBoundaryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataServiceTxBoundaryTest.java index f83113d3..e89c4486 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataServiceTxBoundaryTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataServiceTxBoundaryTest.java @@ -264,6 +264,30 @@ class CollectDataServiceTxBoundaryTest { verify(transactionTemplate, never()).execute(any()); } + /** taskId 28599：失败时已收到的分片仍要组装成结果文件，让用户能下载已采集的数据。 */ + @Test + void failedSubmitWithReceivedChunksEnqueuesPartialWorkbook() { + when(taskChunkMapper.selectCount(any())).thenReturn(170L); + CollectDataSubmitResultRequest request = submitRequest(); + request.setError("中间分批回传失败，终止本次采集以避免服务端数据残缺"); + + service.submitResult(TASK_ID, request); + + verify(taskFileJobService).enqueueAssembleResult( + eq(TASK_ID), eq("COLLECT_DATA"), eq(5001L), anyString()); + } + + @Test + void failedSubmitWithoutReceivedChunksEnqueuesNothing() { + CollectDataSubmitResultRequest request = submitRequest(); + request.setError("采集端启动失败"); + + service.submitResult(TASK_ID, request); + + verify(taskFileJobService, never()) + .enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString()); + } + private CollectDataSubmitResultRequest submitRequest() { CollectDataSubmitRowDto row = new CollectDataSubmitRowDto(); row.setAsin("B0COLLECT1"); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/SuccessTimingContractTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/SuccessTimingContractTest.java index 901b2680..686c5adf 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/SuccessTimingContractTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/SuccessTimingContractTest.java @@ -107,6 +107,8 @@ class SuccessTimingContractTest { .thenReturn("oss://shufuai/collect-data/6868.xlsx"); lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1); lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + // 终态条件更新的返回值为「匹配行数」：任务非 FAILED 时匹配 1 行（默认用例都基于 RUNNING 任务）。 + lenient().when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); } private TaskFileJobEntity job() { @@ -124,6 +126,8 @@ class SuccessTimingContractTest { void successHappensOnlyAfterFileGeneratedAndUploaded() { service.processResultFileJob(job()); + // 契约核心：结果文件「生成并上传」必须先于任何落库（结果行与任务终态）。 + // 保持结果行先写：若进程在这两步之间退出，任务仍是 RUNNING，陈旧巡检可重新组装（自愈）。 var order = inOrder(ossStorageService, fileResultMapper, fileTaskMapper); order.verify(ossStorageService).uploadResultFile(any(), eq("COLLECT_DATA")); order.verify(fileResultMapper).updateById(any(FileResultEntity.class)); @@ -220,6 +224,35 @@ class SuccessTimingContractTest { verify(fileTaskMapper, org.mockito.Mockito.times(1)).update(isNull(), any(LambdaUpdateWrapper.class)); } + /** + * taskId 28599：组装完成时任务已是 FAILED（客户端上报失败或陈旧判死在前）—— + * 条件更新不匹配，任务保持失败态与真实原因，但结果文件的 url 照常落库， + * 用户「看到失败」的同时仍能下载已采集的部分结果。 + */ + @Test + void failedTaskKeepsFailureButStillCarriesDownloadUrl() { + String reason = "中间分批回传失败，终止本次采集以避免服务端数据残缺"; + when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(0); + FileTaskEntity failedTask = runningTask(); + failedTask.setStatus("FAILED"); + failedTask.setErrorMessage(reason); + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(failedTask); + FileResultEntity failedResult = runningResult(); + failedResult.setErrorMessage(reason); + when(fileResultMapper.selectById(RESULT_ID)).thenReturn(failedResult); + + service.processResultFileJob(job()); + + ArgumentCaptor resultCaptor = ArgumentCaptor.forClass(FileResultEntity.class); + // 失败态是补偿写：先乐观写成功，条件更新不匹配后再写回失败语义 + verify(fileResultMapper, org.mockito.Mockito.times(2)).updateById(resultCaptor.capture()); + FileResultEntity result = resultCaptor.getAllValues().get(1); + assertEquals("oss://shufuai/collect-data/6868.xlsx", result.getResultFileUrl(), + "失败任务也要有可下载的结果文件"); + assertEquals(0, result.getSuccess(), "任务失败则结果记录保持失败语义，前端显示「失败」而非「已完成」"); + assertEquals(reason, result.getErrorMessage(), "真实失败原因不能被组装流程清掉"); + } + private FileTaskEntity runningTask() { FileTaskEntity task = new FileTaskEntity(); task.setId(TASK_ID); diff --git a/frontend-vue/src/pages/brand/components/BrandCollectDataTab.vue b/frontend-vue/src/pages/brand/components/BrandCollectDataTab.vue index 8f4de4f0..691ee184 100644 --- a/frontend-vue/src/pages/brand/components/BrandCollectDataTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandCollectDataTab.vue @@ -596,7 +596,9 @@ function taskKeywordProgress(item: CollectDataHistoryItem) { function canDownload(item: CollectDataHistoryItem) { const status = normalizeTaskStatus(item) - return Boolean(item.downloadUrl && (item.success || status === 'SUCCESS')) + // 失败任务只要已生成结果文件也允许下载：失败时后端按已收到的分片组装部分结果， + // 已采集的数据不应随任务失败一起不可用（taskId 28599 事故）。 + return Boolean(item.downloadUrl && (item.success || status === 'SUCCESS' || status === 'FAILED')) } async function downloadResult(item: CollectDataHistoryItem) { diff --git a/frontend-vue/src/pages/brand/components/BrandPriceTrackTab.vue b/frontend-vue/src/pages/brand/components/BrandPriceTrackTab.vue index 3887bc95..040225a7 100644 --- a/frontend-vue/src/pages/brand/components/BrandPriceTrackTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandPriceTrackTab.vue @@ -1658,7 +1658,8 @@ function statusClass(item: PriceTrackHistoryItem) { function canDownload(item: PriceTrackHistoryItem) { if (!item.resultId || (!item.fileReady && !item.downloadUrl)) return false const st = resolvedTaskStatus(item) - return st === 'SUCCESS' || item.success === true + // 失败任务也可能有已跑出来的部分结果文件（后端按已收分片组装），有文件就允许下载 + return st === 'SUCCESS' || item.success === true || st === 'FAILED' } async function downloadResult(item: PriceTrackHistoryItem) { diff --git a/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue b/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue index 63950d1a..f9c13343 100644 --- a/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue @@ -1018,7 +1018,8 @@ function statusClass(item: ProductRiskHistoryItem) { function canDownload(item: ProductRiskHistoryItem) { if (!item.resultId || (!item.fileReady && !item.downloadUrl)) return false const st = resolvedTaskStatus(item) - return st === 'SUCCESS' || item.success === true + // 失败任务也可能有已跑出来的部分结果文件（后端按已收分片组装），有文件就允许下载 + return st === 'SUCCESS' || item.success === true || st === 'FAILED' } async function downloadResult(item: ProductRiskHistoryItem) { diff --git a/frontend-vue/src/pages/brand/components/BrandShopMatchTab.vue b/frontend-vue/src/pages/brand/components/BrandShopMatchTab.vue index 537e45f4..806185d9 100644 --- a/frontend-vue/src/pages/brand/components/BrandShopMatchTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandShopMatchTab.vue @@ -531,7 +531,8 @@ function currentTaskStageText(item: ShopMatchHistoryItem) { const snapshot = tas function nextScheduledDisplay(item: ShopMatchHistoryItem) { const snapshot = taskSnapshotOf(item); const task = snapshot?.task; const stages = task?.scheduleStages || []; if (typeof task?.currentStageIndex === 'number') { const stage = stages.find((entry) => entry.stageIndex === task.currentStageIndex); if (stage?.scheduledAt) return formatMonthDayTime(stage.scheduledAt) } if (item.scheduledAt) return formatMonthDayTime(item.scheduledAt); return '' } function statusText(item: ShopMatchHistoryItem) { const status = resolvedTaskStatus(item); if (status === 'SCHEDULED') return '待执行'; if (status === 'RUNNING') return '执行中'; if (status === 'SUCCESS' || status === 'COMPLETED') return '已完成'; if (status === 'FAILED') return '失败'; return item.success ? '已完成' : '未知' } function statusClass(item: ShopMatchHistoryItem) { const status = resolvedTaskStatus(item); return status === 'SUCCESS' || status === 'COMPLETED' ? 'success' : status === 'FAILED' ? 'failed' : 'running' } -function canDownload(item: ShopMatchHistoryItem) { const status = resolvedTaskStatus(item); return !!item.resultId && (!!item.fileReady || !!item.downloadUrl) && (status === 'SUCCESS' || status === 'COMPLETED') } +// 失败任务也可能有已跑出来的部分结果文件（后端按已收分片组装），有文件就允许下载 +function canDownload(item: ShopMatchHistoryItem) { const status = resolvedTaskStatus(item); return !!item.resultId && (!!item.fileReady || !!item.downloadUrl) && (status === 'SUCCESS' || status === 'COMPLETED' || status === 'FAILED') } async function downloadResult(item: ShopMatchHistoryItem) { if (!item.resultId) return; const url = getShopMatchResultDownloadUrl(item.resultId); const filename = item.outputFilename || `${item.shopName || 'result'}.xlsx`; const result = await saveUrlWithProgress(url, filename, `shop-match:${item.resultId}`); if (result.success) ElMessage.success(`已保存: ${result.path || filename}`); else if (result.error && result.error !== '用户取消') ElMessage.error(result.error) } async function deleteTaskRecord(item: ShopMatchHistoryItem) { const taskId = normalizeTaskId(item.taskId)