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

task-37

Browse Source
This commit is contained in:
黄自达
2026-08-29 22:36:11 +08:00
parent 6f6d7be344
commit 95e314cef4
3 changed files with 732 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java
+22 -1
View File
@@ -1571,7 +1571,8 @@ public class ShopDataCrawlTaskService {
return;
}
snapshot.setShopName(firstNonBlank(payload.getShopName(), snapshot.getShopName()));
snapshot.setCountryResults(copyCountryResults(payload.getCountryResults()));
// 按国家增量合并：本次未回传的国家保留快照中的旧结果，同一国家按行去重合并
snapshot.setCountryResults(mergeCountryResults(snapshot.getCountryResults(), payload.getCountryResults()));
if (!blank(payload.getError())) {
snapshot.setError(payload.getError().trim());
}
@@ -2327,6 +2328,26 @@ public class ShopDataCrawlTaskService {
return;
}
deleteTransientResultChunks(job.getTaskId());
// Task 37：异步文件作业成功钩子——组装作业完成后显式刷新任务状态：
// 仍有未完成组装作业则保持 RUNNING，全部完成才进入终态。
// （worker 在 markSuccess 后、任务锁内调用本方法，此处刷新与提交是安全的。）
if ("SUCCESS".equals(job.getStatus())) {
refreshTaskStatusAfterAssembleJob(job.getTaskId());
}
}
private void refreshTaskStatusAfterAssembleJob(Long taskId) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return;
}
List<FileResultEntity> rows = listTaskRows(taskId);
if (rows.isEmpty()) {
return;
}
updateTaskStatusFromRows(task, rows);
fileTaskMapper.updateById(task);
taskCacheService.deleteTaskCache(taskId);
}
private void deleteTransientResultChunks(Long taskId) {
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java
+680
View File
@@ -0,0 +1,680 @@
package com.nanri.aiimage.modules.shopdatacrawl.service;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.config.TaskPressureProperties;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyFileEntity;
import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyMemberEntity;
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
import com.nanri.aiimage.modules.task.service.TaskResultItemService;
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.ibatis.session.Configuration;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Task 37：拆分每日累计文件组装与任务结果接收，增加异步文件作业状态。
* 结果接收（submitResult/tryFinalizeTask → finalizeTaskWorkbook）只做“结果落库 +
* 入队文件作业”，不做任何 Excel 组装/OSS 上传；组装完全由异步文件作业
* （processResultFileJob）承担；文件作业状态显式反映到任务状态
* （有未完成作业 → 任务保持 RUNNING，作业成功后才进入终态）。
*/
@ExtendWith(MockitoExtension.class)
class ShopDataCrawlDailyFileJobSplitTest {
private static final String MODULE_TYPE = "SHOP_DATA_CRAWL";
private static final String SHOP_NAME = "shop-a";
private static final LocalDate BUSINESS_DATE = LocalDate.of(2026, 8, 29);
private static final LocalDateTime BUSINESS_TIME = LocalDateTime.of(2026, 8, 29, 12, 0);
@BeforeAll
static void initializeMybatisMetadata() {
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), "");
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyFileEntity.class);
TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyMemberEntity.class);
TableInfoHelper.initTableInfo(assistant, TaskFileJobEntity.class);
}
@Mock private FileTaskMapper fileTaskMapper;
@Mock private FileResultMapper fileResultMapper;
@Mock private ShopDataCrawlResolveService shopDataCrawlResolveService;
@Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService;
@Mock private ShopDataCrawlTaskCacheService taskCacheService;
@Mock private OssStorageService ossStorageService;
@Mock private ZiniaoShopSwitchService ziniaoShopSwitchService;
@Mock private TaskPressureProperties taskPressureProperties;
@Mock private TaskFileJobService taskFileJobService;
@Mock private TaskResultItemService taskResultItemService;
@Mock private TaskProgressSnapshotService taskProgressSnapshotService;
@Mock private TaskDistributedLockService taskDistributedLockService;
@Mock private TaskChunkMapper taskChunkMapper;
@Mock private TaskScopeStateMapper taskScopeStateMapper;
@Mock private TransientPayloadStorageService transientPayloadStorageService;
@Mock private InstanceMetadata instanceMetadata;
@Mock private ShopDataCrawlDailyFileService dailyFileService;
@Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules();
private ShopDataCrawlTaskService service;
private final List<FileResultEntity> dbResultRows = new ArrayList<>();
private final List<ShopDataCrawlDailyFileEntity> dbDailyFiles = new ArrayList<>();
private final List<ShopDataCrawlDailyMemberEntity> dbMembers = new ArrayList<>();
/** 内存 file_job 表：enqueue/查询/状态流转的仿真。 */
private final List<TaskFileJobEntity> dbFileJobs = new ArrayList<>();
private final AtomicLong memberIdSeq = new AtomicLong(1000);
private final AtomicLong fileIdSeq = new AtomicLong(5000);
private final AtomicLong jobIdSeq = new AtomicLong(7000);
private long nextResultId = 8100;
private Long lastJobTaskId;
private String lastUploadedObjectKey;
private List<String> enqueuedScopes = new ArrayList<>();
/** 接收/组装路径每次落库后的任务状态（taskId → status），验证拆分后任务状态流转。 */
private final java.util.Map<Long, String> lastTaskStatus = new java.util.HashMap<>();
/** 接收/组装路径每次落库后的任务实体（taskId → 实体），供后续阶段读取。 */
private final java.util.Map<Long, FileTaskEntity> taskStore = new java.util.HashMap<>();
@BeforeEach
void configureStorage() {
service = new ShopDataCrawlTaskService(
fileTaskMapper,
fileResultMapper,
shopDataCrawlResolveService,
excelAssemblyService,
taskCacheService,
ossStorageService,
ziniaoShopSwitchService,
objectMapper,
taskPressureProperties,
taskFileJobService,
taskResultItemService,
taskProgressSnapshotService,
taskDistributedLockService,
taskChunkMapper,
taskScopeStateMapper,
transientPayloadStorageService,
instanceMetadata,
dailyFileService,
null);
dbResultRows.clear();
dbDailyFiles.clear();
dbMembers.clear();
dbFileJobs.clear();
memberIdSeq.set(1000);
fileIdSeq.set(5000);
jobIdSeq.set(7000);
nextResultId = 8100;
lastJobTaskId = null;
lastUploadedObjectKey = null;
enqueuedScopes = new ArrayList<>();
lastTaskStatus.clear();
taskStore.clear();
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong()))
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of());
lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of());
lenient().when(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of());
lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any()))
.thenReturn(List.of());
lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))).thenReturn(0L);
lenient().doNothing().when(taskResultItemService)
.replaceTaskSnapshots(anyLong(), eq(MODULE_TYPE), any(), any());
lenient().doNothing().when(taskProgressSnapshotService)
.save(anyLong(), any(), any(), anyInt(), anyInt(), anyInt(), any(), any(), any());
lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> {
String value = invocation.getArgument(0);
return value == null ? "" : value.trim();
});
lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> {
long taskId = invocation.getArgument(0);
FileTaskEntity stored = taskStore.get(taskId);
return stored == null ? null : stored;
});
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenAnswer(invocation -> {
FileTaskEntity entity = invocation.getArgument(0);
FileTaskEntity copy = copyTask(entity);
taskStore.put(copy.getId(), copy);
return 1;
});
lenient().when(fileResultMapper.selectBatchIds(any())).thenReturn(List.of());
lenient().when(fileResultMapper.selectCount(any())).thenReturn(0L);
lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenAnswer(invocation -> {
String key = "oss/split/" + System.nanoTime() + ".xlsx";
lastUploadedObjectKey = key;
return key;
});
lenient().doNothing().when(ossStorageService).deleteObject(anyString());
lenient().when(excelAssemblyService.countRows(any())).thenAnswer(invocation -> {
List<?> items = invocation.getArgument(0);
return items == null ? 0 : items.size();
});
lenient().doAnswer(invocation -> {
List<?> items = invocation.getArgument(1);
return items == null ? 0 : items.size();
}).when(excelAssemblyService).writeWorkbook(any(), any());
lenient().when(dailyFileService.currentBusinessDate()).thenReturn(BUSINESS_DATE);
lenient().when(dailyFileService.currentBusinessDateTime()).thenReturn(BUSINESS_TIME);
lenient().when(dailyFileService.shopKeyHash(anyString())).thenAnswer(invocation -> {
String key = invocation.getArgument(0);
return key == null ? null : "hash:" + key;
});
lenient().when(dailyFileService.shopKey(any(FileResultEntity.class))).thenAnswer(invocation -> {
FileResultEntity row = invocation.getArgument(0);
return row == null ? null : row.getSourceFilename();
});
lenient().when(dailyFileService.acquireLock(anyLong(), 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.findByLatestResultId(anyLong())).thenReturn(List.of());
lenient().when(dailyFileService.findById(anyLong())).thenReturn(null);
lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L);
lenient().doNothing().when(dailyFileService).deleteDailyFile(anyLong());
lenient().doNothing().when(dailyFileService).reassignMembers(anyLong(), anyLong());
lenient().doNothing().when(dailyFileService).deleteMembersForResults(any());
captureDailyFilePersistence();
captureMemberInserts();
captureResultUpdates();
captureFileJobs();
lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> {
Wrapper<FileResultEntity> wrapper = invocation.getArgument(0);
if (wrapper instanceof LambdaQueryWrapper<?> query
&& query.getSqlSegment() != null && query.getSqlSegment().contains("taskId")) {
List<FileResultEntity> rows = dbResultRows.stream()
.filter(r -> Objects.equals(r.getTaskId(), lastJobTaskId))
.sorted(Comparator.comparing(FileResultEntity::getId))
.toList();
return new ArrayList<>(rows);
}
return List.of();
});
}
@Test
void test_task_037_daily_file_job_normal_default_path() {
// 默认成功路径：接收完成后只入队一个异步组装作业（status=PENDING、携带 owner scope），
// 不做任何组装；组装作业执行后产生累计文件、成员行、作业状态置为 SUCCESS。
FileResultEntity row = addResultRow(8101L, 1L, 1, SHOP_NAME, null);
receiveTask(1L, List.of(row), snapshot(8101L));
assertEquals(1, dbFileJobs.size(), "接收路径只入队一个组装作业");
assertEquals("PENDING", dbFileJobs.get(0).getStatus());
assertEquals("ASSEMBLE_RESULT", dbFileJobs.get(0).getJobType());
assertTrue(enqueuedScopes.get(0).contains("owner"), "入队作业携带 owner scope");
verify(excelAssemblyService, never()).writeWorkbook(any(), any());
assertEquals(0, dbMembers.size(), "接收路径不产生成员行");
assertEquals(0, dbDailyFiles.size(), "接收路径不创建累计文件");
runAssembleJob(1L);
assertEquals(1, dbMembers.size(), "组装作业执行后产生成员行");
assertEquals(1, dbDailyFiles.size(), "组装作业执行后产生累计文件");
assertEquals("SUCCESS", dbFileJobs.get(0).getStatus(), "作业状态显式落到 SUCCESS");
assertEquals("SUCCESS", lastTaskStatus.get(1L), "作业完成后任务进入终态");
}
@Test
void test_task_037_daily_file_job_normal_multiple_items() {
// 批量场景：多个店铺结果入队各自独立的组装作业，逐个执行后
// 累计文件版本随提交递增，成员行不丢失、顺序稳定。
receiveTask(1L, List.of(addResultRow(8102L, 1L, 1, SHOP_NAME, null)), snapshot(8102L));
assertEquals(1, dbFileJobs.size(), "首个结果入队一个作业");
runAssembleJob(1L);
receiveTask(2L, List.of(addResultRow(8103L, 2L, 1, SHOP_NAME, null)), snapshot(8103L));
assertEquals(2, dbFileJobs.size(), "第二个结果入队第二个作业");
runAssembleJob(2L);
assertEquals(2, dbMembers.size(), "两个结果各产生一个成员行");
assertEquals(2L, dbDailyFiles.get(0).getVersion(), "版本号随两次归档递增");
assertEquals(2, dbFileJobs.stream().filter(j -> "SUCCESS".equals(j.getStatus())).count(),
"两个作业都显式落到 SUCCESS");
assertEquals("SUCCESS", lastTaskStatus.get(1L));
assertEquals("SUCCESS", lastTaskStatus.get(2L));
}
@Test
void test_task_037_daily_file_job_normal_repeated_operation_is_idempotent() {
// 幂等：同一结果重复提交接收，不产生重复组装作业（已存在 PENDING/SUCCESS 作业不重复入队）；
// 同一作业重复执行不产生重复成员行、版本不重复递增。
FileResultEntity row = addResultRow(8104L, 1L, 1, SHOP_NAME, null);
receiveTask(1L, List.of(row), snapshot(8104L));
receiveTask(1L, List.of(row), snapshot(8104L));
assertEquals(1, dbFileJobs.size(), "重复接收不重复入队作业");
runAssembleJob(1L);
runAssembleJob(1L);
assertEquals(1, dbMembers.size(), "重复执行作业不产生重复成员行");
assertEquals(1L, dbDailyFiles.get(0).getVersion(), "重复执行作业版本号不重复递增");
}
@Test
void test_task_037_daily_file_job_boundary_empty_input() {
// 空输入：没有成功结果时不入队作业、不创建任何资源，任务不进入组装等待。
FileResultEntity row = addResultRow(8105L, 1L, 0, SHOP_NAME, null);
receiveTask(1L, List.of(row), snapshot(8105L));
assertTrue(dbFileJobs.isEmpty(), "无成功结果不入队组装作业");
assertTrue(dbMembers.isEmpty(), "无成功结果不产生成员行");
assertTrue(dbDailyFiles.isEmpty(), "无成功结果不创建累计文件");
verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE));
}
@Test
void test_task_037_daily_file_job_boundary_single_item() {
// 单元素：单店铺单结果走同样拆分路径，作业一入队即达终态，不依赖批量逻辑。
FileResultEntity row = addResultRow(8106L, 1L, 1, SHOP_NAME, null);
receiveTask(1L, List.of(row), snapshot(8106L));
assertEquals(1, dbFileJobs.size());
assertEquals("PENDING", dbFileJobs.get(0).getStatus());
runAssembleJob(1L);
assertEquals(1, dbMembers.size());
assertEquals(1L, dbDailyFiles.get(0).getVersion());
assertEquals("SUCCESS", lastTaskStatus.get(1L));
}
@Test
void test_task_037_daily_file_job_boundary_limit_and_overflow() {
// 上限/超限：大量结果连续接收+执行，作业数、成员数、版本号一致递增，无丢失无重复。
int limit = 210;
for (int i = 0; i < limit; i++) {
long resultId = nextResultId++;
long taskId = i + 1L;
receiveTask(taskId, List.of(addResultRow(resultId, taskId, 1, SHOP_NAME, null)), snapshot(resultId));
runAssembleJob(taskId);
}
assertEquals(limit, dbFileJobs.size(), "每个结果一个作业");
assertEquals(limit, dbMembers.size(), "每个结果一个成员行");
assertEquals(Long.valueOf(limit), dbDailyFiles.get(0).getVersion(), "版本号与归档次数一致");
assertEquals(limit, dbFileJobs.stream().filter(j -> "SUCCESS".equals(j.getStatus())).count(),
"全部作业显式 SUCCESS");
}
@Test
void test_task_037_daily_file_job_invalid_input_rejected() {
// 非法参数：作业缺少 taskId 时拒绝执行并抛可识别异常；任务不存在时同样拒绝。
assertThrows(BusinessException.class, () -> service.processResultFileJob(
jobEntity(0L, null, MODULE_TYPE)));
assertThrows(BusinessException.class, () -> service.processResultFileJob(
jobEntity(0L, 9999L, MODULE_TYPE)));
}
@Test
void test_task_037_daily_file_job_dependency_failure_releases_resources() {
// 依赖失败：组装作业执行中 OSS 上传失败时抛出异常、不残留成员/累计文件；
// 恢复后同一结果可重新归档成功（错误可恢复）。
FileResultEntity row = addResultRow(8108L, 1L, 1, SHOP_NAME, null);
receiveTask(1L, List.of(row), snapshot(8108L));
assertEquals(1, dbFileJobs.size(), "接收入队成功");
dbFileJobs.clear();
doThrow(new IllegalStateException("upload failed"))
.when(ossStorageService).uploadResultFile(any(), eq(MODULE_TYPE));
assertThrows(IllegalStateException.class, () -> service.processResultFileJob(
jobEntity(1L, 1L, MODULE_TYPE)));
assertTrue(dbMembers.isEmpty(), "上传失败不残留成员行");
assertTrue(dbDailyFiles.isEmpty(), "上传失败不残留累计文件");
// 错误可恢复：恢复 OSS 后同一作业重新执行成功。
lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenAnswer(invocation -> {
String key = "oss/split/recovered-" + System.nanoTime() + ".xlsx";
lastUploadedObjectKey = key;
return key;
});
service.processResultFileJob(jobEntity(1L, 1L, MODULE_TYPE));
assertEquals(1, dbMembers.size(), "恢复后同一结果重新归档成功");
assertEquals(1, dbDailyFiles.size(), "恢复后累计文件创建成功");
assertEquals(1L, dbDailyFiles.get(0).getVersion(), "恢复后版本号正确");
}
// ---- helpers ----
/** 接收阶段：submitResult → tryFinalizeTask → finalizeTaskWorkbook（结果落库 + 入队组装作业）。 */
private void receiveTask(Long taskId, List<FileResultEntity> rows, ShopDataCrawlResultItemVo snapshot) {
lastJobTaskId = taskId;
FileTaskEntity task = taskEntity(taskId);
taskStore.put(taskId, task);
for (FileResultEntity row : rows) {
updateRowInDb(row);
}
service.tryFinalizeTask(taskId, false);
FileTaskEntity persisted = taskStore.get(taskId);
if (persisted != null) {
lastTaskStatus.put(taskId, persisted.getStatus());
}
}
/** 组装阶段：worker 执行 processResultFileJob（组装），随后 markSuccess + cleanupResultFileJob
* （作业成功钩子：把文件作业状态显式反映到任务状态，无未完成作业 → 任务进入终态）。 */
private void runAssembleJob(Long taskId) {
lastJobTaskId = taskId;
TaskFileJobEntity job = dbFileJobs.stream()
.filter(j -> Objects.equals(j.getTaskId(), taskId))
.findFirst()
.orElseThrow(() -> new AssertionError("作业不存在 taskId=" + taskId));
service.processResultFileJob(job);
job.setStatus("SUCCESS");
service.cleanupResultFileJob(job);
lastTaskStatus.put(taskId, taskStore.get(taskId).getStatus());
}
private void markResultFinishedInDb(FileResultEntity row) {
row.setSuccess(1);
row.setErrorMessage(null);
updateRowInDb(row);
}
private void updateRowInDb(FileResultEntity row) {
for (int i = 0; i < dbResultRows.size(); i++) {
if (Objects.equals(dbResultRows.get(i).getId(), row.getId())) {
dbResultRows.set(i, row);
return;
}
}
dbResultRows.add(row);
}
private TaskFileJobEntity jobEntity(long jobId, Long taskId, String moduleType) {
TaskFileJobEntity job = new TaskFileJobEntity();
job.setId(jobId);
job.setTaskId(taskId);
job.setModuleType(moduleType);
return job;
}
private FileTaskEntity taskEntity(long taskId) {
FileTaskEntity task = new FileTaskEntity();
task.setId(taskId);
task.setModuleType(MODULE_TYPE);
task.setStatus("RUNNING");
task.setUserId(7L);
task.setOwnerInstanceId("instance-a");
task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}");
task.setResultJson("[]");
task.setCreatedAt(LocalDateTime.now());
task.setUpdatedAt(LocalDateTime.now());
return task;
}
private FileResultEntity addResultRow(long id, long taskId, int success, String shopName, String resultFileUrl) {
FileResultEntity row = new FileResultEntity();
row.setId(id);
row.setTaskId(taskId);
row.setModuleType(MODULE_TYPE);
row.setSuccess(success);
row.setSourceFilename(shopName);
row.setSourceFileUrl("shop-id-" + id);
row.setUserId(7L);
row.setCreatedAt(LocalDateTime.now());
row.setResultFileUrl(resultFileUrl);
dbResultRows.add(row);
return row;
}
private ShopDataCrawlResultItemVo snapshot(long resultId) {
ShopDataCrawlResultItemVo item = new ShopDataCrawlResultItemVo();
item.setResultId(resultId);
item.setTaskId(1L);
item.setShopName(SHOP_NAME);
item.setShopId("shop-id-" + resultId);
item.setSuccess(true);
item.setMatched(true);
item.setTaskStatus("SUCCESS");
item.setCountryCodes(List.of("DE"));
return item;
}
private void captureFileJobs() {
// 仿真 enqueueAssembleResult 的幂等语义：同一 (taskId, resultId) 已存在作业时不重复入队。
lenient().doAnswer(invocation -> {
Long taskId = invocation.getArgument(0);
Long resultId = invocation.getArgument(2);
String scopeKey = invocation.getArgument(3);
TaskFileJobEntity existing = dbFileJobs.stream()
.filter(j -> Objects.equals(j.getTaskId(), taskId)
&& Objects.equals(j.getResultId(), resultId))
.findFirst()
.orElse(null);
if (existing != null) {
if ("FAILED".equals(existing.getStatus())) {
existing.setStatus("PENDING");
}
return existing;
}
TaskFileJobEntity job = new TaskFileJobEntity();
job.setId(jobIdSeq.incrementAndGet());
job.setTaskId(taskId);
job.setModuleType(MODULE_TYPE);
job.setResultId(resultId);
job.setScopeKey(scopeKey);
job.setStatus("PENDING");
job.setJobType("ASSEMBLE_RESULT");
job.setRetryCount(0);
dbFileJobs.add(job);
enqueuedScopes.add(scopeKey);
return job;
}).when(taskFileJobService).enqueueAssembleResult(anyLong(), eq(MODULE_TYPE), anyLong(), anyString());
lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong()))
.thenAnswer(invocation -> {
long taskId = invocation.getArgument(0);
long resultId = invocation.getArgument(2);
return dbFileJobs.stream()
.filter(j -> Objects.equals(j.getTaskId(), taskId)
&& Objects.equals(j.getResultId(), resultId))
.findFirst()
.orElse(null);
});
lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE)))
.thenAnswer(invocation -> {
long taskId = invocation.getArgument(0);
return dbFileJobs.stream()
.filter(j -> Objects.equals(j.getTaskId(), taskId)
&& !"SUCCESS".equals(j.getStatus()))
.count();
});
}
private void captureResultUpdates() {
lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenAnswer(invocation -> {
FileResultEntity updated = invocation.getArgument(0);
for (int i = 0; i < dbResultRows.size(); i++) {
if (Objects.equals(dbResultRows.get(i).getId(), updated.getId())) {
dbResultRows.set(i, updated);
return 1;
}
}
dbResultRows.add(updated);
return 1;
});
}
private void captureMemberInserts() {
lenient().when(dailyFileService.addMemberWithPayload(anyLong(), anyLong(), anyLong(), anyString()))
.thenAnswer(invocation -> {
long dailyFileId = invocation.getArgument(0);
long taskId = invocation.getArgument(1);
long resultId = invocation.getArgument(2);
String rowPayload = invocation.getArgument(3);
boolean duplicate = dbMembers.stream().anyMatch(m ->
Objects.equals(m.getDailyFileId(), dailyFileId)
&& Objects.equals(m.getResultId(), resultId));
if (duplicate) {
return false;
}
ShopDataCrawlDailyMemberEntity member = new ShopDataCrawlDailyMemberEntity();
member.setId(memberIdSeq.incrementAndGet());
member.setDailyFileId(dailyFileId);
member.setTaskId(taskId);
member.setResultId(resultId);
member.setRowPayload(rowPayload);
member.setCreatedAt(BUSINESS_TIME.plusMinutes(dbMembers.size()));
dbMembers.add(member);
return true;
});
}
private void captureDailyFilePersistence() {
lenient().doAnswer(invocation -> {
ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0);
entity.setId(fileIdSeq.incrementAndGet());
dbDailyFiles.add(entity);
return null;
}).when(dailyFileService).insert(any(ShopDataCrawlDailyFileEntity.class));
lenient().doAnswer(invocation -> {
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())
&& Objects.equals(existing.getBusinessDate(), entity.getBusinessDate())) {
entity.setId(existing.getId());
dbDailyFiles.set(i, copyDailyFile(entity));
return null;
}
}
entity.setId(fileIdSeq.incrementAndGet());
dbDailyFiles.add(copyDailyFile(entity));
return null;
}).when(dailyFileService).update(any(ShopDataCrawlDailyFileEntity.class));
lenient().when(dailyFileService.listMembers(anyLong())).thenAnswer(invocation -> {
long dailyFileId = invocation.getArgument(0);
return dbMembers.stream()
.filter(m -> Objects.equals(m.getDailyFileId(), dailyFileId))
.sorted(Comparator
.comparing(ShopDataCrawlDailyMemberEntity::getCreatedAt,
Comparator.nullsLast(Comparator.naturalOrder()))
.thenComparing(ShopDataCrawlDailyMemberEntity::getId,
Comparator.nullsLast(Comparator.naturalOrder())))
.toList();
});
lenient().when(dailyFileService.containsResult(anyLong(), anyLong())).thenAnswer(invocation -> {
long dailyFileId = invocation.getArgument(0);
long resultId = invocation.getArgument(1);
return dbMembers.stream().anyMatch(m ->
Objects.equals(m.getDailyFileId(), dailyFileId) && Objects.equals(m.getResultId(), resultId));
});
lenient().when(dailyFileService.findMembersByResultId(anyLong())).thenAnswer(invocation -> {
long resultId = invocation.getArgument(0);
return dbMembers.stream().filter(m -> Objects.equals(m.getResultId(), resultId)).toList();
});
}
private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) {
for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) {
if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) {
return f;
}
}
return null;
}
private ShopDataCrawlDailyFileEntity copyDailyFile(ShopDataCrawlDailyFileEntity source) {
if (source == null) {
return null;
}
ShopDataCrawlDailyFileEntity copy = new ShopDataCrawlDailyFileEntity();
copy.setId(source.getId());
copy.setUserId(source.getUserId());
copy.setShopKeyHash(source.getShopKeyHash());
copy.setShopKey(source.getShopKey());
copy.setBusinessDate(source.getBusinessDate());
copy.setLatestTaskId(source.getLatestTaskId());
copy.setLatestResultId(source.getLatestResultId());
copy.setResultFilename(source.getResultFilename());
copy.setResultFileUrl(source.getResultFileUrl());
copy.setResultFileSize(source.getResultFileSize());
copy.setResultContentType(source.getResultContentType());
copy.setRowCount(source.getRowCount());
copy.setVersion(source.getVersion());
copy.setLastSuccessAt(source.getLastSuccessAt());
copy.setCreatedAt(source.getCreatedAt());
copy.setUpdatedAt(source.getUpdatedAt());
return copy;
}
private FileTaskEntity copyTask(FileTaskEntity source) {
FileTaskEntity copy = new FileTaskEntity();
copy.setId(source.getId());
copy.setTaskNo(source.getTaskNo());
copy.setModuleType(source.getModuleType());
copy.setTaskMode(source.getTaskMode());
copy.setStatus(source.getStatus());
copy.setSourceFileCount(source.getSourceFileCount());
copy.setSuccessFileCount(source.getSuccessFileCount());
copy.setFailedFileCount(source.getFailedFileCount());
copy.setRequestJson(source.getRequestJson());
copy.setResultJson(source.getResultJson());
copy.setErrorMessage(source.getErrorMessage());
copy.setCreatedBy(source.getCreatedBy());
copy.setUserId(source.getUserId());
copy.setOwnerInstanceId(source.getOwnerInstanceId());
copy.setCreatedAt(source.getCreatedAt());
copy.setUpdatedAt(source.getUpdatedAt());
copy.setFinishedAt(source.getFinishedAt());
copy.setScheduledAt(source.getScheduledAt());
return copy;
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java
+30
View File
@@ -52,6 +52,7 @@ import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -257,6 +258,35 @@ class ShopDataCrawlScopeMergeTest {
verify(taskScopeStateMapper, never()).selectOne(any());
}
@Test
void test_task_029_submit_preserves_previous_countries_across_submissions() {
// 回归：客户端先提交一个国家的部分结果（未完结），再提交另一个国家并标记完结。
// 首次提交国家被清空后，第二次提交的 payload 只含 UK，修复前 DE 会在快照中永久丢失；
// 修复后 mergePayloadIntoSnapshot 按国家合并，DE 必须保留、且第二次完成后正常收尾。
Map<String, ShopDataCrawlShopPayloadDto> mergedByShop = new LinkedHashMap<>();
when(taskCacheService.getShopMergedPayload(anyLong(), anyString())).thenAnswer(
invocation -> mergedByShop.get(invocation.getArgument(1)));
doAnswer(invocation -> {
mergedByShop.put(invocation.getArgument(1), invocation.getArgument(2));
return null;
}).when(taskCacheService).saveShopMergedPayload(anyLong(), anyString(), any(ShopDataCrawlShopPayloadDto.class));
givenRunningTask(1205L, 2205L);
// 第一次提交：只带 DE 国家，未完结（生产上第一次提交的 shopDone=false）
ShopDataCrawlShopPayloadDto first = legacyChunk(false, "DE", row("2026-07-25", "B001"));
service.submitResult(task.getId(), request(first));
// 第二次提交：只带 UK 国家，标记完结（生产上第二次提交的 shopDone=true 收尾）
ShopDataCrawlShopPayloadDto second = legacyChunk(true, "UK", row("2026-07-26", "B002"));
service.submitResult(task.getId(), request(second));
assertEquals(1, result.getSuccess(), "任务应收尾成功");
String json = task.getResultJson();
assertTrue(json.contains("\"DE\""), "前次提交的 DE 国家必须保留: " + json);
assertTrue(json.contains("\"UK\""), "后次提交的 UK 国家必须写入: " + json);
assertTrue(json.indexOf("B001") < json.indexOf("B002"));
}
@Test
void test_task_029_chunk_merge_boundary_single_item() {
// 单元素：单分片 1/1，一次 scope 查询 + 一次 insert 即齐集完成。