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

提交打包

Browse Source
This commit is contained in:
super
2026-05-07 16:21:35 +08:00
parent afeeb8327c
commit 683a3934fc
106 changed files with 19062 additions and 23684 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12372
backend-java/aiimage-backend-a.log.2026-05-06.0 Normal file
View File

File diff suppressed because it is too large Load Diff

5171
backend-java/aiimage-backend-b.log.2026-05-06.0 Normal file
View File

File diff suppressed because it is too large Load Diff

30
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/controller/AppearancePatentController.java
View File

@@ -3,6 +3,8 @@ package com.nanri.aiimage.modules.appearancepatent.controller;
import com.nanri.aiimage.common.api.ApiResponse;
import com.nanri.aiimage.common.util.DownloadHeaderUtil;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParseRequest;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedGroupPageDto;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedPayloadDto;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentSubmitResultRequest;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentTaskBatchRequest;
import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentDashboardVo;
@@ -44,6 +46,34 @@ public class AppearancePatentController {
return ApiResponse.success(service.parseAndCreateTask(request));
}
@GetMapping("/tasks/{taskId}/parsed-payload")
@Operation(summary = "获取外观专利完整解析载荷", description = "供 Python 队列消费端使用。parse 接口仅返回轻量结果，完整行数据通过本接口按 taskId 拉取。")
public ApiResponse<AppearancePatentParsedPayloadDto> parsedPayload(
@Parameter(description = "外观专利任务 ID", required = true, example = "7004")
@PathVariable Long taskId,
@Parameter(description = "当前用户 ID", required = true, example = "1")
@RequestParam("user_id") Long userId) {
return ApiResponse.success(service.parsedPayload(taskId, userId));
}
@GetMapping("/tasks/{taskId}/queue-payload")
@Operation(summary = "获取外观专利 Python 队列载荷", description = "只返回 Python 当前需要的 groups，避免前端搬运完整解析大数据。")
public ApiResponse<AppearancePatentParsedPayloadDto> queuePayload(
@PathVariable Long taskId,
@RequestParam("user_id") Long userId) {
return ApiResponse.success(service.queuePayload(taskId, userId));
}
@GetMapping("/tasks/{taskId}/parsed-groups")
@Operation(summary = "分页获取外观专利解析分组", description = "供后续优化使用，按页返回解析分组。")
public ApiResponse<AppearancePatentParsedGroupPageDto> parsedGroups(
@PathVariable Long taskId,
@RequestParam("user_id") Long userId,
@RequestParam(value = "page", required = false, defaultValue = "1") Integer page,
@RequestParam(value = "page_size", required = false, defaultValue = "50") Integer pageSize) {
return ApiResponse.success(service.parsedGroupPage(taskId, userId, page, pageSize));
}
@GetMapping("/dashboard")
@Operation(summary = "查询外观专利检测总览", description = "查询当前用户的运行中、成功、失败和已结束任务数量。页面进入时请求一次即可，不需要持续轮询。")
public ApiResponse<AppearancePatentDashboardVo> dashboard(

23
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedGroupManifestDto.java Normal file
View File

@@ -0,0 +1,23 @@
package com.nanri.aiimage.modules.appearancepatent.model.dto;
import lombok.Data;
import java.util.ArrayList;
import java.util.List;
@Data
public class AppearancePatentParsedGroupManifestDto {
private String aiPrompt;
private String apiKey;
private Integer pageSize;
private Integer totalGroups;
private Integer totalRows;
private List<PageRef> pages = new ArrayList<>();
@Data
public static class PageRef {
private Integer page;
private Integer groupCount;
private String ref;
}
}

19
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedGroupPageDto.java Normal file
View File

@@ -0,0 +1,19 @@
package com.nanri.aiimage.modules.appearancepatent.model.dto;
import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentParsedGroupVo;
import lombok.Data;
import java.util.ArrayList;
import java.util.List;
@Data
public class AppearancePatentParsedGroupPageDto {
private String aiPrompt;
private String apiKey;
private Integer page;
private Integer pageSize;
private Integer totalGroups;
private Integer totalRows;
private Boolean hasNext;
private List<AppearancePatentParsedGroupVo> groups = new ArrayList<>();
}

9
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentParsedRowVo.java
View File

@@ -3,9 +3,6 @@ package com.nanri.aiimage.modules.appearancepatent.model.vo;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.util.LinkedHashMap;
import java.util.Map;
@Data
@Schema(description = "外观专利检测解析出的单行数据")
public class AppearancePatentParsedRowVo {
@@ -36,12 +33,12 @@ public class AppearancePatentParsedRowVo {
@Schema(description = "国家或站点。", example = "英国")
private String country;
@Schema(description = "价格。", example = "12.99")
private String price;
@Schema(description = "商品图片 URL 或商品 URL，供 Coze 检测使用。", example = "https://webstatic.aiproxy.vip/output/demo.jpg")
private String url;
@Schema(description = "商品标题。", example = "Women Floral Dress Summer Casual")
private String title;
@Schema(description = "该 Excel 行的原始列值映射。最终生成 xlsx 时可以从这里读取价格等字段。", example = "{\"id\":\"2_1\",\"asin\":\"B0CJ8SNXXV\",\"国家\":\"英国\",\"价格\":\"12.99\"}")
private Map<String, String> values = new LinkedHashMap<>();
}

130
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java
View File

@@ -13,6 +13,7 @@ import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.config.StorageProperties;
import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentCozeClient;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParseRequest;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedGroupPageDto;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedPayloadDto;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultGroupDto;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto;
@@ -175,7 +176,7 @@ public class AppearancePatentTaskService {
throw new BusinessException("未解析到有效 ASIN 数据");
}
long parsedAt = System.nanoTime();
List<AppearancePatentParsedGroupVo> groups = buildParsedGroups(allRows);
int groupCount = countParsedGroups(allRows);
long groupedAt = System.nanoTime();
FileTaskEntity task = new FileTaskEntity();
@@ -201,7 +202,7 @@ public class AppearancePatentTaskService {
String aggregateScopeKey = buildAggregateScopeKey(sourceFiles);
String sourceScopeHash = DigestUtil.sha256Hex(aggregateScopeKey);
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), request.getApiKey(), sourceFiles, mergedHeaders, groups, allRows);
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), request.getApiKey(), sourceFiles, mergedHeaders, allRows);
long payloadBuiltAt = System.nanoTime();
String parsedPayloadPointer = storeParsedPayload(task.getId(), sourceScopeHash, parsedPayload);
long payloadStoredAt = System.nanoTime();
@@ -242,16 +243,16 @@ public class AppearancePatentTaskService {
vo.setTotalRows(totalRows);
vo.setAcceptedRows(allRows.size());
vo.setDroppedRows(droppedRows);
vo.setGroupCount(groups.size());
vo.setGroupCount(groupCount);
vo.setAiPrompt(normalize(request.getAiPrompt()));
vo.setItems(allRows);
vo.setGroups(groups);
vo.setItems(List.of());
vo.setGroups(List.of());
long finishedAt = System.nanoTime();
log.info("[appearance-patent] parse timing taskId={} files={} rows={} groups={} totalMs={} parseMs={} groupMs={} taskInsertMs={} payloadJsonMs={} payloadStoreMs={} persistMs={} responseMs={}",
task.getId(),
sourceFiles.size(),
allRows.size(),
groups.size(),
groupCount,
elapsedMs(startedAt, finishedAt),
elapsedMs(parseStartedAt, parsedAt),
elapsedMs(parsedAt, groupedAt),
@@ -322,15 +323,7 @@ public class AppearancePatentTaskService {
.map(FileResultEntity::getId)
.filter(Objects::nonNull)
.toList());
List<FileResultEntity> sortedRows = new ArrayList<>();
for (FileResultEntity row : rows) {
FileTaskEntity task = taskMap.get(row.getTaskId());
String taskStatus = task == null ? null : task.getStatus();
if (STATUS_PENDING.equals(taskStatus)) {
continue;
}
sortedRows.add(row);
}
List<FileResultEntity> sortedRows = new ArrayList<>(rows);
sortedRows.sort(Comparator
.comparingInt((FileResultEntity row) -> historyPriority(row, taskMap.get(row.getTaskId()), jobMap.get(row.getId())))
.thenComparing((FileResultEntity row) -> historyActivityTime(row, taskMap.get(row.getTaskId()), jobMap.get(row.getId())),
@@ -968,6 +961,10 @@ public class AppearancePatentTaskService {
return groups;
}
private int countParsedGroups(List<AppearancePatentParsedRowVo> rows) {
return groupRowsByBaseId(rows).size();
}
private Map<String, List<AppearancePatentParsedRowVo>> groupRowsByBaseId(List<AppearancePatentParsedRowVo> rows) {
Map<String, List<AppearancePatentParsedRowVo>> result = new LinkedHashMap<>();
if (rows == null) {
@@ -1997,7 +1994,7 @@ public class AppearancePatentTaskService {
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getDisplayId(), parsedRow.getSourceId()));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getAsin(), ""));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getCountry(), ""));
row.createCell(col++).setCellValue(readValueByHeader(parsedRow, "价格", "price"));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getPrice(), ""));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getTitle(), "") : firstNonBlank(resultRow.getTitle(), parsedRow.getTitle()));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getUrl(), "") : firstNonBlank(resultRow.getUrl(), parsedRow.getUrl()));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : userFacingCozeCellValue(resultRow, resultRow.getTitleRisk()));
@@ -2081,10 +2078,10 @@ public class AppearancePatentTaskService {
throw new BusinessException("Excel 表头为空");
}
Map<String, Integer> headerMap = buildHeaderMap(header, formatter);
List<String> headers = readHeaders(header, formatter);
int idCol = findRequiredHeader(headerMap, "id");
int asinCol = findRequiredHeader(headerMap, "asin");
int countryCol = findRequiredHeader(headerMap, "国家", "country");
int priceCol = findOptionalHeaderExact(headerMap, "价格", "price");
int urlCol = findOptionalHeaderExact(headerMap,
"url", "rul", "link", "image", "img", "pic", "picture",
"链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接");
@@ -2127,16 +2124,16 @@ public class AppearancePatentTaskService {
vo.setRowToken(buildRowToken(source.getFileKey(), vo.getRowIndex()));
vo.setAsin(asin);
vo.setCountry(country);
vo.setPrice(priceCol >= 0 ? cell(row, priceCol, formatter) : "");
vo.setUrl(urlCol >= 0 ? cell(row, urlCol, formatter) : "");
vo.setTitle(titleCol >= 0 ? cell(row, titleCol, formatter) : "");
vo.setValues(readRowValues(row, headers, formatter));
allRows.add(vo);
}
hydratePromptFields(allRows);
if (allRows.isEmpty()) {
throw new BusinessException("no valid appearance patent rows");
}
return new ParsedWorkbook(total, dropped, headers, allRows);
return new ParsedWorkbook(total, dropped, List.of(), allRows);
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
@@ -2202,23 +2199,6 @@ public class AppearancePatentTaskService {
return map;
}
private List<String> readHeaders(Row header, DataFormatter formatter) {
List<String> headers = new ArrayList<>();
for (int i = 0; i < header.getLastCellNum(); i++) {
String val = normalize(formatter.formatCellValue(header.getCell(i)));
headers.add(val.isBlank() ? "列" + (i + 1) : val);
}
return headers;
}
private Map<String, String> readRowValues(Row row, List<String> headers, DataFormatter formatter) {
Map<String, String> values = new LinkedHashMap<>();
for (int i = 0; i < headers.size(); i++) {
values.put(headers.get(i), cell(row, i, formatter));
}
return values;
}
private int findRequiredHeader(Map<String, Integer> map, String... names) {
int idx = findOptionalHeader(map, names);
if (idx < 0) {
@@ -2368,7 +2348,7 @@ public class AppearancePatentTaskService {
private int historyPriority(FileResultEntity row, FileTaskEntity task, TaskFileJobEntity job) {
String taskStatus = task == null ? null : task.getStatus();
if (STATUS_RUNNING.equals(taskStatus) || isHistoryFileBuilding(row, taskStatus, job)) {
if (STATUS_PENDING.equals(taskStatus) || STATUS_RUNNING.equals(taskStatus) || isHistoryFileBuilding(row, taskStatus, job)) {
return 0;
}
return 1;
@@ -2476,15 +2456,15 @@ public class AppearancePatentTaskService {
return val == null ? "" : val.replace(String.valueOf((char) 0xFEFF), "").replace((char) 0x3000, ' ').trim().replaceAll("\\s+", " ");
}
private String buildParsedPayloadJson(String aiPrompt, String apiKey, List<AppearancePatentSourceFileDto> sourceFiles, List<String> headers, List<AppearancePatentParsedGroupVo> groups, List<AppearancePatentParsedRowVo> allRows) {
private String buildParsedPayloadJson(String aiPrompt, String apiKey, List<AppearancePatentSourceFileDto> sourceFiles, List<String> headers, List<AppearancePatentParsedRowVo> allRows) {
AppearancePatentParsedPayloadDto payload = new AppearancePatentParsedPayloadDto();
payload.setAiPrompt(normalize(aiPrompt));
payload.setApiKey(normalize(apiKey));
payload.setSourceFiles(sourceFiles == null ? List.of() : sourceFiles);
payload.setHeaders(headers == null ? List.of() : headers);
payload.setItems(List.of());
payload.setGroups(groups == null ? List.of() : groups);
payload.setAllItems(List.of());
payload.setGroups(List.of());
payload.setAllItems(allRows == null ? List.of() : allRows);
return writeJson(payload, "保存解析结果失败");
}
@@ -2552,6 +2532,57 @@ public class AppearancePatentTaskService {
return hydrateParsedPayloadRows(new AppearancePatentParsedPayloadDto());
}
public AppearancePatentParsedPayloadDto parsedPayload(Long taskId, Long userId) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) {
throw new BusinessException("任务不存在");
}
return readParsedPayload(task);
}
public AppearancePatentParsedPayloadDto queuePayload(Long taskId, Long userId) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) {
throw new BusinessException("任务不存在");
}
AppearancePatentParsedPayloadDto payload = readParsedPayload(task);
AppearancePatentParsedPayloadDto queuePayload = new AppearancePatentParsedPayloadDto();
queuePayload.setAiPrompt(payload.getAiPrompt());
queuePayload.setApiKey(payload.getApiKey());
queuePayload.setGroups(payload.getGroups() == null ? List.of() : payload.getGroups());
queuePayload.setItems(List.of());
queuePayload.setAllItems(List.of());
queuePayload.setHeaders(List.of());
queuePayload.setSourceFiles(List.of());
return queuePayload;
}
public AppearancePatentParsedGroupPageDto parsedGroupPage(Long taskId, Long userId, Integer page, Integer pageSize) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) {
throw new BusinessException("任务不存在");
}
int safePage = Math.max(1, page == null ? 1 : page);
int safePageSize = Math.max(1, Math.min(pageSize == null ? 50 : pageSize, 200));
AppearancePatentParsedPayloadDto payload = readParsedPayload(task);
List<AppearancePatentParsedGroupVo> groups = payload.getGroups() == null ? List.of() : payload.getGroups();
int totalGroups = groups.size();
int totalRows = payload.getAllItems() == null ? 0 : payload.getAllItems().size();
int fromIndex = Math.min((safePage - 1) * safePageSize, totalGroups);
int toIndex = Math.min(fromIndex + safePageSize, totalGroups);
AppearancePatentParsedGroupPageDto vo = new AppearancePatentParsedGroupPageDto();
vo.setAiPrompt(payload.getAiPrompt());
vo.setApiKey(payload.getApiKey());
vo.setPage(safePage);
vo.setPageSize(safePageSize);
vo.setTotalGroups(totalGroups);
vo.setTotalRows(totalRows);
vo.setHasNext(toIndex < totalGroups);
vo.setGroups(fromIndex >= toIndex ? List.of() : groups.subList(fromIndex, toIndex));
return vo;
}
private AppearancePatentParsedPayloadDto hydrateParsedPayloadRows(AppearancePatentParsedPayloadDto payload) {
if (payload == null) {
return new AppearancePatentParsedPayloadDto();
@@ -2575,6 +2606,9 @@ public class AppearancePatentTaskService {
if (payload.getItems() == null || payload.getItems().isEmpty()) {
payload.setItems(rows);
}
if ((payload.getGroups() == null || payload.getGroups().isEmpty()) && !rows.isEmpty()) {
payload.setGroups(buildParsedGroups(rows));
}
return payload;
}
@@ -2656,22 +2690,6 @@ public class AppearancePatentTaskService {
return normalize(id) + "::" + normalize(asin).toUpperCase(Locale.ROOT) + "::" + normalize(country);
}
private String readValueByHeader(AppearancePatentParsedRowVo row, String... candidates) {
if (row == null || row.getValues() == null || row.getValues().isEmpty() || candidates == null) {
return "";
}
for (Map.Entry<String, String> entry : row.getValues().entrySet()) {
String header = normalize(entry.getKey()).toLowerCase(Locale.ROOT);
for (String candidate : candidates) {
String expected = normalize(candidate).toLowerCase(Locale.ROOT);
if (!expected.isBlank() && header.contains(expected)) {
return entry.getValue() == null ? "" : entry.getValue();
}
}
}
return "";
}
private String userFacingCozeCellValue(AppearancePatentResultRowDto row, String value) {
String normalizedValue = normalize(value);
if (!normalizedValue.isBlank() && !isTechnicalCozeFailure(normalizedValue)) {

197
backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java
View File

@@ -221,107 +221,113 @@ public class BrandTaskService {
}
public void submitCrawlResult(Long taskId, BrandCrawlResultRequest request) {
long startedAt = System.currentTimeMillis();
BrandCrawlTaskEntity task = requireActiveTask(taskId);
List<BrandSourceFileDto> sourceFiles = parseSourceFiles(task.getFilePaths());
if (sourceFiles.isEmpty()) {
throw new BusinessException("任务没有源文件");
if (taskId == null || taskId <= 0) {
throw new BusinessException("taskId invalid");
}
Map<String, BrandSourceFileDto> sourceByUrl = new LinkedHashMap<>();
Map<String, Integer> sourceIndexByUrl = new LinkedHashMap<>();
for (int i = 0; i < sourceFiles.size(); i++) {
BrandSourceFileDto file = sourceFiles.get(i);
sourceByUrl.put(file.getFileUrl(), file);
sourceIndexByUrl.put(file.getFileUrl(), i + 1);
}
List<BrandParsedFileCacheDto> cachedFiles = brandTaskStorageService.getParsedPayload(taskId);
if (cachedFiles == null || cachedFiles.isEmpty()) {
throw new BusinessException("任务原始数据已过期，请重新创建任务");
}
Map<String, BrandParsedFileCacheDto> cachedByUrl = new LinkedHashMap<>();
for (BrandParsedFileCacheDto cachedFile : cachedFiles) {
cachedByUrl.put(cachedFile.getFileUrl(), cachedFile);
}
List<BrandCrawlResultFileDto> resultFiles = request.getFiles() == null ? List.of() : request.getFiles();
if (resultFiles.isEmpty()) {
throw new BusinessException("files 不能为空");
}
ensureNoDuplicateResultFiles(resultFiles);
int totalCount = sourceFiles.size();
markTaskRunning(taskId, totalCount);
afterTaskStarted(taskId, totalCount);
log.info("[brand-submit] taskId={} chunks={} files={} thread={}",
taskId,
resultFiles.size(),
totalCount,
Thread.currentThread().getName());
int finishedCount = brandTaskStorageService.countCompletedFiles(taskId);
for (BrandCrawlResultFileDto resultFile : resultFiles) {
String fileUrl = blankToNull(resultFile.getFileUrl());
if (fileUrl == null) {
throw new BusinessException("fileUrl 不能为空");
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId, TaskDistributedLockService.DEFAULT_WAIT_MILLIS)) {
long startedAt = System.currentTimeMillis();
BrandCrawlTaskEntity task = requireActiveTask(taskId);
List<BrandSourceFileDto> sourceFiles = parseSourceFiles(task.getFilePaths());
if (sourceFiles.isEmpty()) {
throw new BusinessException("task source files missing");
}
BrandSourceFileDto sourceFile = sourceByUrl.get(fileUrl);
if (sourceFile == null) {
throw new BusinessException("存在未知 fileUrl: " + fileUrl);
Map<String, BrandSourceFileDto> sourceByUrl = new LinkedHashMap<>();
Map<String, Integer> sourceIndexByUrl = new LinkedHashMap<>();
for (int i = 0; i < sourceFiles.size(); i++) {
BrandSourceFileDto file = sourceFiles.get(i);
sourceByUrl.put(file.getFileUrl(), file);
sourceIndexByUrl.put(file.getFileUrl(), i + 1);
}
BrandParsedFileCacheDto cachedFile = cachedByUrl.get(fileUrl);
if (cachedFile == null) {
throw new BusinessException("缺少原始缓存数据: " + fileUrl);
List<BrandParsedFileCacheDto> cachedFiles = brandTaskStorageService.getParsedPayload(taskId);
if (cachedFiles == null || cachedFiles.isEmpty()) {
throw new BusinessException("task parsed payload missing");
}
int totalLines = cachedFile.getRows() == null ? 0 : cachedFile.getRows().size();
if (resultFile.getTotalLines() == null || resultFile.getTotalLines() <= 0) {
resultFile.setTotalLines(totalLines);
} else if (totalLines > 0 && resultFile.getTotalLines() > totalLines) {
throw new BusinessException("totalLines 超过原始文件行数: " + fileUrl);
Map<String, BrandParsedFileCacheDto> cachedByUrl = new LinkedHashMap<>();
for (BrandParsedFileCacheDto cachedFile : cachedFiles) {
cachedByUrl.put(cachedFile.getFileUrl(), cachedFile);
}
BrandTaskStorageService.ChunkStoreResult storeResult = brandTaskStorageService.storeChunk(taskId, resultFile);
BrandTaskStorageService.ChunkStoreResult refreshResult = brandTaskStorageService.refreshFileAggregate(taskId, fileUrl);
BrandFileAggregateCacheDto aggregate = refreshResult.aggregate() == null ? storeResult.aggregate() : refreshResult.aggregate();
finishedCount = Math.max(finishedCount, storeResult.finishedFiles());
finishedCount = Math.max(finishedCount, refreshResult.finishedFiles());
log.info("[brand-submit] taskId={} fileUrl={} chunk={}/{} stored={} newlyCompletedFile={} finishedFiles={} receivedChunks={} processedLines={} totalLines={}",
List<BrandCrawlResultFileDto> resultFiles = request.getFiles() == null ? List.of() : request.getFiles();
if (resultFiles.isEmpty()) {
throw new BusinessException("files is empty");
}
ensureNoDuplicateResultFiles(resultFiles);
int totalCount = sourceFiles.size();
markTaskRunning(taskId, totalCount);
afterTaskStarted(taskId, totalCount);
log.info("[brand-submit] taskId={} chunks={} files={} thread={}",
taskId,
fileUrl,
resultFile.getChunkIndex(),
resultFile.getChunkTotal(),
storeResult.stored(),
storeResult.newlyCompletedFile() || refreshResult.newlyCompletedFile(),
finishedCount,
aggregate == null ? 0 : defaultInteger(aggregate.getReceivedChunkCount()),
aggregate == null ? 0 : defaultInteger(aggregate.getProcessedLineCount()),
aggregate == null ? 0 : defaultInteger(aggregate.getTotalLines()));
String fileName = blankToDefault(sourceFile.getOriginalFilename(), sourceFile.getFileUrl());
int fileIndex = sourceIndexByUrl.getOrDefault(fileUrl, 0);
int currentLine = aggregate == null ? 0 : defaultInteger(aggregate.getProcessedLineCount());
int currentTotalLines = aggregate == null ? 0 : Math.max(defaultInteger(aggregate.getTotalLines()), currentLine);
brandTaskProgressCacheService.saveProgressFromResult(taskId,
fileUrl,
fileIndex,
resultFiles.size(),
totalCount,
fileName,
currentLine,
currentTotalLines,
finishedCount);
updateTaskProgress(taskId, finishedCount, totalCount);
saveBrandProgressSnapshot(taskId, STATUS_RUNNING, totalCount, finishedCount, 0, null);
}
Thread.currentThread().getName());
tryFinalizeTask(taskId, request.getStrategy(), sourceFiles, cachedByUrl, finishedCount, totalCount);
log.info("[brand-submit] taskId={} finishedFiles={}/{} elapsedMs={} thread={}",
taskId,
finishedCount,
totalCount,
System.currentTimeMillis() - startedAt,
Thread.currentThread().getName());
int finishedCount = brandTaskStorageService.countCompletedFiles(taskId);
for (BrandCrawlResultFileDto resultFile : resultFiles) {
String fileUrl = blankToNull(resultFile.getFileUrl());
if (fileUrl == null) {
throw new BusinessException("fileUrl is blank");
}
BrandSourceFileDto sourceFile = sourceByUrl.get(fileUrl);
if (sourceFile == null) {
throw new BusinessException("unknown fileUrl: " + fileUrl);
}
BrandParsedFileCacheDto cachedFile = cachedByUrl.get(fileUrl);
if (cachedFile == null) {
throw new BusinessException("parsed cache missing: " + fileUrl);
}
int totalLines = cachedFile.getRows() == null ? 0 : cachedFile.getRows().size();
if (resultFile.getTotalLines() == null || resultFile.getTotalLines() <= 0) {
resultFile.setTotalLines(totalLines);
} else if (totalLines > 0 && resultFile.getTotalLines() > totalLines) {
throw new BusinessException("totalLines exceeds source rows: " + fileUrl);
}
BrandTaskStorageService.ChunkStoreResult storeResult = brandTaskStorageService.storeChunk(taskId, resultFile);
BrandTaskStorageService.ChunkStoreResult refreshResult = brandTaskStorageService.refreshFileAggregate(taskId, fileUrl);
BrandFileAggregateCacheDto aggregate = refreshResult.aggregate() == null ? storeResult.aggregate() : refreshResult.aggregate();
finishedCount = Math.max(finishedCount, storeResult.finishedFiles());
finishedCount = Math.max(finishedCount, refreshResult.finishedFiles());
log.info("[brand-submit] taskId={} fileUrl={} chunk={}/{} stored={} newlyCompletedFile={} finishedFiles={} receivedChunks={} processedLines={} totalLines={}",
taskId,
fileUrl,
resultFile.getChunkIndex(),
resultFile.getChunkTotal(),
storeResult.stored(),
storeResult.newlyCompletedFile() || refreshResult.newlyCompletedFile(),
finishedCount,
aggregate == null ? 0 : defaultInteger(aggregate.getReceivedChunkCount()),
aggregate == null ? 0 : defaultInteger(aggregate.getProcessedLineCount()),
aggregate == null ? 0 : defaultInteger(aggregate.getTotalLines()));
String fileName = blankToDefault(sourceFile.getOriginalFilename(), sourceFile.getFileUrl());
int fileIndex = sourceIndexByUrl.getOrDefault(fileUrl, 0);
int currentLine = aggregate == null ? 0 : defaultInteger(aggregate.getProcessedLineCount());
int currentTotalLines = aggregate == null ? 0 : Math.max(defaultInteger(aggregate.getTotalLines()), currentLine);
brandTaskProgressCacheService.saveProgressFromResult(taskId,
fileUrl,
fileIndex,
totalCount,
fileName,
currentLine,
currentTotalLines,
finishedCount);
updateTaskProgress(taskId, finishedCount, totalCount);
saveBrandProgressSnapshot(taskId, STATUS_RUNNING, totalCount, finishedCount, 0, null);
}
tryFinalizeTask(taskId, request.getStrategy(), sourceFiles, cachedByUrl, finishedCount, totalCount);
log.info("[brand-submit] taskId={} finishedFiles={}/{} elapsedMs={} thread={}",
taskId,
finishedCount,
totalCount,
System.currentTimeMillis() - startedAt,
Thread.currentThread().getName());
}
}
public void cancelTask(Long taskId) {
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId, TaskDistributedLockService.DEFAULT_WAIT_MILLIS)) {
requireTask(taskId);
int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getId, taskId)
@@ -332,9 +338,11 @@ public class BrandTaskService {
}
brandTaskStorageService.deleteTaskData(taskId);
brandTaskProgressCacheService.delete(taskId);
}
}
public void deleteTask(Long taskId) {
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId, TaskDistributedLockService.DEFAULT_WAIT_MILLIS)) {
requireTask(taskId);
int deleted = brandCrawlTaskMapper.delete(new LambdaQueryWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getId, taskId)
@@ -344,6 +352,7 @@ public class BrandTaskService {
}
brandTaskStorageService.deleteTaskData(taskId);
brandTaskProgressCacheService.delete(taskId);
}
}
public String resolveDownloadUrl(Long taskId) {
@@ -1253,6 +1262,14 @@ public class BrandTaskService {
return lockHandle;
}
private TaskDistributedLockService.LockHandle acquireTaskLockOrThrow(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, waitMillis);
if (lockHandle == null) {
throw new BusinessException("task lock is busy");
}
return lockHandle;
}
private boolean tryRecoverCompletedRunningTask(BrandCrawlTaskEntity task) {
if (task == null || task.getId() == null) {
return false;

278
backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java
View File

@@ -35,6 +35,7 @@ import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService;
@@ -54,6 +55,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator;
@@ -71,6 +73,8 @@ public class DeleteBrandRunService {
private static final String MODULE_TYPE = "DELETE_BRAND";
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final Duration TASK_LOCK_TTL = TaskDistributedLockService.DEFAULT_LOCK_TTL;
private static final long TASK_LOCK_WAIT_MILLIS = TaskDistributedLockService.DEFAULT_WAIT_MILLIS;
private final FileTaskMapper fileTaskMapper;
private final FileResultMapper fileResultMapper;
@@ -82,6 +86,7 @@ public class DeleteBrandRunService {
private final ObjectMapper objectMapper;
private final DeleteBrandProgressProperties deleteBrandProgressProperties;
private final TaskPressureProperties taskPressureProperties;
private final TaskDistributedLockService taskDistributedLockService;
private final TaskFileJobService taskFileJobService;
private boolean isUsableIndexedMatch(ZiniaoShopMatchResultVo matchResult) {
@@ -493,13 +498,15 @@ public class DeleteBrandRunService {
throw new BusinessException("记录不存在");
}
Long taskId = entity.getTaskId();
fileResultMapper.deleteById(resultId);
if (taskId != null && taskId > 0) {
Long remaining = fileResultMapper.selectCount(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
if (remaining == null || remaining <= 0) {
deleteBrandTaskStorageService.deleteTaskData(taskId);
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
fileResultMapper.deleteById(resultId);
if (taskId != null && taskId > 0) {
Long remaining = fileResultMapper.selectCount(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
if (remaining == null || remaining <= 0) {
deleteBrandTaskStorageService.deleteTaskData(taskId);
}
}
}
}
@@ -1249,99 +1256,96 @@ public class DeleteBrandRunService {
@Transactional
public void submitResult(Long taskId, DeleteBrandSubmitResultRequest request) {
log.info("[DeleteBrand] Received submitResult for taskId: {}, files count: {}", taskId, request == null || request.getFiles() == null ? 0 : request.getFiles().size());
if (taskId == null || taskId <= 0) {
throw new BusinessException("taskId 不合法");
}
if (request == null || request.getFiles() == null || request.getFiles().isEmpty()) {
throw new BusinessException("files 不能为空");
throw new BusinessException("taskId invalid");
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId, TASK_LOCK_WAIT_MILLIS)) {
log.info("[DeleteBrand] Received submitResult for taskId: {}, files count: {}", taskId, request == null || request.getFiles() == null ? 0 : request.getFiles().size());
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[DeleteBrand] submitResult rejected because task is terminal -> taskId: {}, status: {}, createdAt: {}, updatedAt: {}, finishedAt: {}, error: {}, sourceFiles: {}, successFiles: {}, failedFiles: {}",
task.getId(),
task.getStatus(),
task.getCreatedAt(),
task.getUpdatedAt(),
task.getFinishedAt(),
task.getErrorMessage(),
task.getSourceFileCount(),
task.getSuccessFileCount(),
task.getFailedFileCount());
throw new BusinessException("任务已结束，拒绝继续提交结果");
}
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload = deleteBrandTaskStorageService.loadParsedPayload(taskId);
if (parsedPayload == null || parsedPayload.isEmpty()) {
throw new BusinessException("任务原始数据已过期，请重新执行解析");
}
if (!"RUNNING".equals(task.getStatus())) {
task.setStatus("RUNNING");
task.setErrorMessage(null);
task.setUpdatedAt(LocalDateTime.now());
task.setFinishedAt(null);
fileTaskMapper.updateById(task);
deleteBrandTaskCacheService.saveTaskCache(task);
}
boolean shouldTryFinalize = false;
for (DeleteBrandResultFileDto fileDto : request.getFiles()) {
log.info("[DeleteBrand] submitResult processing chunk -> taskId: {}, file: {}, chunkIndex: {}/{}, processedRows: {}/{}",
taskId, fileDto.getSourceFilename(), fileDto.getChunkIndex(), fileDto.getChunkTotal(),
fileDto.getProcessedRows(), fileDto.getTotalRows());
String fileIdentity = resolveFileIdentity(fileDto);
if (fileIdentity.isBlank()) {
throw new BusinessException("fileKey/sourceFilename 不能为空");
}
DeleteBrandParsedFileCacheDto parsedFile = parsedPayload.get(fileIdentity);
if (parsedFile == null) {
log.warn("[DeleteBrand] parsed payload not found for identity: {}", fileIdentity);
throw new BusinessException("文件标识不匹配: " + fileIdentity);
if (request == null || request.getFiles() == null || request.getFiles().isEmpty()) {
throw new BusinessException("files is empty");
}
// 更早执行 duplicate fast-path，尽量在较重校验和进度写入前直接短路
validateChunk(fileDto, parsedFile);
validateResultFileAgainstParsed(fileDto, parsedFile);
boolean stored = deleteBrandTaskStorageService.storeResultChunkIfChanged(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto);
if (!stored) {
log.info("[DeleteBrand] Duplicate chunk ignored for taskId: {}, file: {}, chunkIndex: {}", taskId, fileIdentity, fileDto.getChunkIndex());
continue;
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("task not found");
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[DeleteBrand] submitResult rejected because task is terminal -> taskId: {}, status: {}, createdAt: {}, updatedAt: {}, finishedAt: {}, error: {}, sourceFiles: {}, successFiles: {}, failedFiles: {}",
task.getId(),
task.getStatus(),
task.getCreatedAt(),
task.getUpdatedAt(),
task.getFinishedAt(),
task.getErrorMessage(),
task.getSourceFileCount(),
task.getSuccessFileCount(),
task.getFailedFileCount());
throw new BusinessException("task already finished");
}
Map<String, String> progress = new LinkedHashMap<>();
progress.put("phase", DeleteBrandTaskCacheService.PHASE_CRAWLING);
progress.put("file_index", String.valueOf(defaultInt(fileDto.getFileIndex(), 0)));
progress.put("file_total", String.valueOf(defaultInt(fileDto.getFileTotal(), parsedPayload.size())));
progress.put("file_name", blankToEmpty(fileDto.getSourceFilename()));
progress.put("current_line", String.valueOf(defaultInt(fileDto.getProcessedRows(), 0)));
progress.put("total_lines", String.valueOf(defaultInt(fileDto.getTotalRows(), parsedFile.getTotalRows() == null ? 0 : parsedFile.getTotalRows())));
progress.put("current_country", blankToEmpty(fileDto.getCurrentCountry()));
progress.put("current_asin", blankToEmpty(fileDto.getCurrentAsin()));
progress.put("updated_at", String.valueOf(System.currentTimeMillis()));
progress.put("last_heartbeat_at", String.valueOf(System.currentTimeMillis()));
deleteBrandTaskCacheService.saveProgress(taskId, progress, true);
// 如果该文件已完成，立即更新成品计数的 Redis 提示，让后续 tryFinalizeTask 更准确
if (isFileCompleted(fileDto)) {
log.info("[DeleteBrand] File chunk transmission fully reached via chunkTotal for taskId: {}, file: {}", taskId, fileDto.getSourceFilename());
enqueueCompletedFileAssembleJob(task, fileIdentity, parsedFile);
shouldTryFinalize = true;
// 这里暂不直接 increment，而是标记需要在 tryFinalizeTask 中重新计算
// 为了保险，直接在 tryFinalizeTask 里重新遍历分片状态
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload = deleteBrandTaskStorageService.loadParsedPayload(taskId);
if (parsedPayload == null || parsedPayload.isEmpty()) {
throw new BusinessException("task parsed payload missing");
}
}
if (!"RUNNING".equals(task.getStatus())) {
task.setStatus("RUNNING");
task.setErrorMessage(null);
task.setUpdatedAt(LocalDateTime.now());
task.setFinishedAt(null);
fileTaskMapper.updateById(task);
deleteBrandTaskCacheService.saveTaskCache(task);
}
if (shouldTryFinalize) {
tryFinalizeTask(taskId, false);
boolean shouldTryFinalize = false;
for (DeleteBrandResultFileDto fileDto : request.getFiles()) {
log.info("[DeleteBrand] submitResult processing chunk -> taskId: {}, file: {}, chunkIndex: {}/{}, processedRows: {}/{}",
taskId, fileDto.getSourceFilename(), fileDto.getChunkIndex(), fileDto.getChunkTotal(),
fileDto.getProcessedRows(), fileDto.getTotalRows());
String fileIdentity = resolveFileIdentity(fileDto);
if (fileIdentity.isBlank()) {
throw new BusinessException("file identity is blank");
}
DeleteBrandParsedFileCacheDto parsedFile = parsedPayload.get(fileIdentity);
if (parsedFile == null) {
log.warn("[DeleteBrand] parsed payload not found for identity: {}", fileIdentity);
throw new BusinessException("parsed file not found: " + fileIdentity);
}
validateChunk(fileDto, parsedFile);
validateResultFileAgainstParsed(fileDto, parsedFile);
boolean stored = deleteBrandTaskStorageService.storeResultChunkIfChanged(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto);
if (!stored) {
log.info("[DeleteBrand] Duplicate chunk ignored for taskId: {}, file: {}, chunkIndex: {}", taskId, fileIdentity, fileDto.getChunkIndex());
continue;
}
Map<String, String> progress = new LinkedHashMap<>();
progress.put("phase", DeleteBrandTaskCacheService.PHASE_CRAWLING);
progress.put("file_index", String.valueOf(defaultInt(fileDto.getFileIndex(), 0)));
progress.put("file_total", String.valueOf(defaultInt(fileDto.getFileTotal(), parsedPayload.size())));
progress.put("file_name", blankToEmpty(fileDto.getSourceFilename()));
progress.put("current_line", String.valueOf(defaultInt(fileDto.getProcessedRows(), 0)));
progress.put("total_lines", String.valueOf(defaultInt(fileDto.getTotalRows(), parsedFile.getTotalRows() == null ? 0 : parsedFile.getTotalRows())));
progress.put("current_country", blankToEmpty(fileDto.getCurrentCountry()));
progress.put("current_asin", blankToEmpty(fileDto.getCurrentAsin()));
progress.put("updated_at", String.valueOf(System.currentTimeMillis()));
progress.put("last_heartbeat_at", String.valueOf(System.currentTimeMillis()));
deleteBrandTaskCacheService.saveProgress(taskId, progress, true);
if (isFileCompleted(fileDto)) {
log.info("[DeleteBrand] File chunk transmission fully reached via chunkTotal for taskId: {}, file: {}", taskId, fileDto.getSourceFilename());
enqueueCompletedFileAssembleJob(task, fileIdentity, parsedFile);
shouldTryFinalize = true;
}
}
if (shouldTryFinalize) {
tryFinalizeTask(taskId, false);
}
}
}
@@ -1398,44 +1402,45 @@ public class DeleteBrandRunService {
if (taskId == null || taskId <= 0) {
return;
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId, 0L)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return;
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
return;
}
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return;
int expectedFiles = deleteBrandTaskStorageService.countParsedPayloadScopes(taskId);
if (expectedFiles <= 0) {
tryFinalizeTaskFromTerminalResults(task);
return;
}
int finishedFiles = deleteBrandTaskStorageService.countCompletedScopes(taskId);
if (finishedFiles < expectedFiles) {
updateTaskFinalizeProgress(task, finishedFiles, fromCompensation);
log.info("[DeleteBrand] tryFinalizeTask skipping -> finishedFiles ({}) < expectedFiles ({})", finishedFiles, expectedFiles);
return;
}
updateTaskFinalizeProgress(task, finishedFiles, true);
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload = deleteBrandTaskStorageService.loadParsedPayload(taskId);
if (parsedPayload == null || parsedPayload.isEmpty()) {
tryFinalizeTaskFromTerminalResults(task);
return;
}
Map<String, List<DeleteBrandResultFileDto>> mergedByFile = deleteBrandTaskStorageService.loadMergedChunks(taskId);
log.info("[DeleteBrand] tryFinalizeTask calling finalizeTask -> finishedFiles: {}, expectedFiles: {}", finishedFiles, expectedFiles);
finalizeTask(task, parsedPayload, mergedByFile);
} catch (BusinessException ex) {
if (ex.getMessage() != null && ex.getMessage().contains("task lock is busy")) {
log.info("[DeleteBrand] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return;
}
throw ex;
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
return;
}
int expectedFiles = deleteBrandTaskStorageService.countParsedPayloadScopes(taskId);
if (expectedFiles <= 0) {
tryFinalizeTaskFromTerminalResults(task);
return;
}
// 移除 finishedFilesHint 的强依赖，因为 Redis 里的 finished_files 可能是旧值
// 既然进入 tryFinalizeTask，就说明有新分片到达，应该以 loadMergedChunks 为准
int finishedFiles = deleteBrandTaskStorageService.countCompletedScopes(taskId);
if (finishedFiles < expectedFiles) {
updateTaskFinalizeProgress(task, finishedFiles, fromCompensation);
log.info("[DeleteBrand] tryFinalizeTask skipping -> finishedFiles ({}) < expectedFiles ({})", finishedFiles, expectedFiles);
return;
}
updateTaskFinalizeProgress(task, finishedFiles, true);
Map<String, DeleteBrandParsedFileCacheDto> parsedPayload = deleteBrandTaskStorageService.loadParsedPayload(taskId);
if (parsedPayload == null || parsedPayload.isEmpty()) {
tryFinalizeTaskFromTerminalResults(task);
return;
}
Map<String, List<DeleteBrandResultFileDto>> mergedByFile = deleteBrandTaskStorageService.loadMergedChunks(taskId);
log.info("[DeleteBrand] tryFinalizeTask calling finalizeTask -> finishedFiles: {}, expectedFiles: {}", finishedFiles, expectedFiles);
finalizeTask(task, parsedPayload, mergedByFile);
}
private void updateTaskFinalizeProgress(FileTaskEntity task, int finishedFiles, boolean fromCompensation) {
@@ -1974,6 +1979,27 @@ public class DeleteBrandRunService {
String normalized = blankToEmpty(value);
return normalized.isBlank() ? defaultValue : normalized;
}
private TaskDistributedLockService.LockHandle acquireTaskLockOrThrow(Long taskId) {
return acquireTaskLockOrThrow(taskId, TASK_LOCK_WAIT_MILLIS);
}
private TaskDistributedLockService.LockHandle acquireTaskLockOrThrow(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, waitMillis);
if (lockHandle == null) {
throw new BusinessException("task lock is busy");
}
return lockHandle;
}
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle =
taskDistributedLockService.acquire(MODULE_TYPE, taskId, TASK_LOCK_TTL, waitMillis);
if (lockHandle == null) {
log.info("[DeleteBrand] task lock busy taskId={} waitMillis={}", taskId, waitMillis);
}
return lockHandle;
}
/**
* 落库用 result_json：去掉国家分组、预览行等大字段，保留匹配摘要与计数。
* 索引命中后单次任务 JSON 体积会明显膨胀，容易触发 max_allowed_packet 或更新失败。

95
backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java
View File

@@ -5,6 +5,8 @@ import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.DeleteBrandProgressProperties;
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskCacheService;
import com.nanri.aiimage.modules.queryasin.service.QueryAsinTaskCacheService;
import com.nanri.aiimage.modules.queryasin.service.QueryAsinTaskService;
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService;
import com.nanri.aiimage.modules.patroldelete.service.PatrolDeleteTaskCacheService;
import com.nanri.aiimage.modules.patroldelete.service.PatrolDeleteTaskService;
@@ -42,6 +44,7 @@ public class DeleteBrandStaleTaskService {
private static final String MODULE_TYPE_PRICE_TRACK = "PRICE_TRACK";
private static final String MODULE_TYPE_SHOP_MATCH = "SHOP_MATCH";
private static final String MODULE_TYPE_PATROL_DELETE = "PATROL_DELETE";
private static final String MODULE_TYPE_QUERY_ASIN = "QUERY_ASIN";
private static final Duration STALE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
private static final Duration FINALIZE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
private static final Duration TEMP_DIR_CLEANUP_LOCK_TTL = Duration.ofMinutes(10);
@@ -58,6 +61,8 @@ public class DeleteBrandStaleTaskService {
private final ShopMatchTaskCacheService shopMatchTaskCacheService;
private final PatrolDeleteTaskService patrolDeleteTaskService;
private final PatrolDeleteTaskCacheService patrolDeleteTaskCacheService;
private final QueryAsinTaskService queryAsinTaskService;
private final QueryAsinTaskCacheService queryAsinTaskCacheService;
private final DeleteBrandProgressProperties deleteBrandProgressProperties;
private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService;
@@ -82,6 +87,7 @@ public class DeleteBrandStaleTaskService {
PriceTrackStaleCheckStats priceTrackStats = failStalePriceTrackTasks();
ShopMatchStaleCheckStats shopMatchStats = failStaleShopMatchTasks();
ShopMatchStaleCheckStats patrolDeleteStats = failStalePatrolDeleteTasks();
ShopMatchStaleCheckStats queryAsinStats = failStaleQueryAsinTasks();
log.info("[stale-check] product-risk summary scanned={} finalized={} failed={} skipped={} elapsedMs={} thread={}",
stats.scannedTaskCount,
stats.finalizedTaskCount,
@@ -110,6 +116,13 @@ public class DeleteBrandStaleTaskService {
patrolDeleteStats.skippedTaskCount,
System.currentTimeMillis() - startedAt,
Thread.currentThread().getName());
log.info("[stale-check] query-asin summary scanned={} finalized={} failed={} skipped={} elapsedMs={} thread={}",
queryAsinStats.scannedTaskCount,
queryAsinStats.finalizedTaskCount,
queryAsinStats.failedTaskCount,
queryAsinStats.skippedTaskCount,
System.currentTimeMillis() - startedAt,
Thread.currentThread().getName());
}
}
@@ -525,6 +538,85 @@ public class DeleteBrandStaleTaskService {
return stats;
}
private ShopMatchStaleCheckStats failStaleQueryAsinTasks() {
ShopMatchStaleCheckStats stats = new ShopMatchStaleCheckStats();
long minutes = Math.max(1L, deleteBrandProgressProperties.getPatrolDeleteStaleTimeoutMinutes());
long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getPatrolDeleteInitialTimeoutMinutes());
long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis();
long nowMillis = System.currentTimeMillis();
LocalDateTime now = LocalDateTime.now();
LocalDateTime threshold = now.minusMinutes(minutes);
LocalDateTime initialThreshold = now.minusMinutes(initialMinutes);
java.util.List<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_QUERY_ASIN)
.eq(FileTaskEntity::getStatus, "RUNNING")
.lt(FileTaskEntity::getUpdatedAt, threshold)
.last("limit 200"));
stats.scannedTaskCount = runningTasks.size();
if (runningTasks.isEmpty()) {
return stats;
}
Map<Long, Long> heartbeatByTaskId = queryAsinTaskCacheService.getTaskHeartbeatMillisBatch(
runningTasks.stream().map(FileTaskEntity::getId).toList());
for (FileTaskEntity task : runningTasks) {
long lastHeartbeatMillis = heartbeatByTaskId.getOrDefault(task.getId(), 0L);
boolean hasFinishedRows = (task.getSuccessFileCount() != null && task.getSuccessFileCount() > 0)
|| (task.getFailedFileCount() != null && task.getFailedFileCount() > 0);
if (lastHeartbeatMillis > 0L && nowMillis - lastHeartbeatMillis < staleTimeoutMillis) {
stats.skippedTaskCount++;
continue;
}
boolean hasStartedProgress = hasFinishedRows;
if (!hasStartedProgress) {
hasStartedProgress = queryAsinTaskCacheService.hasAnyShopMergedPayload(task.getId());
}
if (hasPendingAssembleJobs(task.getId(), MODULE_TYPE_QUERY_ASIN)) {
stats.skippedTaskCount++;
log.info("[stale-check] query-asin skip pending-assemble-jobs taskId={} unfinishedJobs={} activeJobs={}",
task.getId(),
taskFileJobService.countUnfinishedAssembleJobs(task.getId(), MODULE_TYPE_QUERY_ASIN),
taskFileJobService.countActiveAssembleJobs(task.getId(), MODULE_TYPE_QUERY_ASIN));
continue;
}
if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) {
stats.skippedTaskCount++;
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_QUERY_ASIN, task.getId());
if (taskLockHandle == null) {
stats.skippedTaskCount++;
continue;
}
try (taskLockHandle) {
try {
if (queryAsinTaskService.tryFinalizeTask(task.getId(), true)) {
stats.finalizedTaskCount++;
log.info("[stale-check] query-asin finalized taskId={}", task.getId());
continue;
}
} catch (Exception ex) {
log.warn("[stale-check] query-asin finalize threw taskId={} msg={}", task.getId(), ex.getMessage());
}
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, task.getId())
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_QUERY_ASIN)
.eq(FileTaskEntity::getStatus, "RUNNING")
.set(FileTaskEntity::getStatus, "FAILED")
.set(FileTaskEntity::getErrorMessage, "闀挎椂闂存湭鏀跺埌 Python 缁撴灉鍥炰紶锛屼换鍔″凡鑷姩澶辫触")
.set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
if (updated > 0) {
stats.failedTaskCount++;
queryAsinTaskCacheService.deleteTaskCache(task.getId());
log.warn("[stale-check] query-asin failed taskId={} reason=timeout", task.getId());
} else {
stats.skippedTaskCount++;
}
}
}
return stats;
}
private boolean isWithinInitialGrace(FileTaskEntity task, LocalDateTime initialThreshold) {
if (task == null || initialThreshold == null) {
return false;
@@ -595,7 +687,8 @@ public class DeleteBrandStaleTaskService {
"patrol-delete-result",
"delete-brand-result",
"product-risk-result",
"price-track-result"
"price-track-result",
"query-asin-result"
);
for (String dirName : dirNames) {
cleanupTempRoot(dirName, cutoff);

11
backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteTaskService.java
View File

@@ -330,6 +330,7 @@ public class PatrolDeleteTaskService {
throw new BusinessException("shops 不能为空");
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
log.info("[patrol-delete] ignore result callback for deleted task taskId={}", taskId);
@@ -373,12 +374,19 @@ public class PatrolDeleteTaskService {
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
tryFinalizeTask(taskId, false);
}
}
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) {
if (taskId == null || taskId <= 0) {
return false;
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId);
if (lockHandle == null) {
log.info("[patrol-delete] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return false;
}
try (lockHandle) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false;
@@ -449,6 +457,7 @@ public class PatrolDeleteTaskService {
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
return changed;
}
}
@Transactional
@@ -459,6 +468,7 @@ public class PatrolDeleteTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
long taskLoadedAt = System.nanoTime();
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
@@ -479,6 +489,7 @@ public class PatrolDeleteTaskService {
elapsedMs(resultsDeletedAt, auxiliaryDeletedAt),
elapsedMs(auxiliaryDeletedAt, taskDeletedAt),
elapsedMs(taskDeletedAt, finishedAt));
}
}
@Transactional

307
backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackTaskService.java
View File

@@ -258,16 +258,18 @@ public class PriceTrackTaskService {
@Transactional
public void deleteTask(Long taskId, Long userId) {
if (userId == null || userId <= 0) throw new BusinessException("user_id 不合法");
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
}
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
priceTrackTaskCacheService.deleteTaskCache(taskId);
priceTrackLoopRunService.syncLoopRunAfterChildRemoved(taskId);
}
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
priceTrackTaskCacheService.deleteTaskCache(taskId);
priceTrackLoopRunService.syncLoopRunAfterChildRemoved(taskId);
}
@Transactional
@@ -458,154 +460,163 @@ public class PriceTrackTaskService {
if (request == null || request.getShops() == null || request.getShops().isEmpty()) {
throw new BusinessException("shops 不能为空");
}
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[price-track] submitResult rejected taskId={} status={}", taskId, task.getStatus());
throw new BusinessException(40901, "任务已结束，拒绝重复提交");
}
priceTrackTaskCacheService.touchTaskHeartbeat(taskId);
List<FileResultEntity> resultRows = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
Map<String, PriceTrackSubmitResultRequest.ShopResult> payloadByShop = new LinkedHashMap<>();
for (PriceTrackSubmitResultRequest.ShopResult sr : request.getShops()) {
if (sr == null) continue;
String key = ziniaoShopSwitchService.normalizeShopName(sr.getShopName());
if (!key.isBlank()) payloadByShop.put(key, sr);
}
log.info("[price-track] submitResult received taskId={} payloadShopCount={} payloadKeys={}",
taskId, payloadByShop.size(), payloadByShop.keySet());
boolean changed = false;
for (FileResultEntity fr : resultRows) {
String shopKey = fr.getSourceFilename();
PriceTrackSubmitResultRequest.ShopResult payload = shopKey == null ? null : payloadByShop.get(shopKey);
if (payload == null) {
continue;
}
changed = true;
if (payload.getError() != null && !payload.getError().isBlank()) {
markResultFailed(fr, payload.getError());
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
if (Boolean.FALSE.equals(payload.getSuccess())) {
markResultFailed(fr, "shop processing failed");
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
handleSkipAsinDeletionSignals(shopKey, payload);
PriceTrackSubmitResultRequest.ShopResult merged = mergeShopPayload(taskId, shopKey, payload);
if (!Boolean.TRUE.equals(merged.getSuccess())) {
continue;
}
if (countPayloadRows(merged) <= 0) {
markResultFailed(fr, "no usable price-track rows received");
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
try {
enqueueResultFileAssembly(fr, shopKey, merged);
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
} catch (Exception ex) {
markResultFailed(fr, ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage());
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
}
}
if (!changed) {
return;
}
List<FileResultEntity> latest = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest);
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[price-track] submitResult rejected taskId={} status={}", taskId, task.getStatus());
throw new BusinessException(40901, "任务已结束，拒绝重复提交");
}
priceTrackTaskCacheService.touchTaskHeartbeat(taskId);
List<FileResultEntity> resultRows = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
Map<String, PriceTrackSubmitResultRequest.ShopResult> payloadByShop = new LinkedHashMap<>();
for (PriceTrackSubmitResultRequest.ShopResult sr : request.getShops()) {
if (sr == null) continue;
String key = ziniaoShopSwitchService.normalizeShopName(sr.getShopName());
if (!key.isBlank()) payloadByShop.put(key, sr);
}
log.info("[price-track] submitResult received taskId={} payloadShopCount={} payloadKeys={}",
taskId, payloadByShop.size(), payloadByShop.keySet());
boolean changed = false;
for (FileResultEntity fr : resultRows) {
String shopKey = fr.getSourceFilename();
PriceTrackSubmitResultRequest.ShopResult payload = shopKey == null ? null : payloadByShop.get(shopKey);
if (payload == null) {
continue;
}
changed = true;
if (payload.getError() != null && !payload.getError().isBlank()) {
markResultFailed(fr, payload.getError());
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
if (Boolean.FALSE.equals(payload.getSuccess())) {
markResultFailed(fr, "shop processing failed");
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
handleSkipAsinDeletionSignals(shopKey, payload);
PriceTrackSubmitResultRequest.ShopResult merged = mergeShopPayload(taskId, shopKey, payload);
if (!Boolean.TRUE.equals(merged.getSuccess())) {
continue;
}
if (countPayloadRows(merged) <= 0) {
markResultFailed(fr, "no usable price-track rows received");
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
try {
enqueueResultFileAssembly(fr, shopKey, merged);
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
} catch (Exception ex) {
markResultFailed(fr, ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage());
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
}
}
if (!changed) {
return;
}
List<FileResultEntity> latest = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest);
}
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) {
if (taskId == null || taskId <= 0) return false;
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) return false;
if (!"RUNNING".equals(task.getStatus())) {
log.warn("[price-track] stale finalize skipped taskId={} status={} fromCompensation={}",
taskId, task.getStatus(), fromCompensation);
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId);
if (lockHandle == null) {
log.info("[price-track] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return false;
}
List<FileResultEntity> latest = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (latest.isEmpty()) {
task.setStatus("FAILED");
task.setErrorMessage("task details missing, cannot finalize automatically");
task.setUpdatedAt(LocalDateTime.now());
task.setFinishedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
try (lockHandle) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) return false;
if (!"RUNNING".equals(task.getStatus())) {
log.warn("[price-track] stale finalize skipped taskId={} status={} fromCompensation={}",
taskId, task.getStatus(), fromCompensation);
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
return false;
}
List<FileResultEntity> latest = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (latest.isEmpty()) {
task.setStatus("FAILED");
task.setErrorMessage("task details missing, cannot finalize automatically");
task.setUpdatedAt(LocalDateTime.now());
task.setFinishedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
return true;
}
Map<String, PriceTrackSubmitResultRequest.ShopResult> cachedPayloadByShop =
priceTrackTaskCacheService.getAllShopMergedPayload(taskId);
boolean changed = false;
for (FileResultEntity fr : latest) {
boolean success = fr.getSuccess() != null && fr.getSuccess() == 1;
boolean failed = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank();
if (success || failed) continue;
String shopKey = fr.getSourceFilename();
PriceTrackSubmitResultRequest.ShopResult cachedPayload = shopKey == null ? null : cachedPayloadByShop.get(shopKey);
if (cachedPayload == null) {
markResultFailed(fr, "Python interrupted before this shop finished uploading results");
changed = true;
log.warn("[price-track] stale finalize failed without cache taskId={} shop={} fromCompensation={}",
taskId, shopKey, fromCompensation);
continue;
}
if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) {
markResultFailed(fr, cachedPayload.getError());
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
continue;
}
if (countPayloadRows(cachedPayload) <= 0) {
markResultFailed(fr, "no usable price-track rows received before interruption");
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[price-track] stale finalize finished without data taskId={} shop={} fromCompensation={}",
taskId, shopKey, fromCompensation);
continue;
}
try {
enqueueResultFileAssembly(fr, shopKey, cachedPayload);
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[price-track] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={} rowCount={}",
taskId, shopKey, fromCompensation, Boolean.TRUE.equals(cachedPayload.getSuccess()), countPayloadRows(cachedPayload));
} catch (Exception ex) {
String message = ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage();
markResultFailed(fr, message);
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[price-track] stale finalize assemble failed taskId={} shop={} msg={}",
taskId, shopKey, message);
}
}
if (!changed && cachedPayloadByShop.isEmpty()) {
return false;
}
List<FileResultEntity> refreshed = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, refreshed);
return true;
}
Map<String, PriceTrackSubmitResultRequest.ShopResult> cachedPayloadByShop =
priceTrackTaskCacheService.getAllShopMergedPayload(taskId);
boolean changed = false;
for (FileResultEntity fr : latest) {
boolean success = fr.getSuccess() != null && fr.getSuccess() == 1;
boolean failed = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank();
if (success || failed) continue;
String shopKey = fr.getSourceFilename();
PriceTrackSubmitResultRequest.ShopResult cachedPayload = shopKey == null ? null : cachedPayloadByShop.get(shopKey);
if (cachedPayload == null) {
markResultFailed(fr, "Python interrupted before this shop finished uploading results");
changed = true;
log.warn("[price-track] stale finalize failed without cache taskId={} shop={} fromCompensation={}",
taskId, shopKey, fromCompensation);
continue;
}
if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) {
markResultFailed(fr, cachedPayload.getError());
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
continue;
}
if (countPayloadRows(cachedPayload) <= 0) {
markResultFailed(fr, "no usable price-track rows received before interruption");
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[price-track] stale finalize finished without data taskId={} shop={} fromCompensation={}",
taskId, shopKey, fromCompensation);
continue;
}
try {
enqueueResultFileAssembly(fr, shopKey, cachedPayload);
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[price-track] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={} rowCount={}",
taskId, shopKey, fromCompensation, Boolean.TRUE.equals(cachedPayload.getSuccess()), countPayloadRows(cachedPayload));
} catch (Exception ex) {
String message = ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage();
markResultFailed(fr, message);
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[price-track] stale finalize assemble failed taskId={} shop={} msg={}",
taskId, shopKey, message);
}
}
if (!changed && cachedPayloadByShop.isEmpty()) {
return false;
}
List<FileResultEntity> refreshed = fileResultMapper.selectList(
new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, refreshed);
return true;
}
public String resolveResultDownloadUrl(Long resultId, Long userId) {

301
backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java
View File

@@ -267,15 +267,17 @@ public class ProductRiskTaskService {
if (userId == null || userId <= 0) {
throw new BusinessException("user_id 不合法");
}
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
}
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
cleanupTaskCacheIfTerminal(taskId, "DELETE_EMPTY");
}
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
cleanupTaskCacheIfTerminal(taskId, "DELETE_EMPTY");
}
/**
@@ -550,146 +552,150 @@ public class ProductRiskTaskService {
if (request == null || request.getShops() == null || request.getShops().isEmpty()) {
throw new BusinessException("shops 不能为空");
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[product-risk] submitResult rejected taskId={} status={} payloadShopCount={}",
taskId, task.getStatus(), request.getShops() == null ? 0 : request.getShops().size());
throw new BusinessException(40901, "任务已结束，拒绝重复提交");
}
String oldStatus = task.getStatus();
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
List<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
Map<String, ProductRiskShopPayloadDto> payloadByShop = new LinkedHashMap<>();
for (ProductRiskShopPayloadDto p : request.getShops()) {
if (p == null) {
continue;
}
String key = ziniaoShopSwitchService.normalizeShopName(p.getShopName());
if (!key.isBlank()) {
payloadByShop.put(key, p);
}
}
log.info("[product-risk] submitResult received taskId={} payloadShopCount={} payloadKeys={}",
taskId, payloadByShop.size(), payloadByShop.keySet());
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "product-risk-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
int matchedShopCount = 0;
int skippedUnmatchedCount = 0;
int waitingCount = 0;
int assembledCount = 0;
int fallbackMatchedCount = 0;
for (FileResultEntity fr : resultRows) {
String shopKey = fr.getSourceFilename();
ProductRiskShopPayloadDto payload = shopKey == null ? null : payloadByShop.get(shopKey);
if (payload == null && resultRows.size() == 1 && payloadByShop.size() == 1) {
payload = payloadByShop.values().iterator().next();
fallbackMatchedCount++;
log.warn("[product-risk] single-shop fallback matched taskId={} dbShop={} payloadShop={}",
taskId, shopKey, payload.getShopName());
}
if (payload == null) {
skippedUnmatchedCount++;
continue;
}
matchedShopCount++;
if (payload.getError() != null && !payload.getError().isBlank()) {
markResultFailed(fr, payload.getError());
batchErrors.add(shopKey + ": " + payload.getError());
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
ProductRiskShopPayloadDto mergedPayload = mergeShopPayload(taskId, shopKey, payload);
boolean incomingShopDone = hasShopDoneSignal(payload);
int mergedRows = countPayloadRows(mergedPayload);
boolean shopDone = isShopPayloadCompleted(mergedPayload);
log.info("[product-risk] shop merged taskId={} shop={} incomingShopDone={} mergedDone={} mergedRows={} payloadCountries={} mergedShopDoneFlag={}",
taskId,
shopKey,
incomingShopDone,
shopDone,
mergedRows,
mergedPayload.getCountries() == null ? 0 : mergedPayload.getCountries().size(),
mergedPayload.getShopDone());
if (incomingShopDone && mergedRows <= 0) {
markResultNoData(fr, "没有数据");
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
log.info("[product-risk] shop finished without data taskId={} shop={}", taskId, shopKey);
continue;
}
if (!shopDone) {
waitingCount++;
continue;
}
try {
enqueueResultFileAssembly(fr, shopKey, mergedPayload);
assembledCount++;
log.info("[product-risk] shop assembled taskId={} shop={} rowCount={} file={}",
taskId, shopKey, fr.getRowCount(), fr.getResultFilename());
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
} catch (Exception ex) {
log.warn("[product-risk] shop assemble failed taskId={} shop={} msg={}", taskId, shopKey, ex.getMessage());
markResultFailed(fr, ex.getMessage() == null ? "组装失败" : ex.getMessage());
batchErrors.add(shopKey + ": " + ex.getMessage());
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
}
}
List<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
log.warn("[product-risk] submitResult status evaluated taskId={} oldStatus={} newStatus={} matchedShopCount={} skippedUnmatchedCount={} waitingCount={} assembledCount={} fallbackMatchedCount={} batchErrors={}",
taskId, oldStatus, task.getStatus(), matchedShopCount, skippedUnmatchedCount,
waitingCount, assembledCount, fallbackMatchedCount, batchErrors);
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[product-risk] submitResult rejected taskId={} status={} payloadShopCount={}",
taskId, task.getStatus(), request.getShops() == null ? 0 : request.getShops().size());
throw new BusinessException(40901, "任务已结束，拒绝重复提交");
}
String oldStatus = task.getStatus();
List<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
Map<String, ProductRiskShopPayloadDto> payloadByShop = new LinkedHashMap<>();
for (ProductRiskShopPayloadDto p : request.getShops()) {
if (p == null) {
continue;
}
String key = ziniaoShopSwitchService.normalizeShopName(p.getShopName());
if (!key.isBlank()) {
payloadByShop.put(key, p);
}
}
log.info("[product-risk] submitResult received taskId={} payloadShopCount={} payloadKeys={}",
taskId, payloadByShop.size(), payloadByShop.keySet());
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "product-risk-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
int matchedShopCount = 0;
int skippedUnmatchedCount = 0;
int waitingCount = 0;
int assembledCount = 0;
int fallbackMatchedCount = 0;
for (FileResultEntity fr : resultRows) {
String shopKey = fr.getSourceFilename();
ProductRiskShopPayloadDto payload = shopKey == null ? null : payloadByShop.get(shopKey);
// 兼容单店任务：如果规范化店名偶发不一致导致未命中，且任务与回传都只有 1 个店时按唯一店铺回填。
if (payload == null && resultRows.size() == 1 && payloadByShop.size() == 1) {
payload = payloadByShop.values().iterator().next();
fallbackMatchedCount++;
log.warn("[product-risk] single-shop fallback matched taskId={} dbShop={} payloadShop={}",
taskId, shopKey, payload.getShopName());
}
if (payload == null) {
skippedUnmatchedCount++;
// 与删除品牌一致：支持队列逐条处理、分次回传，本次 payload 未包含的店铺保持待处理。
continue;
}
matchedShopCount++;
if (payload.getError() != null && !payload.getError().isBlank()) {
markResultFailed(fr, payload.getError());
batchErrors.add(shopKey + ": " + payload.getError());
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
continue;
}
// 分次上报累积：合并到 Redis 缓存，再按“合并后是否完成”决定是否出包。
ProductRiskShopPayloadDto mergedPayload = mergeShopPayload(taskId, shopKey, payload);
boolean incomingShopDone = hasShopDoneSignal(payload);
int mergedRows = countPayloadRows(mergedPayload);
boolean shopDone = isShopPayloadCompleted(mergedPayload);
log.info("[product-risk] shop merged taskId={} shop={} incomingShopDone={} mergedDone={} mergedRows={} payloadCountries={} mergedShopDoneFlag={}",
taskId,
shopKey,
incomingShopDone,
shopDone,
mergedRows,
mergedPayload.getCountries() == null ? 0 : mergedPayload.getCountries().size(),
mergedPayload.getShopDone());
if (incomingShopDone && mergedRows <= 0) {
markResultNoData(fr, "没有数据");
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
log.info("[product-risk] shop finished without data taskId={} shop={}", taskId, shopKey);
continue;
}
if (!shopDone) {
waitingCount++;
continue;
}
try {
enqueueResultFileAssembly(fr, shopKey, mergedPayload);
assembledCount++;
log.info("[product-risk] shop assembled taskId={} shop={} rowCount={} file={}",
taskId, shopKey, fr.getRowCount(), fr.getResultFilename());
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
} catch (Exception ex) {
log.warn("[product-risk] shop assemble failed taskId={} shop={} msg={}", taskId, shopKey, ex.getMessage());
markResultFailed(fr, ex.getMessage() == null ? "组装失败" : ex.getMessage());
batchErrors.add(shopKey + ": " + ex.getMessage());
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
}
}
List<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
log.warn("[product-risk] submitResult status evaluated taskId={} oldStatus={} newStatus={} matchedShopCount={} skippedUnmatchedCount={} waitingCount={} assembledCount={} fallbackMatchedCount={} batchErrors={}",
taskId, oldStatus, task.getStatus(), matchedShopCount, skippedUnmatchedCount,
waitingCount, assembledCount, fallbackMatchedCount, batchErrors);
}
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) {
if (taskId == null || taskId <= 0) {
return false;
}
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId);
if (lockHandle == null) {
log.info("[product-risk] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return false;
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[product-risk] stale finalize skipped taskId={} status={} fromCompensation={}",
taskId, task.getStatus(), fromCompensation);
return true;
}
String oldStatus = task.getStatus();
try (lockHandle) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false;
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
log.warn("[product-risk] stale finalize skipped taskId={} status={} fromCompensation={}",
taskId, task.getStatus(), fromCompensation);
return true;
}
String oldStatus = task.getStatus();
List<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (resultRows.isEmpty()) {
return false;
}
List<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (resultRows.isEmpty()) {
return false;
}
Map<String, ProductRiskShopPayloadDto> cachedPayloadByShop = productRiskTaskCacheService.getAllShopMergedPayload(taskId);
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "product-risk-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
boolean changed = false;
Map<String, ProductRiskShopPayloadDto> cachedPayloadByShop = productRiskTaskCacheService.getAllShopMergedPayload(taskId);
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "product-risk-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
boolean changed = false;
for (FileResultEntity fr : resultRows) {
boolean success = fr.getSuccess() != null && fr.getSuccess() == 1;
@@ -754,18 +760,19 @@ public class ProductRiskTaskService {
}
}
if (!changed && cachedPayloadByShop.isEmpty()) {
return false;
}
if (!changed && cachedPayloadByShop.isEmpty()) {
return false;
}
List<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
log.warn("[product-risk] stale finalize status evaluated taskId={} oldStatus={} newStatus={} changed={} cachedPayloadCount={} fromCompensation={} batchErrors={}",
taskId, oldStatus, task.getStatus(), changed, cachedPayloadByShop.size(), fromCompensation, batchErrors);
return true;
List<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
log.warn("[product-risk] stale finalize status evaluated taskId={} oldStatus={} newStatus={} changed={} cachedPayloadCount={} fromCompensation={} batchErrors={}",
taskId, oldStatus, task.getStatus(), changed, cachedPayloadByShop.size(), fromCompensation, batchErrors);
return true;
}
}
private void markResultFailed(FileResultEntity fr, String message) {

36
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinTaskCacheService.java
View File

@@ -14,6 +14,7 @@ import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -79,6 +80,41 @@ public class QueryAsinTaskCacheService {
}
}
public Map<Long, Long> getTaskHeartbeatMillisBatch(List<Long> taskIds) {
Map<Long, Long> result = new LinkedHashMap<>();
if (taskIds == null || taskIds.isEmpty()) {
return result;
}
List<Long> normalized = taskIds.stream()
.filter(id -> id != null && id > 0)
.distinct()
.toList();
if (normalized.isEmpty()) {
return result;
}
List<String> keys = normalized.stream().map(this::buildTaskHeartbeatKey).toList();
List<String> values;
try {
values = stringRedisTemplate.opsForValue().multiGet(keys);
} catch (Exception ex) {
log.warn("[query-asin-cache] batch get heartbeat degraded taskIds={} msg={}", normalized, ex.getMessage());
return result;
}
for (int i = 0; i < normalized.size(); i++) {
String raw = values != null && i < values.size() ? values.get(i) : null;
if (raw == null || raw.isBlank()) {
result.put(normalized.get(i), 0L);
continue;
}
try {
result.put(normalized.get(i), Long.parseLong(raw));
} catch (NumberFormatException ignored) {
result.put(normalized.get(i), 0L);
}
}
return result;
}
public void touchTaskHeartbeat(Long taskId) {
try {
stringRedisTemplate.opsForValue().set(

11
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinTaskService.java
View File

@@ -328,6 +328,7 @@ public class QueryAsinTaskService {
throw new BusinessException("shops 不能为空");
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
@@ -370,12 +371,19 @@ public class QueryAsinTaskService {
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
tryFinalizeTask(taskId, false);
}
}
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) {
if (taskId == null || taskId <= 0) {
return false;
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId);
if (lockHandle == null) {
log.info("[query-asin] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return false;
}
try (lockHandle) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false;
@@ -446,6 +454,7 @@ public class QueryAsinTaskService {
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
return changed;
}
}
@Transactional
@@ -455,11 +464,13 @@ public class QueryAsinTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
}
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
taskCacheService.deleteTaskCache(taskId);
}
}
@Transactional

141
backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java
View File

@@ -275,11 +275,13 @@ public class ShopMatchTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
}
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
shopMatchTaskCacheService.deleteTaskCache(taskId);
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
fileTaskMapper.deleteById(taskId);
shopMatchTaskCacheService.deleteTaskCache(taskId);
}
}
private void reconcileTaskAfterResultRemoval(Long taskId) {
@@ -452,42 +454,44 @@ public class ShopMatchTaskService {
if (stageIndex == null || stageIndex < 0) {
throw new BusinessException("stage_index 不合法");
}
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
throw new BusinessException("任务不存在");
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
throw new BusinessException("任务状态不正确");
}
if (!"SCHEDULED".equals(task.getStatus())) {
throw new BusinessException("任务状态不正确");
}
LocalDateTime now = now();
if (task.getScheduledAt() != null && now.isBefore(task.getScheduledAt().minusSeconds(90))) {
log.warn("[shop-match] activate rejected taskId={} stageIndex={} now={} scheduledAt={} earliestActivateAt={}",
taskId, stageIndex, now, task.getScheduledAt(), task.getScheduledAt().minusSeconds(90));
throw new BusinessException("未到定时执行时间");
}
ShopMatchCreateTaskRequest state = parseTaskRequest(task);
int currentStageIndex = state.getCurrentStageIndex() == null ? 0 : state.getCurrentStageIndex();
if (currentStageIndex != stageIndex) {
throw new BusinessException("当前轮次已变化，请刷新后重试");
}
if (state.getScheduleTimes() == null || stageIndex >= state.getScheduleTimes().size()) {
throw new BusinessException("轮次配置缺失");
}
FileTaskEntity runningTask = findOtherRunningTask(userId, taskId);
if (runningTask != null) {
log.warn("[shop-match] activate serialized taskId={} stageIndex={} blockedByTaskId={} blockedUpdatedAt={}",
taskId, stageIndex, runningTask.getId(), runningTask.getUpdatedAt());
throw new BusinessException("当前已有定时匹配任务执行中，请等待上一任务执行结束后再重试");
}
state.setActiveStageIndex(stageIndex);
task.setStatus("RUNNING");
task.setUpdatedAt(now);
persistTaskRequest(task, state);
updateTaskAndRefreshCache(task);
shopMatchTaskCacheService.touchTaskHeartbeat(taskId);
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
throw new BusinessException("任务状态不正确");
}
if (!"SCHEDULED".equals(task.getStatus())) {
throw new BusinessException("任务状态不正确");
}
LocalDateTime now = now();
if (task.getScheduledAt() != null && now.isBefore(task.getScheduledAt().minusSeconds(90))) {
log.warn("[shop-match] activate rejected taskId={} stageIndex={} now={} scheduledAt={} earliestActivateAt={}",
taskId, stageIndex, now, task.getScheduledAt(), task.getScheduledAt().minusSeconds(90));
throw new BusinessException("未到定时执行时间");
}
ShopMatchCreateTaskRequest state = parseTaskRequest(task);
int currentStageIndex = state.getCurrentStageIndex() == null ? 0 : state.getCurrentStageIndex();
if (currentStageIndex != stageIndex) {
throw new BusinessException("当前轮次已变化，请刷新后重试");
}
if (state.getScheduleTimes() == null || stageIndex >= state.getScheduleTimes().size()) {
throw new BusinessException("轮次配置缺失");
}
FileTaskEntity runningTask = findOtherRunningTask(userId, taskId);
if (runningTask != null) {
log.warn("[shop-match] activate serialized taskId={} stageIndex={} blockedByTaskId={} blockedUpdatedAt={}",
taskId, stageIndex, runningTask.getId(), runningTask.getUpdatedAt());
throw new BusinessException("当前已有定时匹配任务执行中，请等待上一任务执行结束后再重试");
}
state.setActiveStageIndex(stageIndex);
task.setStatus("RUNNING");
task.setUpdatedAt(now);
persistTaskRequest(task, state);
updateTaskAndRefreshCache(task);
shopMatchTaskCacheService.touchTaskHeartbeat(taskId);
}
@Transactional
@@ -605,26 +609,32 @@ public class ShopMatchTaskService {
if (taskId == null || taskId <= 0) {
return false;
}
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId);
if (lockHandle == null) {
log.info("[shop-match] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return false;
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
return true;
}
try (lockHandle) {
FileTaskEntity task = loadTaskForExecution(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false;
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
return true;
}
List<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (resultRows.isEmpty()) {
return false;
}
List<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (resultRows.isEmpty()) {
return false;
}
Map<String, ShopMatchShopPayloadDto> cachedPayloadByShop = shopMatchTaskCacheService.getAllShopMergedPayload(taskId);
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "shop-match-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
boolean changed = false;
Map<String, ShopMatchShopPayloadDto> cachedPayloadByShop = shopMatchTaskCacheService.getAllShopMergedPayload(taskId);
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "shop-match-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
boolean changed = false;
for (FileResultEntity result : resultRows) {
boolean success = result.getSuccess() != null && result.getSuccess() == 1;
@@ -689,16 +699,17 @@ public class ShopMatchTaskService {
}
}
if (!changed && cachedPayloadByShop.isEmpty()) {
return false;
}
if (!changed && cachedPayloadByShop.isEmpty()) {
return false;
}
List<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
return true;
List<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
return true;
}
}
private void assembleShopResult(FileResultEntity result, String shopKey, ShopMatchShopPayloadDto payload, File workRoot) {

48
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
View File

@@ -316,6 +316,7 @@ public class SimilarAsinTaskService {
@Transactional
public void activateTask(Long taskId, Long userId) {
try (TaskDistributedLockService.LockHandle ignored = requireTaskLock(taskId, TASK_LOCK_WAIT_MILLIS)) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) {
throw new BusinessException("任务不存在");
@@ -327,6 +328,7 @@ public class SimilarAsinTaskService {
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
taskCacheService.touchTaskHeartbeat(taskId);
}
}
public SimilarAsinDashboardVo dashboard(Long userId) {
@@ -546,6 +548,7 @@ public class SimilarAsinTaskService {
@Transactional
public void deleteTask(Long taskId, Long userId) {
try (TaskDistributedLockService.LockHandle ignored = requireTaskLock(taskId, TASK_LOCK_WAIT_MILLIS)) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !Objects.equals(userId, task.getUserId())) {
throw new BusinessException("任务不存在");
@@ -556,6 +559,7 @@ public class SimilarAsinTaskService {
taskChunkMapper.delete(new LambdaQueryWrapper<TaskChunkEntity>().eq(TaskChunkEntity::getTaskId, taskId).eq(TaskChunkEntity::getModuleType, MODULE_TYPE));
taskCacheService.deleteTaskCache(taskId);
fileTaskMapper.deleteById(taskId);
}
}
public void deleteHistory(Long resultId, Long userId) {
@@ -1568,6 +1572,14 @@ public class SimilarAsinTaskService {
return lockHandle;
}
private TaskDistributedLockService.LockHandle requireTaskLock(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, waitMillis);
if (lockHandle == null) {
throw new BusinessException(40902, "Task is busy, please retry later");
}
return lockHandle;
}
private void completeCozeFileJob(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
@@ -2279,6 +2291,7 @@ public class SimilarAsinTaskService {
if (row == null || row.getTaskId() == null) {
return;
}
Long taskId = row.getTaskId();
if (Boolean.TRUE.equals(vo.getFileReady())) {
vo.setFileProgressCurrent(1);
vo.setFileProgressTotal(1);
@@ -2286,8 +2299,29 @@ public class SimilarAsinTaskService {
vo.setFileProgressMessage("结果文件已生成");
return;
}
TaskProgressSnapshotEntity snapshot = taskProgressSnapshotService.find(row.getTaskId(), MODULE_TYPE);
int cozeCompleted = countCompletedCozeStates(taskId);
int cozePending = countPendingCozeStates(taskId);
if (job != null && STATUS_RUNNING.equals(job.getStatus()) && cozePending > 0) {
int total = Math.max(1, cozeCompleted + cozePending);
int current = Math.max(0, Math.min(cozeCompleted, total));
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(Math.max(0, Math.min(100, (int) Math.floor(current * 100.0 / total))));
vo.setFileProgressMessage("Coze 已完成 " + current + "/" + total + "，等待回流");
return;
}
TaskProgressSnapshotEntity snapshot = taskProgressSnapshotService.find(taskId, MODULE_TYPE);
if (snapshot == null) {
if (job != null && STATUS_RUNNING.equals(job.getStatus()) && cozeCompleted + cozePending > 0) {
int total = Math.max(1, cozeCompleted + cozePending);
int current = Math.max(0, Math.min(cozeCompleted, total));
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(Math.max(0, Math.min(100, (int) Math.floor(current * 100.0 / total))));
vo.setFileProgressMessage(cozePending > 0
? "Coze 已完成 " + current + "/" + total + "，等待回流"
: "Coze 处理中");
}
return;
}
int total = snapshot.getTotalCount() == null ? 0 : snapshot.getTotalCount();
@@ -2296,22 +2330,12 @@ public class SimilarAsinTaskService {
return;
}
current = Math.max(0, Math.min(current, total));
int percent = Math.max(1, Math.min(99, (int) Math.floor(current * 100.0 / total)));
if (job != null && STATUS_RUNNING.equals(job.getStatus())) {
LocalDateTime baseTime = snapshot.getUpdatedAt() != null ? snapshot.getUpdatedAt() : job.getUpdatedAt();
long elapsedSeconds = baseTime == null ? 0 : Math.max(0, Duration.between(baseTime, LocalDateTime.now()).getSeconds());
if (current <= 0) {
percent = Math.max(percent, Math.min(35, 8 + (int) (elapsedSeconds / 6)));
} else if (current < total) {
percent = Math.max(percent, Math.min(92, percent + (int) (elapsedSeconds / 10)));
}
}
int percent = Math.max(0, Math.min(100, (int) Math.floor(current * 100.0 / total)));
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(percent);
vo.setFileProgressMessage(snapshot.getMessage());
}
private String fmt(LocalDateTime t) {
return t == null ? null : t.toString();
}

51
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java
View File

@@ -13,8 +13,14 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.Base64;
import java.util.Locale;
import java.util.Objects;
import java.util.UUID;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
@Service
@RequiredArgsConstructor
@@ -57,6 +63,10 @@ public class TransientPayloadStorageService {
return store("task-parsed", moduleType, taskId, scopeHash, "latest", content, encodeAsJsonString, false);
}
public String storeParsedPayloadEntry(String moduleType, Long taskId, String scopeHash, String entryKey, String content, boolean encodeAsJsonString) {
return store("task-parsed", moduleType, taskId, scopeHash, entryKey, content, encodeAsJsonString, false);
}
public String storeResultPayload(String moduleType, Long taskId, String scopeHash, String submissionId, String content) {
return store("task-result-payload", moduleType, taskId, scopeHash, submissionId, content, true);
}
@@ -84,13 +94,13 @@ public class TransientPayloadStorageService {
}
try {
if (pointer.startsWith(LOCAL_POINTER_PREFIX)) {
return readLocalPayload(pointer.substring(LOCAL_POINTER_PREFIX.length()));
return decodeStoredPayload(readLocalPayload(pointer.substring(LOCAL_POINTER_PREFIX.length())));
}
if (pointer.startsWith(RUSTFS_POINTER_PREFIX)) {
return rustfsObjectStorageService.readObjectAsString(pointer.substring(RUSTFS_POINTER_PREFIX.length()));
return decodeStoredPayload(rustfsObjectStorageService.readObjectAsString(pointer.substring(RUSTFS_POINTER_PREFIX.length())));
}
if (pointer.startsWith(OSS_POINTER_PREFIX)) {
return ossStorageService.readObjectAsString(pointer.substring(OSS_POINTER_PREFIX.length()));
return decodeStoredPayload(ossStorageService.readObjectAsString(pointer.substring(OSS_POINTER_PREFIX.length())));
}
} catch (Exception ex) {
throw new IllegalStateException(errorMessage + ": " + pointer, ex);
@@ -175,10 +185,11 @@ public class TransientPayloadStorageService {
return content;
}
String objectKey = buildObjectKey(category, moduleType, taskId, scopeHash, entryKey);
String storedContent = encodeStoredPayload(content);
String pointer = null;
if (rustfsObjectStorageService.isConfigured()) {
try {
pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, content, verifyAfterUpload);
pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, storedContent, verifyAfterUpload);
} catch (Exception ex) {
log.warn("[transient-payload] rustfs upload failed objectKey={} err={}",
objectKey, ex.getMessage());
@@ -186,7 +197,7 @@ public class TransientPayloadStorageService {
}
}
if (pointer == null) {
pointer = storeLocal(objectKey, content);
pointer = storeLocal(objectKey, storedContent);
}
if (pointer == null) {
return content;
@@ -231,6 +242,36 @@ public class TransientPayloadStorageService {
}
}
private String encodeStoredPayload(String content) {
try {
byte[] raw = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) {
gzip.write(raw);
}
return "gzip64:" + Base64.getEncoder().encodeToString(baos.toByteArray());
} catch (Exception ex) {
throw new IllegalStateException("failed to encode transient payload", ex);
}
}
private String decodeStoredPayload(String storedContent) {
if (storedContent == null || storedContent.isBlank()) {
return storedContent;
}
if (!storedContent.startsWith("gzip64:")) {
return storedContent;
}
try {
byte[] compressed = Base64.getDecoder().decode(storedContent.substring("gzip64:".length()));
try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) {
return new String(gzip.readAllBytes(), StandardCharsets.UTF_8);
}
} catch (Exception ex) {
throw new IllegalStateException("failed to decode transient payload", ex);
}
}
private void deleteLocalPayload(String objectKey) {
try {
Files.deleteIfExists(resolveLocalPayloadPath(objectKey));