修改完善这个专利部分

This commit is contained in:
super
2026-05-15 16:06:12 +08:00
parent 18b0c2d211
commit 1dc2cac19a
72 changed files with 2930 additions and 2643 deletions

View File

@@ -6,6 +6,9 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import org.springframework.web.context.request.async.AsyncRequestNotUsableException;
import java.io.IOException;
@Slf4j
@RestControllerAdvice
@@ -34,9 +37,33 @@ public class GlobalExceptionHandler {
return ApiResponse.fail(ex.getMessage());
}
@ExceptionHandler(AsyncRequestNotUsableException.class)
public void handleAsyncRequestNotUsableException(AsyncRequestNotUsableException ex) {
log.debug("Client disconnected before response completed: {}", ex.getMessage());
}
@ExceptionHandler(Exception.class)
public ApiResponse<Void> handleException(Exception ex) {
if (isClientAbort(ex)) {
log.debug("Client disconnected before response completed: {}", ex.getMessage());
return ApiResponse.fail("客户端已断开连接");
}
log.error("Unhandled exception", ex);
return ApiResponse.fail("服务异常: " + ex.getMessage());
}
private boolean isClientAbort(Throwable ex) {
Throwable cursor = ex;
while (cursor != null) {
String className = cursor.getClass().getName();
String message = cursor.getMessage() == null ? "" : cursor.getMessage().toLowerCase();
if (cursor instanceof AsyncRequestNotUsableException
|| cursor instanceof IOException && (message.contains("broken pipe") || message.contains("connection reset"))
|| className.contains("ClientAbortException")) {
return true;
}
cursor = cursor.getCause();
}
return false;
}
}

View File

@@ -3,6 +3,9 @@ package com.nanri.aiimage.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.util.ArrayList;
import java.util.List;
@Data
@ConfigurationProperties(prefix = "aiimage.appearance-patent")
public class AppearancePatentProperties {
@@ -11,6 +14,8 @@ public class AppearancePatentProperties {
private String cozeWorkflowHistoryPath = "/v1/workflows/{workflow_id}/run_histories/{execute_id}";
private String cozeWorkflowId = "7632683471312355338";
private String cozeToken = "";
private List<CozeCredential> cozeCredentials = new ArrayList<>();
private int cozeCredentialStripeSize = 5;
private int cozeBatchSize = 50;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
@@ -18,4 +23,11 @@ public class AppearancePatentProperties {
private int cozePollTimeoutMillis = 600000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";
@Data
public static class CozeCredential {
private String name;
private String workflowId;
private String token;
}
}

View File

@@ -3,6 +3,9 @@ package com.nanri.aiimage.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.util.ArrayList;
import java.util.List;
@Data
@ConfigurationProperties(prefix = "aiimage.similar-asin")
public class SimilarAsinProperties {
@@ -11,6 +14,8 @@ public class SimilarAsinProperties {
private String cozeWorkflowHistoryPath = "/v1/workflows/{workflow_id}/run_histories/{execute_id}";
private String cozeWorkflowId = "7635328462404583478";
private String cozeToken = "";
private List<CozeCredential> cozeCredentials = new ArrayList<>();
private int cozeCredentialStripeSize = 5;
private int cozeBatchSize = 50;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
@@ -18,4 +23,11 @@ public class SimilarAsinProperties {
private int cozePollTimeoutMillis = 600000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";
@Data
public static class CozeCredential {
private String name;
private String workflowId;
private String token;
}
}

View File

@@ -9,6 +9,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Configuration
public class TaskFileJobConfig {
@@ -37,7 +38,23 @@ public class TaskFileJobConfig {
}
@Bean("cozeTaskExecutor")
public TaskExecutor cozeTaskExecutor(ExecutorService cozeVirtualThreadExecutor) {
return new ConcurrentTaskExecutor(cozeVirtualThreadExecutor);
public TaskExecutor cozeTaskExecutor(
ExecutorService cozeVirtualThreadExecutor,
@Value("${aiimage.coze-task.max-concurrent:8}") int maxConcurrent) {
Semaphore semaphore = new Semaphore(Math.max(1, maxConcurrent));
return new ConcurrentTaskExecutor(command -> cozeVirtualThreadExecutor.execute(() -> {
boolean acquired = false;
try {
semaphore.acquire();
acquired = true;
command.run();
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
} finally {
if (acquired) {
semaphore.release();
}
}
}));
}
}

View File

@@ -34,7 +34,7 @@ public class TaskOperationLockConfig implements WebMvcConfigurer {
private static final Pattern TASK_PATH = Pattern.compile(".*/api/([^/]+)/tasks/(\\d+)(?:/.*)?$");
private static final Pattern TASK_RESULT_PATH = Pattern.compile(".*/api/([^/]+)/tasks/(\\d+)/result/?$");
private static final Set<String> MUTATING_METHODS = Set.of("POST", "PUT", "PATCH", "DELETE");
private static final long RESULT_SUBMIT_WAIT_MILLIS = 10 * 60 * 1000L;
private static final long RESULT_SUBMIT_WAIT_MILLIS = 30 * 1000L;
private static final long DELETE_WAIT_MILLIS = 60 * 1000L;
private static final String LOCK_ATTRIBUTE = TaskOperationLockInterceptor.class.getName() + ".LOCK";

View File

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.AppearancePatentProperties;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto;
import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpHeaders;
@@ -19,22 +20,26 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
@Component
@RequiredArgsConstructor
@Slf4j
public class AppearancePatentCozeClient {
private static final String MODULE_TYPE = "APPEARANCE_PATENT";
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
private final AppearancePatentProperties properties;
private final ObjectMapper objectMapper;
private final CozeCredentialPoolService cozeCredentialPoolService;
private final AtomicLong credentialCursor = new AtomicLong();
public List<AppearancePatentResultRowDto> inspect(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
if (rows == null || rows.isEmpty()) {
return List.of();
}
if (properties.getCozeToken() == null || properties.getCozeToken().isBlank()) {
if (!hasConfiguredCredential()) {
log.warn("[appearance-patent] coze token not configured, keep raw rows size={}", rows.size());
return rows.stream().map(this::copy).toList();
}
@@ -48,17 +53,31 @@ public class AppearancePatentCozeClient {
}
public CozeSubmitResponse submitWorkflow(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
return submitWorkflow(rows, prompt, apiKey, nextCredential());
}
public CozeSubmitResponse submitWorkflow(List<AppearancePatentResultRowDto> rows,
String prompt,
String apiKey,
CozeCredentialRef credential) throws Exception {
CozeCredentialRef resolvedCredential = resolveCredential(credential);
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey, resolvedCredential));
ensureSuccess(submitRoot);
return new CozeSubmitResponse(
extractExecuteId(submitRoot),
extractResultDataText(submitRoot),
writeJson(submitRoot)
writeJson(submitRoot),
resolvedCredential.name()
);
}
public CozePollResponse pollWorkflow(String executeId) throws Exception {
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId));
return pollWorkflow(executeId, null);
}
public CozePollResponse pollWorkflow(String executeId, CozeCredentialRef credential) throws Exception {
CozeCredentialRef resolvedCredential = resolveCredential(credential);
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId, resolvedCredential));
ensureSuccess(pollRoot);
String status = normalize(resolveWorkflowStatus(pollRoot)).toUpperCase(Locale.ROOT);
String dataText = extractResultDataText(pollRoot);
@@ -66,7 +85,8 @@ public class AppearancePatentCozeClient {
String failureMessage = isFailedWorkflowStatus(status)
? firstNonBlank(resolveFailureMessage(pollRoot), "Coze async workflow failed")
: "";
return new CozePollResponse(executeId, status, dataText, outputText, failureMessage, writeJson(pollRoot));
return new CozePollResponse(executeId, status, dataText, outputText, failureMessage, writeJson(pollRoot),
resolvedCredential.name());
}
public List<AppearancePatentResultRowDto> mergeRowsFromDataText(List<AppearancePatentResultRowDto> rows, String dataText) throws Exception {
@@ -164,7 +184,8 @@ public class AppearancePatentCozeClient {
}
private String runWorkflowAsyncAndWait(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
CozeCredentialRef credential = nextCredential();
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey, credential));
ensureSuccess(submitRoot);
String immediateData = extractResultDataText(submitRoot);
@@ -180,7 +201,7 @@ public class AppearancePatentCozeClient {
long deadline = System.currentTimeMillis() + Math.max(10000, properties.getCozePollTimeoutMillis());
while (System.currentTimeMillis() < deadline) {
ensureNotInterrupted();
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId));
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId, credential));
ensureSuccess(pollRoot);
String dataText = extractResultDataText(pollRoot);
@@ -205,20 +226,24 @@ public class AppearancePatentCozeClient {
throw new IllegalStateException("Coze async workflow poll timeout");
}
private String postWorkflow(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
private String postWorkflow(List<AppearancePatentResultRowDto> rows,
String prompt,
String apiKey,
CozeCredentialRef credential) {
Map<String, Object> parameters = buildParameters(rows, prompt, apiKey);
Map<String, Object> body = new LinkedHashMap<>();
body.put("workflow_id", properties.getCozeWorkflowId());
body.put("workflow_id", credential.workflowId());
body.put("parameters", parameters);
body.put("is_async", Boolean.TRUE);
log.info("[appearance-patent] coze request url={} body={}",
log.info("[appearance-patent] coze request credential={} url={} body={}",
credential.name(),
joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()),
writeJson(maskCozeRequestBody(body)));
RestClient.RequestBodySpec request = restClient().post()
.uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setBearerAuth(stripBearer(credential.token()));
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
});
@@ -233,28 +258,98 @@ public class AppearancePatentCozeClient {
});
}
private String getWorkflowHistory(String executeId) {
private String getWorkflowHistory(String executeId, CozeCredentialRef credential) {
String path = properties.getCozeWorkflowHistoryPath()
.replace("{workflow_id}", properties.getCozeWorkflowId())
.replace("{workflow_id}", credential.workflowId())
.replace("{execute_id}", executeId);
return restClient().get()
.uri(joinUrl(properties.getCozeBaseUrl(), path))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setBearerAuth(stripBearer(credential.token()));
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
})
.exchange((clientRequest, clientResponse) -> {
byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());
String responseText = responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8);
log.info("[appearance-patent] coze history response executeId={} status={} body={}",
executeId,
log.info("[appearance-patent] coze history response credential={} executeId={} status={} body={}",
credential.name(), executeId,
clientResponse.getStatusCode(),
responseText);
return responseText;
});
}
public CozeCredentialRef nextCredential() {
List<CozeCredentialPoolService.CozeCredential> pooledCredentials = cozeCredentialPoolService.listEnabled(MODULE_TYPE);
CozeCredentialPoolService.CozeCredential pooledCredential =
cozeCredentialPoolService.chooseRoundRobin(MODULE_TYPE, pooledCredentials, properties.getCozeCredentialStripeSize());
if (pooledCredential != null) {
return new CozeCredentialRef(pooledCredential.name(), pooledCredential.workflowId(), pooledCredential.token(),
pooledCredential.maxConcurrent());
}
List<CozeCredentialRef> credentials = configuredCredentials();
int stripeSize = Math.max(1, properties.getCozeCredentialStripeSize());
long cursor = Math.max(0L, credentialCursor.getAndIncrement());
int index = (int) ((cursor / stripeSize) % credentials.size());
return credentials.get(index);
}
public CozeCredentialRef credentialByName(String name) {
if (name == null || name.isBlank()) {
return nextCredential();
}
String normalizedName = normalize(name);
for (CozeCredentialRef credential : configuredCredentials()) {
if (normalize(credential.name()).equals(normalizedName)) {
return credential;
}
}
return nextCredential();
}
public boolean hasConfiguredCredential() {
return !configuredCredentials().isEmpty();
}
public int configuredCredentialCount() {
return configuredCredentials().size();
}
private CozeCredentialRef resolveCredential(CozeCredentialRef credential) {
return credential == null ? nextCredential() : credential;
}
private List<CozeCredentialRef> configuredCredentials() {
List<CozeCredentialRef> credentials = new ArrayList<>();
for (CozeCredentialPoolService.CozeCredential credential : cozeCredentialPoolService.listEnabled(MODULE_TYPE)) {
credentials.add(new CozeCredentialRef(credential.name(), credential.workflowId(), credential.token(),
credential.maxConcurrent()));
}
if (!credentials.isEmpty()) {
return credentials;
}
if (properties.getCozeCredentials() != null) {
int index = 1;
for (AppearancePatentProperties.CozeCredential credential : properties.getCozeCredentials()) {
if (credential == null
|| normalize(credential.getWorkflowId()).isBlank()
|| normalize(credential.getToken()).isBlank()) {
continue;
}
String name = firstNonBlank(credential.getName(), "credential-" + index);
credentials.add(new CozeCredentialRef(name, credential.getWorkflowId(), credential.getToken(), Integer.MAX_VALUE));
index++;
}
}
if (credentials.isEmpty()
&& properties.getCozeWorkflowId() != null && !properties.getCozeWorkflowId().isBlank()
&& properties.getCozeToken() != null && !properties.getCozeToken().isBlank()) {
credentials.add(new CozeCredentialRef("default", properties.getCozeWorkflowId(), properties.getCozeToken(), Integer.MAX_VALUE));
}
return credentials;
}
private Map<String, Object> buildParameters(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
List<String> groupKeys = rows.stream().map(row -> nonBlank(row.getGroupKey(), rowKey(row))).toList();
List<String> rowIds = rows.stream().map(row -> nonBlank(row.getId(), "")).toList();
@@ -965,7 +1060,8 @@ public class AppearancePatentCozeClient {
public record CozeSubmitResponse(
String executeId,
String immediateData,
String rawResponse
String rawResponse,
String credentialName
) {
}
@@ -975,7 +1071,8 @@ public class AppearancePatentCozeClient {
String dataText,
String outputText,
String failureMessage,
String rawResponse
String rawResponse,
String credentialName
) {
public boolean hasPayload() {
return dataText != null && !dataText.isBlank() || outputText != null && !outputText.isBlank();
@@ -1001,6 +1098,14 @@ public class AppearancePatentCozeClient {
}
}
public record CozeCredentialRef(
String name,
String workflowId,
String token,
int maxConcurrent
) {
}
private static final class PartialCozeResultException extends RuntimeException {
private final int resolvedCount;

View File

@@ -28,6 +28,7 @@ import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentParse
import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentTaskBatchVo;
import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentTaskDetailVo;
import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentTaskItemVo;
import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService;
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
@@ -69,6 +70,7 @@ import org.springframework.transaction.support.TransactionTemplate;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
@@ -84,6 +86,8 @@ import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.function.Supplier;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
@Service
@RequiredArgsConstructor
@@ -100,15 +104,17 @@ public class AppearancePatentTaskService {
private static final String COZE_STATUS_DONE = "DONE";
private static final String COZE_STATUS_FAILED = "FAILED";
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final String CONTENT_TYPE_ZIP = "application/zip";
private static final int RESULT_ROWS_READ_RETRY_LIMIT = 3;
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 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_WAIT_MILLIS = 1000L;
private static final long COZE_SUBMIT_LOCK_RETRY_DELAY_MILLIS = 500L;
private static final long COZE_SUBMIT_MIN_INTERVAL_MILLIS = 30000L;
private static final long PENDING_COZE_RETRY_INTERVAL_MILLIS = 30000L;
private static final int MAX_COZE_SUBMIT_RETRY_COUNT = 5;
private static final List<String> RESULT_HEADERS = List.of(
"id",
@@ -119,7 +125,7 @@ public class AppearancePatentTaskService {
"外观维度(外观设计专利)",
"专利维度(发明/实用新型专利)",
"结论",
"status"
"状态"
);
private final LocalFileStorageService localFileStorageService;
@@ -140,6 +146,7 @@ public class AppearancePatentTaskService {
private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService;
private final InstanceMetadata instanceMetadata;
private final CozeCredentialPoolService cozeCredentialPoolService;
@Autowired
@Qualifier("cozeTaskExecutor")
private TaskExecutor cozeTaskExecutor;
@@ -165,12 +172,12 @@ public class AppearancePatentTaskService {
long parseStartedAt = System.nanoTime();
for (AppearancePatentSourceFileDto source : sourceFiles) {
if (source.getFileKey() == null || source.getFileKey().isBlank()) {
throw new BusinessException("fileKey 不能为空");
}
File input = localFileStorageService.findLocalSourceFile(source.getFileKey());
if (input == null || !input.exists()) {
throw new BusinessException("源文件不存在");
}
throw new BusinessException("fileKey 不能为空");
}
File input = localFileStorageService.findLocalSourceFile(source.getFileKey());
if (input == null || !input.exists()) {
throw new BusinessException("源文件不存在");
}
ParsedWorkbook parsed = parseWorkbook(input, source);
totalRows += parsed.totalRows();
@@ -748,7 +755,7 @@ public class AppearancePatentTaskService {
touchJavaSideTaskActivity(task.getId());
} else if (isResultSubmissionComplete(task.getId())) {
maybeFinalizeCozeJobLocked(task.getId(), new CozeBatchContext(
job.getId(), result.getId(), null, null, 1, 1, currentInstanceId(), 0));
job.getId(), result.getId(), null, null, 1, 1, currentInstanceId(), 0, null));
}
}
@@ -1429,18 +1436,7 @@ public class AppearancePatentTaskService {
@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)
.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"));
List<TaskScopeStateEntity> states = listOwnedPendingCozeStates();
if (states == null || states.isEmpty()) {
return;
}
@@ -1464,6 +1460,41 @@ public class AppearancePatentTaskService {
}
}
private List<TaskScopeStateEntity> listOwnedPendingCozeStates() {
Map<Long, TaskScopeStateEntity> merged = new LinkedHashMap<>();
for (TaskScopeStateEntity state : queryOwnedPendingCozeStates(false, 50)) {
if (state != null && state.getId() != null) {
merged.put(state.getId(), state);
}
}
for (TaskScopeStateEntity state : queryOwnedPendingCozeStates(true, 50)) {
if (state != null && state.getId() != null) {
merged.putIfAbsent(state.getId(), state);
}
}
return new ArrayList<>(merged.values());
}
private List<TaskScopeStateEntity> queryOwnedPendingCozeStates(boolean oldestFirst, int limit) {
LambdaQueryWrapper<TaskScopeStateEntity> wrapper = new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING))
.and(ownerWrapper -> ownerWrapper
.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()))
.last("limit " + Math.max(1, Math.min(limit, 100)));
if (oldestFirst) {
wrapper.orderByAsc(TaskScopeStateEntity::getUpdatedAt);
} else {
wrapper.orderByDesc(TaskScopeStateEntity::getUpdatedAt);
}
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(wrapper);
return states == null ? List.of() : states;
}
private List<TaskChunkEntity> loadSubmittedChunks(Long taskId) {
if (taskId == null || taskId <= 0) {
return List.of();
@@ -1492,6 +1523,10 @@ public class AppearancePatentTaskService {
if (key.isBlank() || !queuedRowKeys.add(key)) {
continue;
}
AppearancePatentResultRowDto persistedRow = persistedRows.get(key);
if (hasResolvedCozeFields(persistedRow)) {
continue;
}
candidates.add(new CozeCandidate(chunk.getScopeHash(), chunk.getChunkIndex(), row));
}
}
@@ -1530,7 +1565,7 @@ public class AppearancePatentTaskService {
if (chunks == null || chunks.isEmpty()) {
return countPendingCozeStates(task.getId()) > 0;
}
if (properties.getCozeToken() == null || properties.getCozeToken().isBlank()) {
if (!cozeClient.hasConfiguredCredential()) {
log.warn("[appearance-patent] coze token not configured, skip async coze taskId={} jobId={}",
task.getId(), job.getId());
return false;
@@ -1586,8 +1621,10 @@ public class AppearancePatentTaskService {
return COZE_STATUS_SUBMITTED.equals(existing.getCozeStatus())
|| COZE_STATUS_RUNNING.equals(existing.getCozeStatus());
}
AppearancePatentCozeClient.CozeCredentialRef credential = cozeClient.nextCredential();
try {
AppearancePatentCozeClient.CozeSubmitResponse submit = submitCozeWorkflowThrottled(batchRows, prompt, apiKey);
AppearancePatentCozeClient.CozeSubmitResponse submit = submitCozeWorkflowThrottled(
batchRows, prompt, apiKey, credential, true);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<AppearancePatentResultRowDto> cozeRows = cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId);
@@ -1600,19 +1637,84 @@ public class AppearancePatentTaskService {
return false;
}
saveCozeBatchState(task, result, job, batchRows, batchScopeKey, batchScopeHash,
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());
batchIndex, batchTotal, submit.executeId(), submit.credentialName());
log.info("[appearance-patent] coze async submitted taskId={} jobId={} rows={} batch={}/{} credential={} executeId={}",
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal,
submit.credentialName(), 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={} rows={} batch={}/{} err={}",
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal, message);
if (isCozeThrottleLockTimeout(message)) {
savePendingCozeBatchState(task, result, job, batchRows, batchScopeKey, batchScopeHash,
batchIndex, batchTotal, message, credential.name());
taskFileJobService.touchRunning(job.getId());
touchJavaSideTaskActivity(task.getId());
return true;
}
mergeCozeRowsIntoSubmittedChunks(task, cozeClient.markRowsFailed(batchRows, message), allRowsByBaseId);
return false;
}
}
private boolean isCozeThrottleLockTimeout(String message) {
return normalize(message).toLowerCase(Locale.ROOT).contains("coze submit throttle lock timeout");
}
private void savePendingCozeBatchState(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
List<AppearancePatentResultRowDto> batchRows,
String batchScopeKey,
String batchScopeHash,
int batchIndex,
int batchTotal,
String pendingReason,
String credentialName) {
LocalDateTime now = LocalDateTime.now();
CozeBatchContext context = new CozeBatchContext(
job.getId(),
result.getId(),
null,
null,
batchIndex,
batchTotal,
currentInstanceId(),
0,
credentialName
);
String batchPayload = writeJson(batchRows, "serialize pending coze batch payload failed");
String storedBatchPayload = storeSharedCozeBatchPayload(task.getId(), batchScopeHash, batchPayload);
TaskScopeStateEntity state = new TaskScopeStateEntity();
state.setTaskId(task.getId());
state.setModuleType(MODULE_TYPE);
state.setScopeKey(batchScopeKey);
state.setScopeHash(batchScopeHash);
state.setParsedPayloadJson(storedBatchPayload);
state.setStateJson(writeJson(context, "serialize pending coze batch context failed"));
state.setCozeStatus(COZE_STATUS_RUNNING);
state.setCozeSubmittedAt(now);
state.setCozeLastPolledAt(null);
state.setCozeAttemptCount(0);
state.setCozeError(firstNonBlank(pendingReason, "Coze submit queued by throttle"));
state.setChunkTotal(batchTotal);
state.setReceivedChunkCount(batchIndex);
state.setCompleted(0);
state.setCreatedAt(now);
state.setUpdatedAt(now);
try {
taskScopeStateMapper.insert(state);
touchJavaSideTaskActivity(task.getId());
log.info("[appearance-patent] coze async submit queued by throttle taskId={} jobId={} rows={} batch={}/{}",
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal);
} catch (DuplicateKeyException ex) {
transientPayloadStorageService.deletePayloadIfPresent(storedBatchPayload);
log.info("[appearance-patent] duplicate pending coze batch state ignored taskId={} scope={}",
task.getId(), batchScopeKey);
}
}
private void saveCozeBatchState(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
@@ -1621,7 +1723,8 @@ public class AppearancePatentTaskService {
String batchScopeHash,
int batchIndex,
int batchTotal,
String executeId) {
String executeId,
String credentialName) {
LocalDateTime now = LocalDateTime.now();
CozeBatchContext context = new CozeBatchContext(
job.getId(),
@@ -1631,7 +1734,8 @@ public class AppearancePatentTaskService {
batchIndex,
batchTotal,
currentInstanceId(),
0
0,
credentialName
);
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
String storedBatchPayload = storeSharedCozeBatchPayload(task.getId(), batchScopeHash, batchPayload);
@@ -1689,6 +1793,10 @@ public class AppearancePatentTaskService {
try {
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
if (state != null && state.getCozeExecuteId() == null
&& (COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
retryPendingCozeSubmitState(state);
}
return;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
@@ -1715,7 +1823,9 @@ public class AppearancePatentTaskService {
log.info("[appearance-patent] coze poll start taskId={} stateId={} executeId={} jobId={} chunk={} batch={}/{}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(),
context.jobId(), context.chunkIndex(), context.batchIndex(), context.batchTotal());
AppearancePatentCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(state.getCozeExecuteId());
AppearancePatentCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(
state.getCozeExecuteId(),
cozeClient.credentialByName(context.credentialName()));
if (!poll.hasPayload() && !poll.isFinished() && !isCozeStateTimedOut(state)) {
log.info("[appearance-patent] coze poll pending taskId={} stateId={} executeId={} status={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), poll.status());
@@ -1804,7 +1914,8 @@ public class AppearancePatentTaskService {
}
try {
AppearancePatentCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task));
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<AppearancePatentResultRowDto> cozeRows =
@@ -1853,23 +1964,54 @@ public class AppearancePatentTaskService {
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;
String apiKey,
AppearancePatentCozeClient.CozeCredentialRef credential,
boolean allowCredentialFallback) throws Exception {
int attempts = allowCredentialFallback ? Math.max(1, cozeClient.configuredCredentialCount()) : 1;
AppearancePatentCozeClient.CozeCredentialRef currentCredential =
credential == null ? cozeClient.nextCredential() : credential;
Exception lastFailure = null;
for (int i = 0; i < attempts; i++) {
DistributedJobLockService.LockHandle lockHandle = acquireCozeSubmitLock(currentCredential);
if (lockHandle == null) {
lastFailure = new IllegalStateException("Coze submit throttle lock timeout");
currentCredential = cozeClient.nextCredential();
continue;
}
CozeCredentialPoolService.BorrowedCredential borrowedCredential =
cozeCredentialPoolService.borrow(MODULE_TYPE, toPoolCredential(currentCredential));
if (borrowedCredential == null) {
lockHandle.close();
lastFailure = new IllegalStateException("Coze credential concurrency limit reached");
currentCredential = cozeClient.nextCredential();
continue;
}
try (lockHandle; borrowedCredential) {
return cozeClient.submitWorkflow(rows, prompt, apiKey, currentCredential);
} finally {
sleepQuietly(COZE_SUBMIT_MIN_INTERVAL_MILLIS);
}
}
throw lastFailure == null ? new IllegalStateException("Coze submit failed") : lastFailure;
}
private DistributedJobLockService.LockHandle acquireCozeSubmitLock() {
private CozeCredentialPoolService.CozeCredential toPoolCredential(AppearancePatentCozeClient.CozeCredentialRef credential) {
if (credential == null) {
return null;
}
return new CozeCredentialPoolService.CozeCredential(
credential.name(),
credential.workflowId(),
credential.token(),
credential.maxConcurrent());
}
private DistributedJobLockService.LockHandle acquireCozeSubmitLock(AppearancePatentCozeClient.CozeCredentialRef credential) {
long deadline = System.currentTimeMillis() + COZE_SUBMIT_LOCK_WAIT_MILLIS;
String credentialName = credential == null ? "default" : firstNonBlank(credential.name(), "default");
while (System.currentTimeMillis() <= deadline) {
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("appearance-patent:coze-submit", COZE_SUBMIT_LOCK_TTL);
distributedJobLockService.tryLock("appearance-patent:coze-submit:" + credentialName, COZE_SUBMIT_LOCK_TTL);
if (lockHandle != null) {
return lockHandle;
}
@@ -1915,14 +2057,16 @@ public class AppearancePatentTaskService {
int partIndex = 1;
for (List<AppearancePatentResultRowDto> partRows : partitions) {
AppearancePatentCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(partRows, readAiPrompt(task), readApiKey(task));
submitCozeWorkflowThrottled(partRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
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());
saveSplitRetryCozeBatchState(state, context, partRows, partIndex, partitions.size(), retryCount,
submit.executeId(), submit.credentialName());
submittedAny = true;
}
partIndex++;
@@ -1950,7 +2094,8 @@ public class AppearancePatentTaskService {
int partIndex,
int partTotal,
int retryCount,
String executeId) {
String executeId,
String credentialName) {
String scopeKey = parent.getScopeKey() + ":split:" + retryCount + ":" + partIndex;
String scopeHash = DigestUtil.sha256Hex(scopeKey);
CozeBatchContext context = new CozeBatchContext(
@@ -1961,7 +2106,8 @@ public class AppearancePatentTaskService {
partIndex,
partTotal,
parentContext.ownerInstanceId(),
retryCount
retryCount,
firstNonBlank(credentialName, parentContext.credentialName())
);
LocalDateTime now = LocalDateTime.now();
String batchPayload = writeJson(batchRows, "serialize split coze batch payload failed");
@@ -2058,6 +2204,24 @@ public class AppearancePatentTaskService {
.set(TaskScopeStateEntity::getUpdatedAt, now)) > 0;
}
private boolean tryClaimPendingCozeSubmitState(TaskScopeStateEntity state) {
if (state == null || state.getId() == null) {
return false;
}
LocalDateTime now = LocalDateTime.now();
long intervalMillis = Math.max(1000L, PENDING_COZE_RETRY_INTERVAL_MILLIS);
return taskScopeStateMapper.update(null, new LambdaUpdateWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getId, state.getId())
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING))
.and(wrapper -> wrapper
.isNull(TaskScopeStateEntity::getCozeLastPolledAt)
.or()
.le(TaskScopeStateEntity::getCozeLastPolledAt, now.minus(Duration.ofMillis(intervalMillis))))
.set(TaskScopeStateEntity::getCozeStatus, COZE_STATUS_RUNNING)
.set(TaskScopeStateEntity::getCozeLastPolledAt, now)
.set(TaskScopeStateEntity::getUpdatedAt, now)) > 0;
}
private void markCozeStateTerminal(TaskScopeStateEntity state, String status, String error) {
taskScopeStateMapper.update(null, new LambdaUpdateWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getId, state.getId())
@@ -2275,7 +2439,8 @@ public class AppearancePatentTaskService {
context.batchIndex(),
context.batchTotal(),
context.ownerInstanceId(),
submitRetryCount
submitRetryCount,
context.credentialName()
);
}
@@ -2374,6 +2539,115 @@ public class AppearancePatentTaskService {
return "task:" + taskId + ":owner:" + firstNonBlank(ownerFromTask(task), currentInstanceId());
}
private void retryPendingCozeSubmitState(TaskScopeStateEntity state) {
if (state == null || state.getId() == null) {
return;
}
if (!tryClaimPendingCozeSubmitState(state)) {
log.info("[appearance-patent] coze submit retry skipped by claim guard taskId={} stateId={} lastPolledAt={}",
state.getTaskId(), state.getId(), state.getCozeLastPolledAt());
return;
}
CozeBatchContext context = readCozeBatchContext(state);
if (context == null || context.jobId() == null || context.resultId() == null) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit context missing");
return;
}
List<AppearancePatentResultRowDto> batchRows = readCozeBatchRows(state);
if (batchRows.isEmpty()) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit payload missing");
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit task missing");
return;
}
Map<String, AppearancePatentResultRowDto> currentRows = loadPersistedResultRows(task.getId());
List<AppearancePatentResultRowDto> currentBatchRows = batchRows.stream()
.map(row -> currentRows.get(rowKey(row)))
.filter(Objects::nonNull)
.toList();
if (currentBatchRows.size() == batchRows.size()
&& currentBatchRows.stream().allMatch(this::hasResolvedCozeFields)) {
markCozeStateTerminal(state, COZE_STATUS_DONE, "Coze rows already resolved by another batch");
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
log.info("[appearance-patent] coze pending submit skipped because rows already resolved taskId={} stateId={} jobId={} rows={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size());
return;
}
try {
taskFileJobService.touchRunning(context.jobId());
AppearancePatentCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
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);
return;
}
if (submit.executeId() == null || submit.executeId().isBlank()) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze async execute_id missing");
mergeCozeRowsIntoSubmittedChunks(task,
cozeClient.markRowsFailed(batchRows, "Coze async execute_id missing"),
allRowsByBaseId);
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
LocalDateTime now = LocalDateTime.now();
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::getCozeError, null)
.set(TaskScopeStateEntity::getUpdatedAt, now));
touchJavaSideTaskActivity(state.getTaskId());
log.info("[appearance-patent] coze pending submit retried taskId={} stateId={} jobId={} rows={} executeId={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size(), submit.executeId());
} catch (Exception ex) {
String message = firstNonBlank(ex.getMessage(), "Coze pending submit retry failed");
log.warn("[appearance-patent] coze pending submit retry failed taskId={} stateId={} jobId={} rows={} err={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size(), message);
if (isCozeThrottleLockTimeout(message)) {
keepPendingCozeSubmitState(state, message);
return;
}
int nextAttemptCount = cozeAttemptCount(state) + 1;
if (nextAttemptCount >= MAX_COZE_SUBMIT_RETRY_COUNT) {
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
String finalMessage = message + " after " + nextAttemptCount + " submit attempts";
mergeCozeRowsIntoSubmittedChunks(task,
cozeClient.markRowsFailed(batchRows, finalMessage),
allRowsByBaseId);
markCozeStateTerminal(state, COZE_STATUS_FAILED, finalMessage);
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
updateCozeStateRunning(state, message);
}
}
private void keepPendingCozeSubmitState(TaskScopeStateEntity state, String error) {
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::getCozeStatus, COZE_STATUS_RUNNING)
.set(TaskScopeStateEntity::getCozeExecuteId, null)
.set(TaskScopeStateEntity::getCozeError, error)
.set(TaskScopeStateEntity::getUpdatedAt, LocalDateTime.now()));
if (updated > 0) {
touchJavaSideTaskActivity(state.getTaskId());
}
}
private String currentInstanceId() {
String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId();
return firstNonBlank(instanceId, "unknown-instance");
@@ -2475,6 +2749,12 @@ public class AppearancePatentTaskService {
}
int safeTotal = Math.max(1, total);
int safeCompleted = Math.max(0, Math.min(completed, safeTotal));
TaskProgressSnapshotEntity existing = taskProgressSnapshotService.find(task.getId(), MODULE_TYPE);
int displayPercent = calculateDisplayProgressPercent(safeCompleted, safeTotal, job, LocalDateTime.now());
int previousDisplayPercent = calculateSnapshotDisplayPercent(existing, job);
if (previousDisplayPercent > displayPercent && safeCompleted < safeTotal) {
displayPercent = previousDisplayPercent;
}
taskProgressSnapshotService.save(
task.getId(),
MODULE_TYPE,
@@ -2484,10 +2764,57 @@ public class AppearancePatentTaskService {
0,
job.getScopeKey(),
message,
Map.of("phase", "RESULT_FILE", "jobId", job.getId())
Map.of("phase", "RESULT_FILE", "jobId", job.getId(), "displayPercent", displayPercent)
);
}
private int calculateDisplayProgressPercent(int current,
int total,
TaskFileJobEntity job,
LocalDateTime baseTime) {
if (total <= 0) {
return 0;
}
current = Math.max(0, Math.min(current, total));
int percent = Math.max(1, Math.min(99, (int) Math.floor(current * 100.0 / total)));
if (job != null && STATUS_RUNNING.equals(job.getStatus())) {
long elapsedSeconds = baseTime == null ? 0 : Math.max(0, Duration.between(baseTime, LocalDateTime.now()).getSeconds());
if (current <= 0) {
int firstRealProgressPercent = Math.max(1, Math.min(99, (int) Math.floor(100.0 / total)));
int waitingCap = Math.max(8, Math.min(35, firstRealProgressPercent - 1));
percent = Math.max(percent, Math.min(waitingCap, 8 + (int) (elapsedSeconds / 6)));
} else if (current < total) {
percent = Math.max(percent, Math.min(92, percent + (int) (elapsedSeconds / 10)));
}
}
return percent;
}
private int calculateSnapshotDisplayPercent(TaskProgressSnapshotEntity snapshot, TaskFileJobEntity job) {
if (snapshot == null) {
return 0;
}
int total = snapshot.getTotalCount() == null ? 0 : snapshot.getTotalCount();
int current = snapshot.getSuccessCount() == null ? 0 : snapshot.getSuccessCount();
int percent = total <= 0 ? 0 : calculateDisplayProgressPercent(current, total, job, snapshot.getUpdatedAt());
return Math.max(percent, extractSnapshotDisplayPercent(snapshot));
}
private int extractSnapshotDisplayPercent(TaskProgressSnapshotEntity snapshot) {
if (snapshot == null || snapshot.getSnapshotJson() == null || snapshot.getSnapshotJson().isBlank()) {
return 0;
}
try {
JsonNode displayPercent = objectMapper.readTree(snapshot.getSnapshotJson()).path("displayPercent");
if (!displayPercent.isNumber()) {
return 0;
}
return Math.max(0, Math.min(100, displayPercent.asInt()));
} catch (Exception ignored) {
return 0;
}
}
private void saveCozePipelineProgress(FileTaskEntity task, TaskFileJobEntity job) {
if (task == null || task.getId() == null || job == null || job.getId() == null) {
return;
@@ -2565,28 +2892,160 @@ public class AppearancePatentTaskService {
if (!outputDir.exists() && !outputDir.mkdirs()) {
throw new BusinessException("创建结果目录失败");
}
String filename = safeFileStem(result.getSourceFilename()) + "-result.xlsx";
String tempFilename = safeFileStem(result.getSourceFilename())
+ "-" + task.getId()
+ "-" + result.getId()
+ "-" + UUID.randomUUID()
+ "-result.xlsx";
File xlsx = new File(outputDir, tempFilename);
List<SourceRows> sourceRows = splitRowsBySourceFile(parsed, receivedRows, result.getSourceFilename());
List<SourceResultWorkbook> workbooks = new ArrayList<>();
File zip = null;
try {
writeResultWorkbook(xlsx, parsed, receivedRows, resultMap);
String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
for (SourceRows item : sourceRows) {
String filename = safeFileStem(item.sourceFilename()) + "-result.xlsx";
String tempFilename = safeFileStem(item.sourceFilename())
+ "-" + task.getId()
+ "-" + result.getId()
+ "-" + UUID.randomUUID()
+ "-result.xlsx";
File xlsx = new File(outputDir, tempFilename);
writeResultWorkbook(xlsx, parsed, item.rows(), resultMap);
workbooks.add(new SourceResultWorkbook(xlsx, filename, item.rows().size()));
}
if (workbooks.isEmpty()) {
throw new BusinessException("外观专利检测结果为空,请稍后重试生成结果文件");
}
File uploadFile;
String filename;
String contentType;
if (workbooks.size() == 1) {
SourceResultWorkbook workbook = workbooks.get(0);
uploadFile = workbook.file();
filename = workbook.filename();
contentType = CONTENT_TYPE_XLSX;
} else {
filename = safeFileStem(result.getSourceFilename()) + "-result.zip";
String tempFilename = safeFileStem(result.getSourceFilename())
+ "-" + task.getId()
+ "-" + result.getId()
+ "-" + UUID.randomUUID()
+ "-result.zip";
zip = new File(outputDir, tempFilename);
packageResultWorkbooksAsZip(zip, workbooks);
uploadFile = zip;
contentType = CONTENT_TYPE_ZIP;
}
String objectKey = ossStorageService.uploadResultFile(uploadFile, MODULE_TYPE);
result.setResultFilename(filename);
result.setResultFileUrl(objectKey);
result.setResultFileSize(xlsx.length());
result.setResultContentType(CONTENT_TYPE_XLSX);
result.setResultFileSize(uploadFile.length());
result.setResultContentType(contentType);
result.setRowCount(receivedRows.size());
} finally {
if (xlsx.exists() && !xlsx.delete()) {
log.warn("[appearance-patent] delete temp xlsx failed file={}", xlsx);
for (SourceResultWorkbook workbook : workbooks) {
if (workbook.file().exists() && !workbook.file().delete()) {
log.warn("[appearance-patent] delete temp xlsx failed file={}", workbook.file());
}
}
if (zip != null && zip.exists() && !zip.delete()) {
log.warn("[appearance-patent] delete temp zip failed file={}", zip);
}
}
}
private List<SourceRows> splitRowsBySourceFile(AppearancePatentParsedPayloadDto parsed,
List<AppearancePatentParsedRowVo> receivedRows,
String fallbackFilename) {
String defaultFilename = firstNonBlank(fallbackFilename, "appearance-patent");
Map<String, SourceRowsBuilder> builders = new LinkedHashMap<>();
if (parsed != null && parsed.getSourceFiles() != null) {
for (AppearancePatentSourceFileDto sourceFile : parsed.getSourceFiles()) {
if (sourceFile == null) {
continue;
}
String sourceFileKey = normalize(sourceFile.getFileKey());
if (sourceFileKey.isBlank()) {
continue;
}
builders.putIfAbsent(sourceFileKey,
new SourceRowsBuilder(sourceFileKey, firstNonBlank(sourceFile.getOriginalFilename(), defaultFilename)));
}
}
for (AppearancePatentParsedRowVo row : receivedRows == null ? List.<AppearancePatentParsedRowVo>of() : receivedRows) {
if (row == null) {
continue;
}
String sourceFileKey = normalize(row.getSourceFileKey());
if (sourceFileKey.isBlank()) {
sourceFileKey = normalize(row.getSourceFilename());
}
if (sourceFileKey.isBlank()) {
sourceFileKey = "__default__";
}
SourceRowsBuilder builder = builders.computeIfAbsent(sourceFileKey,
key -> new SourceRowsBuilder(key, firstNonBlank(row.getSourceFilename(), defaultFilename)));
if (builder.sourceFilename().isBlank() || Objects.equals(builder.sourceFilename(), defaultFilename)) {
builder.setSourceFilename(firstNonBlank(row.getSourceFilename(), defaultFilename));
}
builder.rows().add(row);
}
return builders.values().stream()
.filter(builder -> !builder.rows().isEmpty())
.map(builder -> new SourceRows(builder.sourceFileKey(), firstNonBlank(builder.sourceFilename(), defaultFilename), builder.rows()))
.toList();
}
private void packageResultWorkbooksAsZip(File zip, List<SourceResultWorkbook> workbooks) {
Set<String> entryNames = new LinkedHashSet<>();
try (ZipOutputStream zos = new ZipOutputStream(new FileOutputStream(zip))) {
byte[] buffer = new byte[8192];
for (SourceResultWorkbook workbook : workbooks) {
writeResultZipEntry(zos, buffer, workbook.file(), uniqueZipEntryName(workbook.filename(), entryNames));
}
} catch (IOException ex) {
throw new BusinessException("生成外观专利检测结果压缩包失败");
}
}
private void writeResultZipEntry(ZipOutputStream zos,
byte[] buffer,
File file,
String entryName) throws IOException {
zos.putNextEntry(new ZipEntry(entryName));
try (FileInputStream inputStream = new FileInputStream(file)) {
int len;
while ((len = inputStream.read(buffer)) > 0) {
zos.write(buffer, 0, len);
}
}
zos.closeEntry();
}
private String uniqueZipEntryName(String filename, Set<String> entryNames) {
String safeFilename = normalizeZipFilename(filename);
if (entryNames.add(safeFilename)) {
return safeFilename;
}
int dotIndex = safeFilename.lastIndexOf('.');
String stem = dotIndex > 0 ? safeFilename.substring(0, dotIndex) : safeFilename;
String extension = dotIndex > 0 ? safeFilename.substring(dotIndex) : "";
int index = 2;
while (true) {
String candidate = stem + "(" + index + ")" + extension;
if (entryNames.add(candidate)) {
return candidate;
}
index++;
}
}
private String normalizeZipFilename(String filename) {
String normalized = firstNonBlank(filename, "appearance-patent-result.xlsx").replace('\\', '/');
while (normalized.startsWith("/")) {
normalized = normalized.substring(1);
}
return normalized.isBlank() ? "appearance-patent-result.xlsx" : normalized;
}
private List<AppearancePatentParsedRowVo> filterReceivedParsedRows(List<AppearancePatentParsedRowVo> parsedRows,
Map<String, AppearancePatentResultRowDto> resultMap) {
if (parsedRows == null || parsedRows.isEmpty() || resultMap == null || resultMap.isEmpty()) {
@@ -3141,19 +3600,10 @@ public class AppearancePatentTaskService {
if (total <= 0) {
return;
}
current = Math.max(0, Math.min(current, total));
int percent = Math.max(1, Math.min(99, (int) Math.floor(current * 100.0 / total)));
if (job != null && STATUS_RUNNING.equals(job.getStatus())) {
LocalDateTime baseTime = snapshot.getUpdatedAt() != null ? snapshot.getUpdatedAt() : job.getUpdatedAt();
long elapsedSeconds = baseTime == null ? 0 : Math.max(0, Duration.between(baseTime, LocalDateTime.now()).getSeconds());
if (current <= 0) {
int firstRealProgressPercent = Math.max(1, Math.min(99, (int) Math.floor(100.0 / total)));
int waitingCap = Math.max(8, Math.min(35, firstRealProgressPercent - 1));
percent = Math.max(percent, Math.min(waitingCap, 8 + (int) (elapsedSeconds / 6)));
} else if (current < total) {
percent = Math.max(percent, Math.min(92, percent + (int) (elapsedSeconds / 10)));
}
}
LocalDateTime baseTime = snapshot.getUpdatedAt() != null ? snapshot.getUpdatedAt() : (job == null ? null : job.getUpdatedAt());
int percent = calculateDisplayProgressPercent(current, total, job, baseTime);
percent = Math.max(percent, extractSnapshotDisplayPercent(snapshot));
percent = Boolean.TRUE.equals(vo.getFileReady()) ? 100 : Math.min(99, percent);
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(percent);
@@ -3514,7 +3964,8 @@ public class AppearancePatentTaskService {
Integer batchIndex,
Integer batchTotal,
String ownerInstanceId,
Integer submitRetryCount) {
Integer submitRetryCount,
String credentialName) {
}
private record CozeCandidate(String chunkScopeHash,
@@ -3522,6 +3973,43 @@ public class AppearancePatentTaskService {
AppearancePatentResultRowDto row) {
}
private static class SourceRowsBuilder {
private final String sourceFileKey;
private String sourceFilename;
private final List<AppearancePatentParsedRowVo> rows = new ArrayList<>();
private SourceRowsBuilder(String sourceFileKey, String sourceFilename) {
this.sourceFileKey = sourceFileKey;
this.sourceFilename = sourceFilename;
}
private String sourceFileKey() {
return sourceFileKey;
}
private String sourceFilename() {
return sourceFilename;
}
private void setSourceFilename(String sourceFilename) {
this.sourceFilename = sourceFilename;
}
private List<AppearancePatentParsedRowVo> rows() {
return rows;
}
}
private record SourceRows(String sourceFileKey,
String sourceFilename,
List<AppearancePatentParsedRowVo> rows) {
}
private record SourceResultWorkbook(File file,
String filename,
int rowCount) {
}
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<AppearancePatentParsedRowVo> allRows) {
}
}

View File

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

View File

@@ -0,0 +1,25 @@
package com.nanri.aiimage.modules.coze.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_coze_credential")
public class CozeCredentialEntity {
@TableId(type = IdType.AUTO)
private Long id;
private String moduleType;
private String credentialName;
private String workflowId;
private String token;
private Integer enabled;
private Integer maxConcurrent;
private Integer sortOrder;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
}

View File

@@ -0,0 +1,182 @@
package com.nanri.aiimage.modules.coze.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.nanri.aiimage.modules.coze.mapper.CozeCredentialMapper;
import com.nanri.aiimage.modules.coze.model.entity.CozeCredentialEntity;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.util.List;
import java.util.Objects;
@Slf4j
@Service
@RequiredArgsConstructor
public class CozeCredentialPoolService {
private static final Duration INFLIGHT_TTL = Duration.ofMinutes(30);
private final CozeCredentialMapper cozeCredentialMapper;
private final StringRedisTemplate stringRedisTemplate;
public List<CozeCredential> listEnabled(String moduleType) {
if (moduleType == null || moduleType.isBlank()) {
return List.of();
}
try {
List<CozeCredentialEntity> rows = cozeCredentialMapper.selectList(new LambdaQueryWrapper<CozeCredentialEntity>()
.eq(CozeCredentialEntity::getModuleType, moduleType)
.eq(CozeCredentialEntity::getEnabled, 1)
.orderByAsc(CozeCredentialEntity::getSortOrder)
.orderByAsc(CozeCredentialEntity::getId));
if (rows == null || rows.isEmpty()) {
return List.of();
}
return rows.stream()
.filter(Objects::nonNull)
.filter(row -> !blank(row.getCredentialName())
&& !blank(row.getWorkflowId())
&& !blank(row.getToken()))
.map(row -> new CozeCredential(
row.getCredentialName().trim(),
row.getWorkflowId().trim(),
row.getToken().trim(),
row.getMaxConcurrent() == null || row.getMaxConcurrent() <= 0
? Integer.MAX_VALUE
: row.getMaxConcurrent()))
.toList();
} catch (Exception ex) {
log.warn("[coze-credential] list enabled failed moduleType={} err={}", moduleType, ex.getMessage());
return List.of();
}
}
public CozeCredential chooseLeastInflight(String moduleType, List<CozeCredential> credentials) {
return chooseRoundRobin(moduleType, credentials, 1);
}
public CozeCredential chooseRoundRobin(String moduleType, List<CozeCredential> credentials, int stripeSize) {
if (credentials == null || credentials.isEmpty()) {
return null;
}
int safeStripeSize = Math.max(1, stripeSize);
long cursor = nextCursor(moduleType);
int index = (int) ((Math.max(0L, cursor) / safeStripeSize) % credentials.size());
return credentials.get(index);
}
public BorrowedCredential borrow(String moduleType, CozeCredential credential) {
if (credential == null) {
return null;
}
String key = inflightKey(moduleType, credential.name());
try {
Long value = stringRedisTemplate.opsForValue().increment(key);
stringRedisTemplate.expire(key, INFLIGHT_TTL);
long inflight = value == null ? 0L : value;
if (inflight > credential.maxConcurrent()) {
release(moduleType, credential.name());
return null;
}
return new BorrowedCredential(this, moduleType, credential.name());
} catch (Exception ex) {
log.warn("[coze-credential] borrow failed moduleType={} credential={} err={}",
moduleType, credential.name(), ex.getMessage());
return BorrowedCredential.noop();
}
}
public void release(String moduleType, String credentialName) {
if (blank(moduleType) || blank(credentialName)) {
return;
}
try {
Long value = stringRedisTemplate.opsForValue().decrement(inflightKey(moduleType, credentialName));
if (value != null && value <= 0L) {
stringRedisTemplate.delete(inflightKey(moduleType, credentialName));
}
} catch (Exception ex) {
log.warn("[coze-credential] release failed moduleType={} credential={} err={}",
moduleType, credentialName, ex.getMessage());
}
}
private long inflight(String moduleType, String credentialName) {
try {
String raw = stringRedisTemplate.opsForValue().get(inflightKey(moduleType, credentialName));
return raw == null || raw.isBlank() ? 0L : Long.parseLong(raw);
} catch (Exception ex) {
return 0L;
}
}
private long nextCursor(String moduleType) {
if (blank(moduleType)) {
return 0L;
}
try {
Long value = stringRedisTemplate.opsForValue().increment(cursorKey(moduleType));
stringRedisTemplate.expire(cursorKey(moduleType), Duration.ofDays(7));
return value == null ? 0L : Math.max(0L, value - 1L);
} catch (Exception ex) {
log.warn("[coze-credential] cursor increment failed moduleType={} err={}", moduleType, ex.getMessage());
return System.nanoTime();
}
}
private String inflightKey(String moduleType, String credentialName) {
return "coze:credential:inflight:" + moduleType + ":" + credentialName;
}
private String cursorKey(String moduleType) {
return "coze:credential:cursor:" + moduleType;
}
private boolean blank(String value) {
return value == null || value.isBlank();
}
public record CozeCredential(String name,
String workflowId,
String token,
int maxConcurrent) {
}
public static final class BorrowedCredential implements AutoCloseable {
private final CozeCredentialPoolService owner;
private final String moduleType;
private final String credentialName;
private final boolean noop;
private boolean released;
private BorrowedCredential(CozeCredentialPoolService owner, String moduleType, String credentialName) {
this.owner = owner;
this.moduleType = moduleType;
this.credentialName = credentialName;
this.noop = false;
}
private BorrowedCredential() {
this.owner = null;
this.moduleType = null;
this.credentialName = null;
this.noop = true;
}
private static BorrowedCredential noop() {
return new BorrowedCredential();
}
@Override
public void close() {
if (released || noop) {
return;
}
released = true;
owner.release(moduleType, credentialName);
}
}
}

View File

@@ -16,6 +16,7 @@ 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.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
@@ -28,8 +29,26 @@ public class ProductCategoryController {
@GetMapping("/product-categories")
@Operation(summary = "查询商品类目树")
public ApiResponse<ProductCategoryListVo> list() {
return ApiResponse.success(productCategoryService.list());
public ApiResponse<ProductCategoryListVo> list(@RequestParam(value = "keyword", required = false) String keyword) {
return ApiResponse.success(productCategoryService.list(keyword));
}
@GetMapping("/product-categories/children")
@Operation(summary = "分页查询指定父级下的商品类目")
public ApiResponse<ProductCategoryListVo> children(
@RequestParam(value = "parentId", required = false) Long parentId,
@RequestParam(value = "page", defaultValue = "1") Long page,
@RequestParam(value = "pageSize", defaultValue = "20") Long pageSize) {
return ApiResponse.success(productCategoryService.children(parentId, page, pageSize));
}
@GetMapping("/product-categories/search")
@Operation(summary = "分页搜索商品类目")
public ApiResponse<ProductCategoryListVo> search(
@RequestParam(value = "keyword", required = false) String keyword,
@RequestParam(value = "page", defaultValue = "1") Long page,
@RequestParam(value = "pageSize", defaultValue = "20") Long pageSize) {
return ApiResponse.success(productCategoryService.search(keyword, page, pageSize));
}
@PostMapping("/product-category")

View File

@@ -8,4 +8,8 @@ import java.util.List;
public class ProductCategoryListVo {
private List<ProductCategoryItemVo> tree;
private List<ProductCategoryItemVo> items;
private Long total;
private Long page;
private Long pageSize;
private Boolean hasMore;
}

View File

@@ -17,6 +17,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
@@ -27,13 +28,52 @@ public class ProductCategoryService {
private final ProductCategoryMapper productCategoryMapper;
public ProductCategoryListVo list() {
return list(null);
}
public ProductCategoryListVo list(String keyword) {
List<ProductCategoryEntity> rows = productCategoryMapper.selectList(new LambdaQueryWrapper<ProductCategoryEntity>()
.orderByAsc(ProductCategoryEntity::getParentId)
.orderByAsc(ProductCategoryEntity::getSortOrder)
.orderByAsc(ProductCategoryEntity::getId));
rows = filterRowsByKeyword(rows, keyword);
return buildListVo(rows);
}
public ProductCategoryListVo children(Long parentId, long page, long pageSize) {
long safePage = Math.max(page, 1);
long safePageSize = normalizePageSize(pageSize);
Long normalizedParentId = normalizeParentId(parentId);
LambdaQueryWrapper<ProductCategoryEntity> countQuery = new LambdaQueryWrapper<>();
applyParentFilter(countQuery, normalizedParentId);
Long total = productCategoryMapper.selectCount(countQuery);
LambdaQueryWrapper<ProductCategoryEntity> pageQuery = new LambdaQueryWrapper<ProductCategoryEntity>()
.orderByAsc(ProductCategoryEntity::getSortOrder)
.orderByAsc(ProductCategoryEntity::getId)
.last("LIMIT " + ((safePage - 1) * safePageSize) + ", " + safePageSize);
applyParentFilter(pageQuery, normalizedParentId);
List<ProductCategoryEntity> rows = productCategoryMapper.selectList(pageQuery);
return buildPageVo(rows, total, safePage, safePageSize);
}
public ProductCategoryListVo search(String keyword, long page, long pageSize) {
String normalized = normalizeSearchKeyword(keyword);
if (normalized.isBlank()) {
return emptyPageVo(Math.max(page, 1), normalizePageSize(pageSize));
}
long safePage = Math.max(page, 1);
long safePageSize = normalizePageSize(pageSize);
LambdaQueryWrapper<ProductCategoryEntity> countQuery = buildSearchQuery(normalized);
Long total = productCategoryMapper.selectCount(countQuery);
List<ProductCategoryEntity> rows = productCategoryMapper.selectList(buildSearchQuery(normalized)
.orderByAsc(ProductCategoryEntity::getParentId)
.orderByAsc(ProductCategoryEntity::getSortOrder)
.orderByAsc(ProductCategoryEntity::getId)
.last("LIMIT " + ((safePage - 1) * safePageSize) + ", " + safePageSize));
return buildSearchPageVo(rows, total, safePage, safePageSize);
}
@Transactional
public ProductCategoryItemVo create(ProductCategorySaveRequest request) {
String name = normalizeRequired(request.getName(), "类目名称不能为空");
@@ -80,11 +120,13 @@ public class ProductCategoryService {
}
private ProductCategoryItemVo findItem(Long id) {
ProductCategoryListVo listVo = list();
return listVo.getItems().stream()
.filter(item -> id.equals(item.getId()))
.findFirst()
.orElseThrow(() -> new BusinessException("类目不存在"));
ProductCategoryEntity entity = getById(id);
Long childCount = productCategoryMapper.selectCount(new LambdaQueryWrapper<ProductCategoryEntity>()
.eq(ProductCategoryEntity::getParentId, entity.getId()));
ProductCategoryItemVo item = toItemVo(entity, childCount == null ? 0 : childCount.intValue());
item.setLevel(resolveLevel(entity));
item.setPath(resolvePath(entity));
return item;
}
private ProductCategoryListVo buildListVo(List<ProductCategoryEntity> rows) {
@@ -105,9 +147,153 @@ public class ProductCategoryService {
ProductCategoryListVo vo = new ProductCategoryListVo();
vo.setTree(tree);
vo.setItems(flat);
vo.setTotal((long) flat.size());
vo.setPage(1L);
vo.setPageSize((long) flat.size());
vo.setHasMore(false);
return vo;
}
private ProductCategoryListVo buildPageVo(List<ProductCategoryEntity> rows, Long total, long page, long pageSize) {
Map<Long, Integer> childCountById = buildChildCountById(rows);
List<ProductCategoryItemVo> items = rows.stream()
.map(row -> toItemVo(row, childCountById.getOrDefault(row.getId(), 0)))
.toList();
ProductCategoryListVo vo = new ProductCategoryListVo();
vo.setItems(items);
vo.setTree(List.of());
vo.setTotal(total == null ? 0L : total);
vo.setPage(page);
vo.setPageSize(pageSize);
vo.setHasMore(page * pageSize < vo.getTotal());
return vo;
}
private ProductCategoryListVo buildSearchPageVo(List<ProductCategoryEntity> rows, Long total, long page, long pageSize) {
Map<Long, Integer> childCountById = buildChildCountById(rows);
List<ProductCategoryItemVo> items = rows.stream()
.map(row -> {
ProductCategoryItemVo item = toItemVo(row, childCountById.getOrDefault(row.getId(), 0));
item.setLevel(resolveLevel(row));
item.setPath(resolvePath(row));
return item;
})
.toList();
ProductCategoryListVo vo = new ProductCategoryListVo();
vo.setItems(items);
vo.setTree(List.of());
vo.setTotal(total == null ? 0L : total);
vo.setPage(page);
vo.setPageSize(pageSize);
vo.setHasMore(page * pageSize < vo.getTotal());
return vo;
}
private ProductCategoryListVo emptyPageVo(long page, long pageSize) {
ProductCategoryListVo vo = new ProductCategoryListVo();
vo.setItems(List.of());
vo.setTree(List.of());
vo.setTotal(0L);
vo.setPage(page);
vo.setPageSize(pageSize);
vo.setHasMore(false);
return vo;
}
private Map<Long, Integer> buildChildCountById(List<ProductCategoryEntity> rows) {
Map<Long, Integer> childCountById = new HashMap<>();
if (rows.isEmpty()) {
return childCountById;
}
for (ProductCategoryEntity row : rows) {
Long childCount = productCategoryMapper.selectCount(new LambdaQueryWrapper<ProductCategoryEntity>()
.eq(ProductCategoryEntity::getParentId, row.getId()));
childCountById.put(row.getId(), childCount == null ? 0 : childCount.intValue());
}
return childCountById;
}
private void applyParentFilter(LambdaQueryWrapper<ProductCategoryEntity> query, Long parentId) {
if (parentId == null) {
query.isNull(ProductCategoryEntity::getParentId);
} else {
query.eq(ProductCategoryEntity::getParentId, parentId);
}
}
private LambdaQueryWrapper<ProductCategoryEntity> buildSearchQuery(String keyword) {
return new LambdaQueryWrapper<ProductCategoryEntity>()
.like(ProductCategoryEntity::getName, keyword)
.or()
.like(ProductCategoryEntity::getCategoryKey, keyword)
.or()
.like(ProductCategoryEntity::getDescription, keyword);
}
private List<ProductCategoryEntity> filterRowsByKeyword(List<ProductCategoryEntity> rows, String keyword) {
String normalized = normalizeSearchKeyword(keyword);
if (normalized.isBlank()) {
return rows;
}
Map<Long, ProductCategoryEntity> byId = new HashMap<>();
Map<Long, List<ProductCategoryEntity>> byParent = new HashMap<>();
for (ProductCategoryEntity row : rows) {
byId.put(row.getId(), row);
byParent.computeIfAbsent(row.getParentId(), ignored -> new ArrayList<>()).add(row);
}
Set<Long> includedIds = new HashSet<>();
for (ProductCategoryEntity row : rows) {
if (!matchesKeyword(row, normalized)) {
continue;
}
includeAncestors(row, byId, includedIds);
includeDescendants(row, byParent, includedIds);
}
if (includedIds.isEmpty()) {
return List.of();
}
return rows.stream()
.filter(row -> includedIds.contains(row.getId()))
.toList();
}
private String normalizeSearchKeyword(String keyword) {
return keyword == null ? "" : keyword.trim().toLowerCase(Locale.ROOT);
}
private boolean matchesKeyword(ProductCategoryEntity row, String keyword) {
return containsKeyword(row.getName(), keyword)
|| containsKeyword(row.getCategoryKey(), keyword)
|| containsKeyword(row.getDescription(), keyword);
}
private boolean containsKeyword(String value, String keyword) {
return value != null && value.toLowerCase(Locale.ROOT).contains(keyword);
}
private void includeAncestors(ProductCategoryEntity row,
Map<Long, ProductCategoryEntity> byId,
Set<Long> includedIds) {
ProductCategoryEntity cursor = row;
Set<Long> visited = new HashSet<>();
while (cursor != null && cursor.getId() != null && visited.add(cursor.getId())) {
includedIds.add(cursor.getId());
cursor = cursor.getParentId() == null ? null : byId.get(cursor.getParentId());
}
}
private void includeDescendants(ProductCategoryEntity row,
Map<Long, List<ProductCategoryEntity>> byParent,
Set<Long> includedIds) {
if (row == null || row.getId() == null) {
return;
}
includedIds.add(row.getId());
for (ProductCategoryEntity child : byParent.getOrDefault(row.getId(), List.of())) {
includeDescendants(child, byParent, includedIds);
}
}
private List<ProductCategoryItemVo> buildChildren(
Long parentId,
int level,
@@ -159,6 +345,36 @@ public class ProductCategoryService {
return vo;
}
private int resolveLevel(ProductCategoryEntity entity) {
int level = 0;
ProductCategoryEntity cursor = entity;
Set<Long> visited = new HashSet<>();
while (cursor != null && cursor.getParentId() != null && visited.add(cursor.getId())) {
ProductCategoryEntity parent = productCategoryMapper.selectById(cursor.getParentId());
if (parent == null) {
break;
}
level += 1;
cursor = parent;
}
return level;
}
private String resolvePath(ProductCategoryEntity entity) {
List<String> names = new ArrayList<>();
ProductCategoryEntity cursor = entity;
Set<Long> visited = new HashSet<>();
while (cursor != null && cursor.getId() != null && visited.add(cursor.getId())) {
names.add(cursor.getName());
cursor = cursor.getParentId() == null ? null : productCategoryMapper.selectById(cursor.getParentId());
}
List<String> ordered = new ArrayList<>();
for (int i = names.size() - 1; i >= 0; i--) {
ordered.add(names.get(i));
}
return String.join(" / ", ordered);
}
private ProductCategoryEntity getById(Long id) {
ProductCategoryEntity entity = productCategoryMapper.selectById(id);
if (entity == null) {
@@ -225,6 +441,10 @@ public class ProductCategoryService {
return value == null ? 100 : value;
}
private long normalizePageSize(long value) {
return Math.min(Math.max(value, 1), 100);
}
private String normalizeDescription(String value) {
String normalized = value == null ? "" : value.trim();
return normalized.length() > 512 ? normalized.substring(0, 512) : normalized;

View File

@@ -1,6 +1,7 @@
package com.nanri.aiimage.modules.shopkey.controller;
import com.nanri.aiimage.common.api.ApiResponse;
import com.nanri.aiimage.common.util.DownloadHeaderUtil;
import com.nanri.aiimage.modules.shopkey.model.dto.QueryAsinCountryUpdateRequest;
import com.nanri.aiimage.modules.shopkey.model.dto.QueryAsinCreateRequest;
import com.nanri.aiimage.modules.shopkey.model.vo.QueryAsinImportProgressVo;
@@ -13,6 +14,9 @@ import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
@@ -24,12 +28,17 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
@RestController
@RequiredArgsConstructor
@RequestMapping("/api/admin/query-asins")
@Tag(name = "查询 ASIN", description = "维护店铺按国家查询的 ASIN")
public class QueryAsinController {
private static final DateTimeFormatter EXPORT_FILENAME_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss");
private final QueryAsinService queryAsinService;
@GetMapping
@@ -52,6 +61,23 @@ public class QueryAsinController {
Boolean.TRUE.equals(superAdmin)));
}
@GetMapping("/export")
@Operation(summary = "导出查询 ASIN")
public ResponseEntity<byte[]> export(
@Parameter(description = "分组ID") @RequestParam(required = false) Long groupId,
@Parameter(description = "店铺名") @RequestParam(required = false) String shopName,
@Parameter(description = "ASIN") @RequestParam(required = false) String asin,
@Parameter(description = "当前操作人用户ID") @RequestParam(required = false) Long operatorId,
@Parameter(description = "是否超级管理员") @RequestParam(defaultValue = "false") Boolean superAdmin) {
byte[] bytes = queryAsinService.export(groupId, shopName, asin, operatorId, Boolean.TRUE.equals(superAdmin));
String filename = "query-asin-" + LocalDateTime.now().format(EXPORT_FILENAME_FORMATTER) + ".xlsx";
return ResponseEntity.ok()
.header(HttpHeaders.CONTENT_DISPOSITION, DownloadHeaderUtil.contentDisposition(filename))
.contentType(MediaType.parseMediaType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"))
.contentLength(bytes.length)
.body(bytes);
}
@PostMapping
@Operation(summary = "新增或覆盖查询 ASIN")
public ApiResponse<QueryAsinItemVo> create(

View File

@@ -16,12 +16,18 @@ import com.nanri.aiimage.modules.shopkey.model.vo.QueryAsinPageVo;
import com.nanri.aiimage.modules.shopkey.model.vo.ShopManageGroupItemVo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.xssf.streaming.SXSSFWorkbook;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.nio.file.Files;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -41,6 +47,7 @@ public class QueryAsinService {
private static final List<String> SUPPORTED_COUNTRIES = List.of("DE", "UK", "FR", "IT", "ES");
private static final String KEY_SEPARATOR = Character.toString((char) 1);
private static final String UTF8_BOM = String.valueOf((char) 0xFEFF);
private static final DateTimeFormatter EXPORT_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private final QueryAsinMapper queryAsinMapper;
private final ShopManageGroupService shopManageGroupService;
@@ -52,32 +59,15 @@ public class QueryAsinService {
Long operatorId, boolean superAdmin) {
long safePage = Math.max(page, 1);
long safePageSize = Math.min(Math.max(pageSize, 1), 100);
String safeShopName = normalizeBlank(shopName);
String safeAsin = normalizeBlank(asin);
Set<Long> accessibleGroupIds = superAdmin ? Set.of() : shopManageGroupService.listAccessibleGroupIds(operatorId, false);
LambdaQueryWrapper<QueryAsinEntity> query = new LambdaQueryWrapper<QueryAsinEntity>()
.eq(groupId != null && groupId > 0, QueryAsinEntity::getGroupId, groupId)
.like(!safeShopName.isEmpty(), QueryAsinEntity::getShopName, safeShopName)
.and(!safeAsin.isEmpty(), wrapper -> wrapper
.like(QueryAsinEntity::getAsinDe, safeAsin)
.or().like(QueryAsinEntity::getAsinUk, safeAsin)
.or().like(QueryAsinEntity::getAsinFr, safeAsin)
.or().like(QueryAsinEntity::getAsinIt, safeAsin)
.or().like(QueryAsinEntity::getAsinEs, safeAsin))
.orderByDesc(QueryAsinEntity::getId);
if (!superAdmin) {
if (accessibleGroupIds.isEmpty()) {
query.eq(QueryAsinEntity::getId, -1L);
} else {
query.in(QueryAsinEntity::getGroupId, accessibleGroupIds);
}
}
Long total = queryAsinMapper.selectCount(query);
List<QueryAsinEntity> rows = queryAsinMapper
.selectList(query.last("LIMIT " + ((safePage - 1) * safePageSize) + ", " + safePageSize));
Long total = countFilteredRows(groupId, shopName, asin, operatorId, superAdmin);
List<QueryAsinEntity> rows = listFilteredRows(
groupId,
shopName,
asin,
operatorId,
superAdmin,
(safePage - 1) * safePageSize,
safePageSize);
Map<Long, String> groupNameById = shopManageGroupService.buildGroupNameMap(rows.stream()
.map(QueryAsinEntity::getGroupId)
@@ -93,6 +83,110 @@ public class QueryAsinService {
return vo;
}
public byte[] export(Long groupId, String shopName, String asin, Long operatorId, boolean superAdmin) {
List<QueryAsinEntity> rows = listFilteredRows(groupId, shopName, asin, operatorId, superAdmin, null, null);
Map<Long, String> groupNameById = shopManageGroupService.buildGroupNameMap(rows.stream()
.map(QueryAsinEntity::getGroupId)
.filter(id -> id != null && id > 0)
.distinct()
.toList());
try (SXSSFWorkbook workbook = new SXSSFWorkbook(100);
ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
Sheet sheet = workbook.createSheet("QueryAsin");
writeExportHeader(sheet);
for (int index = 0; index < rows.size(); index++) {
writeExportRow(sheet, index + 1, rows.get(index), groupNameById.get(rows.get(index).getGroupId()));
}
applyExportColumnWidths(sheet);
workbook.write(outputStream);
workbook.dispose();
return outputStream.toByteArray();
} catch (Exception ex) {
throw new BusinessException("导出查询 ASIN 失败");
}
}
private Long countFilteredRows(Long groupId, String shopName, String asin, Long operatorId, boolean superAdmin) {
return queryAsinMapper.selectCount(buildFilterQuery(groupId, shopName, asin, operatorId, superAdmin));
}
private List<QueryAsinEntity> listFilteredRows(Long groupId, String shopName, String asin,
Long operatorId, boolean superAdmin,
Long offset, Long limit) {
LambdaQueryWrapper<QueryAsinEntity> query = buildFilterQuery(groupId, shopName, asin, operatorId, superAdmin);
if (offset != null && limit != null) {
query.last("LIMIT " + Math.max(0L, offset) + ", " + Math.max(1L, limit));
}
return queryAsinMapper.selectList(query);
}
private LambdaQueryWrapper<QueryAsinEntity> buildFilterQuery(Long groupId, String shopName, String asin,
Long operatorId, boolean superAdmin) {
String safeShopName = normalizeBlank(shopName);
String safeAsin = normalizeBlank(asin);
Set<Long> accessibleGroupIds = superAdmin ? Set.of() : shopManageGroupService.listAccessibleGroupIds(operatorId, false);
LambdaQueryWrapper<QueryAsinEntity> query = new LambdaQueryWrapper<QueryAsinEntity>()
.eq(groupId != null && groupId > 0, QueryAsinEntity::getGroupId, groupId)
.like(!safeShopName.isEmpty(), QueryAsinEntity::getShopName, safeShopName)
.and(!safeAsin.isEmpty(), wrapper -> wrapper
.like(QueryAsinEntity::getAsinDe, safeAsin)
.or().like(QueryAsinEntity::getAsinUk, safeAsin)
.or().like(QueryAsinEntity::getAsinFr, safeAsin)
.or().like(QueryAsinEntity::getAsinIt, safeAsin)
.or().like(QueryAsinEntity::getAsinEs, safeAsin))
.orderByDesc(QueryAsinEntity::getId);
if (!superAdmin) {
if (accessibleGroupIds.isEmpty()) {
query.eq(QueryAsinEntity::getId, -1L);
} else {
query.in(QueryAsinEntity::getGroupId, accessibleGroupIds);
}
}
return query;
}
private void writeExportHeader(Sheet sheet) {
Row header = sheet.createRow(0);
String[] headers = {
"序号", "分组", "店铺名",
"德国ASIN", "英国ASIN", "法国ASIN", "意大利ASIN", "西班牙ASIN",
"创建时间", "更新时间"
};
for (int i = 0; i < headers.length; i++) {
header.createCell(i).setCellValue(headers[i]);
}
}
private void writeExportRow(Sheet sheet, int rowNo, QueryAsinEntity entity, String groupName) {
Row row = sheet.createRow(rowNo);
int col = 0;
row.createCell(col++).setCellValue(rowNo);
row.createCell(col++).setCellValue(blankToEmpty(groupName));
row.createCell(col++).setCellValue(blankToEmpty(entity.getShopName()));
row.createCell(col++).setCellValue(blankToEmpty(entity.getAsinDe()));
row.createCell(col++).setCellValue(blankToEmpty(entity.getAsinUk()));
row.createCell(col++).setCellValue(blankToEmpty(entity.getAsinFr()));
row.createCell(col++).setCellValue(blankToEmpty(entity.getAsinIt()));
row.createCell(col++).setCellValue(blankToEmpty(entity.getAsinEs()));
row.createCell(col++).setCellValue(formatExportTime(entity.getCreatedAt()));
row.createCell(col).setCellValue(formatExportTime(entity.getUpdatedAt()));
}
private void applyExportColumnWidths(Sheet sheet) {
int[] widths = {
2800, 5200, 5200,
4600, 4600, 4600, 4600, 4600,
5200, 5200
};
for (int i = 0; i < widths.length; i++) {
sheet.setColumnWidth(i, widths[i]);
}
}
private String formatExportTime(LocalDateTime value) {
return value == null ? "" : value.format(EXPORT_TIME_FORMATTER);
}
@Transactional
public QueryAsinItemVo createOrUpdate(QueryAsinCreateRequest request, Long operatorId, boolean superAdmin) {
ShopManageGroupEntity group = shopManageGroupService.getAccessibleById(request.getGroupId(), operatorId, superAdmin);

View File

@@ -3,6 +3,7 @@ package com.nanri.aiimage.modules.similarasin.client;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.SimilarAsinProperties;
import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService;
import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -19,22 +20,26 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
@Component
@RequiredArgsConstructor
@Slf4j
public class SimilarAsinCozeClient {
private static final String MODULE_TYPE = "SIMILAR_ASIN";
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
private final SimilarAsinProperties properties;
private final ObjectMapper objectMapper;
private final CozeCredentialPoolService cozeCredentialPoolService;
private final AtomicLong credentialCursor = new AtomicLong();
public List<SimilarAsinResultRowDto> inspect(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
if (rows == null || rows.isEmpty()) {
return List.of();
}
if (properties.getCozeToken() == null || properties.getCozeToken().isBlank()) {
if (!hasConfiguredCredential()) {
log.warn("[similar-asin] coze token not configured, keep raw rows size={}", rows.size());
return rows.stream().map(this::copy).toList();
}
@@ -48,17 +53,31 @@ public class SimilarAsinCozeClient {
}
public CozeSubmitResponse submitWorkflow(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
return submitWorkflow(rows, prompt, apiKey, nextCredential());
}
public CozeSubmitResponse submitWorkflow(List<SimilarAsinResultRowDto> rows,
String prompt,
String apiKey,
CozeCredentialRef credential) throws Exception {
CozeCredentialRef resolvedCredential = resolveCredential(credential);
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey, resolvedCredential));
ensureSuccess(submitRoot);
return new CozeSubmitResponse(
extractExecuteId(submitRoot),
extractResultDataText(submitRoot),
writeJson(submitRoot)
writeJson(submitRoot),
resolvedCredential.name()
);
}
public CozePollResponse pollWorkflow(String executeId) throws Exception {
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId));
return pollWorkflow(executeId, null);
}
public CozePollResponse pollWorkflow(String executeId, CozeCredentialRef credential) throws Exception {
CozeCredentialRef resolvedCredential = resolveCredential(credential);
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId, resolvedCredential));
ensureSuccess(pollRoot);
String status = normalize(resolveWorkflowStatus(pollRoot)).toUpperCase(Locale.ROOT);
String dataText = extractResultDataText(pollRoot);
@@ -66,7 +85,8 @@ public class SimilarAsinCozeClient {
String failureMessage = isFailedWorkflowStatus(status)
? firstNonBlank(resolveFailureMessage(pollRoot), "Coze async workflow failed")
: "";
return new CozePollResponse(executeId, status, dataText, outputText, failureMessage, writeJson(pollRoot));
return new CozePollResponse(executeId, status, dataText, outputText, failureMessage, writeJson(pollRoot),
resolvedCredential.name());
}
public List<SimilarAsinResultRowDto> mergeRowsFromDataText(List<SimilarAsinResultRowDto> rows, String dataText) throws Exception {
@@ -161,7 +181,8 @@ public class SimilarAsinCozeClient {
}
private String runWorkflowAsyncAndWait(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
CozeCredentialRef credential = nextCredential();
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey, credential));
ensureSuccess(submitRoot);
String immediateData = extractResultDataText(submitRoot);
@@ -177,7 +198,7 @@ public class SimilarAsinCozeClient {
long deadline = System.currentTimeMillis() + Math.max(10000, properties.getCozePollTimeoutMillis());
while (System.currentTimeMillis() < deadline) {
ensureNotInterrupted();
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId));
JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId, credential));
ensureSuccess(pollRoot);
String dataText = extractResultDataText(pollRoot);
@@ -202,23 +223,27 @@ public class SimilarAsinCozeClient {
throw new IllegalStateException("Coze async workflow poll timeout");
}
private String postWorkflow(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
private String postWorkflow(List<SimilarAsinResultRowDto> rows,
String prompt,
String apiKey,
CozeCredentialRef credential) {
Map<String, Object> parameters = buildParameters(rows, prompt, apiKey);
Map<String, Object> body = new LinkedHashMap<>();
body.put("workflow_id", properties.getCozeWorkflowId());
body.put("workflow_id", credential.workflowId());
body.put("parameters", parameters);
if (apiKey != null && !apiKey.isBlank()) {
body.put("api_key", apiKey.trim());
}
body.put("is_async", Boolean.TRUE);
log.info("[similar-asin] coze request url={} body={}",
log.info("[similar-asin] coze request credential={} url={} body={}",
credential.name(),
joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()),
writeJson(maskCozeRequestBody(body)));
RestClient.RequestBodySpec request = restClient().post()
.uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setBearerAuth(stripBearer(credential.token()));
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
});
@@ -233,28 +258,98 @@ public class SimilarAsinCozeClient {
});
}
private String getWorkflowHistory(String executeId) {
private String getWorkflowHistory(String executeId, CozeCredentialRef credential) {
String path = properties.getCozeWorkflowHistoryPath()
.replace("{workflow_id}", properties.getCozeWorkflowId())
.replace("{workflow_id}", credential.workflowId())
.replace("{execute_id}", executeId);
return restClient().get()
.uri(joinUrl(properties.getCozeBaseUrl(), path))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setBearerAuth(stripBearer(credential.token()));
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
})
.exchange((clientRequest, clientResponse) -> {
byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());
String responseText = responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8);
log.info("[similar-asin] coze history response executeId={} status={} body={}",
executeId,
log.info("[similar-asin] coze history response credential={} executeId={} status={} body={}",
credential.name(), executeId,
clientResponse.getStatusCode(),
responseText);
return responseText;
});
}
public CozeCredentialRef nextCredential() {
List<CozeCredentialPoolService.CozeCredential> pooledCredentials = cozeCredentialPoolService.listEnabled(MODULE_TYPE);
CozeCredentialPoolService.CozeCredential pooledCredential =
cozeCredentialPoolService.chooseRoundRobin(MODULE_TYPE, pooledCredentials, properties.getCozeCredentialStripeSize());
if (pooledCredential != null) {
return new CozeCredentialRef(pooledCredential.name(), pooledCredential.workflowId(), pooledCredential.token(),
pooledCredential.maxConcurrent());
}
List<CozeCredentialRef> credentials = configuredCredentials();
int stripeSize = Math.max(1, properties.getCozeCredentialStripeSize());
long cursor = Math.max(0L, credentialCursor.getAndIncrement());
int index = (int) ((cursor / stripeSize) % credentials.size());
return credentials.get(index);
}
public CozeCredentialRef credentialByName(String name) {
if (name == null || name.isBlank()) {
return nextCredential();
}
String normalizedName = normalize(name);
for (CozeCredentialRef credential : configuredCredentials()) {
if (normalize(credential.name()).equals(normalizedName)) {
return credential;
}
}
return nextCredential();
}
public boolean hasConfiguredCredential() {
return !configuredCredentials().isEmpty();
}
public int configuredCredentialCount() {
return configuredCredentials().size();
}
private CozeCredentialRef resolveCredential(CozeCredentialRef credential) {
return credential == null ? nextCredential() : credential;
}
private List<CozeCredentialRef> configuredCredentials() {
List<CozeCredentialRef> credentials = new ArrayList<>();
for (CozeCredentialPoolService.CozeCredential credential : cozeCredentialPoolService.listEnabled(MODULE_TYPE)) {
credentials.add(new CozeCredentialRef(credential.name(), credential.workflowId(), credential.token(),
credential.maxConcurrent()));
}
if (!credentials.isEmpty()) {
return credentials;
}
if (properties.getCozeCredentials() != null) {
int index = 1;
for (SimilarAsinProperties.CozeCredential credential : properties.getCozeCredentials()) {
if (credential == null
|| normalize(credential.getWorkflowId()).isBlank()
|| normalize(credential.getToken()).isBlank()) {
continue;
}
String name = firstNonBlank(credential.getName(), "credential-" + index);
credentials.add(new CozeCredentialRef(name, credential.getWorkflowId(), credential.getToken(), Integer.MAX_VALUE));
index++;
}
}
if (credentials.isEmpty()
&& properties.getCozeWorkflowId() != null && !properties.getCozeWorkflowId().isBlank()
&& properties.getCozeToken() != null && !properties.getCozeToken().isBlank()) {
credentials.add(new CozeCredentialRef("default", properties.getCozeWorkflowId(), properties.getCozeToken(), Integer.MAX_VALUE));
}
return credentials;
}
private Map<String, Object> buildParameters(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
List<String> asins = rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList();
List<String> skus = rows.stream().map(row -> nonBlank(row.getSku(), "")).toList();
@@ -1045,7 +1140,8 @@ public class SimilarAsinCozeClient {
public record CozeSubmitResponse(
String executeId,
String immediateData,
String rawResponse
String rawResponse,
String credentialName
) {
}
@@ -1055,7 +1151,8 @@ public class SimilarAsinCozeClient {
String dataText,
String outputText,
String failureMessage,
String rawResponse
String rawResponse,
String credentialName
) {
public boolean hasPayload() {
return dataText != null && !dataText.isBlank() || outputText != null && !outputText.isBlank();
@@ -1081,6 +1178,14 @@ public class SimilarAsinCozeClient {
}
}
public record CozeCredentialRef(
String name,
String workflowId,
String token,
int maxConcurrent
) {
}
private static final class PartialCozeResultException extends RuntimeException {
private final int resolvedCount;

View File

@@ -11,6 +11,7 @@ import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.config.SimilarAsinProperties;
import com.nanri.aiimage.config.StorageProperties;
import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService;
import com.nanri.aiimage.modules.similarasin.mapper.SimilarAsinFilterConditionMapper;
import com.nanri.aiimage.modules.similarasin.client.SimilarAsinCozeClient;
import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinFilterConditionAddRequest;
@@ -72,6 +73,7 @@ import org.springframework.transaction.support.TransactionTemplate;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
@@ -86,6 +88,8 @@ import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.function.Supplier;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
@Service
@RequiredArgsConstructor
@@ -102,12 +106,13 @@ public class SimilarAsinTaskService {
private static final String COZE_STATUS_DONE = "DONE";
private static final String COZE_STATUS_FAILED = "FAILED";
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final String CONTENT_TYPE_ZIP = "application/zip";
private static final int RESULT_ROWS_READ_RETRY_LIMIT = 3;
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_WAIT_MILLIS = 1000L;
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;
@@ -122,7 +127,7 @@ public class SimilarAsinTaskService {
"是否符合类目",
"不符合理由",
"产品类目",
"status"
"状态"
);
private final LocalFileStorageService localFileStorageService;
@@ -144,6 +149,7 @@ public class SimilarAsinTaskService {
private final PlatformTransactionManager transactionManager;
private final DistributedJobLockService distributedJobLockService;
private final InstanceMetadata instanceMetadata;
private final CozeCredentialPoolService cozeCredentialPoolService;
@Autowired
@Qualifier("cozeTaskExecutor")
private TaskExecutor cozeTaskExecutor;
@@ -821,7 +827,7 @@ public class SimilarAsinTaskService {
touchJavaSideTaskActivity(task.getId());
} else if (isResultSubmissionComplete(task.getId())) {
maybeFinalizeCozeJobLocked(task.getId(), new CozeBatchContext(
job.getId(), result.getId(), context.scopeHash(), context.chunkIndex(), 1, 1, currentInstanceId(), 0));
job.getId(), result.getId(), context.scopeHash(), context.chunkIndex(), 1, 1, currentInstanceId(), 0, null));
}
}
@@ -1431,12 +1437,7 @@ public class SimilarAsinTaskService {
return;
}
try (lockHandle) {
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)
.last("limit 50"));
List<TaskScopeStateEntity> states = listOwnedPendingCozeStates();
if (states == null || states.isEmpty()) {
return;
}
@@ -1458,7 +1459,7 @@ public class SimilarAsinTaskService {
if (chunks == null || chunks.isEmpty()) {
return countPendingCozeStates(task.getId()) > 0;
}
if (properties.getCozeToken() == null || properties.getCozeToken().isBlank()) {
if (!cozeClient.hasConfiguredCredential()) {
log.warn("[similar-asin] coze token not configured, skip async coze taskId={} jobId={}",
task.getId(), job.getId());
return false;
@@ -1512,8 +1513,10 @@ public class SimilarAsinTaskService {
return COZE_STATUS_SUBMITTED.equals(existing.getCozeStatus())
|| COZE_STATUS_RUNNING.equals(existing.getCozeStatus());
}
SimilarAsinCozeClient.CozeCredentialRef credential = cozeClient.nextCredential();
try {
SimilarAsinCozeClient.CozeSubmitResponse submit = submitCozeWorkflowThrottled(batchRows, prompt, apiKey);
SimilarAsinCozeClient.CozeSubmitResponse submit = submitCozeWorkflowThrottled(
batchRows, prompt, apiKey, credential, true);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows = cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
String emptyResultMessage = emptyCozeResultMessage(cozeRows, batchRows.size());
@@ -1532,14 +1535,22 @@ public class SimilarAsinTaskService {
return false;
}
saveCozeBatchState(task, result, job, chunk, batchRows, batchScopeKey, batchScopeHash,
batchIndex, batchTotal, submit.executeId());
log.info("[similar-asin] coze async submitted taskId={} jobId={} chunk={} batch={}/{} executeId={}",
task.getId(), job.getId(), chunk.getChunkIndex(), batchIndex, batchTotal, submit.executeId());
batchIndex, batchTotal, submit.executeId(), submit.credentialName());
log.info("[similar-asin] coze async submitted taskId={} jobId={} chunk={} batch={}/{} credential={} executeId={}",
task.getId(), job.getId(), chunk.getChunkIndex(), batchIndex, batchTotal,
submit.credentialName(), submit.executeId());
return true;
} catch (Exception ex) {
String message = firstNonBlank(ex.getMessage(), "Coze submit failed");
log.warn("[similar-asin] coze async submit failed taskId={} jobId={} chunk={} batch={}/{} err={}",
task.getId(), job.getId(), chunk.getChunkIndex(), batchIndex, batchTotal, message);
if (isCozeThrottleLockTimeout(message)) {
savePendingCozeBatchState(task, result, job, chunk, batchRows, batchScopeKey, batchScopeHash,
batchIndex, batchTotal, message, credential.name());
taskFileJobService.touchRunning(job.getId());
touchJavaSideTaskActivity(task.getId());
return true;
}
mergeCozeRowsIntoChunk(task,
chunk.getScopeHash(),
chunk.getChunkIndex(),
@@ -1549,6 +1560,65 @@ public class SimilarAsinTaskService {
}
}
private boolean isCozeThrottleLockTimeout(String message) {
return normalize(message).toLowerCase(Locale.ROOT).contains("coze submit throttle lock timeout");
}
private void savePendingCozeBatchState(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
TaskChunkEntity chunk,
List<SimilarAsinResultRowDto> batchRows,
String batchScopeKey,
String batchScopeHash,
int batchIndex,
int batchTotal,
String pendingReason,
String credentialName) {
LocalDateTime now = LocalDateTime.now();
CozeBatchContext context = new CozeBatchContext(
job.getId(),
result.getId(),
chunk.getScopeHash(),
chunk.getChunkIndex(),
batchIndex,
batchTotal,
currentInstanceId(),
0,
credentialName
);
String batchPayload = writeJson(batchRows, "serialize pending coze batch payload failed");
String storedBatchPayload = transientPayloadStorageService.storeParsedPayloadFast(
MODULE_TYPE, task.getId(), batchScopeHash, batchPayload, true);
TaskScopeStateEntity state = new TaskScopeStateEntity();
state.setTaskId(task.getId());
state.setModuleType(MODULE_TYPE);
state.setScopeKey(batchScopeKey);
state.setScopeHash(batchScopeHash);
state.setParsedPayloadJson(storedBatchPayload);
state.setStateJson(writeJson(context, "serialize pending coze batch context failed"));
state.setCozeStatus(COZE_STATUS_RUNNING);
state.setCozeSubmittedAt(now);
state.setCozeLastPolledAt(null);
state.setCozeAttemptCount(0);
state.setCozeError(firstNonBlank(pendingReason, "Coze submit queued by throttle"));
state.setChunkTotal(batchTotal);
state.setReceivedChunkCount(batchIndex);
state.setCompleted(0);
state.setCreatedAt(now);
state.setUpdatedAt(now);
try {
taskScopeStateMapper.insert(state);
touchJavaSideTaskActivity(task.getId());
log.info("[similar-asin] coze async submit queued by throttle taskId={} jobId={} chunk={} rows={} batch={}/{}",
task.getId(), job.getId(), chunk.getChunkIndex(), batchRows.size(), batchIndex, batchTotal);
} catch (DuplicateKeyException ex) {
transientPayloadStorageService.deletePayloadIfPresent(storedBatchPayload);
log.info("[similar-asin] duplicate pending coze batch state ignored taskId={} scope={}",
task.getId(), batchScopeKey);
}
}
private void saveCozeBatchState(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
@@ -1558,7 +1628,8 @@ public class SimilarAsinTaskService {
String batchScopeHash,
int batchIndex,
int batchTotal,
String executeId) {
String executeId,
String credentialName) {
LocalDateTime now = LocalDateTime.now();
CozeBatchContext context = new CozeBatchContext(
job.getId(),
@@ -1568,7 +1639,8 @@ public class SimilarAsinTaskService {
batchIndex,
batchTotal,
currentInstanceId(),
0
0,
credentialName
);
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
String storedBatchPayload = transientPayloadStorageService.storeParsedPayloadFast(
@@ -1602,6 +1674,10 @@ public class SimilarAsinTaskService {
private void pollPendingCozeState(Long stateId) {
TaskScopeStateEntity lockState = taskScopeStateMapper.selectById(stateId);
if (lockState == null || lockState.getCozeExecuteId() == null || lockState.getCozeExecuteId().isBlank()) {
if (lockState != null && lockState.getCozeExecuteId() == null
&& (COZE_STATUS_SUBMITTED.equals(lockState.getCozeStatus()) || COZE_STATUS_RUNNING.equals(lockState.getCozeStatus()))) {
retryPendingCozeSubmitState(lockState);
}
return;
}
if (!isCozeStateOwnedByCurrentInstance(lockState)) {
@@ -1621,6 +1697,10 @@ public class SimilarAsinTaskService {
private void pollPendingCozeStateLocked(Long stateId) {
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
if (state != null && state.getCozeExecuteId() == null
&& (COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
retryPendingCozeSubmitState(state);
}
return;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
@@ -1644,7 +1724,9 @@ public class SimilarAsinTaskService {
}
taskFileJobService.touchRunning(context.jobId());
try {
SimilarAsinCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(state.getCozeExecuteId());
SimilarAsinCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(
state.getCozeExecuteId(),
cozeClient.credentialByName(context.credentialName()));
if (!poll.hasPayload() && !poll.isFinished() && !isCozeStateTimedOut(state)) {
updateCozeStateRunning(state, null);
return;
@@ -1725,7 +1807,8 @@ public class SimilarAsinTaskService {
}
try {
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task));
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows =
@@ -1774,23 +1857,54 @@ public class SimilarAsinTaskService {
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;
String apiKey,
SimilarAsinCozeClient.CozeCredentialRef credential,
boolean allowCredentialFallback) throws Exception {
int attempts = allowCredentialFallback ? Math.max(1, cozeClient.configuredCredentialCount()) : 1;
SimilarAsinCozeClient.CozeCredentialRef currentCredential =
credential == null ? cozeClient.nextCredential() : credential;
Exception lastFailure = null;
for (int i = 0; i < attempts; i++) {
DistributedJobLockService.LockHandle lockHandle = acquireCozeSubmitLock(currentCredential);
if (lockHandle == null) {
lastFailure = new IllegalStateException("Coze submit throttle lock timeout");
currentCredential = cozeClient.nextCredential();
continue;
}
CozeCredentialPoolService.BorrowedCredential borrowedCredential =
cozeCredentialPoolService.borrow(MODULE_TYPE, toPoolCredential(currentCredential));
if (borrowedCredential == null) {
lockHandle.close();
lastFailure = new IllegalStateException("Coze credential concurrency limit reached");
currentCredential = cozeClient.nextCredential();
continue;
}
try (lockHandle; borrowedCredential) {
return cozeClient.submitWorkflow(rows, prompt, apiKey, currentCredential);
} finally {
sleepQuietly(COZE_SUBMIT_MIN_INTERVAL_MILLIS);
}
}
throw lastFailure == null ? new IllegalStateException("Coze submit failed") : lastFailure;
}
private DistributedJobLockService.LockHandle acquireCozeSubmitLock() {
private CozeCredentialPoolService.CozeCredential toPoolCredential(SimilarAsinCozeClient.CozeCredentialRef credential) {
if (credential == null) {
return null;
}
return new CozeCredentialPoolService.CozeCredential(
credential.name(),
credential.workflowId(),
credential.token(),
credential.maxConcurrent());
}
private DistributedJobLockService.LockHandle acquireCozeSubmitLock(SimilarAsinCozeClient.CozeCredentialRef credential) {
long deadline = System.currentTimeMillis() + COZE_SUBMIT_LOCK_WAIT_MILLIS;
String credentialName = credential == null ? "default" : firstNonBlank(credential.name(), "default");
while (System.currentTimeMillis() <= deadline) {
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("similar-asin:coze-submit", COZE_SUBMIT_LOCK_TTL);
distributedJobLockService.tryLock("similar-asin:coze-submit:" + credentialName, COZE_SUBMIT_LOCK_TTL);
if (lockHandle != null) {
return lockHandle;
}
@@ -1836,14 +1950,16 @@ public class SimilarAsinTaskService {
int partIndex = 1;
for (List<SimilarAsinResultRowDto> partRows : partitions) {
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(partRows, readAiPrompt(task), readApiKey(task));
submitCozeWorkflowThrottled(partRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
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());
saveSplitRetryCozeBatchState(state, context, partRows, partIndex, partitions.size(), retryCount,
submit.executeId(), submit.credentialName());
submittedAny = true;
}
partIndex++;
@@ -1871,7 +1987,8 @@ public class SimilarAsinTaskService {
int partIndex,
int partTotal,
int retryCount,
String executeId) {
String executeId,
String credentialName) {
String scopeKey = parent.getScopeKey() + ":split:" + retryCount + ":" + partIndex;
String scopeHash = DigestUtil.sha256Hex(scopeKey);
CozeBatchContext context = new CozeBatchContext(
@@ -1882,7 +1999,8 @@ public class SimilarAsinTaskService {
partIndex,
partTotal,
parentContext.ownerInstanceId(),
retryCount
retryCount,
firstNonBlank(credentialName, parentContext.credentialName())
);
LocalDateTime now = LocalDateTime.now();
String batchPayload = writeJson(batchRows, "serialize split coze batch payload failed");
@@ -1926,6 +2044,130 @@ public class SimilarAsinTaskService {
|| normalized.contains("\u8c03\u7528\u8d85\u65f6");
}
private void retryPendingCozeSubmitState(TaskScopeStateEntity state) {
if (state == null || state.getId() == null) {
return;
}
if (!tryClaimCozeStateForPoll(state)) {
log.info("[similar-asin] coze submit retry skipped by claim guard taskId={} stateId={} lastPolledAt={}",
state.getTaskId(), state.getId(), state.getCozeLastPolledAt());
return;
}
CozeBatchContext context = readCozeBatchContext(state);
if (context == null || context.jobId() == null || context.resultId() == null) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit context missing");
return;
}
List<SimilarAsinResultRowDto> batchRows = readCozeBatchRows(state);
if (batchRows.isEmpty()) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit payload missing");
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit task missing");
return;
}
TaskChunkEntity currentChunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, task.getId())
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.eq(TaskChunkEntity::getScopeHash, context.chunkScopeHash())
.eq(TaskChunkEntity::getChunkIndex, context.chunkIndex())
.last("limit 1"));
Map<String, SimilarAsinResultRowDto> currentRows = readChunkRows(currentChunk);
List<SimilarAsinResultRowDto> currentBatchRows = batchRows.stream()
.map(row -> currentRows.get(rowKey(row)))
.filter(Objects::nonNull)
.toList();
if (currentBatchRows.size() == batchRows.size()
&& currentBatchRows.stream().allMatch(this::hasResolvedCozeFields)) {
markCozeStateTerminal(state, COZE_STATUS_DONE, "Coze rows already resolved by another batch");
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
log.info("[similar-asin] coze pending submit skipped because rows already resolved taskId={} stateId={} jobId={} rows={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size());
return;
}
try {
taskFileJobService.touchRunning(context.jobId());
SimilarAsinCozeClient.CozeSubmitResponse submit =
submitCozeWorkflowThrottled(batchRows, readAiPrompt(task), readApiKey(task),
cozeClient.credentialByName(context.credentialName()), false);
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows =
cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
String emptyResultMessage = emptyCozeResultMessage(cozeRows, batchRows.size());
if (!emptyResultMessage.isBlank()) {
throw new IllegalStateException(emptyResultMessage);
}
mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
markCozeStateTerminal(state, COZE_STATUS_DONE, null);
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
if (submit.executeId() == null || submit.executeId().isBlank()) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze async execute_id missing");
mergeCozeRowsIntoChunk(task,
context.chunkScopeHash(),
context.chunkIndex(),
cozeClient.markRowsFailed(batchRows, "Coze async execute_id missing"),
allRowsByBaseId);
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
LocalDateTime now = LocalDateTime.now();
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::getCozeError, null)
.set(TaskScopeStateEntity::getUpdatedAt, now));
touchJavaSideTaskActivity(state.getTaskId());
log.info("[similar-asin] coze pending submit retried taskId={} stateId={} jobId={} rows={} executeId={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size(), submit.executeId());
} catch (Exception ex) {
String message = firstNonBlank(ex.getMessage(), "Coze pending submit retry failed");
log.warn("[similar-asin] coze pending submit retry failed taskId={} stateId={} jobId={} rows={} err={}",
state.getTaskId(), state.getId(), context.jobId(), batchRows.size(), message);
if (isCozeThrottleLockTimeout(message)) {
keepPendingCozeSubmitState(state, message);
return;
}
int nextAttemptCount = cozeAttemptCount(state) + 1;
if (nextAttemptCount >= MAX_COZE_SUBMIT_RETRY_COUNT) {
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
String finalMessage = message + " after " + nextAttemptCount + " submit attempts";
mergeCozeRowsIntoChunk(task,
context.chunkScopeHash(),
context.chunkIndex(),
cozeClient.markRowsFailed(batchRows, finalMessage),
allRowsByBaseId);
markCozeStateTerminal(state, COZE_STATUS_FAILED, finalMessage);
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
updateCozeStateRunning(state, message);
}
}
private void keepPendingCozeSubmitState(TaskScopeStateEntity state, String error) {
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::getCozeStatus, COZE_STATUS_RUNNING)
.set(TaskScopeStateEntity::getCozeExecuteId, null)
.set(TaskScopeStateEntity::getCozeLastPolledAt, LocalDateTime.now())
.set(TaskScopeStateEntity::getCozeError, error)
.set(TaskScopeStateEntity::getUpdatedAt, LocalDateTime.now()));
if (updated > 0) {
touchJavaSideTaskActivity(state.getTaskId());
}
}
private boolean isRetryableCozeFailure(String failureMessage) {
String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
return normalized.contains("rate limit")
@@ -2173,7 +2415,8 @@ public class SimilarAsinTaskService {
context.batchIndex(),
context.batchTotal(),
context.ownerInstanceId(),
submitRetryCount
submitRetryCount,
context.credentialName()
);
}
@@ -2338,6 +2581,12 @@ public class SimilarAsinTaskService {
}
int safeTotal = Math.max(1, total);
int safeCompleted = Math.max(0, Math.min(completed, safeTotal));
TaskProgressSnapshotEntity existing = taskProgressSnapshotService.find(task.getId(), MODULE_TYPE);
int displayPercent = calculateDisplayProgressPercent(safeCompleted, safeTotal, job, LocalDateTime.now());
int previousDisplayPercent = calculateSnapshotDisplayPercent(existing, job);
if (previousDisplayPercent > displayPercent && safeCompleted < safeTotal) {
displayPercent = previousDisplayPercent;
}
taskProgressSnapshotService.save(
task.getId(),
MODULE_TYPE,
@@ -2347,10 +2596,57 @@ public class SimilarAsinTaskService {
0,
job.getScopeKey(),
message,
Map.of("phase", "RESULT_FILE", "jobId", job.getId())
Map.of("phase", "RESULT_FILE", "jobId", job.getId(), "displayPercent", displayPercent)
);
}
private int calculateDisplayProgressPercent(int current,
int total,
TaskFileJobEntity job,
LocalDateTime baseTime) {
if (total <= 0) {
return 0;
}
current = Math.max(0, Math.min(current, total));
int percent = Math.max(1, Math.min(99, (int) Math.floor(current * 100.0 / total)));
if (job != null && STATUS_RUNNING.equals(job.getStatus())) {
long elapsedSeconds = baseTime == null ? 0 : Math.max(0, Duration.between(baseTime, LocalDateTime.now()).getSeconds());
if (current <= 0) {
int firstRealProgressPercent = Math.max(1, Math.min(99, (int) Math.floor(100.0 / total)));
int waitingCap = Math.max(8, Math.min(35, firstRealProgressPercent - 1));
percent = Math.max(percent, Math.min(waitingCap, 8 + (int) (elapsedSeconds / 6)));
} else if (current < total) {
percent = Math.max(percent, Math.min(92, percent + (int) (elapsedSeconds / 10)));
}
}
return percent;
}
private int calculateSnapshotDisplayPercent(TaskProgressSnapshotEntity snapshot, TaskFileJobEntity job) {
if (snapshot == null) {
return 0;
}
int total = snapshot.getTotalCount() == null ? 0 : snapshot.getTotalCount();
int current = snapshot.getSuccessCount() == null ? 0 : snapshot.getSuccessCount();
int percent = total <= 0 ? 0 : calculateDisplayProgressPercent(current, total, job, snapshot.getUpdatedAt());
return Math.max(percent, extractSnapshotDisplayPercent(snapshot));
}
private int extractSnapshotDisplayPercent(TaskProgressSnapshotEntity snapshot) {
if (snapshot == null || snapshot.getSnapshotJson() == null || snapshot.getSnapshotJson().isBlank()) {
return 0;
}
try {
JsonNode displayPercent = objectMapper.readTree(snapshot.getSnapshotJson()).path("displayPercent");
if (!displayPercent.isNumber()) {
return 0;
}
return Math.max(0, Math.min(100, displayPercent.asInt()));
} catch (Exception ignored) {
return 0;
}
}
public void cleanupResultFileJob(TaskFileJobEntity job) {
if (job == null || job.getTaskId() == null) {
return;
@@ -2379,28 +2675,195 @@ public class SimilarAsinTaskService {
if (!outputDir.exists() && !outputDir.mkdirs()) {
throw new BusinessException("创建结果目录失败");
}
String filename = safeFileStem(result.getSourceFilename()) + "-result.xlsx";
String tempFilename = safeFileStem(result.getSourceFilename())
+ "-" + task.getId()
+ "-" + result.getId()
+ "-" + UUID.randomUUID()
+ "-result.xlsx";
File xlsx = new File(outputDir, tempFilename);
List<SourceRows> sourceRows = splitRowsBySourceFile(parsed, parsed.getAllItems(), result.getSourceFilename());
List<SourceResultWorkbook> workbooks = new ArrayList<>();
File zip = null;
try {
writeResultWorkbook(xlsx, parsed, resultMap);
String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
for (SourceRows item : sourceRows) {
String filename = safeFileStem(item.sourceFilename()) + "-result.xlsx";
String tempFilename = safeFileStem(item.sourceFilename())
+ "-" + task.getId()
+ "-" + result.getId()
+ "-" + UUID.randomUUID()
+ "-result.xlsx";
File xlsx = new File(outputDir, tempFilename);
writeResultWorkbook(xlsx, item.rows(), resultMap);
workbooks.add(new SourceResultWorkbook(xlsx, filename, item.rows().size()));
}
if (workbooks.isEmpty()) {
throw new BusinessException("相似ASIN检测结果为空请稍后重试生成结果文件");
}
File uploadFile;
String filename;
String contentType;
if (workbooks.size() == 1) {
SourceResultWorkbook workbook = workbooks.get(0);
uploadFile = workbook.file();
filename = workbook.filename();
contentType = CONTENT_TYPE_XLSX;
} else {
filename = safeFileStem(result.getSourceFilename()) + "-result.zip";
String tempFilename = safeFileStem(result.getSourceFilename())
+ "-" + task.getId()
+ "-" + result.getId()
+ "-" + UUID.randomUUID()
+ "-result.zip";
zip = new File(outputDir, tempFilename);
packageResultWorkbooksAsZip(zip, workbooks);
uploadFile = zip;
contentType = CONTENT_TYPE_ZIP;
}
String objectKey = ossStorageService.uploadResultFile(uploadFile, MODULE_TYPE);
result.setResultFilename(filename);
result.setResultFileUrl(objectKey);
result.setResultFileSize(xlsx.length());
result.setResultContentType(CONTENT_TYPE_XLSX);
result.setResultFileSize(uploadFile.length());
result.setResultContentType(contentType);
result.setRowCount(parsed.getAllItems().size());
} finally {
if (xlsx.exists() && !xlsx.delete()) {
log.warn("[similar-asin] delete temp xlsx failed file={}", xlsx);
for (SourceResultWorkbook workbook : workbooks) {
if (workbook.file().exists() && !workbook.file().delete()) {
log.warn("[similar-asin] delete temp xlsx failed file={}", workbook.file());
}
}
if (zip != null && zip.exists() && !zip.delete()) {
log.warn("[similar-asin] delete temp zip failed file={}", zip);
}
}
}
private List<TaskScopeStateEntity> listOwnedPendingCozeStates() {
Map<Long, TaskScopeStateEntity> merged = new LinkedHashMap<>();
for (TaskScopeStateEntity state : queryOwnedPendingCozeStates(false, 50)) {
if (state != null && state.getId() != null) {
merged.put(state.getId(), state);
}
}
for (TaskScopeStateEntity state : queryOwnedPendingCozeStates(true, 50)) {
if (state != null && state.getId() != null) {
merged.putIfAbsent(state.getId(), state);
}
}
return new ArrayList<>(merged.values());
}
private List<TaskScopeStateEntity> queryOwnedPendingCozeStates(boolean oldestFirst, int limit) {
LambdaQueryWrapper<TaskScopeStateEntity> wrapper = new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING))
.and(query -> query
.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()))
.last("limit " + Math.max(1, Math.min(limit, 100)));
if (oldestFirst) {
wrapper.orderByAsc(TaskScopeStateEntity::getUpdatedAt);
} else {
wrapper.orderByDesc(TaskScopeStateEntity::getUpdatedAt);
}
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(wrapper);
return states == null ? List.of() : states;
}
private List<SourceRows> splitRowsBySourceFile(SimilarAsinParsedPayloadDto parsed,
List<SimilarAsinParsedRowVo> rows,
String fallbackFilename) {
String defaultFilename = firstNonBlank(fallbackFilename, "similar-asin");
Map<String, SourceRowsBuilder> builders = new LinkedHashMap<>();
if (parsed != null && parsed.getSourceFiles() != null) {
for (SimilarAsinSourceFileDto sourceFile : parsed.getSourceFiles()) {
if (sourceFile == null) {
continue;
}
String sourceFileKey = normalize(sourceFile.getFileKey());
if (sourceFileKey.isBlank()) {
continue;
}
builders.putIfAbsent(sourceFileKey,
new SourceRowsBuilder(sourceFileKey, firstNonBlank(sourceFile.getOriginalFilename(), defaultFilename)));
}
}
for (SimilarAsinParsedRowVo row : rows == null ? List.<SimilarAsinParsedRowVo>of() : rows) {
if (row == null) {
continue;
}
String sourceFileKey = normalize(row.getSourceFileKey());
if (sourceFileKey.isBlank()) {
sourceFileKey = normalize(row.getSourceFilename());
}
if (sourceFileKey.isBlank()) {
sourceFileKey = "__default__";
}
SourceRowsBuilder builder = builders.computeIfAbsent(sourceFileKey,
key -> new SourceRowsBuilder(key, firstNonBlank(row.getSourceFilename(), defaultFilename)));
if (builder.sourceFilename().isBlank() || Objects.equals(builder.sourceFilename(), defaultFilename)) {
builder.setSourceFilename(firstNonBlank(row.getSourceFilename(), defaultFilename));
}
builder.rows().add(row);
}
return builders.values().stream()
.filter(builder -> !builder.rows().isEmpty())
.map(builder -> new SourceRows(builder.sourceFileKey(), firstNonBlank(builder.sourceFilename(), defaultFilename), builder.rows()))
.toList();
}
private void packageResultWorkbooksAsZip(File zip, List<SourceResultWorkbook> workbooks) {
Set<String> entryNames = new LinkedHashSet<>();
try (ZipOutputStream zos = new ZipOutputStream(new FileOutputStream(zip))) {
byte[] buffer = new byte[8192];
for (SourceResultWorkbook workbook : workbooks) {
writeResultZipEntry(zos, buffer, workbook.file(), uniqueZipEntryName(workbook.filename(), entryNames));
}
} catch (IOException ex) {
throw new BusinessException("生成相似ASIN检测结果压缩包失败");
}
}
private void writeResultZipEntry(ZipOutputStream zos,
byte[] buffer,
File file,
String entryName) throws IOException {
zos.putNextEntry(new ZipEntry(entryName));
try (FileInputStream inputStream = new FileInputStream(file)) {
int len;
while ((len = inputStream.read(buffer)) > 0) {
zos.write(buffer, 0, len);
}
}
zos.closeEntry();
}
private String uniqueZipEntryName(String filename, Set<String> entryNames) {
String safeFilename = normalizeZipFilename(filename);
if (entryNames.add(safeFilename)) {
return safeFilename;
}
int dotIndex = safeFilename.lastIndexOf('.');
String stem = dotIndex > 0 ? safeFilename.substring(0, dotIndex) : safeFilename;
String extension = dotIndex > 0 ? safeFilename.substring(dotIndex) : "";
int index = 2;
while (true) {
String candidate = stem + "(" + index + ")" + extension;
if (entryNames.add(candidate)) {
return candidate;
}
index++;
}
}
private String normalizeZipFilename(String filename) {
String normalized = firstNonBlank(filename, "similar-asin-result.xlsx").replace('\\', '/');
while (normalized.startsWith("/")) {
normalized = normalized.substring(1);
}
return normalized.isBlank() ? "similar-asin-result.xlsx" : normalized;
}
private Map<String, SimilarAsinResultRowDto> loadPersistedResultRows(Long taskId) {
Map<String, SimilarAsinResultRowDto> result = new LinkedHashMap<>();
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
@@ -2450,7 +2913,9 @@ public class SimilarAsinTaskService {
}
}
private void writeResultWorkbook(File xlsx, SimilarAsinParsedPayloadDto parsed, Map<String, SimilarAsinResultRowDto> resultMap) {
private void writeResultWorkbook(File xlsx,
List<SimilarAsinParsedRowVo> rowsToWrite,
Map<String, SimilarAsinResultRowDto> resultMap) {
try (SXSSFWorkbook workbook = new SXSSFWorkbook(200); FileOutputStream fos = new FileOutputStream(xlsx)) {
Sheet sheet = workbook.createSheet("相似asin检测");
CellStyle headerStyle = workbook.createCellStyle();
@@ -2467,7 +2932,7 @@ public class SimilarAsinTaskService {
}
int rowIndex = 1;
for (SimilarAsinParsedRowVo parsedRow : parsed.getAllItems()) {
for (SimilarAsinParsedRowVo parsedRow : rowsToWrite == null ? List.<SimilarAsinParsedRowVo>of() : rowsToWrite) {
SimilarAsinResultRowDto resultRow = findResultRow(parsedRow, resultMap);
Row row = sheet.createRow(rowIndex++);
int col = 0;
@@ -2807,23 +3272,25 @@ public class SimilarAsinTaskService {
}
int cozeCompleted = countCompletedCozeStates(taskId);
int cozePending = countPendingCozeStates(taskId);
TaskProgressSnapshotEntity snapshot = taskProgressSnapshotService.find(taskId, MODULE_TYPE);
if (job != null && STATUS_RUNNING.equals(job.getStatus()) && cozePending > 0) {
int total = Math.max(1, cozeCompleted + cozePending);
int current = Math.max(0, Math.min(cozeCompleted, total));
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(Math.max(0, Math.min(100, (int) Math.floor(current * 100.0 / total))));
int percent = calculateDisplayProgressPercent(current, total, job, snapshot == null ? null : snapshot.getUpdatedAt());
percent = Math.max(percent, calculateSnapshotDisplayPercent(snapshot, job));
vo.setFileProgressPercent(Math.min(99, percent));
vo.setFileProgressMessage("Coze 已完成 " + current + "/" + total + ",等待回流");
return;
}
TaskProgressSnapshotEntity snapshot = taskProgressSnapshotService.find(taskId, MODULE_TYPE);
if (snapshot == null) {
if (job != null && STATUS_RUNNING.equals(job.getStatus()) && cozeCompleted + cozePending > 0) {
int total = Math.max(1, cozeCompleted + cozePending);
int current = Math.max(0, Math.min(cozeCompleted, total));
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(Math.max(0, Math.min(100, (int) Math.floor(current * 100.0 / total))));
vo.setFileProgressPercent(Math.min(99, calculateDisplayProgressPercent(current, total, job, job.getUpdatedAt())));
vo.setFileProgressMessage(cozePending > 0
? "Coze 已完成 " + current + "/" + total + ",等待回流"
: "Coze 处理中");
@@ -2835,8 +3302,10 @@ public class SimilarAsinTaskService {
if (total <= 0) {
return;
}
current = Math.max(0, Math.min(current, total));
int percent = Math.max(0, Math.min(100, (int) Math.floor(current * 100.0 / total)));
LocalDateTime baseTime = snapshot.getUpdatedAt() != null ? snapshot.getUpdatedAt() : (job == null ? null : job.getUpdatedAt());
int percent = calculateDisplayProgressPercent(current, total, job, baseTime);
percent = Math.max(percent, extractSnapshotDisplayPercent(snapshot));
percent = Boolean.TRUE.equals(vo.getFileReady()) ? 100 : Math.min(99, percent);
vo.setFileProgressCurrent(current);
vo.setFileProgressTotal(total);
vo.setFileProgressPercent(percent);
@@ -3138,7 +3607,45 @@ public class SimilarAsinTaskService {
Integer batchIndex,
Integer batchTotal,
String ownerInstanceId,
Integer submitRetryCount) {
Integer submitRetryCount,
String credentialName) {
}
private static class SourceRowsBuilder {
private final String sourceFileKey;
private String sourceFilename;
private final List<SimilarAsinParsedRowVo> rows = new ArrayList<>();
private SourceRowsBuilder(String sourceFileKey, String sourceFilename) {
this.sourceFileKey = sourceFileKey;
this.sourceFilename = sourceFilename;
}
private String sourceFileKey() {
return sourceFileKey;
}
private String sourceFilename() {
return sourceFilename;
}
private void setSourceFilename(String sourceFilename) {
this.sourceFilename = sourceFilename;
}
private List<SimilarAsinParsedRowVo> rows() {
return rows;
}
}
private record SourceRows(String sourceFileKey,
String sourceFilename,
List<SimilarAsinParsedRowVo> rows) {
}
private record SourceResultWorkbook(File file,
String filename,
int rowCount) {
}
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<SimilarAsinParsedRowVo> allRows) {

View File

@@ -4,6 +4,8 @@ import com.nanri.aiimage.modules.task.model.dto.TaskFileJobDispatchEvent;
import com.nanri.aiimage.modules.task.model.dto.TaskFileJobMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
@@ -15,6 +17,8 @@ public class TaskFileJobDispatchCoordinator {
private final TaskFileJobPublisher taskFileJobPublisher;
private final TaskFileJobLocalDispatcher taskFileJobLocalDispatcher;
@Qualifier("taskFileJobDispatchExecutor")
private final TaskExecutor taskFileJobDispatchExecutor;
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
public void onTaskFileJobDispatch(TaskFileJobDispatchEvent event) {
@@ -29,6 +33,16 @@ public class TaskFileJobDispatchCoordinator {
message.setScopeKey(event.scopeKey());
message.setJobType(event.jobType());
message.setCreatedAt(event.createdAt());
try {
taskFileJobDispatchExecutor.execute(() -> dispatchAfterCommit(event, message));
} catch (RuntimeException ex) {
log.warn("[task-file-job] dispatch executor rejected after commit jobId={} taskId={} moduleType={} msg={}",
event.jobId(), event.taskId(), event.moduleType(), ex.getMessage(), ex);
dispatchAfterCommit(event, message);
}
}
private void dispatchAfterCommit(TaskFileJobDispatchEvent event, TaskFileJobMessage message) {
boolean published = taskFileJobPublisher.publish(message);
if (published) {
return;

View File

@@ -8,12 +8,17 @@ import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@Slf4j
@Service
@RequiredArgsConstructor
public class TaskFileJobPublisher {
private final ObjectProvider<RocketMQTemplate> rocketMQTemplateProvider;
private final AtomicInteger consecutiveFailures = new AtomicInteger();
private final AtomicLong circuitOpenUntilMillis = new AtomicLong();
@Value("${aiimage.result-file-job.mq-enabled:false}")
private boolean mqEnabled;
@@ -21,6 +26,15 @@ public class TaskFileJobPublisher {
@Value("${aiimage.result-file-job.topic:aiimage-result-file-job}")
private String topic;
@Value("${aiimage.result-file-job.mq-send-timeout-ms:1000}")
private long mqSendTimeoutMillis;
@Value("${aiimage.result-file-job.mq-failure-threshold:3}")
private int mqFailureThreshold;
@Value("${aiimage.result-file-job.mq-circuit-open-ms:60000}")
private long mqCircuitOpenMillis;
public boolean publish(TaskFileJobMessage message) {
if (message == null || message.getJobId() == null) {
return false;
@@ -30,6 +44,13 @@ public class TaskFileJobPublisher {
message.getJobId(), message.getTaskId(), message.getModuleType());
return false;
}
long now = System.currentTimeMillis();
long openUntil = circuitOpenUntilMillis.get();
if (openUntil > now) {
log.info("[task-file-job] RocketMQ publish circuit open, fallback to local dispatcher/scheduled worker jobId={} taskId={} moduleType={} remainingMs={}",
message.getJobId(), message.getTaskId(), message.getModuleType(), openUntil - now);
return false;
}
RocketMQTemplate template = rocketMQTemplateProvider.getIfAvailable();
if (template == null) {
log.warn("[task-file-job] RocketMQTemplate unavailable, jobId={} taskId={} moduleType={}",
@@ -37,13 +58,25 @@ public class TaskFileJobPublisher {
return false;
}
try {
template.convertAndSend(topic, message);
template.syncSend(topic, message, Math.max(100L, mqSendTimeoutMillis));
consecutiveFailures.set(0);
circuitOpenUntilMillis.set(0L);
log.info("[task-file-job] mq published jobId={} taskId={} moduleType={} topic={}",
message.getJobId(), message.getTaskId(), message.getModuleType(), topic);
return true;
} catch (Exception ex) {
log.warn("[task-file-job] publish RocketMQ failed, fallback to local dispatcher/scheduled worker jobId={} taskId={} moduleType={} msg={}",
message.getJobId(), message.getTaskId(), message.getModuleType(), ex.getMessage());
int failures = consecutiveFailures.incrementAndGet();
int threshold = Math.max(1, mqFailureThreshold);
if (failures >= threshold) {
long nextOpenUntil = System.currentTimeMillis() + Math.max(1000L, mqCircuitOpenMillis);
circuitOpenUntilMillis.set(nextOpenUntil);
log.warn("[task-file-job] publish RocketMQ failed, circuit opened and fallback to local dispatcher/scheduled worker jobId={} taskId={} moduleType={} failures={} openMs={} msg={}",
message.getJobId(), message.getTaskId(), message.getModuleType(), failures,
Math.max(1000L, mqCircuitOpenMillis), ex.getMessage());
} else {
log.warn("[task-file-job] publish RocketMQ failed, fallback to local dispatcher/scheduled worker jobId={} taskId={} moduleType={} failures={}/{} msg={}",
message.getJobId(), message.getTaskId(), message.getModuleType(), failures, threshold, ex.getMessage());
}
return false;
}
}

View File

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -76,6 +77,41 @@ public class TaskFileJobService {
.last("limit " + Math.max(1, Math.min(limit, 100))));
}
public List<TaskFileJobEntity> listRunnableJobsForOwner(int limit, String owner) {
String normalizedOwner = owner == null ? "" : owner.trim();
if (normalizedOwner.isBlank()) {
return listRunnableJobs(limit);
}
String ownerMarker = ":owner:" + normalizedOwner;
int safeLimit = Math.max(1, Math.min(limit, 100));
List<TaskFileJobEntity> ownerJobs = taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
.lt(TaskFileJobEntity::getRetryCount, 5)
.in(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN"))
.like(TaskFileJobEntity::getScopeKey, ownerMarker)
.orderByAsc(TaskFileJobEntity::getUpdatedAt)
.last("limit " + safeLimit));
if (ownerJobs.size() >= safeLimit) {
return ownerJobs;
}
List<TaskFileJobEntity> jobs = new ArrayList<>(ownerJobs);
LambdaQueryWrapper<TaskFileJobEntity> genericWrapper = new LambdaQueryWrapper<TaskFileJobEntity>()
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
.lt(TaskFileJobEntity::getRetryCount, 5)
.and(wrapper -> wrapper
.notIn(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN"))
.or()
.isNull(TaskFileJobEntity::getScopeKey)
.or()
.notLike(TaskFileJobEntity::getScopeKey, ":owner:"))
.orderByAsc(TaskFileJobEntity::getUpdatedAt)
.last("limit " + (safeLimit - jobs.size()));
jobs.addAll(taskFileJobMapper.selectList(genericWrapper));
return jobs;
}
public List<TaskFileJobEntity> listJobs(String status, String moduleType, Long taskId, int limit) {
LambdaQueryWrapper<TaskFileJobEntity> wrapper = new LambdaQueryWrapper<TaskFileJobEntity>()
.orderByDesc(TaskFileJobEntity::getUpdatedAt)
@@ -130,7 +166,8 @@ public class TaskFileJobService {
.eq(TaskFileJobEntity::getStatus, "RUNNING")
.set(TaskFileJobEntity::getStatus, "FAILED")
.set(TaskFileJobEntity::getErrorMessage, "文件生成任务运行超时")
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()));
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())
.set(TaskFileJobEntity::getFinishedAt, LocalDateTime.now()));
continue;
}
int updated = taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
@@ -168,6 +205,19 @@ public class TaskFileJobService {
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()));
}
public boolean deferRunning(Long jobId, String message) {
if (jobId == null || jobId <= 0) {
return false;
}
return taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
.eq(TaskFileJobEntity::getId, jobId)
.eq(TaskFileJobEntity::getStatus, "RUNNING")
.set(TaskFileJobEntity::getStatus, "PENDING")
.set(TaskFileJobEntity::getErrorMessage, message)
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())
.set(TaskFileJobEntity::getFinishedAt, null)) > 0;
}
@Transactional
public boolean requeue(Long jobId, String message) {
if (jobId == null || jobId <= 0) {
@@ -198,14 +248,23 @@ public class TaskFileJobService {
.set(TaskFileJobEntity::getFinishedAt, LocalDateTime.now()));
}
public TaskFileJobEntity findById(Long jobId) {
if (jobId == null || jobId <= 0) {
return null;
}
return taskFileJobMapper.selectById(jobId);
}
public void markFailed(TaskFileJobEntity job, String message) {
int retryCount = job.getRetryCount() == null ? 0 : job.getRetryCount() + 1;
taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
.eq(TaskFileJobEntity::getId, job.getId())
.ne(TaskFileJobEntity::getStatus, "SUCCESS")
.set(TaskFileJobEntity::getStatus, "FAILED")
.set(TaskFileJobEntity::getRetryCount, retryCount)
.set(TaskFileJobEntity::getErrorMessage, message == null ? "结果文件生成失败" : message)
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()));
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())
.set(TaskFileJobEntity::getFinishedAt, retryCount >= 5 ? LocalDateTime.now() : null));
}
public TaskFileJobEntity findAssembleJob(Long taskId, String moduleType, Long resultId) {

View File

@@ -62,7 +62,7 @@ public class TaskResultFileJobWorker {
if (!localWorkerEnabled) {
return;
}
List<TaskFileJobEntity> jobs = taskFileJobService.listRunnableJobs(batchSize);
List<TaskFileJobEntity> jobs = taskFileJobService.listRunnableJobsForOwner(batchSize, currentInstanceId());
if (jobs == null || jobs.isEmpty()) {
return;
}
@@ -111,6 +111,12 @@ public class TaskResultFileJobWorker {
private void processInternal(TaskFileJobEntity job) {
long startedAt = System.currentTimeMillis();
try {
TaskFileJobEntity latest = taskFileJobService.findById(job.getId());
if (latest != null && "SUCCESS".equals(latest.getStatus())) {
log.info("[task-file-job] process skipped because job already succeeded jobId={} taskId={} moduleType={} resultId={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId());
return;
}
TaskDistributedLockService.LockHandle lockHandle =
taskDistributedLockService.acquire(job.getModuleType(), job.getTaskId(), TaskDistributedLockService.DEFAULT_WAIT_MILLIS);
if (lockHandle == null) {
@@ -120,9 +126,22 @@ public class TaskResultFileJobWorker {
return;
}
try (lockHandle) {
latest = taskFileJobService.findById(job.getId());
if (latest != null && "SUCCESS".equals(latest.getStatus())) {
log.info("[task-file-job] process skipped after lock because job already succeeded jobId={} taskId={} moduleType={} resultId={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId());
return;
}
boolean completed = dispatch(job);
if (!completed) {
taskFileJobService.touchRunning(job.getId());
if (isOwnerScopedCozeJob(job)) {
taskFileJobService.touchRunning(job.getId());
log.info("[task-file-job] process waiting for async coze result jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt);
return;
}
taskFileJobService.deferRunning(job.getId(), "Waiting for Coze/file assembly to continue");
log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt);
@@ -136,6 +155,12 @@ public class TaskResultFileJobWorker {
System.currentTimeMillis() - startedAt, resultFileUrl);
}
} catch (Exception ex) {
TaskFileJobEntity latest = taskFileJobService.findById(job.getId());
if (latest != null && "SUCCESS".equals(latest.getStatus())) {
log.info("[task-file-job] process failure ignored because job already succeeded jobId={} taskId={} moduleType={} resultId={} msg={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), ex.getMessage());
return;
}
String message = ex.getMessage() == null ? "结果文件生成失败" : ex.getMessage();
log.warn("[task-file-job] process failed jobId={} taskId={} moduleType={} resultId={} msg={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message);

View File

@@ -136,6 +136,9 @@ aiimage:
mq-enabled: ${AIIMAGE_RESULT_FILE_JOB_MQ_ENABLED:true}
topic: ${AIIMAGE_RESULT_FILE_JOB_TOPIC:aiimage-result-file-job}
consumer-group: ${AIIMAGE_RESULT_FILE_JOB_CONSUMER_GROUP:aiimage-result-file-job-consumer}
mq-send-timeout-ms: ${AIIMAGE_RESULT_FILE_JOB_MQ_SEND_TIMEOUT_MS:1000}
mq-failure-threshold: ${AIIMAGE_RESULT_FILE_JOB_MQ_FAILURE_THRESHOLD:3}
mq-circuit-open-ms: ${AIIMAGE_RESULT_FILE_JOB_MQ_CIRCUIT_OPEN_MS:60000}
local-dispatch-enabled: ${AIIMAGE_RESULT_FILE_JOB_LOCAL_DISPATCH_ENABLED:true}
local-dispatch-pool-size: ${AIIMAGE_RESULT_FILE_JOB_LOCAL_DISPATCH_POOL_SIZE:2}
local-dispatch-queue-capacity: ${AIIMAGE_RESULT_FILE_JOB_LOCAL_DISPATCH_QUEUE_CAPACITY:200}
@@ -144,12 +147,14 @@ aiimage:
stuck-scan-delay-ms: ${AIIMAGE_RESULT_FILE_JOB_STUCK_SCAN_DELAY_MS:60000}
stuck-timeout-minutes: ${AIIMAGE_RESULT_FILE_JOB_STUCK_TIMEOUT_MINUTES:30}
batch-size: ${AIIMAGE_RESULT_FILE_JOB_BATCH_SIZE:20}
coze-task:
max-concurrent: ${AIIMAGE_COZE_TASK_MAX_CONCURRENT:8}
appearance-patent:
coze-base-url: ${AIIMAGE_APPEARANCE_PATENT_COZE_BASE_URL:https://api.coze.cn}
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:50}
coze-workflow-id: ${AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID:7639685157562089513}
coze-token: ${AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN:Bearer sat_CztofPRhIKFKeaPZBIRxocfckqhsdCZZ45NPhkf7WOZjbX36vnVSfVV30T6u2rSX}
coze-batch-size: ${AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE:20}
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:30000}
@@ -159,9 +164,9 @@ aiimage:
similar-asin:
coze-base-url: ${AIIMAGE_SIMILAR_ASIN_COZE_BASE_URL:https://api.coze.cn}
coze-workflow-path: ${AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_PATH:/v1/workflow/run}
coze-workflow-id: ${AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_ID:7635328462404583478}
coze-token: ${AIIMAGE_SIMILAR_ASIN_COZE_TOKEN:Bearer sat_gsInckdsN0qqhnZSK7K4Zevw6neJCHmYHyIbm85oGgJIGEmSdkH5OFctSKLsAHvT}
coze-batch-size: ${AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE:50}
coze-workflow-id: ${AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_ID:7639708860686024756}
coze-token: ${AIIMAGE_SIMILAR_ASIN_COZE_TOKEN:Bearer sat_vqmb97BiXbMW3XVyG9htWQKjZ2F55CaW2FRrT9bDdPha7a3hxmXnMNCE1XNHdCVU}
coze-batch-size: ${AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE:20}
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:30000}

View File

@@ -0,0 +1,29 @@
CREATE TABLE IF NOT EXISTS biz_coze_credential (
id BIGINT NOT NULL AUTO_INCREMENT,
module_type VARCHAR(64) NOT NULL,
credential_name VARCHAR(128) NOT NULL,
workflow_id VARCHAR(128) NOT NULL,
token VARCHAR(512) NOT NULL,
enabled TINYINT NOT NULL DEFAULT 1,
max_concurrent INT NOT NULL DEFAULT 0,
sort_order INT NOT NULL DEFAULT 0,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY uk_coze_credential_module_name (module_type, credential_name),
KEY idx_coze_credential_module_enabled (module_type, enabled, sort_order, id)
);
INSERT INTO biz_coze_credential (module_type, credential_name, workflow_id, token, enabled, max_concurrent, sort_order)
VALUES
('APPEARANCE_PATENT', 'appearance-1', '7639685157562089513', 'Bearer sat_CztofPRhIKFKeaPZBIRxocfckqhsdCZZ45NPhkf7WOZjbX36vnVSfVV30T6u2rSX', 1, 0, 10),
('APPEARANCE_PATENT', 'appearance-2', '7639688221823778850', 'Bearer sat_2VsTpfWXYG3EkCo8OQfLwJEWqhduCL4NEw7sSCsABppkW61kkSRPIac9coNlEnHE', 1, 0, 20),
('SIMILAR_ASIN', 'similar-1', '7639708860686024756', 'Bearer sat_vqmb97BiXbMW3XVyG9htWQKjZ2F55CaW2FRrT9bDdPha7a3hxmXnMNCE1XNHdCVU', 1, 0, 10),
('SIMILAR_ASIN', 'similar-2', '7635328462404583478', 'Bearer sat_gsInckdsN0qqhnZSK7K4Zevw6neJCHmYHyIbm85oGgJIGEmSdkH5OFctSKLsAHvT', 1, 0, 20)
ON DUPLICATE KEY UPDATE
workflow_id = VALUES(workflow_id),
token = VALUES(token),
enabled = VALUES(enabled),
max_concurrent = VALUES(max_concurrent),
sort_order = VALUES(sort_order),
updated_at = CURRENT_TIMESTAMP;

View File

@@ -4,7 +4,7 @@
Source Server : AI-线
Source Server Type : MySQL
Source Server Version : 80045 (8.0.45)
Source Host : 8.136.19.173:3306
Source Host : 47.111.163.154:3306
Source Schema : aiimage
Target Server Type : MySQL