diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyFileEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyFileEntity.java index 6ea5f184..f9746802 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyFileEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyFileEntity.java @@ -24,8 +24,10 @@ public class ShopDataCrawlDailyFileEntity { private String resultFileUrl; private Long resultFileSize; private String resultContentType; + private String countryCodesJson; private Integer rowCount; private Long version; + private Integer compensationDone; private LocalDateTime lastSuccessAt; private LocalDateTime createdAt; private LocalDateTime updatedAt; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyCompensationRunner.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyCompensationRunner.java new file mode 100644 index 00000000..78cdbb4a --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyCompensationRunner.java @@ -0,0 +1,29 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.ApplicationArguments; +import org.springframework.boot.ApplicationRunner; +import org.springframework.stereotype.Component; + +/** + * V95 店铺级累计文件迁移的启动补偿：对 compensation_done=0 的每日累计文件， + * 按成员 result 快照重建合并工作簿（覆盖国家语义），补齐国家列表并清理旧对象。 + * 失败仅记日志，不阻断应用启动与后续补偿。 + */ +@Component +@RequiredArgsConstructor +@Slf4j +public class ShopDataCrawlDailyCompensationRunner implements ApplicationRunner { + + private final ShopDataCrawlTaskService taskService; + + @Override + public void run(ApplicationArguments args) { + try { + taskService.compensateDailyWorkbooksAfterMigration(); + } catch (Exception ex) { + log.warn("[shop-data-crawl] daily compensation runner failed: {}", ex.getMessage(), ex); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java index 57f51013..579492d8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java @@ -73,34 +73,47 @@ public class ShopDataCrawlDailyFileService { return HexFormat.of().formatHex(digest); } - public TaskDistributedLockService.LockHandle acquireLock(Long userId, String shopKey) { - if (userId == null || userId <= 0 || shopKey == null || shopKey.isBlank()) { + public TaskDistributedLockService.LockHandle acquireLock(String shopKey) { + if (shopKey == null || shopKey.isBlank()) { return null; } - // The lock protects the shop's whole daily-file lifecycle. Including the - // date would allow yesterday and today to update the same shop together. - String identity = userId + "|" + shopKey; + // 店铺级锁：只按店铺串行，跨账号同店并发抓取时互斥（锁在 Redis，与 DB 行解耦） + String identity = shopKey; String lockModule = DAILY_LOCK_MODULE_PREFIX + shopKeyHash(identity); return taskDistributedLockService.acquire(lockModule, 1L, DAILY_LOCK_TTL, DAILY_LOCK_WAIT_MILLIS); } - public ShopDataCrawlDailyFileEntity findForUpdate(Long userId, String shopKeyHash, LocalDate businessDate) { - if (userId == null || shopKeyHash == null || businessDate == null) { + public ShopDataCrawlDailyFileEntity findForUpdate(String shopKeyHash, LocalDate businessDate) { + if (shopKeyHash == null || businessDate == null) { return null; } return dailyFileMapper.selectOne(new LambdaQueryWrapper() - .eq(ShopDataCrawlDailyFileEntity::getUserId, userId) .eq(ShopDataCrawlDailyFileEntity::getShopKeyHash, shopKeyHash) .eq(ShopDataCrawlDailyFileEntity::getBusinessDate, businessDate) .last("FOR UPDATE")); } - public List findOlder(Long userId, String shopKeyHash, LocalDate businessDate) { - if (userId == null || shopKeyHash == null || businessDate == null) { + public List findForCompensation() { + return dailyFileMapper.selectList(new LambdaQueryWrapper() + .eq(ShopDataCrawlDailyFileEntity::getCompensationDone, 0) + .orderByAsc(ShopDataCrawlDailyFileEntity::getBusinessDate) + .orderByAsc(ShopDataCrawlDailyFileEntity::getId) + .last("LIMIT 500")); + } + + public void markCompensationDone(ShopDataCrawlDailyFileEntity entity) { + if (entity == null || entity.getId() == null) { + return; + } + entity.setCompensationDone(1); + update(entity); + } + + public List findOlder(String shopKeyHash, LocalDate businessDate) { + if (shopKeyHash == null || businessDate == null) { return List.of(); } return dailyFileMapper.selectList(new LambdaQueryWrapper() - .eq(ShopDataCrawlDailyFileEntity::getUserId, userId) .eq(ShopDataCrawlDailyFileEntity::getShopKeyHash, shopKeyHash) .lt(ShopDataCrawlDailyFileEntity::getBusinessDate, businessDate) .orderByDesc(ShopDataCrawlDailyFileEntity::getBusinessDate) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index 671b676a..96e60d7d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -63,6 +63,7 @@ import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -676,7 +677,7 @@ public class ShopDataCrawlTaskService { ensureTaskOwnedByCurrentInstance(task, "delete shop data crawl task"); try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) { List taskRows = listTaskRows(taskId); - try (DailyLockSet dailyLocks = acquireDailyLocks(task.getUserId(), taskRows)) { + try (DailyLockSet dailyLocks = acquireDailyLocks(taskRows)) { ensureDailySyncCompletedBeforeDelete(taskRows); Set removedResultIds = taskRows.stream() .map(FileResultEntity::getId) @@ -739,7 +740,7 @@ public class ShopDataCrawlTaskService { if (latestEntity == null || !MODULE_TYPE.equals(latestEntity.getModuleType()) || !userId.equals(latestEntity.getUserId())) { throw new BusinessException("记录不存在"); } - try (DailyLockSet dailyLocks = acquireDailyLocks(userId, List.of(latestEntity))) { + try (DailyLockSet dailyLocks = acquireDailyLocks(List.of(latestEntity))) { deleteResultHistoryRow(latestEntity); } } @@ -816,7 +817,7 @@ public class ShopDataCrawlTaskService { continue; } ShopDataCrawlDailyFileEntity lockedDailyFile = dailyFileService.findForUpdate( - dailyFile.getUserId(), dailyFile.getShopKeyHash(), dailyFile.getBusinessDate()); + dailyFile.getShopKeyHash(), dailyFile.getBusinessDate()); if (lockedDailyFile != null) { dailyFile = lockedDailyFile; } @@ -960,6 +961,7 @@ public class ShopDataCrawlTaskService { dailyFile.setResultFileSize(latest.result().getResultFileSize()); dailyFile.setResultContentType(CONTENT_TYPE_XLSX); dailyFile.setRowCount(rowCount); + dailyFile.setCountryCodesJson(countryCodesJson(collectCountryCodes(snapshots))); dailyFile.setVersion(Math.max(0L, Objects.requireNonNullElse(dailyFile.getVersion(), 0L)) + 1L); dailyFile.setLastSuccessAt(now); dailyFile.setUpdatedAt(now); @@ -1801,10 +1803,9 @@ public class ShopDataCrawlTaskService { FileResultEntity row, ShopDataCrawlResultItemVo snapshot, LocalDate businessDate) { - Long userId = row.getUserId() != null ? row.getUserId() : task.getUserId(); String shopKey = dailyFileService.shopKey(row); String shopKeyHash = dailyFileService.shopKeyHash(shopKey); - if (userId == null || shopKeyHash == null) { + if (shopKeyHash == null) { throw new BusinessException("店铺累计文件归属信息不完整"); } // Task 36：版本号/CAS 短临界区。店铺级锁只覆盖“准备/提交”两个毫秒级短事务 @@ -1814,13 +1815,13 @@ public class ShopDataCrawlTaskService { DailyWorkbookArtifact artifact = null; for (int attempt = 1; attempt <= MAX_DAILY_AGGREGATION_ATTEMPTS; attempt++) { DailyAggregationPreparation preparation; - TaskDistributedLockService.LockHandle prepareLock = dailyFileService.acquireLock(userId, shopKey); + TaskDistributedLockService.LockHandle prepareLock = dailyFileService.acquireLock(shopKey); if (prepareLock == null) { throw new BusinessException("店铺当天累计文件正在处理中，请稍后重试"); } try { preparation = executeShortTransaction( - () -> prepareDailyAggregation(userId, shopKeyHash, businessDate, row)); + () -> prepareDailyAggregation(shopKeyHash, businessDate, row)); } finally { prepareLock.close(); } @@ -1828,7 +1829,7 @@ public class ShopDataCrawlTaskService { return new DailyAggregationResult(List.of(), false); } ShopDataCrawlDailyFileEntity baseForAttempt = resolveBaseDailyFile( - preparation.dailyFile(), userId, shopKeyHash, businessDate); + preparation.dailyFile(), shopKeyHash, businessDate); // 组装在锁外执行。每次尝试都用本次准备阶段读到的最新 base 组装 // （冲突重试时 base 已变化，复用过期的组装结果会把并发写入的行丢在对象外）； // 零行引用对象不重复上传。 @@ -1836,7 +1837,7 @@ public class ShopDataCrawlTaskService { artifact = assembleDailyWorkbook(task, snapshot, baseForAttempt, excelAssemblyService.countRows(List.of(snapshot))); } - TaskDistributedLockService.LockHandle commitLock = dailyFileService.acquireLock(userId, shopKey); + TaskDistributedLockService.LockHandle commitLock = dailyFileService.acquireLock(shopKey); if (commitLock == null) { throw new BusinessException("店铺当天累计文件正在处理中，请稍后重试"); } @@ -1845,7 +1846,7 @@ public class ShopDataCrawlTaskService { DailyWorkbookArtifact artifactForAttempt = artifact; try { persistedResult = executeShortTransaction(() -> persistDailyAggregation( - task, row, userId, shopKey, shopKeyHash, businessDate, + task, row, shopKey, shopKeyHash, businessDate, preparationForAttempt, artifactForAttempt, snapshot, preparationForAttempt.dailyFile())); if (persistedResult.discardUploadedObject() && artifactForAttempt.uploaded()) { @@ -1876,12 +1877,11 @@ public class ShopDataCrawlTaskService { throw new BusinessException("店铺累计文件并发更新冲突，请稍后重试"); } - private DailyAggregationPreparation prepareDailyAggregation(Long userId, - String shopKeyHash, + private DailyAggregationPreparation prepareDailyAggregation(String shopKeyHash, LocalDate businessDate, FileResultEntity row) { ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate( - userId, shopKeyHash, businessDate); + shopKeyHash, businessDate); if (handleExistingDailyMembership(row, dailyFile)) { return new DailyAggregationPreparation(dailyFile, true); } @@ -1889,13 +1889,12 @@ public class ShopDataCrawlTaskService { } private ShopDataCrawlDailyFileEntity resolveBaseDailyFile(ShopDataCrawlDailyFileEntity currentDailyFile, - Long userId, String shopKeyHash, LocalDate businessDate) { if (currentDailyFile != null) { return currentDailyFile; } - List older = dailyFileService.findOlder(userId, shopKeyHash, businessDate); + List older = dailyFileService.findOlder(shopKeyHash, businessDate); return older == null || older.isEmpty() ? null : older.get(0); } @@ -1916,7 +1915,8 @@ public class ShopDataCrawlTaskService { Math.max(0L, Objects.requireNonNullElse(baseDailyFile.getResultFileSize(), 0L)), false, filename, - Math.max(0, Objects.requireNonNullElse(baseDailyFile.getRowCount(), 0))); + Math.max(0, Objects.requireNonNullElse(baseDailyFile.getRowCount(), 0)), + parseCountryCodesJson(baseDailyFile.getCountryCodesJson())); } File workRoot = FileUtil.mkdir(FileUtil.file( @@ -1935,7 +1935,8 @@ public class ShopDataCrawlTaskService { if (blank(objectKey)) { throw new BusinessException("累计文件上传后未返回文件地址"); } - return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, rowCount); + List countryCodes = collectCountryCodes(accumulatedItems); + return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, rowCount, countryCodes); } finally { FileUtil.del(outputXlsx); FileUtil.del(workRoot); @@ -1943,8 +1944,10 @@ public class ShopDataCrawlTaskService { } /** - * 从数据层累积重建每日累计文件的快照列表：既有成员行按 (createdAt, id) 升序读取， - * 优先用行级 payload；payload 缺失（历史数据）时按结果快照兜底；新结果追加在末尾。 + * 数据层增量模型：既有成员行按 (createdAt, id) 升序读取，优先用行级 payload； + * payload 缺失（历史数据）时按结果快照兜底；新结果追加在末尾。 + * 随后按国家覆盖：同一国家以最后一次出现的成员结果为准（本次回传的国家替换旧行， + * 未回传的国家保留），item 结构与顺序保持稳定。 */ private List buildDailyFileFromData(ShopDataCrawlDailyFileEntity baseDailyFile, List appended) { @@ -1967,7 +1970,93 @@ public class ShopDataCrawlTaskService { accumulated.add(item); } } - return accumulated; + return applyCountryCoverage(accumulated); + } + + /** + * 按国家覆盖：从后往前记录每个国家最后一次出现的成员下标；每个成员只保留 + * "该国最后一次出现" 的国家结果（非该次出现的行被剔除）。空结果国家同样覆盖旧行。 + * item 结构（resultId/shopName 等元数据）与顺序保持稳定。 + */ + private List applyCountryCoverage(List items) { + if (items == null || items.isEmpty()) { + return items == null ? List.of() : items; + } + Map lastCountryItemIndex = new HashMap<>(); + for (int i = 0; i < items.size(); i++) { + ShopDataCrawlResultItemVo item = items.get(i); + if (item == null) { + continue; + } + for (ShopDataCrawlCountryResultDto countryResult : item.getCountryResults() == null + ? List.of() : item.getCountryResults()) { + if (countryResult != null && !blank(countryResult.getCountry())) { + lastCountryItemIndex.put(countryResult.getCountry().trim().toUpperCase(), i); + } + } + } + if (lastCountryItemIndex.isEmpty()) { + return items; + } + for (int i = 0; i < items.size(); i++) { + ShopDataCrawlResultItemVo item = items.get(i); + if (item == null || item.getCountryResults() == null || item.getCountryResults().isEmpty()) { + continue; + } + List kept = new ArrayList<>(); + for (ShopDataCrawlCountryResultDto countryResult : item.getCountryResults()) { + if (countryResult == null || blank(countryResult.getCountry())) { + continue; + } + if (Objects.equals(lastCountryItemIndex.get(countryResult.getCountry().trim().toUpperCase()), i)) { + kept.add(countryResult); + } + } + item.setCountryResults(kept); + } + return items; + } + + private List collectCountryCodes(List items) { + LinkedHashSet codes = new LinkedHashSet<>(); + if (items != null) { + for (ShopDataCrawlResultItemVo item : items) { + if (item == null) { + continue; + } + for (ShopDataCrawlCountryResultDto countryResult : item.getCountryResults() == null + ? List.of() : item.getCountryResults()) { + if (countryResult != null && !blank(countryResult.getCountry())) { + codes.add(countryResult.getCountry().trim().toUpperCase()); + } + } + } + } + return new ArrayList<>(codes); + } + + private String countryCodesJson(List countryCodes) { + if (countryCodes == null || countryCodes.isEmpty()) { + return "[]"; + } + try { + return objectMapper.writeValueAsString(countryCodes); + } catch (Exception ex) { + return "[]"; + } + } + + private List parseCountryCodesJson(String json) { + if (blank(json)) { + return List.of(); + } + try { + List parsed = objectMapper.readValue(json, new TypeReference>() { + }); + return parsed == null ? List.of() : parsed; + } catch (Exception ex) { + return List.of(); + } } private ShopDataCrawlResultItemVo snapshotFromPayload(ShopDataCrawlDailyMemberEntity member) { @@ -2023,7 +2112,6 @@ public class ShopDataCrawlTaskService { private DailyAggregationResult persistDailyAggregation(FileTaskEntity task, FileResultEntity row, - Long userId, String shopKey, String shopKeyHash, LocalDate businessDate, @@ -2032,7 +2120,7 @@ public class ShopDataCrawlTaskService { ShopDataCrawlResultItemVo snapshot, ShopDataCrawlDailyFileEntity expectedBase) { ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate( - userId, shopKeyHash, businessDate); + shopKeyHash, businessDate); if (handleExistingDailyMembership(row, dailyFile)) { return new DailyAggregationResult(List.of(), true); } @@ -2044,7 +2132,7 @@ public class ShopDataCrawlTaskService { } List olderFiles = dailyFileService.findOlder( - userId, shopKeyHash, businessDate); + shopKeyHash, businessDate); Set obsoleteObjectKeys = new HashSet<>(); collectObjectKey(obsoleteObjectKeys, dailyFile == null ? null : dailyFile.getResultFileUrl()); for (ShopDataCrawlDailyFileEntity older : olderFiles) { @@ -2053,8 +2141,6 @@ public class ShopDataCrawlTaskService { String objectKey = artifact.objectKey(); String filename = artifact.filename(); - List shopRows = findShopResultRows(userId, row); - clearShopResultPointers(shopRows, row.getId()); row.setResultFilename(filename); row.setResultFileUrl(objectKey); row.setResultFileSize(artifact.fileSize()); @@ -2065,7 +2151,6 @@ public class ShopDataCrawlTaskService { LocalDateTime now = dailyFileService.currentBusinessDateTime(); if (dailyFile == null) { dailyFile = new ShopDataCrawlDailyFileEntity(); - dailyFile.setUserId(userId); dailyFile.setShopKeyHash(shopKeyHash); dailyFile.setShopKey(shopKey); dailyFile.setBusinessDate(businessDate); @@ -2081,6 +2166,8 @@ public class ShopDataCrawlTaskService { dailyFile.setResultFileSize(row.getResultFileSize()); dailyFile.setResultContentType(CONTENT_TYPE_XLSX); dailyFile.setRowCount(row.getRowCount()); + dailyFile.setCountryCodesJson(countryCodesJson(artifact.countryCodes())); + dailyFile.setCompensationDone(1); dailyFile.setLastSuccessAt(now); dailyFile.setUpdatedAt(now); if (dailyFile.getId() == null) { @@ -2179,60 +2266,6 @@ public class ShopDataCrawlTaskService { return template.execute(status -> action.get()); } - private List findShopResultRows(Long userId, FileResultEntity sourceRow) { - if (userId == null || sourceRow == null) { - return List.of(); - } - String shopId = trimToNull(sourceRow.getSourceFileUrl()); - String shopName = trimToNull(sourceRow.getSourceFilename()); - LambdaQueryWrapper wrapper = new LambdaQueryWrapper() - .eq(FileResultEntity::getModuleType, MODULE_TYPE) - .and(owner -> owner.eq(FileResultEntity::getUserId, userId) - .or().isNull(FileResultEntity::getUserId)); - if (shopId != null) { - wrapper.eq(FileResultEntity::getSourceFileUrl, shopId); - } else if (shopName != null) { - wrapper.eq(FileResultEntity::getSourceFilename, shopName) - .apply("TRIM(COALESCE(source_file_url, '')) = ''"); - } else { - return List.of(); - } - List candidates = fileResultMapper.selectList(wrapper); - if (candidates == null || candidates.isEmpty()) { - return List.of(); - } - Map legacyTaskOwners = loadTaskMapByIds(candidates.stream() - .filter(candidate -> candidate != null && candidate.getUserId() == null) - .map(FileResultEntity::getTaskId) - .filter(Objects::nonNull) - .distinct() - .toList()); - return candidates.stream() - .filter(Objects::nonNull) - .filter(candidate -> Objects.equals(userId, candidate.getUserId()) - || Objects.equals(userId, - legacyTaskOwners.get(candidate.getTaskId()) == null - ? null - : legacyTaskOwners.get(candidate.getTaskId()).getUserId())) - .toList(); - } - - private void clearShopResultPointers(List rows, Long keepResultId) { - if (rows == null) { - return; - } - for (FileResultEntity candidate : rows) { - if (candidate == null || Objects.equals(candidate.getId(), keepResultId) - || blank(candidate.getResultFileUrl())) { - continue; - } - candidate.setResultFileUrl(null); - candidate.setResultFileSize(null); - candidate.setResultContentType(null); - fileResultMapper.updateById(candidate); - } - } - private void collectObjectKey(Set target, String value) { if (target == null || blank(value)) { return; @@ -2286,19 +2319,18 @@ public class ShopDataCrawlTaskService { } } - private DailyLockSet acquireDailyLocks(Long fallbackUserId, List rows) { + private DailyLockSet acquireDailyLocks(List rows) { Map requests = new TreeMap<>(); if (rows != null) { for (FileResultEntity row : rows) { if (row == null) { continue; } - Long userId = row.getUserId() != null ? row.getUserId() : fallbackUserId; String shopKey = dailyFileService.shopKey(row); - if (userId == null || userId <= 0 || blank(shopKey)) { + if (blank(shopKey)) { continue; } - requests.putIfAbsent(userId + "|" + shopKey, new DailyLockRequest(userId, shopKey)); + requests.putIfAbsent(shopKey, new DailyLockRequest(shopKey)); } } if (requests.isEmpty()) { @@ -2307,8 +2339,7 @@ public class ShopDataCrawlTaskService { List handles = new ArrayList<>(); try { for (DailyLockRequest request : requests.values()) { - TaskDistributedLockService.LockHandle handle = dailyFileService.acquireLock( - request.userId(), request.shopKey()); + TaskDistributedLockService.LockHandle handle = dailyFileService.acquireLock(request.shopKey()); if (handle == null) { throw new BusinessException("店铺累计文件正在处理中，请稍后重试"); } @@ -2321,7 +2352,7 @@ public class ShopDataCrawlTaskService { } } - private record DailyLockRequest(Long userId, String shopKey) { + private record DailyLockRequest(String shopKey) { } private record DailyMemberData(ShopDataCrawlDailyMemberEntity member, @@ -2374,10 +2405,11 @@ public class ShopDataCrawlTaskService { long fileSize, boolean uploaded, String filename, - int rowCount) { + int rowCount, + List countryCodes) { - DailyWorkbookArtifact(String objectKey, long fileSize, boolean uploaded, String filename) { - this(objectKey, fileSize, uploaded, filename, 0); + DailyWorkbookArtifact(String objectKey, long fileSize, boolean uploaded, String filename, int rowCount) { + this(objectKey, fileSize, uploaded, filename, rowCount, List.of()); } } @@ -2443,6 +2475,133 @@ public class ShopDataCrawlTaskService { } } + /** + * V95 迁移后启动补偿：对 compensation_done=0 的每日累计文件， + * 按成员 result 快照重建一份合并工作簿（覆盖国家语义），上传新对象、 + * 更新 daily_file 与最新 result 行指针、清理不再被引用的旧对象。 + * 单文件失败仅记日志，不阻断其余文件与启动。 + */ + public void compensateDailyWorkbooksAfterMigration() { + List pending = dailyFileService.findForCompensation(); + if (pending.isEmpty()) { + return; + } + log.info("[shop-data-crawl] daily compensation start, pending={}", pending.size()); + int processed = 0; + for (ShopDataCrawlDailyFileEntity dailyFile : pending) { + if (dailyFile == null || dailyFile.getId() == null) { + continue; + } + try { + compensateOneDailyWorkbook(dailyFile); + processed++; + } catch (Exception ex) { + log.warn("[shop-data-crawl] daily compensation failed dailyFileId={} shopKey={} date={} msg={}", + dailyFile.getId(), dailyFile.getShopKey(), dailyFile.getBusinessDate(), safeMessage(ex)); + } + } + log.info("[shop-data-crawl] daily compensation done, processed={}/{}", processed, pending.size()); + } + + private void compensateOneDailyWorkbook(ShopDataCrawlDailyFileEntity dailyFile) { + List members = dailyFileService.listMembers(dailyFile.getId()); + List sorted = sortedDailyMembers(members); + if (sorted.isEmpty()) { + // 无成员行：标记完成，避免每次启动重试 + dailyFileService.markCompensationDone(dailyFile); + return; + } + List snapshots = new ArrayList<>(); + for (ShopDataCrawlDailyMemberEntity member : sorted) { + FileResultEntity result = fileResultMapper.selectById(member.getResultId()); + if (result == null) { + log.warn("[shop-data-crawl] compensation member result missing dailyFileId={} resultId={}", + dailyFile.getId(), member.getResultId()); + continue; + } + ShopDataCrawlResultItemVo snapshot = loadSnapshotForDailyMember(result); + if (snapshot == null) { + log.warn("[shop-data-crawl] compensation snapshot missing dailyFileId={} resultId={} skip", + dailyFile.getId(), member.getResultId()); + continue; + } + snapshots.add(snapshot); + } + if (snapshots.isEmpty()) { + dailyFileService.markCompensationDone(dailyFile); + return; + } + + String filename = blank(dailyFile.getResultFilename()) + ? buildTaskWorkbookFilename(loadCompensationTask(dailyFile)) + : dailyFile.getResultFilename(); + String oldObjectKey = dailyFile.getResultFileUrl(); + File workRoot = FileUtil.mkdir(FileUtil.file( + System.getProperty("java.io.tmpdir"), + "shop-data-crawl-result", + "compensation-" + dailyFile.getId())); + File outputXlsx = FileUtil.file(workRoot, filename); + try { + int rowCount = excelAssemblyService.writeWorkbook(outputXlsx, snapshots); + String newObjectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE); + if (blank(newObjectKey)) { + throw new BusinessException("累计文件补偿上传结果为空"); + } + LocalDateTime now = dailyFileService.currentBusinessDateTime(); + dailyFile.setLatestTaskId(dailyFile.getLatestTaskId()); + dailyFile.setResultFilename(filename); + dailyFile.setResultFileUrl(newObjectKey); + dailyFile.setResultFileSize(outputXlsx.length()); + dailyFile.setResultContentType(CONTENT_TYPE_XLSX); + dailyFile.setRowCount(rowCount); + dailyFile.setCountryCodesJson(countryCodesJson(collectCountryCodes(snapshots))); + dailyFile.setVersion(Math.max(0L, Objects.requireNonNullElse(dailyFile.getVersion(), 0L)) + 1L); + dailyFile.setCompensationDone(1); + dailyFile.setLastSuccessAt(now); + dailyFile.setUpdatedAt(now); + dailyFileService.update(dailyFile); + + FileResultEntity latestResult = fileResultMapper.selectById(dailyFile.getLatestResultId()); + if (latestResult != null) { + latestResult.setResultFilename(filename); + latestResult.setResultFileUrl(newObjectKey); + latestResult.setResultFileSize(outputXlsx.length()); + latestResult.setResultContentType(CONTENT_TYPE_XLSX); + latestResult.setRowCount(rowCount); + fileResultMapper.updateById(latestResult); + } + // 清理旧对象（引用计数：其他 result/daily_file 仍引用则保留） + if (!blank(oldObjectKey) && !Objects.equals(oldObjectKey, newObjectKey)) { + deleteResultObjectNowIfUnreferenced(oldObjectKey); + } + log.info("[shop-data-crawl] compensation rebuilt dailyFileId={} shopKey={} rows={} object={}", + dailyFile.getId(), dailyFile.getShopKey(), rowCount, newObjectKey); + } finally { + FileUtil.del(outputXlsx); + FileUtil.del(workRoot); + } + } + + private FileTaskEntity loadCompensationTask(ShopDataCrawlDailyFileEntity dailyFile) { + if (dailyFile.getLatestTaskId() != null) { + FileTaskEntity task = fileTaskMapper.selectById(dailyFile.getLatestTaskId()); + if (task != null) { + return task; + } + } + List members = dailyFileService.listMembers(dailyFile.getId()); + for (ShopDataCrawlDailyMemberEntity member : sortedDailyMembers(members)) { + FileTaskEntity task = fileTaskMapper.selectById(member.getTaskId()); + if (task != null) { + return task; + } + } + FileTaskEntity fallback = new FileTaskEntity(); + fallback.setId(0L); + fallback.setTaskNo(MODULE_TYPE + "-" + IdUtil.getSnowflakeNextIdStr()); + return fallback; + } + private boolean isTaskWorkbookPending(FileTaskEntity task, List rows) { if (task == null || task.getId() == null || rows == null || rows.isEmpty()) { return false; diff --git a/backend-java/src/main/resources/db/V95__shop_data_crawl_daily_file_shop_level.sql b/backend-java/src/main/resources/db/V95__shop_data_crawl_daily_file_shop_level.sql new file mode 100644 index 00000000..78a601c5 --- /dev/null +++ b/backend-java/src/main/resources/db/V95__shop_data_crawl_daily_file_shop_level.sql @@ -0,0 +1,70 @@ +-- V95: 店铺数据抓取累计文件改为店铺级共享（去 user 维度） +-- 问题：同店同日在不同账号（尾号5578江秀珍1~5）下产生多份互不相干的累计文件， +-- 管理页每店只显示最新一份，其他账号抓的国家（如英国）看起来"没了"。 +-- 目标：每店每天只保留一份累计文件，(shop_key, business_date) 唯一； +-- 聚合层按国家覆盖（见 ShopDataCrawlTaskService），未更新的国家保留。 +-- +-- 存量合并策略： +-- 1) 回填 shop_key（店铺名，来自 daily_member.result_id -> biz_file_result.source_filename） +-- 2) 同店同日多份 daily_file 合并为一个分组（保留各 result 的成员行）， +-- 由启动补偿组件按成员 result 快照重建一份累计文件 +-- 3) 唯一键改为 (shop_key, business_date) + +-- 1. 新增列（country_codes_json 供后台管理页展示累计文件实际包含的国家） +ALTER TABLE biz_shop_data_crawl_daily_file + ADD COLUMN `shop_key` VARCHAR(1000) NULL DEFAULT NULL COMMENT '店铺名（去 user 后的店铺级归属键）' AFTER `shop_key_hash`, + ADD COLUMN `country_codes_json` VARCHAR(512) NULL DEFAULT NULL COMMENT '累计文件实际包含的国家代码 JSON（后台管理页展示用）' AFTER `result_content_type`, + ADD COLUMN `compensation_done` TINYINT NOT NULL DEFAULT 0 COMMENT '0=启动补偿组件待重建，1=已按成员快照重建' AFTER `version`; + +-- 2. 回填 shop_key：daily_member 记录了每个结果行属于哪个 daily_file +UPDATE biz_shop_data_crawl_daily_file df +JOIN biz_shop_data_crawl_daily_member m ON m.daily_file_id = df.id +JOIN biz_file_result r ON r.id = m.result_id +SET df.shop_key = TRIM(r.source_filename) +WHERE df.shop_key IS NULL OR df.shop_key = ''; + +-- 3. 兜底：无成员的 daily_file 用 latest_result_id 回填 +UPDATE biz_shop_data_crawl_daily_file df +LEFT JOIN biz_shop_data_crawl_daily_member m ON m.daily_file_id = df.id +JOIN biz_file_result r ON r.id = df.latest_result_id +SET df.shop_key = TRIM(r.source_filename) +WHERE m.id IS NULL AND (df.shop_key IS NULL OR df.shop_key = ''); + +-- 4. 清理仍为空的（无任何可反解依据的历史残留） +DELETE df FROM biz_shop_data_crawl_daily_file df +WHERE df.shop_key IS NULL OR df.shop_key = ''; + +-- 5. 交换唯一键：先删旧 (user_id, shop_key_hash, business_date)，再加 (shop_key, business_date) +ALTER TABLE biz_shop_data_crawl_daily_file + DROP KEY uk_shop_data_crawl_daily_file; + +-- 6. 同店同日合并成员行归属：保留最新一份 daily_file，其余 daily_file 的成员行迁到它名下 +UPDATE biz_shop_data_crawl_daily_member m +JOIN biz_shop_data_crawl_daily_file cur ON cur.id = m.daily_file_id +JOIN ( + SELECT shop_key, business_date, MAX(id) AS max_id + FROM biz_shop_data_crawl_daily_file + GROUP BY shop_key, business_date + HAVING COUNT(*) > 1 +) g ON g.shop_key = cur.shop_key AND g.business_date = cur.business_date AND g.max_id != cur.id +SET m.daily_file_id = g.max_id; + +-- 7. 删除同店同日合并后的多余 daily_file 行（对象存储文件由补偿组件按引用计数清理） +DELETE df FROM biz_shop_data_crawl_daily_file df +WHERE df.id NOT IN ( + SELECT keep_id FROM ( + SELECT MAX(id) AS keep_id + FROM biz_shop_data_crawl_daily_file + GROUP BY shop_key, business_date + ) t +); + +-- 8. 唯一键（shop_key 已非空；兜底统一改为未命名） +UPDATE biz_shop_data_crawl_daily_file SET shop_key = '未命名' WHERE TRIM(shop_key) = ''; + +-- 9. country_codes_json 由启动补偿组件加载成员 result 快照后回填 +-- （biz_task_result_item.payload_json 是 rustfs 指针，SQL 无法解析内容） + +-- 10. 新唯一键 +ALTER TABLE biz_shop_data_crawl_daily_file + ADD UNIQUE KEY uk_shop_data_crawl_daily_file (shop_key, business_date); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java index 9028f6db..b197b515 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java @@ -278,11 +278,11 @@ class ShopDataCrawlCleanupTest { FileResultEntity row = invocation.getArgument(0); return row == null ? null : row.getSourceFilename(); }); - lenient().when(dailyFileService.acquireLock(anyLong(), anyString())) + lenient().when(dailyFileService.acquireLock(anyString())) .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); - lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())).thenAnswer(invocation -> - copyDailyFile(findDailyFile(invocation.getArgument(0), invocation.getArgument(1)))); - lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findForUpdate(anyString(), any())).thenAnswer(invocation -> + copyDailyFile(findDailyFile(invocation.getArgument(0)))); + lenient().when(dailyFileService.findOlder(anyString(), any())).thenReturn(List.of()); lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); lenient().when(dailyFileService.countObjectReferences(anyString())).thenAnswer(invocation -> { @@ -423,7 +423,7 @@ class ShopDataCrawlCleanupTest { taskStore.put(1L, task); lenient().when(dailyFileService.findByLatestResultId(8102L)) - .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + .thenReturn(List.of(copyDailyFile(findDailyFile("hash:" + SHOP_NAME)))); String newObjectKey = "oss/daily/rebuilt.xlsx"; lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn(newObjectKey); doAnswer(invocation -> { @@ -459,7 +459,7 @@ class ShopDataCrawlCleanupTest { taskStore.put(1L, task); lenient().when(dailyFileService.findByLatestResultId(8105L)) - .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + .thenReturn(List.of(copyDailyFile(findDailyFile("hash:" + SHOP_NAME)))); lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8103L, ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(8103L)); lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8104L, @@ -531,7 +531,7 @@ class ShopDataCrawlCleanupTest { FileTaskEntity task = taskEntity(1L, "SUCCESS"); taskStore.put(1L, task); lenient().when(dailyFileService.findByLatestResultId(8107L)) - .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + .thenReturn(List.of(copyDailyFile(findDailyFile("hash:" + SHOP_NAME)))); service.deleteHistory(8107L, USER_ID); @@ -588,7 +588,7 @@ class ShopDataCrawlCleanupTest { FileTaskEntity task = taskEntity(1L, "SUCCESS"); taskStore.put(1L, task); lenient().when(dailyFileService.findByLatestResultId(resultIds[11])) - .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + .thenReturn(List.of(copyDailyFile(findDailyFile("hash:" + SHOP_NAME)))); for (long resultId : resultIds) { service.deleteHistory(resultId, USER_ID); @@ -652,7 +652,7 @@ class ShopDataCrawlCleanupTest { FileTaskEntity task = taskEntity(1L, "SUCCESS"); taskStore.put(1L, task); lenient().when(dailyFileService.findByLatestResultId(8112L)) - .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + .thenReturn(List.of(copyDailyFile(findDailyFile("hash:" + SHOP_NAME)))); lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8111L, ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(8111L)); doAnswer(invocation -> { @@ -824,15 +824,14 @@ class ShopDataCrawlCleanupTest { lastJobTaskId = 2L; TaskFileJobEntity job2 = jobEntity(2L, 2L, MODULE_TYPE, 8402L); AtomicLong calls = new AtomicLong(); - lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())).thenAnswer(invocation -> { - Long userId = invocation.getArgument(0); - String shopKeyHash = invocation.getArgument(1); + lenient().when(dailyFileService.findForUpdate(anyString(), any())).thenAnswer(invocation -> { + String shopKeyHash = invocation.getArgument(0); if (calls.incrementAndGet() == 2L) { - ShopDataCrawlDailyFileEntity conflicting = copyDailyFile(findDailyFile(userId, shopKeyHash)); + ShopDataCrawlDailyFileEntity conflicting = copyDailyFile(findDailyFile(shopKeyHash)); conflicting.setVersion(99L); return conflicting; } - return copyDailyFile(findDailyFile(userId, shopKeyHash)); + return copyDailyFile(findDailyFile(shopKeyHash)); }); service.processResultFileJob(job2); @@ -1171,9 +1170,9 @@ class ShopDataCrawlCleanupTest { return DigestUtil.sha256Hex("result-chunks:" + SHOP_NAME); } - private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + private ShopDataCrawlDailyFileEntity findDailyFile(String shopKeyHash) { for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { - if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + if (Objects.equals(f.getShopKeyHash(), shopKeyHash)) { return f; } } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java index 0ea85f9f..acdaa8e6 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java @@ -194,7 +194,7 @@ class ShopDataCrawlDailyFileIncrementalTest { }); // 店铺级锁：每次返回独立 mock 句柄，供失败路径验证 close()。 - lenient().when(dailyFileService.acquireLock(anyLong(), anyString())).thenAnswer(invocation -> { + lenient().when(dailyFileService.acquireLock(anyString())).thenAnswer(invocation -> { TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); lastLock.set(handle); return handle; @@ -209,9 +209,9 @@ class ShopDataCrawlDailyFileIncrementalTest { FileResultEntity row = invocation.getArgument(0); return row == null ? null : row.getSourceFilename(); }); - lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())) - .thenAnswer(invocation -> findDailyFile(invocation.getArgument(0), invocation.getArgument(1))); - lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findForUpdate(anyString(), any())) + .thenAnswer(invocation -> findDailyFile(invocation.getArgument(0))); + lenient().when(dailyFileService.findOlder(anyString(), any())).thenReturn(List.of()); lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); @@ -482,8 +482,7 @@ class ShopDataCrawlDailyFileIncrementalTest { ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); for (int i = 0; i < dbDailyFiles.size(); i++) { ShopDataCrawlDailyFileEntity existing = dbDailyFiles.get(i); - if (Objects.equals(existing.getUserId(), entity.getUserId()) - && Objects.equals(existing.getShopKeyHash(), entity.getShopKeyHash()) + if (Objects.equals(existing.getShopKeyHash(), entity.getShopKeyHash()) && Objects.equals(existing.getBusinessDate(), entity.getBusinessDate())) { entity.setId(existing.getId()); dbDailyFiles.set(i, entity); @@ -517,9 +516,9 @@ class ShopDataCrawlDailyFileIncrementalTest { }); } - private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + private ShopDataCrawlDailyFileEntity findDailyFile(String shopKeyHash) { for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { - if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + if (Objects.equals(f.getShopKeyHash(), shopKeyHash)) { return f; } } @@ -535,4 +534,57 @@ class ShopDataCrawlDailyFileIncrementalTest { FileResultEntity row = addResultRow(900L, 900L, 1, SHOP_NAME, null); processJob(900L, List.of(row), snapshot(900L, SHOP_NAME, 1)); } + + @Test + void test_task_095_shop_level_country_coverage_keeps_unupdated_countries() { + // V95 店铺级共享：跨账号同店结果汇聚到同一份累计文件， + // 同国家覆盖更新，未更新的国家保留。 + // 账号A（user 7）先抓英国；账号B（user 8）抓德国；账号A再抓英国（覆盖）。 + FileResultEntity uk1 = addResultRow(9501L, 9501L, 1, SHOP_NAME, null); + processJob(9501L, List.of(uk1), snapshotWithCountry(9501L, "UK", 100)); + + FileResultEntity de1 = addResultRow(9502L, 9502L, 1, SHOP_NAME, null); + de1.setUserId(8L); // 不同账号 + processJob(9502L, List.of(de1), snapshotWithCountry(9502L, "DE", 200)); + + FileResultEntity uk2 = addResultRow(9503L, 9503L, 1, SHOP_NAME, null); + processJob(9503L, List.of(uk2), snapshotWithCountry(9503L, "UK", 150)); + + assertEquals(1, dbDailyFiles.size(), "跨账号同店同日只保留一份累计文件"); + assertEquals(3, dbMembers.size(), "三个结果各有一个成员行"); + + // 整表重建：英国以最后一次（9503）为准，德国（9502）保留 + triggerRebuild(); + assertEquals(4, lastAssembledItems.size(), "成员快照按创建顺序累积，item 结构保持"); + Map countryRows = new java.util.HashMap<>(); + for (ShopDataCrawlResultItemVo item : lastAssembledItems) { + if (item.getCountryResults() == null) continue; + for (com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto cr + : item.getCountryResults()) { + countryRows.put(cr.getCountry(), cr.getItems() == null ? 0 : cr.getItems().size()); + } + } + assertEquals(150, countryRows.getOrDefault("UK", -1), "英国以最后一次抓取为准（覆盖更新）"); + assertEquals(200, countryRows.getOrDefault("DE", -1), "德国未更新，保留之前抓取的数据"); + verify(ossStorageService, never()).readObjectBytes(anyString()); + } + + private ShopDataCrawlResultItemVo snapshotWithCountry(long resultId, String country, int rows) { + ShopDataCrawlResultItemVo item = snapshot(resultId, SHOP_NAME, rows); + com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto countryResult = + new com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto(); + countryResult.setCountry(country); + List rowList = new ArrayList<>(); + for (int i = 0; i < rows; i++) { + com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto row = + new com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto(); + row.setAsin("B0" + String.format("%08d", i)); + row.setDate("2026-07-25"); + rowList.add(row); + } + countryResult.setItems(rowList); + item.setCountryResults(List.of(countryResult)); + item.setCountryCodes(List.of(country)); + return item; + } } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java index e4eacc94..6e5f30b7 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java @@ -221,11 +221,11 @@ class ShopDataCrawlDailyFileJobSplitTest { FileResultEntity row = invocation.getArgument(0); return row == null ? null : row.getSourceFilename(); }); - lenient().when(dailyFileService.acquireLock(anyLong(), anyString())) + lenient().when(dailyFileService.acquireLock(anyString())) .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); - lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())).thenAnswer(invocation -> - copyDailyFile(findDailyFile(invocation.getArgument(0), invocation.getArgument(1)))); - lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findForUpdate(anyString(), any())).thenAnswer(invocation -> + copyDailyFile(findDailyFile(invocation.getArgument(0)))); + lenient().when(dailyFileService.findOlder(anyString(), any())).thenReturn(List.of()); lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); @@ -622,9 +622,9 @@ class ShopDataCrawlDailyFileJobSplitTest { }); } - private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + private ShopDataCrawlDailyFileEntity findDailyFile(String shopKeyHash) { for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { - if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + if (Objects.equals(f.getShopKeyHash(), shopKeyHash)) { return f; } } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java index e7b98f4c..c8626f18 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java @@ -200,7 +200,7 @@ class ShopDataCrawlDailyFileLockTest { }).when(excelAssemblyService).writeWorkbook(any(), any()); // 店铺级锁：每次获取返回独立句柄并计数（两次短临界区各取一次）。 - lenient().when(dailyFileService.acquireLock(anyLong(), anyString())).thenAnswer(invocation -> { + lenient().when(dailyFileService.acquireLock(anyString())).thenAnswer(invocation -> { TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); lockAcquireCount.incrementAndGet(); lastLock.set(handle); @@ -217,10 +217,9 @@ class ShopDataCrawlDailyFileLockTest { FileResultEntity row = invocation.getArgument(0); return row == null ? null : row.getSourceFilename(); }); - lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())) - .thenAnswer(invocation -> copyDailyFile(findDailyFile( - invocation.getArgument(0), invocation.getArgument(1)))); - lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findForUpdate(anyString(), any())) + .thenAnswer(invocation -> copyDailyFile(findDailyFile(invocation.getArgument(0)))); + lenient().when(dailyFileService.findOlder(anyString(), any())).thenReturn(List.of()); lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); @@ -345,14 +344,14 @@ class ShopDataCrawlDailyFileLockTest { seedDailyFile(); FileResultEntity row = addResultRow(7207L, 1L, 1, SHOP_NAME, null); lenient().doAnswer(invocation -> { - ShopDataCrawlDailyFileEntity current = findDailyFile(invocation.getArgument(0), invocation.getArgument(1)); + ShopDataCrawlDailyFileEntity current = findDailyFile(invocation.getArgument(0)); if (current == null) { return null; } ShopDataCrawlDailyFileEntity readCopy = copyDailyFile(current); current.setVersion(readCopy.getVersion() + 1L); return readCopy; - }).when(dailyFileService).findForUpdate(anyLong(), anyString(), any()); + }).when(dailyFileService).findForUpdate(anyString(), any()); Exception ex = assertThrows(BusinessException.class, () -> processJob(1L, List.of(row), snapshot(7207L))); @@ -566,9 +565,9 @@ class ShopDataCrawlDailyFileLockTest { }); } - private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + private ShopDataCrawlDailyFileEntity findDailyFile(String shopKeyHash) { for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { - if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + if (Objects.equals(f.getShopKeyHash(), shopKeyHash)) { return f; } } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java index bea9bd54..06af59b6 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java @@ -127,10 +127,10 @@ class ShopDataCrawlTaskServiceRetentionTest { when(dailyFileService.currentBusinessDateTime()).thenReturn(BUSINESS_TIME); when(dailyFileService.shopKey(any())).thenReturn("shop-id:shop-1"); when(dailyFileService.shopKeyHash(anyString())).thenReturn("hash-1"); - when(dailyFileService.acquireLock(eq(USER_ID), eq("shop-id:shop-1"))) + when(dailyFileService.acquireLock("shop-id:shop-1")) .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of()); - when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of()); + when(dailyFileService.findOlder("hash-1", BUSINESS_DATE)).thenReturn(List.of()); when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); when(dailyFileService.addMemberWithPayload(anyLong(), anyLong(), anyLong(), anyString())).thenReturn(true); when(transactionManager.getTransaction(any())).thenReturn(transactionStatus); @@ -144,7 +144,7 @@ class ShopDataCrawlTaskServiceRetentionTest { @Test void firstSuccessCreatesDailyWorkbookAndMembership() { when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(null); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(1); @@ -176,7 +176,7 @@ class ShopDataCrawlTaskServiceRetentionTest { + "\"shopId\":\"shop-1\",\"success\":true,\"countryResults\":[]}"); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(previous, currentRow)); when(fileTaskMapper.selectBatchIds(List.of(100L))).thenReturn(List.of(previousTask)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); when(dailyFileService.listMembers(301L)).thenReturn(List.of(previousMember)); when(taskResultItemService.getResultSnapshot( @@ -192,7 +192,8 @@ class ShopDataCrawlTaskServiceRetentionTest { verify(ossStorageService, never()).readObjectBytes(anyString()); verify(dailyFileService).update(daily); verify(ossStorageService).deleteObject("result/old.xlsx"); - assertNull(previous.getResultFileUrl()); + // 店铺级共享：历史结果行保留旧对象指针（可继续下载），累计文件本体被新对象取代 + assertEquals("result/old.xlsx", previous.getResultFileUrl()); assertEquals("result/new.xlsx", currentRow.getResultFileUrl()); assertEquals(2, currentRow.getRowCount()); } @@ -215,7 +216,7 @@ class ShopDataCrawlTaskServiceRetentionTest { ShopDataCrawlDailyFileEntity daily = daily("result/old.xlsx", 2); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); doAnswer(invocation -> { assertFalse(transactionActive.get(), "row counting must run outside the database transaction"); @@ -248,7 +249,7 @@ class ShopDataCrawlTaskServiceRetentionTest { ShopDataCrawlDailyFileEntity daily = daily("result/current.xlsx", 3); daily.setLatestResultId(RESULT_ID); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(true); service.processResultFileJob(job); @@ -264,7 +265,7 @@ class ShopDataCrawlTaskServiceRetentionTest { void zeroNewRowsReuseDailyObjectWithoutWorkbookIo() { ShopDataCrawlDailyFileEntity daily = daily("result/current.xlsx", 3); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); when(excelAssemblyService.countRows(any())).thenReturn(0); @@ -291,7 +292,7 @@ class ShopDataCrawlTaskServiceRetentionTest { when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(previous, currentRow)); when(fileTaskMapper.selectBatchIds(List.of(100L))).thenReturn(List.of(previousTask)); when(fileResultMapper.selectCount(any())).thenReturn(1L); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); when(dailyFileService.listMembers(301L)).thenReturn(List.of()); when(taskResultItemService.getResultSnapshot( @@ -314,7 +315,7 @@ class ShopDataCrawlTaskServiceRetentionTest { yesterday.setLatestResultId(RESULT_ID); ShopDataCrawlDailyMemberEntity member = member(301L, TASK_ID, RESULT_ID, BUSINESS_TIME.minusDays(1)); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(today); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(today); when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of(member)); when(dailyFileService.findById(301L)).thenReturn(yesterday); @@ -333,8 +334,8 @@ class ShopDataCrawlTaskServiceRetentionTest { yesterday.setId(300L); yesterday.setBusinessDate(BUSINESS_DATE.minusDays(1)); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(previous, currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); - when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(null); + when(dailyFileService.findOlder("hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); when(dailyFileService.listMembers(300L)).thenReturn(List.of()); when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(1); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/today.xlsx"); @@ -350,7 +351,8 @@ class ShopDataCrawlTaskServiceRetentionTest { verify(ossStorageService).deleteObject("result/yesterday.xlsx"); assertEquals("result/today.xlsx", currentRow.getResultFileUrl()); assertEquals(1, currentRow.getRowCount()); - assertNull(previous.getResultFileUrl()); + // 店铺级共享：旧日历史结果行保留旧对象指针（可下载），仅累计文件行被删除 + assertEquals("result/yesterday.xlsx", previous.getResultFileUrl()); } @Test @@ -359,8 +361,8 @@ class ShopDataCrawlTaskServiceRetentionTest { yesterday.setId(300L); yesterday.setBusinessDate(BUSINESS_DATE.minusDays(1)); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); - when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(null); + when(dailyFileService.findOlder("hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); when(dailyFileService.listMembers(300L)).thenReturn(List.of()); when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(4); doThrow(new IllegalStateException("upload failed")) @@ -378,8 +380,8 @@ class ShopDataCrawlTaskServiceRetentionTest { yesterday.setId(300L); yesterday.setBusinessDate(BUSINESS_DATE.minusDays(1)); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); - when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(null); + when(dailyFileService.findOlder("hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); when(dailyFileService.listMembers(300L)).thenReturn(List.of()); when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(4); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); @@ -418,7 +420,7 @@ class ShopDataCrawlTaskServiceRetentionTest { when(dailyFileService.findByLatestResultId(RESULT_ID)).thenReturn(List.of(daily)); when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of(removedMember)); when(dailyFileService.findById(301L)).thenReturn(daily); - when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.findForUpdate("hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.listMembers(301L)).thenReturn(List.of(removedMember, previousMember)); when(taskResultItemService.getResultSnapshot( 100L, MODULE_TYPE, 200L, ShopDataCrawlResultItemVo.class)).thenReturn(previousSnapshot); diff --git a/backend/blueprints/admin_api.py b/backend/blueprints/admin_api.py index 761704d5..274c9aaa 100644 --- a/backend/blueprints/admin_api.py +++ b/backend/blueprints/admin_api.py @@ -1362,6 +1362,7 @@ _SHOP_DATA_CRAWL_ADMIN_COLUMNS = f""" t.task_no, t.status AS task_status, t.request_json, t.result_json, t.error_message AS task_error, t.created_at, t.updated_at, t.finished_at, {_SHOP_DATA_CRAWL_LATEST_TIME_SQL} AS latest_file_updated_at, + df.country_codes_json, COALESCE(df.row_count, r.row_count) AS row_count_display, u.username """ @@ -1378,6 +1379,14 @@ def _shop_data_crawl_country_codes(request_json): return [str(value).strip().upper() for value in raw if str(value or '').strip()] +def _shop_data_crawl_country_codes_from_json(json_value): + """从 daily_file.country_codes_json 解析国家代码列表（店铺累计文件实际包含的国家）。""" + value = _parse_json_value(json_value, None) + if isinstance(value, list): + return [str(item).strip().upper() for item in value if str(item or '').strip()] + return [] + + def _shop_data_crawl_group_names(cursor, rows): shop_names = sorted({ _shop_data_crawl_shop_key(row.get('shop_name')) @@ -1437,7 +1446,8 @@ def _shop_data_crawl_admin_item(row, group_names=None): 'status': row.get('task_status') or '', 'success': success, 'error': row.get('result_error') or row.get('task_error') or row.get('file_error') or '', - 'country_codes': _shop_data_crawl_country_codes(row.get('request_json')), + 'country_codes': _shop_data_crawl_country_codes_from_json(row.get('country_codes_json')) + or _shop_data_crawl_country_codes(row.get('request_json')), 'output_filename': row.get('result_filename') or '', 'result_file_url': row.get('result_file_url') or '', 'file_ready': file_ready, @@ -1445,7 +1455,7 @@ def _shop_data_crawl_admin_item(row, group_names=None): 'file_status': file_status, 'file_error': row.get('file_error') or '', 'file_size': int(row.get('result_file_size') or 0), - 'row_count': int(row.get('row_count') or 0), + 'row_count': int(row.get('row_count_display') or row.get('row_count') or 0), 'created_at': _format_admin_datetime(row.get('created_at') or row.get('result_created_at')), 'updated_at': _format_admin_datetime(row.get('updated_at')), 'finished_at': _format_admin_datetime(row.get('finished_at')),