super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(品牌检测): 熔断中止落真实原因+部分结果可下载，熔断加持续时长门槛

Browse Source 
- 熔断改为「连续失败持续 6 分钟未恢复」才中止：原 8 次即中止，5 线程一轮
  就能凑满，重跑机制没机会生效导致任务失败率过高（任务 2309 复盘）
- 期间每 30s 冷却重试，重跑轮次 10→30；限流窗口恢复后任务自动跑完
- 失败终态上报内部接口 /api/internal/brand/tasks/{id}/abort：落真实原因 +
  用已收分片部分组装结果（未检测品牌单独成 sheet），不再悬挂到心跳超时被
  判「前端长时间无响应」且已跑数据无法下载
- 组装前从分片重建聚合（缓存快照可能缺后加字段如 keptBrands）
- 前端：失败任务有结果文件即显示「下载结果」
This commit is contained in:
黄自达
2026-09-14 17:04:19 +08:00
parent 24c5a09c7f
commit b54f72d3f6
7 changed files with 500 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/brand/controller/InternalBrandTaskController.java
+56
View File
@@ -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 +
* 真实原因，避免任务悬挂到心跳超时被判「前端长时间无响应」、已跑出的数据无法下载。
*
* <p>鉴权：仅凭 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<Map<String, Object>> 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));
}
}
backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java
+3
View File
@@ -18,4 +18,7 @@ public class BrandFileAggregateCacheDto {
private Boolean completed = false;
private List<BrandInvalidBrandDto> invalidBrands = new ArrayList<>();
private List<String> queryFailedBrands = new ArrayList<>();
/** 已判定保留的品牌：与 invalidBrands / queryFailedBrands 一起构成「已检测品牌」，
* 失败任务的未检测品牌 = 源文件品牌 - 三者并集（部分组装时写「未检测品牌」sheet）。 */
private List<String> keptBrands = new ArrayList<>();
}
backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandTaskAbortRequest.java
+13
View File
@@ -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;
}
backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java
+167 -17
View File
@@ -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<OutputEntry> 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<String, Object> resultPaths = buildAndUploadResult(taskId, outputEntries);
Map<String, Object> resultPaths = assembleAndUploadResult(taskId, strategy, sourceFiles, cachedByUrl, aggregates, false);
int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.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<String, Object> assembleAndUploadResult(Long taskId,
String strategy,
List<BrandSourceFileDto> sourceFiles,
Map<String, BrandParsedFileCacheDto> cachedByUrl,
Map<String, BrandFileAggregateCacheDto> aggregates,
boolean partial) throws IOException {
File outputDir = FileUtil.mkdir(FileUtil.file(storageProperties.getLocalTempDir(), "brand-result", String.valueOf(taskId)));
List<OutputEntry> 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<String> 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<String> resolveUndetectedBrands(BrandParsedFileCacheDto cachedFile, BrandFileAggregateCacheDto aggregate) {
List<Map<String, Object>> sourceRows = cachedFile.getRows() == null ? List.of() : cachedFile.getRows();
Set<String> handled = new LinkedHashSet<>();
handled.addAll(normalizeBrandSet(aggregate.getKeptBrands()));
handled.addAll(normalizeBrandSetFromInvalids(aggregate.getInvalidBrands()));
handled.addAll(normalizeBrandSet(aggregate.getQueryFailedBrands()));
LinkedHashSet<String> undetected = new LinkedHashSet<>();
for (Map<String, Object> 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<String, Object> 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<String, Object> 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<BrandSourceFileDto> sourceFiles = parseSourceFiles(task.getFilePaths());
int totalCount = sourceFiles.size();
Map<String, BrandParsedFileCacheDto> cachedByUrl =
indexCachedFiles(brandTaskStorageService.getParsedPayload(taskId));
// 先从分片重建聚合再读取：缓存的 state_json 可能是旧版本（缺后加字段，
// 如 keptBrands）或与已落库分片不一致，直接读会把已检测品牌误判为未检测
for (BrandSourceFileDto sourceFile : sourceFiles) {
brandTaskStorageService.refreshFileAggregate(taskId, sourceFile.getFileUrl());
}
Map<String, BrandFileAggregateCacheDto> aggregates = brandTaskStorageService.getAllFileAggregates(taskId);
boolean hasChunk = aggregates.values().stream()
.anyMatch(item -> item != null && defaultInteger(item.getReceivedChunkCount()) > 0);
Map<String, Object> 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<BrandCrawlTaskEntity> wrapper = new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.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<BrandCrawlResultFileDto> resultFiles) {
Set<String> 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<String> undetectedBrands) throws IOException {
String actualStrategy = normalizeStrategy(strategy);
SXSSFWorkbook workbook = new SXSSFWorkbook(200);
workbook.setCompressTempFiles(true);
@@ -1048,6 +1173,10 @@ public class BrandTaskService {
}
Set<String> invalidBrandSet = normalizeBrandSetFromInvalids(resultFile.getInvalidBrands());
// 未检测品牌（任务中止、分片没到齐）：主 sheet 剔除这些行、整体挪到独立 sheet，
// 避免用户把「没查过」的行误当成「已通过检测」上架
Set<String> undetectedBrandSet = normalizeBrandSet(undetectedBrands);
List<Map<String, Object>> undetectedRows = new ArrayList<>();
List<Map<String, Object>> sourceRows = cachedFile.getRows() == null ? List.of() : cachedFile.getRows();
int writeRowIndex = 1;
for (Map<String, Object> 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<String, Object> 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("品牌");
backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskStorageService.java
+8
View File
@@ -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;
}