diff --git a/app/.env b/app/.env index 9ffac95..c9e68e1 100644 --- a/app/.env +++ b/app/.env @@ -13,7 +13,7 @@ client_name=ShuFuAI # java_api_base=http://127.0.0.1:18080 -java_api_base=http://8.136.19.173:18081 -# java_api_base=http://8.136.19.173:18080 +# java_api_base=http://8.136.19.173:18081 +java_api_base=http://8.136.19.173:18080 diff --git a/app/__pycache__/ali_oss.cpython-312.pyc b/app/__pycache__/ali_oss.cpython-312.pyc index f2690ed..553a210 100644 Binary files a/app/__pycache__/ali_oss.cpython-312.pyc and b/app/__pycache__/ali_oss.cpython-312.pyc differ diff --git a/backend-java/brand_chunk_load_test.py b/backend-java/brand_chunk_load_test.py new file mode 100644 index 0000000..da8ccd3 --- /dev/null +++ b/backend-java/brand_chunk_load_test.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +import json +import sys +import time +from urllib import request, error + + +def build_chunk(file_url: str, chunk_index: int, chunk_total: int, total_lines: int): + return { + "fileUrl": file_url, + "originalFilename": "demo.xlsx", + "relativePath": None, + "mainSheetName": "Sheet1", + "chunkIndex": chunk_index, + "chunkTotal": chunk_total, + "totalLines": total_lines, + "keptRows": [f"brand-{chunk_index}"], + "invalidBrands": [], + "queryFailedBrands": [] + } + + +def post_json(url: str, payload: dict): + data = json.dumps(payload).encode("utf-8") + req = request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST") + with request.urlopen(req, timeout=60) as resp: + return resp.status, resp.read().decode("utf-8", errors="ignore") + + +def main(): + if len(sys.argv) < 4: + print("usage: python brand_chunk_load_test.py [file_url]") + sys.exit(1) + base_url = sys.argv[1].rstrip("/") + task_id = int(sys.argv[2]) + chunk_total = int(sys.argv[3]) + file_url = sys.argv[4] if len(sys.argv) > 4 else "https://example.com/demo.xlsx" + + submit_url = f"{base_url}/api/brand/tasks/{task_id}/result" + started = time.time() + for i in range(1, chunk_total + 1): + payload = { + "strategy": "Terms", + "files": [build_chunk(file_url, i, chunk_total, chunk_total)] + } + status, body = post_json(submit_url, payload) + if i == 1 or i == chunk_total or i % 100 == 0: + print(f"chunk {i}/{chunk_total} status={status} elapsed={time.time() - started:.2f}s") + if status < 200 or status >= 300: + print(body) + sys.exit(2) + print(f"done {chunk_total} chunks in {time.time() - started:.2f}s") + + +if __name__ == "__main__": + main() 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 index 77934a4..bd44b8d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java @@ -7,7 +7,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "aiimage.delete-brand-progress") public class DeleteBrandProgressProperties { private long heartbeatTimeoutMinutes = 15; - private String staleCheckCron = "0 */2 * * * *"; + private String staleCheckCron = "*/30 * * * * *"; /** * 定时补偿 finalize：用于“分片其实已齐，但最后一次提交断开导致没触发 finalize”的场景。 @@ -15,7 +15,8 @@ public class DeleteBrandProgressProperties { private String finalizeCheckCron = "30 */2 * * * *"; /** - * 商品风险（PRODUCT_RISK_RESOLVE）RUNNING 超时自动失败：与删除品牌共用同一定时调度。 + * 商品风险（PRODUCT_RISK_RESOLVE）RUNNING 超时自动收尾/失败，按分钟计算； + * 与删除品牌共用同一定时调度。 */ - private long productRiskStaleTimeoutHours = 48; + private long productRiskStaleTimeoutMinutes = 1; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java new file mode 100644 index 0000000..f93fc47 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java @@ -0,0 +1,24 @@ +package com.nanri.aiimage.config; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +@Configuration +@Slf4j +public class SchedulingConfig { + + @Bean + public TaskScheduler taskScheduler() { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(4); + scheduler.setThreadNamePrefix("aiimage-scheduling-"); + scheduler.setWaitForTasksToCompleteOnShutdown(true); + scheduler.setAwaitTerminationSeconds(30); + scheduler.setErrorHandler(ex -> log.warn("[scheduling] task execution failed: {}", ex.getMessage(), ex)); + scheduler.initialize(); + return scheduler; + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java new file mode 100644 index 0000000..4cbf6cc --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java @@ -0,0 +1,21 @@ +package com.nanri.aiimage.modules.brand.model.dto; + +import lombok.Data; + +import java.util.ArrayList; +import java.util.List; + +@Data +public class BrandFileAggregateCacheDto { + private String fileUrl; + private String originalFilename; + private String relativePath; + private String mainSheetName; + private Integer chunkTotal; + private Integer totalLines; + private Integer receivedChunkCount = 0; + private Integer processedLineCount = 0; + private Boolean completed = false; + private List invalidBrands = new ArrayList<>(); + private List queryFailedBrands = new ArrayList<>(); +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java index 613bd28..b5bdd07 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java @@ -5,10 +5,12 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.config.BrandProgressProperties; import com.nanri.aiimage.modules.brand.model.dto.BrandCrawlResultFileDto; -import lombok.RequiredArgsConstructor; +import com.nanri.aiimage.modules.brand.model.dto.BrandFileAggregateCacheDto; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; import java.time.Duration; import java.time.Instant; import java.util.ArrayList; @@ -17,18 +19,26 @@ import java.util.List; import java.util.Map; @Service -@RequiredArgsConstructor public class BrandTaskProgressCacheService { public static final String PHASE_CRAWLING = "crawling"; public static final String PHASE_ASSEMBLING = "assembling"; public static final String PHASE_UPLOADING = "uploading"; public static final String PHASE_FAILED = "failed"; + private static final Duration FINALIZE_LOCK_TTL = Duration.ofMinutes(10); private final StringRedisTemplate stringRedisTemplate; private final BrandProgressProperties brandProgressProperties; private final ObjectMapper objectMapper; + public BrandTaskProgressCacheService(StringRedisTemplate stringRedisTemplate, + BrandProgressProperties brandProgressProperties, + ObjectMapper objectMapper) { + this.stringRedisTemplate = stringRedisTemplate; + this.brandProgressProperties = brandProgressProperties; + this.objectMapper = objectMapper; + } + public void saveProgressFromResult(Long taskId, String fileUrl, int fileIndex, @@ -51,7 +61,7 @@ public class BrandTaskProgressCacheService { values.put("updated_at", now); values.put("last_heartbeat_at", now); stringRedisTemplate.opsForHash().putAll(key, values); - stringRedisTemplate.expire(key, Duration.ofHours(brandProgressProperties.getTtlHours())); + stringRedisTemplate.expire(key, ttl()); } public void updatePhase(Long taskId, String phase, int finishedFiles, int fileTotal) { @@ -64,7 +74,7 @@ public class BrandTaskProgressCacheService { values.put("updated_at", now); values.put("last_heartbeat_at", now); stringRedisTemplate.opsForHash().putAll(key, values); - stringRedisTemplate.expire(key, Duration.ofHours(brandProgressProperties.getTtlHours())); + stringRedisTemplate.expire(key, ttl()); } public void markFailed(Long taskId, String message) { @@ -84,7 +94,7 @@ public class BrandTaskProgressCacheService { public void saveParsedPayload(Long taskId, Object payload) { try { - stringRedisTemplate.opsForValue().set(buildPayloadKey(taskId), objectMapper.writeValueAsString(payload), Duration.ofHours(brandProgressProperties.getTtlHours())); + stringRedisTemplate.opsForValue().set(buildPayloadKey(taskId), objectMapper.writeValueAsString(payload), ttl()); } catch (Exception ex) { throw new BusinessException("暂存品牌原始数据失败"); } @@ -102,67 +112,279 @@ public class BrandTaskProgressCacheService { } } - public void delete(Long taskId) { - stringRedisTemplate.delete(buildKey(taskId)); - stringRedisTemplate.delete(buildResultKey(taskId)); - stringRedisTemplate.delete(buildPayloadKey(taskId)); - } - - public void mergeResultChunks(Long taskId, List incomingFiles) { - String resultKey = buildResultKey(taskId); - for (BrandCrawlResultFileDto file : incomingFiles) { - if (file.getChunkIndex() == null || file.getChunkTotal() == null || file.getChunkIndex() <= 0 || file.getChunkTotal() <= 0 || file.getChunkIndex() > file.getChunkTotal()) { - throw new BusinessException("分片参数不合法"); + public ChunkStoreResult storeChunk(Long taskId, BrandCrawlResultFileDto file) { + validateChunk(file); + String fileUrl = normalizeFileUrl(file.getFileUrl()); + String chunkDigest = digestChunk(file); + String digestField = String.valueOf(file.getChunkIndex()); + String digestKey = buildChunkDigestKey(taskId, fileUrl); + Object existingDigest = stringRedisTemplate.opsForHash().get(digestKey, digestField); + if (existingDigest instanceof String existing && !existing.isBlank()) { + refreshFileKeys(taskId, fileUrl); + if (existing.equals(chunkDigest)) { + BrandFileAggregateCacheDto aggregate = getFileAggregate(taskId, fileUrl); + int finishedFiles = countCompletedFiles(taskId); + return new ChunkStoreResult(false, false, finishedFiles, aggregate); } - String field = buildChunkField(file.getFileUrl(), file.getChunkIndex()); - try { - stringRedisTemplate.opsForHash().put(resultKey, field, objectMapper.writeValueAsString(file)); - } catch (Exception ex) { - throw new BusinessException("暂存结果分片失败"); + throw new BusinessException("同一分片重复提交但内容不一致: " + fileUrl + "#" + file.getChunkIndex()); + } + + BrandFileAggregateCacheDto aggregate = getFileAggregate(taskId, fileUrl); + if (aggregate == null) { + aggregate = createAggregate(file); + } else { + ensureChunkConsistency(aggregate, file); + } + mergeAggregate(aggregate, file); + + stringRedisTemplate.opsForSet().add(buildChunkReceiptKey(taskId, fileUrl), digestField); + Long receivedSize = stringRedisTemplate.opsForSet().size(buildChunkReceiptKey(taskId, fileUrl)); + int receivedCount = receivedSize == null ? 0 : receivedSize.intValue(); + if (receivedCount <= 0) { + receivedCount = aggregate.getReceivedChunkCount() == null ? 0 : aggregate.getReceivedChunkCount(); + } + aggregate.setReceivedChunkCount(Math.max(receivedCount, aggregate.getReceivedChunkCount() == null ? 0 : aggregate.getReceivedChunkCount())); + boolean fileCompleted = aggregate.getChunkTotal() != null + && aggregate.getChunkTotal() > 0 + && aggregate.getReceivedChunkCount() >= aggregate.getChunkTotal(); + aggregate.setCompleted(fileCompleted); + saveFileAggregate(taskId, aggregate); + stringRedisTemplate.opsForHash().put(digestKey, digestField, chunkDigest); + refreshFileKeys(taskId, fileUrl); + + boolean newlyCompleted = false; + int finishedFiles = countCompletedFiles(taskId); + if (fileCompleted) { + Long added = stringRedisTemplate.opsForSet().add(buildCompletedFilesKey(taskId), fileUrl); + refreshTaskKey(buildCompletedFilesKey(taskId)); + newlyCompleted = added != null && added > 0; + if (newlyCompleted) { + finishedFiles = countCompletedFiles(taskId); } } - stringRedisTemplate.expire(resultKey, Duration.ofHours(brandProgressProperties.getTtlHours())); + return new ChunkStoreResult(true, newlyCompleted, finishedFiles, aggregate); } - public Map> groupResultChunksByFile(Long taskId) { - Map stored = stringRedisTemplate.opsForHash().entries(buildResultKey(taskId)); - Map> grouped = new LinkedHashMap<>(); - for (Object value : stored.values()) { - if (!(value instanceof String raw) || raw.isBlank()) { + public BrandFileAggregateCacheDto getFileAggregate(Long taskId, String fileUrl) { + String normalizedFileUrl = normalizeFileUrl(fileUrl); + Object raw = stringRedisTemplate.opsForHash().get(buildFileAggregateKey(taskId), normalizedFileUrl); + if (!(raw instanceof String json) || json.isBlank()) { + return null; + } + try { + return objectMapper.readValue(json, BrandFileAggregateCacheDto.class); + } catch (Exception ex) { + throw new BusinessException("读取品牌文件聚合缓存失败"); + } + } + + public Map getAllFileAggregates(Long taskId) { + Map stored = stringRedisTemplate.opsForHash().entries(buildFileAggregateKey(taskId)); + Map result = new LinkedHashMap<>(); + for (Map.Entry entry : stored.entrySet()) { + if (!(entry.getKey() instanceof String fileUrl) || !(entry.getValue() instanceof String json) || json.isBlank()) { continue; } try { - BrandCrawlResultFileDto file = objectMapper.readValue(raw, BrandCrawlResultFileDto.class); - grouped.computeIfAbsent(file.getFileUrl(), ignored -> new ArrayList<>()).add(file); + result.put(fileUrl, objectMapper.readValue(json, BrandFileAggregateCacheDto.class)); } catch (Exception ignored) { } } - grouped.values().forEach(list -> list.sort(java.util.Comparator.comparing(BrandCrawlResultFileDto::getChunkIndex))); - return grouped; + return result; } - public void clearResultChunks(Long taskId) { - stringRedisTemplate.delete(buildResultKey(taskId)); + public int countCompletedFiles(Long taskId) { + Long count = stringRedisTemplate.opsForSet().size(buildCompletedFilesKey(taskId)); + return count == null ? 0 : count.intValue(); + } + + public boolean acquireFinalizeLock(Long taskId) { + Boolean ok = stringRedisTemplate.opsForValue().setIfAbsent(buildFinalizeLockKey(taskId), String.valueOf(Instant.now().toEpochMilli()), FINALIZE_LOCK_TTL); + return Boolean.TRUE.equals(ok); + } + + public void releaseFinalizeLock(Long taskId) { + stringRedisTemplate.delete(buildFinalizeLockKey(taskId)); + } + + public void delete(Long taskId) { + stringRedisTemplate.delete(buildKey(taskId)); + stringRedisTemplate.delete(buildPayloadKey(taskId)); + stringRedisTemplate.delete(buildFileAggregateKey(taskId)); + stringRedisTemplate.delete(buildCompletedFilesKey(taskId)); + stringRedisTemplate.delete(buildFinalizeLockKey(taskId)); + stringRedisTemplate.delete(buildLegacyResultKey(taskId)); + } + + public void deleteFileState(Long taskId, String fileUrl) { + String normalizedFileUrl = normalizeFileUrl(fileUrl); + stringRedisTemplate.opsForHash().delete(buildFileAggregateKey(taskId), normalizedFileUrl); + stringRedisTemplate.delete(buildChunkReceiptKey(taskId, normalizedFileUrl)); + stringRedisTemplate.delete(buildChunkDigestKey(taskId, normalizedFileUrl)); } public String buildKey(Long taskId) { return "brand:task:progress:" + taskId; } - private String buildResultKey(Long taskId) { - return "brand:task:result-chunks:" + taskId; + public long getHeartbeatTimeoutMinutes() { + return brandProgressProperties.getHeartbeatTimeoutMinutes(); } - private String buildChunkField(String fileUrl, Integer chunkIndex) { - return fileUrl + "#" + chunkIndex; + private void saveFileAggregate(Long taskId, BrandFileAggregateCacheDto aggregate) { + try { + stringRedisTemplate.opsForHash().put(buildFileAggregateKey(taskId), normalizeFileUrl(aggregate.getFileUrl()), objectMapper.writeValueAsString(aggregate)); + refreshTaskKey(buildFileAggregateKey(taskId)); + } catch (Exception ex) { + throw new BusinessException("暂存品牌文件聚合结果失败"); + } + } + + private void validateChunk(BrandCrawlResultFileDto file) { + if (file == null) { + throw new BusinessException("分片不能为空"); + } + if (file.getChunkIndex() == null || file.getChunkTotal() == null || file.getChunkIndex() <= 0 || file.getChunkTotal() <= 0 || file.getChunkIndex() > file.getChunkTotal()) { + throw new BusinessException("分片参数不合法"); + } + if (file.getFileUrl() == null || file.getFileUrl().isBlank()) { + throw new BusinessException("fileUrl 不能为空"); + } + } + + private BrandFileAggregateCacheDto createAggregate(BrandCrawlResultFileDto file) { + BrandFileAggregateCacheDto aggregate = new BrandFileAggregateCacheDto(); + aggregate.setFileUrl(normalizeFileUrl(file.getFileUrl())); + aggregate.setOriginalFilename(file.getOriginalFilename()); + aggregate.setRelativePath(file.getRelativePath()); + aggregate.setMainSheetName(file.getMainSheetName()); + aggregate.setChunkTotal(file.getChunkTotal()); + aggregate.setTotalLines(safePositive(file.getTotalLines())); + aggregate.setReceivedChunkCount(0); + aggregate.setProcessedLineCount(0); + aggregate.setCompleted(false); + aggregate.setInvalidBrands(new ArrayList<>()); + aggregate.setQueryFailedBrands(new ArrayList<>()); + return aggregate; + } + + private void ensureChunkConsistency(BrandFileAggregateCacheDto aggregate, BrandCrawlResultFileDto file) { + if (aggregate.getChunkTotal() != null && file.getChunkTotal() != null && !aggregate.getChunkTotal().equals(file.getChunkTotal())) { + throw new BusinessException("同一文件的 chunkTotal 不一致: " + aggregate.getFileUrl()); + } + if (aggregate.getChunkTotal() == null) { + aggregate.setChunkTotal(file.getChunkTotal()); + } + int incomingTotalLines = safePositive(file.getTotalLines()); + if (incomingTotalLines > 0) { + aggregate.setTotalLines(Math.max(safePositive(aggregate.getTotalLines()), incomingTotalLines)); + } + if ((aggregate.getOriginalFilename() == null || aggregate.getOriginalFilename().isBlank()) && file.getOriginalFilename() != null && !file.getOriginalFilename().isBlank()) { + aggregate.setOriginalFilename(file.getOriginalFilename()); + } + if ((aggregate.getRelativePath() == null || aggregate.getRelativePath().isBlank()) && file.getRelativePath() != null && !file.getRelativePath().isBlank()) { + aggregate.setRelativePath(file.getRelativePath()); + } + if ((aggregate.getMainSheetName() == null || aggregate.getMainSheetName().isBlank()) && file.getMainSheetName() != null && !file.getMainSheetName().isBlank()) { + aggregate.setMainSheetName(file.getMainSheetName()); + } + } + + private void mergeAggregate(BrandFileAggregateCacheDto aggregate, BrandCrawlResultFileDto file) { + int processed = safePositive(aggregate.getProcessedLineCount()) + + sizeOf(file.getKeptRows()) + + sizeOf(file.getInvalidBrands()) + + sizeOf(file.getQueryFailedBrands()); + aggregate.setProcessedLineCount(processed); + if (file.getInvalidBrands() != null && !file.getInvalidBrands().isEmpty()) { + aggregate.getInvalidBrands().addAll(file.getInvalidBrands()); + } + if (file.getQueryFailedBrands() != null && !file.getQueryFailedBrands().isEmpty()) { + aggregate.getQueryFailedBrands().addAll(file.getQueryFailedBrands()); + } + } + + private int sizeOf(List list) { + return list == null ? 0 : list.size(); + } + + private int safePositive(Integer value) { + return value == null || value <= 0 ? 0 : value; + } + + private String digestChunk(BrandCrawlResultFileDto file) { + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + byte[] bytes = digest.digest(objectMapper.writeValueAsString(file).getBytes(StandardCharsets.UTF_8)); + StringBuilder sb = new StringBuilder(bytes.length * 2); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } catch (Exception ex) { + throw new BusinessException("计算品牌结果分片摘要失败"); + } + } + + private void refreshFileKeys(Long taskId, String fileUrl) { + refreshTaskKey(buildChunkReceiptKey(taskId, fileUrl)); + refreshTaskKey(buildChunkDigestKey(taskId, fileUrl)); + refreshTaskKey(buildFileAggregateKey(taskId)); + } + + private void refreshTaskKey(String key) { + stringRedisTemplate.expire(key, ttl()); + } + + private Duration ttl() { + return Duration.ofHours(brandProgressProperties.getTtlHours()); } private String buildPayloadKey(Long taskId) { return "brand:task:parsed-payload:" + taskId; } - public long getHeartbeatTimeoutMinutes() { - return brandProgressProperties.getHeartbeatTimeoutMinutes(); + private String buildFileAggregateKey(Long taskId) { + return "brand:task:file-aggregate:" + taskId; + } + + private String buildChunkReceiptKey(Long taskId, String fileUrl) { + return "brand:task:file-chunks:" + taskId + ":" + hashFileUrl(fileUrl); + } + + private String buildChunkDigestKey(Long taskId, String fileUrl) { + return "brand:task:file-chunk-digests:" + taskId + ":" + hashFileUrl(fileUrl); + } + + private String buildCompletedFilesKey(Long taskId) { + return "brand:task:completed-files:" + taskId; + } + + private String buildFinalizeLockKey(Long taskId) { + return "brand:task:finalizing:" + taskId; + } + + private String buildLegacyResultKey(Long taskId) { + return "brand:task:result-chunks:" + taskId; + } + + private String normalizeFileUrl(String fileUrl) { + return fileUrl == null ? "" : fileUrl.trim(); + } + + private String hashFileUrl(String fileUrl) { + String normalized = normalizeFileUrl(fileUrl); + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + byte[] bytes = digest.digest(normalized.getBytes(StandardCharsets.UTF_8)); + StringBuilder sb = new StringBuilder(bytes.length * 2); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } catch (Exception ex) { + throw new BusinessException("生成文件缓存键失败"); + } } private String normalizePhase(String phase) { @@ -179,4 +401,7 @@ public class BrandTaskProgressCacheService { private String blankToEmpty(String value) { return value == null ? "" : value.trim(); } + + public record ChunkStoreResult(boolean stored, boolean newlyCompletedFile, int finishedFiles, BrandFileAggregateCacheDto aggregate) { + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index af865c7..7c7e77d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -12,6 +12,7 @@ import com.nanri.aiimage.config.StorageProperties; import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper; import com.nanri.aiimage.modules.brand.model.dto.BrandCrawlResultFileDto; import com.nanri.aiimage.modules.brand.model.dto.BrandCrawlResultRequest; +import com.nanri.aiimage.modules.brand.model.dto.BrandFileAggregateCacheDto; import com.nanri.aiimage.modules.brand.model.dto.BrandInvalidBrandDto; import com.nanri.aiimage.modules.brand.model.dto.BrandParsedFileCacheDto; import com.nanri.aiimage.modules.brand.model.dto.BrandQueryFailedDto; @@ -30,6 +31,7 @@ import com.nanri.aiimage.modules.brand.model.vo.LegacyBrandTaskItemVo; import com.nanri.aiimage.modules.brand.model.vo.LegacyBrandTaskListVo; import com.nanri.aiimage.modules.file.service.oss.OssStorageService; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.usermodel.Cell; import org.apache.poi.ss.usermodel.DataFormatter; import org.apache.poi.ss.usermodel.Row; @@ -63,6 +65,7 @@ import java.util.zip.ZipOutputStream; @Service @RequiredArgsConstructor +@Slf4j public class BrandTaskService { private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm"); @@ -205,18 +208,19 @@ public class BrandTaskService { } public void submitCrawlResult(Long taskId, BrandCrawlResultRequest request) { - BrandCrawlTaskEntity task = requireTask(taskId); - if (STATUS_CANCELLED.equalsIgnoreCase(blankToDefault(task.getStatus(), STATUS_PENDING))) { - throw new BusinessException("任务已取消"); - } + long startedAt = System.currentTimeMillis(); + BrandCrawlTaskEntity task = requireActiveTask(taskId); List sourceFiles = parseSourceFiles(task.getFilePaths()); if (sourceFiles.isEmpty()) { throw new BusinessException("任务没有源文件"); } Map sourceByUrl = new LinkedHashMap<>(); - for (BrandSourceFileDto file : sourceFiles) { + Map 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 cachedFiles = brandTaskProgressCacheService.getParsedPayload(taskId, new TypeReference>() { @@ -236,82 +240,67 @@ public class BrandTaskService { ensureNoDuplicateResultFiles(resultFiles); int totalCount = sourceFiles.size(); - int started = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() - .eq(BrandCrawlTaskEntity::getId, taskId) - .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) - .set(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING) - .set(BrandCrawlTaskEntity::getProgressTotal, totalCount) - .set(BrandCrawlTaskEntity::getUpdatedAt, LocalDateTime.now()) - .set(BrandCrawlTaskEntity::getErrorMessage, null)); - if (started == 0) { - throw new BusinessException("任务已取消"); + markTaskRunning(taskId, totalCount); + + log.info("[brand-submit] taskId={} chunks={} files={} thread={}", + taskId, + resultFiles.size(), + totalCount, + Thread.currentThread().getName()); + + int finishedCount = brandTaskProgressCacheService.countCompletedFiles(taskId); + for (BrandCrawlResultFileDto resultFile : resultFiles) { + String fileUrl = blankToNull(resultFile.getFileUrl()); + if (fileUrl == null) { + throw new BusinessException("fileUrl 不能为空"); + } + BrandSourceFileDto sourceFile = sourceByUrl.get(fileUrl); + if (sourceFile == null) { + throw new BusinessException("存在未知 fileUrl: " + fileUrl); + } + BrandParsedFileCacheDto cachedFile = cachedByUrl.get(fileUrl); + if (cachedFile == null) { + throw new BusinessException("缺少原始缓存数据: " + 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 超过原始文件行数: " + fileUrl); + } + BrandTaskProgressCacheService.ChunkStoreResult storeResult = brandTaskProgressCacheService.storeChunk(taskId, resultFile); + finishedCount = Math.max(finishedCount, storeResult.finishedFiles()); + log.info("[brand-submit] taskId={} fileUrl={} chunk={}/{} stored={} newlyCompletedFile={} finishedFiles={}", + taskId, + fileUrl, + resultFile.getChunkIndex(), + resultFile.getChunkTotal(), + storeResult.stored(), + storeResult.newlyCompletedFile(), + finishedCount); + BrandFileAggregateCacheDto aggregate = storeResult.aggregate(); + 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); } - brandTaskProgressCacheService.mergeResultChunks(taskId, resultFiles); - Map> groupedResults = brandTaskProgressCacheService.groupResultChunksByFile(taskId); - int finishedCount = countCompletedFiles(groupedResults); - updateResultDrivenProgress(taskId, sourceFiles, sourceByUrl, groupedResults, finishedCount, totalCount); - updateTaskProgress(taskId, finishedCount, totalCount); - - if (finishedCount < totalCount) { - return; - } - - brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_ASSEMBLING, finishedCount, totalCount); - - File outputDir = FileUtil.mkdir(FileUtil.file(storageProperties.getLocalTempDir(), "brand-result", String.valueOf(taskId))); - List outputEntries = new ArrayList<>(); - - try { - for (List fileChunks : groupedResults.values()) { - BrandCrawlResultFileDto resultFile = mergeChunks(fileChunks); - BrandSourceFileDto sourceFile = sourceByUrl.get(resultFile.getFileUrl()); - if (sourceFile == null) { - throw new BusinessException("存在未知 fileUrl: " + resultFile.getFileUrl()); - } - BrandParsedFileCacheDto cachedFile = cachedByUrl.get(resultFile.getFileUrl()); - if (cachedFile == null) { - throw new BusinessException("缺少原始缓存数据: " + resultFile.getFileUrl()); - } - File sourceLocalFile = resolveSourceFile(sourceFile); - String originalFilename = resolveOriginalFilename(sourceFile, sourceLocalFile); - File outputFile = buildNamedOutputFile(outputDir, buildResultFilename(originalFilename)); - writeBrandWorkbook(outputFile, request.getStrategy(), cachedFile, resultFile); - outputEntries.add(new OutputEntry( - sourceLocalFile, - originalFilename, - outputFile, - originalFilename)); - } - - brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_UPLOADING, finishedCount, totalCount); - Map resultPaths = buildAndUploadResult(taskId, outputEntries); - int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() - .eq(BrandCrawlTaskEntity::getId, taskId) - .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) - .set(BrandCrawlTaskEntity::getStatus, STATUS_SUCCESS) - .set(BrandCrawlTaskEntity::getResultPaths, JSONUtil.toJsonStr(resultPaths)) - .set(BrandCrawlTaskEntity::getProgressCurrent, finishedCount) - .set(BrandCrawlTaskEntity::getProgressTotal, totalCount) - .set(BrandCrawlTaskEntity::getErrorMessage, null)); - if (updated == 0) { - throw new BusinessException("任务已取消"); - } - brandTaskProgressCacheService.delete(taskId); - } catch (Exception ex) { - brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() - .eq(BrandCrawlTaskEntity::getId, taskId) - .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) - .set(BrandCrawlTaskEntity::getStatus, STATUS_FAILED) - .set(BrandCrawlTaskEntity::getProgressCurrent, finishedCount) - .set(BrandCrawlTaskEntity::getProgressTotal, totalCount) - .set(BrandCrawlTaskEntity::getErrorMessage, ex.getMessage())); - brandTaskProgressCacheService.markFailed(taskId, ex.getMessage()); - if (ex instanceof BusinessException businessException) { - throw businessException; - } - throw new BusinessException(ex.getMessage()); - } + 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) { @@ -498,6 +487,139 @@ public class BrandTaskService { return Objects.equals(taskType, 1) ? 1 : 2; } + private void markTaskRunning(Long taskId, int totalCount) { + int started = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(BrandCrawlTaskEntity::getId, taskId) + .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) + .set(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING) + .set(BrandCrawlTaskEntity::getProgressTotal, totalCount) + .set(BrandCrawlTaskEntity::getUpdatedAt, LocalDateTime.now()) + .set(BrandCrawlTaskEntity::getErrorMessage, null)); + if (started == 0) { + throw new BusinessException("任务已取消"); + } + } + + private BrandCrawlTaskEntity requireActiveTask(Long taskId) { + BrandCrawlTaskEntity task = requireTask(taskId); + String status = blankToDefault(task.getStatus(), STATUS_PENDING); + if (STATUS_CANCELLED.equalsIgnoreCase(status)) { + throw new BusinessException("任务已取消"); + } + if (STATUS_SUCCESS.equalsIgnoreCase(status) || STATUS_FAILED.equalsIgnoreCase(status)) { + throw new BusinessException("任务已结束，不能继续提交结果"); + } + return task; + } + + private void tryFinalizeTask(Long taskId, + String strategy, + List sourceFiles, + Map cachedByUrl, + int finishedCount, + int totalCount) { + if (finishedCount < totalCount) { + return; + } + log.info("[brand-finalize] taskId={} all files completed, trying finalize lock", taskId); + if (!brandTaskProgressCacheService.acquireFinalizeLock(taskId)) { + log.info("[brand-finalize] taskId={} finalize lock busy, skip", taskId); + return; + } + try { + finalizeTask(taskId, strategy, sourceFiles, cachedByUrl, totalCount); + } finally { + brandTaskProgressCacheService.releaseFinalizeLock(taskId); + } + } + + private void finalizeTask(Long taskId, + String strategy, + List sourceFiles, + Map cachedByUrl, + int totalCount) { + long startedAt = System.currentTimeMillis(); + BrandCrawlTaskEntity freshTask = requireTask(taskId); + String freshStatus = blankToDefault(freshTask.getStatus(), STATUS_PENDING); + if (STATUS_CANCELLED.equalsIgnoreCase(freshStatus)) { + throw new BusinessException("任务已取消"); + } + if (STATUS_SUCCESS.equalsIgnoreCase(freshStatus)) { + return; + } + if (STATUS_FAILED.equalsIgnoreCase(freshStatus)) { + throw new BusinessException("任务已结束，不能继续组装结果"); + } + Map aggregates = brandTaskProgressCacheService.getAllFileAggregates(taskId); + log.info("[brand-finalize] taskId={} aggregateFiles={} expectedFiles={} thread={}", + taskId, + aggregates.size(), + totalCount, + Thread.currentThread().getName()); + if (aggregates.size() < totalCount) { + return; + } + for (BrandSourceFileDto sourceFile : sourceFiles) { + BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl()); + if (aggregate == null || !Boolean.TRUE.equals(aggregate.getCompleted())) { + return; + } + } + + brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_ASSEMBLING, totalCount, totalCount); + File outputDir = FileUtil.mkdir(FileUtil.file(storageProperties.getLocalTempDir(), "brand-result", String.valueOf(taskId))); + List outputEntries = new ArrayList<>(); + try { + for (BrandSourceFileDto sourceFile : sourceFiles) { + BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl()); + BrandParsedFileCacheDto cachedFile = cachedByUrl.get(sourceFile.getFileUrl()); + if (cachedFile == null) { + throw new BusinessException("缺少原始缓存数据: " + sourceFile.getFileUrl()); + } + File sourceLocalFile = resolveSourceFile(sourceFile); + String originalFilename = resolveOriginalFilename(sourceFile, sourceLocalFile); + File outputFile = buildNamedOutputFile(outputDir, buildResultFilename(originalFilename)); + writeBrandWorkbook(outputFile, strategy, cachedFile, aggregate); + outputEntries.add(new OutputEntry(sourceLocalFile, originalFilename, outputFile, originalFilename)); + } + brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_UPLOADING, totalCount, totalCount); + Map resultPaths = buildAndUploadResult(taskId, outputEntries); + int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(BrandCrawlTaskEntity::getId, taskId) + .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) + .set(BrandCrawlTaskEntity::getStatus, STATUS_SUCCESS) + .set(BrandCrawlTaskEntity::getResultPaths, JSONUtil.toJsonStr(resultPaths)) + .set(BrandCrawlTaskEntity::getProgressCurrent, totalCount) + .set(BrandCrawlTaskEntity::getProgressTotal, totalCount) + .set(BrandCrawlTaskEntity::getErrorMessage, null)); + if (updated == 0) { + throw new BusinessException("任务已取消"); + } + for (BrandSourceFileDto sourceFile : sourceFiles) { + brandTaskProgressCacheService.deleteFileState(taskId, sourceFile.getFileUrl()); + } + brandTaskProgressCacheService.delete(taskId); + log.info("[brand-finalize] taskId={} finalized success files={} elapsedMs={}", + taskId, + totalCount, + System.currentTimeMillis() - startedAt); + } catch (Exception ex) { + log.warn("[brand-finalize] taskId={} finalize failed msg={}", taskId, ex.getMessage()); + brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(BrandCrawlTaskEntity::getId, taskId) + .ne(BrandCrawlTaskEntity::getStatus, STATUS_CANCELLED) + .set(BrandCrawlTaskEntity::getStatus, STATUS_FAILED) + .set(BrandCrawlTaskEntity::getProgressCurrent, brandTaskProgressCacheService.countCompletedFiles(taskId)) + .set(BrandCrawlTaskEntity::getProgressTotal, totalCount) + .set(BrandCrawlTaskEntity::getErrorMessage, ex.getMessage())); + brandTaskProgressCacheService.markFailed(taskId, ex.getMessage()); + if (ex instanceof BusinessException businessException) { + throw businessException; + } + throw new BusinessException(ex.getMessage()); + } + } + private void ensureNoDuplicateResultFiles(List resultFiles) { Set seen = new LinkedHashSet<>(); for (BrandCrawlResultFileDto resultFile : resultFiles) { @@ -505,8 +627,12 @@ public class BrandTaskService { if (fileUrl == null) { throw new BusinessException("fileUrl 不能为空"); } - if (!seen.add(fileUrl)) { - throw new BusinessException("存在重复 fileUrl: " + fileUrl); + if (resultFile.getChunkIndex() == null || resultFile.getChunkIndex() <= 0) { + throw new BusinessException("chunkIndex 不合法"); + } + String chunkKey = fileUrl + "#" + resultFile.getChunkIndex(); + if (!seen.add(chunkKey)) { + throw new BusinessException("存在重复分片: " + chunkKey); } } } @@ -522,81 +648,6 @@ public class BrandTaskService { .set(BrandCrawlTaskEntity::getErrorMessage, null)); } - private void updateResultDrivenProgress(Long taskId, - List sourceFiles, - Map sourceByUrl, - Map> groupedResults, - int finishedCount, - int totalCount) { - if (groupedResults.isEmpty()) { - return; - } - String currentFileUrl = null; - List currentChunks = null; - int currentFileIndex = 0; - for (int i = sourceFiles.size() - 1; i >= 0; i--) { - BrandSourceFileDto sourceFile = sourceFiles.get(i); - List chunks = groupedResults.get(sourceFile.getFileUrl()); - if (chunks == null || chunks.isEmpty()) { - continue; - } - currentFileUrl = sourceFile.getFileUrl(); - currentChunks = chunks; - currentFileIndex = i + 1; - break; - } - if (currentFileUrl == null || currentChunks == null) { - return; - } - BrandSourceFileDto sourceFile = sourceByUrl.get(currentFileUrl); - String fileName = sourceFile == null ? "" : blankToDefault(sourceFile.getOriginalFilename(), sourceFile.getFileUrl()); - int currentLine = countProcessedLines(currentChunks); - int totalLines = Math.max(inferTotalLines(currentChunks), currentLine); - brandTaskProgressCacheService.saveProgressFromResult(taskId, - currentFileUrl, - currentFileIndex, - totalCount, - fileName, - currentLine, - totalLines, - finishedCount); - } - - private int countProcessedLines(List chunks) { - int currentLine = 0; - for (BrandCrawlResultFileDto chunk : chunks) { - if (chunk.getKeptRows() != null) { - currentLine += chunk.getKeptRows().size(); - } - if (chunk.getInvalidBrands() != null) { - currentLine += chunk.getInvalidBrands().size(); - } - if (chunk.getQueryFailedBrands() != null) { - currentLine += chunk.getQueryFailedBrands().size(); - } - } - return currentLine; - } - - private int inferTotalLines(List chunks) { - if (chunks == null || chunks.isEmpty()) { - return 0; - } - BrandCrawlResultFileDto last = chunks.get(chunks.size() - 1); - if (last.getTotalLines() != null && last.getTotalLines() > 0) { - return last.getTotalLines(); - } - if (last.getChunkTotal() == null || last.getChunkTotal() <= 0) { - return countProcessedLines(chunks); - } - int processed = countProcessedLines(chunks); - if (Objects.equals(chunks.size(), last.getChunkTotal())) { - return processed; - } - int averagePerChunk = Math.max(processed / Math.max(chunks.size(), 1), 1); - return averagePerChunk * last.getChunkTotal(); - } - private String buildDesc(String strategy, List files) { String joined = files.stream() .map(file -> blankToDefault(file.getOriginalFilename(), file.getFileUrl())) @@ -769,7 +820,7 @@ public class BrandTaskService { private void writeBrandWorkbook(File outputFile, String strategy, BrandParsedFileCacheDto cachedFile, - BrandCrawlResultFileDto resultFile) throws IOException { + BrandFileAggregateCacheDto resultFile) throws IOException { String actualStrategy = normalizeStrategy(strategy); try (XSSFWorkbook workbook = new XSSFWorkbook()) { String mainSheetName = blankToDefault(cachedFile.getSheetName(), "Sheet1"); @@ -986,52 +1037,6 @@ public class BrandTaskService { return vo; } - private int countCompletedFiles(Map> groupedResults) { - int count = 0; - for (List chunks : groupedResults.values()) { - if (!chunks.isEmpty() && Objects.equals(chunks.size(), chunks.get(0).getChunkTotal())) { - count++; - } - } - return count; - } - - private BrandCrawlResultFileDto mergeChunks(List chunks) { - if (chunks == null || chunks.isEmpty()) { - throw new BusinessException("结果分片为空"); - } - BrandCrawlResultFileDto first = chunks.get(0); - if (!Objects.equals(chunks.size(), first.getChunkTotal())) { - throw new BusinessException("文件结果分片未完成"); - } - BrandCrawlResultFileDto merged = new BrandCrawlResultFileDto(); - merged.setFileUrl(first.getFileUrl()); - merged.setOriginalFilename(first.getOriginalFilename()); - merged.setRelativePath(first.getRelativePath()); - merged.setMainSheetName(first.getMainSheetName()); - merged.setChunkIndex(first.getChunkIndex()); - merged.setChunkTotal(first.getChunkTotal()); - merged.setTotalLines(first.getTotalLines()); - List keptRows = new ArrayList<>(); - List invalidBrands = new ArrayList<>(); - List queryFailedBrands = new ArrayList<>(); - for (BrandCrawlResultFileDto chunk : chunks) { - if (chunk.getKeptRows() != null) { - keptRows.addAll(chunk.getKeptRows()); - } - if (chunk.getInvalidBrands() != null) { - invalidBrands.addAll(chunk.getInvalidBrands()); - } - if (chunk.getQueryFailedBrands() != null) { - queryFailedBrands.addAll(chunk.getQueryFailedBrands()); - } - } - merged.setKeptRows(keptRows); - merged.setInvalidBrands(invalidBrands); - merged.setQueryFailedBrands(queryFailedBrands); - return merged; - } - private Integer parseInteger(Object value) { if (value == null) { return 0; 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 index 891b2de..c9fa8e8 100644 --- 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 @@ -3,9 +3,11 @@ package com.nanri.aiimage.modules.deletebrand.service; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.nanri.aiimage.config.DeleteBrandProgressProperties; +import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; @@ -19,6 +21,7 @@ import java.util.Map; */ @Service @RequiredArgsConstructor +@Slf4j public class DeleteBrandStaleTaskService { private static final String MODULE_TYPE_DELETE_BRAND = "DELETE_BRAND"; @@ -27,12 +30,22 @@ public class DeleteBrandStaleTaskService { private final FileTaskMapper fileTaskMapper; private final DeleteBrandTaskCacheService deleteBrandTaskCacheService; private final DeleteBrandRunService deleteBrandRunService; + private final ProductRiskTaskService productRiskTaskService; private final DeleteBrandProgressProperties deleteBrandProgressProperties; @Scheduled(cron = "${aiimage.delete-brand-progress.stale-check-cron:0 */2 * * * *}") public void failStaleRunningTasks() { + long startedAt = System.currentTimeMillis(); + log.info("[stale-check] scan started thread={}", Thread.currentThread().getName()); failStaleDeleteBrandTasks(); - failStaleProductRiskResolveTasks(); + ProductRiskStaleCheckStats stats = failStaleProductRiskResolveTasks(); + log.info("[stale-check] product-risk summary scanned={} finalized={} failed={} skipped={} elapsedMs={} thread={}", + stats.scannedTaskCount, + stats.finalizedTaskCount, + stats.failedTaskCount, + stats.skippedTaskCount, + System.currentTimeMillis() - startedAt, + Thread.currentThread().getName()); } private void failStaleDeleteBrandTasks() { @@ -99,16 +112,32 @@ public class DeleteBrandStaleTaskService { } } - private void failStaleProductRiskResolveTasks() { - long hours = Math.max(1L, deleteBrandProgressProperties.getProductRiskStaleTimeoutHours()); - LocalDateTime threshold = LocalDateTime.now().minusHours(hours); + private ProductRiskStaleCheckStats failStaleProductRiskResolveTasks() { + ProductRiskStaleCheckStats stats = new ProductRiskStaleCheckStats(); + long minutes = Math.max(1L, deleteBrandProgressProperties.getProductRiskStaleTimeoutMinutes()); + LocalDateTime threshold = LocalDateTime.now().minusMinutes(minutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRODUCT_RISK) .eq(FileTaskEntity::getStatus, "RUNNING") .lt(FileTaskEntity::getUpdatedAt, threshold) .last("limit 200")); + stats.scannedTaskCount = runningTasks.size(); + if (runningTasks.isEmpty()) { + return stats; + } + log.info("[stale-check] product-risk candidates={} threshold={} timeoutMinutes={}", + runningTasks.size(), threshold, minutes); for (FileTaskEntity task : runningTasks) { - fileTaskMapper.update(null, new LambdaUpdateWrapper() + try { + if (productRiskTaskService.tryFinalizeTask(task.getId(), true)) { + stats.finalizedTaskCount++; + log.info("[stale-check] product-risk finalized taskId={}", task.getId()); + continue; + } + } catch (Exception ex) { + log.warn("[stale-check] product-risk finalize threw taskId={} msg={}", task.getId(), ex.getMessage()); + } + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() .eq(FileTaskEntity::getId, task.getId()) .eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRODUCT_RISK) .eq(FileTaskEntity::getStatus, "RUNNING") @@ -116,7 +145,14 @@ public class DeleteBrandStaleTaskService { .set(FileTaskEntity::getErrorMessage, "长时间未收到 Python 结果，任务已自动失败") .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()) .set(FileTaskEntity::getFinishedAt, LocalDateTime.now())); + if (updated > 0) { + stats.failedTaskCount++; + log.warn("[stale-check] product-risk failed taskId={} reason=timeout", task.getId()); + } else { + stats.skippedTaskCount++; + } } + return stats; } @Scheduled(cron = "${aiimage.delete-brand-progress.finalize-check-cron:30 */2 * * * *}") @@ -135,4 +171,11 @@ public class DeleteBrandStaleTaskService { } } } + + private static final class ProductRiskStaleCheckStats { + private int scannedTaskCount; + private int finalizedTaskCount; + private int failedTaskCount; + private int skippedTaskCount; + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskRowDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskRowDto.java index 214cf0b..a0a059f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskRowDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskRowDto.java @@ -21,7 +21,7 @@ public class ProductRiskRowDto { @JsonProperty("done") @Schema(description = "是否完成等标记列") - private String done; + private Boolean done; @JsonProperty("removeAsin") @Schema(description = "待移除 ASIN 等") diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskExcelAssemblyService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskExcelAssemblyService.java index 06e1602..301ee7c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskExcelAssemblyService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskExcelAssemblyService.java @@ -49,7 +49,7 @@ public class ProductRiskExcelAssemblyService { createTextCell(row, 0, firstNonBlank(dto == null ? null : dto.getShopName(), shopDisplayName)); createTextCell(row, 1, dto == null ? null : dto.getProductAsinSku()); createTextCell(row, 2, dto == null ? null : dto.getStatus()); - createTextCell(row, 3, dto == null ? null : dto.getDone()); + createTextCell(row, 3, dto == null || dto.getDone() == null ? null : String.valueOf(dto.getDone())); createTextCell(row, 4, dto == null ? null : dto.getRemoveAsin()); createTextCell(row, 5, dto == null ? null : dto.getRemoveStatus()); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java index 4b1a1df..fe24cc6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java @@ -441,43 +441,32 @@ public class ProductRiskTaskService { // 分次上报累积：合并到 Redis 缓存，再按“合并后是否完成”决定是否出包。 ProductRiskShopPayloadDto mergedPayload = mergeShopPayload(taskId, shopKey, payload); + boolean incomingDone = payloadHasAnyDoneTrue(payload); + int mergedRows = countPayloadRows(mergedPayload); boolean shopDone = isShopPayloadCompleted(mergedPayload); log.info("[product-risk] shop merged taskId={} shop={} incomingDone={} mergedDone={} mergedRows={} payloadCountries={}", taskId, shopKey, - payloadHasAnyDoneTrue(payload), + incomingDone, shopDone, - countPayloadRows(mergedPayload), + mergedRows, mergedPayload.getCountries() == null ? 0 : mergedPayload.getCountries().size()); + if (incomingDone && 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 { - Map> countries = excelAssemblyService.normalizeCountriesMap(mergedPayload.getCountries()); - String displayName = mergedPayload.getShopName() != null && !mergedPayload.getShopName().isBlank() - ? mergedPayload.getShopName().trim() - : shopKey; - String stem = safeFileStem(displayName); - File xlsx = FileUtil.file(workRoot, stem + ".xlsx"); - excelAssemblyService.writeWorkbook(xlsx, displayName, countries); - File zip = FileUtil.file(workRoot, stem + ".zip"); - ZipUtil.zip(zip, false, xlsx); - String objectKey = ossStorageService.uploadResultFile(zip, MODULE_TYPE); - fr.setResultFilename(stem + ".zip"); - fr.setResultFileUrl(objectKey); - fr.setResultFileSize(zip.length()); - fr.setResultContentType(CONTENT_TYPE_ZIP); - fr.setRowCount(excelAssemblyService.countRows(countries)); - fr.setSuccess(1); - fr.setErrorMessage(null); - fileResultMapper.updateById(fr); + assembleShopResult(fr, shopKey, mergedPayload, workRoot); assembledCount++; log.info("[product-risk] shop assembled taskId={} shop={} rowCount={} file={}", taskId, shopKey, fr.getRowCount(), fr.getResultFilename()); - FileUtil.del(xlsx); - FileUtil.del(zip); productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); } catch (Exception ex) { log.warn("[product-risk] shop assemble failed taskId={} shop={} msg={}", taskId, shopKey, ex.getMessage()); @@ -538,15 +527,202 @@ public class ProductRiskTaskService { fileTaskMapper.updateById(task); } + @Transactional + public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) { + if (taskId == null || taskId <= 0) { + return false; + } + FileTaskEntity task = fileTaskMapper.selectById(taskId); + if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { + return false; + } + if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) { + return true; + } + + List resultRows = fileResultMapper.selectList(new LambdaQueryWrapper() + .eq(FileResultEntity::getTaskId, taskId) + .eq(FileResultEntity::getModuleType, MODULE_TYPE) + .orderByAsc(FileResultEntity::getId)); + if (resultRows.isEmpty()) { + return false; + } + + Map cachedPayloadByShop = productRiskTaskCacheService.getAllShopMergedPayload(taskId); + File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "product-risk-result", String.valueOf(taskId))); + List batchErrors = new ArrayList<>(); + boolean changed = false; + + for (FileResultEntity fr : resultRows) { + boolean success = fr.getSuccess() != null && fr.getSuccess() == 1; + boolean failed = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank(); + if (success || failed) { + continue; + } + + String shopKey = fr.getSourceFilename(); + ProductRiskShopPayloadDto cachedPayload = shopKey == null ? null : cachedPayloadByShop.get(shopKey); + + if (cachedPayload == null) { + markResultFailed(fr, "Python interrupted before this shop finished uploading results"); + batchErrors.add(shopKey + ": Python interrupted before this shop finished uploading results"); + changed = true; + continue; + } + + if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) { + markResultFailed(fr, cachedPayload.getError()); + batchErrors.add(shopKey + ": " + cachedPayload.getError()); + productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + changed = true; + continue; + } + + if (countPayloadRows(cachedPayload) <= 0 && payloadHasAnyDoneTrue(cachedPayload)) { + markResultNoData(fr, "没有数据"); + productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + changed = true; + log.warn("[product-risk] stale finalize finished without data taskId={} shop={} fromCompensation={}", + taskId, shopKey, fromCompensation); + continue; + } + + if (countPayloadRows(cachedPayload) <= 0) { + markResultFailed(fr, "Python interrupted before any assembleable rows were uploaded"); + batchErrors.add(shopKey + ": Python interrupted before any assembleable rows were uploaded"); + productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + changed = true; + continue; + } + + try { + assembleShopResult(fr, shopKey, cachedPayload, workRoot); + productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + changed = true; + log.warn("[product-risk] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={}", + taskId, shopKey, fromCompensation, isShopPayloadCompleted(cachedPayload)); + } catch (Exception ex) { + String message = ex.getMessage() == null ? "assemble failed" : ex.getMessage(); + markResultFailed(fr, message); + batchErrors.add(shopKey + ": " + message); + productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey); + changed = true; + log.warn("[product-risk] stale finalize failed taskId={} shop={} msg={}", taskId, shopKey, message); + } + } + + if (!changed && cachedPayloadByShop.isEmpty()) { + return false; + } + + List latest = fileResultMapper.selectList(new LambdaQueryWrapper() + .eq(FileResultEntity::getTaskId, taskId) + .eq(FileResultEntity::getModuleType, MODULE_TYPE) + .orderByAsc(FileResultEntity::getId)); + updateTaskStatusFromLatestRows(task, latest, batchErrors); + return true; + } + private void markResultFailed(FileResultEntity fr, String message) { fr.setSuccess(0); fr.setErrorMessage(message); fr.setResultFilename(null); fr.setResultFileUrl(null); fr.setResultFileSize(0L); + fr.setRowCount(0); + fr.setResultContentType(null); fileResultMapper.updateById(fr); } + private void markResultNoData(FileResultEntity fr, String message) { + fr.setSuccess(1); + fr.setErrorMessage(message); + fr.setResultFilename(null); + fr.setResultFileUrl(null); + fr.setResultFileSize(0L); + fr.setRowCount(0); + fr.setResultContentType(null); + fileResultMapper.updateById(fr); + } + + private void assembleShopResult(FileResultEntity fr, + String shopKey, + ProductRiskShopPayloadDto payload, + File workRoot) { + Map> countries = excelAssemblyService.normalizeCountriesMap(payload.getCountries()); + String displayName = payload.getShopName() != null && !payload.getShopName().isBlank() + ? payload.getShopName().trim() + : shopKey; + String stem = safeFileStem(displayName); + File xlsx = FileUtil.file(workRoot, stem + ".xlsx"); + File zip = FileUtil.file(workRoot, stem + ".zip"); + try { + excelAssemblyService.writeWorkbook(xlsx, displayName, countries); + ZipUtil.zip(zip, false, xlsx); + String objectKey = ossStorageService.uploadResultFile(zip, MODULE_TYPE); + fr.setResultFilename(stem + ".zip"); + fr.setResultFileUrl(objectKey); + fr.setResultFileSize(zip.length()); + fr.setResultContentType(CONTENT_TYPE_ZIP); + fr.setRowCount(excelAssemblyService.countRows(countries)); + fr.setSuccess(1); + fr.setErrorMessage(null); + fileResultMapper.updateById(fr); + } finally { + FileUtil.del(xlsx); + FileUtil.del(zip); + } + } + + private void updateTaskStatusFromLatestRows(FileTaskEntity task, + List latest, + List batchErrors) { + int ok = 0; + int fail = 0; + List allErrors = new ArrayList<>(); + boolean allDone = true; + for (FileResultEntity fr : latest) { + boolean success = fr.getSuccess() != null && fr.getSuccess() == 1; + boolean failed = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank(); + if (success) { + ok++; + } else if (failed) { + fail++; + allErrors.add(fr.getSourceFilename() + ": " + fr.getErrorMessage()); + } else { + allDone = false; + } + } + + task.setSuccessFileCount(ok); + task.setFailedFileCount(fail); + task.setUpdatedAt(LocalDateTime.now()); + + if (!allDone) { + task.setStatus("RUNNING"); + task.setErrorMessage(null); + task.setFinishedAt(null); + } else if (ok > 0 && fail == 0) { + task.setStatus("SUCCESS"); + task.setErrorMessage(null); + task.setFinishedAt(LocalDateTime.now()); + } else if (ok > 0) { + task.setStatus("SUCCESS"); + task.setErrorMessage(String.join("; ", allErrors.isEmpty() ? batchErrors : allErrors)); + task.setFinishedAt(LocalDateTime.now()); + } else { + task.setStatus("FAILED"); + task.setErrorMessage(allErrors.isEmpty() ? "all shops failed" : String.join("; ", allErrors)); + task.setFinishedAt(LocalDateTime.now()); + } + try { + task.setResultJson(objectMapper.writeValueAsString(buildSnapshotFromDb(task.getId(), task.getStatus()))); + } catch (Exception ex) { + log.warn("[product-risk] compact result json failed: {}", ex.getMessage()); + } + fileTaskMapper.updateById(task); + } + private ProductRiskTaskDetailVo buildTaskDetail(FileTaskEntity task) { ProductRiskTaskDetailVo detail = new ProductRiskTaskDetailVo(); detail.setTask(toTaskItemVo(task)); @@ -739,8 +915,8 @@ public class ProductRiskTaskService { for (Map.Entry entry : byKey.entrySet()) { ProductRiskRowDto existed = entry.getValue(); ProductRiskRowDto doneMerged = mergeRow(existed, row); - if (doneMerged.getDone() == null || doneMerged.getDone().isBlank()) { - doneMerged.setDone("true"); + if (doneMerged.getDone() == null) { + doneMerged.setDone(Boolean.TRUE); } entry.setValue(doneMerged); } @@ -779,7 +955,7 @@ public class ProductRiskTaskService { merged.setShopName(firstNonBlank(incoming.getShopName(), merged.getShopName())); merged.setProductAsinSku(firstNonBlank(incoming.getProductAsinSku(), merged.getProductAsinSku())); merged.setStatus(firstNonBlank(incoming.getStatus(), merged.getStatus())); - merged.setDone(firstNonBlank(incoming.getDone(), merged.getDone())); + merged.setDone(incoming.getDone() != null ? incoming.getDone() : merged.getDone()); merged.setRemoveAsin(firstNonBlank(incoming.getRemoveAsin(), merged.getRemoveAsin())); merged.setRemoveStatus(firstNonBlank(incoming.getRemoveStatus(), merged.getRemoveStatus())); return merged; @@ -823,25 +999,7 @@ public class ProductRiskTaskService { } private boolean isShopPayloadCompleted(ProductRiskShopPayloadDto payload) { - if (payload == null || payload.getCountries() == null || payload.getCountries().isEmpty()) { - return false; - } - boolean hasAnyBusinessRow = false; - for (List rows : payload.getCountries().values()) { - if (rows == null || rows.isEmpty()) { - continue; - } - for (ProductRiskRowDto row : rows) { - if (row == null || isEmptyBusinessRow(row)) { - continue; - } - hasAnyBusinessRow = true; - if (!isDoneValue(row.getDone())) { - return false; - } - } - } - return hasAnyBusinessRow; + return countPayloadRows(payload) > 0 && payloadHasAnyDoneTrue(payload); } private int countPayloadRows(ProductRiskShopPayloadDto payload) { @@ -872,7 +1030,7 @@ public class ProductRiskTaskService { continue; } for (ProductRiskRowDto row : rows) { - if (row == null || isEmptyBusinessRow(row)) { + if (row == null) { continue; } if (isDoneValue(row.getDone())) { @@ -883,20 +1041,7 @@ public class ProductRiskTaskService { return false; } - private boolean isDoneValue(String value) { - if (value == null) { - return false; - } - String normalized = value.trim().toLowerCase(Locale.ROOT); - if (normalized.isBlank()) { - return false; - } - return "true".equals(normalized) - || "1".equals(normalized) - || "yes".equals(normalized) - || "y".equals(normalized) - || "是".equals(normalized) - || "完成".equals(normalized) - || "已完成".equals(normalized); + private boolean isDoneValue(Boolean value) { + return Boolean.TRUE.equals(value); } } diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 8f44981..a1fd856 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -79,9 +79,9 @@ aiimage: 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 * * * *} + stale-check-cron: ${AIIMAGE_DELETE_BRAND_STALE_CHECK_CRON:*/30 * * * * *} finalize-check-cron: ${AIIMAGE_DELETE_BRAND_FINALIZE_CHECK_CRON:30 */2 * * * *} - product-risk-stale-timeout-hours: ${AIIMAGE_PRODUCT_RISK_STALE_TIMEOUT_HOURS:48} + product-risk-stale-timeout-minutes: ${AIIMAGE_PRODUCT_RISK_STALE_TIMEOUT_MINUTES:1} 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/BrandProductRiskTab.vue b/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue index d3d648a..b0457a6 100644 --- a/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue @@ -475,6 +475,13 @@ function rowKeyForMatch(row: ProductRiskShopQueueItem) { return `${name}\u0001${id}` } +function removeMatchedRowsLocally(rows: ProductRiskShopQueueItem[]) { + if (!rows.length) return + const keys = new Set(rows.map((row) => rowKeyForMatch(row))) + matchedItems.value = matchedItems.value.filter((row) => !keys.has(rowKeyForMatch(row))) + saveMatchedItemsToStorage() +} + async function removeMatchedRow(row: ProductRiskShopQueueItem) { const key = rowKeyForMatch(row) const backup = [...matchedItems.value] @@ -608,14 +615,24 @@ function isTaskTerminalById(taskId?: number) { return s === 'SUCCESS' || s === 'FAILED' } -const currentSectionItems = computed(() => - historyItems.value.filter( +const currentSectionItems = computed(() => { + const out = historyItems.value.filter( (row) => row.taskId && pollingTaskIds.value.includes(row.taskId) && !isTaskTerminalById(row.taskId), - ), -) + ) + + const existingTaskIds = new Set(out.map((row) => row.taskId).filter((id): id is number => typeof id === 'number')) + for (const taskId of pollingTaskIds.value) { + if (existingTaskIds.has(taskId)) continue + const snapshot = taskSnapshots.value[taskId] + const items = snapshot?.items || [] + const first = items[0] + if (first) out.push(first) + } + return out +}) const historySectionItems = computed(() => historyItems.value.filter( @@ -706,6 +723,37 @@ function stopPolling() { } } +async function waitForTaskTerminal(taskId: number) { + while (true) { + const batch = await getProductRiskTasksBatch([taskId]) + const detail = (batch.items || []).find((d) => d.task?.id === taskId) + const status = detail?.task?.status || '' + + if (detail) { + taskSnapshots.value = { + ...taskSnapshots.value, + [taskId]: detail, + } + saveTaskSnapshotsToStorage() + } + if (status) { + taskDetails.value[taskId] = status + saveTaskDetailsToStorage() + } + + if (status === 'SUCCESS' || status === 'FAILED') { + removePollingTask(taskId) + await loadHistory() + await loadDashboard() + return status + } + + await new Promise((resolve) => { + window.setTimeout(() => resolve(), getPollIntervalMs()) + }) + } +} + async function confirmAdd() { const name = shopInput.value.trim() if (!name) { @@ -808,9 +856,14 @@ async function runMatch() { } } -function statusText(item: ProductRiskHistoryItem) { +function resolvedTaskStatus(item: ProductRiskHistoryItem) { const tid = item.taskId - const st = tid ? taskStatusOf(tid) : item.taskStatus + if (!tid) return item.taskStatus || '' + return taskStatusOf(tid) || item.taskStatus || '' +} + +function statusText(item: ProductRiskHistoryItem) { + const st = resolvedTaskStatus(item) if (st === 'SUCCESS') return '已完成' if (st === 'FAILED') return '失败' if (st === 'RUNNING') return '执行中' @@ -819,17 +872,15 @@ function statusText(item: ProductRiskHistoryItem) { } function statusClass(item: ProductRiskHistoryItem) { - const tid = item.taskId - const st = tid ? taskStatusOf(tid) : item.taskStatus + const st = resolvedTaskStatus(item) if (st === 'SUCCESS' || item.success) return 'success' if (st === 'FAILED') return 'failed' return 'running' } function canDownload(item: ProductRiskHistoryItem) { - if (!item.resultId) return false - const tid = item.taskId - const st = tid ? taskStatusOf(tid) : item.taskStatus + if (!item.resultId || !item.downloadUrl) return false + const st = resolvedTaskStatus(item) return st === 'SUCCESS' || item.success === true } @@ -894,26 +945,27 @@ async function pushToPythonQueue() { } pushing.value = true queuePayloadText.value = '' - let createdTaskId: number | null = null try { - const created = await createProductRiskTask(toPush) - createdTaskId = created.taskId - taskSnapshots.value = { - ...taskSnapshots.value, - [created.taskId]: { - task: { id: created.taskId, status: 'RUNNING' }, - items: created.items, - }, - } - saveTaskSnapshotsToStorage() - addPollingTask(created.taskId) for (let i = 0; i < toPush.length; i++) { const item = toPush[i] + const created = await createProductRiskTask([item]) + const taskId = created.taskId + taskSnapshots.value = { + ...taskSnapshots.value, + [taskId]: { + task: { id: taskId, status: 'RUNNING' }, + items: created.items, + }, + } + saveTaskSnapshotsToStorage() + addPollingTask(taskId) + ensurePolling(true) + const payload = { type: 'product-risk-resolve-run', ts: Date.now(), data: { - taskId: created.taskId, + taskId, items: [item], country_codes: [...orderedCountryCodes.value], risk_listing_filter: productRiskListingFilter.value, @@ -922,18 +974,32 @@ async function pushToPythonQueue() { queuePayloadText.value = JSON.stringify(payload, null, 2) const pushResult = await api.enqueue_json(payload) if (!pushResult?.success) { - removePollingTask(created.taskId) + removePollingTask(taskId) queuePushResult.value = `第 ${i + 1}/${toPush.length} 条推送失败：${pushResult?.error || '未知错误'}` ElMessage.error(queuePushResult.value) return } - queuePushResult.value = `任务 ${created.taskId}：已入队 ${i + 1}/${toPush.length} 条店铺，当前队列长度：${pushResult.queue_size ?? '-'}` + + removeMatchedRowsLocally([item]) + queuePushResult.value = toPush.length > 1 + ? `任务 ${taskId}：第 ${i + 1}/${toPush.length} 条店铺已入队，等待执行完成...` + : `任务 ${taskId} 已入队，等待执行完成...` + + const finalStatus = await waitForTaskTerminal(taskId) + if (finalStatus !== 'SUCCESS') { + queuePushResult.value = `任务 ${taskId} 执行失败，已停止后续店铺推送` + ElMessage.error(queuePushResult.value) + return + } + + queuePushResult.value = i + 1 < toPush.length + ? `任务 ${taskId} 已完成，继续推送下一条（${i + 1}/${toPush.length}）` + : `任务 ${taskId} 已完成` } - ElMessage.success(`已创建任务 ${created.taskId}，${toPush.length} 条店铺已逐条入队`) + ElMessage.success(`已按顺序完成 ${toPush.length} 条店铺推送`) await loadHistory() ensurePolling(true) } catch (e) { - if (createdTaskId != null) removePollingTask(createdTaskId) queuePushResult.value = e instanceof Error ? e.message : '推送异常' ElMessage.error(queuePushResult.value) } finally { diff --git a/frontend-vue/tsconfig.json b/frontend-vue/tsconfig.json index 98e4ec6..5fef477 100644 --- a/frontend-vue/tsconfig.json +++ b/frontend-vue/tsconfig.json @@ -9,13 +9,20 @@ "resolveJsonModule": true, "isolatedModules": true, "esModuleInterop": true, - "lib": ["ES2020", "DOM", "DOM.Iterable"], + "lib": [ + "ES2020", + "DOM", + "DOM.Iterable" + ], "skipLibCheck": true, - "baseUrl": ".", "paths": { - "@/*": ["src/*"] + "@/*": [ + "./src/*" + ] }, - "types": ["vite/client"] + "types": [ + "vite/client" + ] }, "include": [ "src/**/*.ts", @@ -25,5 +32,9 @@ "auto-imports.d.ts", "components.d.ts" ], - "references": [{ "path": "./tsconfig.node.json" }] -} \ No newline at end of file + "references": [ + { + "path": "./tsconfig.node.json" + } + ] +}