更新处理相关内容

This commit is contained in:
super
2026-05-09 00:20:52 +08:00
parent a2d0bd3c1c
commit 52095eb992
54 changed files with 636585 additions and 21916 deletions

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -11,10 +11,10 @@ public class AppearancePatentProperties {
private String cozeWorkflowHistoryPath = "/v1/workflows/{workflow_id}/run_histories/{execute_id}";
private String cozeWorkflowId = "7632683471312355338";
private String cozeToken = "";
private int cozeBatchSize = 10;
private int cozeBatchSize = 50;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
private int cozePollIntervalMillis = 5000;
private int cozePollIntervalMillis = 30000;
private int cozePollTimeoutMillis = 600000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";

View File

@@ -11,10 +11,10 @@ public class SimilarAsinProperties {
private String cozeWorkflowHistoryPath = "/v1/workflows/{workflow_id}/run_histories/{execute_id}";
private String cozeWorkflowId = "7632683471312355338";
private String cozeToken = "";
private int cozeBatchSize = 10;
private int cozeBatchSize = 50;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
private int cozePollIntervalMillis = 5000;
private int cozePollIntervalMillis = 30000;
private int cozePollTimeoutMillis = 600000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";

View File

@@ -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() {

View File

@@ -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) {

View File

@@ -33,6 +33,7 @@ public class PermissionMenuSchemaInitializer {
new DefaultAdminMenu("店铺管理", "admin_shop_manage", "shop-manage", 50),
new DefaultAdminMenu("跳过跟价ASIN", "admin_skip_price_asin", "skip-price-asin", 60),
new DefaultAdminMenu("查询ASIN", "admin_query_asin", "query-asin", 65),
new DefaultAdminMenu("商品类目", "admin_product_categories", "product-categories", 66),
new DefaultAdminMenu("查看生成记录", "admin_history", "history", 70),
new DefaultAdminMenu("版本管理", "admin_version", "version", 80)
);

View File

@@ -0,0 +1,54 @@
package com.nanri.aiimage.modules.productcategory.controller;
import com.nanri.aiimage.common.api.ApiResponse;
import com.nanri.aiimage.modules.productcategory.model.dto.ProductCategorySaveRequest;
import com.nanri.aiimage.modules.productcategory.model.vo.ProductCategoryItemVo;
import com.nanri.aiimage.modules.productcategory.model.vo.ProductCategoryListVo;
import com.nanri.aiimage.modules.productcategory.service.ProductCategoryService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequiredArgsConstructor
@RequestMapping("/api/admin")
@Tag(name = "商品类目管理", description = "维护商品类目树")
public class ProductCategoryController {
private final ProductCategoryService productCategoryService;
@GetMapping("/product-categories")
@Operation(summary = "查询商品类目树")
public ApiResponse<ProductCategoryListVo> list() {
return ApiResponse.success(productCategoryService.list());
}
@PostMapping("/product-category")
@Operation(summary = "新增商品类目")
public ApiResponse<ProductCategoryItemVo> create(@Valid @RequestBody ProductCategorySaveRequest request) {
return ApiResponse.success("创建成功", productCategoryService.create(request));
}
@PutMapping("/product-category/{id}")
@Operation(summary = "更新商品类目")
public ApiResponse<ProductCategoryItemVo> update(@PathVariable Long id,
@Valid @RequestBody ProductCategorySaveRequest request) {
return ApiResponse.success("保存成功", productCategoryService.update(id, request));
}
@DeleteMapping("/product-category/{id}")
@Operation(summary = "删除商品类目")
public ApiResponse<Void> delete(@PathVariable Long id) {
productCategoryService.delete(id);
return ApiResponse.success("删除成功", null);
}
}

View File

@@ -0,0 +1,9 @@
package com.nanri.aiimage.modules.productcategory.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.productcategory.model.entity.ProductCategoryEntity;
import org.apache.ibatis.annotations.Mapper;
@Mapper
public interface ProductCategoryMapper extends BaseMapper<ProductCategoryEntity> {
}

View File

@@ -0,0 +1,26 @@
package com.nanri.aiimage.modules.productcategory.model.dto;
import com.fasterxml.jackson.annotation.JsonAlias;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
@Data
@Schema(description = "商品类目保存请求")
public class ProductCategorySaveRequest {
@JsonAlias("parent_id")
@Schema(description = "父级类目 ID空表示顶级")
private Long parentId;
@NotBlank(message = "类目名称不能为空")
@Schema(description = "类目名称", requiredMode = Schema.RequiredMode.REQUIRED)
private String name;
@JsonAlias("sort_order")
@Schema(description = "排序值")
private Integer sortOrder;
@Schema(description = "备注")
private String description;
}

View File

@@ -0,0 +1,24 @@
package com.nanri.aiimage.modules.productcategory.model.entity;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.time.LocalDateTime;
@Data
@TableName("biz_product_category")
public class ProductCategoryEntity {
@TableId(type = IdType.AUTO)
private Long id;
private Long parentId;
private String name;
private String categoryKey;
private Integer sortOrder;
private String description;
private Boolean isBuiltin;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
}

View File

@@ -0,0 +1,24 @@
package com.nanri.aiimage.modules.productcategory.model.vo;
import lombok.Data;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
@Data
public class ProductCategoryItemVo {
private Long id;
private Long parentId;
private String name;
private String categoryKey;
private Integer sortOrder;
private String description;
private Boolean isBuiltin;
private Integer childCount;
private Integer level;
private String path;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
private List<ProductCategoryItemVo> children = new ArrayList<>();
}

View File

@@ -0,0 +1,11 @@
package com.nanri.aiimage.modules.productcategory.model.vo;
import lombok.Data;
import java.util.List;
@Data
public class ProductCategoryListVo {
private List<ProductCategoryItemVo> tree;
private List<ProductCategoryItemVo> items;
}

View File

@@ -0,0 +1,237 @@
package com.nanri.aiimage.modules.productcategory.service;
import cn.hutool.crypto.digest.DigestUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.productcategory.mapper.ProductCategoryMapper;
import com.nanri.aiimage.modules.productcategory.model.dto.ProductCategorySaveRequest;
import com.nanri.aiimage.modules.productcategory.model.entity.ProductCategoryEntity;
import com.nanri.aiimage.modules.productcategory.model.vo.ProductCategoryItemVo;
import com.nanri.aiimage.modules.productcategory.model.vo.ProductCategoryListVo;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@Service
@RequiredArgsConstructor
public class ProductCategoryService {
private final ProductCategoryMapper productCategoryMapper;
public ProductCategoryListVo list() {
List<ProductCategoryEntity> rows = productCategoryMapper.selectList(new LambdaQueryWrapper<ProductCategoryEntity>()
.orderByAsc(ProductCategoryEntity::getParentId)
.orderByAsc(ProductCategoryEntity::getSortOrder)
.orderByAsc(ProductCategoryEntity::getId));
return buildListVo(rows);
}
@Transactional
public ProductCategoryItemVo create(ProductCategorySaveRequest request) {
String name = normalizeRequired(request.getName(), "类目名称不能为空");
Long parentId = normalizeParentId(request.getParentId());
ensureParentExists(parentId, null);
ensureSiblingNameUnique(parentId, name, null);
ProductCategoryEntity entity = new ProductCategoryEntity();
entity.setParentId(parentId);
entity.setName(name);
entity.setCategoryKey(generateCategoryKey(parentId, name));
entity.setSortOrder(resolveSortOrder(request.getSortOrder()));
entity.setDescription(normalizeDescription(request.getDescription()));
entity.setIsBuiltin(false);
productCategoryMapper.insert(entity);
return findItem(entity.getId());
}
@Transactional
public ProductCategoryItemVo update(Long id, ProductCategorySaveRequest request) {
ProductCategoryEntity entity = getById(id);
String name = normalizeRequired(request.getName(), "类目名称不能为空");
Long parentId = normalizeParentId(request.getParentId());
ensureParentExists(parentId, id);
ensureSiblingNameUnique(parentId, name, id);
entity.setParentId(parentId);
entity.setName(name);
entity.setSortOrder(resolveSortOrder(request.getSortOrder()));
entity.setDescription(normalizeDescription(request.getDescription()));
productCategoryMapper.updateById(entity);
return findItem(id);
}
@Transactional
public void delete(Long id) {
ProductCategoryEntity entity = getById(id);
Long childCount = productCategoryMapper.selectCount(new LambdaQueryWrapper<ProductCategoryEntity>()
.eq(ProductCategoryEntity::getParentId, entity.getId()));
if (childCount != null && childCount > 0) {
throw new BusinessException("请先删除该类目下的子类目");
}
productCategoryMapper.deleteById(entity.getId());
}
private ProductCategoryItemVo findItem(Long id) {
ProductCategoryListVo listVo = list();
return listVo.getItems().stream()
.filter(item -> id.equals(item.getId()))
.findFirst()
.orElseThrow(() -> new BusinessException("类目不存在"));
}
private ProductCategoryListVo buildListVo(List<ProductCategoryEntity> rows) {
Map<Long, Integer> childCountById = new HashMap<>();
Map<Long, List<ProductCategoryEntity>> byParent = new HashMap<>();
for (ProductCategoryEntity row : rows) {
byParent.computeIfAbsent(row.getParentId(), ignored -> new ArrayList<>()).add(row);
if (row.getParentId() != null) {
childCountById.merge(row.getParentId(), 1, Integer::sum);
}
}
byParent.values().forEach(list -> list.sort(Comparator
.comparing(ProductCategoryEntity::getSortOrder, Comparator.nullsLast(Integer::compareTo))
.thenComparing(ProductCategoryEntity::getId, Comparator.nullsLast(Long::compareTo))));
List<ProductCategoryItemVo> flat = new ArrayList<>();
List<ProductCategoryItemVo> tree = buildChildren(null, 0, "", byParent, childCountById, flat);
ProductCategoryListVo vo = new ProductCategoryListVo();
vo.setTree(tree);
vo.setItems(flat);
return vo;
}
private List<ProductCategoryItemVo> buildChildren(
Long parentId,
int level,
String parentPath,
Map<Long, List<ProductCategoryEntity>> byParent,
Map<Long, Integer> childCountById,
List<ProductCategoryItemVo> flat) {
List<ProductCategoryItemVo> result = new ArrayList<>();
for (ProductCategoryEntity entity : byParent.getOrDefault(parentId, List.of())) {
ProductCategoryItemVo item = toItemVo(entity, childCountById.getOrDefault(entity.getId(), 0));
item.setLevel(level);
item.setPath(parentPath.isBlank() ? item.getName() : parentPath + " / " + item.getName());
flat.add(copyWithoutChildren(item));
item.setChildren(buildChildren(entity.getId(), level + 1, item.getPath(), byParent, childCountById, flat));
result.add(item);
}
return result;
}
private ProductCategoryItemVo toItemVo(ProductCategoryEntity entity, int childCount) {
ProductCategoryItemVo vo = new ProductCategoryItemVo();
vo.setId(entity.getId());
vo.setParentId(entity.getParentId());
vo.setName(entity.getName());
vo.setCategoryKey(entity.getCategoryKey());
vo.setSortOrder(entity.getSortOrder());
vo.setDescription(entity.getDescription());
vo.setIsBuiltin(Boolean.TRUE.equals(entity.getIsBuiltin()));
vo.setChildCount(childCount);
vo.setCreatedAt(entity.getCreatedAt());
vo.setUpdatedAt(entity.getUpdatedAt());
return vo;
}
private ProductCategoryItemVo copyWithoutChildren(ProductCategoryItemVo source) {
ProductCategoryItemVo vo = new ProductCategoryItemVo();
vo.setId(source.getId());
vo.setParentId(source.getParentId());
vo.setName(source.getName());
vo.setCategoryKey(source.getCategoryKey());
vo.setSortOrder(source.getSortOrder());
vo.setDescription(source.getDescription());
vo.setIsBuiltin(source.getIsBuiltin());
vo.setChildCount(source.getChildCount());
vo.setLevel(source.getLevel());
vo.setPath(source.getPath());
vo.setCreatedAt(source.getCreatedAt());
vo.setUpdatedAt(source.getUpdatedAt());
return vo;
}
private ProductCategoryEntity getById(Long id) {
ProductCategoryEntity entity = productCategoryMapper.selectById(id);
if (entity == null) {
throw new BusinessException("类目不存在");
}
return entity;
}
private void ensureParentExists(Long parentId, Long currentId) {
if (parentId == null) {
return;
}
if (currentId != null && currentId.equals(parentId)) {
throw new BusinessException("父级类目不能选择自己");
}
ProductCategoryEntity parent = productCategoryMapper.selectById(parentId);
if (parent == null) {
throw new BusinessException("父级类目不存在");
}
Set<Long> visited = new HashSet<>();
Long cursor = parent.getParentId();
while (cursor != null) {
if (!visited.add(cursor)) {
throw new BusinessException("类目层级存在循环");
}
if (currentId != null && currentId.equals(cursor)) {
throw new BusinessException("父级类目不能选择自己的子级");
}
ProductCategoryEntity row = productCategoryMapper.selectById(cursor);
cursor = row == null ? null : row.getParentId();
}
}
private void ensureSiblingNameUnique(Long parentId, String name, Long excludeId) {
LambdaQueryWrapper<ProductCategoryEntity> query = new LambdaQueryWrapper<ProductCategoryEntity>()
.eq(ProductCategoryEntity::getName, name);
if (parentId == null) {
query.isNull(ProductCategoryEntity::getParentId);
} else {
query.eq(ProductCategoryEntity::getParentId, parentId);
}
if (excludeId != null) {
query.ne(ProductCategoryEntity::getId, excludeId);
}
Long count = productCategoryMapper.selectCount(query);
if (count != null && count > 0) {
throw new BusinessException("同级类目名称已存在");
}
}
private String normalizeRequired(String value, String message) {
String normalized = value == null ? "" : value.trim();
if (normalized.isBlank()) {
throw new BusinessException(message);
}
return normalized;
}
private Long normalizeParentId(Long value) {
return value == null || value <= 0 ? null : value;
}
private Integer resolveSortOrder(Integer value) {
return value == null ? 100 : value;
}
private String normalizeDescription(String value) {
String normalized = value == null ? "" : value.trim();
return normalized.length() > 512 ? normalized.substring(0, 512) : normalized;
}
private String generateCategoryKey(Long parentId, String name) {
String source = (parentId == null ? "root" : String.valueOf(parentId)) + ":" + name;
return "custom_" + DigestUtil.sha1Hex(source).substring(0, 20);
}
}

View File

@@ -1,9 +1,15 @@
package com.nanri.aiimage.modules.productrisk.model.vo;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.nanri.aiimage.modules.shopkey.model.dto.SkipPriceAsinDetailDto;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@Data
@Schema(description = "单店结果项:列表接口与任务详情中共用,对应 biz_file_result 一行 + 任务/请求衍生字段")
public class ProductRiskResultItemVo {
@@ -65,4 +71,12 @@ public class ProductRiskResultItemVo {
@JsonProperty("scheduledAt")
@Schema(description = "定时执行时间,未设置时为空")
private String scheduledAt;
@JsonAlias("skip_asins_by_country")
@JsonProperty("skipAsinsByCountry")
private Map<String, List<String>> skipAsinsByCountry = new LinkedHashMap<>();
@JsonAlias("skip_asin_details_by_country")
@JsonProperty("skipAsinDetailsByCountry")
private Map<String, List<SkipPriceAsinDetailDto>> skipAsinDetailsByCountry = new LinkedHashMap<>();
}

View File

@@ -1,12 +1,16 @@
package com.nanri.aiimage.modules.productrisk.model.vo;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.nanri.aiimage.modules.queryasin.model.dto.QueryAsinCountryAsinsDto;
import com.nanri.aiimage.modules.shopkey.model.dto.SkipPriceAsinDetailDto;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* 与 Python 队列约定字段对齐,供前端原样推入 enqueue_json。
@@ -52,4 +56,12 @@ public class ProductRiskShopQueueItemVo {
@JsonProperty("queryAsins")
@Schema(description = "查询 ASIN 模块返回的后台维护 ASIN 数据;按全表聚合返回,不按当前店铺过滤")
private List<QueryAsinCountryAsinsDto> queryAsins = new ArrayList<>();
@JsonAlias("skip_asins_by_country")
@JsonProperty("skipAsinsByCountry")
private Map<String, List<String>> skipAsinsByCountry = new LinkedHashMap<>();
@JsonAlias("skip_asin_details_by_country")
@JsonProperty("skipAsinDetailsByCountry")
private Map<String, List<SkipPriceAsinDetailDto>> skipAsinDetailsByCountry = new LinkedHashMap<>();
}

View File

@@ -0,0 +1,17 @@
package com.nanri.aiimage.modules.shopkey.model.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
@Data
@Schema(description = "跳过 ASIN 明细")
public class SkipPriceAsinDetailDto {
@Schema(description = "ASIN")
private String asin;
@JsonProperty("minimumPrice")
@Schema(description = "最低价,未配置时为空字符串")
private String minimumPrice;
}

View File

@@ -6,6 +6,7 @@ import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.shopkey.mapper.SkipPriceAsinMapper;
import com.nanri.aiimage.modules.shopkey.mapper.ShopManageMapper;
import com.nanri.aiimage.modules.shopkey.model.dto.SkipPriceAsinCreateRequest;
import com.nanri.aiimage.modules.shopkey.model.dto.SkipPriceAsinDetailDto;
import com.nanri.aiimage.modules.shopkey.model.entity.ShopManageEntity;
import com.nanri.aiimage.modules.shopkey.model.entity.ShopManageGroupEntity;
import com.nanri.aiimage.modules.shopkey.model.entity.SkipPriceAsinEntity;
@@ -833,6 +834,72 @@ public class SkipPriceAsinService {
return result;
}
public Map<String, Map<String, List<SkipPriceAsinDetailDto>>> listSkipAsinDetailsByShopAndCountries(
List<String> shopNames, List<String> countryCodes) {
if (shopNames == null || shopNames.isEmpty()) {
return Map.of();
}
LinkedHashSet<String> normalizedShopNames = new LinkedHashSet<>();
for (String shopName : shopNames) {
String normalized = normalizeBlank(shopName);
if (!normalized.isEmpty()) {
normalizedShopNames.add(normalized);
}
}
if (normalizedShopNames.isEmpty()) {
return Map.of();
}
LinkedHashSet<String> normalizedCountries = new LinkedHashSet<>();
if (countryCodes != null) {
for (String countryCode : countryCodes) {
String normalized = normalizeBlank(countryCode).toUpperCase(Locale.ROOT);
if (SUPPORTED_COUNTRIES.contains(normalized)) {
normalizedCountries.add(normalized);
}
}
}
if (normalizedCountries.isEmpty()) {
normalizedCountries.addAll(SUPPORTED_COUNTRIES);
}
List<SkipPriceAsinEntity> rows = skipPriceAsinMapper.selectList(new LambdaQueryWrapper<SkipPriceAsinEntity>()
.in(SkipPriceAsinEntity::getShopName, normalizedShopNames)
.orderByDesc(SkipPriceAsinEntity::getId));
Map<String, Map<String, LinkedHashMap<String, SkipPriceAsinDetailDto>>> grouped = new LinkedHashMap<>();
for (SkipPriceAsinEntity row : rows) {
String shopName = normalizeBlank(row.getShopName());
if (shopName.isEmpty()) {
continue;
}
Map<String, LinkedHashMap<String, SkipPriceAsinDetailDto>> countryMap =
grouped.computeIfAbsent(shopName, ignored -> new LinkedHashMap<>());
for (String country : normalizedCountries) {
String asin = getCountryAsin(row, country);
if (asin.isEmpty()) {
continue;
}
SkipPriceAsinDetailDto detail = new SkipPriceAsinDetailDto();
detail.setAsin(asin);
BigDecimal minimumPrice = getCountryMinimumPrice(row, country);
detail.setMinimumPrice(minimumPrice == null ? "" : minimumPrice.toPlainString());
countryMap.computeIfAbsent(country, ignored -> new LinkedHashMap<>()).putIfAbsent(asin, detail);
}
}
Map<String, Map<String, List<SkipPriceAsinDetailDto>>> result = new LinkedHashMap<>();
for (Map.Entry<String, Map<String, LinkedHashMap<String, SkipPriceAsinDetailDto>>> shopEntry : grouped.entrySet()) {
Map<String, List<SkipPriceAsinDetailDto>> countryResult = new LinkedHashMap<>();
for (String country : normalizedCountries) {
LinkedHashMap<String, SkipPriceAsinDetailDto> details = shopEntry.getValue().get(country);
if (details != null && !details.isEmpty()) {
countryResult.put(country, List.copyOf(details.values()));
}
}
result.put(shopEntry.getKey(), countryResult);
}
return result;
}
public Map<String, List<String>> listAllSkipAsinsByCountry() {
List<SkipPriceAsinEntity> rows = skipPriceAsinMapper.selectList(
new LambdaQueryWrapper<SkipPriceAsinEntity>()

View File

@@ -1,5 +1,6 @@
package com.nanri.aiimage.modules.shopmatch.model.dto;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
@@ -11,6 +12,11 @@ public class ShopMatchRowDto {
@Schema(description = "ASIN")
private String asin;
@JsonAlias("minimum_price")
@JsonProperty("minimumPrice")
@Schema(description = "最低价")
private String minimumPrice;
@Schema(description = "状态")
private String status;

View File

@@ -1,7 +1,9 @@
package com.nanri.aiimage.modules.shopmatch.model.dto;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.nanri.aiimage.modules.productrisk.model.enums.ProductRiskCountryCode;
import com.nanri.aiimage.modules.shopkey.model.dto.SkipPriceAsinDetailDto;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
@@ -22,4 +24,14 @@ public class ShopMatchShopPayloadDto {
@Schema(description = "按国家分组的 ASIN/状态 列表")
private Map<ProductRiskCountryCode, List<ShopMatchRowDto>> countries = new LinkedHashMap<>();
@JsonAlias("skip_asin_details_by_country")
@JsonProperty("skipAsinDetailsByCountry")
@Schema(description = "Python 回传的跳过 ASIN 明细,按国家携带 asin 和最低价")
private Map<ProductRiskCountryCode, List<SkipPriceAsinDetailDto>> skipAsinDetailsByCountry = new LinkedHashMap<>();
@JsonAlias("skip_asins_by_country")
@JsonProperty("skipAsinsByCountry")
@Schema(description = "Python 回传的跳过 ASIN 列表,兼容旧格式")
private Map<ProductRiskCountryCode, List<String>> skipAsinsByCountry = new LinkedHashMap<>();
}

View File

@@ -30,8 +30,9 @@ public class ShopMatchExcelAssemblyService {
String sheetName = code.getSheetName();
Sheet sheet = workbook.createSheet(sheetName);
Row header = sheet.createRow(0);
header.createCell(0).setCellValue(HEADER[0]);
header.createCell(1).setCellValue(HEADER[1]);
header.createCell(0).setCellValue("ASIN");
header.createCell(1).setCellValue("最低价");
header.createCell(2).setCellValue("状态");
List<ShopMatchRowDto> rows = safe.getOrDefault(sheetName, List.of());
int rowIndex = 1;
for (ShopMatchRowDto item : rows) {
@@ -40,10 +41,12 @@ public class ShopMatchExcelAssemblyService {
}
Row row = sheet.createRow(rowIndex++);
row.createCell(0).setCellValue(item.getAsin() == null ? "" : item.getAsin());
row.createCell(1).setCellValue(item.getStatus() == null ? "" : item.getStatus());
row.createCell(1).setCellValue(item.getMinimumPrice() == null ? "" : item.getMinimumPrice());
row.createCell(2).setCellValue(item.getStatus() == null ? "" : item.getStatus());
}
sheet.setColumnWidth(0, 20 * 256);
sheet.setColumnWidth(1, 18 * 256);
sheet.setColumnWidth(2, 18 * 256);
}
workbook.write(fos);
} catch (Exception ex) {

View File

@@ -24,6 +24,8 @@ import com.nanri.aiimage.modules.shopmatch.model.dto.ShopMatchShopPayloadDto;
import com.nanri.aiimage.modules.shopmatch.model.dto.ShopMatchStageCompleteRequest;
import com.nanri.aiimage.modules.shopmatch.model.dto.ShopMatchSubmitResultRequest;
import com.nanri.aiimage.modules.shopmatch.model.entity.ShopMatchShopCandidateEntity;
import com.nanri.aiimage.modules.shopkey.model.dto.SkipPriceAsinDetailDto;
import com.nanri.aiimage.modules.shopkey.service.SkipPriceAsinService;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
@@ -69,6 +71,7 @@ public class ShopMatchTaskService {
private final TaskResultPayloadService taskResultPayloadService;
private final TaskFileJobService taskFileJobService;
private final TaskDistributedLockService taskDistributedLockService;
private final SkipPriceAsinService skipPriceAsinService;
private FileTaskEntity loadTaskForExecution(Long taskId) {
Map<Long, FileTaskEntity> cachedTasks = shopMatchTaskCacheService.getTaskCacheBatch(List.of(taskId));
@@ -381,6 +384,7 @@ public class ShopMatchTaskService {
}
List<String> countryCodes = normalizeCountryCodes(request.getCountryCodes());
enrichItemsWithSkipAsins(uniqueItems, countryCodes);
List<LocalDateTime> scheduleTimes = normalizeScheduleTimes(request.getScheduleTimes());
LocalDateTime now = now();
@@ -962,6 +966,8 @@ public class ShopMatchTaskService {
vo.setMatched(item.isMatched());
vo.setMatchStatus(item.getMatchStatus());
vo.setMatchMessage(item.getMatchMessage());
vo.setSkipAsinsByCountry(item.getSkipAsinsByCountry());
vo.setSkipAsinDetailsByCountry(item.getSkipAsinDetailsByCountry());
vo.setTaskStatus(taskStatus);
vo.setSuccess(false);
vo.setError(null);
@@ -1078,6 +1084,8 @@ public class ShopMatchTaskService {
vo.setCompanyName(item.getCompanyName());
vo.setMatchStatus(item.getMatchStatus());
vo.setMatchMessage(item.getMatchMessage());
vo.setSkipAsinsByCountry(item.getSkipAsinsByCountry());
vo.setSkipAsinDetailsByCountry(item.getSkipAsinDetailsByCountry());
break;
}
}
@@ -1164,6 +1172,36 @@ public class ShopMatchTaskService {
return new ArrayList<>(byNorm.values());
}
private void enrichItemsWithSkipAsins(List<ProductRiskShopQueueItemVo> items, List<String> countryCodes) {
List<String> shopNames = items.stream()
.map(ProductRiskShopQueueItemVo::getShopName)
.filter(name -> name != null && !name.isBlank())
.distinct()
.toList();
Map<String, Map<String, List<SkipPriceAsinDetailDto>>> detailsByShop =
skipPriceAsinService.listSkipAsinDetailsByShopAndCountries(shopNames, countryCodes);
for (ProductRiskShopQueueItemVo item : items) {
Map<String, List<SkipPriceAsinDetailDto>> detailMap =
detailsByShop.getOrDefault(item.getShopName(), Map.of());
Map<String, List<SkipPriceAsinDetailDto>> normalizedDetails = new LinkedHashMap<>();
Map<String, List<String>> asinsByCountry = new LinkedHashMap<>();
for (String countryCode : countryCodes) {
List<SkipPriceAsinDetailDto> details = detailMap.getOrDefault(countryCode, List.of());
if (details.isEmpty()) {
continue;
}
normalizedDetails.put(countryCode, details);
asinsByCountry.put(countryCode, details.stream()
.map(SkipPriceAsinDetailDto::getAsin)
.filter(asin -> asin != null && !asin.isBlank())
.distinct()
.toList());
}
item.setSkipAsinDetailsByCountry(normalizedDetails);
item.setSkipAsinsByCountry(asinsByCountry);
}
}
private List<String> normalizeCountryCodes(List<String> raw) {
if (raw == null || raw.isEmpty()) {
throw new BusinessException("country_codes 不能为空");
@@ -1234,6 +1272,7 @@ public class ShopMatchTaskService {
if (incoming.getError() != null && !incoming.getError().isBlank()) {
merged.setError(incoming.getError().trim());
}
mergeSkipAsinPayload(merged, incoming);
Map<com.nanri.aiimage.modules.productrisk.model.enums.ProductRiskCountryCode, List<ShopMatchRowDto>> nextCountries =
merged.getCountries() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(merged.getCountries());
if (incoming.getCountries() != null && !incoming.getCountries().isEmpty()) {
@@ -1245,10 +1284,131 @@ public class ShopMatchTaskService {
}
}
merged.setCountries(nextCountries);
appendSkipAsinRows(merged);
shopMatchTaskCacheService.saveShopMergedPayload(taskId, shopKey, merged);
return merged;
}
private void mergeSkipAsinPayload(ShopMatchShopPayloadDto merged, ShopMatchShopPayloadDto incoming) {
Map<ProductRiskCountryCode, List<SkipPriceAsinDetailDto>> nextDetails =
merged.getSkipAsinDetailsByCountry() == null ? new LinkedHashMap<>()
: new LinkedHashMap<>(merged.getSkipAsinDetailsByCountry());
if (incoming.getSkipAsinDetailsByCountry() != null && !incoming.getSkipAsinDetailsByCountry().isEmpty()) {
for (Map.Entry<ProductRiskCountryCode, List<SkipPriceAsinDetailDto>> entry : incoming.getSkipAsinDetailsByCountry().entrySet()) {
if (entry.getKey() == null || entry.getValue() == null || entry.getValue().isEmpty()) {
continue;
}
nextDetails.put(entry.getKey(), mergeSkipAsinDetails(nextDetails.get(entry.getKey()), entry.getValue()));
}
}
Map<ProductRiskCountryCode, List<String>> nextAsins =
merged.getSkipAsinsByCountry() == null ? new LinkedHashMap<>()
: new LinkedHashMap<>(merged.getSkipAsinsByCountry());
if (incoming.getSkipAsinsByCountry() != null && !incoming.getSkipAsinsByCountry().isEmpty()) {
for (Map.Entry<ProductRiskCountryCode, List<String>> entry : incoming.getSkipAsinsByCountry().entrySet()) {
if (entry.getKey() == null || entry.getValue() == null || entry.getValue().isEmpty()) {
continue;
}
nextAsins.put(entry.getKey(), mergeSkipAsins(nextAsins.get(entry.getKey()), entry.getValue()));
}
}
merged.setSkipAsinDetailsByCountry(nextDetails);
merged.setSkipAsinsByCountry(nextAsins);
}
private List<SkipPriceAsinDetailDto> mergeSkipAsinDetails(List<SkipPriceAsinDetailDto> existing,
List<SkipPriceAsinDetailDto> incoming) {
LinkedHashMap<String, SkipPriceAsinDetailDto> byAsin = new LinkedHashMap<>();
if (existing != null) {
for (SkipPriceAsinDetailDto detail : existing) {
putSkipAsinDetail(byAsin, detail);
}
}
if (incoming != null) {
for (SkipPriceAsinDetailDto detail : incoming) {
putSkipAsinDetail(byAsin, detail);
}
}
return new ArrayList<>(byAsin.values());
}
private void putSkipAsinDetail(Map<String, SkipPriceAsinDetailDto> byAsin, SkipPriceAsinDetailDto detail) {
String asin = normalizeAsin(detail == null ? null : detail.getAsin());
if (asin.isBlank()) {
return;
}
SkipPriceAsinDetailDto normalized = new SkipPriceAsinDetailDto();
normalized.setAsin(asin);
normalized.setMinimumPrice(detail.getMinimumPrice() == null ? "" : detail.getMinimumPrice().trim());
byAsin.put(asin, normalized);
}
private List<String> mergeSkipAsins(List<String> existing, List<String> incoming) {
LinkedHashSet<String> asins = new LinkedHashSet<>();
if (existing != null) {
existing.stream().map(this::normalizeAsin).filter(asin -> !asin.isBlank()).forEach(asins::add);
}
if (incoming != null) {
incoming.stream().map(this::normalizeAsin).filter(asin -> !asin.isBlank()).forEach(asins::add);
}
return new ArrayList<>(asins);
}
private void appendSkipAsinRows(ShopMatchShopPayloadDto payload) {
Map<ProductRiskCountryCode, List<ShopMatchRowDto>> countries =
payload.getCountries() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(payload.getCountries());
Map<ProductRiskCountryCode, List<SkipPriceAsinDetailDto>> detailsByCountry =
payload.getSkipAsinDetailsByCountry() == null ? Map.of() : payload.getSkipAsinDetailsByCountry();
Map<ProductRiskCountryCode, List<String>> asinsByCountry =
payload.getSkipAsinsByCountry() == null ? Map.of() : payload.getSkipAsinsByCountry();
LinkedHashSet<ProductRiskCountryCode> countryCodes = new LinkedHashSet<>();
countryCodes.addAll(detailsByCountry.keySet());
countryCodes.addAll(asinsByCountry.keySet());
for (ProductRiskCountryCode countryCode : countryCodes) {
if (countryCode == null) {
continue;
}
LinkedHashMap<String, ShopMatchRowDto> rowsByAsin = new LinkedHashMap<>();
List<ShopMatchRowDto> existingRows = countries.get(countryCode);
if (existingRows != null) {
for (ShopMatchRowDto row : existingRows) {
if (row == null || isEmptyBusinessRow(row)) {
continue;
}
rowsByAsin.put(normalizeAsin(row.getAsin()), cloneRow(row));
}
}
for (SkipPriceAsinDetailDto detail : detailsByCountry.getOrDefault(countryCode, List.of())) {
upsertSkipAsinRow(rowsByAsin, detail == null ? null : detail.getAsin(),
detail == null ? null : detail.getMinimumPrice());
}
for (String asin : asinsByCountry.getOrDefault(countryCode, List.of())) {
upsertSkipAsinRow(rowsByAsin, asin, null);
}
countries.put(countryCode, new ArrayList<>(rowsByAsin.values()));
}
payload.setCountries(countries);
}
private void upsertSkipAsinRow(Map<String, ShopMatchRowDto> rowsByAsin, String asin, String minimumPrice) {
String normalizedAsin = normalizeAsin(asin);
if (normalizedAsin.isBlank()) {
return;
}
ShopMatchRowDto row = rowsByAsin.get(normalizedAsin);
if (row == null) {
row = new ShopMatchRowDto();
row.setAsin(normalizedAsin);
rowsByAsin.put(normalizedAsin, row);
}
if (minimumPrice != null && !minimumPrice.isBlank()) {
row.setMinimumPrice(minimumPrice.trim());
}
row.setStatus("跳过");
}
private List<ShopMatchRowDto> mergeCountryRows(List<ShopMatchRowDto> existingRows, List<ShopMatchRowDto> incomingRows) {
LinkedHashMap<String, ShopMatchRowDto> byKey = new LinkedHashMap<>();
if (existingRows != null) {
@@ -1289,6 +1449,9 @@ public class ShopMatchTaskService {
if (incoming.getAsin() != null && !incoming.getAsin().isBlank()) {
merged.setAsin(incoming.getAsin());
}
if (incoming.getMinimumPrice() != null && !incoming.getMinimumPrice().isBlank()) {
merged.setMinimumPrice(incoming.getMinimumPrice());
}
if (incoming.getStatus() != null && !incoming.getStatus().isBlank()) {
merged.setStatus(incoming.getStatus());
}
@@ -1300,12 +1463,17 @@ public class ShopMatchTaskService {
ShopMatchRowDto out = new ShopMatchRowDto();
if (row != null) {
out.setAsin(row.getAsin());
out.setMinimumPrice(row.getMinimumPrice());
out.setStatus(row.getStatus());
out.setDone(row.getDone());
}
return out;
}
private String normalizeAsin(String asin) {
return asin == null ? "" : asin.trim().toUpperCase(Locale.ROOT);
}
private boolean isShopPayloadCompleted(ShopMatchShopPayloadDto payload) {
return countPayloadRows(payload) > 0 && payloadHasAnyDoneTrue(payload);
}
@@ -1329,9 +1497,12 @@ public class ShopMatchTaskService {
}
private boolean payloadHasAnyDoneTrue(ShopMatchShopPayloadDto payload) {
if (payload == null || payload.getCountries() == null) {
if (payload == null) {
return false;
}
if (payload.getCountries() == null || payload.getCountries().isEmpty()) {
return hasSkipAsinPayload(payload);
}
for (List<ShopMatchRowDto> rows : payload.getCountries().values()) {
if (rows == null) {
continue;
@@ -1345,6 +1516,27 @@ public class ShopMatchTaskService {
return false;
}
private boolean hasSkipAsinPayload(ShopMatchShopPayloadDto payload) {
if (payload == null) {
return false;
}
if (payload.getSkipAsinDetailsByCountry() != null) {
for (List<SkipPriceAsinDetailDto> details : payload.getSkipAsinDetailsByCountry().values()) {
if (details != null && details.stream().anyMatch(detail -> detail != null && !normalizeAsin(detail.getAsin()).isBlank())) {
return true;
}
}
}
if (payload.getSkipAsinsByCountry() != null) {
for (List<String> asins : payload.getSkipAsinsByCountry().values()) {
if (asins != null && asins.stream().anyMatch(asin -> !normalizeAsin(asin).isBlank())) {
return true;
}
}
}
return false;
}
private boolean isEmptyBusinessRow(ShopMatchRowDto row) {
return row == null || row.getAsin() == null || row.getAsin().trim().isEmpty();
}

View File

@@ -572,6 +572,11 @@ public class SimilarAsinCozeClient {
|| message.contains("Read timed out")
|| message.contains("Connection reset")
|| message.contains("I/O error on POST request")
|| message.contains("429")
|| message.toLowerCase(Locale.ROOT).contains("rate limit")
|| message.toLowerCase(Locale.ROOT).contains("retry later")
|| message.contains("\u9650\u6d41")
|| message.contains("\u7a0d\u540e\u91cd\u8bd5")
|| message.toLowerCase(Locale.ROOT).contains("timeout");
}
@@ -779,7 +784,9 @@ public class SimilarAsinCozeClient {
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"))));
String message = firstNonBlank(
findTextByFieldName(dataNode, "error_message", "error", "msg"),
findTextByFieldName(root, "error_message", "error", "msg"));
return message == null ? "" : message;
}

View File

@@ -15,5 +15,6 @@ public class SimilarAsinFilterConditionEntity {
private Long id;
private Long userId;
private String conditionText;
private String conditionHash;
private LocalDateTime createdAt;
}

View File

@@ -106,6 +106,11 @@ public class SimilarAsinTaskService {
private static final long RESULT_ROWS_READ_RETRY_DELAY_MS = 500L;
private static final Duration TASK_LOCK_TTL = Duration.ofMinutes(5);
private static final long TASK_LOCK_WAIT_MILLIS = 10000L;
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 int PARSE_RESPONSE_PREVIEW_LIMIT = 100;
private static final List<String> RESULT_HEADERS = List.of(
"id",
@@ -159,9 +164,11 @@ public class SimilarAsinTaskService {
if (text.isBlank()) {
throw new BusinessException("筛选条件不能为空");
}
String conditionHash = DigestUtil.sha256Hex(text);
SimilarAsinFilterConditionEntity existing = filterConditionMapper.selectOne(
new LambdaQueryWrapper<SimilarAsinFilterConditionEntity>()
.eq(SimilarAsinFilterConditionEntity::getUserId, request.getUserId())
.eq(SimilarAsinFilterConditionEntity::getConditionHash, conditionHash)
.eq(SimilarAsinFilterConditionEntity::getConditionText, text)
.last("limit 1"));
if (existing != null) {
@@ -170,8 +177,22 @@ public class SimilarAsinTaskService {
SimilarAsinFilterConditionEntity entity = new SimilarAsinFilterConditionEntity();
entity.setUserId(request.getUserId());
entity.setConditionText(text);
entity.setConditionHash(conditionHash);
entity.setCreatedAt(LocalDateTime.now());
filterConditionMapper.insert(entity);
try {
filterConditionMapper.insert(entity);
} catch (DuplicateKeyException ex) {
existing = filterConditionMapper.selectOne(
new LambdaQueryWrapper<SimilarAsinFilterConditionEntity>()
.eq(SimilarAsinFilterConditionEntity::getUserId, request.getUserId())
.eq(SimilarAsinFilterConditionEntity::getConditionHash, conditionHash)
.eq(SimilarAsinFilterConditionEntity::getConditionText, text)
.last("limit 1"));
if (existing != null) {
return toFilterConditionVo(existing);
}
throw ex;
}
return toFilterConditionVo(entity);
}
@@ -461,6 +482,7 @@ public class SimilarAsinTaskService {
completeSubmittedChunk(context);
return null;
});
submitCozeForSubmittedChunk(context);
return;
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
@@ -538,12 +560,15 @@ public class SimilarAsinTaskService {
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
@@ -706,7 +731,7 @@ public class SimilarAsinTaskService {
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) {
@@ -728,6 +753,43 @@ public class SimilarAsinTaskService {
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;
}
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, task.getId())
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.eq(TaskChunkEntity::getScopeHash, context.scopeHash())
.eq(TaskChunkEntity::getChunkIndex, context.chunkIndex())
.last("limit 1"));
if (chunk == null || readChunkRows(chunk).isEmpty()) {
return;
}
FileResultEntity result = findOrCreateResultRecordForAssembly(task, allRowCount(task));
if (result == null) {
return;
}
TaskFileJobEntity job = taskFileJobService.enqueueAssembleResult(
task.getId(), MODULE_TYPE, result.getId(), buildTaskOwnerScopeKey(task));
if (job == null || "SUCCESS".equals(job.getStatus())) {
return;
}
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
boolean pendingCoze = submitCozeBatches(task, result, job, List.of(chunk), allRowsByBaseId);
if (pendingCoze) {
taskFileJobService.touchRunning(job.getId());
touchJavaSideTaskActivity(task.getId());
} else if (isResultSubmissionComplete(task.getId())) {
maybeFinalizeCozeJobLocked(task.getId(), new CozeBatchContext(
job.getId(), result.getId(), chunk.getScopeHash(), chunk.getChunkIndex(), 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())) {
@@ -1127,13 +1189,19 @@ public class SimilarAsinTaskService {
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) {
@@ -1155,20 +1223,67 @@ public class SimilarAsinTaskService {
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));
if (shouldAssembleResult) {
enqueueResultAssembly(task, result, true);
} else if (assembleWorkbook && failed) {
log.warn("[similar-asin] 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));
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) {
SimilarAsinParsedPayloadDto payload = readParsedPayload(task);
List<SimilarAsinSourceFileDto> 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);
result.setUserId(task.getUserId());
result.setCreatedAt(LocalDateTime.now());
fileResultMapper.insert(result);
return result;
}
private boolean shouldAssembleSynchronously() {
return false;
}
@@ -1197,6 +1312,12 @@ public class SimilarAsinTaskService {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
int cozeWorkUnits = countCozeWorkUnits(chunks, Math.max(1, properties.getCozeBatchSize()));
int totalProgressUnits = Math.max(3, cozeWorkUnits + 3);
if (countPendingCozeStates(task.getId()) > 0) {
taskFileJobService.touchRunning(job.getId());
touchJavaSideTaskActivity(task.getId());
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze submitted, waiting for result");
return false;
}
saveFileBuildProgress(task, job, totalProgressUnits, 0, "Submitting Coze");
boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId);
if (pendingCoze) {
@@ -1205,11 +1326,17 @@ public class SimilarAsinTaskService {
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze submitted, waiting for result");
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), "Waiting for Python upload");
return false;
}
completeCozeFileJob(task, result, job, totalProgressUnits, cozeWorkUnits);
return true;
}
@Scheduled(fixedDelayString = "${aiimage.similar-asin.coze-poll-delay-ms:5000}")
@Scheduled(fixedDelayString = "${aiimage.similar-asin.coze-poll-delay-ms:30000}")
public void pollPendingCozeJobs() {
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("similar-asin:coze-poll", Duration.ofMinutes(1));
@@ -1299,7 +1426,7 @@ public class SimilarAsinTaskService {
|| COZE_STATUS_RUNNING.equals(existing.getCozeStatus());
}
try {
SimilarAsinCozeClient.CozeSubmitResponse submit = cozeClient.submitWorkflow(batchRows, prompt, apiKey);
SimilarAsinCozeClient.CozeSubmitResponse submit = submitCozeWorkflowThrottled(batchRows, prompt, apiKey);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows = cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
mergeCozeRowsIntoChunk(task, chunk.getScopeHash(), chunk.getChunkIndex(), cozeRows, allRowsByBaseId);
@@ -1349,7 +1476,8 @@ public class SimilarAsinTaskService {
chunk.getChunkIndex(),
batchIndex,
batchTotal,
currentInstanceId()
currentInstanceId(),
0
);
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
String storedBatchPayload = transientPayloadStorageService.storeParsedPayloadFast(
@@ -1442,6 +1570,12 @@ public class SimilarAsinTaskService {
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<SimilarAsinResultRowDto> cozeRows = failureMessage.isBlank()
? cozeClient.mergeRowsFromDataText(batchRows, poll.resolvedPayloadText())
: cozeClient.markRowsFailed(batchRows, failureMessage);
@@ -1477,6 +1611,241 @@ public class SimilarAsinTaskService {
}
}
private boolean retryFailedCozeBatchState(TaskScopeStateEntity state,
CozeBatchContext context,
List<SimilarAsinResultRowDto> 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 {
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task));
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows =
cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
markCozeStateTerminal(state, COZE_STATUS_DONE, null);
maybeFinalizeCozeJob(state.getTaskId(), context);
log.info("[similar-asin] 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("[similar-asin] 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("[similar-asin] coze retry submit failed taskId={} stateId={} executeId={} err={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), firstNonBlank(ex.getMessage(), "Coze retry failed"));
}
return false;
}
private SimilarAsinCozeClient.CozeSubmitResponse submitCozeWorkflowThrottled(
List<SimilarAsinResultRowDto> rows,
String prompt,
String apiKey) throws Exception {
DistributedJobLockService.LockHandle lockHandle = acquireCozeSubmitLock();
if (lockHandle == null) {
throw new IllegalStateException("Coze submit throttle lock timeout");
}
try (lockHandle) {
SimilarAsinCozeClient.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("similar-asin: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<SimilarAsinResultRowDto> 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<SimilarAsinResultRowDto>> partitions = List.<List<SimilarAsinResultRowDto>>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<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
int partIndex = 1;
for (List<SimilarAsinResultRowDto> partRows : partitions) {
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(partRows, readAiPrompt(task), readApiKey(task));
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows =
cozeClient.mergeRowsFromDataText(partRows, submit.immediateData());
mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), 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());
maybeFinalizeCozeJob(state.getTaskId(), context);
log.info("[similar-asin] 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("[similar-asin] 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<SimilarAsinResultRowDto> 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 = transientPayloadStorageService.storeParsedPayloadFast(
MODULE_TYPE, parent.getTaskId(), scopeHash, batchPayload, true);
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("[similar-asin] 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())
@@ -1538,22 +1907,7 @@ public class SimilarAsinTaskService {
return;
}
try (lockHandle) {
if (countPendingCozeStates(taskId) > 0) {
return;
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
FileResultEntity result = fileResultMapper.selectById(context.resultId());
TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId());
if (task == null || result == null || job == null || "SUCCESS".equals(job.getStatus())) {
return;
}
int cozeWorkUnits = countCompletedCozeStates(taskId);
int totalProgressUnits = Math.max(3, cozeWorkUnits + 3);
completeCozeFileJob(task, result, job, totalProgressUnits, cozeWorkUnits);
taskFileJobService.markSuccess(job, result.getResultFileUrl());
cleanupResultFileJob(job);
log.info("[similar-asin] coze async job finalized taskId={} jobId={} resultId={} resultFileUrl={}",
taskId, job.getId(), result.getId(), result.getResultFileUrl());
maybeFinalizeCozeJobLocked(taskId, context);
} catch (Exception ex) {
TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId());
if (job != null) {
@@ -1564,6 +1918,37 @@ public class SimilarAsinTaskService {
}
}
private void maybeFinalizeCozeJobLocked(Long taskId, CozeBatchContext context) {
if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) {
return;
}
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("similar-asin:coze-finalize:" + taskId, Duration.ofMinutes(5));
if (lockHandle == null) {
return;
}
try (lockHandle) {
if (countPendingCozeStates(taskId) > 0) {
return;
}
TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId());
if (job == null || "SUCCESS".equals(job.getStatus())) {
return;
}
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 results ready, assembling xlsx");
if (requeued) {
log.info("[similar-asin] coze async results ready, result file job requeued taskId={} jobId={} resultId={}",
taskId, job.getId(), context.resultId());
}
}
}
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, TASK_LOCK_TTL, waitMillis);
if (lockHandle == null) {
@@ -1593,8 +1978,13 @@ public class SimilarAsinTaskService {
assembleResultWorkbook(task, result);
saveFileBuildProgress(task, job, totalProgressUnits, totalProgressUnits - 1, "Uploading result file");
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);
@@ -1655,6 +2045,23 @@ public class SimilarAsinTaskService {
}
}
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;
@@ -1689,6 +2096,19 @@ public class SimilarAsinTaskService {
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;
@@ -1878,6 +2298,16 @@ public class SimilarAsinTaskService {
return result;
}
private boolean hasPersistedResultRows(Long taskId) {
if (taskId == null || taskId <= 0) {
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, SimilarAsinResultRowDto> loadPersistedResultRowsWithRetry(Long taskId, int expectedParsedRows) {
Map<String, SimilarAsinResultRowDto> result = loadPersistedResultRows(taskId);
if (expectedParsedRows <= 0 || !result.isEmpty()) {
@@ -2620,6 +3050,7 @@ public class SimilarAsinTaskService {
private record SubmitContext(FileTaskEntity task,
String scopeKey,
String scopeHash,
Integer chunkIndex,
boolean forceFlush,
String error) {
}
@@ -2630,7 +3061,8 @@ public class SimilarAsinTaskService {
Integer chunkIndex,
Integer batchIndex,
Integer batchTotal,
String ownerInstanceId) {
String ownerInstanceId,
Integer submitRetryCount) {
}
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<SimilarAsinParsedRowVo> allRows) {

View File

@@ -30,7 +30,7 @@ AIIMAGE_APPEARANCE_PATENT_COZE_BASE_URL=https://api.coze.cn
AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_PATH=/v1/workflow/run
AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID=7632683471312355338
AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN=
AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE=10
AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE=50
AIIMAGE_APPEARANCE_PATENT_COZE_READ_TIMEOUT_MILLIS=60000
AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES=20
@@ -38,7 +38,7 @@ AIIMAGE_SIMILAR_ASIN_COZE_BASE_URL=https://api.coze.cn
AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_PATH=/v1/workflow/run
AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_ID=7632683471312355338
AIIMAGE_SIMILAR_ASIN_COZE_TOKEN=
AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE=10
AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE=50
AIIMAGE_SIMILAR_ASIN_COZE_READ_TIMEOUT_MILLIS=60000
AIIMAGE_SIMILAR_ASIN_STALE_TIMEOUT_MINUTES=20

View File

@@ -147,10 +147,10 @@ aiimage:
coze-workflow-path: ${AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_PATH:/v1/workflow/run}
coze-workflow-id: ${AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID:7632683471312355338}
coze-token: ${AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN:Bearer sat_gsInckdsN0qqhnZSK7K4Zevw6neJCHmYHyIbm85oGgJIGEmSdkH5OFctSKLsAHvT}
coze-batch-size: ${AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE:10}
coze-batch-size: ${AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE:50}
coze-connect-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_CONNECT_TIMEOUT_MILLIS:10000}
coze-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_READ_TIMEOUT_MILLIS:60000}
coze-poll-interval-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_INTERVAL_MILLIS:5000}
coze-poll-interval-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_INTERVAL_MILLIS:30000}
coze-poll-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_TIMEOUT_MILLIS:600000}
stale-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES:20}
stale-finalize-cron: ${AIIMAGE_APPEARANCE_PATENT_STALE_FINALIZE_CRON:0 */2 * * * *}
@@ -159,10 +159,10 @@ aiimage:
coze-workflow-path: ${AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_PATH:/v1/workflow/run}
coze-workflow-id: ${AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_ID:7632683471312355338}
coze-token: ${AIIMAGE_SIMILAR_ASIN_COZE_TOKEN:Bearer sat_gsInckdsN0qqhnZSK7K4Zevw6neJCHmYHyIbm85oGgJIGEmSdkH5OFctSKLsAHvT}
coze-batch-size: ${AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE:10}
coze-batch-size: ${AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE:50}
coze-connect-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_CONNECT_TIMEOUT_MILLIS:10000}
coze-read-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_READ_TIMEOUT_MILLIS:60000}
coze-poll-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_INTERVAL_MILLIS:5000}
coze-poll-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_INTERVAL_MILLIS:30000}
coze-poll-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_TIMEOUT_MILLIS:600000}
stale-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_STALE_TIMEOUT_MINUTES:20}
stale-finalize-cron: ${AIIMAGE_SIMILAR_ASIN_STALE_FINALIZE_CRON:0 */2 * * * *}

View File

@@ -0,0 +1,68 @@
SET @db_name = DATABASE();
SET @idx_exists := (
SELECT COUNT(*)
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = @db_name
AND TABLE_NAME = 'biz_similar_asin_filter_condition'
AND INDEX_NAME = 'uk_similar_asin_filter_condition_user_text'
);
SET @sql := IF(@idx_exists > 0,
'ALTER TABLE biz_similar_asin_filter_condition DROP INDEX uk_similar_asin_filter_condition_user_text',
'SELECT 1'
);
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @condition_text_type := (
SELECT DATA_TYPE
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = @db_name
AND TABLE_NAME = 'biz_similar_asin_filter_condition'
AND COLUMN_NAME = 'condition_text'
);
SET @sql := IF(@condition_text_type <> 'mediumtext',
'ALTER TABLE biz_similar_asin_filter_condition MODIFY condition_text MEDIUMTEXT NOT NULL',
'SELECT 1'
);
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @condition_hash_exists := (
SELECT COUNT(*)
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = @db_name
AND TABLE_NAME = 'biz_similar_asin_filter_condition'
AND COLUMN_NAME = 'condition_hash'
);
SET @sql := IF(@condition_hash_exists = 0,
'ALTER TABLE biz_similar_asin_filter_condition ADD COLUMN condition_hash CHAR(64) NULL AFTER condition_text',
'SELECT 1'
);
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
UPDATE biz_similar_asin_filter_condition
SET condition_hash = SHA2(condition_text, 256)
WHERE condition_hash IS NULL OR condition_hash = '';
ALTER TABLE biz_similar_asin_filter_condition
MODIFY condition_hash CHAR(64) NOT NULL;
SET @idx_exists := (
SELECT COUNT(*)
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = @db_name
AND TABLE_NAME = 'biz_similar_asin_filter_condition'
AND INDEX_NAME = 'uk_similar_asin_filter_condition_user_hash'
);
SET @sql := IF(@idx_exists = 0,
'ALTER TABLE biz_similar_asin_filter_condition ADD UNIQUE KEY uk_similar_asin_filter_condition_user_hash (user_id, condition_hash)',
'SELECT 1'
);
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

View File

@@ -0,0 +1,12 @@
INSERT INTO `columns` (`name`, `column_key`, `menu_type`, `route_path`, `sort_order`)
SELECT '商品类目', 'admin_product_categories', 'admin', 'product-categories', 66
WHERE NOT EXISTS (
SELECT 1 FROM `columns` WHERE `column_key` = 'admin_product_categories'
);
UPDATE `columns`
SET `name` = '商品类目',
`menu_type` = 'admin',
`route_path` = 'product-categories',
`sort_order` = 66
WHERE `column_key` = 'admin_product_categories';

View File

@@ -0,0 +1,91 @@
CREATE TABLE IF NOT EXISTS biz_product_category (
id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '主键',
parent_id BIGINT NULL COMMENT '父级类目ID',
name VARCHAR(128) NOT NULL COMMENT '类目名称',
category_key VARCHAR(128) NOT NULL COMMENT '类目标识',
sort_order INT NOT NULL DEFAULT 0 COMMENT '排序',
description VARCHAR(512) NOT NULL DEFAULT '' COMMENT '备注',
is_builtin TINYINT(1) NOT NULL DEFAULT 0 COMMENT '是否内置',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
UNIQUE KEY uk_product_category_key (category_key),
KEY idx_product_category_parent_sort (parent_id, sort_order, id)
) COMMENT='商品类目树';
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT NULL, '护肤品', 'skincare', 10, 1
WHERE NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'skincare');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT NULL, '彩妆', 'makeup', 20, 1
WHERE NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'makeup');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT NULL, '洗护', 'haircare', 30, 1
WHERE NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'haircare');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT NULL, '功效类', 'functional', 40, 1
WHERE NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'functional');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT NULL, '香氛', 'fragrance', 50, 1
WHERE NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'fragrance');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '面霜', 'skincare_cream', 10, 1 FROM biz_product_category p
WHERE p.category_key = 'skincare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'skincare_cream');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '精华', 'skincare_essence', 20, 1 FROM biz_product_category p
WHERE p.category_key = 'skincare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'skincare_essence');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '面膜', 'skincare_mask', 30, 1 FROM biz_product_category p
WHERE p.category_key = 'skincare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'skincare_mask');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '爽肤水', 'skincare_toner', 40, 1 FROM biz_product_category p
WHERE p.category_key = 'skincare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'skincare_toner');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '身体乳', 'skincare_body_lotion', 50, 1 FROM biz_product_category p
WHERE p.category_key = 'skincare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'skincare_body_lotion');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '口红', 'makeup_lipstick', 10, 1 FROM biz_product_category p
WHERE p.category_key = 'makeup' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'makeup_lipstick');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '粉底', 'makeup_foundation', 20, 1 FROM biz_product_category p
WHERE p.category_key = 'makeup' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'makeup_foundation');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '眼影', 'makeup_eyeshadow', 30, 1 FROM biz_product_category p
WHERE p.category_key = 'makeup' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'makeup_eyeshadow');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '睫毛', 'makeup_mascara', 40, 1 FROM biz_product_category p
WHERE p.category_key = 'makeup' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'makeup_mascara');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '洗发水', 'haircare_shampoo', 10, 1 FROM biz_product_category p
WHERE p.category_key = 'haircare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'haircare_shampoo');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '护发素', 'haircare_conditioner', 20, 1 FROM biz_product_category p
WHERE p.category_key = 'haircare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'haircare_conditioner');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '染发剂', 'haircare_hair_dye', 30, 1 FROM biz_product_category p
WHERE p.category_key = 'haircare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'haircare_hair_dye');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '烫发膏', 'haircare_perm_cream', 40, 1 FROM biz_product_category p
WHERE p.category_key = 'haircare' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'haircare_perm_cream');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '防晒', 'functional_sunscreen', 10, 1 FROM biz_product_category p
WHERE p.category_key = 'functional' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'functional_sunscreen');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '祛斑', 'functional_spot_removal', 20, 1 FROM biz_product_category p
WHERE p.category_key = 'functional' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'functional_spot_removal');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '美白', 'functional_whitening', 30, 1 FROM biz_product_category p
WHERE p.category_key = 'functional' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'functional_whitening');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, is_builtin)
SELECT p.id, '祛痘产品', 'functional_acne_care', 40, 1 FROM biz_product_category p
WHERE p.category_key = 'functional' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'functional_acne_care');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, description, is_builtin)
SELECT p.id, '香水', 'fragrance_perfume', 10, '', 1 FROM biz_product_category p
WHERE p.category_key = 'fragrance' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'fragrance_perfume');
INSERT INTO biz_product_category (parent_id, name, category_key, sort_order, description, is_builtin)
SELECT p.id, '香薰精油(高酒精属于危险品)', 'fragrance_essential_oil_hazardous', 20, '高酒精属于危险品', 1 FROM biz_product_category p
WHERE p.category_key = 'fragrance' AND NOT EXISTS (SELECT 1 FROM biz_product_category WHERE category_key = 'fragrance_essential_oil_hazardous');