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

菜单修改优化

Browse Source
This commit is contained in:
supernijia
2026-07-28 13:40:19 +08:00
parent 4ddb8b47b0
commit 7c3c9e53a0
90 changed files with 9388 additions and 1545 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/publish/controller/PublishController.java
+2 -2
View File
@@ -106,8 +106,8 @@ public class PublishController {
@PostMapping("/tasks/{taskId}/result")
@Operation(
summary = "Python 按文件回传当前店铺完整数据",
description = "请求只需 taskId 和 files，user_id 为兼容旧客户端的可选字段；后端按 taskId 反查任务所属用户，传入 user_id 时会校验归属。成功回传必须包含该文件全部原始行，可使用 rows 数组，或在 rows 为空时使用 countries 按国家分组；行数少于原始数据、包含 null 行或八列全空白对象时会被拒绝且不会覆盖已解析数据。error 非空时将该文件标记为 FAILED。全部文件进入终态后，只要至少一个文件成功就异步组装结果；全部失败则不生成结果文件。")
summary = "Python 按文件分片回传上架结果",
description = "请求只需 taskId 和 files，user_id 为兼容旧客户端的可选字段；后端始终按 taskId 处理任务数据。每个文件通过 chunk_index/chunk_total 声明从 1 开始的分片，不传时按 1/1 兼容旧客户端。正常分片先写入 RustFS，同一文件的全部分片到齐后才按序合并、校验总行数并覆盖解析数据；重复分片内容相同则幂等接受，内容不同或 chunk_total 不一致则拒绝。error 非空时直接将文件标记为 FAILED。全部文件进入终态后，只要至少一个文件成功就异步组装结果；全部失败则不生成结果文件。")
public ApiResponse<Void> submitResult(
@Parameter(description = "上架任务 ID", required = true, example = "9001")
@PathVariable Long taskId,
backend-java/src/main/java/com/nanri/aiimage/modules/publish/model/dto/PublishResultFileDto.java
+17 -6
View File
@@ -1,6 +1,7 @@
package com.nanri.aiimage.modules.publish.model.dto;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.Valid;
import lombok.Data;
@@ -11,29 +12,39 @@ import java.util.List;
import java.util.Map;
@Data
@Schema(description = "单个文件的 Python 处理结果。fileId、fileKey、sourceFilename 至少提供一个用于定位文件。成功时必须提交完整 rows 或 countries；失败时填写 error。同一次请求中，不同定位字段指向同一任务文件也视为重复提交。")
@Schema(description = "单个文件的一次 Python 分片回传。fileId/file_key/source_filename 三者可用于定位文件，rows 或 countries 提供当前分片数据，error 用于返回失败。")
public class PublishResultFileDto {
@JsonAlias("file_id")
@Schema(description = "任务内文件 ID；同时兼容 file_id，优先使用该字段定位", example = "9101")
private Long fileId;
@JsonAlias("file_key")
@Schema(description = "源文件标识；同时兼容 file_key，可在 fileId 缺失时定位文件", example = "uploads/20260724/uuid/郭亚庆.xlsx")
@Schema(description = "任务内文件标识；兼容 file_key，fileId 缺失时可通过该字段定位", example = "uploads/20260724/uuid/郭亚庆.xlsx")
private String fileKey;
@JsonAlias("source_filename")
@Schema(description = "原始文件名；同时兼容 source_filename，可在 fileId/fileKey 缺失时定位文件", example = "郭亚庆.xlsx")
@Schema(description = "原始文件名；兼容 source_filename，fileId/fileKey 都缺失时可定位文件", example = "郭亚庆.xlsx")
private String sourceFilename;
@Schema(description = "文件级失败原因。非空时文件标记为 FAILED，rows/countries 不会覆盖原始数据", example = "打开店铺失败")
@JsonProperty("chunk_index")
@JsonAlias("chunkIndex")
@Schema(description = "分片序号，从 1 开始；不传时与 chunk_total 同时未传表示 1/1，兼容旧版。", example = "1", requiredMode = Schema.RequiredMode.NOT_REQUIRED)
private Integer chunkIndex;
@JsonProperty("chunk_total")
@JsonAlias("chunkTotal")
@Schema(description = "该文件的分片总数；不传时按 1 处理。分片先暂存到 RustFS，全部到齐后再按序合并。", example = "36", requiredMode = Schema.RequiredMode.NOT_REQUIRED)
private Integer chunkTotal;
@Schema(description = "文件处理结果；FAILED 时通过 error 携带错误信息，rows/countries 可为空", example = "打开文件失败")
private String error;
@Valid
@JsonAlias({"items", "data"})
@Schema(description = "当前店铺的完整结果行；同时兼容 items/data。成功时行数不能少于原始 Excel 数据行数，且不能包含 null 行或八列全空白对象")
@Schema(description = "当前分片的结果行，可为一条或多条；rows 中每条均不能为空对象")
private List<PublishRowDto> rows = new ArrayList<>();
@Valid
@Schema(description = "按国家名称或代码分组的完整结果，可替代 rows。仅当 rows 为空时读取；缺少国家字段的行会使用当前 Map key；分组内不能包含 null 行或八列全空白对象")
@Schema(description = "当前分片按国家分组的结果；rows 为空时可使用 countries，key 为国家或店铺名")
private Map<String, List<PublishRowDto>> countries = new LinkedHashMap<>();
}
backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java
+384 -11
View File
@@ -2,8 +2,10 @@ package com.nanri.aiimage.modules.publish.service;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.crypto.digest.DigestUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
@@ -31,18 +33,24 @@ import com.nanri.aiimage.modules.publish.model.vo.PublishTaskDetailVo;
import com.nanri.aiimage.modules.publish.model.vo.PublishTaskVo;
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.dto.TaskHeartbeatRequest;
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.TransientPayloadStorageService;
import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -84,8 +92,11 @@ public class PublishTaskService {
private final PublishItemMapper publishItemMapper;
private final FileTaskMapper fileTaskMapper;
private final FileResultMapper fileResultMapper;
private final TaskChunkMapper taskChunkMapper;
private final TaskScopeStateMapper taskScopeStateMapper;
private final TaskFileJobService taskFileJobService;
private final TaskDistributedLockService taskDistributedLockService;
private final TransientPayloadStorageService transientPayloadStorageService;
private final OssStorageService ossStorageService;
private final ObjectMapper objectMapper;
private final TransactionTemplate transactionTemplate;
@@ -239,7 +250,23 @@ public class PublishTaskService {
if (lock == null) {
throw new BusinessException("task lock is busy");
}
transactionTemplate.executeWithoutResult(status -> submitResultLocked(taskId, request));
List<String> storedPayloads = new ArrayList<>();
boolean[] cleanupAfterCommit = {false};
try {
transactionTemplate.executeWithoutResult(
status -> cleanupAfterCommit[0] = submitResultLocked(taskId, request, storedPayloads));
} catch (RuntimeException ex) {
deleteRolledBackPayloads(storedPayloads);
throw ex;
}
if (cleanupAfterCommit[0]) {
try {
deleteTransientResultChunks(taskId);
} catch (Exception ex) {
log.warn("[publish] failed-task chunk cleanup failed taskId={} msg={}",
taskId, safeMessage(ex));
}
}
}
}
@@ -455,7 +482,7 @@ public class PublishTaskService {
@Transactional
public void deleteTask(Long taskId, Long userId) {
FileTaskEntity task = requireTask(taskId, userId);
FileTaskEntity task = requireTaskForDeletion(taskId, userId);
List<FileResultEntity> results = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
@@ -470,6 +497,7 @@ public class PublishTaskService {
}
}
taskFileJobService.deleteTaskJobs(taskId, MODULE_TYPE);
deleteTransientResultChunks(taskId);
publishItemMapper.delete(new LambdaQueryWrapper<PublishItemEntity>()
.eq(PublishItemEntity::getTaskId, taskId));
publishFileMapper.delete(new LambdaQueryWrapper<PublishFileEntity>()
@@ -486,8 +514,49 @@ public class PublishTaskService {
if (result == null || !MODULE_TYPE.equals(result.getModuleType())) {
throw new BusinessException("结果不存在");
}
FileTaskEntity task = requireTask(result.getTaskId(), userId);
deleteTask(task.getId(), userId);
deleteTask(result.getTaskId(), userId);
}
public void cleanupResultFileJob(TaskFileJobEntity job) {
if (job == null || job.getTaskId() == null) {
return;
}
deleteTransientResultChunks(job.getTaskId());
}
private void deleteTransientResultChunks(Long taskId) {
if (taskId == null || taskId <= 0) {
return;
}
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.select(TaskChunkEntity::getPayloadJson)
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE));
if (chunks != null) {
for (TaskChunkEntity chunk : chunks) {
transientPayloadStorageService.deletePayloadIfPresent(chunk.getPayloadJson());
}
}
taskChunkMapper.delete(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE));
taskScopeStateMapper.delete(new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE));
}
private void deleteRolledBackPayloads(List<String> storedPayloads) {
if (storedPayloads == null || storedPayloads.isEmpty()) {
return;
}
for (String storedPayload : storedPayloads) {
try {
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
} catch (Exception ex) {
log.warn("[publish] rolled-back RustFS payload cleanup failed pointer={} msg={}",
transientPayloadStorageService.extractPointer(storedPayload), safeMessage(ex));
}
}
}
private PreparedFile prepareFile(PublishSourceFileDto source) {
@@ -538,6 +607,26 @@ public class PublishTaskService {
}
LocalDateTime now = LocalDateTime.now();
String error = "任务心跳超时";
List<PublishFileEntity> files = listTaskFiles(taskId);
for (PublishFileEntity file : files) {
if (!STATUS_PENDING.equals(file.getStatus()) && !STATUS_RUNNING.equals(file.getStatus())) {
continue;
}
List<PublishRowDto> recoveredRows = loadReceivedResultRows(taskId, file.getId());
if (recoveredRows.isEmpty()) {
continue;
}
replaceRows(taskId, file.getId(), recoveredRows);
file.setStatus(STATUS_SUCCESS);
file.setTotalRows(recoveredRows.size());
file.setProcessedRows(recoveredRows.size());
file.setErrorMessage("任务心跳超时，已保留 " + recoveredRows.size() + " 行已回传数据");
file.setUpdatedAt(now);
file.setFinishedAt(now);
publishFileMapper.updateById(file);
log.info("[publish] recovered partial result from stale task taskId={} fileId={} rows={}",
taskId, file.getId(), recoveredRows.size());
}
publishFileMapper.update(null, new LambdaUpdateWrapper<PublishFileEntity>()
.eq(PublishFileEntity::getTaskId, taskId)
.in(PublishFileEntity::getStatus, List.of(STATUS_PENDING, STATUS_RUNNING))
@@ -545,7 +634,7 @@ public class PublishTaskService {
.set(PublishFileEntity::getErrorMessage, error)
.set(PublishFileEntity::getUpdatedAt, now)
.set(PublishFileEntity::getFinishedAt, now));
List<PublishFileEntity> files = listTaskFiles(taskId);
files = listTaskFiles(taskId);
int successCount = (int) files.stream().filter(file -> STATUS_SUCCESS.equals(file.getStatus())).count();
task.setSuccessFileCount(successCount);
task.setFailedFileCount(Math.max(0, files.size() - successCount));
@@ -615,10 +704,12 @@ public class PublishTaskService {
return new PersistedTask(task, result, savedFiles);
}
private void submitResultLocked(Long taskId, PublishSubmitResultRequest request) {
private boolean submitResultLocked(Long taskId,
PublishSubmitResultRequest request,
List<String> storedPayloads) {
FileTaskEntity task = requireTask(taskId, request.getUserId());
if (STATUS_SUCCESS.equals(task.getStatus())) {
return;
return false;
}
if (STATUS_FAILED.equals(task.getStatus())) {
throw new BusinessException("任务已失败，拒绝继续回传");
@@ -629,6 +720,9 @@ public class PublishTaskService {
Set<Long> submittedFileIds = new LinkedHashSet<>();
for (PublishResultFileDto incoming : request.getFiles()) {
if (incoming == null) {
throw new BusinessException("files 不能包含空对象");
}
PublishFileEntity file = findCallbackFile(taskId, incoming);
if (!submittedFileIds.add(file.getId())) {
throw new BusinessException("同一文件不能在一次请求中重复提交");
@@ -641,7 +735,16 @@ public class PublishTaskService {
file.setProcessedRows(0);
file.setErrorMessage(incoming.getError().trim());
} else {
List<PublishRowDto> rows = flattenRows(incoming);
ResultChunkReceipt receipt = persistResultChunk(taskId, file, incoming, storedPayloads);
if (!receipt.completed()) {
file.setStatus(STATUS_RUNNING);
file.setErrorMessage(null);
file.setUpdatedAt(LocalDateTime.now());
file.setFinishedAt(null);
publishFileMapper.updateById(file);
continue;
}
List<PublishRowDto> rows = loadCompleteResultRows(taskId, receipt);
validateCompleteResultRows(taskId, file.getId(), rows);
replaceRows(taskId, file.getId(), rows);
file.setStatus(STATUS_SUCCESS);
@@ -666,11 +769,11 @@ public class PublishTaskService {
if (terminalCount < files.size()) {
task.setStatus(STATUS_RUNNING);
fileTaskMapper.updateById(task);
return;
return false;
}
if (successCount <= 0) {
markTaskAndResultFailed(task, result, "全部文件处理失败");
return;
return true;
}
task.setStatus(STATUS_RUNNING);
@@ -678,6 +781,7 @@ public class PublishTaskService {
task.setFinishedAt(null);
fileTaskMapper.updateById(task);
taskFileJobService.enqueueAssembleResult(taskId, MODULE_TYPE, result.getId(), ownerScopeKey(taskId));
return false;
}
private List<PublishTaskDetailVo> loadTaskDetails(List<FileTaskEntity> tasks) {
@@ -805,6 +909,20 @@ public class PublishTaskService {
}
private FileTaskEntity requireTask(Long taskId, Long userId) {
FileTaskEntity task = requireTaskRecord(taskId, userId);
ensureTaskOwnedByCurrentInstance(task, "access publish task");
return task;
}
private FileTaskEntity requireTaskForDeletion(Long taskId, Long userId) {
FileTaskEntity task = requireTaskRecord(taskId, userId);
if (!isTerminal(task.getStatus())) {
ensureTaskOwnedByCurrentInstance(task, "delete publish task");
}
return task;
}
private FileTaskEntity requireTaskRecord(Long taskId, Long userId) {
if (taskId == null || taskId <= 0) {
throw new BusinessException("taskId 不合法");
}
@@ -813,7 +931,6 @@ public class PublishTaskService {
|| (userId != null && !userId.equals(task.getUserId()))) {
throw new BusinessException("任务不存在");
}
ensureTaskOwnedByCurrentInstance(task, "access publish task");
return task;
}
@@ -888,6 +1005,257 @@ public class PublishTaskService {
return file;
}
private ResultChunkReceipt persistResultChunk(Long taskId,
PublishFileEntity file,
PublishResultFileDto incoming,
List<String> storedPayloads) {
int chunkIndex = incoming.getChunkIndex() == null ? 1 : incoming.getChunkIndex();
int chunkTotal = incoming.getChunkTotal() == null ? 1 : incoming.getChunkTotal();
validateChunkMetadata(chunkIndex, chunkTotal);
String scopeKey = "file:" + file.getId();
String scopeHash = DigestUtil.sha256Hex(scopeKey);
TaskScopeStateEntity scope = findResultScope(taskId, scopeHash);
validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal);
List<PublishRowDto> rows = flattenRows(incoming);
String payloadJson = writeJson(rows, "序列化上架结果分片失败");
String payloadHash = DigestUtil.sha256Hex(payloadJson);
TaskChunkEntity existing = findResultChunk(taskId, scopeHash, chunkIndex);
if (existing != null) {
validateExistingChunk(existing, chunkTotal, payloadHash);
int receivedChunkCount = countResultChunks(taskId, scopeHash);
persistResultScope(taskId, scopeKey, scopeHash, chunkTotal, receivedChunkCount);
return new ResultChunkReceipt(scopeHash, chunkTotal,
receivedChunkCount >= chunkTotal);
}
ensureRustfsPayloadStorageEnabled();
String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(
MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
requireRustfsPayload(storedPayload, "上架结果分片必须写入 RustFS");
TaskChunkEntity chunk = new TaskChunkEntity();
chunk.setTaskId(taskId);
chunk.setModuleType(MODULE_TYPE);
chunk.setScopeKey(scopeKey);
chunk.setScopeHash(scopeHash);
chunk.setChunkIndex(chunkIndex);
chunk.setChunkTotal(chunkTotal);
chunk.setPayloadJson(storedPayload);
chunk.setPayloadHash(payloadHash);
chunk.setCreatedAt(LocalDateTime.now());
chunk.setUpdatedAt(LocalDateTime.now());
try {
taskChunkMapper.insert(chunk);
storedPayloads.add(storedPayload);
} catch (DuplicateKeyException ex) {
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
TaskChunkEntity winner = findResultChunk(taskId, scopeHash, chunkIndex);
if (winner == null) {
throw new BusinessException("上架结果分片并发写入失败，请重试");
}
validateExistingChunk(winner, chunkTotal, payloadHash);
} catch (RuntimeException ex) {
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
throw ex;
}
int receivedChunkCount = countResultChunks(taskId, scopeHash);
persistResultScope(taskId, scopeKey, scopeHash, chunkTotal, receivedChunkCount);
log.info("[publish] result chunk received taskId={} fileId={} chunk={}/{} received={}",
taskId, file.getId(), chunkIndex, chunkTotal, receivedChunkCount);
return new ResultChunkReceipt(scopeHash, chunkTotal,
receivedChunkCount >= chunkTotal);
}
private void validateChunkMetadata(int chunkIndex, int chunkTotal) {
if (chunkIndex <= 0) {
throw new BusinessException("chunk_index 必须从 1 开始");
}
if (chunkTotal <= 0) {
throw new BusinessException("chunk_total 必须大于 0");
}
if (chunkIndex > chunkTotal) {
throw new BusinessException("chunk_index 不能大于 chunk_total");
}
}
private TaskScopeStateEntity findResultScope(Long taskId, String scopeHash) {
return taskScopeStateMapper.selectOne(new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
.eq(TaskScopeStateEntity::getScopeHash, scopeHash)
.last("limit 1"));
}
private TaskChunkEntity findResultChunk(Long taskId, String scopeHash, int chunkIndex) {
return taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.eq(TaskChunkEntity::getScopeHash, scopeHash)
.eq(TaskChunkEntity::getChunkIndex, chunkIndex)
.last("limit 1"));
}
private void validateExistingChunk(TaskChunkEntity existing, int chunkTotal, String payloadHash) {
validateChunkTotal(existing.getChunkTotal(), chunkTotal);
if (!Objects.equals(existing.getPayloadHash(), payloadHash)) {
throw new BusinessException("同一 chunk_index 已回传不同内容，拒绝覆盖");
}
}
private void validateChunkTotal(Integer existingChunkTotal, int chunkTotal) {
if (existingChunkTotal != null && existingChunkTotal > 0 && existingChunkTotal != chunkTotal) {
throw new BusinessException("同一文件的 chunk_total 必须保持一致");
}
}
private int countResultChunks(Long taskId, String scopeHash) {
Long count = taskChunkMapper.selectCount(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.eq(TaskChunkEntity::getScopeHash, scopeHash));
return count == null ? 0 : count.intValue();
}
private void persistResultScope(Long taskId,
String scopeKey,
String scopeHash,
int chunkTotal,
int receivedChunkCount) {
TaskScopeStateEntity scope = findResultScope(taskId, scopeHash);
validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal);
LocalDateTime now = LocalDateTime.now();
if (scope == null) {
scope = new TaskScopeStateEntity();
scope.setTaskId(taskId);
scope.setModuleType(MODULE_TYPE);
scope.setScopeKey(scopeKey);
scope.setScopeHash(scopeHash);
scope.setCreatedAt(now);
}
boolean completed = receivedChunkCount >= chunkTotal;
scope.setChunkTotal(chunkTotal);
scope.setReceivedChunkCount(receivedChunkCount);
scope.setCompleted(completed ? 1 : 0);
scope.setLastChunkAt(now);
scope.setLastError(null);
scope.setStateJson(completed ? "{\"phase\":\"COMPLETE\"}" : "{\"phase\":\"RECEIVING\"}");
scope.setUpdatedAt(now);
if (scope.getId() != null) {
taskScopeStateMapper.updateById(scope);
return;
}
try {
taskScopeStateMapper.insert(scope);
} catch (DuplicateKeyException ex) {
TaskScopeStateEntity winner = findResultScope(taskId, scopeHash);
if (winner == null) {
throw new BusinessException("上架结果分片状态写入失败，请重试");
}
validateChunkTotal(winner.getChunkTotal(), chunkTotal);
winner.setChunkTotal(chunkTotal);
winner.setReceivedChunkCount(receivedChunkCount);
winner.setCompleted(completed ? 1 : 0);
winner.setLastChunkAt(now);
winner.setLastError(null);
winner.setStateJson(scope.getStateJson());
winner.setUpdatedAt(now);
taskScopeStateMapper.updateById(winner);
}
}
private List<PublishRowDto> loadCompleteResultRows(Long taskId, ResultChunkReceipt receipt) {
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.eq(TaskChunkEntity::getScopeHash, receipt.scopeHash())
.orderByAsc(TaskChunkEntity::getChunkIndex));
if (chunks == null || chunks.size() != receipt.chunkTotal()) {
throw new BusinessException("上架结果分片尚未完整，暂不能合并");
}
List<PublishRowDto> rows = new ArrayList<>();
TypeReference<List<PublishRowDto>> listType = new TypeReference<>() {
};
for (int i = 0; i < chunks.size(); i++) {
TaskChunkEntity chunk = chunks.get(i);
int expectedIndex = i + 1;
if (!Objects.equals(chunk.getChunkIndex(), expectedIndex)) {
throw new BusinessException("上架结果缺少第 " + expectedIndex + " 个分片");
}
validateChunkTotal(chunk.getChunkTotal(), receipt.chunkTotal());
List<PublishRowDto> chunkRows = readResultChunkRows(chunk, listType);
for (PublishRowDto row : chunkRows) {
rows.add(copyRequiredRow(row));
}
}
return rows;
}
private List<PublishRowDto> loadReceivedResultRows(Long taskId, Long fileId) {
String scopeHash = DigestUtil.sha256Hex("file:" + fileId);
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.eq(TaskChunkEntity::getScopeHash, scopeHash)
.orderByAsc(TaskChunkEntity::getChunkIndex));
if (chunks == null || chunks.isEmpty()) {
return List.of();
}
List<PublishRowDto> rows = new ArrayList<>();
TypeReference<List<PublishRowDto>> listType = new TypeReference<>() {
};
for (TaskChunkEntity chunk : chunks) {
try {
for (PublishRowDto row : readResultChunkRows(chunk, listType)) {
rows.add(copyRequiredRow(row));
}
} catch (Exception ex) {
log.warn("[publish] skip unreadable stale result chunk taskId={} fileId={} chunkIndex={} msg={}",
taskId, fileId, chunk.getChunkIndex(), safeMessage(ex));
}
}
return rows;
}
private List<PublishRowDto> readResultChunkRows(TaskChunkEntity chunk,
TypeReference<List<PublishRowDto>> listType) {
int chunkIndex = chunk.getChunkIndex() == null ? 0 : chunk.getChunkIndex();
String pointer = transientPayloadStorageService.extractPointer(chunk.getPayloadJson());
if (pointer == null || !pointer.startsWith("rustfs:")) {
throw new BusinessException("上架结果分片不是 RustFS 数据，拒绝合并");
}
try {
String payloadJson = transientPayloadStorageService.resolvePayload(
chunk.getPayloadJson(), "读取上架结果分片失败");
List<PublishRowDto> rows = objectMapper.readValue(payloadJson, listType);
return rows == null ? List.of() : rows;
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
throw new BusinessException("读取上架结果第 " + chunkIndex + " 个分片失败: "
+ safeMessage(ex));
}
}
private void ensureRustfsPayloadStorageEnabled() {
if (!transientPayloadStorageService.isSharedWriteEnabled()) {
throw new BusinessException("RustFS 未配置，上架结果分片暂不可接收");
}
}
private void requireRustfsPayload(String storedPayload, String message) {
String pointer = transientPayloadStorageService.extractPointer(storedPayload);
if (pointer != null && pointer.startsWith("rustfs:")) {
return;
}
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
throw new BusinessException(message);
}
private List<PublishRowDto> flattenRows(PublishResultFileDto incoming) {
if (incoming.getRows() != null && !incoming.getRows().isEmpty()) {
return incoming.getRows().stream().map(this::copyRequiredRow).toList();
@@ -1266,4 +1634,9 @@ public class PublishTaskService {
private record TaskOptions(String publishCountry, List<String> syncCountries) {
}
private record ResultChunkReceipt(String scopeHash,
int chunkTotal,
boolean completed) {
}
}