From b54f72d3f6ff5917c93968228f9c3b6d2ed620d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Mon, 14 Sep 2026 17:04:19 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E5=93=81=E7=89=8C=E6=A3=80=E6=B5=8B):=20?= =?UTF-8?q?=E7=86=94=E6=96=AD=E4=B8=AD=E6=AD=A2=E8=90=BD=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E5=8E=9F=E5=9B=A0+=E9=83=A8=E5=88=86=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E5=8F=AF=E4=B8=8B=E8=BD=BD=EF=BC=8C=E7=86=94=E6=96=AD=E5=8A=A0?= =?UTF-8?q?=E6=8C=81=E7=BB=AD=E6=97=B6=E9=95=BF=E9=97=A8=E6=A7=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 熔断改为「连续失败持续 6 分钟未恢复」才中止：原 8 次即中止，5 线程一轮 就能凑满，重跑机制没机会生效导致任务失败率过高（任务 2309 复盘） - 期间每 30s 冷却重试，重跑轮次 10→30；限流窗口恢复后任务自动跑完 - 失败终态上报内部接口 /api/internal/brand/tasks/{id}/abort：落真实原因 + 用已收分片部分组装结果（未检测品牌单独成 sheet），不再悬挂到心跳超时被 判「前端长时间无响应」且已跑数据无法下载 - 组装前从分片重建聚合（缓存快照可能缺后加字段如 keptBrands） - 前端：失败任务有结果文件即显示「下载结果」 --- .../InternalBrandTaskController.java | 56 ++++ .../model/dto/BrandFileAggregateCacheDto.java | 3 + .../model/dto/BrandTaskAbortRequest.java | 13 + .../brand/service/BrandTaskService.java | 184 +++++++++++-- .../service/BrandTaskStorageService.java | 8 + .../service/BrandTaskAbortServiceTest.java | 249 ++++++++++++++++++ .../pages/brand/components/BrandBrandTab.vue | 5 +- 7 files changed, 500 insertions(+), 18 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/brand/controller/InternalBrandTaskController.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandTaskAbortRequest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskAbortServiceTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/controller/InternalBrandTaskController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/controller/InternalBrandTaskController.java new file mode 100644 index 00000000..152b52bd --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/controller/InternalBrandTaskController.java @@ -0,0 +1,56 @@ +package com.nanri.aiimage.modules.brand.controller; + +import com.nanri.aiimage.common.api.ApiResponse; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.security.AdminAuthSupport; +import com.nanri.aiimage.modules.brand.model.dto.BrandTaskAbortRequest; +import com.nanri.aiimage.modules.brand.service.BrandTaskService; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.servlet.http.HttpServletRequest; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.Map; + +/** + * 内部接口：品牌任务中止上报，供主机 A 品牌检测服务（15126）在爬取不可继续时调用 + * （如 WIPO 连续限流熔断）。Java 侧用已收到的结果分片部分组装结果文件并落 failed + + * 真实原因，避免任务悬挂到心跳超时被判「前端长时间无响应」、已跑出的数据无法下载。 + * + *

鉴权：仅凭 X-Internal-Token（与容器 AIIMAGE_INTERNAL_TOKEN / 宿主机 + * ~/.aiimage/internal-token 同值）。/api/internal 前缀虽在 AdminApiGuardFilter 兜底 + * 名单内、可信令牌会放行，controller 仍须自校验——防止配置漂移时匿名可达。 + */ +@RestController +@RequiredArgsConstructor +@Slf4j +@RequestMapping("/api/internal/brand") +@Tag(name = "内部接口", description = "服务间调用（X-Internal-Token 鉴权）。") +public class InternalBrandTaskController { + + private final BrandTaskService brandTaskService; + private final AdminAuthSupport adminAuthSupport; + + @PostMapping("/tasks/{taskId}/abort") + @Operation(summary = "上报品牌任务中止（爬取方调用）", + description = "用已收到的结果分片部分组装结果文件（未检测品牌单独成 sheet）并落 failed + 真实原因；幂等，终态任务直接返回。") + public ApiResponse> abortTask(HttpServletRequest request, + @PathVariable Long taskId, + @RequestBody(required = false) BrandTaskAbortRequest body) { + if (!adminAuthSupport.isTrustedInternalToken(request)) { + log.warn("[internal-brand-abort] 拒绝未携带可信内部令牌的请求 taskId={} remoteAddr={}", + taskId, request.getRemoteAddr()); + throw new BusinessException(401, "未授权"); + } + String errorMessage = body == null ? null : body.getErrorMessage(); + log.info("[internal-brand-abort] 收到中止上报 taskId={} remoteAddr={} msg={}", + taskId, request.getRemoteAddr(), errorMessage); + return ApiResponse.success(brandTaskService.abortTask(taskId, errorMessage)); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java index 4cbf6cc1..2b27e24c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java @@ -18,4 +18,7 @@ public class BrandFileAggregateCacheDto { private Boolean completed = false; private List invalidBrands = new ArrayList<>(); private List queryFailedBrands = new ArrayList<>(); + /** 已判定保留的品牌：与 invalidBrands / queryFailedBrands 一起构成「已检测品牌」， + * 失败任务的未检测品牌 = 源文件品牌 - 三者并集（部分组装时写「未检测品牌」sheet）。 */ + private List keptBrands = new ArrayList<>(); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandTaskAbortRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandTaskAbortRequest.java new file mode 100644 index 00000000..8880227b --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandTaskAbortRequest.java @@ -0,0 +1,13 @@ +package com.nanri.aiimage.modules.brand.model.dto; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +@Data +@Schema(description = "品牌任务中止上报请求（内部接口，由爬取方 15126 调用）。") +public class BrandTaskAbortRequest { + + @Schema(description = "中止原因，会原样写入任务 error_message，前端任务列表展示该文案。", + example = "连续 8 次请求被 WIPO 限流（返回 Forbidden），已中止任务；请检查代理配置或错峰重跑") + private String errorMessage; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index 1115bab5..fc69220b 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -686,23 +686,8 @@ public class BrandTaskService { } brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_ASSEMBLING, totalCount, totalCount); - File outputDir = FileUtil.mkdir(FileUtil.file(storageProperties.getLocalTempDir(), "brand-result", String.valueOf(taskId))); - List outputEntries = new ArrayList<>(); try { - for (BrandSourceFileDto sourceFile : sourceFiles) { - BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl()); - BrandParsedFileCacheDto cachedFile = cachedByUrl.get(sourceFile.getFileUrl()); - if (cachedFile == null) { - throw new BusinessException("缺少原始缓存数据: " + sourceFile.getFileUrl()); - } - File sourceLocalFile = resolveSourceFile(sourceFile); - String originalFilename = resolveOriginalFilename(sourceFile, sourceLocalFile); - File outputFile = buildNamedOutputFile(outputDir, buildResultFilename(originalFilename)); - writeBrandWorkbook(outputFile, strategy, cachedFile, aggregate); - outputEntries.add(new OutputEntry(sourceLocalFile, originalFilename, outputFile, originalFilename)); - } - brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_UPLOADING, totalCount, totalCount); - Map resultPaths = buildAndUploadResult(taskId, outputEntries); + Map resultPaths = assembleAndUploadResult(taskId, strategy, sourceFiles, cachedByUrl, aggregates, false); int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() .eq(BrandCrawlTaskEntity::getId, taskId) .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) @@ -736,11 +721,150 @@ public class BrandTaskService { throw businessException; } throw new BusinessException(ex.getMessage()); + } + } + + /** + * 组装并上传结果文件。partial=false 为成功收尾（要求全部文件分片收齐，由调用方校验）； + * partial=true 为失败中止的部分组装：用已收到的分片出结果，未检测品牌单独成 sheet。 + */ + private Map assembleAndUploadResult(Long taskId, + String strategy, + List sourceFiles, + Map cachedByUrl, + Map aggregates, + boolean partial) throws IOException { + File outputDir = FileUtil.mkdir(FileUtil.file(storageProperties.getLocalTempDir(), "brand-result", String.valueOf(taskId))); + List outputEntries = new ArrayList<>(); + try { + for (BrandSourceFileDto sourceFile : sourceFiles) { + BrandParsedFileCacheDto cachedFile = cachedByUrl.get(sourceFile.getFileUrl()); + if (cachedFile == null) { + if (!partial) { + throw new BusinessException("缺少原始缓存数据: " + sourceFile.getFileUrl()); + } + log.warn("[brand-assemble] taskId={} 原始缓存数据缺失，跳过该文件 fileUrl={}", taskId, sourceFile.getFileUrl()); + continue; + } + BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl()); + if (aggregate == null) { + if (!partial) { + throw new BusinessException("缺少结果聚合数据: " + sourceFile.getFileUrl()); + } + aggregate = new BrandFileAggregateCacheDto(); + aggregate.setFileUrl(sourceFile.getFileUrl()); + } + List undetectedBrands = partial ? resolveUndetectedBrands(cachedFile, aggregate) : List.of(); + File sourceLocalFile = resolveSourceFile(sourceFile); + String originalFilename = resolveOriginalFilename(sourceFile, sourceLocalFile); + File outputFile = buildNamedOutputFile(outputDir, buildResultFilename(originalFilename)); + writeBrandWorkbook(outputFile, strategy, cachedFile, aggregate, undetectedBrands); + outputEntries.add(new OutputEntry(sourceLocalFile, originalFilename, outputFile, originalFilename)); + } + if (outputEntries.isEmpty()) { + throw new BusinessException("没有可组装的结果文件"); + } + brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_UPLOADING, + sourceFiles.size(), sourceFiles.size()); + return buildAndUploadResult(taskId, outputEntries); } finally { cleanupBrandResultTempFiles(outputEntries, outputDir); } } + /** 未检测品牌 = 源文件品牌 -（保留 ∪ 不符合品牌 ∪ 查询失败品牌）；按源文件出现顺序去重。 */ + private List resolveUndetectedBrands(BrandParsedFileCacheDto cachedFile, BrandFileAggregateCacheDto aggregate) { + List> sourceRows = cachedFile.getRows() == null ? List.of() : cachedFile.getRows(); + Set handled = new LinkedHashSet<>(); + handled.addAll(normalizeBrandSet(aggregate.getKeptBrands())); + handled.addAll(normalizeBrandSetFromInvalids(aggregate.getInvalidBrands())); + handled.addAll(normalizeBrandSet(aggregate.getQueryFailedBrands())); + LinkedHashSet undetected = new LinkedHashSet<>(); + for (Map rowData : sourceRows) { + String brand = normalizeCellText(Objects.toString(rowData.getOrDefault("品牌", ""), "")); + if (!brand.isBlank() && !handled.contains(brand)) { + undetected.add(brand); + } + } + return new ArrayList<>(undetected); + } + + /** + * 爬取方（15126）中止上报：任务不可能再收到剩余分片时调用（如 WIPO 连续限流熔断）。 + * 用已收到的分片部分组装结果文件并落 failed + 真实原因——否则 Java 侧任务会悬挂到 + * 心跳超时被判「前端长时间无响应」，且已跑出的数据因没有 result_paths 无法下载。 + * 幂等：任务已是终态时直接返回，不覆盖既有结果。 + */ + public Map abortTask(Long taskId, String errorMessage) { + if (taskId == null || taskId <= 0) { + throw new BusinessException("taskId invalid"); + } + try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId, RESULT_SUBMIT_WAIT_MILLIS)) { + BrandCrawlTaskEntity task = requireTask(taskId); + String status = blankToDefault(task.getStatus(), STATUS_PENDING); + Map data = new LinkedHashMap<>(); + data.put("taskId", taskId); + boolean hasResult = task.getResultPaths() != null && !task.getResultPaths().isBlank(); + // success/cancelled 不动；failed 已有结果也不重复组装。failed 且无结果 + // （如被心跳超时兜底判失败的历史任务）允许补组装——存量补救路径。 + if (STATUS_SUCCESS.equalsIgnoreCase(status) || STATUS_CANCELLED.equalsIgnoreCase(status) + || (STATUS_FAILED.equalsIgnoreCase(status) && hasResult)) { + log.info("[brand-abort] taskId={} 已是终态且无需补组装 status={} hasResult={}，跳过", + taskId, status, hasResult); + data.put("status", status); + data.put("resultGenerated", false); + return data; + } + String message = blankToDefault(errorMessage, + blankToDefault(task.getErrorMessage(), "品牌检测任务已中止")); + List sourceFiles = parseSourceFiles(task.getFilePaths()); + int totalCount = sourceFiles.size(); + Map cachedByUrl = + indexCachedFiles(brandTaskStorageService.getParsedPayload(taskId)); + // 先从分片重建聚合再读取：缓存的 state_json 可能是旧版本（缺后加字段， + // 如 keptBrands）或与已落库分片不一致，直接读会把已检测品牌误判为未检测 + for (BrandSourceFileDto sourceFile : sourceFiles) { + brandTaskStorageService.refreshFileAggregate(taskId, sourceFile.getFileUrl()); + } + Map aggregates = brandTaskStorageService.getAllFileAggregates(taskId); + boolean hasChunk = aggregates.values().stream() + .anyMatch(item -> item != null && defaultInteger(item.getReceivedChunkCount()) > 0); + Map resultPaths = null; + if (hasChunk && !cachedByUrl.isEmpty() && !sourceFiles.isEmpty()) { + try { + resultPaths = assembleAndUploadResult(taskId, normalizeStrategy(task.getStrategy()), + sourceFiles, cachedByUrl, aggregates, true); + log.info("[brand-abort] taskId={} 部分结果组装完成 files={}", taskId, sourceFiles.size()); + } catch (Exception ex) { + log.warn("[brand-abort] taskId={} 部分结果组装失败（仅标记失败） msg={}", taskId, ex.getMessage(), ex); + } + } else { + log.info("[brand-abort] taskId={} 无已收到分片，跳过结果组装 receivedAggregates={}", taskId, aggregates.size()); + } + int finishedCount = brandTaskStorageService.countCompletedFiles(taskId); + LambdaUpdateWrapper wrapper = new LambdaUpdateWrapper() + .eq(BrandCrawlTaskEntity::getId, taskId) + .in(BrandCrawlTaskEntity::getStatus, STATUS_PENDING, STATUS_RUNNING, STATUS_FAILED) + .set(BrandCrawlTaskEntity::getStatus, STATUS_FAILED) + .set(BrandCrawlTaskEntity::getErrorMessage, message) + .set(BrandCrawlTaskEntity::getProgressCurrent, finishedCount) + .set(BrandCrawlTaskEntity::getProgressTotal, totalCount); + if (resultPaths != null) { + wrapper.set(BrandCrawlTaskEntity::getResultPaths, JSONUtil.toJsonStr(resultPaths)); + } + int updated = brandCrawlTaskMapper.update(null, wrapper); + if (updated > 0) { + brandTaskProgressCacheService.markFailed(taskId, message); + saveBrandProgressSnapshot(taskId, STATUS_FAILED, totalCount, finishedCount, 1, message); + } + log.info("[brand-abort] taskId={} aborted updated={} resultGenerated={} finishedFiles={}/{} msg={}", + taskId, updated, resultPaths != null, finishedCount, totalCount, message); + data.put("status", STATUS_FAILED); + data.put("resultGenerated", resultPaths != null); + return data; + } + } + private void ensureNoDuplicateResultFiles(List resultFiles) { Set seen = new LinkedHashSet<>(); for (BrandCrawlResultFileDto resultFile : resultFiles) { @@ -1034,7 +1158,8 @@ public class BrandTaskService { private void writeBrandWorkbook(File outputFile, String strategy, BrandParsedFileCacheDto cachedFile, - BrandFileAggregateCacheDto resultFile) throws IOException { + BrandFileAggregateCacheDto resultFile, + List undetectedBrands) throws IOException { String actualStrategy = normalizeStrategy(strategy); SXSSFWorkbook workbook = new SXSSFWorkbook(200); workbook.setCompressTempFiles(true); @@ -1048,6 +1173,10 @@ public class BrandTaskService { } Set invalidBrandSet = normalizeBrandSetFromInvalids(resultFile.getInvalidBrands()); + // 未检测品牌（任务中止、分片没到齐）：主 sheet 剔除这些行、整体挪到独立 sheet， + // 避免用户把「没查过」的行误当成「已通过检测」上架 + Set undetectedBrandSet = normalizeBrandSet(undetectedBrands); + List> undetectedRows = new ArrayList<>(); List> sourceRows = cachedFile.getRows() == null ? List.of() : cachedFile.getRows(); int writeRowIndex = 1; for (Map rowData : sourceRows) { @@ -1055,6 +1184,10 @@ public class BrandTaskService { if (!brand.isBlank() && invalidBrandSet.contains(brand)) { continue; } + if (!brand.isBlank() && undetectedBrandSet.contains(brand)) { + undetectedRows.add(rowData); + continue; + } var row = mainSheet.createRow(writeRowIndex++); for (int colIndex = 0; colIndex < columns.size(); colIndex++) { String column = columns.get(colIndex); @@ -1062,6 +1195,23 @@ public class BrandTaskService { } } + if (!undetectedRows.isEmpty()) { + var undetectedSheet = workbook.createSheet("未检测品牌"); + var undetectedHeader = undetectedSheet.createRow(0); + for (int i = 0; i < columns.size(); i++) { + undetectedHeader.createCell(i).setCellValue(columns.get(i)); + } + for (int i = 0; i < undetectedRows.size(); i++) { + Map rowData = undetectedRows.get(i); + var row = undetectedSheet.createRow(i + 1); + for (int colIndex = 0; colIndex < columns.size(); colIndex++) { + String column = columns.get(colIndex); + row.createCell(colIndex).setCellValue(Objects.toString(rowData.getOrDefault(column, ""), "")); + } + } + applyBrandSheetWidths(undetectedSheet, columns.size()); + } + var invalidSheet = workbook.createSheet("不符合品牌"); var invalidHeader = invalidSheet.createRow(0); invalidHeader.createCell(0).setCellValue("品牌"); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskStorageService.java index 1931b02f..a55526a9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskStorageService.java @@ -325,6 +325,7 @@ public class BrandTaskStorageService { aggregate.setCompleted(false); aggregate.setInvalidBrands(new ArrayList<>()); aggregate.setQueryFailedBrands(new ArrayList<>()); + aggregate.setKeptBrands(new ArrayList<>()); return aggregate; } @@ -362,6 +363,12 @@ public class BrandTaskStorageService { if (file.getQueryFailedBrands() != null && !file.getQueryFailedBrands().isEmpty()) { aggregate.getQueryFailedBrands().addAll(file.getQueryFailedBrands()); } + if (file.getKeptRows() != null && !file.getKeptRows().isEmpty()) { + if (aggregate.getKeptBrands() == null) { + aggregate.setKeptBrands(new ArrayList<>()); + } + aggregate.getKeptBrands().addAll(file.getKeptRows()); + } } private BrandFileAggregateCacheDto rebuildAggregate(Long taskId, String scopeKey, String scopeHash) { @@ -379,6 +386,7 @@ public class BrandTaskStorageService { aggregate.setCompleted(false); aggregate.setInvalidBrands(new ArrayList<>()); aggregate.setQueryFailedBrands(new ArrayList<>()); + aggregate.setKeptBrands(new ArrayList<>()); } return aggregate; } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskAbortServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskAbortServiceTest.java new file mode 100644 index 00000000..4533d3b7 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskAbortServiceTest.java @@ -0,0 +1,249 @@ +package com.nanri.aiimage.modules.brand.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.config.BrandProgressProperties; +import com.nanri.aiimage.config.StorageProperties; +import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper; +import com.nanri.aiimage.modules.brand.model.dto.BrandCrawlResultFileDto; +import com.nanri.aiimage.modules.brand.model.dto.BrandFileAggregateCacheDto; +import com.nanri.aiimage.modules.brand.model.dto.BrandParsedFileCacheDto; +import com.nanri.aiimage.modules.brand.model.dto.BrandSourceFileDto; +import com.nanri.aiimage.modules.brand.model.entity.BrandCrawlTaskEntity; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 品牌任务中止上报（abortTask）：熔断中止时用已收到的分片部分组装结果， + * 落 failed + 真实原因，避免任务悬挂到心跳超时、已跑出的数据无法下载。 + */ +@ExtendWith(MockitoExtension.class) +class BrandTaskAbortServiceTest { + + @Mock private BrandCrawlTaskMapper brandCrawlTaskMapper; + @Mock private OssStorageService ossStorageService; + @Mock private StorageProperties storageProperties; + @Mock private BrandProgressProperties brandProgressProperties; + @Mock private BrandTaskProgressCacheService brandTaskProgressCacheService; + @Mock private BrandTaskStorageService brandTaskStorageService; + @Mock private DistributedJobLockService distributedJobLockService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + + @TempDir + Path tempDir; + + @BeforeAll + static void initializeTableInfo() { + TableInfoHelper.initTableInfo( + new MapperBuilderAssistant(new MybatisConfiguration(), ""), + BrandCrawlTaskEntity.class); + } + + private BrandTaskService service() { + return new BrandTaskService(brandCrawlTaskMapper, ossStorageService, storageProperties, + brandProgressProperties, brandTaskProgressCacheService, brandTaskStorageService, + distributedJobLockService, taskDistributedLockService, new ObjectMapper(), + taskFileJobService, taskProgressSnapshotService); + } + + private BrandCrawlTaskEntity task(String status) { + return task(status, "https://oss.example/source.xlsx"); + } + + private BrandCrawlTaskEntity task(String status, String fileUrl) { + BrandCrawlTaskEntity entity = new BrandCrawlTaskEntity(); + entity.setId(2309L); + entity.setUserId(1050L); + entity.setStatus(status); + entity.setStrategy("Simple"); + BrandSourceFileDto source = new BrandSourceFileDto(); + source.setFileUrl(fileUrl); + source.setOriginalFilename("待检测品牌.xlsx"); + entity.setFilePaths(new ObjectMapper().valueToTree(List.of(source)).toString()); + return entity; + } + + @SuppressWarnings("unchecked") + private LambdaUpdateWrapper captureUpdateWrapper() { + ArgumentCaptor> captor = ArgumentCaptor.forClass(Wrapper.class); + verify(brandCrawlTaskMapper).update(isNull(), captor.capture()); + return (LambdaUpdateWrapper) captor.getValue(); + } + + @Test + void terminalTaskIsSkippedIdempotently() { + when(brandCrawlTaskMapper.selectById(2309L)).thenReturn(task("success")); + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + when(taskDistributedLockService.acquire(anyString(), any(), anyLong())).thenReturn(handle); + + Map result = service().abortTask(2309L, "限流中止"); + + assertEquals("success", result.get("status")); + assertFalse((Boolean) result.get("resultGenerated")); + verify(brandCrawlTaskMapper, never()).update(any(), any()); + verify(brandTaskProgressCacheService, never()).markFailed(anyLong(), anyString()); + } + + @Test + void abortWithoutChunksMarksFailedOnly() { + when(brandCrawlTaskMapper.selectById(2309L)).thenReturn(task("running")); + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + when(taskDistributedLockService.acquire(anyString(), any(), anyLong())).thenReturn(handle); + when(brandTaskStorageService.getParsedPayload(2309L)).thenReturn(List.of()); + when(brandTaskStorageService.getAllFileAggregates(2309L)).thenReturn(Map.of()); + when(brandCrawlTaskMapper.update(isNull(), any())).thenReturn(1); + + Map result = service().abortTask(2309L, "连续 8 次请求被 WIPO 限流"); + + assertEquals("failed", result.get("status")); + assertFalse((Boolean) result.get("resultGenerated")); + LambdaUpdateWrapper wrapper = captureUpdateWrapper(); + assertFalse(wrapper.getSqlSet().contains("result_paths")); + verify(brandTaskProgressCacheService).markFailed(2309L, "连续 8 次请求被 WIPO 限流"); + verify(ossStorageService, never()).uploadResultFile(any(), anyString()); + } + + @Test + void abortWithChunksAssemblesPartialResult() throws Exception { + assertPartialAssembly("running"); + } + + /** 存量补救：已被心跳兜底判失败、无结果文件的旧任务（如任务 2309）补组装。 */ + @Test + void failedTaskWithoutResultIsRepairedByAssembly() throws Exception { + assertPartialAssembly("failed"); + } + + @Test + void failedTaskWithExistingResultIsSkipped() { + BrandCrawlTaskEntity entity = task("failed"); + entity.setResultPaths("{\"zip_url\":\"https://oss.example/existing.zip\"}"); + when(brandCrawlTaskMapper.selectById(2309L)).thenReturn(entity); + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + when(taskDistributedLockService.acquire(anyString(), any(), anyLong())).thenReturn(handle); + + Map result = service().abortTask(2309L, "限流中止"); + + assertEquals("failed", result.get("status")); + assertFalse((Boolean) result.get("resultGenerated")); + verify(brandCrawlTaskMapper, never()).update(any(), any()); + } + + private void assertPartialAssembly(String initialStatus) throws Exception { + File sourceFile = tempDir.resolve("source.xlsx").toFile(); + Files.write(sourceFile.toPath(), new byte[]{1, 2, 3}); + String fileUrl = sourceFile.getAbsolutePath(); + when(brandCrawlTaskMapper.selectById(2309L)).thenReturn(task(initialStatus, fileUrl)); + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + when(taskDistributedLockService.acquire(anyString(), any(), anyLong())).thenReturn(handle); + when(storageProperties.getLocalTempDir()).thenReturn(tempDir.toString()); + + BrandParsedFileCacheDto cached = new BrandParsedFileCacheDto(); + cached.setFileUrl(fileUrl); + cached.setSheetName("Sheet1"); + cached.setColumns(new ArrayList<>(List.of("品牌", "ASIN"))); + cached.setRows(new ArrayList<>(List.of( + row("A"), row("B"), row("C")))); + when(brandTaskStorageService.getParsedPayload(2309L)).thenReturn(List.of(cached)); + + BrandFileAggregateCacheDto aggregate = new BrandFileAggregateCacheDto(); + aggregate.setFileUrl(fileUrl); + aggregate.setChunkTotal(3); + aggregate.setReceivedChunkCount(1); + aggregate.setCompleted(false); + aggregate.setKeptBrands(new ArrayList<>(List.of("A"))); + Map aggregates = new LinkedHashMap<>(); + aggregates.put(fileUrl, aggregate); + when(brandTaskStorageService.getAllFileAggregates(2309L)).thenReturn(aggregates); + java.util.concurrent.atomic.AtomicReference capturedWorkbook = + new java.util.concurrent.atomic.AtomicReference<>(); + when(ossStorageService.uploadResultFile(any(File.class), anyString())).thenAnswer(invocation -> { + File uploaded = invocation.getArgument(0); + if (uploaded.getName().endsWith(".xlsx") && capturedWorkbook.get() == null) { + capturedWorkbook.set(org.apache.poi.ss.usermodel.WorkbookFactory.create(uploaded)); + } + return "result-key"; + }); + when(ossStorageService.generateFreshDownloadUrl(anyString())).thenReturn("https://oss.example/result-key"); + when(brandCrawlTaskMapper.update(isNull(), any())).thenReturn(1); + lenient().when(brandTaskStorageService.countCompletedFiles(2309L)).thenReturn(0); + + Map result = service().abortTask(2309L, "连续 8 次请求被 WIPO 限流"); + + assertEquals("failed", result.get("status")); + assertTrue((Boolean) result.get("resultGenerated")); + LambdaUpdateWrapper wrapper = captureUpdateWrapper(); + assertTrue(wrapper.getSqlSet().contains("result_paths")); + verify(ossStorageService, org.mockito.Mockito.atLeastOnce()).uploadResultFile(any(File.class), anyString()); + + org.apache.poi.ss.usermodel.Workbook workbook = capturedWorkbook.get(); + assertNotNull(workbook, "应生成结果工作簿"); + org.apache.poi.ss.usermodel.Sheet undetected = workbook.getSheet("未检测品牌"); + assertNotNull(undetected, "部分结果应包含「未检测品牌」sheet"); + assertEquals("B", undetected.getRow(1).getCell(0).getStringCellValue()); + assertEquals("C", undetected.getRow(2).getCell(0).getStringCellValue()); + assertEquals(1, workbook.getSheet("Sheet1").getLastRowNum(), "主 sheet 只剩表头 + 已检测品牌 A 一行"); + } + + @Test + void abortReportsRealErrorMessageInsteadOfStalePlaceholder() throws Exception { + when(brandCrawlTaskMapper.selectById(2309L)).thenReturn(task("running")); + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + when(taskDistributedLockService.acquire(anyString(), any(), anyLong())).thenReturn(handle); + when(brandTaskStorageService.getParsedPayload(2309L)).thenReturn(List.of()); + when(brandTaskStorageService.getAllFileAggregates(2309L)).thenReturn(Map.of()); + when(brandCrawlTaskMapper.update(isNull(), any())).thenReturn(1); + String realMessage = "连续 8 次请求被 WIPO 限流（返回 Forbidden），已中止任务；请检查代理配置或错峰重跑"; + + service().abortTask(2309L, realMessage); + + LambdaUpdateWrapper wrapper = captureUpdateWrapper(); + assertTrue(wrapper.getParamNameValuePairs().containsValue(realMessage), + "error_message 应写入熔断的真实原因，而不是心跳超时兜底文案"); + } + + private Map row(String brand) { + Map row = new LinkedHashMap<>(); + row.put("品牌", brand); + row.put("ASIN", "B0" + brand); + return row; + } +} diff --git a/frontend-vue/src/pages/brand/components/BrandBrandTab.vue b/frontend-vue/src/pages/brand/components/BrandBrandTab.vue index df725f9e..a1ec5a95 100644 --- a/frontend-vue/src/pages/brand/components/BrandBrandTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandBrandTab.vue @@ -246,7 +246,10 @@ function canDelete(item: BrandTaskItem) { function canDownload(item: BrandTaskItem) { const status = (item.status || '').toLowerCase() - return status === 'success' && Boolean(item.result_paths?.zip_url) + if (!item.result_paths?.zip_url) return false + // 失败任务也可能有结果文件：熔断中止时会用已跑出的数据组装部分结果 + //（未检测品牌单独成 sheet），同样允许下载 + return status === 'success' || status === 'failed' } function taskProgress(item: BrandTaskItem) {