diff --git a/app/.env b/app/.env index a8104b3..8a4270d 100644 --- a/app/.env +++ b/app/.env @@ -13,7 +13,7 @@ client_name=ShuFuAI # java_api_base=http://47.111.163.154:18080 -java_api_base=http://127.0.0.1:18080 -# java_api_base=http://8.136.19.173:18080 +# java_api_base=http://127.0.0.1:18080 +java_api_base=http://8.136.19.173:18080 diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java b/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java new file mode 100644 index 0000000..5ae7736 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java @@ -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 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"; + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index 7c7e77d..8b8dbc1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -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 runningTasks = brandCrawlTaskMapper.selectList(new LambdaQueryWrapper() .eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING) @@ -1075,6 +1085,7 @@ public class BrandTaskService { continue; } failStaleRunningTask(task.getId(), "前端长时间无响应，任务已自动失败"); + } } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java index afd8b27..0a8d949 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java @@ -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 runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .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 runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() - .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 runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .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. + } } } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskShopPayloadDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskShopPayloadDto.java index 835d788..ee8a286 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskShopPayloadDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/model/dto/ProductRiskShopPayloadDto.java @@ -30,4 +30,6 @@ public class ProductRiskShopPayloadDto { description = "五国数据：JSON 对象的键为枚举 ProductRiskCountryCode（DE、FR、ES、IT、UK），" + "值为该国 sheet 的行列表；省略的键表示该国无数据行。") private Map> countries = new LinkedHashMap<>(); + + private Boolean shopDone; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskCacheService.java index 2a48521..c509033 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskCacheService.java @@ -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; + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java index ef4448f..f841151 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/productrisk/service/ProductRiskTaskService.java @@ -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 taskIds) { @@ -338,6 +337,7 @@ public class ProductRiskTaskService { task.setCreatedAt(LocalDateTime.now()); task.setUpdatedAt(LocalDateTime.now()); fileTaskMapper.insert(task); + productRiskTaskCacheService.touchTaskHeartbeat(task.getId()); List 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> 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 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) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskCacheService.java index 7933547..31a908a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskCacheService.java @@ -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) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java index dea5548..b37b684 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopmatch/service/ShopMatchTaskService.java @@ -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 countryCodes = normalizeCountryCodes(request.getCountryCodes()); List 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 dedupeQueueItems(List raw) { Map byNorm = new LinkedHashMap<>(); for (ProductRiskShopQueueItemVo item : raw) { @@ -857,7 +871,7 @@ public class ShopMatchTaskService { return List.of(); } List out = new ArrayList<>(); - LocalDateTime now = LocalDateTime.now(); + LocalDateTime now = now(); LocalDateTime previous = null; for (LocalDateTime time : raw) { if (time == null) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java index 17177b4..69513d4 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java @@ -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 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 moduleTypes = moduleCleanupProperties.getModuleTypes(); - if (moduleTypes == null || moduleTypes.isEmpty()) { - return; - } - - List moduleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() - .in(FileTaskEntity::getModuleType, moduleTypes) - .select(FileTaskEntity::getId, FileTaskEntity::getModuleType, FileTaskEntity::getStatus)); - - List cleanupTaskIds = new ArrayList<>(); - List 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 moduleTypes = moduleCleanupProperties.getModuleTypes(); + if (moduleTypes == null || moduleTypes.isEmpty()) { + return; } + + List moduleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .in(FileTaskEntity::getModuleType, moduleTypes) + .select(FileTaskEntity::getId, FileTaskEntity::getModuleType, FileTaskEntity::getStatus)); + + List cleanupTaskIds = new ArrayList<>(); + List 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() + .in(FileResultEntity::getModuleType, moduleTypes) + .in(FileResultEntity::getTaskId, cleanupTaskIds)); + + int resetTasks = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .in(FileTaskEntity::getId, cleanupTaskIds) + .set(FileTaskEntity::getResultJson, null) + .set(FileTaskEntity::getRequestJson, null) + .set(FileTaskEntity::getErrorMessage, null)); + + int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper() + .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() - .in(FileResultEntity::getModuleType, moduleTypes) - .in(FileResultEntity::getTaskId, cleanupTaskIds)); - - int resetTasks = fileTaskMapper.update(null, new LambdaUpdateWrapper() - .in(FileTaskEntity::getId, cleanupTaskIds) - .set(FileTaskEntity::getResultJson, null) - .set(FileTaskEntity::getRequestJson, null) - .set(FileTaskEntity::getErrorMessage, null)); - - int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper() - .in(FileTaskEntity::getId, cleanupTaskIds)); - - log.info("模块历史清理完成: moduleTypes={}, deletedResults={}, resetTasks={}, deletedTasks={}, skippedActiveTaskIds={}", - moduleTypes, deletedResults, resetTasks, deletedTasks, skippedActiveTaskIds); } private boolean isTerminalStatus(String status) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexRefreshService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexRefreshService.java index 8d3a90b..11e2ee2 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexRefreshService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexRefreshService.java @@ -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()); + } } } } diff --git a/backend/web_source/admin.html b/backend/web_source/admin.html index cfca98b..3092098 100644 --- a/backend/web_source/admin.html +++ b/backend/web_source/admin.html @@ -198,6 +198,10 @@ +
+ + +

@@ -210,6 +214,7 @@ ID 栏目名 栏目标识 + 菜单路由 创建时间 操作 @@ -522,6 +527,10 @@ +
+ + +

@@ -771,6 +780,71 @@ updateShopManageAccess(); }); } + function applyTabAccess(tabName, canUse) { + var tab = document.querySelector('.tab[data-tab="' + tabName + '"]'); + var panel = document.getElementById('panel-' + tabName); + if (tab) tab.style.display = canUse ? '' : 'none'; + if (panel) panel.style.display = canUse ? '' : 'none'; + if (!canUse && tab && tab.classList.contains('active')) { + tab.classList.remove('active'); + if (panel) panel.classList.remove('active'); + var usersTab = document.querySelector('.tab[data-tab="users"]'); + var usersPanel = document.getElementById('panel-users'); + if (usersTab) usersTab.classList.add('active'); + if (usersPanel) usersPanel.classList.add('active'); + } + } + function updateAdminOnlyAccess() { + var canUse = currentUserRole === 'super_admin'; + applyTabAccess('columns', canUse); + applyTabAccess('shop-keys', canUse); + } + function updateDedupeTotalDataAccess() { + var canUse = currentUserRole === 'super_admin' || + !!currentUserAdminPermissionKeys['admin_dedupe_total_data'] || + !!currentUserAdminPermissionRoutes['dedupe-total-data']; + applyTabAccess('dedupe-total-data', canUse); + } + function updateShopManageAccess() { + var canUse = currentUserRole === 'super_admin' || + !!currentUserAdminPermissionKeys['admin_shop_manage'] || + !!currentUserAdminPermissionRoutes['shop-manage']; + applyTabAccess('shop-manage', canUse); + } + function loadCurrentUserAdminPermissions() { + currentUserAdminPermissionKeys = {}; + currentUserAdminPermissionRoutes = {}; + updateAdminOnlyAccess(); + if (!currentUserId || currentUserRole === 'super_admin') { + updateDedupeTotalDataAccess(); + updateShopManageAccess(); + return; + } + fetch('/api/admin/user/' + currentUserId + '/column-permissions?menu_type=admin') + .then(function(r) { return r.json(); }) + .then(function(res) { + if (!res.success) { + updateAdminOnlyAccess(); + updateDedupeTotalDataAccess(); + updateShopManageAccess(); + return; + } + (res.items || []).forEach(function(item) { + var columnKey = (item.column_key || '').trim(); + var routePath = (item.route_path || '').trim(); + if (columnKey) currentUserAdminPermissionKeys[columnKey] = true; + if (routePath) currentUserAdminPermissionRoutes[routePath] = true; + }); + updateAdminOnlyAccess(); + updateDedupeTotalDataAccess(); + updateShopManageAccess(); + }) + .catch(function() { + updateAdminOnlyAccess(); + updateDedupeTotalDataAccess(); + updateShopManageAccess(); + }); + } var allColumnsList = []; function loadColumnsForPermission() { fetch('/api/admin/columns') @@ -1994,6 +2068,115 @@ document.getElementById('editColumnModal').classList.remove('show'); }; + function loadColumns() { + fetch('/api/admin/columns') + .then(function(r) { return r.json(); }) + .then(function(res) { + var tbody = document.getElementById('columnListBody'); + if (!res.success) { + tbody.innerHTML = '加载失败: ' + (res.error || '') + ''; + return; + } + allColumnsList = res.items || []; + if (allColumnsList.length === 0) { + tbody.innerHTML = '暂无栏目，请先在上方新增'; + } else { + tbody.innerHTML = allColumnsList.map(function(c) { + return '' + c.id + '' + (c.name || '') + '' + (c.column_key || '') + '' + (c.route_path || '') + '' + (c.created_at || '') + '' + + ' ' + + ''; + }).join(''); + } + bindColumnActions(); + }) + .catch(function() { + document.getElementById('columnListBody').innerHTML = '请求失败'; + }); + } + function bindColumnActions() { + document.querySelectorAll('[data-column-edit]').forEach(function(btn) { + btn.onclick = function() { + document.getElementById('editColumnId').value = btn.dataset.columnEdit || ''; + document.getElementById('editColumnName').value = (btn.dataset.name || '').replace(/"/g, '"'); + document.getElementById('editColumnKey').value = (btn.dataset.key || '').replace(/"/g, '"'); + document.getElementById('editColumnRoutePath').value = (btn.dataset.route || '').replace(/"/g, '"'); + document.getElementById('msgEditColumn').textContent = ''; + document.getElementById('editColumnModal').classList.add('show'); + }; + }); + document.querySelectorAll('[data-column-delete]').forEach(function(btn) { + btn.onclick = function() { + if (!confirm('确定删除栏目「' + (btn.dataset.name || '').replace(/"/g, '"') + '」吗？')) return; + fetch('/api/admin/column/' + btn.dataset.columnDelete, { method: 'DELETE' }) + .then(function(r) { return r.json(); }) + .then(function(res) { + if (res.success) { loadColumns(); loadColumnsForPermission(); } + else { alert(res.error || '删除失败'); } + }); + }; + }); + } + document.getElementById('btnAddColumn').onclick = function() { + var name = (document.getElementById('columnName').value || '').trim(); + var key = (document.getElementById('columnKey').value || '').trim(); + var routePath = (document.getElementById('columnRoutePath').value || '').trim(); + var msgEl = document.getElementById('msgColumn'); + msgEl.textContent = ''; + msgEl.className = 'msg'; + if (!name) { msgEl.textContent = '请填写栏目名'; msgEl.classList.add('err'); return; } + if (!key) { msgEl.textContent = '请填写栏目标识'; msgEl.classList.add('err'); return; } + if (!routePath) { msgEl.textContent = '请填写菜单路由'; msgEl.classList.add('err'); return; } + fetch('/api/admin/column', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ name: name, column_key: key, route_path: routePath, menu_type: 'admin' }) + }) + .then(function(r) { return r.json(); }) + .then(function(res) { + if (res.success) { + msgEl.textContent = res.msg || '新增成功'; + msgEl.classList.add('ok'); + document.getElementById('columnName').value = ''; + document.getElementById('columnKey').value = ''; + document.getElementById('columnRoutePath').value = ''; + loadColumns(); + loadColumnsForPermission(); + } else { + msgEl.textContent = res.error || '新增失败'; + msgEl.classList.add('err'); + } + }) + .catch(function() { msgEl.textContent = '请求失败'; msgEl.classList.add('err'); }); + }; + document.getElementById('btnSaveColumn').onclick = function() { + var cid = document.getElementById('editColumnId').value; + var name = (document.getElementById('editColumnName').value || '').trim(); + var key = (document.getElementById('editColumnKey').value || '').trim(); + var routePath = (document.getElementById('editColumnRoutePath').value || '').trim(); + var msgEl = document.getElementById('msgEditColumn'); + msgEl.textContent = ''; + msgEl.className = 'msg'; + if (!name) { msgEl.textContent = '请填写栏目名'; msgEl.classList.add('err'); return; } + if (!key) { msgEl.textContent = '请填写栏目标识'; msgEl.classList.add('err'); return; } + if (!routePath) { msgEl.textContent = '请填写菜单路由'; msgEl.classList.add('err'); return; } + fetch('/api/admin/column/' + cid, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ name: name, column_key: key, route_path: routePath, menu_type: 'admin' }) + }) + .then(function(r) { return r.json(); }) + .then(function(res) { + if (res.success) { + document.getElementById('editColumnModal').classList.remove('show'); + loadColumns(); + loadColumnsForPermission(); + } else { + msgEl.textContent = res.error || '保存失败'; + msgEl.classList.add('err'); + } + }); + }; + // ========== 分页 ========== function renderPagination(elId, total, page, pageSize, onPage) { var el = document.getElementById(elId); diff --git a/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue b/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue index 9e3f585..e9277b2 100644 --- a/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandDeleteBrandTab.vue @@ -54,6 +54,10 @@ + {{ running ? '正在读取删除品牌数据，请稍候…' : '解析后会在右侧展示按国家分组的去重结果' }} @@ -133,9 +137,6 @@ {{ getTaskStatusInfo(item).text }} -
保存中...
+
商品列表筛选
+

与“推送到 Python 队列”一并下发，供 Python 区分处理场景。

+
+ + + +
定时执行
@@ -100,6 +107,7 @@ import { computed, onMounted, onUnmounted, ref } from 'vue' import { ElMessage } from 'element-plus' import BrandTopBar from '@/pages/brand/components/BrandTopBar.vue' +import { LISTING_FILTER_OPTIONS, type ListingFilterValue } from '@/pages/brand/components/listingFilters' import { activateShopMatchTask, addShopMatchCandidate, createShopMatchTask, deleteShopMatchCandidate, deleteShopMatchHistory, deleteShopMatchTask, getShopMatchCountryPreference, getShopMatchDashboard, getShopMatchHistory, getShopMatchResultDownloadUrl, getShopMatchTasksBatch, listShopMatchCandidates, matchShopMatchShops, putShopMatchCountryPreference, type ShopMatchCandidateVo, type ShopMatchDashboardVo, type ShopMatchHistoryItem, type ShopMatchShopQueueItem, type ShopMatchTaskDetailVo } from '@/shared/api/java-modules' import { getPywebviewApi } from '@/shared/bridges/pywebview' @@ -108,6 +116,7 @@ const shopInput = ref('') const candidates = ref([]) const selectedCandidates = ref([]) const matchedItems = ref([]) +const shopMatchListingFilter = ref('Active') const adding = ref(false) const matching = ref(false) const pushing = ref(false) @@ -144,6 +153,7 @@ function sessionKey() { return `shop-match:tasks:${uidForStorage()}` } function taskStatusStorageKey() { return `shop-match:task-status:${uidForStorage()}` } function taskSnapshotsStorageKey() { return `shop-match:task-snapshots:${uidForStorage()}` } function matchedItemsStorageKey() { return `shop-match:matched-items:${uidForStorage()}` } +function setStorageJson(key: string, value: unknown, shouldRemove: boolean) { if (shouldRemove) window.localStorage.removeItem(key); else window.localStorage.setItem(key, JSON.stringify(value)) } function rowKeyForMatch(row: ShopMatchShopQueueItem) { return `${(row.shopName || '').trim()}::${row.shopId || ''}` } function taskRowKey(item: ShopMatchHistoryItem) { return `${normalizeTaskId(item.taskId)}-${item.resultId || 0}-${item.shopName || ''}` } function countryLabel(code: string) { return COUNTRY_OPTIONS.find((item) => item.code === code)?.label || code } @@ -164,13 +174,13 @@ function formatDateTime(value?: string) { return value ? value.replace('T', ' ') function formatTimeOnly(value?: string) { return value ? value.slice(11, 16) : '-' } function uniqueTaskRows(rows: ShopMatchHistoryItem[], snapshots: Record, section: 'current' | 'history') { const map = new Map(); for (const row of rows) map.set(taskRowKey(row), row); for (const [taskIdText, snapshot] of Object.entries(snapshots)) { const taskId = normalizeTaskId(taskIdText); const first = snapshot.items?.[0]; if (!first || !taskId) continue; const terminal = isTaskTerminalById(taskId); if ((section === 'current' && terminal) || (section === 'history' && !terminal)) continue; const key = taskRowKey({ ...first, taskId }); if (!map.has(key)) map.set(key, { ...first, taskId }) } return Array.from(map.values()) } function loadPollingIdsFromStorage() { try { const raw = window.localStorage.getItem(sessionKey()); const parsed = raw ? JSON.parse(raw) : []; pollingTaskIds.value = Array.isArray(parsed) ? parsed.filter((item): item is number => typeof item === 'number' && item > 0) : [] } catch { pollingTaskIds.value = [] } } -function savePollingIds() { window.localStorage.setItem(sessionKey(), JSON.stringify(pollingTaskIds.value)) } +function savePollingIds() { setStorageJson(sessionKey(), pollingTaskIds.value, pollingTaskIds.value.length === 0) } function loadTaskDetailsFromStorage() { try { const raw = window.localStorage.getItem(taskStatusStorageKey()); taskDetails.value = raw ? JSON.parse(raw) : {} } catch { taskDetails.value = {} } } -function saveTaskDetailsToStorage() { window.localStorage.setItem(taskStatusStorageKey(), JSON.stringify(taskDetails.value)) } +function saveTaskDetailsToStorage() { setStorageJson(taskStatusStorageKey(), taskDetails.value, Object.keys(taskDetails.value).length === 0) } function loadTaskSnapshotsFromStorage() { try { const raw = window.localStorage.getItem(taskSnapshotsStorageKey()); taskSnapshots.value = raw ? JSON.parse(raw) : {} } catch { taskSnapshots.value = {} } } -function saveTaskSnapshotsToStorage() { window.localStorage.setItem(taskSnapshotsStorageKey(), JSON.stringify(taskSnapshots.value)) } +function saveTaskSnapshotsToStorage() { setStorageJson(taskSnapshotsStorageKey(), taskSnapshots.value, Object.keys(taskSnapshots.value).length === 0) } function loadMatchedItemsFromStorage() { try { const raw = window.localStorage.getItem(matchedItemsStorageKey()); matchedItems.value = raw ? JSON.parse(raw) : [] } catch { matchedItems.value = [] } } -function saveMatchedItemsToStorage() { window.localStorage.setItem(matchedItemsStorageKey(), JSON.stringify(matchedItems.value)) } +function saveMatchedItemsToStorage() { setStorageJson(matchedItemsStorageKey(), matchedItems.value, matchedItems.value.length === 0) } function syncPollingIdsWithTaskState() { const nextIds = pollingTaskIds.value.filter((taskId) => (taskDetails.value[taskId] || taskSnapshots.value[taskId]?.task?.status) === 'RUNNING'); if (nextIds.length === pollingTaskIds.value.length) return; pollingTaskIds.value = nextIds; savePollingIds() } function addPollingTask(taskId: number) { if (!pollingTaskIds.value.includes(taskId)) { pollingTaskIds.value = [...pollingTaskIds.value, taskId]; savePollingIds() } } function clearDispatchTimer(taskId: number) { const timer = dispatchTimers.get(taskId); if (timer) { window.clearTimeout(timer); dispatchTimers.delete(taskId) } } @@ -209,7 +219,7 @@ async function runMatch() { const names = selectedCandidates.value.map((item) => function removeMatchedRow(row: ShopMatchShopQueueItem) { matchedItems.value = matchedItems.value.filter((item) => rowKeyForMatch(item) !== rowKeyForMatch(row)); saveMatchedItemsToStorage() } function removeMatchedRowsLocally(rows: ShopMatchHistoryItem[] | ShopMatchShopQueueItem[]) { const keys = new Set(rows.map((row) => rowKeyForMatch(row as ShopMatchShopQueueItem))); matchedItems.value = matchedItems.value.filter((item) => !keys.has(rowKeyForMatch(item))); saveMatchedItemsToStorage() } function parseScheduleValues() { if (!scheduleEnabled.value) return undefined; const values = schedulePickerValues.value.map((item) => item.trim()).filter(Boolean); if (!values.length) throw new Error('请至少选择一个执行时间'); const withTime = values.map((value) => { const normalized = normalizeScheduleValue(value); const date = resolveScheduleDateTime(normalized); if (!date) throw new Error(`时间格式无效: ${value}`); return { raw: formatScheduleDateTime(date), time: date.getTime() } }).sort((a, b) => a.time - b.time); for (let i = 1; i < withTime.length; i += 1) if (withTime[i].time === withTime[i - 1].time) throw new Error('执行时间不能重复'); return withTime.map((item) => item.raw) } -function buildQueuePayload(taskId: number, item: Pick, countryCodes: string[], stageIndex?: number, finalStage = true) { return { type: 'shop-match-run', ts: Date.now(), data: { taskId, user_id: currentUserId(), items: [{ shopName: item.shopName, shopId: item.shopId, platform: item.platform, companyName: item.companyName, matched: true, matchStatus: item.matchStatus, matchMessage: item.matchMessage }], country_codes: [...countryCodes], stage_index: stageIndex, final_stage: finalStage } } } +function buildQueuePayload(taskId: number, item: Pick, countryCodes: string[], stageIndex?: number, finalStage = true) { return { type: 'shop-match-run', ts: Date.now(), data: { taskId, user_id: currentUserId(), items: [{ shopName: item.shopName, shopId: item.shopId, platform: item.platform, companyName: item.companyName, matched: true, matchStatus: item.matchStatus, matchMessage: item.matchMessage }], country_codes: [...countryCodes], risk_listing_filter: shopMatchListingFilter.value, stage_index: stageIndex, final_stage: finalStage } } } function nextScheduleStage(taskId: number) { const task = taskSnapshots.value[taskId]?.task; const stages = task?.scheduleStages || []; const currentStageIndex = task?.currentStageIndex; if (typeof currentStageIndex !== 'number') return null; const stage = stages.find((item) => item.stageIndex === currentStageIndex); if (!stage?.scheduledAt) return null; return { stageIndex: currentStageIndex, scheduledAt: stage.scheduledAt, finalStage: currentStageIndex === stages.length - 1 } } async function dispatchScheduledTask(taskId: number, item: ShopMatchHistoryItem) { const api = getPywebviewApi(); if (!api?.enqueue_json) throw new Error('当前客户端未提供 enqueue_json'); const stage = nextScheduleStage(taskId); const snapshot = taskSnapshots.value[taskId]; const countryCodes = snapshot?.task?.countryCodes || orderedCountryCodes.value; if (!stage) throw new Error(`任务 ${taskId} 缺少待执行阶段`); clearDispatchTimer(taskId); await activateShopMatchTask(taskId, stage.stageIndex); taskDetails.value = { ...taskDetails.value, [taskId]: 'RUNNING' }; if (snapshot?.task) taskSnapshots.value = { ...taskSnapshots.value, [taskId]: { ...snapshot, task: { ...snapshot.task, status: 'RUNNING', activeStageIndex: stage.stageIndex } } }; saveTaskSnapshotsToStorage(); saveTaskDetailsToStorage(); const payload = buildQueuePayload(taskId, item, countryCodes, stage.stageIndex, stage.finalStage); queuePayloadText.value = JSON.stringify(payload, null, 2); const result = await api.enqueue_json(payload); if (!result?.success) throw new Error(result?.error || `任务 ${taskId} 推送失败`); addPollingTask(taskId); queuePushResult.value = `任务 ${taskId} 已推送第 ${stage.stageIndex + 1} 轮`; ensurePolling(true) } function hasRunningTask() { return Object.values(taskDetails.value).some((status) => status === 'RUNNING') || Object.values(taskSnapshots.value).some((detail) => detail.task?.status === 'RUNNING') } @@ -247,6 +257,7 @@ async function deleteTaskRecord(item: ShopMatchHistoryItem) { } else if (resultId > 0) { await deleteShopMatchHistory(resultId) historyItems.value = historyItems.value.filter((row) => Number(row.resultId || 0) !== resultId) + removeMatchedRowsLocally([item]) } else { throw new Error('缺少可删除的任务标识') } diff --git a/frontend-vue/src/pages/brand/components/BrandTopBar.vue b/frontend-vue/src/pages/brand/components/BrandTopBar.vue index ea54b7c..b2a07c7 100644 --- a/frontend-vue/src/pages/brand/components/BrandTopBar.vue +++ b/frontend-vue/src/pages/brand/components/BrandTopBar.vue @@ -5,22 +5,33 @@ 返回首页
-
@@ -28,16 +39,33 @@