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

店铺数据抓取累计文件改为店铺级共享，按国家覆盖更新

Browse Source 
同店同日在不同账号下产生多份互不相干的累计文件（唯一键含 user_id），
后台管理页每店只显示最新一份，其他账号抓的国家看起来丢失。

- V95 迁移：加 shop_key/country_codes_json/compensation_done 列，按店铺
  合并存量 daily_file 与成员行，唯一键改为 (shop_key, business_date)
- DailyFileService：findForUpdate/findOlder/acquireLock 去 user 维度，
  店铺级锁跨账号串行
- TaskService：聚合按店铺定位；applyCountryCoverage 按国家覆盖（本次
  回传的国家替换旧行，未更新的国家保留）；启动补偿组件按成员快照重建
  合并文件并回填国家列表
- 管理页：country_codes 改从 daily_file.country_codes_json 取，行数用
  累计文件实际值
This commit is contained in:
黄自达
2026-08-30 16:30:09 +08:00
parent ae95b294ec
commit 97ae8a7c51
11 changed files with 495 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyFileEntity.java
+2
View File
@@ -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;
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyCompensationRunner.java
+29
View File
@@ -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);
}
}
}
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java
+24 -11
View File
@@ -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<ShopDataCrawlDailyFileEntity>()
.eq(ShopDataCrawlDailyFileEntity::getUserId, userId)
.eq(ShopDataCrawlDailyFileEntity::getShopKeyHash, shopKeyHash)
.eq(ShopDataCrawlDailyFileEntity::getBusinessDate, businessDate)
.last("FOR UPDATE"));
}
public List<ShopDataCrawlDailyFileEntity> findOlder(Long userId, String shopKeyHash, LocalDate businessDate) {
if (userId == null || shopKeyHash == null || businessDate == null) {
public List<ShopDataCrawlDailyFileEntity> findForCompensation() {
return dailyFileMapper.selectList(new LambdaQueryWrapper<ShopDataCrawlDailyFileEntity>()
.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<ShopDataCrawlDailyFileEntity> findOlder(String shopKeyHash, LocalDate businessDate) {
if (shopKeyHash == null || businessDate == null) {
return List.of();
}
return dailyFileMapper.selectList(new LambdaQueryWrapper<ShopDataCrawlDailyFileEntity>()
.eq(ShopDataCrawlDailyFileEntity::getUserId, userId)
.eq(ShopDataCrawlDailyFileEntity::getShopKeyHash, shopKeyHash)
.lt(ShopDataCrawlDailyFileEntity::getBusinessDate, businessDate)
.orderByDesc(ShopDataCrawlDailyFileEntity::getBusinessDate)
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java
+249 -90
View File
@@ -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<FileResultEntity> taskRows = listTaskRows(taskId);
try (DailyLockSet dailyLocks = acquireDailyLocks(task.getUserId(), taskRows)) {
try (DailyLockSet dailyLocks = acquireDailyLocks(taskRows)) {
ensureDailySyncCompletedBeforeDelete(taskRows);
Set<Long> 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<ShopDataCrawlDailyFileEntity> older = dailyFileService.findOlder(userId, shopKeyHash, businessDate);
List<ShopDataCrawlDailyFileEntity> 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<String> 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<ShopDataCrawlResultItemVo> buildDailyFileFromData(ShopDataCrawlDailyFileEntity baseDailyFile,
List<ShopDataCrawlResultItemVo> appended) {
@@ -1967,7 +1970,93 @@ public class ShopDataCrawlTaskService {
accumulated.add(item);
}
}
return accumulated;
return applyCountryCoverage(accumulated);
}
/**
* 按国家覆盖：从后往前记录每个国家最后一次出现的成员下标；每个成员只保留
* "该国最后一次出现" 的国家结果（非该次出现的行被剔除）。空结果国家同样覆盖旧行。
* item 结构（resultId/shopName 等元数据）与顺序保持稳定。
*/
private List<ShopDataCrawlResultItemVo> applyCountryCoverage(List<ShopDataCrawlResultItemVo> items) {
if (items == null || items.isEmpty()) {
return items == null ? List.of() : items;
}
Map<String, Integer> 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.<ShopDataCrawlCountryResultDto>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<ShopDataCrawlCountryResultDto> 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<String> collectCountryCodes(List<ShopDataCrawlResultItemVo> items) {
LinkedHashSet<String> codes = new LinkedHashSet<>();
if (items != null) {
for (ShopDataCrawlResultItemVo item : items) {
if (item == null) {
continue;
}
for (ShopDataCrawlCountryResultDto countryResult : item.getCountryResults() == null
? List.<ShopDataCrawlCountryResultDto>of() : item.getCountryResults()) {
if (countryResult != null && !blank(countryResult.getCountry())) {
codes.add(countryResult.getCountry().trim().toUpperCase());
}
}
}
}
return new ArrayList<>(codes);
}
private String countryCodesJson(List<String> countryCodes) {
if (countryCodes == null || countryCodes.isEmpty()) {
return "[]";
}
try {
return objectMapper.writeValueAsString(countryCodes);
} catch (Exception ex) {
return "[]";
}
}
private List<String> parseCountryCodesJson(String json) {
if (blank(json)) {
return List.of();
}
try {
List<String> parsed = objectMapper.readValue(json, new TypeReference<List<String>>() {
});
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<ShopDataCrawlDailyFileEntity> olderFiles = dailyFileService.findOlder(
userId, shopKeyHash, businessDate);
shopKeyHash, businessDate);
Set<String> 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<FileResultEntity> 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<FileResultEntity> findShopResultRows(Long userId, FileResultEntity sourceRow) {
if (userId == null || sourceRow == null) {
return List.of();
}
String shopId = trimToNull(sourceRow.getSourceFileUrl());
String shopName = trimToNull(sourceRow.getSourceFilename());
LambdaQueryWrapper<FileResultEntity> wrapper = new LambdaQueryWrapper<FileResultEntity>()
.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<FileResultEntity> candidates = fileResultMapper.selectList(wrapper);
if (candidates == null || candidates.isEmpty()) {
return List.of();
}
Map<Long, FileTaskEntity> 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<FileResultEntity> 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<String> target, String value) {
if (target == null || blank(value)) {
return;
@@ -2286,19 +2319,18 @@ public class ShopDataCrawlTaskService {
}
}
private DailyLockSet acquireDailyLocks(Long fallbackUserId, List<FileResultEntity> rows) {
private DailyLockSet acquireDailyLocks(List<FileResultEntity> rows) {
Map<String, DailyLockRequest> 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<TaskDistributedLockService.LockHandle> 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<String> 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<ShopDataCrawlDailyFileEntity> 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<ShopDataCrawlDailyMemberEntity> members = dailyFileService.listMembers(dailyFile.getId());
List<ShopDataCrawlDailyMemberEntity> sorted = sortedDailyMembers(members);
if (sorted.isEmpty()) {
// 无成员行：标记完成，避免每次启动重试
dailyFileService.markCompensationDone(dailyFile);
return;
}
List<ShopDataCrawlResultItemVo> 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<ShopDataCrawlDailyMemberEntity> 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<FileResultEntity> rows) {
if (task == null || task.getId() == null || rows == null || rows.isEmpty()) {
return false;
backend-java/src/main/resources/db/V95__shop_data_crawl_daily_file_shop_level.sql
+70
View File
@@ -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);
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java
+15 -16
View File
@@ -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;
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java
+60 -8
View File
@@ -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<String, Integer> 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<com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto> 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;
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java
+6 -6
View File
@@ -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;
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java
+8 -9
View File
@@ -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;
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java
+20 -18
View File
@@ -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);