完善删除品牌asin
This commit is contained in:
@@ -62,4 +62,7 @@ public class DeleteBrandResultItemVo {
|
||||
|
||||
@Schema(description = "错误信息")
|
||||
private String error;
|
||||
|
||||
@Schema(description = "由于列表来自 history 接口,增加所属任务的状态(RUNNING/SUCCESS/FAILED),协助前端准确定位执行中任务")
|
||||
private String taskStatus;
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import 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;
|
||||
@@ -55,6 +56,7 @@ import java.util.Set;
|
||||
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
public class DeleteBrandRunService {
|
||||
|
||||
private static final String MODULE_TYPE = "DELETE_BRAND";
|
||||
@@ -86,6 +88,7 @@ public class DeleteBrandRunService {
|
||||
task.setCreatedAt(LocalDateTime.now());
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
fileTaskMapper.insert(task);
|
||||
deleteBrandTaskCacheService.saveTaskCache(task);
|
||||
|
||||
List<DeleteBrandResultItemVo> items = new ArrayList<>();
|
||||
int parsedSuccessCount = 0;
|
||||
@@ -128,7 +131,8 @@ public class DeleteBrandRunService {
|
||||
if (message != null && (message.contains("code=40004")
|
||||
|| message.contains("userId存在无效的参数值")
|
||||
|| message.contains("无权限")
|
||||
|| message.contains("没有权限"))) {
|
||||
|| message.contains("没有权限")) && !message.contains("白名单")) {
|
||||
log.warn("[ziniao-match] suppressed upstream permission error for {}, msg: {}", item.getShopName(), message);
|
||||
item.setMatched(false);
|
||||
} else {
|
||||
throw ex;
|
||||
@@ -193,6 +197,9 @@ public class DeleteBrandRunService {
|
||||
task.setFinishedAt(LocalDateTime.now());
|
||||
}
|
||||
fileTaskMapper.updateById(task);
|
||||
if ("RUNNING".equals(task.getStatus())) {
|
||||
deleteBrandTaskCacheService.saveTaskCache(task);
|
||||
}
|
||||
|
||||
if (!parsedPayloadByFileIdentity.isEmpty()) {
|
||||
deleteBrandTaskCacheService.saveParsedPayload(task.getId(), parsedPayloadByFileIdentity);
|
||||
@@ -202,7 +209,7 @@ public class DeleteBrandRunService {
|
||||
vo.setTotal(request.getFiles().size());
|
||||
vo.setSuccessCount(parsedSuccessCount);
|
||||
vo.setFailedCount(parsedFailedCount);
|
||||
vo.setItems(items);
|
||||
vo.setItems(items.stream().peek(item -> item.setTaskStatus("RUNNING")).toList());
|
||||
return vo;
|
||||
}
|
||||
|
||||
@@ -214,8 +221,9 @@ public class DeleteBrandRunService {
|
||||
.orderByDesc(FileResultEntity::getCreatedAt)
|
||||
.last("limit 100"));
|
||||
|
||||
// 补充从 task.resultJson 还原匹配信息/真实失败原因(历史表本身不存这些字段)
|
||||
// 补充信息
|
||||
Map<Long, List<DeleteBrandResultItemVo>> itemsByTaskId = new LinkedHashMap<>();
|
||||
Map<Long, String> statusByTaskId = new LinkedHashMap<>();
|
||||
List<Long> taskIds = entities.stream()
|
||||
.map(FileResultEntity::getTaskId)
|
||||
.filter(id -> id != null && id > 0)
|
||||
@@ -224,13 +232,13 @@ public class DeleteBrandRunService {
|
||||
if (!taskIds.isEmpty()) {
|
||||
List<FileTaskEntity> tasks = fileTaskMapper.selectBatchIds(taskIds);
|
||||
for (FileTaskEntity task : tasks) {
|
||||
if (task == null || task.getResultJson() == null || task.getResultJson().isBlank()) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
List<DeleteBrandResultItemVo> taskItems = objectMapper.readValue(task.getResultJson(), new TypeReference<List<DeleteBrandResultItemVo>>() {
|
||||
});
|
||||
itemsByTaskId.put(task.getId(), taskItems);
|
||||
statusByTaskId.put(task.getId(), task.getStatus());
|
||||
if (task.getResultJson() != null && !task.getResultJson().isBlank()) {
|
||||
List<DeleteBrandResultItemVo> taskItems = objectMapper.readValue(task.getResultJson(), new TypeReference<List<DeleteBrandResultItemVo>>() {
|
||||
});
|
||||
itemsByTaskId.put(task.getId(), taskItems);
|
||||
}
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
@@ -250,7 +258,9 @@ public class DeleteBrandRunService {
|
||||
item.setError(entity.getErrorMessage());
|
||||
|
||||
// 以 task.resultJson 中的项为准补全 matched/shopId/platform/openStoreUrl/错误信息
|
||||
// 真实补充
|
||||
if (entity.getTaskId() != null) {
|
||||
item.setTaskStatus(statusByTaskId.get(entity.getTaskId()));
|
||||
List<DeleteBrandResultItemVo> taskItems = itemsByTaskId.get(entity.getTaskId());
|
||||
if (taskItems != null && entity.getSourceFilename() != null && !entity.getSourceFilename().isBlank()) {
|
||||
for (DeleteBrandResultItemVo candidate : taskItems) {
|
||||
@@ -456,9 +466,27 @@ public class DeleteBrandRunService {
|
||||
return vo;
|
||||
}
|
||||
|
||||
java.util.Map<Long, FileTaskEntity> taskById = fileTaskMapper.selectBatchIds(normalizedTaskIds).stream()
|
||||
.filter(task -> task != null && MODULE_TYPE.equals(task.getModuleType()))
|
||||
.collect(java.util.stream.Collectors.toMap(FileTaskEntity::getId, task -> task, (left, right) -> left, java.util.LinkedHashMap::new));
|
||||
java.util.Map<Long, FileTaskEntity> cachedTasks = deleteBrandTaskCacheService.getTaskCacheBatch(normalizedTaskIds);
|
||||
java.util.List<Long> missingTaskIds = new java.util.ArrayList<>();
|
||||
for (Long taskId : normalizedTaskIds) {
|
||||
if (!cachedTasks.containsKey(taskId)) {
|
||||
missingTaskIds.add(taskId);
|
||||
}
|
||||
}
|
||||
|
||||
java.util.Map<Long, FileTaskEntity> taskById = new java.util.LinkedHashMap<>(cachedTasks);
|
||||
if (!missingTaskIds.isEmpty()) {
|
||||
java.util.List<FileTaskEntity> dbTasks = fileTaskMapper.selectBatchIds(missingTaskIds).stream()
|
||||
.filter(task -> task != null && MODULE_TYPE.equals(task.getModuleType()))
|
||||
.toList();
|
||||
for (FileTaskEntity dbTask : dbTasks) {
|
||||
taskById.put(dbTask.getId(), dbTask);
|
||||
// Save back to cache
|
||||
if ("RUNNING".equals(dbTask.getStatus())) {
|
||||
deleteBrandTaskCacheService.saveTaskCache(dbTask);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
java.util.Map<Long, java.util.Map<Object, Object>> progressByTaskId = deleteBrandTaskCacheService.getProgressBatch(normalizedTaskIds);
|
||||
|
||||
@@ -858,6 +886,7 @@ public class DeleteBrandRunService {
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
task.setFinishedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
deleteBrandTaskCacheService.saveTaskCache(task);
|
||||
|
||||
deleteBrandTaskCacheService.saveProgress(task.getId(), Map.of(
|
||||
"phase", "success",
|
||||
@@ -872,6 +901,7 @@ public class DeleteBrandRunService {
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
task.setFinishedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
deleteBrandTaskCacheService.saveTaskCache(task);
|
||||
deleteBrandTaskCacheService.saveProgress(task.getId(), Map.of(
|
||||
"phase", DeleteBrandTaskCacheService.PHASE_FAILED,
|
||||
"finished_files", String.valueOf(task.getSuccessFileCount() == null ? 0 : task.getSuccessFileCount()),
|
||||
|
||||
@@ -37,22 +37,30 @@ public class DeleteBrandStaleTaskService {
|
||||
|
||||
for (FileTaskEntity task : runningTasks) {
|
||||
Map<Object, Object> progress = deleteBrandTaskCacheService.getProgress(task.getId());
|
||||
if (progress == null || progress.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
boolean hasProgress = progress != null && !progress.isEmpty() && progress.containsKey("last_heartbeat_at");
|
||||
|
||||
long lastHeartbeatAt = 0L;
|
||||
try {
|
||||
lastHeartbeatAt = Long.parseLong(String.valueOf(progress.getOrDefault("last_heartbeat_at", "0")));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
if (lastHeartbeatAt <= 0) {
|
||||
continue;
|
||||
if (hasProgress) {
|
||||
try {
|
||||
lastHeartbeatAt = Long.parseLong(String.valueOf(progress.get("last_heartbeat_at")));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
LocalDateTime lastHeartbeat = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
|
||||
if (lastHeartbeat.isAfter(threshold)) {
|
||||
continue;
|
||||
LocalDateTime lastActivityTime;
|
||||
if (lastHeartbeatAt > 0) {
|
||||
// 有心跳,用最后心跳时间对比 15 分钟
|
||||
lastActivityTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
|
||||
if (lastActivityTime.isAfter(threshold)) {
|
||||
continue; // 没超时
|
||||
}
|
||||
} else {
|
||||
// 没有心跳,说明任务一直没被 Python 取出来执行(处在排队状态中)
|
||||
// 排队的宽限期我们设为 12 个小时,如果建了 12 个小时都没执行,再杀死
|
||||
LocalDateTime queuedThreshold = LocalDateTime.now().minusHours(12);
|
||||
if (task.getCreatedAt() != null && task.getCreatedAt().isAfter(queuedThreshold)) {
|
||||
continue; // 排队中,不要杀它!
|
||||
}
|
||||
}
|
||||
|
||||
int updated = fileTaskMapper.update(null, new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<FileTaskEntity>()
|
||||
@@ -65,10 +73,7 @@ public class DeleteBrandStaleTaskService {
|
||||
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
|
||||
|
||||
if (updated > 0) {
|
||||
deleteBrandTaskCacheService.saveProgress(task.getId(), java.util.Map.of(
|
||||
"phase", DeleteBrandTaskCacheService.PHASE_FAILED,
|
||||
"updated_at", String.valueOf(System.currentTimeMillis())
|
||||
));
|
||||
deleteBrandTaskCacheService.delete(task.getId());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -181,6 +181,42 @@ public class DeleteBrandTaskCacheService {
|
||||
stringRedisTemplate.delete(buildPayloadKey(taskId));
|
||||
stringRedisTemplate.delete(buildProgressKey(taskId));
|
||||
stringRedisTemplate.delete(buildResultChunksKey(taskId));
|
||||
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
|
||||
}
|
||||
|
||||
public void saveTaskCache(com.nanri.aiimage.modules.task.model.entity.FileTaskEntity task) {
|
||||
if (task == null || task.getId() == null) return;
|
||||
try {
|
||||
stringRedisTemplate.opsForValue().set(buildTaskEntityKey(task.getId()), objectMapper.writeValueAsString(task), Duration.ofHours(PAYLOAD_TTL_HOURS));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
public java.util.Map<Long, com.nanri.aiimage.modules.task.model.entity.FileTaskEntity> getTaskCacheBatch(java.util.List<Long> taskIds) {
|
||||
java.util.Map<Long, com.nanri.aiimage.modules.task.model.entity.FileTaskEntity> result = new java.util.LinkedHashMap<>();
|
||||
if (taskIds == null || taskIds.isEmpty()) return result;
|
||||
|
||||
java.util.List<Long> normalized = taskIds.stream().filter(id -> id != null && id > 0).distinct().toList();
|
||||
if (normalized.isEmpty()) return result;
|
||||
|
||||
java.util.List<String> keys = normalized.stream().map(this::buildTaskEntityKey).toList();
|
||||
java.util.List<String> values = stringRedisTemplate.opsForValue().multiGet(keys);
|
||||
|
||||
for (int i = 0; i < normalized.size(); i++) {
|
||||
Long taskId = normalized.get(i);
|
||||
String val = values != null && i < values.size() ? values.get(i) : null;
|
||||
if (val != null && !val.isBlank()) {
|
||||
try {
|
||||
result.put(taskId, objectMapper.readValue(val, com.nanri.aiimage.modules.task.model.entity.FileTaskEntity.class));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private String buildTaskEntityKey(Long taskId) {
|
||||
return "delete-brand:task:entity:" + taskId;
|
||||
}
|
||||
|
||||
private record LocalParsedPayloadCacheEntry(long cachedAtMillis, Object value) {
|
||||
|
||||
@@ -472,6 +472,9 @@ public class ZiniaoAuthService {
|
||||
if (message == null || message.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
if (message.contains("白名单") || message.contains("IP") || message.contains("whitelist")) {
|
||||
return false;
|
||||
}
|
||||
return message.contains("isv.invalid-api-key")
|
||||
|| message.contains("无效的apiKey参数")
|
||||
|| message.contains("非法的参数");
|
||||
|
||||
Reference in New Issue
Block a user