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

后端架构更新

Browse Source
This commit is contained in:
super
2026-04-27 09:18:10 +08:00
parent 0caf62c3d2
commit 995e2ee65a
141 changed files with 6957 additions and 772 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/model/vo/QueryAsinResultItemVo.java
+4
View File
@@ -43,6 +43,10 @@ public class QueryAsinResultItemVo {
private LocalDateTime finishedAt;
private String outputFilename;
private String downloadUrl;
private Long fileJobId;
private String fileStatus;
private String fileError;
private Boolean fileReady;
@JsonProperty("queryAsins")
private List<QueryAsinCountryAsinsDto> queryAsins = new ArrayList<>();
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinTaskCacheService.java
+30 -8
View File
@@ -6,6 +6,7 @@ import com.nanri.aiimage.modules.queryasin.model.dto.QueryAsinShopPayloadDto;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
@@ -18,6 +19,7 @@ import java.util.concurrent.ConcurrentHashMap;
@Service
@RequiredArgsConstructor
@Slf4j
public class QueryAsinTaskCacheService {
private static final String MODULE_TYPE = "QUERY_ASIN";
@@ -60,7 +62,13 @@ public class QueryAsinTaskCacheService {
if (taskId == null || taskId <= 0) {
return 0L;
}
String raw = stringRedisTemplate.opsForValue().get(buildTaskHeartbeatKey(taskId));
String raw;
try {
raw = stringRedisTemplate.opsForValue().get(buildTaskHeartbeatKey(taskId));
} catch (Exception ex) {
log.warn("[query-asin-cache] get heartbeat degraded taskId={} msg={}", taskId, ex.getMessage());
return 0L;
}
if (raw == null || raw.isBlank()) {
return 0L;
}
@@ -72,10 +80,14 @@ public class QueryAsinTaskCacheService {
}
public void touchTaskHeartbeat(Long taskId) {
stringRedisTemplate.opsForValue().set(
buildTaskHeartbeatKey(taskId),
String.valueOf(Instant.now().toEpochMilli()),
Duration.ofHours(PAYLOAD_TTL_HOURS));
try {
stringRedisTemplate.opsForValue().set(
buildTaskHeartbeatKey(taskId),
String.valueOf(Instant.now().toEpochMilli()),
Duration.ofHours(PAYLOAD_TTL_HOURS));
} catch (Exception ex) {
log.warn("[query-asin-cache] touch heartbeat degraded taskId={} msg={}", taskId, ex.getMessage());
}
}
public void deleteTaskCache(Long taskId) {
@@ -83,8 +95,12 @@ public class QueryAsinTaskCacheService {
return;
}
taskEntityLocalCache.remove(taskId);
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
try {
stringRedisTemplate.delete(buildTaskHeartbeatKey(taskId));
stringRedisTemplate.delete(buildTaskEntityKey(taskId));
} catch (Exception ex) {
log.warn("[query-asin-cache] delete cache degraded taskId={} msg={}", taskId, ex.getMessage());
}
taskScopePayloadStorageService.deleteTaskScopePayloads(taskId, MODULE_TYPE);
}
@@ -133,7 +149,13 @@ public class QueryAsinTaskCacheService {
return result;
}
java.util.List<String> keys = missingIds.stream().map(this::buildTaskEntityKey).toList();
java.util.List<String> values = stringRedisTemplate.opsForValue().multiGet(keys);
java.util.List<String> values;
try {
values = stringRedisTemplate.opsForValue().multiGet(keys);
} catch (Exception ex) {
log.warn("[query-asin-cache] batch get task cache degraded taskIds={} msg={}", missingIds, ex.getMessage());
return result;
}
for (int i = 0; i < missingIds.size(); i++) {
Long taskId = missingIds.get(i);
String val = values != null && i < values.size() ? values.get(i) : null;
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinTaskService.java
+129 -18
View File
@@ -24,6 +24,10 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
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.TaskFileJobEntity;
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.ziniao.service.ZiniaoShopSwitchService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -56,6 +60,9 @@ public class QueryAsinTaskService {
private final ZiniaoShopSwitchService ziniaoShopSwitchService;
private final ObjectMapper objectMapper;
private final TaskPressureProperties taskPressureProperties;
private final TaskFileJobService taskFileJobService;
private final TaskResultItemService taskResultItemService;
private final TaskProgressSnapshotService taskProgressSnapshotService;
private FileTaskEntity loadTaskForExecution(Long taskId) {
Map<Long, FileTaskEntity> cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId));
@@ -185,8 +192,7 @@ public class QueryAsinTaskService {
batch.getMissingTaskIds().add(taskId);
continue;
}
List<QueryAsinResultItemVo> snapshots = buildSnapshotFromDb(task, taskRows);
batch.getItems().addAll(snapshots);
batch.getItems().addAll(buildProgressItems(task, taskRows));
}
return batch;
}
@@ -502,7 +508,11 @@ public class QueryAsinTaskService {
private Map<Long, Map<Long, QueryAsinResultItemVo>> buildSnapshotMap(Map<Long, FileTaskEntity> taskMap) {
Map<Long, Map<Long, QueryAsinResultItemVo>> out = new LinkedHashMap<>();
for (Map.Entry<Long, FileTaskEntity> entry : taskMap.entrySet()) {
out.put(entry.getKey(), indexSnapshotByResultId(parseTaskSnapshots(entry.getValue().getResultJson())));
List<QueryAsinResultItemVo> snapshots = taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, QueryAsinResultItemVo.class);
if (snapshots.isEmpty()) {
snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
}
out.put(entry.getKey(), indexSnapshotByResultId(snapshots));
}
return out;
}
@@ -519,9 +529,8 @@ public class QueryAsinTaskService {
item.setCreatedAt(entity.getCreatedAt());
item.setFinishedAt(task != null ? task.getFinishedAt() : item.getFinishedAt());
item.setOutputFilename(firstNonBlank(item.getOutputFilename(), entity.getResultFilename()));
item.setDownloadUrl(blank(entity.getResultFileUrl())
? item.getDownloadUrl()
: ossStorageService.generateFreshDownloadUrl(entity.getResultFileUrl()));
item.setDownloadUrl(null);
attachFileJobState(item, entity);
if (item.getCountryResults() == null) {
item.setCountryResults(new ArrayList<>());
}
@@ -531,6 +540,18 @@ public class QueryAsinTaskService {
return item;
}
private void attachFileJobState(QueryAsinResultItemVo item, FileResultEntity entity) {
TaskFileJobEntity job = taskFileJobService.findAssembleJob(entity.getTaskId(), MODULE_TYPE, entity.getId());
item.setFileReady(!blank(entity.getResultFileUrl()));
if (job == null) {
item.setFileStatus(Boolean.TRUE.equals(item.getFileReady()) ? "SUCCESS" : null);
return;
}
item.setFileJobId(job.getId());
item.setFileStatus(job.getStatus());
item.setFileError(job.getErrorMessage());
}
private List<QueryAsinTaskItemDto> dedupeItems(List<QueryAsinTaskItemDto> items) {
LinkedHashMap<String, QueryAsinTaskItemDto> map = new LinkedHashMap<>();
for (QueryAsinTaskItemDto item : items) {
@@ -575,6 +596,7 @@ public class QueryAsinTaskService {
try {
task.setRequestJson(objectMapper.writeValueAsString(requestItems));
task.setResultJson(objectMapper.writeValueAsString(snapshots));
syncSnapshotTables(task, snapshots);
fileTaskMapper.updateById(task);
} catch (Exception ex) {
throw new BusinessException("查询ASIN任务快照保存失败");
@@ -591,6 +613,27 @@ public class QueryAsinTaskService {
return list;
}
private List<QueryAsinResultItemVo> buildProgressItems(FileTaskEntity task, List<FileResultEntity> rows) {
List<QueryAsinResultItemVo> list = new ArrayList<>();
for (FileResultEntity row : rows) {
QueryAsinResultItemVo item = new QueryAsinResultItemVo();
item.setResultId(row.getId());
item.setTaskId(row.getTaskId());
item.setShopName(row.getSourceFilename());
item.setShopId(row.getSourceFileUrl());
item.setTaskStatus(task == null ? null : task.getStatus());
item.setSuccess(row.getSuccess() == null ? null : row.getSuccess() == 1);
item.setError(row.getErrorMessage());
item.setCreatedAt(row.getCreatedAt());
item.setFinishedAt(task == null ? null : task.getFinishedAt());
item.setOutputFilename(row.getResultFilename());
item.setDownloadUrl(null);
attachFileJobState(item, row);
list.add(item);
}
return list;
}
private Map<Long, QueryAsinResultItemVo> indexSnapshotByResultId(List<QueryAsinResultItemVo> snapshots) {
Map<Long, QueryAsinResultItemVo> map = new LinkedHashMap<>();
if (snapshots == null) {
@@ -773,27 +816,25 @@ public class QueryAsinTaskService {
.filter(item -> Boolean.TRUE.equals(item.getSuccess()))
.toList();
if (!successItems.isEmpty()) {
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "query-asin-result", String.valueOf(task.getId())));
String filename = safeFileStem("查询ASIN-" + task.getId()) + ".xlsx";
File xlsx = FileUtil.file(workRoot, filename);
try {
excelAssemblyService.writeWorkbook(xlsx, successItems);
String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
long fileSize = xlsx.length();
int rowCount = excelAssemblyService.countRows(successItems);
String filename = safeFileStem("query-asin-" + task.getId()) + ".xlsx";
int rowCount = excelAssemblyService.countRows(successItems);
FileResultEntity firstSuccessRow = null;
for (FileResultEntity row : rows) {
if (Integer.valueOf(1).equals(row.getSuccess())) {
row.setResultFilename(filename);
row.setResultFileUrl(objectKey);
row.setResultFileSize(fileSize);
row.setResultFileUrl(null);
row.setResultFileSize(0L);
row.setResultContentType(CONTENT_TYPE_XLSX);
row.setRowCount(rowCount);
fileResultMapper.updateById(row);
if (firstSuccessRow == null) {
firstSuccessRow = row;
}
}
}
if (firstSuccessRow != null) {
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, firstSuccessRow.getId(), "task:" + task.getId());
snapshots = buildSnapshotFromDb(task, rows);
} finally {
FileUtil.del(xlsx);
}
}
@@ -802,6 +843,48 @@ public class QueryAsinTaskService {
taskCacheService.deleteTaskCache(task.getId());
}
public void processResultFileJob(TaskFileJobEntity job) {
if (job == null || job.getTaskId() == null) {
throw new BusinessException("结果文件任务参数不完整");
}
FileTaskEntity task = fileTaskMapper.selectById(job.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
}
List<FileResultEntity> rows = listTaskRows(job.getTaskId());
List<QueryAsinResultItemVo> snapshots = taskResultItemService.listResultSnapshots(job.getTaskId(), MODULE_TYPE, QueryAsinResultItemVo.class);
if (snapshots.isEmpty()) {
snapshots = buildSnapshotFromDb(task, rows);
}
List<QueryAsinResultItemVo> successItems = snapshots.stream()
.filter(item -> Boolean.TRUE.equals(item.getSuccess()))
.toList();
if (successItems.isEmpty()) {
throw new BusinessException("没有可生成的查询ASIN结果");
}
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "query-asin-result", String.valueOf(task.getId())));
String filename = safeFileStem("query-asin-" + task.getId()) + ".xlsx";
File xlsx = FileUtil.file(workRoot, filename);
try {
excelAssemblyService.writeWorkbook(xlsx, successItems);
String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
long fileSize = xlsx.length();
int rowCount = excelAssemblyService.countRows(successItems);
for (FileResultEntity row : rows) {
if (Integer.valueOf(1).equals(row.getSuccess())) {
row.setResultFilename(filename);
row.setResultFileUrl(objectKey);
row.setResultFileSize(fileSize);
row.setResultContentType(CONTENT_TYPE_XLSX);
row.setRowCount(rowCount);
fileResultMapper.updateById(row);
}
}
} finally {
FileUtil.del(xlsx);
}
}
private void markResultSuccess(FileResultEntity row) {
row.setSuccess(1);
row.setErrorMessage(null);
@@ -838,11 +921,39 @@ public class QueryAsinTaskService {
private void persistSnapshotJson(FileTaskEntity task, List<QueryAsinResultItemVo> snapshots) {
try {
task.setResultJson(objectMapper.writeValueAsString(snapshots == null ? List.of() : snapshots));
syncSnapshotTables(task, snapshots);
} catch (Exception ex) {
throw new BusinessException("查询ASIN任务快照保存失败");
}
}
private void syncSnapshotTables(FileTaskEntity task, List<QueryAsinResultItemVo> snapshots) {
List<QueryAsinResultItemVo> safe = snapshots == null ? List.of() : snapshots;
taskResultItemService.replaceTaskSnapshots(task.getId(), MODULE_TYPE, safe, new TaskResultItemService.SnapshotKeyResolver() {
@Override
public Long resultId(Object snapshot) {
return ((QueryAsinResultItemVo) snapshot).getResultId();
}
@Override
public String scopeKey(Object snapshot) {
QueryAsinResultItemVo item = (QueryAsinResultItemVo) snapshot;
return firstNonBlank(item.getShopName(), "result:" + item.getResultId());
}
});
int successCount = 0;
int failedCount = 0;
for (QueryAsinResultItemVo item : safe) {
if (Boolean.TRUE.equals(item.getSuccess())) {
successCount++;
} else if (Boolean.FALSE.equals(item.getSuccess())) {
failedCount++;
}
}
taskProgressSnapshotService.save(task.getId(), MODULE_TYPE, firstNonBlank(task.getStatus(), "RUNNING"),
safe.size(), successCount, failedCount, null, task.getErrorMessage(), null);
}
private List<QueryAsinCountryResultDto> copyCountryResults(List<QueryAsinCountryResultDto> results) {
List<QueryAsinCountryResultDto> copy = new ArrayList<>();
if (results == null) {