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-12 17:15:15 +08:00
parent 7589b4e418
commit e53eaf738e
17 changed files with 814 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java Normal file
View File

@@ -0,0 +1,84 @@
package com.nanri.aiimage.common.service;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import java.net.InetAddress;
import java.time.Duration;
import java.util.Collections;
import java.util.UUID;
@Service
@Slf4j
public class DistributedJobLockService {
private static final DefaultRedisScript<Long> RELEASE_SCRIPT = new DefaultRedisScript<>(
"if redis.call('get', KEYS[1]) == ARGV[1] then "
+ "return redis.call('del', KEYS[1]) "
+ "else return 0 end",
Long.class
);
private final StringRedisTemplate stringRedisTemplate;
private final String ownerPrefix;
public DistributedJobLockService(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
this.ownerPrefix = resolveOwnerPrefix();
}
public LockHandle tryLock(String jobName, Duration ttl) {
if (jobName == null || jobName.isBlank()) {
return null;
}
Duration actualTtl = (ttl == null || ttl.isNegative() || ttl.isZero()) ? Duration.ofMinutes(10) : ttl;
String key = buildLockKey(jobName);
String token = ownerPrefix + ":" + UUID.randomUUID();
Boolean locked = stringRedisTemplate.opsForValue().setIfAbsent(key, token, actualTtl);
if (!Boolean.TRUE.equals(locked)) {
return null;
}
return new LockHandle(key, token, jobName);
}
public final class LockHandle implements AutoCloseable {
private final String key;
private final String token;
private final String jobName;
private boolean released;
private LockHandle(String key, String token, String jobName) {
this.key = key;
this.token = token;
this.jobName = jobName;
}
@Override
public void close() {
if (released) {
return;
}
released = true;
try {
stringRedisTemplate.execute(RELEASE_SCRIPT, Collections.singletonList(key), token);
} catch (Exception ex) {
log.warn("[job-lock] release failed jobName={} key={} msg={}", jobName, key, ex.getMessage());
}
}
}
private String buildLockKey(String jobName) {
return "job-lock:" + jobName;
}
private String resolveOwnerPrefix() {
try {
return InetAddress.getLocalHost().getHostName();
} catch (Exception ignored) {
return "unknown-host";
}
}
}

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

@@ -7,6 +7,7 @@ import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.BrandProgressProperties;
import com.nanri.aiimage.config.StorageProperties;
import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper;
@@ -49,6 +50,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.nio.file.Files;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -69,6 +71,7 @@ import java.util.zip.ZipOutputStream;
public class BrandTaskService {
private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm");
private static final Duration STALE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
private static final String STATUS_PENDING = "pending";
private static final String STATUS_RUNNING = "running";
private static final String STATUS_SUCCESS = "success";
@@ -80,6 +83,7 @@ public class BrandTaskService {
private final StorageProperties storageProperties;
private final BrandProgressProperties brandProgressProperties;
private final BrandTaskProgressCacheService brandTaskProgressCacheService;
private final DistributedJobLockService distributedJobLockService;
private final ObjectMapper objectMapper;
public BrandTaskCreateVo createTask(Long userId, BrandTaskCreateRequest request) {
@@ -1051,6 +1055,12 @@ public class BrandTaskService {
@Transactional
@org.springframework.scheduling.annotation.Scheduled(cron = "${aiimage.brand-progress.stale-check-cron:0 */2 * * * *}")
public void failStaleRunningTasks() {
DistributedJobLockService.LockHandle lockHandle = distributedJobLockService.tryLock("brand:stale-check", STALE_CHECK_LOCK_TTL);
if (lockHandle == null) {
log.info("[brand-stale-check] skip because another instance holds the distributed lock");
return;
}
try (lockHandle) {
LocalDateTime threshold = LocalDateTime.now().minusMinutes(brandProgressProperties.getHeartbeatTimeoutMinutes());
List<BrandCrawlTaskEntity> runningTasks = brandCrawlTaskMapper.selectList(new LambdaQueryWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
@@ -1075,6 +1085,7 @@ public class BrandTaskService {
continue;
}
failStaleRunningTask(task.getId(), "前端长时间无响应，任务已自动失败");
}
}
}

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

@@ -2,6 +2,7 @@ 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.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.DeleteBrandProgressProperties;
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskCacheService;
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService;
@@ -14,15 +15,12 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Map;
/**
* 删除品牌任务的超时回收与 finalize 补偿。
* 同时复用这套定时调度，对商品风险和匹配店铺任务做超时兜底收尾。
*/
@Service
@RequiredArgsConstructor
@Slf4j
@@ -31,6 +29,8 @@ public class DeleteBrandStaleTaskService {
private static final String MODULE_TYPE_DELETE_BRAND = "DELETE_BRAND";
private static final String MODULE_TYPE_PRODUCT_RISK = "PRODUCT_RISK_RESOLVE";
private static final String MODULE_TYPE_SHOP_MATCH = "SHOP_MATCH";
private static final Duration STALE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
private static final Duration FINALIZE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
private final FileTaskMapper fileTaskMapper;
private final DeleteBrandTaskCacheService deleteBrandTaskCacheService;
@@ -40,34 +40,42 @@ public class DeleteBrandStaleTaskService {
private final ShopMatchTaskService shopMatchTaskService;
private final ShopMatchTaskCacheService shopMatchTaskCacheService;
private final DeleteBrandProgressProperties deleteBrandProgressProperties;
private final DistributedJobLockService distributedJobLockService;
@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();
ProductRiskStaleCheckStats stats = failStaleProductRiskResolveTasks();
ShopMatchStaleCheckStats shopMatchStats = failStaleShopMatchTasks();
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());
log.info("[stale-check] shop-match summary scanned={} finalized={} failed={} skipped={} elapsedMs={} thread={}",
shopMatchStats.scannedTaskCount,
shopMatchStats.finalizedTaskCount,
shopMatchStats.failedTaskCount,
shopMatchStats.skippedTaskCount,
System.currentTimeMillis() - startedAt,
Thread.currentThread().getName());
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("delete-brand:stale-check", STALE_CHECK_LOCK_TTL);
if (lockHandle == null) {
log.info("[stale-check] skip because another instance holds the distributed lock");
return;
}
try (lockHandle) {
long startedAt = System.currentTimeMillis();
log.info("[stale-check] scan started thread={}", Thread.currentThread().getName());
failStaleDeleteBrandTasks();
ProductRiskStaleCheckStats stats = failStaleProductRiskResolveTasks();
ShopMatchStaleCheckStats shopMatchStats = failStaleShopMatchTasks();
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());
log.info("[stale-check] shop-match summary scanned={} finalized={} failed={} skipped={} elapsedMs={} thread={}",
shopMatchStats.scannedTaskCount,
shopMatchStats.finalizedTaskCount,
shopMatchStats.failedTaskCount,
shopMatchStats.skippedTaskCount,
System.currentTimeMillis() - startedAt,
Thread.currentThread().getName());
}
}
private void failStaleDeleteBrandTasks() {
LocalDateTime threshold = LocalDateTime.now().minusMinutes(deleteBrandProgressProperties.getHeartbeatTimeoutMinutes());
// 先按 DB updatedAt 粗筛，避免全表扫描。
java.util.List<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND)
.eq(FileTaskEntity::getStatus, "RUNNING")
@@ -96,15 +104,12 @@ public class DeleteBrandStaleTaskService {
}
}
LocalDateTime lastActivityTime;
// 只有 has_progress=true 且心跳超时，才判定为异常卡住。
if (lastHeartbeatAt > 0 && isActive) {
lastActivityTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
LocalDateTime lastActivityTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
if (lastActivityTime.isAfter(threshold)) {
continue;
}
} else {
// 没有心跳，或仍处于静默等待阶段时，使用更长的初始宽限期。
LocalDateTime queuedThreshold = LocalDateTime.now().minusHours(12);
if (task.getCreatedAt() != null && task.getCreatedAt().isAfter(queuedThreshold)) {
continue;
@@ -130,6 +135,8 @@ public class DeleteBrandStaleTaskService {
ProductRiskStaleCheckStats stats = new ProductRiskStaleCheckStats();
long minutes = Math.max(1L, deleteBrandProgressProperties.getProductRiskStaleTimeoutMinutes());
long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getProductRiskInitialTimeoutMinutes());
long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis();
long nowMillis = System.currentTimeMillis();
LocalDateTime now = LocalDateTime.now();
LocalDateTime threshold = now.minusMinutes(minutes);
LocalDateTime initialThreshold = now.minusMinutes(initialMinutes);
@@ -146,9 +153,16 @@ public class DeleteBrandStaleTaskService {
runningTasks.size(), threshold, minutes, initialThreshold, initialMinutes);
for (FileTaskEntity task : runningTasks) {
boolean hasUploadedPayload = productRiskTaskCacheService.hasAnyShopMergedPayload(task.getId());
long lastPayloadHeartbeatMillis = productRiskTaskCacheService.getTaskHeartbeatMillis(task.getId());
boolean hasFinishedRows = (task.getSuccessFileCount() != null && task.getSuccessFileCount() > 0)
|| (task.getFailedFileCount() != null && task.getFailedFileCount() > 0);
boolean hasStartedProgress = hasUploadedPayload || hasFinishedRows;
if (lastPayloadHeartbeatMillis > 0L && nowMillis - lastPayloadHeartbeatMillis < staleTimeoutMillis) {
stats.skippedTaskCount++;
log.info("[stale-check] product-risk skip recent-task-heartbeat taskId={} lastPayloadHeartbeatMillis={} timeoutMinutes={} updatedAt={}",
task.getId(), lastPayloadHeartbeatMillis, minutes, task.getUpdatedAt());
continue;
}
if (!hasStartedProgress && task.getCreatedAt() != null && task.getCreatedAt().isAfter(initialThreshold)) {
stats.skippedTaskCount++;
log.info("[stale-check] product-risk skip initial-grace taskId={} createdAt={} initialThreshold={}",
@@ -241,17 +255,25 @@ public class DeleteBrandStaleTaskService {
@Scheduled(cron = "${aiimage.delete-brand-progress.finalize-check-cron:30 */2 * * * *}")
public void finalizeCompletedRunningTasks() {
java.util.List<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND)
.eq(FileTaskEntity::getStatus, "RUNNING")
.orderByAsc(FileTaskEntity::getUpdatedAt)
.last("limit 100"));
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("delete-brand:finalize-check", FINALIZE_CHECK_LOCK_TTL);
if (lockHandle == null) {
log.info("[finalize-check] skip because another instance holds the distributed lock");
return;
}
try (lockHandle) {
java.util.List<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND)
.eq(FileTaskEntity::getStatus, "RUNNING")
.orderByAsc(FileTaskEntity::getUpdatedAt)
.last("limit 100"));
for (FileTaskEntity task : runningTasks) {
try {
deleteBrandRunService.tryFinalizeTask(task.getId(), true);
} catch (Exception ignored) {
// 定时补偿不影响主流程；下次调度或人工重试时再处理。
for (FileTaskEntity task : runningTasks) {
try {
deleteBrandRunService.tryFinalizeTask(task.getId(), true);
} catch (Exception ignored) {
// Keep compensation best-effort; the next schedule can retry.
}
}
}
}

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

@@ -30,4 +30,6 @@ public class ProductRiskShopPayloadDto {
description = "五国数据：JSON 对象的键为枚举 ProductRiskCountryCode（DE、FR、ES、IT、UK），"
+ "值为该国 sheet 的行列表；省略的键表示该国无数据行。")
private Map<ProductRiskCountryCode, List<ProductRiskRowDto>> countries = new LinkedHashMap<>();
private Boolean shopDone;
}

38
backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskCacheService.java
View File

@@ -1,6 +1,5 @@
package com.nanri.aiimage.modules.productrisk.service;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.productrisk.model.dto.ProductRiskShopPayloadDto;
@@ -9,6 +8,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.Instant;
import java.util.Map;
@Service
@@ -42,6 +42,7 @@ public class ProductRiskTaskCacheService {
try {
stringRedisTemplate.opsForHash().put(buildShopPayloadKey(taskId), shopKey, objectMapper.writeValueAsString(payload));
stringRedisTemplate.expire(buildShopPayloadKey(taskId), Duration.ofHours(PAYLOAD_TTL_HOURS));
touchTaskHeartbeat(taskId);
} catch (Exception ex) {
throw new BusinessException("暂存商品风险店铺缓存失败");
}
@@ -81,14 +82,49 @@ public class ProductRiskTaskCacheService {
return size != null && size > 0;
}
public long countShopMergedPayload(Long taskId) {
if (taskId == null || taskId <= 0) {
return 0L;
}
Long size = stringRedisTemplate.opsForHash().size(buildShopPayloadKey(taskId));
return size == null ? 0L : size;
}
public long getTaskHeartbeatMillis(Long taskId) {
if (taskId == null || taskId <= 0) {
return 0L;
}
String raw = stringRedisTemplate.opsForValue().get(buildTaskHeartbeatKey(taskId));
if (raw == null || raw.isBlank()) {
return 0L;
}
try {
return Long.parseLong(raw);
} catch (NumberFormatException ignored) {
return 0L;
}
}
public void deleteTaskCache(Long taskId) {
if (taskId == null || taskId <= 0) {
return;
}
stringRedisTemplate.delete(buildShopPayloadKey(taskId));
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
}
private String buildShopPayloadKey(Long taskId) {
return "product-risk:task:shop-payload:" + taskId;
}
public void touchTaskHeartbeat(Long taskId) {
stringRedisTemplate.opsForValue().set(
buildTaskHeartbeatKey(taskId),
String.valueOf(Instant.now().toEpochMilli()),
Duration.ofHours(PAYLOAD_TTL_HOURS));
}
private String buildTaskHeartbeatKey(Long taskId) {
return "product-risk:task:heartbeat:" + taskId;
}
}

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

@@ -208,6 +208,7 @@ public class ProductRiskTaskService {
.orderByAsc(FileResultEntity::getId));
if (latest.isEmpty()) {
log.warn("[product-risk] reconcile delete empty taskId={} oldStatus={} -> deleting task", taskId, oldStatus);
cleanupTaskCacheIfTerminal(taskId, "DELETE_EMPTY");
fileTaskMapper.deleteById(taskId);
return;
}
@@ -260,9 +261,7 @@ public class ProductRiskTaskService {
log.warn("[product-risk] reconcile status changed taskId={} oldStatus={} newStatus={} ok={} fail={} allDone={} errorMessage={}",
taskId, oldStatus, task.getStatus(), ok, fail, allDone, task.getErrorMessage());
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
productRiskTaskCacheService.deleteTaskCache(taskId);
}
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
}
public ProductRiskTaskBatchVo getTaskDetailsBatch(List<Long> taskIds) {
@@ -338,6 +337,7 @@ public class ProductRiskTaskService {
task.setCreatedAt(LocalDateTime.now());
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.insert(task);
productRiskTaskCacheService.touchTaskHeartbeat(task.getId());
List<ProductRiskResultItemVo> snapshot = new ArrayList<>();
for (ProductRiskShopQueueItemVo item : uniqueItems) {
@@ -448,17 +448,18 @@ public class ProductRiskTaskService {
// 分次上报累积：合并到 Redis 缓存，再按“合并后是否完成”决定是否出包。
ProductRiskShopPayloadDto mergedPayload = mergeShopPayload(taskId, shopKey, payload);
boolean incomingDone = payloadHasAnyDoneTrue(payload);
boolean incomingShopDone = hasShopDoneSignal(payload);
int mergedRows = countPayloadRows(mergedPayload);
boolean shopDone = isShopPayloadCompleted(mergedPayload);
log.info("[product-risk] shop merged taskId={} shop={} incomingDone={} mergedDone={} mergedRows={} payloadCountries={}",
log.info("[product-risk] shop merged taskId={} shop={} incomingShopDone={} mergedDone={} mergedRows={} payloadCountries={} mergedShopDoneFlag={}",
taskId,
shopKey,
incomingDone,
incomingShopDone,
shopDone,
mergedRows,
mergedPayload.getCountries() == null ? 0 : mergedPayload.getCountries().size());
if (incomingDone && mergedRows <= 0) {
mergedPayload.getCountries() == null ? 0 : mergedPayload.getCountries().size(),
mergedPayload.getShopDone());
if (incomingShopDone && mergedRows <= 0) {
markResultNoData(fr, "没有数据");
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
log.info("[product-risk] shop finished without data taskId={} shop={}", taskId, shopKey);
@@ -535,6 +536,7 @@ public class ProductRiskTaskService {
log.warn("[product-risk] submitResult status evaluated taskId={} oldStatus={} newStatus={} ok={} fail={} allDone={} matchedShopCount={} skippedUnmatchedCount={} waitingCount={} assembledCount={} fallbackMatchedCount={} batchErrors={}",
taskId, oldStatus, task.getStatus(), ok, fail, allDone, matchedShopCount, skippedUnmatchedCount,
waitingCount, assembledCount, fallbackMatchedCount, batchErrors);
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
}
@Transactional
@@ -577,8 +579,8 @@ public class ProductRiskTaskService {
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");
markResultFailed(fr, "Python 在该店铺结果上传完成前中断");
batchErrors.add(shopKey + ": Python 在该店铺结果上传完成前中断");
changed = true;
continue;
}
@@ -591,7 +593,7 @@ public class ProductRiskTaskService {
continue;
}
if (countPayloadRows(cachedPayload) <= 0 && payloadHasAnyDoneTrue(cachedPayload)) {
if (countPayloadRows(cachedPayload) <= 0 && isShopPayloadCompleted(cachedPayload)) {
markResultNoData(fr, "没有数据");
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
@@ -601,21 +603,26 @@ public class ProductRiskTaskService {
}
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");
markResultFailed(fr, "Python 在上传可组装结果前中断");
batchErrors.add(shopKey + ": Python 在上传可组装结果前中断");
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
continue;
}
if (!isShopPayloadCompleted(cachedPayload)) {
log.warn("[product-risk] stale finalize assembling partial payload taskId={} shop={} rowCount={} fromCompensation={}",
taskId, shopKey, countPayloadRows(cachedPayload), fromCompensation);
}
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));
log.warn("[product-risk] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={} rowCount={}",
taskId, shopKey, fromCompensation, isShopPayloadCompleted(cachedPayload), countPayloadRows(cachedPayload));
} catch (Exception ex) {
String message = ex.getMessage() == null ? "assemble failed" : ex.getMessage();
String message = ex.getMessage() == null ? "结果组装失败" : ex.getMessage();
markResultFailed(fr, message);
batchErrors.add(shopKey + ": " + message);
productRiskTaskCacheService.removeShopMergedPayload(taskId, shopKey);
@@ -739,6 +746,19 @@ public class ProductRiskTaskService {
fileTaskMapper.updateById(task);
log.warn("[product-risk] status updated from latest rows taskId={} oldStatus={} newStatus={} ok={} fail={} allDone={} batchErrors={} errorMessage={}",
task.getId(), oldStatus, task.getStatus(), ok, fail, allDone, batchErrors, task.getErrorMessage());
cleanupTaskCacheIfTerminal(task.getId(), task.getStatus());
}
private void cleanupTaskCacheIfTerminal(Long taskId, String taskStatus) {
if (!"SUCCESS".equals(taskStatus) && !"FAILED".equals(taskStatus) && !"DELETE_EMPTY".equals(taskStatus)) {
return;
}
long cachedShopCount = productRiskTaskCacheService.countShopMergedPayload(taskId);
if (cachedShopCount > 0) {
log.warn("[product-risk] clearing merged payload cache taskId={} status={} cachedShopCount={}",
taskId, taskStatus, cachedShopCount);
}
productRiskTaskCacheService.deleteTaskCache(taskId);
}
private ProductRiskTaskDetailVo buildTaskDetail(FileTaskEntity task) {
@@ -879,6 +899,7 @@ public class ProductRiskTaskService {
if (merged == null) {
merged = new ProductRiskShopPayloadDto();
}
boolean incomingShopDone = hasShopDoneSignal(incoming);
if (incoming.getShopName() != null && !incoming.getShopName().isBlank()) {
merged.setShopName(incoming.getShopName().trim());
@@ -889,6 +910,9 @@ public class ProductRiskTaskService {
if (incoming.getError() != null && !incoming.getError().isBlank()) {
merged.setError(incoming.getError());
}
if (Boolean.TRUE.equals(merged.getShopDone()) || incomingShopDone || Boolean.TRUE.equals(incoming.getShopDone())) {
merged.setShopDone(Boolean.TRUE);
}
Map<com.nanri.aiimage.modules.productrisk.model.enums.ProductRiskCountryCode, List<ProductRiskRowDto>> nextCountries =
merged.getCountries() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(merged.getCountries());
@@ -1017,7 +1041,27 @@ public class ProductRiskTaskService {
}
private boolean isShopPayloadCompleted(ProductRiskShopPayloadDto payload) {
return countPayloadRows(payload) > 0 && payloadHasAnyDoneTrue(payload);
return Boolean.TRUE.equals(payload == null ? null : payload.getShopDone());
}
private boolean hasShopDoneSignal(ProductRiskShopPayloadDto payload) {
if (payload == null || payload.getCountries() == null) {
return false;
}
for (List<ProductRiskRowDto> rows : payload.getCountries().values()) {
if (rows == null) {
continue;
}
for (ProductRiskRowDto row : rows) {
if (row == null) {
continue;
}
if (isDoneValue(row.getDone()) && isEmptyBusinessRow(row)) {
return true;
}
}
}
return false;
}
private int countPayloadRows(ProductRiskShopPayloadDto payload) {

8
backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskCacheService.java
View File

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.shopmatch.model.dto.ShopMatchShopPayloadDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
@@ -13,6 +14,7 @@ import java.util.Map;
@Service
@RequiredArgsConstructor
@Slf4j
public class ShopMatchTaskCacheService {
private static final long PAYLOAD_TTL_HOURS = 24;
@@ -78,7 +80,11 @@ public class ShopMatchTaskCacheService {
if (taskId == null || taskId <= 0) {
return;
}
stringRedisTemplate.delete(buildShopPayloadKey(taskId));
try {
stringRedisTemplate.delete(buildShopPayloadKey(taskId));
} catch (Exception ex) {
log.warn("[shop-match] delete task cache failed taskId={} msg={}", taskId, ex.getMessage());
}
}
private String buildShopPayloadKey(Long taskId) {

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

@@ -35,6 +35,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.io.File;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -49,6 +50,7 @@ public class ShopMatchTaskService {
private static final String MODULE_TYPE = "SHOP_MATCH";
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai");
private final FileTaskMapper fileTaskMapper;
private final FileResultMapper fileResultMapper;
@@ -224,7 +226,7 @@ public class ShopMatchTaskService {
List<String> countryCodes = normalizeCountryCodes(request.getCountryCodes());
List<LocalDateTime> scheduleTimes = normalizeScheduleTimes(request.getScheduleTimes());
LocalDateTime now = LocalDateTime.now();
LocalDateTime now = now();
ShopMatchCreateTaskRequest persistedRequest = new ShopMatchCreateTaskRequest();
persistedRequest.setUserId(request.getUserId());
@@ -263,7 +265,7 @@ public class ShopMatchTaskService {
result.setSourceFileUrl(item.getShopId());
result.setUserId(request.getUserId());
result.setSuccess(0);
result.setCreatedAt(LocalDateTime.now());
result.setCreatedAt(now);
fileResultMapper.insert(result);
snapshot.add(toSnapshotVo(result, item, task.getStatus()));
}
@@ -302,7 +304,10 @@ public class ShopMatchTaskService {
if (!"SCHEDULED".equals(task.getStatus())) {
throw new BusinessException("任务状态不正确");
}
if (task.getScheduledAt() != null && LocalDateTime.now().isBefore(task.getScheduledAt().minusSeconds(90))) {
LocalDateTime now = now();
if (task.getScheduledAt() != null && now.isBefore(task.getScheduledAt().minusSeconds(90))) {
log.warn("[shop-match] activate rejected taskId={} stageIndex={} now={} scheduledAt={} earliestActivateAt={}",
taskId, stageIndex, now, task.getScheduledAt(), task.getScheduledAt().minusSeconds(90));
throw new BusinessException("未到定时执行时间");
}
ShopMatchCreateTaskRequest state = parseTaskRequest(task);
@@ -315,7 +320,7 @@ public class ShopMatchTaskService {
}
state.setActiveStageIndex(stageIndex);
task.setStatus("RUNNING");
task.setUpdatedAt(LocalDateTime.now());
task.setUpdatedAt(now);
persistTaskRequest(task, state);
fileTaskMapper.updateById(task);
}
@@ -350,7 +355,7 @@ public class ShopMatchTaskService {
state.setActiveStageIndex(null);
task.setStatus("SCHEDULED");
task.setScheduledAt(state.getScheduleTimes().get(nextIndex));
task.setUpdatedAt(LocalDateTime.now());
task.setUpdatedAt(now());
task.setFinishedAt(null);
persistTaskRequest(task, state);
fileTaskMapper.updateById(task);
@@ -499,12 +504,17 @@ public class ShopMatchTaskService {
continue;
}
if (!isShopPayloadCompleted(cachedPayload)) {
log.warn("[shop-match] stale finalize assembling partial payload taskId={} shop={} rowCount={} fromCompensation={}",
taskId, shopKey, countPayloadRows(cachedPayload), fromCompensation);
}
try {
assembleShopResult(result, shopKey, cachedPayload, workRoot);
shopMatchTaskCacheService.removeShopMergedPayload(taskId, shopKey);
changed = true;
log.warn("[shop-match] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={}",
taskId, shopKey, fromCompensation, isShopPayloadCompleted(cachedPayload));
log.warn("[shop-match] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={} rowCount={}",
taskId, shopKey, fromCompensation, isShopPayloadCompleted(cachedPayload), countPayloadRows(cachedPayload));
} catch (Exception ex) {
String message = ex.getMessage() == null ? "结果组装失败" : ex.getMessage();
markResultFailed(result, message);
@@ -584,7 +594,7 @@ public class ShopMatchTaskService {
}
task.setSuccessFileCount(ok);
task.setFailedFileCount(fail);
task.setUpdatedAt(LocalDateTime.now());
task.setUpdatedAt(now());
if (!allDone) {
task.setStatus("RUNNING");
task.setErrorMessage(null);
@@ -592,15 +602,15 @@ public class ShopMatchTaskService {
} else if (ok > 0 && fail == 0) {
task.setStatus("SUCCESS");
task.setErrorMessage(null);
task.setFinishedAt(LocalDateTime.now());
task.setFinishedAt(now());
} else if (ok > 0) {
task.setStatus("SUCCESS");
task.setErrorMessage(String.join("；", errors.isEmpty() ? batchErrors : errors));
task.setFinishedAt(LocalDateTime.now());
task.setFinishedAt(now());
} else {
task.setStatus("FAILED");
task.setErrorMessage(errors.isEmpty() ? "全部店铺处理失败" : String.join("；", errors));
task.setFinishedAt(LocalDateTime.now());
task.setFinishedAt(now());
}
try {
task.setResultJson(objectMapper.writeValueAsString(buildSnapshotFromDb(task.getId(), task.getStatus())));
@@ -815,6 +825,10 @@ public class ShopMatchTaskService {
return time == null ? null : time.toString();
}
private LocalDateTime now() {
return LocalDateTime.now(BUSINESS_ZONE);
}
private List<ProductRiskShopQueueItemVo> dedupeQueueItems(List<ProductRiskShopQueueItemVo> raw) {
Map<String, ProductRiskShopQueueItemVo> byNorm = new LinkedHashMap<>();
for (ProductRiskShopQueueItemVo item : raw) {
@@ -857,7 +871,7 @@ public class ShopMatchTaskService {
return List.of();
}
List<LocalDateTime> out = new ArrayList<>();
LocalDateTime now = LocalDateTime.now();
LocalDateTime now = now();
LocalDateTime previous = null;
for (LocalDateTime time : raw) {
if (time == null) {

94
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java
View File

@@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.task.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.ModuleCleanupProperties;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
@@ -13,6 +14,7 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
@@ -24,60 +26,70 @@ import java.util.Set;
public class ModuleHistoryCleanupService {
private static final Set<String> TERMINAL_STATUSES = Set.of("SUCCESS", "FAILED", "CANCELLED", "CANCELED");
private static final Duration CLEANUP_LOCK_TTL = Duration.ofHours(2);
private final ModuleCleanupProperties moduleCleanupProperties;
private final FileTaskMapper fileTaskMapper;
private final FileResultMapper fileResultMapper;
private final DistributedJobLockService distributedJobLockService;
@Transactional
@Scheduled(cron = "${aiimage.module-cleanup.cron:0 0 0 * * *}")
public void cleanupConfiguredModules() {
if (!moduleCleanupProperties.isEnabled()) {
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("module-history:cleanup", CLEANUP_LOCK_TTL);
if (lockHandle == null) {
log.info("[module-cleanup] skip because another instance holds the distributed lock");
return;
}
List<String> moduleTypes = moduleCleanupProperties.getModuleTypes();
if (moduleTypes == null || moduleTypes.isEmpty()) {
return;
}
List<FileTaskEntity> moduleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.in(FileTaskEntity::getModuleType, moduleTypes)
.select(FileTaskEntity::getId, FileTaskEntity::getModuleType, FileTaskEntity::getStatus));
List<Long> cleanupTaskIds = new ArrayList<>();
List<Long> skippedActiveTaskIds = new ArrayList<>();
for (FileTaskEntity task : moduleTasks) {
if (task == null || task.getId() == null) {
continue;
try (lockHandle) {
if (!moduleCleanupProperties.isEnabled()) {
return;
}
if (isTerminalStatus(task.getStatus())) {
cleanupTaskIds.add(task.getId());
} else {
skippedActiveTaskIds.add(task.getId());
List<String> moduleTypes = moduleCleanupProperties.getModuleTypes();
if (moduleTypes == null || moduleTypes.isEmpty()) {
return;
}
List<FileTaskEntity> moduleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.in(FileTaskEntity::getModuleType, moduleTypes)
.select(FileTaskEntity::getId, FileTaskEntity::getModuleType, FileTaskEntity::getStatus));
List<Long> cleanupTaskIds = new ArrayList<>();
List<Long> skippedActiveTaskIds = new ArrayList<>();
for (FileTaskEntity task : moduleTasks) {
if (task == null || task.getId() == null) {
continue;
}
if (isTerminalStatus(task.getStatus())) {
cleanupTaskIds.add(task.getId());
} else {
skippedActiveTaskIds.add(task.getId());
}
}
if (cleanupTaskIds.isEmpty()) {
log.info("模块历史清理跳过: moduleTypes={}, activeTaskIds={}, reason=no-terminal-tasks",
moduleTypes, skippedActiveTaskIds);
return;
}
int deletedResults = fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.in(FileResultEntity::getModuleType, moduleTypes)
.in(FileResultEntity::getTaskId, cleanupTaskIds));
int resetTasks = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.in(FileTaskEntity::getId, cleanupTaskIds)
.set(FileTaskEntity::getResultJson, null)
.set(FileTaskEntity::getRequestJson, null)
.set(FileTaskEntity::getErrorMessage, null));
int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper<FileTaskEntity>()
.in(FileTaskEntity::getId, cleanupTaskIds));
log.info("模块历史清理完成: moduleTypes={}, deletedResults={}, resetTasks={}, deletedTasks={}, skippedActiveTaskIds={}",
moduleTypes, deletedResults, resetTasks, deletedTasks, skippedActiveTaskIds);
}
if (cleanupTaskIds.isEmpty()) {
log.info("模块历史清理跳过: moduleTypes={}, activeTaskIds={}, reason=no-terminal-tasks",
moduleTypes, skippedActiveTaskIds);
return;
}
int deletedResults = fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
.in(FileResultEntity::getModuleType, moduleTypes)
.in(FileResultEntity::getTaskId, cleanupTaskIds));
int resetTasks = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.in(FileTaskEntity::getId, cleanupTaskIds)
.set(FileTaskEntity::getResultJson, null)
.set(FileTaskEntity::getRequestJson, null)
.set(FileTaskEntity::getErrorMessage, null));
int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper<FileTaskEntity>()
.in(FileTaskEntity::getId, cleanupTaskIds));
log.info("模块历史清理完成: moduleTypes={}, deletedResults={}, resetTasks={}, deletedTasks={}, skippedActiveTaskIds={}",
moduleTypes, deletedResults, resetTasks, deletedTasks, skippedActiveTaskIds);
}
private boolean isTerminalStatus(String status) {

21
backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexRefreshService.java
View File

@@ -1,23 +1,36 @@
package com.nanri.aiimage.modules.ziniao.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
@Service
@RequiredArgsConstructor
@Slf4j
public class ZiniaoShopIndexRefreshService {
private static final Duration REFRESH_LOCK_TTL = Duration.ofMinutes(30);
private final ZiniaoShopIndexService ziniaoShopIndexService;
private final DistributedJobLockService distributedJobLockService;
@Scheduled(cron = "${aiimage.ziniao.shop-index-refresh-cron:0 */10 * * * *}")
public void refreshShopIndex() {
try {
ziniaoShopIndexService.refreshShopIndex();
} catch (Exception ex) {
log.warn("[ziniao-index] refresh failed: {}", ex.getMessage());
DistributedJobLockService.LockHandle lockHandle = distributedJobLockService.tryLock("ziniao:shop-index-refresh", REFRESH_LOCK_TTL);
if (lockHandle == null) {
log.info("[ziniao-index] skip refresh because another instance holds the distributed lock");
return;
}
try (lockHandle) {
try {
ziniaoShopIndexService.refreshShopIndex();
} catch (Exception ex) {
log.warn("[ziniao-index] refresh failed: {}", ex.getMessage());
}
}
}
}