diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java new file mode 100644 index 0000000..d0ded5a --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java @@ -0,0 +1,16 @@ +package com.nanri.aiimage.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +@Data +@ConfigurationProperties(prefix = "aiimage.delete-brand-progress") +public class DeleteBrandProgressProperties { + private long heartbeatTimeoutMinutes = 15; + private String staleCheckCron = "0 */2 * * * *"; + + /** + * 定时补偿 finalize：用于“分片其实已齐，但最后一次提交断开导致没触发 finalize”的场景。 + */ + private String finalizeCheckCron = "30 */2 * * * *"; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java index f2de092..24b5973 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java @@ -4,6 +4,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.context.annotation.Configuration; @Configuration -@EnableConfigurationProperties({OssProperties.class, StorageProperties.class, BrandProgressProperties.class, ZiniaoProperties.class, ModuleCleanupProperties.class}) +@EnableConfigurationProperties({OssProperties.class, StorageProperties.class, BrandProgressProperties.class, DeleteBrandProgressProperties.class, ZiniaoProperties.class, ModuleCleanupProperties.class}) public class PropertiesConfig { } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java index 34b6600..49d3d4c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java @@ -3,8 +3,10 @@ package com.nanri.aiimage.modules.deletebrand.controller; import com.nanri.aiimage.common.api.ApiResponse; import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandRunRequest; import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandSubmitResultRequest; +import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandTaskBatchRequest; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandHistoryVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandRunVo; +import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo; import com.nanri.aiimage.modules.deletebrand.service.DeleteBrandRunService; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; @@ -75,6 +77,12 @@ public class DeleteBrandRunController { return ApiResponse.success(deleteBrandRunService.getTaskDetail(taskId)); } + @PostMapping("/tasks/batch") + @Operation(summary = "批量获取删除品牌任务详情", description = "合并多个 taskId 的详情查询，减少轮询期对 DB/Redis 的压力。") + public ApiResponse getTasksBatch(@Valid @RequestBody DeleteBrandTaskBatchRequest request) { + return ApiResponse.success(deleteBrandRunService.getTaskDetails(request.getTaskIds())); + } + @PostMapping("/tasks/{taskId}/result") @Operation(summary = "提交删除品牌处理结果", description = "插件/前端分片回传处理结果；后端累计分片并推进进度，全部完成后再执行最终组装与落库。") public ApiResponse submitResult( diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/dto/DeleteBrandTaskBatchRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/dto/DeleteBrandTaskBatchRequest.java new file mode 100644 index 0000000..e6f3018 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/dto/DeleteBrandTaskBatchRequest.java @@ -0,0 +1,17 @@ +package com.nanri.aiimage.modules.deletebrand.model.dto; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotEmpty; +import lombok.Data; + +import java.util.ArrayList; +import java.util.List; + +@Data +@Schema(description = "删除品牌任务批量详情请求") +public class DeleteBrandTaskBatchRequest { + + @NotEmpty(message = "taskIds 不能为空") + @Schema(description = "任务ID列表") + private List taskIds = new ArrayList<>(); +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/vo/DeleteBrandTaskBatchVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/vo/DeleteBrandTaskBatchVo.java new file mode 100644 index 0000000..2bcfe52 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/vo/DeleteBrandTaskBatchVo.java @@ -0,0 +1,18 @@ +package com.nanri.aiimage.modules.deletebrand.model.vo; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.util.ArrayList; +import java.util.List; + +@Data +@Schema(description = "删除品牌任务批量详情响应") +public class DeleteBrandTaskBatchVo { + + @Schema(description = "任务详情列表，按请求顺序返回") + private List items = new ArrayList<>(); + + @Schema(description = "不存在/无权限的 taskId") + private List missingTaskIds = new ArrayList<>(); +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java index 29205c4..26ae1f6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java @@ -7,6 +7,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.config.DeleteBrandProgressProperties; import com.nanri.aiimage.modules.deletebrand.model.cache.DeleteBrandParsedFileCacheDto; import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandCountryResultItemDto; import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandProcessedCountryDto; @@ -66,6 +67,7 @@ public class DeleteBrandRunService { private final com.nanri.aiimage.modules.ziniao.service.ZiniaoAuthService ziniaoAuthService; private final OssStorageService ossStorageService; private final ObjectMapper objectMapper; + private final DeleteBrandProgressProperties deleteBrandProgressProperties; public DeleteBrandRunVo run(DeleteBrandRunRequest request) { if (request.getUserId() == null || request.getUserId() <= 0) { @@ -206,27 +208,72 @@ public class DeleteBrandRunService { public DeleteBrandHistoryVo listHistory(Long userId) { DeleteBrandHistoryVo vo = new DeleteBrandHistoryVo(); - vo.setItems(fileResultMapper.selectList(new LambdaQueryWrapper() - .eq(FileResultEntity::getModuleType, MODULE_TYPE) - .eq(FileResultEntity::getUserId, userId) - .orderByDesc(FileResultEntity::getCreatedAt) - .last("limit 100")) - .stream() - .map(entity -> { - DeleteBrandResultItemVo item = new DeleteBrandResultItemVo(); - item.setResultId(entity.getId()); - item.setTaskId(entity.getTaskId()); - item.setFileKey(entity.getSourceFileUrl()); - item.setSourceFilename(entity.getSourceFilename()); - item.setShopName(entity.getSourceFilename() == null ? null : FileUtil.mainName(entity.getSourceFilename())); - item.setOutputFilename(entity.getResultFilename()); - item.setDownloadUrl(entity.getResultFileUrl() == null ? null : ossStorageService.getPublicUrl(entity.getResultFileUrl())); - item.setTotalRows(entity.getRowCount()); - item.setSuccess(entity.getSuccess() != null && entity.getSuccess() == 1); - item.setError(entity.getErrorMessage()); - return item; - }) - .toList()); + List entities = fileResultMapper.selectList(new LambdaQueryWrapper() + .eq(FileResultEntity::getModuleType, MODULE_TYPE) + .eq(FileResultEntity::getUserId, userId) + .orderByDesc(FileResultEntity::getCreatedAt) + .last("limit 100")); + + // 补充从 task.resultJson 还原匹配信息/真实失败原因（历史表本身不存这些字段） + Map> itemsByTaskId = new LinkedHashMap<>(); + List taskIds = entities.stream() + .map(FileResultEntity::getTaskId) + .filter(id -> id != null && id > 0) + .distinct() + .toList(); + if (!taskIds.isEmpty()) { + List tasks = fileTaskMapper.selectBatchIds(taskIds); + for (FileTaskEntity task : tasks) { + if (task == null || task.getResultJson() == null || task.getResultJson().isBlank()) { + continue; + } + try { + List taskItems = objectMapper.readValue(task.getResultJson(), new TypeReference>() { + }); + itemsByTaskId.put(task.getId(), taskItems); + } catch (Exception ignored) { + } + } + } + + vo.setItems(entities.stream().map(entity -> { + DeleteBrandResultItemVo item = new DeleteBrandResultItemVo(); + item.setResultId(entity.getId()); + item.setTaskId(entity.getTaskId()); + item.setFileKey(entity.getSourceFileUrl()); + item.setSourceFilename(entity.getSourceFilename()); + item.setShopName(entity.getSourceFilename() == null ? null : FileUtil.mainName(entity.getSourceFilename())); + item.setOutputFilename(entity.getResultFilename()); + item.setDownloadUrl(entity.getResultFileUrl() == null ? null : ossStorageService.getPublicUrl(entity.getResultFileUrl())); + item.setTotalRows(entity.getRowCount()); + item.setSuccess(entity.getSuccess() != null && entity.getSuccess() == 1); + item.setError(entity.getErrorMessage()); + + // 以 task.resultJson 中的项为准补全 matched/shopId/platform/openStoreUrl/错误信息 + if (entity.getTaskId() != null) { + List taskItems = itemsByTaskId.get(entity.getTaskId()); + if (taskItems != null && entity.getSourceFilename() != null && !entity.getSourceFilename().isBlank()) { + for (DeleteBrandResultItemVo candidate : taskItems) { + if (candidate == null) { + continue; + } + if (entity.getSourceFilename().equals(candidate.getSourceFilename())) { + item.setMatched(candidate.isMatched()); + item.setShopId(candidate.getShopId()); + item.setPlatform(candidate.getPlatform()); + item.setOpenStoreUrl(candidate.getOpenStoreUrl()); + // 历史表可能只有通用 errorMessage；优先返回任务当时的真实 error + if ((item.getError() == null || item.getError().isBlank()) && candidate.getError() != null && !candidate.getError().isBlank()) { + item.setError(candidate.getError()); + } + break; + } + } + } + } + return item; + }).toList()); + return vo; } @@ -391,7 +438,46 @@ public class DeleteBrandRunService { if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { throw new BusinessException("任务不存在"); } + return buildTaskDetail(task); + } + public com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo getTaskDetails(java.util.List taskIds) { + com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo vo = new com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandTaskBatchVo(); + if (taskIds == null || taskIds.isEmpty()) { + return vo; + } + + java.util.List normalizedTaskIds = taskIds.stream() + .filter(id -> id != null && id > 0) + .distinct() + .limit(50) + .toList(); + if (normalizedTaskIds.isEmpty()) { + return vo; + } + + java.util.Map taskById = fileTaskMapper.selectBatchIds(normalizedTaskIds).stream() + .filter(task -> task != null && MODULE_TYPE.equals(task.getModuleType())) + .collect(java.util.stream.Collectors.toMap(FileTaskEntity::getId, task -> task, (left, right) -> left, java.util.LinkedHashMap::new)); + + java.util.Map> progressByTaskId = deleteBrandTaskCacheService.getProgressBatch(normalizedTaskIds); + + for (Long taskId : normalizedTaskIds) { + FileTaskEntity task = taskById.get(taskId); + if (task == null) { + vo.getMissingTaskIds().add(taskId); + continue; + } + vo.getItems().add(buildTaskDetail(task, progressByTaskId.get(taskId))); + } + return vo; + } + + private DeleteBrandTaskDetailVo buildTaskDetail(FileTaskEntity task) { + return buildTaskDetail(task, null); + } + + private DeleteBrandTaskDetailVo buildTaskDetail(FileTaskEntity task, java.util.Map progress) { DeleteBrandTaskItemVo taskVo = new DeleteBrandTaskItemVo(); taskVo.setId(task.getId()); taskVo.setTaskNo(task.getTaskNo()); @@ -403,10 +489,16 @@ public class DeleteBrandRunService { taskVo.setCreatedAt(task.getCreatedAt()); taskVo.setUpdatedAt(task.getUpdatedAt()); taskVo.setFinishedAt(task.getFinishedAt()); - taskVo.setDownloadUrl(resolveTaskDownloadUrl(task.getId())); - taskVo.setDownloadFilename(resolveTaskDownloadFilename(task.getId())); + // 轮询期避免每次都查 biz_file_result（会放大 DB 压力）；仅任务成功后再提供下载信息 + if ("SUCCESS".equals(task.getStatus())) { + taskVo.setDownloadUrl(resolveTaskDownloadUrl(task.getId())); + taskVo.setDownloadFilename(resolveTaskDownloadFilename(task.getId())); + } else { + taskVo.setDownloadUrl(null); + taskVo.setDownloadFilename(null); + } - DeleteBrandLineProgressVo progressVo = buildLineProgress(taskId); + DeleteBrandLineProgressVo progressVo = buildLineProgress(task.getId(), progress); DeleteBrandTaskDetailVo detail = new DeleteBrandTaskDetailVo(); detail.setTask(taskVo); @@ -415,23 +507,27 @@ public class DeleteBrandRunService { } private DeleteBrandLineProgressVo buildLineProgress(Long taskId) { - Map progress = deleteBrandTaskCacheService.getProgress(taskId); + return buildLineProgress(taskId, null); + } + + private DeleteBrandLineProgressVo buildLineProgress(Long taskId, Map progress) { + Map resolvedProgress = progress != null ? progress : deleteBrandTaskCacheService.getProgress(taskId); DeleteBrandLineProgressVo vo = new DeleteBrandLineProgressVo(); - if (progress == null || progress.isEmpty()) { + if (resolvedProgress == null || resolvedProgress.isEmpty()) { vo.setHas_progress(false); vo.setInfo(null); return vo; } DeleteBrandLineProgressInfoVo info = new DeleteBrandLineProgressInfoVo(); - info.setFile_index(parseInt(progress.get("file_index"))); - info.setFile_total(parseInt(progress.get("file_total"))); - info.setFile_name(stringValue(progress.get("file_name"))); - info.setCurrent_line(parseInt(progress.get("current_line"))); - info.setTotal_lines(parseInt(progress.get("total_lines"))); - info.setCurrent_country(stringValue(progress.get("current_country"))); - info.setCurrent_asin(stringValue(progress.get("current_asin"))); - info.setFinished_files(parseInt(progress.get("finished_files"))); - info.setPhase(stringValue(progress.get("phase"))); + info.setFile_index(parseInt(resolvedProgress.get("file_index"))); + info.setFile_total(parseInt(resolvedProgress.get("file_total"))); + info.setFile_name(stringValue(resolvedProgress.get("file_name"))); + info.setCurrent_line(parseInt(resolvedProgress.get("current_line"))); + info.setTotal_lines(parseInt(resolvedProgress.get("total_lines"))); + info.setCurrent_country(stringValue(resolvedProgress.get("current_country"))); + info.setCurrent_asin(stringValue(resolvedProgress.get("current_asin"))); + info.setFinished_files(parseInt(resolvedProgress.get("finished_files"))); + info.setPhase(stringValue(resolvedProgress.get("phase"))); vo.setHas_progress(true); vo.setInfo(info); return vo; @@ -524,6 +620,9 @@ public class DeleteBrandRunService { if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { throw new BusinessException("任务不存在"); } + if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) { + throw new BusinessException("任务已结束，拒绝继续提交结果"); + } Map parsedPayload = deleteBrandTaskCacheService.getParsedPayload(taskId, new TypeReference>() { @@ -548,9 +647,17 @@ public class DeleteBrandRunService { throw new BusinessException("文件标识不匹配: " + fileIdentity); } + // 更早的 duplicate fast-path：尽量在较重的校验/进度写入之前直接短路 + if (deleteBrandTaskCacheService.hasSameResultChunk(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto)) { + continue; + } + validateChunk(fileDto, parsedFile); validateResultFileAgainstParsed(fileDto, parsedFile); - deleteBrandTaskCacheService.mergeResultChunk(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto); + boolean stored = deleteBrandTaskCacheService.mergeResultChunk(taskId, fileIdentity, fileDto.getChunkIndex(), fileDto); + if (!stored) { + continue; + } Map progress = new LinkedHashMap<>(); progress.put("phase", DeleteBrandTaskCacheService.PHASE_CRAWLING); @@ -566,19 +673,52 @@ public class DeleteBrandRunService { deleteBrandTaskCacheService.saveProgress(taskId, progress); } + tryFinalizeTask(taskId, false); + } + + public void tryFinalizeTask(Long taskId, boolean fromCompensation) { + if (taskId == null || taskId <= 0) { + return; + } + + FileTaskEntity task = fileTaskMapper.selectById(taskId); + if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { + return; + } + if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) { + return; + } + + Map parsedPayload = deleteBrandTaskCacheService.getParsedPayload(taskId, + new TypeReference>() { + }); + if (parsedPayload == null || parsedPayload.isEmpty()) { + return; + } + + Map progressSnapshot = deleteBrandTaskCacheService.getProgress(taskId); + int expectedFiles = parsedPayload.size(); + Integer finishedFilesHint = parseInt(progressSnapshot == null ? null : progressSnapshot.get("finished_files")); + if (finishedFilesHint != null && finishedFilesHint < expectedFiles) { + return; + } + Map> mergedByFile = loadMergedChunks(taskId); int finishedFiles = countCompletedFiles(mergedByFile); Map progress = new LinkedHashMap<>(); progress.put("finished_files", String.valueOf(finishedFiles)); progress.put("updated_at", String.valueOf(System.currentTimeMillis())); + if (fromCompensation) { + progress.put("last_heartbeat_at", String.valueOf(System.currentTimeMillis())); + } deleteBrandTaskCacheService.saveProgress(taskId, progress); task.setSuccessFileCount(finishedFiles); task.setUpdatedAt(LocalDateTime.now()); fileTaskMapper.updateById(task); - if (finishedFiles < parsedPayload.size()) { + if (finishedFiles < expectedFiles) { return; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java new file mode 100644 index 0000000..1689a8c --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java @@ -0,0 +1,92 @@ +package com.nanri.aiimage.modules.deletebrand.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.nanri.aiimage.config.DeleteBrandProgressProperties; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import lombok.RequiredArgsConstructor; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class DeleteBrandStaleTaskService { + + private static final String MODULE_TYPE = "DELETE_BRAND"; + + private final FileTaskMapper fileTaskMapper; + private final DeleteBrandTaskCacheService deleteBrandTaskCacheService; + private final DeleteBrandRunService deleteBrandRunService; + private final DeleteBrandProgressProperties deleteBrandProgressProperties; + + @Scheduled(cron = "${aiimage.delete-brand-progress.stale-check-cron:0 */2 * * * *}") + public void failStaleRunningTasks() { + LocalDateTime threshold = LocalDateTime.now().minusMinutes(deleteBrandProgressProperties.getHeartbeatTimeoutMinutes()); + + // 先按 DB updatedAt 粗筛，避免全表扫描 + java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, "RUNNING") + .lt(FileTaskEntity::getUpdatedAt, threshold) + .last("limit 200")); + + for (FileTaskEntity task : runningTasks) { + Map progress = deleteBrandTaskCacheService.getProgress(task.getId()); + if (progress == null || progress.isEmpty()) { + continue; + } + + long lastHeartbeatAt = 0L; + try { + lastHeartbeatAt = Long.parseLong(String.valueOf(progress.getOrDefault("last_heartbeat_at", "0"))); + } catch (Exception ignored) { + } + if (lastHeartbeatAt <= 0) { + continue; + } + + LocalDateTime lastHeartbeat = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault()); + if (lastHeartbeat.isAfter(threshold)) { + continue; + } + + int updated = fileTaskMapper.update(null, new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, task.getId()) + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, "RUNNING") + .set(FileTaskEntity::getStatus, "FAILED") + .set(FileTaskEntity::getErrorMessage, "结果回传长时间无响应，任务已自动失败") + .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()) + .set(FileTaskEntity::getFinishedAt, LocalDateTime.now())); + + if (updated > 0) { + deleteBrandTaskCacheService.saveProgress(task.getId(), java.util.Map.of( + "phase", DeleteBrandTaskCacheService.PHASE_FAILED, + "updated_at", String.valueOf(System.currentTimeMillis()) + )); + } + } + } + + @Scheduled(cron = "${aiimage.delete-brand-progress.finalize-check-cron:30 */2 * * * *}") + public void finalizeCompletedRunningTasks() { + java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, "RUNNING") + .orderByAsc(FileTaskEntity::getUpdatedAt) + .last("limit 100")); + + for (FileTaskEntity task : runningTasks) { + try { + deleteBrandRunService.tryFinalizeTask(task.getId(), true); + } catch (Exception ignored) { + // 定时补偿不影响主流程；下次调度或人工重试时再处理 + } + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandTaskCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandTaskCacheService.java index d1fb47f..8c38fdc 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandTaskCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandTaskCacheService.java @@ -8,6 +8,7 @@ import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; import java.time.Duration; +import java.util.concurrent.ConcurrentHashMap; @Service @RequiredArgsConstructor @@ -19,25 +20,45 @@ public class DeleteBrandTaskCacheService { public static final String PHASE_FAILED = "failed"; private static final long PAYLOAD_TTL_HOURS = 24; + private static final long LOCAL_PARSED_PAYLOAD_CACHE_MILLIS = 3000; private final StringRedisTemplate stringRedisTemplate; private final ObjectMapper objectMapper; + private final ConcurrentHashMap parsedPayloadLocalCache = new ConcurrentHashMap<>(); public void saveParsedPayload(Long taskId, Object payload) { try { stringRedisTemplate.opsForValue().set(buildPayloadKey(taskId), objectMapper.writeValueAsString(payload), Duration.ofHours(PAYLOAD_TTL_HOURS)); + parsedPayloadLocalCache.put(taskId, new LocalParsedPayloadCacheEntry(System.currentTimeMillis(), payload)); } catch (Exception ex) { throw new BusinessException("暂存删除品牌原始数据失败"); } } public T getParsedPayload(Long taskId, TypeReference typeReference) { + if (taskId == null || taskId <= 0) { + return null; + } + + long now = System.currentTimeMillis(); + LocalParsedPayloadCacheEntry cached = parsedPayloadLocalCache.get(taskId); + if (cached != null && cached.isFresh(now) && cached.value() != null) { + try { + return objectMapper.convertValue(cached.value(), typeReference); + } catch (Exception ignored) { + // fallthrough to redis + } + } + String raw = stringRedisTemplate.opsForValue().get(buildPayloadKey(taskId)); if (raw == null || raw.isBlank()) { + parsedPayloadLocalCache.remove(taskId); return null; } try { - return objectMapper.readValue(raw, typeReference); + T parsed = objectMapper.readValue(raw, typeReference); + parsedPayloadLocalCache.put(taskId, new LocalParsedPayloadCacheEntry(now, parsed)); + return parsed; } catch (Exception ex) { throw new BusinessException("读取删除品牌原始数据失败"); } @@ -53,7 +74,67 @@ public class DeleteBrandTaskCacheService { return stringRedisTemplate.opsForHash().entries(buildProgressKey(taskId)); } - public void mergeResultChunk(Long taskId, String fileIdentity, Integer chunkIndex, Object chunk) { + public java.util.Map> getProgressBatch(java.util.List taskIds) { + java.util.Map> result = new java.util.LinkedHashMap<>(); + if (taskIds == null || taskIds.isEmpty()) { + return result; + } + + java.util.List normalizedTaskIds = taskIds.stream() + .filter(taskId -> taskId != null && taskId > 0) + .distinct() + .toList(); + if (normalizedTaskIds.isEmpty()) { + return result; + } + + java.util.List pipelineResults = stringRedisTemplate.executePipelined((org.springframework.data.redis.core.RedisCallback) connection -> { + org.springframework.data.redis.serializer.RedisSerializer serializer = stringRedisTemplate.getStringSerializer(); + for (Long taskId : normalizedTaskIds) { + connection.hGetAll(serializer.serialize(buildProgressKey(taskId))); + } + return null; + }); + + for (int i = 0; i < normalizedTaskIds.size(); i++) { + Long taskId = normalizedTaskIds.get(i); + Object raw = pipelineResults != null && i < pipelineResults.size() ? pipelineResults.get(i) : null; + if (raw instanceof java.util.Map rawMap) { + java.util.Map converted = new java.util.LinkedHashMap<>(); + for (java.util.Map.Entry entry : rawMap.entrySet()) { + converted.put(entry.getKey(), entry.getValue()); + } + result.put(taskId, converted); + } else { + result.put(taskId, java.util.Collections.emptyMap()); + } + } + return result; + } + + public boolean hasSameResultChunk(Long taskId, String fileIdentity, Integer chunkIndex, Object chunk) { + if (chunkIndex == null || chunkIndex <= 0) { + throw new BusinessException("chunkIndex 不合法"); + } + if (fileIdentity == null || fileIdentity.isBlank()) { + throw new BusinessException("fileIdentity 不能为空"); + } + String field = fileIdentity.trim() + "#" + chunkIndex; + String chunkJson; + try { + chunkJson = objectMapper.writeValueAsString(chunk); + } catch (Exception ex) { + throw new BusinessException("读取删除品牌结果分片失败"); + } + Object existing = stringRedisTemplate.opsForHash().get(buildResultChunksKey(taskId), field); + if (chunkJson.equals(existing)) { + stringRedisTemplate.expire(buildResultChunksKey(taskId), Duration.ofHours(PAYLOAD_TTL_HOURS)); + return true; + } + return false; + } + + public boolean mergeResultChunk(Long taskId, String fileIdentity, Integer chunkIndex, Object chunk) { String resultKey = buildResultChunksKey(taskId); if (chunkIndex == null || chunkIndex <= 0) { throw new BusinessException("chunkIndex 不合法"); @@ -62,12 +143,20 @@ public class DeleteBrandTaskCacheService { throw new BusinessException("fileIdentity 不能为空"); } String field = fileIdentity.trim() + "#" + chunkIndex; + String chunkJson; try { - stringRedisTemplate.opsForHash().put(resultKey, field, objectMapper.writeValueAsString(chunk)); + chunkJson = objectMapper.writeValueAsString(chunk); } catch (Exception ex) { throw new BusinessException("暂存删除品牌结果分片失败"); } + Object existing = stringRedisTemplate.opsForHash().get(resultKey, field); + if (chunkJson.equals(existing)) { + stringRedisTemplate.expire(resultKey, Duration.ofHours(PAYLOAD_TTL_HOURS)); + return false; + } + stringRedisTemplate.opsForHash().put(resultKey, field, chunkJson); stringRedisTemplate.expire(resultKey, Duration.ofHours(PAYLOAD_TTL_HOURS)); + return true; } public java.util.Map> groupResultChunkJsonByFile(Long taskId) { @@ -88,11 +177,18 @@ public class DeleteBrandTaskCacheService { } public void delete(Long taskId) { + parsedPayloadLocalCache.remove(taskId); stringRedisTemplate.delete(buildPayloadKey(taskId)); stringRedisTemplate.delete(buildProgressKey(taskId)); stringRedisTemplate.delete(buildResultChunksKey(taskId)); } + private record LocalParsedPayloadCacheEntry(long cachedAtMillis, Object value) { + private boolean isFresh(long now) { + return now - cachedAtMillis <= LOCAL_PARSED_PAYLOAD_CACHE_MILLIS; + } + } + private String buildPayloadKey(Long taskId) { return "delete-brand:task:parsed-payload:" + taskId; } diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index d5446c1..7ce15a9 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -15,6 +15,11 @@ spring: url: ${AIIMAGE_DB_URL:jdbc:mysql://127.0.0.1:3306/aiimage?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false} username: ${AIIMAGE_DB_USERNAME:root} password: ${AIIMAGE_DB_PASSWORD:change-me} + hikari: + maximum-pool-size: ${AIIMAGE_DB_POOL_MAX_SIZE:30} + minimum-idle: ${AIIMAGE_DB_POOL_MIN_IDLE:5} + connection-timeout: ${AIIMAGE_DB_POOL_CONNECTION_TIMEOUT_MS:30000} + leak-detection-threshold: ${AIIMAGE_DB_POOL_LEAK_DETECT_MS:0} data: redis: host: ${AIIMAGE_REDIS_HOST:127.0.0.1} @@ -72,6 +77,10 @@ aiimage: failed-ttl-hours: ${AIIMAGE_BRAND_PROGRESS_FAILED_TTL_HOURS:2} heartbeat-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_HEARTBEAT_TIMEOUT_MINUTES:15} stale-check-cron: ${AIIMAGE_BRAND_PROGRESS_STALE_CHECK_CRON:0 */2 * * * *} + delete-brand-progress: + heartbeat-timeout-minutes: ${AIIMAGE_DELETE_BRAND_HEARTBEAT_TIMEOUT_MINUTES:15} + stale-check-cron: ${AIIMAGE_DELETE_BRAND_STALE_CHECK_CRON:0 */2 * * * *} + finalize-check-cron: ${AIIMAGE_DELETE_BRAND_FINALIZE_CHECK_CRON:30 */2 * * * *} module-cleanup: enabled: ${AIIMAGE_MODULE_CLEANUP_ENABLED:true} cron: ${AIIMAGE_MODULE_CLEANUP_CRON:0 0 0 * * *} diff --git a/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue b/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue index 3f0fdfc..2169a15 100644 --- a/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue @@ -240,6 +240,7 @@ import { deleteDeleteBrandHistory, getDeleteBrandHistory, getDeleteBrandTaskDetail, + getDeleteBrandTaskDetails, getDeleteBrandTaskDownloadUrl, runDeleteBrand, type DeleteBrandPreviewRow, @@ -260,6 +261,7 @@ const queuePushResult = ref('') const queuePayloadText = ref('') const taskDetails = ref>({}) const pollTimer = ref(null) +const pollingInFlight = ref(false) const displayPaths = computed(() => selectedPaths.value.slice(0, 10)) const currentSectionItems = computed(() => currentRunItems.value) const historySectionItems = computed(() => @@ -276,7 +278,7 @@ const hasVisibleItems = computed(() => currentSectionItems.value.length > 0 || h function normalizeDeleteBrandItems(items: DeleteBrandResultItem[]) { return items.map((item) => { - if (!item.matched) { + if (item.matched === false) { return { ...item, success: false, @@ -351,6 +353,22 @@ function isTaskRunning(taskId: number) { return status === 'RUNNING' } +function isTaskTerminal(taskId: number) { + const status = taskDetails.value[taskId]?.task?.status + return status === 'SUCCESS' || status === 'FAILED' +} + +function getPollingTaskIds() { + return Array.from( + new Set( + currentRunItems.value + .filter((item) => item.matched && (item.taskId || 0) > 0) + .map(getItemTaskId) + .filter((id) => id > 0 && !isTaskTerminal(id)), + ), + ) +} + function canDownloadTaskResult(item: DeleteBrandResultItem) { const taskId = item.taskId if (!taskId || !item.matched) return false @@ -358,32 +376,61 @@ function canDownloadTaskResult(item: DeleteBrandResultItem) { return status === 'SUCCESS' } -async function refreshTaskDetails() { - const taskIds = Array.from(new Set(resultItems.value.map(getItemTaskId).filter((id) => id > 0))) - for (const id of taskIds) { - try { - const detail = await getDeleteBrandTaskDetail(id) - taskDetails.value[id] = detail - } catch { - // ignore polling failures +async function refreshTaskDetails(taskIds?: number[]) { + const ids = taskIds || getPollingTaskIds() + if (!ids.length) return + + try { + const batch = await getDeleteBrandTaskDetails(ids) + for (const detail of batch.items || []) { + const id = detail?.task?.id + if (typeof id === 'number' && id > 0) { + taskDetails.value[id] = detail + } } + } catch { + // ignore polling failures } } +function getPollIntervalMs() { + return document.visibilityState === 'visible' ? 4500 : 10000 +} + +function scheduleNextPoll() { + if (pollTimer.value) return + pollTimer.value = window.setTimeout(async () => { + pollTimer.value = null + if (pollingInFlight.value) { + scheduleNextPoll() + return + } + + const taskIds = getPollingTaskIds() + if (!taskIds.length) return + + pollingInFlight.value = true + try { + await refreshTaskDetails(taskIds) + } finally { + pollingInFlight.value = false + } + + const stillRunning = taskIds.some((id) => isTaskRunning(id)) + if (stillRunning) { + scheduleNextPoll() + } + }, getPollIntervalMs()) +} + function ensurePolling() { if (pollTimer.value) return - pollTimer.value = window.setInterval(async () => { - await refreshTaskDetails() - const stillRunning = Object.keys(taskDetails.value).some((id) => isTaskRunning(Number(id))) - if (!stillRunning) { - stopPolling() - } - }, 1500) + scheduleNextPoll() } function stopPolling() { if (pollTimer.value) { - window.clearInterval(pollTimer.value) + window.clearTimeout(pollTimer.value) pollTimer.value = null } } @@ -464,7 +511,7 @@ async function submitRun() { currentRunItems.value = normalizedItems syncResultState() - await refreshTaskDetails() + await refreshTaskDetails(getPollingTaskIds()) ensurePolling() const api = getPywebviewApi() @@ -537,13 +584,39 @@ async function loadHistory() { } } +function removeRecordFromView(resultId: number) { + const removedFromCurrent = currentRunItems.value.find((item) => item.resultId === resultId) + const removedTaskId = removedFromCurrent?.taskId + + currentRunItems.value = currentRunItems.value.filter((item) => item.resultId !== resultId) + historyItems.value = historyItems.value.filter((item) => item.resultId !== resultId) + + if (removedTaskId && !currentRunItems.value.some((item) => item.taskId === removedTaskId)) { + delete taskDetails.value[removedTaskId] + } + + syncResultState() + + if (getPollingTaskIds().length === 0) { + stopPolling() + } +} + async function deleteHistoryRecord(resultId: number) { try { await deleteDeleteBrandHistory(resultId) + removeRecordFromView(resultId) await loadHistory() ElMessage.success('已删除') } catch (error) { - ElMessage.error(error instanceof Error ? error.message : '删除失败') + const message = error instanceof Error ? error.message : '删除失败' + // 后端可能已被清理/重复点击导致「记录不存在」，前端也应立即移除并停止轮询 + if (typeof message === 'string' && message.includes('记录不存在')) { + removeRecordFromView(resultId) + ElMessage.success('已删除') + return + } + ElMessage.error(message) } } diff --git a/frontend-vue/src/shared/api/java-modules.ts b/frontend-vue/src/shared/api/java-modules.ts index 2ca87ea..0a7f035 100644 --- a/frontend-vue/src/shared/api/java-modules.ts +++ b/frontend-vue/src/shared/api/java-modules.ts @@ -205,12 +205,17 @@ export interface DeleteBrandTaskDetailVo { line_progress: DeleteBrandLineProgress } +export interface DeleteBrandTaskBatchVo { + items: DeleteBrandTaskDetailVo[] + missingTaskIds: number[] +} + export interface DeleteBrandResultItem { resultId?: number fileKey?: string sourceFilename: string shopName?: string - matched: boolean + matched?: boolean shopId?: string platform?: string openStoreUrl?: string @@ -365,6 +370,10 @@ export function getDeleteBrandTaskDetail(taskId: number) { return unwrapJavaResponse(get>(`${JAVA_API_PREFIX}/delete-brand/tasks/${taskId}`)) } +export function getDeleteBrandTaskDetails(taskIds: number[]) { + return unwrapJavaResponse(post, { taskIds: number[] }>(`${JAVA_API_PREFIX}/delete-brand/tasks/batch`, { taskIds })) +} + export function getDeleteBrandTaskDownloadUrl(taskId: number) { return getJavaDownloadUrl(`/delete-brand/tasks/${taskId}/download`) }