task-31: 将任务快照改为轻量进度字段，避免每次写入完整结果 JSON
This commit is contained in:
+29
-4
@@ -494,8 +494,7 @@ public class ShopDataCrawlTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
persistSnapshotJson(task, snapshots);
|
||||
fileTaskMapper.updateById(task);
|
||||
persistProgressOrSnapshot(task, snapshots, resultRows.stream().allMatch(this::isResultFinished));
|
||||
taskCacheService.saveTaskCache(task);
|
||||
tryFinalizeTask(taskId, false);
|
||||
}
|
||||
@@ -599,13 +598,14 @@ public class ShopDataCrawlTaskService {
|
||||
List<FileResultEntity> latestRows = listTaskRows(taskId);
|
||||
updateTaskStatusFromRows(task, latestRows);
|
||||
if (latestRows.stream().allMatch(this::isResultFinished)) {
|
||||
persistSnapshotJson(task, snapshots);
|
||||
if (changed || !snapshotJsonHasCountryRows(task)) {
|
||||
persistSnapshotJson(task, snapshots);
|
||||
}
|
||||
fileTaskMapper.updateById(task);
|
||||
finalizeTaskWorkbook(task, latestRows, snapshots);
|
||||
return true;
|
||||
}
|
||||
|
||||
persistSnapshotJson(task, snapshots);
|
||||
fileTaskMapper.updateById(task);
|
||||
taskCacheService.saveTaskCache(task);
|
||||
return changed;
|
||||
@@ -2332,6 +2332,20 @@ public class ShopDataCrawlTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
private boolean snapshotJsonHasCountryRows(FileTaskEntity task) {
|
||||
if (task == null) {
|
||||
return false;
|
||||
}
|
||||
for (ShopDataCrawlResultItemVo snapshot : parseTaskSnapshots(task.getResultJson())) {
|
||||
if (snapshot != null && snapshot.getCountryResults() != null
|
||||
&& snapshot.getCountryResults().stream().anyMatch(country ->
|
||||
country != null && country.getItems() != null && !country.getItems().isEmpty())) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private void persistSnapshotJson(FileTaskEntity task, List<ShopDataCrawlResultItemVo> snapshots) {
|
||||
try {
|
||||
task.setResultJson(objectMapper.writeValueAsString(snapshots == null ? List.of() : snapshots));
|
||||
@@ -2341,6 +2355,17 @@ public class ShopDataCrawlTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* RUNNING 期间只写轻量进度字段（successFileCount/failedFileCount/status/updatedAt），
|
||||
* 避免每次分片接收都序列化并落库完整结果 JSON；仅终态（全部结果行完成）才写完整快照。
|
||||
*/
|
||||
private void persistProgressOrSnapshot(FileTaskEntity task, List<ShopDataCrawlResultItemVo> snapshots, boolean terminal) {
|
||||
if (terminal) {
|
||||
persistSnapshotJson(task, snapshots);
|
||||
}
|
||||
fileTaskMapper.updateById(task);
|
||||
}
|
||||
|
||||
private void syncSnapshotTables(FileTaskEntity task, List<ShopDataCrawlResultItemVo> snapshots) {
|
||||
List<ShopDataCrawlResultItemVo> safe = snapshots == null ? List.of() : snapshots;
|
||||
taskResultItemService.replaceTaskSnapshots(task.getId(), MODULE_TYPE, safe, new TaskResultItemService.SnapshotKeyResolver() {
|
||||
|
||||
+587
@@ -0,0 +1,587 @@
|
||||
package com.nanri.aiimage.modules.shopdatacrawl.service;
|
||||
|
||||
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.dto.ShopDataCrawlCountryResultDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlSubmitResultRequest;
|
||||
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.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 org.springframework.dao.DuplicateKeyException;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
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.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.ArgumentMatchers.nullable;
|
||||
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 31：将任务快照改为轻量进度字段，避免每次写入完整结果 JSON。
|
||||
* RUNNING 期间每次分片接收只更新任务行上的轻量进度字段（successFileCount /
|
||||
* failedFileCount / status / updatedAt），不再序列化完整结果 JSON，也不写快照表；
|
||||
* 完整结果 JSON（含国家行）只在任务终态（全部行完成）时写入一次。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ShopDataCrawlLightweightProgressTest {
|
||||
|
||||
private static final String MODULE_TYPE = "SHOP_DATA_CRAWL";
|
||||
private static final String SHOP_NAME = "Demo Shop";
|
||||
|
||||
@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);
|
||||
}
|
||||
|
||||
@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<TaskChunkEntity> storedChunks = new ArrayList<>();
|
||||
private final List<TaskScopeStateEntity> storedScopes = new ArrayList<>();
|
||||
private final Map<String, String> rustfsPayloads = new LinkedHashMap<>();
|
||||
private FileTaskEntity task;
|
||||
private FileResultEntity result;
|
||||
private int nextPayloadId;
|
||||
private boolean insertFails;
|
||||
private boolean scopeUpdateFails;
|
||||
|
||||
@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);
|
||||
|
||||
storedChunks.clear();
|
||||
storedScopes.clear();
|
||||
rustfsPayloads.clear();
|
||||
nextPayloadId = 0;
|
||||
insertFails = false;
|
||||
scopeUpdateFails = false;
|
||||
|
||||
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.getAllShopMergedPayload(anyLong())).thenReturn(Map.of());
|
||||
lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of());
|
||||
lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null);
|
||||
lenient().when(excelAssemblyService.countRows(any())).thenReturn(1);
|
||||
|
||||
lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> {
|
||||
Long taskId = invocation.getArgument(0);
|
||||
return task != null && Objects.equals(taskId, task.getId()) ? task : null;
|
||||
});
|
||||
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||
lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> {
|
||||
Long resultId = invocation.getArgument(0);
|
||||
return result != null && Objects.equals(resultId, result.getId()) ? result : null;
|
||||
});
|
||||
lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<FileResultEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
return result != null && Objects.equals(taskId, result.getTaskId()) ? List.of(result) : List.of();
|
||||
});
|
||||
lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1);
|
||||
|
||||
lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> {
|
||||
String value = invocation.getArgument(0);
|
||||
return value == null ? "" : value.trim();
|
||||
});
|
||||
|
||||
configureTransientPayloadStorage();
|
||||
configureChunkMapper();
|
||||
configureScopeMapper();
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_normal_default_path() {
|
||||
// 正常路径：分片到达只更新轻量进度字段（行成功、计数器推进），
|
||||
// 不写完整结果 JSON、不写快照表；终态时完整 JSON 才写入一次。
|
||||
givenRunningTask(1311L, 2311L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
|
||||
assertEquals("RUNNING", task.getStatus(), "未齐集仍为 RUNNING");
|
||||
assertEquals(0, successFileCount(), "未齐集不推进成功计数");
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertTrue(!hasResultRowsInJson(), "RUNNING 期间不写完整结果 JSON");
|
||||
verify(taskResultItemService, never()).replaceTaskSnapshots(any(), any(), any(), any());
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))));
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()), "齐集后成功计数=1");
|
||||
assertEquals(0, Integer.valueOf(task.getFailedFileCount()));
|
||||
assertEquals("SUCCESS", task.getStatus());
|
||||
assertTrue(hasResultRowsInJson(), "终态写入完整结果 JSON");
|
||||
assertTrue(task.getResultJson().indexOf("B001") < task.getResultJson().indexOf("B002"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_normal_multiple_items() {
|
||||
// 批量场景：多分片逐批到达，轻量进度字段持续反映进度，快照表一次不写。
|
||||
givenRunningTask(1312L, 2312L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 3, "DE", row("2026-07-25", "B001"))));
|
||||
service.submitResult(task.getId(), request(chunk(2, 3, "UK", row("2026-07-26", "B002"))));
|
||||
assertEquals("RUNNING", task.getStatus());
|
||||
assertEquals(0, successFileCount());
|
||||
verify(taskResultItemService, never()).replaceTaskSnapshots(any(), any(), any(), any());
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(3, 3, "FR", row("2026-07-27", "B003"))));
|
||||
assertEquals("SUCCESS", task.getStatus());
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()));
|
||||
assertEquals(0, Integer.valueOf(task.getFailedFileCount()), "失败计数保留 0");
|
||||
|
||||
String json = task.getResultJson();
|
||||
assertTrue(json.indexOf("B001") < json.indexOf("B002") && json.indexOf("B002") < json.indexOf("B003"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_normal_repeated_operation_is_idempotent() {
|
||||
// 幂等：同一分片重放不重复推进进度，计数器/状态稳定，无重复快照写。
|
||||
givenRunningTask(1313L, 2313L);
|
||||
ShopDataCrawlSubmitResultRequest request = request(chunk(1, 1, "DE", row("2026-07-25", "B001")));
|
||||
|
||||
service.submitResult(task.getId(), request);
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()));
|
||||
assertEquals("SUCCESS", task.getStatus());
|
||||
String jsonAfterFirst = task.getResultJson();
|
||||
assertTrue(hasResultRowsInJson(), "终态写入完整结果 JSON");
|
||||
|
||||
BusinessException replayError = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request));
|
||||
assertTrue(replayError.getMessage().contains("已结束"), "任务终态后重复提交被拒");
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()), "重放不重复计数");
|
||||
assertEquals(jsonAfterFirst, task.getResultJson(), "重放不放大完整 JSON");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_boundary_empty_input() {
|
||||
// 空输入：空分片在快照写路径之前被拒，无进度更新、无资源创建。
|
||||
givenRunningTask(1314L, 2314L);
|
||||
|
||||
ShopDataCrawlShopPayloadDto emptyItems = new ShopDataCrawlShopPayloadDto();
|
||||
emptyItems.setShopName(SHOP_NAME);
|
||||
emptyItems.setChunkIndex(1);
|
||||
emptyItems.setChunkTotal(1);
|
||||
emptyItems.setCountryResults(List.of(countryWithItems("DE", List.of())));
|
||||
BusinessException error = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(emptyItems)));
|
||||
assertTrue(error.getMessage().contains("内容为空"));
|
||||
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
assertEquals(0, storedScopes.size());
|
||||
assertEquals(0, successFileCount(), "空分片不推进进度");
|
||||
verify(taskResultItemService, never()).replaceTaskSnapshots(any(), any(), any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_boundary_single_item() {
|
||||
// 单元素：单分片 1/1 直接终态，完整 JSON 写入一次，无重复快照写。
|
||||
givenRunningTask(1315L, 2315L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 1, "DE", row("2026-07-25", "B001"))));
|
||||
|
||||
assertEquals("SUCCESS", task.getStatus());
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()));
|
||||
assertTrue(hasResultRowsInJson());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_boundary_limit_and_overflow() {
|
||||
// 上限/超限：5/5 全量到达 + 一次重放，轻量进度最终收敛，完整 JSON 只写一次。
|
||||
givenRunningTask(1316L, 2316L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 5, "UK", row("2026-07-25", "B001"))));
|
||||
service.submitResult(task.getId(), request(chunk(1, 5, "UK", row("2026-07-25", "B001"))));
|
||||
assertEquals("RUNNING", task.getStatus());
|
||||
assertEquals(0, Integer.valueOf(task.getSuccessFileCount()));
|
||||
|
||||
String[] countries = {"DE", "FR", "ES", "IT"};
|
||||
String[] dates = {"2026-07-26", "2026-07-27", "2026-07-28", "2026-07-29"};
|
||||
for (int i = 2; i <= 5; i++) {
|
||||
service.submitResult(task.getId(), request(chunk(i, 5, countries[i - 2], row(dates[i - 2], "B00" + i))));
|
||||
}
|
||||
|
||||
assertEquals("SUCCESS", task.getStatus());
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()));
|
||||
assertEquals(1, storedScopes.size(), "scope 行始终只有一行");
|
||||
assertEquals(5, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(5, countResultJsonRows(), "终态完整 JSON 含全部 5 行");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_invalid_input_rejected() {
|
||||
// 非法参数：非正 chunk_index/chunk_total 在进度写路径之前拒绝，无状态写入。
|
||||
givenRunningTask(1317L, 2317L);
|
||||
|
||||
BusinessException zeroIndex = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(0, 1, "DE", row("2026-07-25", "B001")))));
|
||||
assertTrue(zeroIndex.getMessage().contains("chunk_index"));
|
||||
BusinessException zeroTotal = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(1, 0, "DE", row("2026-07-25", "B001")))));
|
||||
assertTrue(zeroTotal.getMessage().contains("chunk_total"));
|
||||
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, storedScopes.size());
|
||||
assertEquals("RUNNING", task.getStatus());
|
||||
assertEquals(0, successFileCount());
|
||||
verify(taskResultItemService, never()).replaceTaskSnapshots(any(), any(), any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_031_snapshot_progress_dependency_failure_releases_resources() {
|
||||
// 依赖失败：chunk 插入失败 → payload 清理、无进度写入；恢复后重试成功，
|
||||
// 终态完整 JSON 写入一次。
|
||||
givenRunningTask(1318L, 2318L);
|
||||
|
||||
insertFails = true;
|
||||
assertThrows(RuntimeException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))));
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
assertEquals(0, storedScopes.size());
|
||||
insertFails = false;
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))));
|
||||
|
||||
assertEquals("SUCCESS", task.getStatus());
|
||||
assertEquals(1, Integer.valueOf(task.getSuccessFileCount()));
|
||||
assertTrue(hasResultRowsInJson());
|
||||
}
|
||||
|
||||
private int successFileCount() {
|
||||
return task.getSuccessFileCount() == null ? 0 : task.getSuccessFileCount();
|
||||
}
|
||||
|
||||
private boolean hasResultRowsInJson() {
|
||||
return countResultJsonRows() > 0;
|
||||
}
|
||||
|
||||
private int countResultJsonRows() {
|
||||
try {
|
||||
List<ShopDataCrawlResultItemVo> snapshots = objectMapper.readValue(task.getResultJson(),
|
||||
objectMapper.getTypeFactory().constructCollectionType(List.class, ShopDataCrawlResultItemVo.class));
|
||||
int count = 0;
|
||||
for (ShopDataCrawlResultItemVo snapshot : snapshots) {
|
||||
if (snapshot.getCountryResults() == null) {
|
||||
continue;
|
||||
}
|
||||
for (ShopDataCrawlCountryResultDto country : snapshot.getCountryResults()) {
|
||||
count += country.getItems() == null ? 0 : country.getItems().size();
|
||||
}
|
||||
}
|
||||
return count;
|
||||
} catch (Exception ex) {
|
||||
throw new IllegalStateException("解析结果 JSON 失败", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private void configureTransientPayloadStorage() {
|
||||
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||
lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||
anyString(), anyLong(), anyString(), any(), anyString())).thenAnswer(invocation -> {
|
||||
String pointer = "rustfs:payload-" + (++nextPayloadId);
|
||||
rustfsPayloads.put(pointer, invocation.getArgument(4));
|
||||
return pointer;
|
||||
});
|
||||
lenient().when(transientPayloadStorageService.extractPointer(any())).thenAnswer(invocation -> {
|
||||
String value = invocation.getArgument(0);
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
return value.startsWith("rustfs:") || value.startsWith("local:") || value.startsWith("oss:")
|
||||
? value : null;
|
||||
});
|
||||
lenient().when(transientPayloadStorageService.resolvePayload(any(), any())).thenAnswer(invocation -> {
|
||||
String pointer = invocation.getArgument(0);
|
||||
String payload = rustfsPayloads.get(pointer);
|
||||
if (payload == null) {
|
||||
throw new IllegalStateException("missing test RustFS payload: " + pointer);
|
||||
}
|
||||
return payload;
|
||||
});
|
||||
lenient().doAnswer(invocation -> {
|
||||
rustfsPayloads.remove(invocation.getArgument(0));
|
||||
return null;
|
||||
}).when(transientPayloadStorageService).deletePayloadIfPresent(any());
|
||||
}
|
||||
|
||||
private void configureChunkMapper() {
|
||||
lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> {
|
||||
if (insertFails) {
|
||||
throw new RuntimeException("db down");
|
||||
}
|
||||
TaskChunkEntity chunk = invocation.getArgument(0);
|
||||
boolean duplicate = storedChunks.stream().anyMatch(existing ->
|
||||
Objects.equals(existing.getTaskId(), chunk.getTaskId())
|
||||
&& Objects.equals(existing.getScopeHash(), chunk.getScopeHash())
|
||||
&& Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex()));
|
||||
if (duplicate) {
|
||||
throw new DuplicateKeyException("duplicate chunk key: " + chunk.getScopeHash() + "/" + chunk.getChunkIndex());
|
||||
}
|
||||
chunk.setId((long) storedChunks.size() + 1L);
|
||||
storedChunks.add(chunk);
|
||||
return 1;
|
||||
});
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
Integer chunkIndex = queryInteger(query);
|
||||
return storedChunks.stream()
|
||||
.filter(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
});
|
||||
lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
return storedChunks.stream()
|
||||
.filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null))
|
||||
.sorted(Comparator.comparing(TaskChunkEntity::getChunkIndex))
|
||||
.toList();
|
||||
});
|
||||
lenient().when(taskChunkMapper.delete(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
Integer chunkIndex = queryInteger(query);
|
||||
int before = storedChunks.size();
|
||||
storedChunks.removeIf(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex));
|
||||
return before - storedChunks.size();
|
||||
});
|
||||
}
|
||||
|
||||
private void configureScopeMapper() {
|
||||
lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
|
||||
TaskScopeStateEntity scope = invocation.getArgument(0);
|
||||
scope.setId((long) storedScopes.size() + 1L);
|
||||
storedScopes.add(scope);
|
||||
return 1;
|
||||
});
|
||||
lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskScopeStateEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
return storedScopes.stream()
|
||||
.filter(scope -> Objects.equals(taskId, scope.getTaskId()))
|
||||
.filter(scope -> scopeHash == null || Objects.equals(scopeHash, scope.getScopeHash()))
|
||||
.findFirst()
|
||||
.map(ShopDataCrawlLightweightProgressTest::copyScope)
|
||||
.orElse(null);
|
||||
});
|
||||
lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
|
||||
if (scopeUpdateFails) {
|
||||
throw new RuntimeException("scope update down");
|
||||
}
|
||||
TaskScopeStateEntity updated = invocation.getArgument(0);
|
||||
for (int i = 0; i < storedScopes.size(); i++) {
|
||||
if (Objects.equals(storedScopes.get(i).getId(), updated.getId())) {
|
||||
storedScopes.set(i, copyScope(updated));
|
||||
}
|
||||
}
|
||||
return 1;
|
||||
});
|
||||
}
|
||||
|
||||
private static TaskScopeStateEntity copyScope(TaskScopeStateEntity source) {
|
||||
TaskScopeStateEntity copy = new TaskScopeStateEntity();
|
||||
copy.setId(source.getId());
|
||||
copy.setTaskId(source.getTaskId());
|
||||
copy.setModuleType(source.getModuleType());
|
||||
copy.setScopeKey(source.getScopeKey());
|
||||
copy.setScopeHash(source.getScopeHash());
|
||||
copy.setChunkTotal(source.getChunkTotal());
|
||||
copy.setReceivedChunkCount(source.getReceivedChunkCount());
|
||||
copy.setCompleted(source.getCompleted());
|
||||
copy.setLastChunkAt(source.getLastChunkAt());
|
||||
copy.setLastError(source.getLastError());
|
||||
copy.setCreatedAt(source.getCreatedAt());
|
||||
copy.setUpdatedAt(source.getUpdatedAt());
|
||||
return copy;
|
||||
}
|
||||
|
||||
private boolean matchesChunk(TaskChunkEntity chunk, Long taskId, String scopeHash, Integer chunkIndex) {
|
||||
return (taskId == null || Objects.equals(taskId, chunk.getTaskId()))
|
||||
&& (scopeHash == null || Objects.equals(scopeHash, chunk.getScopeHash()))
|
||||
&& (chunkIndex == null || Objects.equals(chunkIndex, chunk.getChunkIndex()));
|
||||
}
|
||||
|
||||
private Long queryLong(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(Long.class::isInstance)
|
||||
.map(Long.class::cast)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private Integer queryInteger(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(Integer.class::isInstance)
|
||||
.map(Integer.class::cast)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private String queryScopeHash(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(String.class::isInstance)
|
||||
.map(String.class::cast)
|
||||
.filter(value -> value.length() == 64)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private void givenRunningTask(long taskId, long resultId) {
|
||||
task = new FileTaskEntity();
|
||||
task.setId(taskId);
|
||||
task.setUserId(7L);
|
||||
task.setModuleType(MODULE_TYPE);
|
||||
task.setStatus("RUNNING");
|
||||
task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}");
|
||||
task.setResultJson("[]");
|
||||
task.setCreatedAt(LocalDateTime.now());
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
|
||||
result = new FileResultEntity();
|
||||
result.setId(resultId);
|
||||
result.setTaskId(taskId);
|
||||
result.setUserId(7L);
|
||||
result.setModuleType(MODULE_TYPE);
|
||||
result.setSourceFilename(SHOP_NAME);
|
||||
result.setSourceFileUrl("shop-1");
|
||||
result.setSuccess(-1);
|
||||
result.setCreatedAt(LocalDateTime.now());
|
||||
}
|
||||
|
||||
private ShopDataCrawlSubmitResultRequest request(ShopDataCrawlShopPayloadDto payload) {
|
||||
ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest();
|
||||
request.setShops(List.of(payload));
|
||||
return request;
|
||||
}
|
||||
|
||||
private ShopDataCrawlShopPayloadDto chunk(int chunkIndex,
|
||||
int chunkTotal,
|
||||
String country,
|
||||
ShopDataCrawlRowDto row) {
|
||||
ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto();
|
||||
payload.setShopName(SHOP_NAME);
|
||||
payload.setChunkIndex(chunkIndex);
|
||||
payload.setChunkTotal(chunkTotal);
|
||||
payload.setCountryResults(List.of(country(country, row)));
|
||||
return payload;
|
||||
}
|
||||
|
||||
private ShopDataCrawlCountryResultDto country(String country, ShopDataCrawlRowDto row) {
|
||||
return countryWithItems(country, List.of(row));
|
||||
}
|
||||
|
||||
private ShopDataCrawlCountryResultDto countryWithItems(String country, List<ShopDataCrawlRowDto> items) {
|
||||
ShopDataCrawlCountryResultDto result = new ShopDataCrawlCountryResultDto();
|
||||
result.setCountry(country);
|
||||
result.setItems(items);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ShopDataCrawlRowDto row(String date, String asin) {
|
||||
ShopDataCrawlRowDto row = new ShopDataCrawlRowDto();
|
||||
row.setDate(date);
|
||||
row.setAsin(asin);
|
||||
row.setCommodityImage("https://m.media-amazon.com/images/I/" + asin + ".jpg");
|
||||
row.setInventorySales("10");
|
||||
row.setSalesRank("20");
|
||||
row.setPageViews("30");
|
||||
row.setUnitsSold("40");
|
||||
row.setPrice("50");
|
||||
row.setRecommendedOffer("60");
|
||||
return row;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user