更新处理相关内容
This commit is contained in:
super
+5
-2
@@ -734,7 +734,10 @@ public class AppearancePatentCozeClient {
|
||||
private String resolveFailureMessage(JsonNode root) {
|
||||
JsonNode dataNode = root.path("data");
|
||||
String message = text(firstNonNull(dataNode.get("error_message"), firstNonNull(dataNode.get("msg"), root.get("msg"))));
|
||||
return message == null ? "" : message;
|
||||
if (message != null && !message.isBlank()) {
|
||||
return message;
|
||||
}
|
||||
return firstNonBlank(findTextByFieldName(root, "error_message", "error", "msg"), "");
|
||||
}
|
||||
|
||||
private String extractExecuteId(JsonNode root) {
|
||||
@@ -978,7 +981,7 @@ public class AppearancePatentCozeClient {
|
||||
|
||||
public boolean isFailed() {
|
||||
String normalized = status == null ? "" : status.trim().toUpperCase(Locale.ROOT);
|
||||
return normalized.contains("FAILED") || normalized.contains("ERROR") || normalized.contains("CANCEL");
|
||||
return normalized.contains("FAIL") || normalized.contains("ERROR") || normalized.contains("CANCEL");
|
||||
}
|
||||
|
||||
public boolean isFinished() {
|
||||
|
||||
+657
-72
@@ -105,6 +105,11 @@ public class AppearancePatentTaskService {
|
||||
private static final Duration TASK_LOCK_TTL = Duration.ofMinutes(5);
|
||||
private static final long TASK_LOCK_WAIT_MILLIS = 10000L;
|
||||
private static final long TASK_LOCK_RETRY_DELAY_MILLIS = 200L;
|
||||
private static final Duration COZE_SUBMIT_LOCK_TTL = Duration.ofMinutes(2);
|
||||
private static final long COZE_SUBMIT_LOCK_WAIT_MILLIS = 180000L;
|
||||
private static final long COZE_SUBMIT_LOCK_RETRY_DELAY_MILLIS = 500L;
|
||||
private static final long COZE_SUBMIT_MIN_INTERVAL_MILLIS = 30000L;
|
||||
private static final int MAX_COZE_SUBMIT_RETRY_COUNT = 5;
|
||||
private static final List<String> RESULT_HEADERS = List.of(
|
||||
"id",
|
||||
"asin",
|
||||
@@ -419,6 +424,7 @@ public class AppearancePatentTaskService {
|
||||
completeSubmittedChunk(context);
|
||||
return null;
|
||||
});
|
||||
submitCozeForSubmittedChunk(context);
|
||||
return;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
@@ -496,12 +502,15 @@ public class AppearancePatentTaskService {
|
||||
taskScopeStateMapper.updateById(scope);
|
||||
}
|
||||
|
||||
SubmitContext context = new SubmitContext(task, scopeKey, scopeHash, chunkIndex,
|
||||
Boolean.TRUE.equals(request.getDone()), request.getError());
|
||||
if (Boolean.TRUE.equals(request.getDone()) || request.getError() != null && !request.getError().isBlank()) {
|
||||
finalizeTask(task, request.getError(), allRowCount(task), true);
|
||||
} else {
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
}
|
||||
submitCozeForSubmittedChunk(context);
|
||||
}
|
||||
|
||||
@Transactional
|
||||
@@ -659,7 +668,7 @@ public class AppearancePatentTaskService {
|
||||
upsertScopeState(taskId, scopeKey, scopeHash, chunkTotal, request.getError(), done, false);
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
return new SubmitContext(task, scopeKey, scopeHash, done, request.getError());
|
||||
return new SubmitContext(task, scopeKey, scopeHash, chunkIndex, done, request.getError());
|
||||
}
|
||||
|
||||
private void completeSubmittedChunk(SubmitContext context) {
|
||||
@@ -681,6 +690,39 @@ public class AppearancePatentTaskService {
|
||||
fileTaskMapper.updateById(task);
|
||||
}
|
||||
|
||||
private void submitCozeForSubmittedChunk(SubmitContext context) {
|
||||
if (context == null || context.task() == null || context.task().getId() == null) {
|
||||
return;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(context.task().getId());
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || STATUS_SUCCESS.equals(task.getStatus())) {
|
||||
return;
|
||||
}
|
||||
List<TaskChunkEntity> chunks = loadSubmittedChunks(task.getId());
|
||||
if (chunks.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
FileResultEntity result = findOrCreateResultRecordForAssembly(task, allRowCount(task));
|
||||
if (result == null) {
|
||||
return;
|
||||
}
|
||||
TaskFileJobEntity job = taskFileJobService.enqueueAssembleResult(
|
||||
task.getId(), MODULE_TYPE, result.getId(), buildTaskOwnerScopeKey(task.getId()));
|
||||
if (job == null || "SUCCESS".equals(job.getStatus())) {
|
||||
return;
|
||||
}
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
|
||||
boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId);
|
||||
saveCozePipelineProgress(task, job);
|
||||
if (pendingCoze) {
|
||||
taskFileJobService.touchRunning(job.getId());
|
||||
touchJavaSideTaskActivity(task.getId());
|
||||
} else if (isResultSubmissionComplete(task.getId())) {
|
||||
maybeFinalizeCozeJobLocked(task.getId(), new CozeBatchContext(
|
||||
job.getId(), result.getId(), null, null, 1, 1, currentInstanceId(), 0));
|
||||
}
|
||||
}
|
||||
|
||||
private void finalizeStaleTask(Long taskId, String error) {
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !STATUS_RUNNING.equals(task.getStatus())) {
|
||||
@@ -1074,13 +1116,19 @@ public class AppearancePatentTaskService {
|
||||
private void finalizeTask(FileTaskEntity task, String error, int rowCount, boolean assembleWorkbook) {
|
||||
String finalError = error;
|
||||
FileResultEntity result = null;
|
||||
boolean hasPersistedResultRows = hasPersistedResultRows(task.getId());
|
||||
List<FileResultEntity> rows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, task.getId())
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.last("limit 1"));
|
||||
if (!rows.isEmpty()) {
|
||||
result = rows.getFirst();
|
||||
if (assembleWorkbook && shouldAssembleSynchronously()) {
|
||||
} else if (assembleWorkbook && hasPersistedResultRows) {
|
||||
result = createResultRecordForAssembly(task, rowCount);
|
||||
}
|
||||
if (result != null) {
|
||||
boolean shouldAssembleResult = assembleWorkbook && (finalError == null || finalError.isBlank() || hasPersistedResultRows);
|
||||
if (shouldAssembleResult && shouldAssembleSynchronously()) {
|
||||
try {
|
||||
assembleResultWorkbook(task, result);
|
||||
} catch (Exception ex) {
|
||||
@@ -1102,20 +1150,69 @@ public class AppearancePatentTaskService {
|
||||
result.setSuccess(failed ? 0 : 1);
|
||||
result.setErrorMessage(finalError);
|
||||
result.setRowCount(rowCount > 0 ? rowCount : result.getRowCount());
|
||||
if (!failed && assembleWorkbook) {
|
||||
boolean shouldAssembleResult = assembleWorkbook && (!failed || hasPersistedResultRows);
|
||||
if (shouldAssembleResult) {
|
||||
result.setResultFilename(safeFileStem(result.getSourceFilename()) + "-result.xlsx");
|
||||
result.setResultFileUrl(null);
|
||||
result.setResultFileSize(0L);
|
||||
result.setResultContentType(CONTENT_TYPE_XLSX);
|
||||
}
|
||||
fileResultMapper.updateById(result);
|
||||
if (!failed && assembleWorkbook) {
|
||||
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), buildTaskOwnerScopeKey(task.getId()));
|
||||
if (shouldAssembleResult) {
|
||||
enqueueResultAssembly(task, result, true);
|
||||
} else if (assembleWorkbook && failed) {
|
||||
log.warn("[appearance-patent] skip failed task workbook assembly because no persisted result rows taskId={}", task.getId());
|
||||
}
|
||||
}
|
||||
taskCacheService.deleteTaskCache(task.getId());
|
||||
}
|
||||
|
||||
private FileResultEntity findOrCreateResultRecordForAssembly(FileTaskEntity task, int rowCount) {
|
||||
if (task == null || task.getId() == null) {
|
||||
return null;
|
||||
}
|
||||
List<FileResultEntity> rows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, task.getId())
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||
.last("limit 1"));
|
||||
if (rows != null && !rows.isEmpty()) {
|
||||
return rows.getFirst();
|
||||
}
|
||||
return createResultRecordForAssembly(task, rowCount);
|
||||
}
|
||||
|
||||
private TaskFileJobEntity enqueueResultAssembly(FileTaskEntity task, FileResultEntity result, boolean dispatchWhenIdle) {
|
||||
if (task == null || task.getId() == null || result == null || result.getId() == null) {
|
||||
return null;
|
||||
}
|
||||
TaskFileJobEntity job = taskFileJobService.enqueueAssembleResult(
|
||||
task.getId(), MODULE_TYPE, result.getId(), buildTaskOwnerScopeKey(task.getId()));
|
||||
if (dispatchWhenIdle
|
||||
&& job != null
|
||||
&& "RUNNING".equals(job.getStatus())
|
||||
&& countPendingCozeStates(task.getId()) == 0) {
|
||||
taskFileJobService.requeue(job.getId(), "Python upload finished, assembling xlsx");
|
||||
}
|
||||
return job;
|
||||
}
|
||||
|
||||
private FileResultEntity createResultRecordForAssembly(FileTaskEntity task, int rowCount) {
|
||||
AppearancePatentParsedPayloadDto payload = readParsedPayload(task);
|
||||
List<AppearancePatentSourceFileDto> sourceFiles = payload.getSourceFiles() == null ? List.of() : payload.getSourceFiles();
|
||||
FileResultEntity result = new FileResultEntity();
|
||||
result.setTaskId(task.getId());
|
||||
result.setModuleType(MODULE_TYPE);
|
||||
result.setSourceFilename(buildAggregateSourceFilenameLabel(sourceFiles));
|
||||
result.setSourceFileUrl(buildAggregateScopeKey(sourceFiles));
|
||||
result.setRowCount(rowCount > 0 ? rowCount : (payload.getAllItems() == null ? 0 : payload.getAllItems().size()));
|
||||
result.setUserId(task.getUserId());
|
||||
result.setCreatedAt(LocalDateTime.now());
|
||||
fileResultMapper.insert(result);
|
||||
log.warn("[appearance-patent] recreated missing result record for workbook assembly taskId={} resultId={}",
|
||||
task.getId(), result.getId());
|
||||
return result;
|
||||
}
|
||||
|
||||
private boolean shouldAssembleSynchronously() {
|
||||
return false;
|
||||
}
|
||||
@@ -1158,29 +1255,41 @@ public class AppearancePatentTaskService {
|
||||
if (countPendingCozeStates(task.getId()) > 0) {
|
||||
taskFileJobService.touchRunning(job.getId());
|
||||
touchJavaSideTaskActivity(task.getId());
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze submitted, waiting for result");
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze 已提交，等待结果回流");
|
||||
return false;
|
||||
}
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, 0, "正在提交 Coze");
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, 0, "Submitting Coze");
|
||||
boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId);
|
||||
if (pendingCoze) {
|
||||
taskFileJobService.touchRunning(job.getId());
|
||||
touchJavaSideTaskActivity(task.getId());
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze submitted, waiting for result");
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze 已提交，等待结果回流");
|
||||
return false;
|
||||
}
|
||||
if (STATUS_RUNNING.equals(task.getStatus()) && !isResultSubmissionComplete(task.getId())) {
|
||||
taskFileJobService.touchRunning(job.getId());
|
||||
touchJavaSideTaskActivity(task.getId());
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, Math.max(1, cozeWorkUnits), "等待 Python 继续回传数据");
|
||||
return false;
|
||||
}
|
||||
completeCozeFileJob(task, result, job, totalProgressUnits, cozeWorkUnits);
|
||||
return true;
|
||||
}
|
||||
|
||||
@Scheduled(fixedDelayString = "${aiimage.appearance-patent.coze-poll-delay-ms:5000}")
|
||||
@Scheduled(fixedDelayString = "${aiimage.appearance-patent.coze-poll-delay-ms:30000}")
|
||||
public void pollPendingCozeJobs() {
|
||||
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
|
||||
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING))
|
||||
.isNotNull(TaskScopeStateEntity::getCozeExecuteId)
|
||||
.orderByAsc(TaskScopeStateEntity::getUpdatedAt)
|
||||
.and(wrapper -> wrapper
|
||||
.apply("JSON_UNQUOTE(JSON_EXTRACT(state_json, '$.ownerInstanceId')) IS NULL")
|
||||
.or()
|
||||
.apply("JSON_UNQUOTE(JSON_EXTRACT(state_json, '$.ownerInstanceId')) = ''")
|
||||
.or()
|
||||
.apply("JSON_UNQUOTE(JSON_EXTRACT(state_json, '$.ownerInstanceId')) = {0}", currentInstanceId()))
|
||||
.orderByDesc(TaskScopeStateEntity::getUpdatedAt)
|
||||
.last("limit 50"));
|
||||
if (states == null || states.isEmpty()) {
|
||||
return;
|
||||
@@ -1205,6 +1314,64 @@ public class AppearancePatentTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
private List<TaskChunkEntity> loadSubmittedChunks(Long taskId) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return List.of();
|
||||
}
|
||||
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
|
||||
.orderByAsc(TaskChunkEntity::getScopeHash)
|
||||
.orderByAsc(TaskChunkEntity::getChunkIndex));
|
||||
return chunks == null ? List.of() : chunks;
|
||||
}
|
||||
|
||||
private List<CozeCandidate> collectPendingCozeCandidates(FileTaskEntity task, List<TaskChunkEntity> chunks) {
|
||||
if (task == null || task.getId() == null || chunks == null || chunks.isEmpty()) {
|
||||
return List.of();
|
||||
}
|
||||
Set<String> queuedRowKeys = new LinkedHashSet<>(loadSubmittedCozeRowKeys(task.getId()));
|
||||
List<CozeCandidate> candidates = new ArrayList<>();
|
||||
for (TaskChunkEntity chunk : chunks) {
|
||||
Map<String, AppearancePatentResultRowDto> persistedRows = readChunkRows(chunk);
|
||||
if (persistedRows.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
for (AppearancePatentResultRowDto row : pickGroupRepresentativesForCoze(persistedRows.values())) {
|
||||
String key = rowKey(row);
|
||||
if (key.isBlank() || !queuedRowKeys.add(key)) {
|
||||
continue;
|
||||
}
|
||||
candidates.add(new CozeCandidate(chunk.getScopeHash(), chunk.getChunkIndex(), row));
|
||||
}
|
||||
}
|
||||
return candidates;
|
||||
}
|
||||
|
||||
private Set<String> loadSubmittedCozeRowKeys(Long taskId) {
|
||||
Set<String> keys = new LinkedHashSet<>();
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return keys;
|
||||
}
|
||||
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||
.select(TaskScopeStateEntity::getId, TaskScopeStateEntity::getTaskId, TaskScopeStateEntity::getParsedPayloadJson)
|
||||
.eq(TaskScopeStateEntity::getTaskId, taskId)
|
||||
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
|
||||
.isNotNull(TaskScopeStateEntity::getCozeStatus));
|
||||
if (states == null || states.isEmpty()) {
|
||||
return keys;
|
||||
}
|
||||
for (TaskScopeStateEntity state : states) {
|
||||
for (AppearancePatentResultRowDto row : readCozeBatchRows(state)) {
|
||||
String key = rowKey(row);
|
||||
if (!key.isBlank()) {
|
||||
keys.add(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
return keys;
|
||||
}
|
||||
|
||||
private boolean submitCozeBatches(FileTaskEntity task,
|
||||
FileResultEntity result,
|
||||
TaskFileJobEntity job,
|
||||
@@ -1221,24 +1388,27 @@ public class AppearancePatentTaskService {
|
||||
String prompt = readAiPrompt(task);
|
||||
String apiKey = readApiKey(task);
|
||||
int batchSize = Math.max(1, properties.getCozeBatchSize());
|
||||
List<CozeCandidate> candidates = collectPendingCozeCandidates(task, chunks);
|
||||
boolean flushRemainder = isResultSubmissionComplete(task.getId());
|
||||
int submitLimit = (candidates.size() / batchSize) * batchSize;
|
||||
if (flushRemainder && submitLimit < candidates.size()) {
|
||||
submitLimit = candidates.size();
|
||||
}
|
||||
if (submitLimit <= 0) {
|
||||
log.info("[appearance-patent] coze batch waiting for more rows taskId={} jobId={} pendingRows={} batchSize={} finalUpload={}",
|
||||
task.getId(), job.getId(), candidates.size(), batchSize, flushRemainder);
|
||||
return countPendingCozeStates(task.getId()) > 0;
|
||||
}
|
||||
boolean pending = false;
|
||||
for (TaskChunkEntity chunk : chunks) {
|
||||
Map<String, AppearancePatentResultRowDto> persistedRows = readChunkRows(chunk);
|
||||
if (persistedRows.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
List<AppearancePatentResultRowDto> unresolvedRows = pickGroupRepresentativesForCoze(persistedRows.values());
|
||||
if (unresolvedRows.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
int batchTotal = Math.max(1, (unresolvedRows.size() + batchSize - 1) / batchSize);
|
||||
int batchIndex = 1;
|
||||
for (int i = 0; i < unresolvedRows.size(); i += batchSize) {
|
||||
List<AppearancePatentResultRowDto> batchRows =
|
||||
unresolvedRows.subList(i, Math.min(i + batchSize, unresolvedRows.size()));
|
||||
pending |= submitCozeBatch(task, result, job, chunk, batchRows, batchIndex, batchTotal, prompt, apiKey, allRowsByBaseId);
|
||||
batchIndex++;
|
||||
}
|
||||
int batchTotal = Math.max(1, (submitLimit + batchSize - 1) / batchSize);
|
||||
int batchIndex = 1;
|
||||
for (int i = 0; i < submitLimit; i += batchSize) {
|
||||
List<CozeCandidate> batchCandidates = candidates.subList(i, Math.min(i + batchSize, submitLimit));
|
||||
List<AppearancePatentResultRowDto> batchRows = batchCandidates.stream()
|
||||
.map(CozeCandidate::row)
|
||||
.toList();
|
||||
pending |= submitCozeBatch(task, result, job, batchRows, batchIndex, batchTotal, prompt, apiKey, allRowsByBaseId);
|
||||
batchIndex++;
|
||||
}
|
||||
return pending || countPendingCozeStates(task.getId()) > 0;
|
||||
}
|
||||
@@ -1246,7 +1416,6 @@ public class AppearancePatentTaskService {
|
||||
private boolean submitCozeBatch(FileTaskEntity task,
|
||||
FileResultEntity result,
|
||||
TaskFileJobEntity job,
|
||||
TaskChunkEntity chunk,
|
||||
List<AppearancePatentResultRowDto> batchRows,
|
||||
int batchIndex,
|
||||
int batchTotal,
|
||||
@@ -1256,7 +1425,7 @@ public class AppearancePatentTaskService {
|
||||
if (batchRows == null || batchRows.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
String batchScopeKey = buildCozeBatchScopeKey(job.getId(), chunk.getScopeHash(), chunk.getChunkIndex(), batchIndex);
|
||||
String batchScopeKey = buildCozeBatchScopeKey(task.getId(), batchRows);
|
||||
String batchScopeHash = DigestUtil.sha256Hex(batchScopeKey);
|
||||
TaskScopeStateEntity existing = taskScopeStateMapper.selectOne(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getTaskId, task.getId())
|
||||
@@ -1268,34 +1437,28 @@ public class AppearancePatentTaskService {
|
||||
|| COZE_STATUS_RUNNING.equals(existing.getCozeStatus());
|
||||
}
|
||||
try {
|
||||
AppearancePatentCozeClient.CozeSubmitResponse submit = cozeClient.submitWorkflow(batchRows, prompt, apiKey);
|
||||
AppearancePatentCozeClient.CozeSubmitResponse submit = submitCozeWorkflowThrottled(batchRows, prompt, apiKey);
|
||||
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
|
||||
List<AppearancePatentResultRowDto> cozeRows = cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
|
||||
mergeCozeRowsIntoChunk(task, chunk.getScopeHash(), chunk.getChunkIndex(), cozeRows, allRowsByBaseId);
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId);
|
||||
return false;
|
||||
}
|
||||
if (submit.executeId() == null || submit.executeId().isBlank()) {
|
||||
mergeCozeRowsIntoChunk(task,
|
||||
chunk.getScopeHash(),
|
||||
chunk.getChunkIndex(),
|
||||
mergeCozeRowsIntoSubmittedChunks(task,
|
||||
cozeClient.markRowsFailed(batchRows, "Coze async execute_id missing"),
|
||||
allRowsByBaseId);
|
||||
return false;
|
||||
}
|
||||
saveCozeBatchState(task, result, job, chunk, batchRows, batchScopeKey, batchScopeHash,
|
||||
saveCozeBatchState(task, result, job, batchRows, batchScopeKey, batchScopeHash,
|
||||
batchIndex, batchTotal, submit.executeId());
|
||||
log.info("[appearance-patent] coze async submitted taskId={} jobId={} chunk={} batch={}/{} executeId={}",
|
||||
task.getId(), job.getId(), chunk.getChunkIndex(), batchIndex, batchTotal, submit.executeId());
|
||||
log.info("[appearance-patent] coze async submitted taskId={} jobId={} rows={} batch={}/{} executeId={}",
|
||||
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal, submit.executeId());
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
String message = firstNonBlank(ex.getMessage(), "Coze submit failed");
|
||||
log.warn("[appearance-patent] coze async submit failed taskId={} jobId={} chunk={} batch={}/{} err={}",
|
||||
task.getId(), job.getId(), chunk.getChunkIndex(), batchIndex, batchTotal, message);
|
||||
mergeCozeRowsIntoChunk(task,
|
||||
chunk.getScopeHash(),
|
||||
chunk.getChunkIndex(),
|
||||
cozeClient.markRowsFailed(batchRows, message),
|
||||
allRowsByBaseId);
|
||||
log.warn("[appearance-patent] coze async submit failed taskId={} jobId={} rows={} batch={}/{} err={}",
|
||||
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal, message);
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeClient.markRowsFailed(batchRows, message), allRowsByBaseId);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -1303,7 +1466,6 @@ public class AppearancePatentTaskService {
|
||||
private void saveCozeBatchState(FileTaskEntity task,
|
||||
FileResultEntity result,
|
||||
TaskFileJobEntity job,
|
||||
TaskChunkEntity chunk,
|
||||
List<AppearancePatentResultRowDto> batchRows,
|
||||
String batchScopeKey,
|
||||
String batchScopeHash,
|
||||
@@ -1314,11 +1476,12 @@ public class AppearancePatentTaskService {
|
||||
CozeBatchContext context = new CozeBatchContext(
|
||||
job.getId(),
|
||||
result.getId(),
|
||||
chunk.getScopeHash(),
|
||||
chunk.getChunkIndex(),
|
||||
null,
|
||||
null,
|
||||
batchIndex,
|
||||
batchTotal,
|
||||
currentInstanceId()
|
||||
currentInstanceId(),
|
||||
0
|
||||
);
|
||||
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
|
||||
String storedBatchPayload = storeSharedCozeBatchPayload(task.getId(), batchScopeHash, batchPayload);
|
||||
@@ -1393,10 +1556,10 @@ public class AppearancePatentTaskService {
|
||||
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing");
|
||||
return;
|
||||
}
|
||||
if (!isOwnerCurrent(context.ownerInstanceId())) {
|
||||
log.info("[appearance-patent] coze poll skipped after context refresh because owner is another instance taskId={} stateId={} owner={} current={}",
|
||||
state.getTaskId(), state.getId(), context.ownerInstanceId(), currentInstanceId());
|
||||
return;
|
||||
if (!isOwnerCurrent(context.ownerInstanceId())) {
|
||||
log.info("[appearance-patent] coze poll skipped after context refresh because owner is another instance taskId={} stateId={} owner={} current={}",
|
||||
state.getTaskId(), state.getId(), context.ownerInstanceId(), currentInstanceId());
|
||||
return;
|
||||
}
|
||||
taskFileJobService.touchRunning(context.jobId());
|
||||
log.info("[appearance-patent] coze poll start taskId={} stateId={} executeId={} jobId={} chunk={} batch={}/{}",
|
||||
@@ -1421,17 +1584,27 @@ public class AppearancePatentTaskService {
|
||||
if (batchRows.isEmpty() && failureMessage.isBlank()) {
|
||||
failureMessage = "Coze batch payload missing";
|
||||
}
|
||||
if (!failureMessage.isBlank() && splitRetryFailedCozeBatchState(state, context, batchRows, failureMessage)) {
|
||||
return;
|
||||
}
|
||||
if (!failureMessage.isBlank() && retryFailedCozeBatchState(state, context, batchRows, failureMessage)) {
|
||||
return;
|
||||
}
|
||||
List<AppearancePatentResultRowDto> cozeRows = failureMessage.isBlank()
|
||||
? cozeClient.mergeRowsFromDataText(batchRows, poll.resolvedPayloadText())
|
||||
: cozeClient.markRowsFailed(batchRows, failureMessage);
|
||||
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
|
||||
if (task != null) {
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
|
||||
mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId);
|
||||
}
|
||||
markCozeStateTerminal(state,
|
||||
failureMessage.isBlank() ? COZE_STATUS_DONE : COZE_STATUS_FAILED,
|
||||
failureMessage.isBlank() ? null : failureMessage);
|
||||
if (task != null) {
|
||||
TaskFileJobEntity progressJob = taskFileJobService.findAssembleJob(state.getTaskId(), MODULE_TYPE, context.resultId());
|
||||
saveCozePipelineProgress(task, progressJob);
|
||||
}
|
||||
log.info("[appearance-patent] coze poll completed taskId={} stateId={} executeId={} status={} batchRows={} mergedRows={} failure={}",
|
||||
state.getTaskId(), state.getId(), state.getCozeExecuteId(),
|
||||
failureMessage.isBlank() ? COZE_STATUS_DONE : COZE_STATUS_FAILED,
|
||||
@@ -1447,9 +1620,7 @@ public class AppearancePatentTaskService {
|
||||
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
|
||||
if (task != null) {
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
|
||||
mergeCozeRowsIntoChunk(task,
|
||||
context.chunkScopeHash(),
|
||||
context.chunkIndex(),
|
||||
mergeCozeRowsIntoSubmittedChunks(task,
|
||||
cozeClient.markRowsFailed(batchRows, message),
|
||||
allRowsByBaseId);
|
||||
}
|
||||
@@ -1467,6 +1638,240 @@ public class AppearancePatentTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
private boolean retryFailedCozeBatchState(TaskScopeStateEntity state,
|
||||
CozeBatchContext context,
|
||||
List<AppearancePatentResultRowDto> batchRows,
|
||||
String failureMessage) {
|
||||
if (state == null || context == null || batchRows == null || batchRows.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
if (!isRetryableCozeFailure(failureMessage) || cozeSubmitRetryCount(context) >= MAX_COZE_SUBMIT_RETRY_COUNT) {
|
||||
return false;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
AppearancePatentCozeClient.CozeSubmitResponse submit =
|
||||
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task));
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
|
||||
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
|
||||
List<AppearancePatentResultRowDto> cozeRows =
|
||||
cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId);
|
||||
markCozeStateTerminal(state, COZE_STATUS_DONE, null);
|
||||
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
|
||||
log.info("[appearance-patent] coze retry returned immediate result taskId={} stateId={} chunk={} batch={}/{}",
|
||||
state.getTaskId(), state.getId(), context.chunkIndex(), context.batchIndex(), context.batchTotal());
|
||||
return true;
|
||||
}
|
||||
if (submit.executeId() == null || submit.executeId().isBlank()) {
|
||||
return false;
|
||||
}
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
CozeBatchContext retryContext = withCozeSubmitRetryCount(context, cozeSubmitRetryCount(context) + 1);
|
||||
int updated = taskScopeStateMapper.update(null, new LambdaUpdateWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getId, state.getId())
|
||||
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING))
|
||||
.set(TaskScopeStateEntity::getCozeExecuteId, submit.executeId())
|
||||
.set(TaskScopeStateEntity::getCozeStatus, COZE_STATUS_SUBMITTED)
|
||||
.set(TaskScopeStateEntity::getCozeSubmittedAt, now)
|
||||
.set(TaskScopeStateEntity::getCozeLastPolledAt, null)
|
||||
.set(TaskScopeStateEntity::getCozeCompletedAt, null)
|
||||
.set(TaskScopeStateEntity::getCozeAttemptCount, 0)
|
||||
.set(TaskScopeStateEntity::getCozeError, "retry after failure: " + firstNonBlank(failureMessage, "unknown"))
|
||||
.set(TaskScopeStateEntity::getStateJson, writeJson(retryContext, "serialize coze batch retry context failed"))
|
||||
.set(TaskScopeStateEntity::getCompleted, 0)
|
||||
.set(TaskScopeStateEntity::getUpdatedAt, now));
|
||||
if (updated > 0) {
|
||||
taskFileJobService.touchRunning(context.jobId());
|
||||
touchJavaSideTaskActivity(state.getTaskId());
|
||||
log.info("[appearance-patent] coze retry submitted taskId={} stateId={} oldExecuteId={} newExecuteId={} chunk={} batch={}/{} retry={}/{} failure={}",
|
||||
state.getTaskId(), state.getId(), state.getCozeExecuteId(), submit.executeId(),
|
||||
context.chunkIndex(), context.batchIndex(), context.batchTotal(),
|
||||
retryContext.submitRetryCount(), MAX_COZE_SUBMIT_RETRY_COUNT, failureMessage);
|
||||
return true;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
log.warn("[appearance-patent] coze retry submit failed taskId={} stateId={} executeId={} err={}",
|
||||
state.getTaskId(), state.getId(), state.getCozeExecuteId(), firstNonBlank(ex.getMessage(), "Coze retry failed"));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private AppearancePatentCozeClient.CozeSubmitResponse submitCozeWorkflowThrottled(
|
||||
List<AppearancePatentResultRowDto> rows,
|
||||
String prompt,
|
||||
String apiKey) throws Exception {
|
||||
DistributedJobLockService.LockHandle lockHandle = acquireCozeSubmitLock();
|
||||
if (lockHandle == null) {
|
||||
throw new IllegalStateException("Coze submit throttle lock timeout");
|
||||
}
|
||||
try (lockHandle) {
|
||||
AppearancePatentCozeClient.CozeSubmitResponse response = cozeClient.submitWorkflow(rows, prompt, apiKey);
|
||||
sleepQuietly(COZE_SUBMIT_MIN_INTERVAL_MILLIS);
|
||||
return response;
|
||||
}
|
||||
}
|
||||
|
||||
private DistributedJobLockService.LockHandle acquireCozeSubmitLock() {
|
||||
long deadline = System.currentTimeMillis() + COZE_SUBMIT_LOCK_WAIT_MILLIS;
|
||||
while (System.currentTimeMillis() <= deadline) {
|
||||
DistributedJobLockService.LockHandle lockHandle =
|
||||
distributedJobLockService.tryLock("appearance-patent:coze-submit", COZE_SUBMIT_LOCK_TTL);
|
||||
if (lockHandle != null) {
|
||||
return lockHandle;
|
||||
}
|
||||
sleepQuietly(COZE_SUBMIT_LOCK_RETRY_DELAY_MILLIS);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private void sleepQuietly(long millis) {
|
||||
if (millis <= 0L) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(millis);
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean splitRetryFailedCozeBatchState(TaskScopeStateEntity state,
|
||||
CozeBatchContext context,
|
||||
List<AppearancePatentResultRowDto> batchRows,
|
||||
String failureMessage) {
|
||||
if (state == null || context == null || batchRows == null || batchRows.size() <= 1) {
|
||||
return false;
|
||||
}
|
||||
if (!shouldSplitCozeBatchForRetry(failureMessage) || cozeSubmitRetryCount(context) >= MAX_COZE_SUBMIT_RETRY_COUNT) {
|
||||
return false;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
|
||||
return false;
|
||||
}
|
||||
int middle = Math.max(1, batchRows.size() / 2);
|
||||
List<List<AppearancePatentResultRowDto>> partitions = List.<List<AppearancePatentResultRowDto>>of(
|
||||
new ArrayList<>(batchRows.subList(0, middle)),
|
||||
new ArrayList<>(batchRows.subList(middle, batchRows.size()))
|
||||
).stream().filter(rows -> rows != null && !rows.isEmpty()).toList();
|
||||
int retryCount = cozeSubmitRetryCount(context) + 1;
|
||||
boolean submittedAny = false;
|
||||
try {
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
|
||||
int partIndex = 1;
|
||||
for (List<AppearancePatentResultRowDto> partRows : partitions) {
|
||||
AppearancePatentCozeClient.CozeSubmitResponse submit =
|
||||
submitCozeWorkflowThrottled(partRows, readAiPrompt(task), readApiKey(task));
|
||||
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
|
||||
List<AppearancePatentResultRowDto> cozeRows =
|
||||
cozeClient.mergeRowsFromDataText(partRows, submit.immediateData());
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId);
|
||||
submittedAny = true;
|
||||
} else if (submit.executeId() != null && !submit.executeId().isBlank()) {
|
||||
saveSplitRetryCozeBatchState(state, context, partRows, partIndex, partitions.size(), retryCount, submit.executeId());
|
||||
submittedAny = true;
|
||||
}
|
||||
partIndex++;
|
||||
}
|
||||
if (submittedAny) {
|
||||
markCozeStateTerminal(state, COZE_STATUS_DONE, "split retry submitted after failure: " + firstNonBlank(failureMessage, "unknown"));
|
||||
taskFileJobService.touchRunning(context.jobId());
|
||||
touchJavaSideTaskActivity(state.getTaskId());
|
||||
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
|
||||
log.info("[appearance-patent] coze split retry submitted taskId={} stateId={} chunk={} batch={}/{} parts={} retry={}/{} failure={}",
|
||||
state.getTaskId(), state.getId(), context.chunkIndex(), context.batchIndex(), context.batchTotal(),
|
||||
partitions.size(), retryCount, MAX_COZE_SUBMIT_RETRY_COUNT, failureMessage);
|
||||
return true;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
log.warn("[appearance-patent] coze split retry submit failed taskId={} stateId={} executeId={} err={}",
|
||||
state.getTaskId(), state.getId(), state.getCozeExecuteId(), firstNonBlank(ex.getMessage(), "Coze split retry failed"));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private void saveSplitRetryCozeBatchState(TaskScopeStateEntity parent,
|
||||
CozeBatchContext parentContext,
|
||||
List<AppearancePatentResultRowDto> batchRows,
|
||||
int partIndex,
|
||||
int partTotal,
|
||||
int retryCount,
|
||||
String executeId) {
|
||||
String scopeKey = parent.getScopeKey() + ":split:" + retryCount + ":" + partIndex;
|
||||
String scopeHash = DigestUtil.sha256Hex(scopeKey);
|
||||
CozeBatchContext context = new CozeBatchContext(
|
||||
parentContext.jobId(),
|
||||
parentContext.resultId(),
|
||||
parentContext.chunkScopeHash(),
|
||||
parentContext.chunkIndex(),
|
||||
partIndex,
|
||||
partTotal,
|
||||
parentContext.ownerInstanceId(),
|
||||
retryCount
|
||||
);
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
String batchPayload = writeJson(batchRows, "serialize split coze batch payload failed");
|
||||
String storedBatchPayload = storeSharedCozeBatchPayload(parent.getTaskId(), scopeHash, batchPayload);
|
||||
TaskScopeStateEntity state = new TaskScopeStateEntity();
|
||||
state.setTaskId(parent.getTaskId());
|
||||
state.setModuleType(MODULE_TYPE);
|
||||
state.setScopeKey(scopeKey);
|
||||
state.setScopeHash(scopeHash);
|
||||
state.setParsedPayloadJson(storedBatchPayload);
|
||||
state.setStateJson(writeJson(context, "serialize split coze batch context failed"));
|
||||
state.setCozeExecuteId(executeId);
|
||||
state.setCozeStatus(COZE_STATUS_SUBMITTED);
|
||||
state.setCozeSubmittedAt(now);
|
||||
state.setCozeAttemptCount(0);
|
||||
state.setChunkTotal(partTotal);
|
||||
state.setReceivedChunkCount(partIndex);
|
||||
state.setCompleted(0);
|
||||
state.setCreatedAt(now);
|
||||
state.setUpdatedAt(now);
|
||||
try {
|
||||
taskScopeStateMapper.insert(state);
|
||||
} catch (DuplicateKeyException ex) {
|
||||
transientPayloadStorageService.deletePayloadIfPresent(storedBatchPayload);
|
||||
log.info("[appearance-patent] duplicate split coze batch state ignored taskId={} scope={}",
|
||||
parent.getTaskId(), scopeKey);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean shouldSplitCozeBatchForRetry(String failureMessage) {
|
||||
String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
|
||||
return normalized.contains("timeout")
|
||||
|| normalized.contains("timed out")
|
||||
|| normalized.contains("out of limit")
|
||||
|| normalized.contains("execution limit")
|
||||
|| normalized.contains("720712008")
|
||||
|| normalized.contains("720701002")
|
||||
|| normalized.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650")
|
||||
|| normalized.contains("\u8c03\u7528\u8d85\u65f6");
|
||||
}
|
||||
|
||||
private boolean isRetryableCozeFailure(String failureMessage) {
|
||||
String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
|
||||
return normalized.contains("rate limit")
|
||||
|| normalized.contains("too many")
|
||||
|| normalized.contains("retry later")
|
||||
|| normalized.contains("timeout")
|
||||
|| normalized.contains("timed out")
|
||||
|| normalized.contains("out of limit")
|
||||
|| normalized.contains("execution limit")
|
||||
|| normalized.contains("702093018")
|
||||
|| normalized.contains("720712008")
|
||||
|| normalized.contains("720701002")
|
||||
|| normalized.contains("plugin limit")
|
||||
|| normalized.contains("\u9650\u6d41")
|
||||
|| normalized.contains("\u7a0d\u540e\u91cd\u8bd5")
|
||||
|| normalized.contains("\u5de5\u4f5c\u6d41\u8282\u70b9\u6267\u884c\u8d85\u9650")
|
||||
|| normalized.contains("\u8c03\u7528\u8d85\u65f6");
|
||||
}
|
||||
|
||||
private void updateCozeStateRunning(TaskScopeStateEntity state, String error) {
|
||||
int updated = taskScopeStateMapper.update(null, new LambdaUpdateWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getId, state.getId())
|
||||
@@ -1546,7 +1951,13 @@ public class AppearancePatentTaskService {
|
||||
if (job == null || "SUCCESS".equals(job.getStatus())) {
|
||||
return;
|
||||
}
|
||||
boolean requeued = taskFileJobService.requeue(job.getId(), "Coze results ready, assembling xlsx");
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task != null && STATUS_RUNNING.equals(task.getStatus()) && !isResultSubmissionComplete(taskId)) {
|
||||
taskFileJobService.touchRunning(job.getId());
|
||||
touchJavaSideTaskActivity(taskId);
|
||||
return;
|
||||
}
|
||||
boolean requeued = taskFileJobService.requeue(job.getId(), "Coze 结果已回流，正在组装 xlsx");
|
||||
if (requeued) {
|
||||
log.info("[appearance-patent] coze async results ready, result file job requeued taskId={} jobId={} resultId={}",
|
||||
taskId, job.getId(), context.resultId());
|
||||
@@ -1570,17 +1981,22 @@ public class AppearancePatentTaskService {
|
||||
throw new BusinessException("Coze 结果仍在处理中，暂不能生成结果文件");
|
||||
}
|
||||
int assembleProgress = Math.max(1, Math.min(totalProgressUnits - 2, cozeWorkUnits));
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, assembleProgress, "Assembling xlsx");
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, assembleProgress, "正在组装 xlsx");
|
||||
assembleResultWorkbook(task, result);
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, totalProgressUnits - 1, "Uploading result file");
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, totalProgressUnits - 1, "正在上传结果文件");
|
||||
fileResultMapper.updateById(result);
|
||||
task.setStatus(STATUS_SUCCESS);
|
||||
task.setErrorMessage(null);
|
||||
boolean taskAlreadyFailed = STATUS_FAILED.equals(task.getStatus())
|
||||
|| (task.getErrorMessage() != null && !task.getErrorMessage().isBlank());
|
||||
String existingError = task.getErrorMessage();
|
||||
task.setStatus(taskAlreadyFailed ? STATUS_FAILED : STATUS_SUCCESS);
|
||||
task.setSuccessFileCount(taskAlreadyFailed ? 0 : 1);
|
||||
task.setFailedFileCount(taskAlreadyFailed ? 1 : 0);
|
||||
task.setErrorMessage(taskAlreadyFailed ? existingError : null);
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
task.setFinishedAt(LocalDateTime.now());
|
||||
fileTaskMapper.updateById(task);
|
||||
taskCacheService.deleteTaskCache(task.getId());
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, totalProgressUnits, "Result file generated");
|
||||
saveFileBuildProgress(task, job, totalProgressUnits, totalProgressUnits, "结果文件已生成");
|
||||
}
|
||||
|
||||
private void mergeCozeRowsIntoChunk(FileTaskEntity task,
|
||||
@@ -1588,16 +2004,74 @@ public class AppearancePatentTaskService {
|
||||
Integer chunkIndex,
|
||||
List<AppearancePatentResultRowDto> cozeRows,
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId) {
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId, chunkScopeHash, chunkIndex);
|
||||
}
|
||||
|
||||
private void mergeCozeRowsIntoSubmittedChunks(FileTaskEntity task,
|
||||
List<AppearancePatentResultRowDto> cozeRows,
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId) {
|
||||
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId, null, null);
|
||||
}
|
||||
|
||||
private void mergeCozeRowsIntoSubmittedChunks(FileTaskEntity task,
|
||||
List<AppearancePatentResultRowDto> cozeRows,
|
||||
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId,
|
||||
String fallbackScopeHash,
|
||||
Integer fallbackChunkIndex) {
|
||||
if (task == null || cozeRows == null || cozeRows.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<String, AppearancePatentResultRowDto> mergedRows = new LinkedHashMap<>();
|
||||
List<TaskChunkEntity> chunks = loadSubmittedChunks(task.getId());
|
||||
if (chunks.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<String, Map<String, AppearancePatentResultRowDto>> rowsByChunk = new LinkedHashMap<>();
|
||||
Map<String, TaskChunkEntity> chunkByKey = new LinkedHashMap<>();
|
||||
for (TaskChunkEntity chunk : chunks) {
|
||||
String chunkKey = chunkStorageKey(chunk.getScopeHash(), chunk.getChunkIndex());
|
||||
rowsByChunk.put(chunkKey, readChunkRows(chunk));
|
||||
chunkByKey.put(chunkKey, chunk);
|
||||
}
|
||||
Map<String, Map<String, AppearancePatentResultRowDto>> mergeRowsByChunk = new LinkedHashMap<>();
|
||||
for (AppearancePatentResultRowDto resultRow : cozeRows) {
|
||||
for (AppearancePatentResultRowDto expandedRow : expandRows(List.of(resultRow), allRowsByBaseId)) {
|
||||
mergedRows.put(rowKey(expandedRow), expandedRow);
|
||||
String rowKey = rowKey(expandedRow);
|
||||
if (rowKey.isBlank()) {
|
||||
continue;
|
||||
}
|
||||
boolean matched = false;
|
||||
for (Map.Entry<String, Map<String, AppearancePatentResultRowDto>> entry : rowsByChunk.entrySet()) {
|
||||
if (entry.getValue().containsKey(rowKey)) {
|
||||
mergeRowsByChunk.computeIfAbsent(entry.getKey(), ignored -> new LinkedHashMap<>())
|
||||
.put(rowKey, expandedRow);
|
||||
matched = true;
|
||||
}
|
||||
}
|
||||
if (!matched && fallbackScopeHash != null && fallbackChunkIndex != null) {
|
||||
String fallbackKey = chunkStorageKey(fallbackScopeHash, fallbackChunkIndex);
|
||||
if (chunkByKey.containsKey(fallbackKey)) {
|
||||
mergeRowsByChunk.computeIfAbsent(fallbackKey, ignored -> new LinkedHashMap<>())
|
||||
.put(rowKey, expandedRow);
|
||||
matched = true;
|
||||
}
|
||||
}
|
||||
if (!matched) {
|
||||
log.warn("[appearance-patent] coze row has no submitted chunk taskId={} rowKey={}",
|
||||
task.getId(), rowKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
mergeChunkPayload(task.getId(), chunkScopeHash, chunkIndex, new ArrayList<>(mergedRows.values()));
|
||||
for (Map.Entry<String, Map<String, AppearancePatentResultRowDto>> entry : mergeRowsByChunk.entrySet()) {
|
||||
TaskChunkEntity chunk = chunkByKey.get(entry.getKey());
|
||||
if (chunk == null || entry.getValue().isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
mergeChunkPayload(task.getId(), chunk.getScopeHash(), chunk.getChunkIndex(), new ArrayList<>(entry.getValue().values()));
|
||||
}
|
||||
}
|
||||
|
||||
private String chunkStorageKey(String scopeHash, Integer chunkIndex) {
|
||||
return firstNonBlank(scopeHash, "") + ":" + (chunkIndex == null ? 0 : chunkIndex);
|
||||
}
|
||||
|
||||
private List<AppearancePatentResultRowDto> readCozeBatchRows(TaskScopeStateEntity state) {
|
||||
@@ -1636,6 +2110,23 @@ public class AppearancePatentTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
private int cozeSubmitRetryCount(CozeBatchContext context) {
|
||||
return context == null || context.submitRetryCount() == null ? 0 : context.submitRetryCount();
|
||||
}
|
||||
|
||||
private CozeBatchContext withCozeSubmitRetryCount(CozeBatchContext context, int submitRetryCount) {
|
||||
return new CozeBatchContext(
|
||||
context.jobId(),
|
||||
context.resultId(),
|
||||
context.chunkScopeHash(),
|
||||
context.chunkIndex(),
|
||||
context.batchIndex(),
|
||||
context.batchTotal(),
|
||||
context.ownerInstanceId(),
|
||||
submitRetryCount
|
||||
);
|
||||
}
|
||||
|
||||
private boolean isCozeStateTimedOut(TaskScopeStateEntity state) {
|
||||
if (state == null || state.getCozeSubmittedAt() == null) {
|
||||
return false;
|
||||
@@ -1670,6 +2161,30 @@ public class AppearancePatentTaskService {
|
||||
return count == null ? 0 : count.intValue();
|
||||
}
|
||||
|
||||
private int countAllCozeStates(Long taskId) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return 0;
|
||||
}
|
||||
Long count = taskScopeStateMapper.selectCount(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getTaskId, taskId)
|
||||
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
|
||||
.isNotNull(TaskScopeStateEntity::getCozeStatus));
|
||||
return count == null ? 0 : count.intValue();
|
||||
}
|
||||
|
||||
private boolean isResultSubmissionComplete(Long taskId) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return false;
|
||||
}
|
||||
Long count = taskScopeStateMapper.selectCount(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getTaskId, taskId)
|
||||
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
|
||||
.isNull(TaskScopeStateEntity::getCozeStatus)
|
||||
.isNotNull(TaskScopeStateEntity::getLastChunkAt)
|
||||
.eq(TaskScopeStateEntity::getCompleted, 1));
|
||||
return count != null && count > 0;
|
||||
}
|
||||
|
||||
private boolean isJavaSideProcessing(Long taskId) {
|
||||
return countPendingCozeStates(taskId) > 0
|
||||
|| taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE) > 0;
|
||||
@@ -1686,11 +2201,20 @@ public class AppearancePatentTaskService {
|
||||
.set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()));
|
||||
}
|
||||
|
||||
private String buildCozeBatchScopeKey(Long jobId, String chunkScopeHash, Integer chunkIndex, int batchIndex) {
|
||||
return "coze:job:" + jobId
|
||||
+ ":chunk:" + (chunkIndex == null ? 0 : chunkIndex)
|
||||
+ ":" + firstNonBlank(chunkScopeHash, "unknown")
|
||||
+ ":batch:" + batchIndex;
|
||||
private String buildCozeBatchScopeKey(Long taskId, List<AppearancePatentResultRowDto> batchRows) {
|
||||
StringBuilder rowKeys = new StringBuilder();
|
||||
if (batchRows != null) {
|
||||
for (AppearancePatentResultRowDto row : batchRows) {
|
||||
String key = rowKey(row);
|
||||
if (!key.isBlank()) {
|
||||
if (!rowKeys.isEmpty()) {
|
||||
rowKeys.append('|');
|
||||
}
|
||||
rowKeys.append(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
return "coze:task:" + taskId + ":rows:" + DigestUtil.sha256Hex(rowKeys.toString());
|
||||
}
|
||||
|
||||
private String buildTaskOwnerScopeKey(Long taskId) {
|
||||
@@ -1811,6 +2335,45 @@ public class AppearancePatentTaskService {
|
||||
);
|
||||
}
|
||||
|
||||
private void saveCozePipelineProgress(FileTaskEntity task, TaskFileJobEntity job) {
|
||||
if (task == null || task.getId() == null || job == null || job.getId() == null) {
|
||||
return;
|
||||
}
|
||||
int totalCoze = countAllCozeStates(task.getId());
|
||||
int completedCoze = countCompletedCozeStates(task.getId());
|
||||
int pendingCoze = countPendingCozeStates(task.getId());
|
||||
boolean uploadComplete = isResultSubmissionComplete(task.getId());
|
||||
if (totalCoze <= 0) {
|
||||
int receivedChunks = countTaskChunks(task.getId());
|
||||
int uploadedProgress = Math.max(1, receivedChunks);
|
||||
saveFileBuildProgress(task, job, Math.max(uploadedProgress + 1, 2), uploadedProgress,
|
||||
"正在接收 Python 数据，累计 50 条后提交 Coze");
|
||||
return;
|
||||
}
|
||||
int total = Math.max(3, totalCoze + 3);
|
||||
int completed = Math.max(0, Math.min(completedCoze, total - 1));
|
||||
String message;
|
||||
if (pendingCoze > 0) {
|
||||
message = "已提交 Coze " + totalCoze + " 批，已完成 " + completedCoze + " 批，等待结果回流";
|
||||
} else if (!uploadComplete) {
|
||||
message = "Coze 已完成 " + completedCoze + " 批，等待 Python 继续回传数据";
|
||||
} else {
|
||||
message = "Coze 结果已回流，正在组装 xlsx";
|
||||
completed = Math.max(completed, Math.min(total - 2, completedCoze));
|
||||
}
|
||||
saveFileBuildProgress(task, job, total, completed, message);
|
||||
}
|
||||
|
||||
private int countTaskChunks(Long taskId) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return 0;
|
||||
}
|
||||
Long count = taskChunkMapper.selectCount(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE));
|
||||
return count == null ? 0 : count.intValue();
|
||||
}
|
||||
|
||||
public void cleanupResultFileJob(TaskFileJobEntity job) {
|
||||
if (job == null || job.getTaskId() == null) {
|
||||
return;
|
||||
@@ -1891,6 +2454,10 @@ public class AppearancePatentTaskService {
|
||||
if (receivedRows == null || receivedRows.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task != null && STATUS_FAILED.equals(task.getStatus())) {
|
||||
return;
|
||||
}
|
||||
int expectedRows = 0;
|
||||
int missingRows = 0;
|
||||
List<String> sampleAsins = new ArrayList<>();
|
||||
@@ -1907,7 +2474,8 @@ public class AppearancePatentTaskService {
|
||||
}
|
||||
}
|
||||
}
|
||||
if (expectedRows > 0 && missingRows > 0) {
|
||||
boolean enforceCompleteCozeCoverage = false;
|
||||
if (enforceCompleteCozeCoverage && expectedRows > 0 && missingRows > 0) {
|
||||
log.warn("[appearance-patent] incomplete coze coverage taskId={} expectedRows={} missingRows={} samples={}",
|
||||
taskId, expectedRows, missingRows, sampleAsins);
|
||||
throw new BusinessException("Coze 结果不完整：缺少 " + missingRows + "/" + expectedRows + " 条检测结果，请等待重试或重新运行任务");
|
||||
@@ -1933,6 +2501,16 @@ public class AppearancePatentTaskService {
|
||||
return result;
|
||||
}
|
||||
|
||||
private boolean hasPersistedResultRows(Long taskId) {
|
||||
if (taskId == null) {
|
||||
return false;
|
||||
}
|
||||
Long count = taskChunkMapper.selectCount(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE));
|
||||
return count != null && count > 0;
|
||||
}
|
||||
|
||||
private Map<String, AppearancePatentResultRowDto> loadPersistedResultRowsWithRetry(Long taskId, int expectedParsedRows) {
|
||||
Map<String, AppearancePatentResultRowDto> result = loadPersistedResultRows(taskId);
|
||||
if (expectedParsedRows <= 0 || !result.isEmpty()) {
|
||||
@@ -2770,6 +3348,7 @@ public class AppearancePatentTaskService {
|
||||
private record SubmitContext(FileTaskEntity task,
|
||||
String scopeKey,
|
||||
String scopeHash,
|
||||
Integer chunkIndex,
|
||||
boolean forceFlush,
|
||||
String error) {
|
||||
}
|
||||
@@ -2780,7 +3359,13 @@ public class AppearancePatentTaskService {
|
||||
Integer chunkIndex,
|
||||
Integer batchIndex,
|
||||
Integer batchTotal,
|
||||
String ownerInstanceId) {
|
||||
String ownerInstanceId,
|
||||
Integer submitRetryCount) {
|
||||
}
|
||||
|
||||
private record CozeCandidate(String chunkScopeHash,
|
||||
Integer chunkIndex,
|
||||
AppearancePatentResultRowDto row) {
|
||||
}
|
||||
|
||||
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<AppearancePatentParsedRowVo> allRows) {
|
||||
|
||||
Reference in New Issue
Block a user