提交打包
This commit is contained in:
super
+159
-148
@@ -258,16 +258,18 @@ public class PriceTrackTaskService {
|
||||
@Transactional
|
||||
public void deleteTask(Long taskId, Long userId) {
|
||||
if (userId == null || userId <= 0) throw new BusinessException("user_id 不合法");
|
||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
|
||||
throw new BusinessException("任务不存在");
|
||||
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
|
||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
|
||||
throw new BusinessException("任务不存在");
|
||||
}
|
||||
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
|
||||
fileTaskMapper.deleteById(taskId);
|
||||
priceTrackTaskCacheService.deleteTaskCache(taskId);
|
||||
priceTrackLoopRunService.syncLoopRunAfterChildRemoved(taskId);
|
||||
}
|
||||
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
|
||||
fileTaskMapper.deleteById(taskId);
|
||||
priceTrackTaskCacheService.deleteTaskCache(taskId);
|
||||
priceTrackLoopRunService.syncLoopRunAfterChildRemoved(taskId);
|
||||
}
|
||||
|
||||
@Transactional
|
||||
@@ -458,154 +460,163 @@ public class PriceTrackTaskService {
|
||||
if (request == null || request.getShops() == null || request.getShops().isEmpty()) {
|
||||
throw new BusinessException("shops 不能为空");
|
||||
}
|
||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
throw new BusinessException("任务不存在");
|
||||
try (TaskDistributedLockService.LockHandle ignored = acquireTaskLockOrThrow(taskId)) {
|
||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
throw new BusinessException("任务不存在");
|
||||
}
|
||||
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
|
||||
log.warn("[price-track] submitResult rejected taskId={} status={}", taskId, task.getStatus());
|
||||
throw new BusinessException(40901, "任务已结束，拒绝重复提交");
|
||||
}
|
||||
priceTrackTaskCacheService.touchTaskHeartbeat(taskId);
|
||||
List<FileResultEntity> resultRows = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
Map<String, PriceTrackSubmitResultRequest.ShopResult> payloadByShop = new LinkedHashMap<>();
|
||||
for (PriceTrackSubmitResultRequest.ShopResult sr : request.getShops()) {
|
||||
if (sr == null) continue;
|
||||
String key = ziniaoShopSwitchService.normalizeShopName(sr.getShopName());
|
||||
if (!key.isBlank()) payloadByShop.put(key, sr);
|
||||
}
|
||||
log.info("[price-track] submitResult received taskId={} payloadShopCount={} payloadKeys={}",
|
||||
taskId, payloadByShop.size(), payloadByShop.keySet());
|
||||
boolean changed = false;
|
||||
for (FileResultEntity fr : resultRows) {
|
||||
String shopKey = fr.getSourceFilename();
|
||||
PriceTrackSubmitResultRequest.ShopResult payload = shopKey == null ? null : payloadByShop.get(shopKey);
|
||||
if (payload == null) {
|
||||
continue;
|
||||
}
|
||||
changed = true;
|
||||
if (payload.getError() != null && !payload.getError().isBlank()) {
|
||||
markResultFailed(fr, payload.getError());
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
continue;
|
||||
}
|
||||
if (Boolean.FALSE.equals(payload.getSuccess())) {
|
||||
markResultFailed(fr, "shop processing failed");
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
continue;
|
||||
}
|
||||
handleSkipAsinDeletionSignals(shopKey, payload);
|
||||
PriceTrackSubmitResultRequest.ShopResult merged = mergeShopPayload(taskId, shopKey, payload);
|
||||
if (!Boolean.TRUE.equals(merged.getSuccess())) {
|
||||
continue;
|
||||
}
|
||||
if (countPayloadRows(merged) <= 0) {
|
||||
markResultFailed(fr, "no usable price-track rows received");
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
enqueueResultFileAssembly(fr, shopKey, merged);
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
} catch (Exception ex) {
|
||||
markResultFailed(fr, ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage());
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
}
|
||||
}
|
||||
if (!changed) {
|
||||
return;
|
||||
}
|
||||
List<FileResultEntity> latest = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
updateTaskStatusFromLatestRows(task, latest);
|
||||
}
|
||||
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
|
||||
log.warn("[price-track] submitResult rejected taskId={} status={}", taskId, task.getStatus());
|
||||
throw new BusinessException(40901, "任务已结束，拒绝重复提交");
|
||||
}
|
||||
priceTrackTaskCacheService.touchTaskHeartbeat(taskId);
|
||||
List<FileResultEntity> resultRows = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
Map<String, PriceTrackSubmitResultRequest.ShopResult> payloadByShop = new LinkedHashMap<>();
|
||||
for (PriceTrackSubmitResultRequest.ShopResult sr : request.getShops()) {
|
||||
if (sr == null) continue;
|
||||
String key = ziniaoShopSwitchService.normalizeShopName(sr.getShopName());
|
||||
if (!key.isBlank()) payloadByShop.put(key, sr);
|
||||
}
|
||||
log.info("[price-track] submitResult received taskId={} payloadShopCount={} payloadKeys={}",
|
||||
taskId, payloadByShop.size(), payloadByShop.keySet());
|
||||
boolean changed = false;
|
||||
for (FileResultEntity fr : resultRows) {
|
||||
String shopKey = fr.getSourceFilename();
|
||||
PriceTrackSubmitResultRequest.ShopResult payload = shopKey == null ? null : payloadByShop.get(shopKey);
|
||||
if (payload == null) {
|
||||
continue;
|
||||
}
|
||||
changed = true;
|
||||
if (payload.getError() != null && !payload.getError().isBlank()) {
|
||||
markResultFailed(fr, payload.getError());
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
continue;
|
||||
}
|
||||
if (Boolean.FALSE.equals(payload.getSuccess())) {
|
||||
markResultFailed(fr, "shop processing failed");
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
continue;
|
||||
}
|
||||
handleSkipAsinDeletionSignals(shopKey, payload);
|
||||
PriceTrackSubmitResultRequest.ShopResult merged = mergeShopPayload(taskId, shopKey, payload);
|
||||
if (!Boolean.TRUE.equals(merged.getSuccess())) {
|
||||
continue;
|
||||
}
|
||||
if (countPayloadRows(merged) <= 0) {
|
||||
markResultFailed(fr, "no usable price-track rows received");
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
enqueueResultFileAssembly(fr, shopKey, merged);
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
} catch (Exception ex) {
|
||||
markResultFailed(fr, ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage());
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
}
|
||||
}
|
||||
if (!changed) {
|
||||
return;
|
||||
}
|
||||
List<FileResultEntity> latest = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
updateTaskStatusFromLatestRows(task, latest);
|
||||
}
|
||||
|
||||
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) {
|
||||
if (taskId == null || taskId <= 0) return false;
|
||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) return false;
|
||||
if (!"RUNNING".equals(task.getStatus())) {
|
||||
log.warn("[price-track] stale finalize skipped taskId={} status={} fromCompensation={}",
|
||||
taskId, task.getStatus(), fromCompensation);
|
||||
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
|
||||
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId);
|
||||
if (lockHandle == null) {
|
||||
log.info("[price-track] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
|
||||
return false;
|
||||
}
|
||||
List<FileResultEntity> latest = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
if (latest.isEmpty()) {
|
||||
task.setStatus("FAILED");
|
||||
task.setErrorMessage("task details missing, cannot finalize automatically");
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
task.setFinishedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
|
||||
try (lockHandle) {
|
||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) return false;
|
||||
if (!"RUNNING".equals(task.getStatus())) {
|
||||
log.warn("[price-track] stale finalize skipped taskId={} status={} fromCompensation={}",
|
||||
taskId, task.getStatus(), fromCompensation);
|
||||
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
|
||||
return false;
|
||||
}
|
||||
List<FileResultEntity> latest = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
if (latest.isEmpty()) {
|
||||
task.setStatus("FAILED");
|
||||
task.setErrorMessage("task details missing, cannot finalize automatically");
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
task.setFinishedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
cleanupTaskCacheIfTerminal(taskId, task.getStatus());
|
||||
return true;
|
||||
}
|
||||
Map<String, PriceTrackSubmitResultRequest.ShopResult> cachedPayloadByShop =
|
||||
priceTrackTaskCacheService.getAllShopMergedPayload(taskId);
|
||||
boolean changed = false;
|
||||
for (FileResultEntity fr : latest) {
|
||||
boolean success = fr.getSuccess() != null && fr.getSuccess() == 1;
|
||||
boolean failed = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank();
|
||||
if (success || failed) continue;
|
||||
String shopKey = fr.getSourceFilename();
|
||||
PriceTrackSubmitResultRequest.ShopResult cachedPayload = shopKey == null ? null : cachedPayloadByShop.get(shopKey);
|
||||
if (cachedPayload == null) {
|
||||
markResultFailed(fr, "Python interrupted before this shop finished uploading results");
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize failed without cache taskId={} shop={} fromCompensation={}",
|
||||
taskId, shopKey, fromCompensation);
|
||||
continue;
|
||||
}
|
||||
if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) {
|
||||
markResultFailed(fr, cachedPayload.getError());
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
continue;
|
||||
}
|
||||
if (countPayloadRows(cachedPayload) <= 0) {
|
||||
markResultFailed(fr, "no usable price-track rows received before interruption");
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize finished without data taskId={} shop={} fromCompensation={}",
|
||||
taskId, shopKey, fromCompensation);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
enqueueResultFileAssembly(fr, shopKey, cachedPayload);
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={} rowCount={}",
|
||||
taskId, shopKey, fromCompensation, Boolean.TRUE.equals(cachedPayload.getSuccess()), countPayloadRows(cachedPayload));
|
||||
} catch (Exception ex) {
|
||||
String message = ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage();
|
||||
markResultFailed(fr, message);
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize assemble failed taskId={} shop={} msg={}",
|
||||
taskId, shopKey, message);
|
||||
}
|
||||
}
|
||||
if (!changed && cachedPayloadByShop.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
List<FileResultEntity> refreshed = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
updateTaskStatusFromLatestRows(task, refreshed);
|
||||
return true;
|
||||
}
|
||||
Map<String, PriceTrackSubmitResultRequest.ShopResult> cachedPayloadByShop =
|
||||
priceTrackTaskCacheService.getAllShopMergedPayload(taskId);
|
||||
boolean changed = false;
|
||||
for (FileResultEntity fr : latest) {
|
||||
boolean success = fr.getSuccess() != null && fr.getSuccess() == 1;
|
||||
boolean failed = fr.getErrorMessage() != null && !fr.getErrorMessage().isBlank();
|
||||
if (success || failed) continue;
|
||||
String shopKey = fr.getSourceFilename();
|
||||
PriceTrackSubmitResultRequest.ShopResult cachedPayload = shopKey == null ? null : cachedPayloadByShop.get(shopKey);
|
||||
if (cachedPayload == null) {
|
||||
markResultFailed(fr, "Python interrupted before this shop finished uploading results");
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize failed without cache taskId={} shop={} fromCompensation={}",
|
||||
taskId, shopKey, fromCompensation);
|
||||
continue;
|
||||
}
|
||||
if (cachedPayload.getError() != null && !cachedPayload.getError().isBlank()) {
|
||||
markResultFailed(fr, cachedPayload.getError());
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
continue;
|
||||
}
|
||||
if (countPayloadRows(cachedPayload) <= 0) {
|
||||
markResultFailed(fr, "no usable price-track rows received before interruption");
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize finished without data taskId={} shop={} fromCompensation={}",
|
||||
taskId, shopKey, fromCompensation);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
enqueueResultFileAssembly(fr, shopKey, cachedPayload);
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize assembled taskId={} shop={} fromCompensation={} completedFlag={} rowCount={}",
|
||||
taskId, shopKey, fromCompensation, Boolean.TRUE.equals(cachedPayload.getSuccess()), countPayloadRows(cachedPayload));
|
||||
} catch (Exception ex) {
|
||||
String message = ex.getMessage() == null ? "assemble price-track result failed" : ex.getMessage();
|
||||
markResultFailed(fr, message);
|
||||
priceTrackTaskCacheService.removeShopMergedPayload(taskId, shopKey);
|
||||
changed = true;
|
||||
log.warn("[price-track] stale finalize assemble failed taskId={} shop={} msg={}",
|
||||
taskId, shopKey, message);
|
||||
}
|
||||
}
|
||||
if (!changed && cachedPayloadByShop.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
List<FileResultEntity> refreshed = fileResultMapper.selectList(
|
||||
new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, taskId)
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(FileResultEntity::getId));
|
||||
updateTaskStatusFromLatestRows(task, refreshed);
|
||||
return true;
|
||||
}
|
||||
|
||||
public String resolveResultDownloadUrl(Long resultId, Long userId) {
|
||||
|
||||
Reference in New Issue
Block a user