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

修复接口内容缺失

Browse Source
This commit is contained in:
super
2026-04-08 16:14:26 +08:00
parent 336338cff5
commit 0327d1cc51
17 changed files with 13738 additions and 1034 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13475
aiimage-backend.log
View File

File diff suppressed because it is too large Load Diff

4
app/.env
View File

@@ -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

BIN
app/__pycache__/ali_oss.cpython-312.pyc
View File

Binary file not shown.

BIN
backend-java/aiimage-backend.log.2026-04-07.0.gz Normal file
View File

Binary file not shown.

56
backend-java/brand_chunk_load_test.py Normal file
View File

@@ -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 <base_url> <task_id> <chunk_total> [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()

7
backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java
View File

@@ -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;
}

24
backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java Normal file
View File

@@ -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;
}
}

21
backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandFileAggregateCacheDto.java Normal file
View File

@@ -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<BrandInvalidBrandDto> invalidBrands = new ArrayList<>();
private List<String> queryFailedBrands = new ArrayList<>();
}

297
backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java
View File

@@ -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 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);
}
throw new BusinessException("同一分片重复提交但内容不一致: " + fileUrl + "#" + file.getChunkIndex());
}
public void mergeResultChunks(Long taskId, List<BrandCrawlResultFileDto> 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("分片参数不合法");
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);
}
}
return new ChunkStoreResult(true, newlyCompleted, finishedFiles, aggregate);
}
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;
}
String field = buildChunkField(file.getFileUrl(), file.getChunkIndex());
try {
stringRedisTemplate.opsForHash().put(resultKey, field, objectMapper.writeValueAsString(file));
return objectMapper.readValue(json, BrandFileAggregateCacheDto.class);
} catch (Exception ex) {
throw new BusinessException("暂存结果分片失败");
throw new BusinessException("读取品牌文件聚合缓存失败");
}
}
stringRedisTemplate.expire(resultKey, Duration.ofHours(brandProgressProperties.getTtlHours()));
}
public Map<String, List<BrandCrawlResultFileDto>> groupResultChunksByFile(Long taskId) {
Map<Object, Object> stored = stringRedisTemplate.opsForHash().entries(buildResultKey(taskId));
Map<String, List<BrandCrawlResultFileDto>> grouped = new LinkedHashMap<>();
for (Object value : stored.values()) {
if (!(value instanceof String raw) || raw.isBlank()) {
public Map<String, BrandFileAggregateCacheDto> getAllFileAggregates(Long taskId) {
Map<Object, Object> stored = stringRedisTemplate.opsForHash().entries(buildFileAggregateKey(taskId));
Map<String, BrandFileAggregateCacheDto> result = new LinkedHashMap<>();
for (Map.Entry<Object, Object> 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) {
}
}

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

@@ -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<BrandSourceFileDto> sourceFiles = parseSourceFiles(task.getFilePaths());
if (sourceFiles.isEmpty()) {
throw new BusinessException("任务没有源文件");
}
Map<String, BrandSourceFileDto> sourceByUrl = new LinkedHashMap<>();
for (BrandSourceFileDto file : sourceFiles) {
Map<String, Integer> sourceIndexByUrl = new LinkedHashMap<>();
for (int i = 0; i < sourceFiles.size(); i++) {
BrandSourceFileDto file = sourceFiles.get(i);
sourceByUrl.put(file.getFileUrl(), file);
sourceIndexByUrl.put(file.getFileUrl(), i + 1);
}
List<BrandParsedFileCacheDto> cachedFiles = brandTaskProgressCacheService.getParsedPayload(taskId,
new TypeReference<List<BrandParsedFileCacheDto>>() {
@@ -236,82 +240,67 @@ public class BrandTaskService {
ensureNoDuplicateResultFiles(resultFiles);
int totalCount = sourceFiles.size();
int started = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.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 不能为空");
}
brandTaskProgressCacheService.mergeResultChunks(taskId, resultFiles);
Map<String, List<BrandCrawlResultFileDto>> 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<OutputEntry> outputEntries = new ArrayList<>();
try {
for (List<BrandCrawlResultFileDto> fileChunks : groupedResults.values()) {
BrandCrawlResultFileDto resultFile = mergeChunks(fileChunks);
BrandSourceFileDto sourceFile = sourceByUrl.get(resultFile.getFileUrl());
BrandSourceFileDto sourceFile = sourceByUrl.get(fileUrl);
if (sourceFile == null) {
throw new BusinessException("存在未知 fileUrl: " + resultFile.getFileUrl());
throw new BusinessException("存在未知 fileUrl: " + fileUrl);
}
BrandParsedFileCacheDto cachedFile = cachedByUrl.get(resultFile.getFileUrl());
BrandParsedFileCacheDto cachedFile = cachedByUrl.get(fileUrl);
if (cachedFile == null) {
throw new BusinessException("缺少原始缓存数据: " + resultFile.getFileUrl());
throw new BusinessException("缺少原始缓存数据: " + fileUrl);
}
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));
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.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_UPLOADING, finishedCount, totalCount);
Map<String, Object> resultPaths = buildAndUploadResult(taskId, outputEntries);
int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.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<BrandCrawlTaskEntity>()
.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<BrandCrawlTaskEntity>()
.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<BrandSourceFileDto> sourceFiles,
Map<String, BrandParsedFileCacheDto> 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<BrandSourceFileDto> sourceFiles,
Map<String, BrandParsedFileCacheDto> 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<String, BrandFileAggregateCacheDto> 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<OutputEntry> outputEntries = new ArrayList<>();
try {
for (BrandSourceFileDto sourceFile : sourceFiles) {
BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl());
BrandParsedFileCacheDto cachedFile = cachedByUrl.get(sourceFile.getFileUrl());
if (cachedFile == null) {
throw new BusinessException("缺少原始缓存数据: " + sourceFile.getFileUrl());
}
File sourceLocalFile = resolveSourceFile(sourceFile);
String originalFilename = resolveOriginalFilename(sourceFile, sourceLocalFile);
File outputFile = buildNamedOutputFile(outputDir, buildResultFilename(originalFilename));
writeBrandWorkbook(outputFile, strategy, cachedFile, aggregate);
outputEntries.add(new OutputEntry(sourceLocalFile, originalFilename, outputFile, originalFilename));
}
brandTaskProgressCacheService.updatePhase(taskId, BrandTaskProgressCacheService.PHASE_UPLOADING, totalCount, totalCount);
Map<String, Object> resultPaths = buildAndUploadResult(taskId, outputEntries);
int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.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<BrandCrawlTaskEntity>()
.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<BrandCrawlResultFileDto> resultFiles) {
Set<String> 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<BrandSourceFileDto> sourceFiles,
Map<String, BrandSourceFileDto> sourceByUrl,
Map<String, List<BrandCrawlResultFileDto>> groupedResults,
int finishedCount,
int totalCount) {
if (groupedResults.isEmpty()) {
return;
}
String currentFileUrl = null;
List<BrandCrawlResultFileDto> currentChunks = null;
int currentFileIndex = 0;
for (int i = sourceFiles.size() - 1; i >= 0; i--) {
BrandSourceFileDto sourceFile = sourceFiles.get(i);
List<BrandCrawlResultFileDto> 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<BrandCrawlResultFileDto> 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<BrandCrawlResultFileDto> 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<BrandSourceFileDto> 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<String, List<BrandCrawlResultFileDto>> groupedResults) {
int count = 0;
for (List<BrandCrawlResultFileDto> chunks : groupedResults.values()) {
if (!chunks.isEmpty() && Objects.equals(chunks.size(), chunks.get(0).getChunkTotal())) {
count++;
}
}
return count;
}
private BrandCrawlResultFileDto mergeChunks(List<BrandCrawlResultFileDto> 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<String> keptRows = new ArrayList<>();
List<BrandInvalidBrandDto> invalidBrands = new ArrayList<>();
List<String> 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;

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

@@ -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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<FileTaskEntity>()
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<FileTaskEntity>()
.eq(FileTaskEntity::getId, task.getId())
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRODUCT_RISK)
.eq(FileTaskEntity::getStatus, "RUNNING")
@@ -116,8 +145,15 @@ 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 * * * *}")
public void finalizeCompletedRunningTasks() {
@@ -135,4 +171,11 @@ public class DeleteBrandStaleTaskService {
}
}
}
private static final class ProductRiskStaleCheckStats {
private int scannedTaskCount;
private int finalizedTaskCount;
private int failedTaskCount;
private int skippedTaskCount;
}
}

2
backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskRowDto.java
View File

@@ -21,7 +21,7 @@ public class ProductRiskRowDto {
@JsonProperty("done")
@Schema(description = "是否完成等标记列")
private String done;
private Boolean done;
@JsonProperty("removeAsin")
@Schema(description = "待移除 ASIN 等")

2
backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskExcelAssemblyService.java
View File

@@ -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());
}

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

@@ -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<String, List<ProductRiskRowDto>> 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<FileResultEntity> resultRows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
if (resultRows.isEmpty()) {
return false;
}
Map<String, ProductRiskShopPayloadDto> cachedPayloadByShop = productRiskTaskCacheService.getAllShopMergedPayload(taskId);
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "product-risk-result", String.valueOf(taskId)));
List<String> batchErrors = new ArrayList<>();
boolean changed = false;
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<FileResultEntity> latest = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.orderByAsc(FileResultEntity::getId));
updateTaskStatusFromLatestRows(task, latest, batchErrors);
return true;
}
private void 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<String, List<ProductRiskRowDto>> 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<FileResultEntity> latest,
List<String> batchErrors) {
int ok = 0;
int fail = 0;
List<String> 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<String, ProductRiskRowDto> 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<ProductRiskRowDto> 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);
}
}

4
backend-java/src/main/resources/application.yml
View File

@@ -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 * * *}

114
frontend-vue/src/pages/brand/components/BrandProductRiskTab.vue
View File

@@ -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<void>((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
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,
[created.taskId]: {
task: { id: created.taskId, status: 'RUNNING' },
[taskId]: {
task: { id: taskId, status: 'RUNNING' },
items: created.items,
},
}
saveTaskSnapshotsToStorage()
addPollingTask(created.taskId)
for (let i = 0; i < toPush.length; i++) {
const item = toPush[i]
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
}
ElMessage.success(`已创建任务 ${created.taskId}，${toPush.length} 条店铺已逐条入队`)
queuePushResult.value = i + 1 < toPush.length
? `任务 ${taskId} 已完成，继续推送下一条（${i + 1}/${toPush.length}）`
: `任务 ${taskId} 已完成`
}
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 {

21
frontend-vue/tsconfig.json
View File

@@ -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" }]
"references": [
{
"path": "./tsconfig.node.json"
}
]
}