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

提交更新

Browse Source
This commit is contained in:
super
2026-05-06 00:01:49 +08:00
parent b16c63225f
commit c3810fdb7b
36 changed files with 40057 additions and 14870 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java
View File

@@ -14,8 +14,8 @@ public class AppearancePatentProperties {
private int cozeBatchSize = 10;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
private int cozePollIntervalMillis = 2000;
private int cozePollTimeoutMillis = 120000;
private int cozePollIntervalMillis = 5000;
private int cozePollTimeoutMillis = 600000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";
}

4
backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java
View File

@@ -14,8 +14,8 @@ public class SimilarAsinProperties {
private int cozeBatchSize = 10;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
private int cozePollIntervalMillis = 2000;
private int cozePollTimeoutMillis = 120000;
private int cozePollIntervalMillis = 5000;
private int cozePollTimeoutMillis = 600000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";
}

16
backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java
View File

@@ -32,7 +32,9 @@ public class TaskOperationLockConfig implements WebMvcConfigurer {
private static final class TaskOperationLockInterceptor implements HandlerInterceptor {
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 String LOCK_ATTRIBUTE = TaskOperationLockInterceptor.class.getName() + ".LOCK";
private final TaskDistributedLockService taskDistributedLockService;
@@ -53,17 +55,25 @@ public class TaskOperationLockConfig implements WebMvcConfigurer {
}
String moduleType = matcher.group(1).trim().toUpperCase(Locale.ROOT).replace('-', '_');
Long taskId = Long.valueOf(matcher.group(2));
long waitMillis = resolveWaitMillis(request.getMethod(), uri);
TaskDistributedLockService.LockHandle lockHandle =
taskDistributedLockService.acquire(moduleType, taskId, TaskDistributedLockService.DEFAULT_WAIT_MILLIS);
taskDistributedLockService.acquire(moduleType, taskId, waitMillis);
if (lockHandle == null) {
log.info("[task-operation-lock] rejected busy task operation method={} uri={} moduleType={} taskId={}",
request.getMethod(), uri, moduleType, taskId);
log.info("[task-operation-lock] rejected busy task operation method={} uri={} moduleType={} taskId={} waitMillis={}",
request.getMethod(), uri, moduleType, taskId, waitMillis);
throw new BusinessException(40902, "TASK_LOCK_BUSY");
}
request.setAttribute(LOCK_ATTRIBUTE, lockHandle);
return true;
}
private long resolveWaitMillis(String method, String uri) {
if ("POST".equals(method) && TASK_RESULT_PATH.matcher(uri == null ? "" : uri).matches()) {
return RESULT_SUBMIT_WAIT_MILLIS;
}
return TaskDistributedLockService.DEFAULT_WAIT_MILLIS;
}
@Override
public void afterCompletion(HttpServletRequest request,
HttpServletResponse response,

71
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java
View File

@@ -30,7 +30,7 @@ public class AppearancePatentCozeClient {
private final AppearancePatentProperties properties;
private final ObjectMapper objectMapper;
public List<AppearancePatentResultRowDto> inspect(List<AppearancePatentResultRowDto> rows, String prompt) {
public List<AppearancePatentResultRowDto> inspect(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
if (rows == null || rows.isEmpty()) {
return List.of();
}
@@ -39,7 +39,7 @@ public class AppearancePatentCozeClient {
return rows.stream().map(this::copy).toList();
}
try {
return inspectWithFallback(rows, prompt);
return inspectWithFallback(rows, prompt, apiKey);
} catch (Exception ex) {
String failureMessage = failureMessage(ex);
log.warn("[appearance-patent] coze batch failed size={} err={}", rows.size(), failureMessage);
@@ -47,8 +47,8 @@ public class AppearancePatentCozeClient {
}
}
public CozeSubmitResponse submitWorkflow(List<AppearancePatentResultRowDto> rows, String prompt) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt));
public CozeSubmitResponse submitWorkflow(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
ensureSuccess(submitRoot);
return new CozeSubmitResponse(
extractExecuteId(submitRoot),
@@ -83,12 +83,12 @@ public class AppearancePatentCozeClient {
return rows.stream().map(this::copy).map(row -> markFailed(row, failureMessage)).toList();
}
private List<AppearancePatentResultRowDto> inspectWithFallback(List<AppearancePatentResultRowDto> rows, String prompt) {
private List<AppearancePatentResultRowDto> inspectWithFallback(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
try {
if (rows.size() == 1) {
return inspectSingleRowWithRetry(rows, prompt);
return inspectSingleRowWithRetry(rows, prompt, apiKey);
}
InspectAttempt attempt = inspectOnce(rows, prompt);
InspectAttempt attempt = inspectOnce(rows, prompt, apiKey);
if (attempt.resolvedCount() < rows.size()) {
throw new PartialCozeResultException(attempt.resolvedCount(), rows.size(), attempt.rawResultCount());
}
@@ -99,17 +99,17 @@ public class AppearancePatentCozeClient {
log.warn("[appearance-patent] coze batch fallback split size={} left={} right={} err={}",
rows.size(), middle, rows.size() - middle, failureMessage(ex));
List<AppearancePatentResultRowDto> merged = new ArrayList<>(rows.size());
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(0, middle), prompt));
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(middle, rows.size()), prompt));
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(0, middle), prompt, apiKey));
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(middle, rows.size()), prompt, apiKey));
return merged;
}
throw propagate(ex);
}
}
private List<AppearancePatentResultRowDto> inspectPartitionWithFailureFallback(List<AppearancePatentResultRowDto> rows, String prompt) {
private List<AppearancePatentResultRowDto> inspectPartitionWithFailureFallback(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
try {
return inspectWithFallback(rows, prompt);
return inspectWithFallback(rows, prompt, apiKey);
} catch (Exception ex) {
String failureMessage = failureMessage(ex);
log.warn("[appearance-patent] coze partition failed size={} err={}", rows.size(), failureMessage);
@@ -117,12 +117,12 @@ public class AppearancePatentCozeClient {
}
}
private List<AppearancePatentResultRowDto> inspectSingleRowWithRetry(List<AppearancePatentResultRowDto> rows, String prompt) throws Exception {
private List<AppearancePatentResultRowDto> inspectSingleRowWithRetry(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) throws Exception {
AppearancePatentResultRowDto row = rows.getFirst();
PartialCozeResultException lastFailure = null;
for (int attemptIndex = 1; attemptIndex <= 3; attemptIndex++) {
try {
InspectAttempt attempt = inspectOnce(rows, prompt);
InspectAttempt attempt = inspectOnce(rows, prompt, apiKey);
if (attempt.resolvedCount() == rows.size()) {
return attempt.mergedRows();
}
@@ -153,8 +153,8 @@ public class AppearancePatentCozeClient {
throw lastFailure == null ? new PartialCozeResultException(0, rows.size(), 0) : lastFailure;
}
private InspectAttempt inspectOnce(List<AppearancePatentResultRowDto> rows, String prompt) throws Exception {
String raw = runWorkflowAsyncAndWait(rows, prompt);
private InspectAttempt inspectOnce(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) throws Exception {
String raw = runWorkflowAsyncAndWait(rows, prompt, apiKey);
List<CozeResult> results = parseResults(raw);
if (rows.size() > 1 && !results.isEmpty() && results.stream().noneMatch(this::hasIdentity)) {
throw new PartialCozeResultException(0, rows.size(), results.size());
@@ -163,8 +163,8 @@ public class AppearancePatentCozeClient {
return new InspectAttempt(raw, merged, resolvedCount(merged), results.size());
}
private String runWorkflowAsyncAndWait(List<AppearancePatentResultRowDto> rows, String prompt) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt));
private String runWorkflowAsyncAndWait(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
ensureSuccess(submitRoot);
String immediateData = extractResultDataText(submitRoot);
@@ -205,15 +205,15 @@ public class AppearancePatentCozeClient {
throw new IllegalStateException("Coze async workflow poll timeout");
}
private String postWorkflow(List<AppearancePatentResultRowDto> rows, String prompt) {
Map<String, Object> parameters = buildParameters(rows, prompt);
private String postWorkflow(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
Map<String, Object> parameters = buildParameters(rows, prompt, apiKey);
Map<String, Object> body = new LinkedHashMap<>();
body.put("workflow_id", properties.getCozeWorkflowId());
body.put("parameters", parameters);
body.put("is_async", Boolean.TRUE);
log.info("[appearance-patent] coze request url={} body={}",
joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()),
writeJson(body));
writeJson(maskCozeRequestBody(body)));
RestClient.RequestBodySpec request = restClient().post()
.uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
@@ -255,7 +255,7 @@ public class AppearancePatentCozeClient {
});
}
private Map<String, Object> buildParameters(List<AppearancePatentResultRowDto> rows, String prompt) {
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();
List<String> asins = rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList();
@@ -268,9 +268,38 @@ public class AppearancePatentCozeClient {
parameters.put("url_list", urls);
parameters.put("items", buildItemObjects(rows, groupKeys, rowIds, asins, countries, titles, urls));
parameters.put("prompt", prompt == null ? "" : prompt);
if (apiKey != null && !apiKey.isBlank()) {
parameters.put("api_key", apiKey.trim());
}
return parameters;
}
@SuppressWarnings("unchecked")
private Map<String, Object> maskCozeRequestBody(Map<String, Object> body) {
Map<String, Object> masked = new LinkedHashMap<>(body);
Object parametersObj = masked.get("parameters");
if (parametersObj instanceof Map<?, ?> parameters) {
Map<String, Object> maskedParameters = new LinkedHashMap<>((Map<String, Object>) parameters);
Object apiKey = maskedParameters.get("api_key");
if (apiKey instanceof String apiKeyText && !apiKeyText.isBlank()) {
maskedParameters.put("api_key", maskSecret(apiKeyText));
}
masked.put("parameters", maskedParameters);
}
return masked;
}
private String maskSecret(String secret) {
String normalized = secret == null ? "" : secret.trim();
if (normalized.isBlank()) {
return "";
}
if (normalized.length() <= 10) {
return "***";
}
return normalized.substring(0, 6) + "***" + normalized.substring(normalized.length() - 4);
}
private List<Map<String, Object>> buildItemObjects(List<AppearancePatentResultRowDto> rows,
List<String> groupKeys,
List<String> rowIds,

7
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParseRequest.java
View File

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
@@ -25,4 +26,10 @@ public class AppearancePatentParseRequest {
@JsonAlias({"aiPrompt", "prompt"})
@Schema(description = "AI 提示词。非必填；为空时前端会使用默认提示词。后端会保存该提示词，并在调用 Coze workflow 时作为 prompt 参数传入。", example = "请排查这些亚马逊商品在英国及欧洲地区是否存在知识产权侵权风险。")
private String aiPrompt;
@JsonProperty("api_key")
@JsonAlias({"apiKey"})
@Schema(description = "传递给 Coze workflow parameters.api_key 的任务级密钥。")
@NotBlank(message = "密钥不能为空")
private String apiKey;
}

3
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedPayloadDto.java
View File

@@ -14,6 +14,9 @@ public class AppearancePatentParsedPayloadDto {
@Schema(description = "AI 提示词")
private String aiPrompt;
@Schema(description = "传递给 Coze workflow parameters.api_key 的任务级密钥")
private String apiKey;
@Schema(description = "本次解析的源文件列表")
private List<AppearancePatentSourceFileDto> sourceFiles = new ArrayList<>();

174
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java
View File

@@ -9,6 +9,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.AppearancePatentProperties;
import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.config.StorageProperties;
import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentCozeClient;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParseRequest;
@@ -131,6 +132,7 @@ public class AppearancePatentTaskService {
private final PlatformTransactionManager transactionManager;
private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService;
private final InstanceMetadata instanceMetadata;
@Autowired
@Qualifier("cozeTaskExecutor")
private TaskExecutor cozeTaskExecutor;
@@ -199,11 +201,11 @@ public class AppearancePatentTaskService {
String aggregateScopeKey = buildAggregateScopeKey(sourceFiles);
String sourceScopeHash = DigestUtil.sha256Hex(aggregateScopeKey);
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), sourceFiles, mergedHeaders, groups, allRows);
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), request.getApiKey(), sourceFiles, mergedHeaders, groups, allRows);
long payloadBuiltAt = System.nanoTime();
String parsedPayloadPointer = storeParsedPayload(task.getId(), sourceScopeHash, parsedPayload);
long payloadStoredAt = System.nanoTime();
task.setResultJson(buildTaskResultJson(request.getAiPrompt(), sourceFiles, parsedPayloadPointer));
task.setResultJson(buildTaskResultJson(request.getAiPrompt(), request.getApiKey(), sourceFiles, parsedPayloadPointer));
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
@@ -267,6 +269,7 @@ public class AppearancePatentTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) {
throw new BusinessException("任务不存在");
}
ensureTaskOwnedByCurrentInstance(task, "activate");
if (STATUS_SUCCESS.equals(task.getStatus()) || STATUS_FAILED.equals(task.getStatus())) {
throw new BusinessException("任务已结束");
}
@@ -429,6 +432,7 @@ public class AppearancePatentTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
}
ensureTaskOwnedByCurrentInstance(task, "submit result");
if (!STATUS_RUNNING.equals(task.getStatus())) {
throw new BusinessException("任务不是运行中状态");
}
@@ -620,6 +624,7 @@ public class AppearancePatentTaskService {
throw new BusinessException("任务不是运行中状态");
}
ensureTaskOwnedByCurrentInstance(task, "submit result");
int chunkIndex = request.getChunkIndex() == null ? 0 : request.getChunkIndex();
int chunkTotal = request.getChunkTotal() == null ? 1 : request.getChunkTotal();
boolean done = Boolean.TRUE.equals(request.getDone());
@@ -772,10 +777,11 @@ public class AppearancePatentTaskService {
return List.of();
}
String prompt = readAiPrompt(task);
String apiKey = readApiKey(task);
int batchSize = Math.max(1, properties.getCozeBatchSize());
List<AppearancePatentResultRowDto> result = new ArrayList<>();
for (int i = 0; i < items.size(); i += batchSize) {
result.addAll(cozeClient.inspect(items.subList(i, Math.min(i + batchSize, items.size())), prompt));
result.addAll(cozeClient.inspect(items.subList(i, Math.min(i + batchSize, items.size())), prompt, apiKey));
if (progressHook != null) {
progressHook.run();
}
@@ -1060,6 +1066,14 @@ public class AppearancePatentTaskService {
}
}
private String readApiKey(FileTaskEntity task) {
try {
return normalize(readParsedPayload(task).getApiKey());
} catch (Exception ignored) {
return "";
}
}
private void finalizeTask(FileTaskEntity task, String error, int rowCount, boolean assembleWorkbook) {
String finalError = error;
FileResultEntity result = null;
@@ -1098,7 +1112,7 @@ public class AppearancePatentTaskService {
}
fileResultMapper.updateById(result);
if (!failed && assembleWorkbook) {
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), "task:" + task.getId());
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), buildTaskOwnerScopeKey(task.getId()));
}
}
taskCacheService.deleteTaskCache(task.getId());
@@ -1112,6 +1126,11 @@ public class AppearancePatentTaskService {
if (job == null || job.getTaskId() == null || job.getResultId() == null) {
throw new BusinessException("result file job arguments are incomplete");
}
if (!isJobOwnedByCurrentInstance(job)) {
log.info("[appearance-patent] skip result file job because owner is another instance jobId={} taskId={} owner={} current={}",
job.getId(), job.getTaskId(), ownerFromScopeKey(job.getScopeKey()), currentInstanceId());
return false;
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(job.getTaskId(), TASK_LOCK_WAIT_MILLIS);
if (lockHandle == null) {
taskFileJobService.requeue(job.getId(), "Task is busy, waiting for appearance patent result merge");
@@ -1159,38 +1178,31 @@ public class AppearancePatentTaskService {
@Scheduled(fixedDelayString = "${aiimage.appearance-patent.coze-poll-delay-ms:5000}")
public void pollPendingCozeJobs() {
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("appearance-patent:coze-poll", Duration.ofMinutes(1));
if (lockHandle == null) {
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"));
if (states == null || states.isEmpty()) {
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"));
if (states == null || states.isEmpty()) {
return;
log.info("[appearance-patent] coze poll picked pending states count={}", states.size());
for (TaskScopeStateEntity state : states) {
if (state == null || state.getId() == null || !isCozeStateOwnedByCurrentInstance(state)) {
continue;
}
log.info("[appearance-patent] coze poll picked pending states count={}", states.size());
for (TaskScopeStateEntity state : states) {
if (state == null || state.getId() == null) {
continue;
}
Long stateId = state.getId();
try {
cozeTaskExecutor.execute(() -> {
log.info("[appearance-patent] coze poll worker entered stateId={} taskId={} executeId={}",
stateId, state.getTaskId(), state.getCozeExecuteId());
pollPendingCozeState(stateId);
});
} catch (Exception ex) {
log.warn("[appearance-patent] coze poll dispatch failed stateId={} taskId={} executeId={} err={}",
stateId, state.getTaskId(), state.getCozeExecuteId(),
firstNonBlank(ex.getMessage(), ex.getClass().getSimpleName()), ex);
}
Long stateId = state.getId();
try {
cozeTaskExecutor.execute(() -> {
log.info("[appearance-patent] coze poll worker entered stateId={} taskId={} executeId={}",
stateId, state.getTaskId(), state.getCozeExecuteId());
pollPendingCozeState(stateId);
});
} catch (Exception ex) {
log.warn("[appearance-patent] coze poll dispatch failed stateId={} taskId={} executeId={} err={}",
stateId, state.getTaskId(), state.getCozeExecuteId(),
firstNonBlank(ex.getMessage(), ex.getClass().getSimpleName()), ex);
}
}
}
@@ -1209,6 +1221,7 @@ public class AppearancePatentTaskService {
return false;
}
String prompt = readAiPrompt(task);
String apiKey = readApiKey(task);
int batchSize = Math.max(1, properties.getCozeBatchSize());
boolean pending = false;
for (TaskChunkEntity chunk : chunks) {
@@ -1225,7 +1238,7 @@ public class AppearancePatentTaskService {
for (int i = 0; i < unresolvedRows.size(); i += batchSize) {
List<AppearancePatentResultRowDto> batchRows =
unresolvedRows.subList(i, Math.min(i + batchSize, unresolvedRows.size()));
pending |= submitCozeBatch(task, result, job, chunk, batchRows, batchIndex, batchTotal, prompt, allRowsByBaseId);
pending |= submitCozeBatch(task, result, job, chunk, batchRows, batchIndex, batchTotal, prompt, apiKey, allRowsByBaseId);
batchIndex++;
}
}
@@ -1240,6 +1253,7 @@ public class AppearancePatentTaskService {
int batchIndex,
int batchTotal,
String prompt,
String apiKey,
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId) {
if (batchRows == null || batchRows.isEmpty()) {
return false;
@@ -1256,7 +1270,7 @@ public class AppearancePatentTaskService {
|| COZE_STATUS_RUNNING.equals(existing.getCozeStatus());
}
try {
AppearancePatentCozeClient.CozeSubmitResponse submit = cozeClient.submitWorkflow(batchRows, prompt);
AppearancePatentCozeClient.CozeSubmitResponse submit = cozeClient.submitWorkflow(batchRows, prompt, apiKey);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<AppearancePatentResultRowDto> cozeRows = cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
mergeCozeRowsIntoChunk(task, chunk.getScopeHash(), chunk.getChunkIndex(), cozeRows, allRowsByBaseId);
@@ -1305,7 +1319,8 @@ public class AppearancePatentTaskService {
chunk.getScopeHash(),
chunk.getChunkIndex(),
batchIndex,
batchTotal
batchTotal,
currentInstanceId()
);
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
String storedBatchPayload = storeSharedCozeBatchPayload(task.getId(), batchScopeHash, batchPayload);
@@ -1341,6 +1356,11 @@ public class AppearancePatentTaskService {
if (lockState != null) {
taskIdForLock = lockState.getTaskId();
}
if (!isCozeStateOwnedByCurrentInstance(lockState)) {
log.info("[appearance-patent] coze poll skipped because owner is another instance taskId={} stateId={} owner={} current={}",
taskIdForLock, stateId, ownerFromCozeState(lockState), currentInstanceId());
return;
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskIdForLock, 0L);
if (lockHandle == null) {
if (taskIdForLock != null) {
@@ -1375,6 +1395,11 @@ public class AppearancePatentTaskService {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing");
return;
}
if (!isOwnerCurrent(context.ownerInstanceId())) {
log.info("[appearance-patent] coze poll skipped after context refresh because owner is another instance taskId={} stateId={} owner={} current={}",
state.getTaskId(), state.getId(), context.ownerInstanceId(), currentInstanceId());
return;
}
taskFileJobService.touchRunning(context.jobId());
log.info("[appearance-patent] coze poll start taskId={} stateId={} executeId={} jobId={} chunk={} batch={}/{}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(),
@@ -1664,6 +1689,69 @@ public class AppearancePatentTaskService {
+ ":batch:" + batchIndex;
}
private String buildTaskOwnerScopeKey(Long taskId) {
return "task:" + taskId + ":owner:" + currentInstanceId();
}
private String currentInstanceId() {
String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId();
return firstNonBlank(instanceId, "unknown-instance");
}
private boolean isJobOwnedByCurrentInstance(TaskFileJobEntity job) {
return job == null || isOwnerCurrent(ownerFromScopeKey(job.getScopeKey()));
}
private void ensureTaskOwnedByCurrentInstance(FileTaskEntity task, String operation) {
String owner = ownerFromTask(task);
if (isOwnerCurrent(owner)) {
return;
}
log.warn("[appearance-patent] reject task operation because owner is another instance taskId={} operation={} owner={} current={}",
task == null ? null : task.getId(), operation, owner, currentInstanceId());
throw new BusinessException(40903, "该任务已绑定到另一台服务实例处理，请通过原实例继续处理");
}
private boolean isCozeStateOwnedByCurrentInstance(TaskScopeStateEntity state) {
return state == null || isOwnerCurrent(ownerFromCozeState(state));
}
private boolean isOwnerCurrent(String owner) {
return owner == null || owner.isBlank() || Objects.equals(owner, currentInstanceId());
}
private String ownerFromCozeState(TaskScopeStateEntity state) {
CozeBatchContext context = readCozeBatchContext(state);
return context == null ? null : context.ownerInstanceId();
}
private String ownerFromTask(FileTaskEntity task) {
if (task == null || task.getResultJson() == null || task.getResultJson().isBlank()) {
return null;
}
try {
String owner = objectMapper.readTree(task.getResultJson()).path("ownerInstanceId").asText("");
return owner.isBlank() ? null : owner;
} catch (Exception ex) {
log.warn("[appearance-patent] read task owner failed taskId={} err={}",
task.getId(), ex.getMessage());
return null;
}
}
private String ownerFromScopeKey(String scopeKey) {
if (scopeKey == null || scopeKey.isBlank()) {
return null;
}
String marker = ":owner:";
int index = scopeKey.lastIndexOf(marker);
if (index < 0) {
return null;
}
String owner = scopeKey.substring(index + marker.length()).trim();
return owner.isBlank() ? null : owner;
}
private String abbreviate(String value, int maxLength) {
if (value == null) {
return "";
@@ -2340,7 +2428,9 @@ public class AppearancePatentTaskService {
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) {
percent = Math.max(percent, Math.min(35, 8 + (int) (elapsedSeconds / 6)));
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)));
}
@@ -2379,9 +2469,10 @@ public class AppearancePatentTaskService {
return val == null ? "" : val.replace(String.valueOf((char) 0xFEFF), "").replace((char) 0x3000, ' ').trim().replaceAll("\\s+", " ");
}
private String buildParsedPayloadJson(String aiPrompt, List<AppearancePatentSourceFileDto> sourceFiles, List<String> headers, List<AppearancePatentParsedGroupVo> groups, List<AppearancePatentParsedRowVo> allRows) {
private String buildParsedPayloadJson(String aiPrompt, String apiKey, List<AppearancePatentSourceFileDto> sourceFiles, List<String> headers, List<AppearancePatentParsedGroupVo> groups, List<AppearancePatentParsedRowVo> allRows) {
AppearancePatentParsedPayloadDto payload = new AppearancePatentParsedPayloadDto();
payload.setAiPrompt(normalize(aiPrompt));
payload.setApiKey(normalize(apiKey));
payload.setSourceFiles(sourceFiles == null ? List.of() : sourceFiles);
payload.setHeaders(headers == null ? List.of() : headers);
payload.setItems(List.of());
@@ -2390,14 +2481,16 @@ public class AppearancePatentTaskService {
return writeJson(payload, "保存解析结果失败");
}
private String buildTaskResultJson(String aiPrompt, List<AppearancePatentSourceFileDto> sourceFiles, String parsedPayloadPointer) {
private String buildTaskResultJson(String aiPrompt, String apiKey, List<AppearancePatentSourceFileDto> sourceFiles, String parsedPayloadPointer) {
Map<String, Object> payload = new LinkedHashMap<>();
payload.put("aiPrompt", normalize(aiPrompt));
payload.put("apiKey", normalize(apiKey));
payload.put("sourceFileKeys", sourceFiles == null ? List.of() : sourceFiles.stream()
.map(AppearancePatentSourceFileDto::getFileKey)
.filter(Objects::nonNull)
.toList());
payload.put("parsedPayloadRef", parsedPayloadPointer);
payload.put("ownerInstanceId", currentInstanceId());
return writeJson(payload, "保存任务结果索引失败");
}
@@ -2661,7 +2754,8 @@ public class AppearancePatentTaskService {
String chunkScopeHash,
Integer chunkIndex,
Integer batchIndex,
Integer batchTotal) {
Integer batchTotal,
String ownerInstanceId) {
}
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<AppearancePatentParsedRowVo> allRows) {

123
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java
View File

@@ -30,7 +30,7 @@ public class SimilarAsinCozeClient {
private final SimilarAsinProperties properties;
private final ObjectMapper objectMapper;
public List<SimilarAsinResultRowDto> inspect(List<SimilarAsinResultRowDto> rows, String prompt) {
public List<SimilarAsinResultRowDto> inspect(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
if (rows == null || rows.isEmpty()) {
return List.of();
}
@@ -39,7 +39,7 @@ public class SimilarAsinCozeClient {
return rows.stream().map(this::copy).toList();
}
try {
return inspectWithFallback(rows, prompt);
return inspectWithFallback(rows, prompt, apiKey);
} catch (Exception ex) {
String failureMessage = failureMessage(ex);
log.warn("[similar-asin] coze batch failed size={} err={}", rows.size(), failureMessage);
@@ -47,8 +47,8 @@ public class SimilarAsinCozeClient {
}
}
public CozeSubmitResponse submitWorkflow(List<SimilarAsinResultRowDto> rows, String prompt) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt));
public CozeSubmitResponse submitWorkflow(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
ensureSuccess(submitRoot);
return new CozeSubmitResponse(
extractExecuteId(submitRoot),
@@ -83,12 +83,12 @@ public class SimilarAsinCozeClient {
return rows.stream().map(this::copy).map(row -> markFailed(row, failureMessage)).toList();
}
private List<SimilarAsinResultRowDto> inspectWithFallback(List<SimilarAsinResultRowDto> rows, String prompt) {
private List<SimilarAsinResultRowDto> inspectWithFallback(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
try {
if (rows.size() == 1) {
return inspectSingleRowWithRetry(rows, prompt);
return inspectSingleRowWithRetry(rows, prompt, apiKey);
}
InspectAttempt attempt = inspectOnce(rows, prompt);
InspectAttempt attempt = inspectOnce(rows, prompt, apiKey);
if (attempt.resolvedCount() < rows.size()) {
throw new PartialCozeResultException(attempt.resolvedCount(), rows.size(), attempt.rawResultCount());
}
@@ -99,17 +99,17 @@ public class SimilarAsinCozeClient {
log.warn("[similar-asin] coze batch fallback split size={} left={} right={} err={}",
rows.size(), middle, rows.size() - middle, failureMessage(ex));
List<SimilarAsinResultRowDto> merged = new ArrayList<>(rows.size());
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(0, middle), prompt));
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(middle, rows.size()), prompt));
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(0, middle), prompt, apiKey));
merged.addAll(inspectPartitionWithFailureFallback(rows.subList(middle, rows.size()), prompt, apiKey));
return merged;
}
throw propagate(ex);
}
}
private List<SimilarAsinResultRowDto> inspectPartitionWithFailureFallback(List<SimilarAsinResultRowDto> rows, String prompt) {
private List<SimilarAsinResultRowDto> inspectPartitionWithFailureFallback(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
try {
return inspectWithFallback(rows, prompt);
return inspectWithFallback(rows, prompt, apiKey);
} catch (Exception ex) {
String failureMessage = failureMessage(ex);
log.warn("[similar-asin] coze partition failed size={} err={}", rows.size(), failureMessage);
@@ -117,12 +117,12 @@ public class SimilarAsinCozeClient {
}
}
private List<SimilarAsinResultRowDto> inspectSingleRowWithRetry(List<SimilarAsinResultRowDto> rows, String prompt) throws Exception {
private List<SimilarAsinResultRowDto> inspectSingleRowWithRetry(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) throws Exception {
SimilarAsinResultRowDto row = rows.getFirst();
PartialCozeResultException lastFailure = null;
for (int attemptIndex = 1; attemptIndex <= 3; attemptIndex++) {
try {
InspectAttempt attempt = inspectOnce(rows, prompt);
InspectAttempt attempt = inspectOnce(rows, prompt, apiKey);
if (attempt.resolvedCount() == rows.size()) {
return attempt.mergedRows();
}
@@ -153,8 +153,8 @@ public class SimilarAsinCozeClient {
throw lastFailure == null ? new PartialCozeResultException(0, rows.size(), 0) : lastFailure;
}
private InspectAttempt inspectOnce(List<SimilarAsinResultRowDto> rows, String prompt) throws Exception {
String raw = runWorkflowAsyncAndWait(rows, prompt);
private InspectAttempt inspectOnce(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) throws Exception {
String raw = runWorkflowAsyncAndWait(rows, prompt, apiKey);
List<CozeResult> results = parseResults(raw);
if (rows.size() > 1 && !results.isEmpty() && results.stream().noneMatch(this::hasIdentity)) {
throw new PartialCozeResultException(0, rows.size(), results.size());
@@ -163,8 +163,8 @@ public class SimilarAsinCozeClient {
return new InspectAttempt(raw, merged, resolvedCount(merged), results.size());
}
private String runWorkflowAsyncAndWait(List<SimilarAsinResultRowDto> rows, String prompt) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt));
private String runWorkflowAsyncAndWait(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) throws Exception {
JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey));
ensureSuccess(submitRoot);
String immediateData = extractResultDataText(submitRoot);
@@ -205,15 +205,15 @@ public class SimilarAsinCozeClient {
throw new IllegalStateException("Coze async workflow poll timeout");
}
private String postWorkflow(List<SimilarAsinResultRowDto> rows, String prompt) {
Map<String, Object> parameters = buildParameters(rows, prompt);
private String postWorkflow(List<SimilarAsinResultRowDto> rows, String prompt, String apiKey) {
Map<String, Object> parameters = buildParameters(rows, prompt, apiKey);
Map<String, Object> body = new LinkedHashMap<>();
body.put("workflow_id", properties.getCozeWorkflowId());
body.put("parameters", parameters);
body.put("is_async", Boolean.TRUE);
log.info("[similar-asin] coze request url={} body={}",
joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()),
writeJson(body));
writeJson(maskCozeRequestBody(body)));
RestClient.RequestBodySpec request = restClient().post()
.uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
@@ -255,11 +255,12 @@ public class SimilarAsinCozeClient {
});
}
private Map<String, Object> buildParameters(List<SimilarAsinResultRowDto> rows, String prompt) {
private Map<String, Object> buildParameters(List<SimilarAsinResultRowDto> 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();
List<String> asins = rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList();
List<String> countries = rows.stream().map(row -> nonBlank(row.getCountry(), "")).toList();
List<String> prices = rows.stream().map(row -> nonBlank(row.getPrice(), "")).toList();
List<String> titles = rows.stream().map(row -> nonBlank(row.getTitle(), row.getAsin())).toList();
List<String> urls = rows.stream().map(row -> nonBlank(row.getUrl(), "")).toList();
List<List<String>> urlLists = rows.stream().map(SimilarAsinResultRowDto::getUrls).toList();
@@ -268,17 +269,48 @@ public class SimilarAsinCozeClient {
parameters.put("title_list", titles);
parameters.put("url_list", urls);
parameters.put("url_lists", urlLists);
parameters.put("items", buildItemObjects(rows, groupKeys, rowIds, asins, countries, titles, urlLists));
parameters.put("items", buildItemObjects(rows, groupKeys, rowIds, asins, countries, prices, titles, urls, urlLists));
parameters.put("prompt", prompt == null ? "" : prompt);
if (apiKey != null && !apiKey.isBlank()) {
parameters.put("api_key", apiKey.trim());
}
return parameters;
}
@SuppressWarnings("unchecked")
private Map<String, Object> maskCozeRequestBody(Map<String, Object> body) {
Map<String, Object> masked = new LinkedHashMap<>(body);
Object parametersObj = masked.get("parameters");
if (parametersObj instanceof Map<?, ?> parameters) {
Map<String, Object> maskedParameters = new LinkedHashMap<>((Map<String, Object>) parameters);
Object apiKey = maskedParameters.get("api_key");
if (apiKey instanceof String apiKeyText && !apiKeyText.isBlank()) {
maskedParameters.put("api_key", maskSecret(apiKeyText));
}
masked.put("parameters", maskedParameters);
}
return masked;
}
private String maskSecret(String secret) {
String normalized = secret == null ? "" : secret.trim();
if (normalized.isBlank()) {
return "";
}
if (normalized.length() <= 10) {
return "***";
}
return normalized.substring(0, 6) + "***" + normalized.substring(normalized.length() - 4);
}
private List<Map<String, Object>> buildItemObjects(List<SimilarAsinResultRowDto> rows,
List<String> groupKeys,
List<String> rowIds,
List<String> asins,
List<String> countries,
List<String> prices,
List<String> titles,
List<String> urls,
List<List<String>> urlLists) {
List<Map<String, Object>> items = new ArrayList<>(rows.size());
for (int i = 0; i < rows.size(); i++) {
@@ -287,9 +319,11 @@ public class SimilarAsinCozeClient {
item.put("row_id", rowIds.get(i));
item.put("asin", asins.get(i));
item.put("country", countries.get(i));
item.put("price", prices.get(i));
item.put("title", titles.get(i));
item.put("url", urlLists.get(i));
item.put("url", urls.get(i));
item.put("urls", urlLists.get(i));
item.put("target_urls", urlLists.get(i));
items.add(item);
}
return items;
@@ -319,7 +353,16 @@ public class SimilarAsinCozeClient {
text(firstNonNull(
firstNonNull(node.get("country"), node.get("site")),
firstNonNull(itemNode.get("country"), itemNode.get("site")))),
text(firstNonNull(
firstNonNull(node.get("price"), node.get("价格")),
firstNonNull(itemNode.get("price"), itemNode.get("价格")))),
text(firstNonNull(node.get("title"), firstNonNull(itemNode.get("title"), itemNode.get("title_risk")))),
text(firstNonNull(
firstNonNull(node.get("is_stock"), firstNonNull(node.get("isStock"), firstNonNull(node.get("stock"), node.get("是否有货")))),
firstNonNull(itemNode.get("is_stock"), firstNonNull(itemNode.get("isStock"), firstNonNull(itemNode.get("stock"), itemNode.get("是否有货")))))),
text(firstNonNull(
firstNonNull(node.get("similarity"), firstNonNull(node.get("similarity_rate"), node.get("相似度"))),
firstNonNull(itemNode.get("similarity"), firstNonNull(itemNode.get("similarity_rate"), itemNode.get("相似度"))))),
text(firstNonNull(node.get("appearance"),
firstNonNull(node.get("appearance_risk"),
firstNonNull(itemNode.get("appearance"), itemNode.get("appearance_risk"))))),
@@ -396,6 +439,9 @@ public class SimilarAsinCozeClient {
if (result == null) {
result = resultByCompositeKey.get(rowKey(row.getId(), row.getAsin(), row.getCountry()));
}
if (result == null && i < results.size() && isIndexedResultForRow(results.get(i), row)) {
result = results.get(i);
}
if (result == null) {
result = resultByAsinCountry.get(asinCountryKey(row.getAsin(), row.getCountry()));
}
@@ -431,11 +477,25 @@ public class SimilarAsinCozeClient {
|| !normalize(result.asin()).isBlank();
}
private boolean isIndexedResultForRow(CozeResult result, SimilarAsinResultRowDto row) {
if (result == null || row == null) {
return false;
}
if (!normalize(result.groupKey()).isBlank() || !normalize(result.rowId()).isBlank()) {
return false;
}
String resultAsin = normalize(result.asin()).toUpperCase(Locale.ROOT);
return resultAsin.isBlank() || resultAsin.equals(normalize(row.getAsin()).toUpperCase(Locale.ROOT));
}
private void applyResult(SimilarAsinResultRowDto row, CozeResult result) {
if (row == null || result == null) {
return;
}
row.setTitleRisk(result.title());
row.setPrice(nonBlank(result.price(), row.getPrice()));
row.setIsStock(result.isStock());
row.setSimilarity(result.similarity());
row.setAppearanceRisk(result.appearance());
row.setPatentRisk(result.patent());
row.setConclusion(result.result());
@@ -460,6 +520,7 @@ public class SimilarAsinCozeClient {
row.setId(source.getId());
row.setAsin(source.getAsin());
row.setCountry(source.getCountry());
row.setPrice(source.getPrice());
row.setUrls(source.getUrls());
row.setTitle(source.getTitle());
row.setError(source.getError());
@@ -468,6 +529,8 @@ public class SimilarAsinCozeClient {
row.setAppearanceRisk(source.getAppearanceRisk());
row.setPatentRisk(source.getPatentRisk());
row.setConclusion(source.getConclusion());
row.setIsStock(source.getIsStock());
row.setSimilarity(source.getSimilarity());
row.setTitleReason(source.getTitleReason());
row.setAppearanceReason(source.getAppearanceReason());
row.setPatentReason(source.getPatentReason());
@@ -529,7 +592,9 @@ public class SimilarAsinCozeClient {
return !normalize(row.getTitleRisk()).isBlank()
|| !normalize(row.getAppearanceRisk()).isBlank()
|| !normalize(row.getPatentRisk()).isBlank()
|| !normalize(row.getConclusion()).isBlank();
|| !normalize(row.getConclusion()).isBlank()
|| !normalize(row.getIsStock()).isBlank()
|| !normalize(row.getSimilarity()).isBlank();
}
private void ensureSuccess(JsonNode root) {
@@ -656,6 +721,13 @@ public class SimilarAsinCozeClient {
|| item.has("patent_risk")
|| item.has("result")
|| item.has("conclusion")
|| item.has("is_stock")
|| item.has("isStock")
|| item.has("stock")
|| item.has("是否有货")
|| item.has("similarity")
|| item.has("similarity_rate")
|| item.has("相似度")
|| item.has("title_reason")
|| item.has("titleReason")
|| item.has("appearance_reason")
@@ -909,7 +981,10 @@ public class SimilarAsinCozeClient {
String rowId,
String asin,
String country,
String price,
String title,
String isStock,
String similarity,
String appearance,
String patent,
String result,

7
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParseRequest.java
View File

@@ -3,6 +3,7 @@ package com.nanri.aiimage.modules.similarasin.model.dto;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
@@ -25,4 +26,10 @@ public class SimilarAsinParseRequest {
@JsonAlias({"aiPrompt", "prompt"})
@Schema(description = "AI 提示词。非必填；为空时前端会使用默认提示词。后端会保存该提示词，并在调用 Coze workflow 时作为 prompt 参数传入。", example = "请排查这些亚马逊商品在英国及欧洲地区是否存在知识产权侵权风险。")
private String aiPrompt;
@JsonProperty("api_key")
@JsonAlias({"apiKey"})
@Schema(description = "传递给 Coze workflow parameters.api_key 的任务级密钥。")
@NotBlank(message = "密钥不能为空")
private String apiKey;
}

3
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParsedPayloadDto.java
View File

@@ -14,6 +14,9 @@ public class SimilarAsinParsedPayloadDto {
@Schema(description = "AI 提示词")
private String aiPrompt;
@Schema(description = "传递给 Coze workflow parameters.api_key 的任务级密钥")
private String apiKey;
@Schema(description = "本次解析的源文件列表")
private List<SimilarAsinSourceFileDto> sourceFiles = new ArrayList<>();

21
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java
View File

@@ -39,6 +39,10 @@ public class SimilarAsinResultRowDto {
@Schema(description = "站点或国家。来自 Excel 的国家列，例如英国、德国、法国。", example = "英国")
private String country;
@JsonAlias({"price", "价格"})
@Schema(description = "商品价格。来自 Excel 解析或 Python 回传。", example = "12.29")
private String price;
@Schema(description = "兼容旧链路的首个图片 URL。新链路允许 url 传字符串或数组；数组会同步写入 urls。", example = "https://webstatic.aiproxy.vip/output/20260425/103322/demo.jpg")
private String url;
@@ -49,23 +53,31 @@ public class SimilarAsinResultRowDto {
@JsonAlias({"productTitle", "product_title", "itemTitle", "item_title", "商品标题", "商品名称", "标题"})
private String title;
@JsonAlias({"is_stock", "isStock", "stock", "是否有货"})
@Schema(description = "Coze 返回的是否有货结果。", example = "有货", accessMode = Schema.AccessMode.READ_ONLY)
private String isStock;
@JsonAlias({"similarity", "similarity_rate", "similarityRate", "相似度"})
@Schema(description = "Coze 返回的相似度结果。", example = "80%", accessMode = Schema.AccessMode.READ_ONLY)
private String similarity;
@Schema(description = "单行错误信息。通常用于记录 Python 单行处理异常；Coze 失败时后端会尽量保留原始行，不强行写入风险结果。", example = "图片地址为空")
private String error;
@Schema(description = "单行完成标记。当前主要使用请求体顶层 done 控制任务收尾，该字段仅作兼容。", example = "true")
private Boolean done;
@Schema(description = "Java 调用 Coze 后生成的标题维度检测结果，对应最终 xlsx 的“标题维度（商标）”列。Python 回传请求中不要传该字段；即使传入，后端也会以 Java/Coze 处理结果为准。", example = "标题未发现明显商标侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
@Schema(description = "兼容旧版 Coze 返回中的标题维度结果字段；当前相似 ASIN 结果文件不再输出该列。", example = "标题未发现明显商标侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
private String titleRisk;
@Schema(description = "Java 调用 Coze 后生成的外观维度检测结果，对应最终 xlsx 的“外观维度（外观设计专利）”列。Python 回传请求中不要传该字段。", example = "未发现明显外观设计专利侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
@Schema(description = "兼容旧版 Coze 返回中的外观维度结果字段；当前相似 ASIN 结果文件不再输出该列。", example = "未发现明显外观设计专利侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
private String appearanceRisk;
@JsonAlias({"patent ", "patent"})
@Schema(description = "Java 调用 Coze 后生成的专利维度检测结果，对应最终 xlsx 的“专利维度（发明/实用新型专利）”列。兼容 Coze 返回字段 patent 和 patent 后带空格的情况；Python 回传请求中不要传该字段。", example = "未发现明显发明或实用新型专利侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
@Schema(description = "兼容旧版 Coze 返回中的专利维度结果字段；当前相似 ASIN 结果文件不再输出该列。", example = "未发现明显发明或实用新型专利侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
private String patentRisk;
@Schema(description = "Java 调用 Coze 后生成的最终结论，对应最终 xlsx 的“结论”列。Python 回传请求中不要传该字段。", example = "未发现明显侵权风险", accessMode = Schema.AccessMode.READ_ONLY)
@Schema(description = "兼容旧版 Coze 返回中的结论字段；当前相似 ASIN 结果文件不再输出该列。", example = "未发现明显侵权风险", accessMode = Schema.AccessMode.READ_ONLY)
private String conclusion;
@JsonAlias({"title_reason", "titleReason"})
@@ -118,6 +130,7 @@ public class SimilarAsinResultRowDto {
"mainImages", "main_images", "mainImageUrls", "main_image_urls",
"productImages", "product_images", "productImageUrls", "product_image_urls",
"images", "imgs", "pics", "pictures", "links", "imageLinks", "image_links",
"target_urls", "targetUrls", "图片链接数组",
"图片链接", "商品图片", "商品主图", "主图", "主图链接"
})
public void setUrls(Object urls) {

3
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinParsedRowVo.java
View File

@@ -36,6 +36,9 @@ public class SimilarAsinParsedRowVo {
@Schema(description = "国家或站点。", example = "英国")
private String country;
@Schema(description = "商品价格。", example = "12.29")
private String price;
@Schema(description = "商品图片 URL 或商品 URL，供 Coze 检测使用。", example = "https://webstatic.aiproxy.vip/output/demo.jpg")
private String url;

307
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
View File

@@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
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.similarasin.client.SimilarAsinCozeClient;
@@ -99,15 +100,15 @@ public class SimilarAsinTaskService {
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
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 List<String> RESULT_HEADERS = List.of(
"id",
"asin",
"国家",
"价格",
"标题维度（商标）",
"外观维度（外观设计专利）",
"专利维度（发明/实用新型专利）",
"结论"
"是否有货",
"相似度"
);
private final LocalFileStorageService localFileStorageService;
@@ -127,6 +128,7 @@ public class SimilarAsinTaskService {
private final TransientPayloadStorageService transientPayloadStorageService;
private final PlatformTransactionManager transactionManager;
private final DistributedJobLockService distributedJobLockService;
private final InstanceMetadata instanceMetadata;
@Autowired
@Qualifier("cozeTaskExecutor")
private TaskExecutor cozeTaskExecutor;
@@ -195,11 +197,11 @@ public class SimilarAsinTaskService {
String aggregateScopeKey = buildAggregateScopeKey(sourceFiles);
String sourceScopeHash = DigestUtil.sha256Hex(aggregateScopeKey);
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), sourceFiles, mergedHeaders, groups, allRows);
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), request.getApiKey(), sourceFiles, mergedHeaders, groups, allRows);
long payloadBuiltAt = System.nanoTime();
String parsedPayloadPointer = storeParsedPayload(task.getId(), sourceScopeHash, parsedPayload);
long payloadStoredAt = System.nanoTime();
task.setResultJson(buildTaskResultJson(request.getAiPrompt(), sourceFiles, parsedPayloadPointer));
task.setResultJson(buildTaskResultJson(request.getAiPrompt(), request.getApiKey(), sourceFiles, parsedPayloadPointer));
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
@@ -386,6 +388,16 @@ public class SimilarAsinTaskService {
}
public void submitResult(Long taskId, SimilarAsinSubmitResultRequest request) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, TASK_LOCK_WAIT_MILLIS);
if (lockHandle == null) {
throw new BusinessException(40902, "Task is processing the previous result chunk, please retry later");
}
try (lockHandle) {
submitResultLocked(taskId, request);
}
}
private void submitResultLocked(Long taskId, SimilarAsinSubmitResultRequest request) {
if (transactionManager != null) {
SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(taskId, request));
inNewTransaction(() -> {
@@ -402,6 +414,7 @@ public class SimilarAsinTaskService {
throw new BusinessException("任务不是运行中状态");
}
ensureTaskOwnedByCurrentInstance(task, "submit result");
int chunkIndex = request.getChunkIndex() == null ? 0 : request.getChunkIndex();
int chunkTotal = request.getChunkTotal() == null ? 1 : request.getChunkTotal();
String scopeKey = firstNonBlank(request.getSubmissionId(), "task:" + taskId);
@@ -530,6 +543,9 @@ public class SimilarAsinTaskService {
.lt(FileTaskEntity::getUpdatedAt, threshold)
.last("limit 50"));
for (FileTaskEntity task : tasks) {
if (!isOwnerCurrent(ownerFromTask(task))) {
continue;
}
if (isJavaSideProcessing(task.getId())) {
touchJavaSideTaskActivity(task.getId());
continue;
@@ -559,6 +575,9 @@ public class SimilarAsinTaskService {
.lt(FileTaskEntity::getUpdatedAt, threshold)
.last("limit 50"));
for (FileTaskEntity task : tasks) {
if (!isOwnerCurrent(ownerFromTask(task))) {
continue;
}
if (isJavaSideProcessing(task.getId())) {
touchJavaSideTaskActivity(task.getId());
continue;
@@ -586,6 +605,7 @@ public class SimilarAsinTaskService {
throw new BusinessException("任务不是运行中状态");
}
ensureTaskOwnedByCurrentInstance(task, "submit result");
int chunkIndex = request.getChunkIndex() == null ? 0 : request.getChunkIndex();
int chunkTotal = request.getChunkTotal() == null ? 1 : request.getChunkTotal();
boolean done = Boolean.TRUE.equals(request.getDone());
@@ -738,10 +758,11 @@ public class SimilarAsinTaskService {
return List.of();
}
String prompt = readAiPrompt(task);
String apiKey = readApiKey(task);
int batchSize = Math.max(1, properties.getCozeBatchSize());
List<SimilarAsinResultRowDto> result = new ArrayList<>();
for (int i = 0; i < items.size(); i += batchSize) {
result.addAll(cozeClient.inspect(items.subList(i, Math.min(i + batchSize, items.size())), prompt));
result.addAll(cozeClient.inspect(items.subList(i, Math.min(i + batchSize, items.size())), prompt, apiKey));
if (progressHook != null) {
progressHook.run();
}
@@ -818,17 +839,9 @@ public class SimilarAsinTaskService {
List<SimilarAsinResultRowDto> result = new ArrayList<>();
Set<String> emittedKeys = new LinkedHashSet<>();
for (SimilarAsinResultRowDto representative : representatives) {
List<SimilarAsinParsedRowVo> siblings = resolveSiblingRows(representative, allRowsByBaseId);
if (siblings == null || siblings.isEmpty()) {
String key = firstNonBlank(normalize(representative.getRowToken()), rowKey(representative));
if (emittedKeys.add(key)) {
result.add(representative);
continue;
}
for (SimilarAsinParsedRowVo sibling : siblings) {
String key = rowKey(sibling);
if (!emittedKeys.add(key)) {
continue;
}
result.add(copyResultToSibling(representative, sibling));
}
}
return result;
@@ -877,32 +890,13 @@ public class SimilarAsinTaskService {
}
private List<SimilarAsinResultRowDto> pickGroupRepresentativesForCoze(java.util.Collection<SimilarAsinResultRowDto> rows) {
Map<String, List<SimilarAsinResultRowDto>> groupedRows = new LinkedHashMap<>();
if (rows == null) {
return List.of();
}
for (SimilarAsinResultRowDto row : rows) {
if (row == null) {
continue;
}
String key = firstNonBlank(normalize(row.getGroupKey()), rowKey(row));
groupedRows.computeIfAbsent(key, ignored -> new ArrayList<>()).add(row);
}
List<SimilarAsinResultRowDto> representatives = new ArrayList<>();
for (List<SimilarAsinResultRowDto> siblings : groupedRows.values()) {
boolean alreadyResolved = siblings.stream().anyMatch(this::hasResolvedCozeFields);
if (alreadyResolved) {
continue;
}
SimilarAsinResultRowDto candidate = null;
for (SimilarAsinResultRowDto sibling : siblings) {
if (shouldInspectRow(sibling)) {
candidate = sibling;
break;
}
}
if (candidate != null) {
representatives.add(candidate);
for (SimilarAsinResultRowDto row : rows) {
if (row != null && !hasResolvedCozeFields(row) && shouldInspectRow(row)) {
representatives.add(row);
}
}
return representatives;
@@ -1004,6 +998,7 @@ public class SimilarAsinTaskService {
row.setId(sibling.getDisplayId());
row.setAsin(sibling.getAsin());
row.setCountry(sibling.getCountry());
row.setPrice(firstNonBlank(representative.getPrice(), sibling.getPrice()));
if (representative.hasImageUrl()) {
row.setUrls(representative.getUrls());
} else {
@@ -1015,6 +1010,8 @@ public class SimilarAsinTaskService {
row.setAppearanceRisk(representative.getAppearanceRisk());
row.setPatentRisk(representative.getPatentRisk());
row.setConclusion(representative.getConclusion());
row.setIsStock(representative.getIsStock());
row.setSimilarity(representative.getSimilarity());
row.setTitleReason(representative.getTitleReason());
row.setAppearanceReason(representative.getAppearanceReason());
row.setPatentReason(representative.getPatentReason());
@@ -1030,6 +1027,14 @@ public class SimilarAsinTaskService {
}
}
private String readApiKey(FileTaskEntity task) {
try {
return normalize(readParsedPayload(task).getApiKey());
} catch (Exception ignored) {
return "";
}
}
private void finalizeTask(FileTaskEntity task, String error, int rowCount, boolean assembleWorkbook) {
String finalError = error;
FileResultEntity result = null;
@@ -1068,7 +1073,7 @@ public class SimilarAsinTaskService {
}
fileResultMapper.updateById(result);
if (!failed && assembleWorkbook) {
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), "task:" + task.getId());
taskFileJobService.enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), buildTaskOwnerScopeKey(task));
}
}
taskCacheService.deleteTaskCache(task.getId());
@@ -1082,6 +1087,11 @@ public class SimilarAsinTaskService {
if (job == null || job.getTaskId() == null || job.getResultId() == null) {
throw new BusinessException("result file job arguments are incomplete");
}
if (!isJobOwnedByCurrentInstance(job)) {
log.info("[similar-asin] skip result file job because owner is another instance jobId={} taskId={} owner={} current={}",
job.getId(), job.getTaskId(), ownerFromScopeKey(job.getScopeKey()), currentInstanceId());
return false;
}
FileTaskEntity task = fileTaskMapper.selectById(job.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("task not found");
@@ -1128,7 +1138,7 @@ public class SimilarAsinTaskService {
}
log.info("[similar-asin] coze poll picked pending states count={}", states.size());
for (TaskScopeStateEntity state : states) {
if (state == null || state.getId() == null) {
if (state == null || state.getId() == null || !isCozeStateOwnedByCurrentInstance(state)) {
continue;
}
cozeTaskExecutor.execute(() -> pollPendingCozeState(state.getId()));
@@ -1150,6 +1160,7 @@ public class SimilarAsinTaskService {
return false;
}
String prompt = readAiPrompt(task);
String apiKey = readApiKey(task);
int batchSize = Math.max(1, properties.getCozeBatchSize());
boolean pending = false;
for (TaskChunkEntity chunk : chunks) {
@@ -1166,7 +1177,7 @@ public class SimilarAsinTaskService {
for (int i = 0; i < unresolvedRows.size(); i += batchSize) {
List<SimilarAsinResultRowDto> batchRows =
unresolvedRows.subList(i, Math.min(i + batchSize, unresolvedRows.size()));
pending |= submitCozeBatch(task, result, job, chunk, batchRows, batchIndex, batchTotal, prompt, allRowsByBaseId);
pending |= submitCozeBatch(task, result, job, chunk, batchRows, batchIndex, batchTotal, prompt, apiKey, allRowsByBaseId);
batchIndex++;
}
}
@@ -1181,6 +1192,7 @@ public class SimilarAsinTaskService {
int batchIndex,
int batchTotal,
String prompt,
String apiKey,
Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId) {
if (batchRows == null || batchRows.isEmpty()) {
return false;
@@ -1197,7 +1209,7 @@ public class SimilarAsinTaskService {
|| COZE_STATUS_RUNNING.equals(existing.getCozeStatus());
}
try {
SimilarAsinCozeClient.CozeSubmitResponse submit = cozeClient.submitWorkflow(batchRows, prompt);
SimilarAsinCozeClient.CozeSubmitResponse submit = cozeClient.submitWorkflow(batchRows, prompt, apiKey);
if (submit.immediateData() != null && !submit.immediateData().isBlank()) {
List<SimilarAsinResultRowDto> cozeRows = cozeClient.mergeRowsFromDataText(batchRows, submit.immediateData());
mergeCozeRowsIntoChunk(task, chunk.getScopeHash(), chunk.getChunkIndex(), cozeRows, allRowsByBaseId);
@@ -1246,7 +1258,8 @@ public class SimilarAsinTaskService {
chunk.getScopeHash(),
chunk.getChunkIndex(),
batchIndex,
batchTotal
batchTotal,
currentInstanceId()
);
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
String storedBatchPayload = transientPayloadStorageService.storeParsedPayloadFast(
@@ -1278,6 +1291,25 @@ public class SimilarAsinTaskService {
}
private void pollPendingCozeState(Long stateId) {
TaskScopeStateEntity lockState = taskScopeStateMapper.selectById(stateId);
if (lockState == null || lockState.getCozeExecuteId() == null || lockState.getCozeExecuteId().isBlank()) {
return;
}
if (!isCozeStateOwnedByCurrentInstance(lockState)) {
log.info("[similar-asin] coze poll skipped because owner is another instance taskId={} stateId={} owner={} current={}",
lockState.getTaskId(), stateId, ownerFromCozeState(lockState), currentInstanceId());
return;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(lockState.getTaskId(), 0L);
if (taskLockHandle == null) {
return;
}
try (taskLockHandle) {
pollPendingCozeStateLocked(stateId);
}
}
private void pollPendingCozeStateLocked(Long stateId) {
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
return;
@@ -1285,16 +1317,7 @@ public class SimilarAsinTaskService {
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
return;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(state.getTaskId(), 0L);
if (taskLockHandle == null) {
return;
}
try (taskLockHandle) {
state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
return;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
if (!isCozeStateOwnedByCurrentInstance(state)) {
return;
}
if (!tryClaimCozeStateForPoll(state)) {
@@ -1305,6 +1328,11 @@ public class SimilarAsinTaskService {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing");
return;
}
if (!isOwnerCurrent(context.ownerInstanceId())) {
log.info("[similar-asin] coze poll skipped after context refresh because owner is another instance taskId={} stateId={} owner={} current={}",
state.getTaskId(), state.getId(), context.ownerInstanceId(), currentInstanceId());
return;
}
taskFileJobService.touchRunning(context.jobId());
try {
SimilarAsinCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(state.getCozeExecuteId());
@@ -1321,6 +1349,9 @@ public class SimilarAsinTaskService {
: "Coze async workflow completed without output";
}
List<SimilarAsinResultRowDto> batchRows = readCozeBatchRows(state);
if (batchRows.isEmpty() && failureMessage.isBlank()) {
failureMessage = "Coze batch payload missing";
}
List<SimilarAsinResultRowDto> cozeRows = failureMessage.isBlank()
? cozeClient.mergeRowsFromDataText(batchRows, poll.resolvedPayloadText())
: cozeClient.markRowsFailed(batchRows, failureMessage);
@@ -1354,7 +1385,6 @@ public class SimilarAsinTaskService {
state.getTaskId(), state.getId(), state.getCozeExecuteId(), message);
updateCozeStateRunning(state, message);
}
}
}
private void updateCozeStateRunning(TaskScopeStateEntity state, String error) {
@@ -1410,6 +1440,9 @@ public class SimilarAsinTaskService {
if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) {
return;
}
if (!isOwnerCurrent(context.ownerInstanceId())) {
return;
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, 0L);
if (lockHandle == null) {
return;
@@ -1442,7 +1475,7 @@ public class SimilarAsinTaskService {
}
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis);
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, TASK_LOCK_TTL, waitMillis);
if (lockHandle == null) {
log.info("[similar-asin] skip task operation because task lock is busy taskId={}", taskId);
}
@@ -1454,6 +1487,9 @@ public class SimilarAsinTaskService {
TaskFileJobEntity job,
int totalProgressUnits,
int cozeWorkUnits) {
if (countPendingCozeStates(task.getId()) > 0) {
throw new BusinessException("Coze result is still processing, cannot generate result file yet");
}
int assembleProgress = Math.max(1, Math.min(totalProgressUnits - 2, cozeWorkUnits));
saveFileBuildProgress(task, job, totalProgressUnits, assembleProgress, "Assembling xlsx");
assembleResultWorkbook(task, result);
@@ -1572,6 +1608,69 @@ public class SimilarAsinTaskService {
+ ":batch:" + batchIndex;
}
private String buildTaskOwnerScopeKey(FileTaskEntity task) {
Long taskId = task == null ? null : task.getId();
return "task:" + taskId + ":owner:" + firstNonBlank(ownerFromTask(task), currentInstanceId());
}
private String currentInstanceId() {
String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId();
return firstNonBlank(instanceId, "unknown-instance");
}
private boolean isJobOwnedByCurrentInstance(TaskFileJobEntity job) {
return job == null || isOwnerCurrent(ownerFromScopeKey(job.getScopeKey()));
}
private void ensureTaskOwnedByCurrentInstance(FileTaskEntity task, String operation) {
String owner = ownerFromTask(task);
if (isOwnerCurrent(owner)) {
return;
}
log.warn("[similar-asin] reject task operation because owner is another instance taskId={} operation={} owner={} current={}",
task == null ? null : task.getId(), operation, owner, currentInstanceId());
throw new BusinessException(40903, "This task is bound to another service instance");
}
private boolean isCozeStateOwnedByCurrentInstance(TaskScopeStateEntity state) {
return state == null || isOwnerCurrent(ownerFromCozeState(state));
}
private boolean isOwnerCurrent(String owner) {
return owner == null || owner.isBlank() || Objects.equals(owner, currentInstanceId());
}
private String ownerFromCozeState(TaskScopeStateEntity state) {
CozeBatchContext context = readCozeBatchContext(state);
return context == null ? null : context.ownerInstanceId();
}
private String ownerFromTask(FileTaskEntity task) {
if (task == null || task.getResultJson() == null || task.getResultJson().isBlank()) {
return null;
}
try {
String owner = objectMapper.readTree(task.getResultJson()).path("ownerInstanceId").asText("");
return owner.isBlank() ? null : owner;
} catch (Exception ex) {
log.warn("[similar-asin] read task owner failed taskId={} err={}", task.getId(), ex.getMessage());
return null;
}
}
private String ownerFromScopeKey(String scopeKey) {
if (scopeKey == null || scopeKey.isBlank()) {
return null;
}
String marker = ":owner:";
int index = scopeKey.lastIndexOf(marker);
if (index < 0) {
return null;
}
String owner = scopeKey.substring(index + marker.length()).trim();
return owner.isBlank() ? null : owner;
}
private int countCozeWorkUnits(List<TaskChunkEntity> chunks, int batchSize) {
if (chunks == null || chunks.isEmpty()) {
return 0;
@@ -1632,11 +1731,8 @@ public class SimilarAsinTaskService {
long resolvedRows = parsed.getAllItems().stream()
.filter(row -> findResultRow(row, resultMap) != null || findResultRowByAsin(row.getAsin(), resultMap) != null)
.count();
long reasonRows = resultMap.values().stream()
.filter(this::hasReasonFields)
.count();
log.info("[similar-asin] assemble workbook taskId={} parsedRows={} resultRows={} resolvedRows={} reasonRows={}",
task.getId(), parsed.getAllItems().size(), resultMap.size(), resolvedRows, reasonRows);
log.info("[similar-asin] assemble workbook taskId={} parsedRows={} resultRows={} resolvedRows={}",
task.getId(), parsed.getAllItems().size(), resultMap.size(), resolvedRows);
if (!parsed.getAllItems().isEmpty() && resultMap.isEmpty()) {
throw new BusinessException("相似ASIN检测结果为空，请稍后重试生成结果文件");
}
@@ -1707,15 +1803,13 @@ public class SimilarAsinTaskService {
private void writeResultWorkbook(File xlsx, SimilarAsinParsedPayloadDto parsed, Map<String, SimilarAsinResultRowDto> resultMap) {
try (SXSSFWorkbook workbook = new SXSSFWorkbook(200); FileOutputStream fos = new FileOutputStream(xlsx)) {
Sheet sheet = workbook.createSheet("相似ASIN检测结果");
Sheet sheet = workbook.createSheet("相似asin检测");
CellStyle headerStyle = workbook.createCellStyle();
Font font = workbook.createFont();
font.setBold(true);
headerStyle.setFont(font);
List<String> resultHeaders = new ArrayList<>(RESULT_HEADERS);
resultHeaders.add(4, "标题");
resultHeaders.add(5, "图片链接");
Row header = sheet.createRow(0);
for (int i = 0; i < resultHeaders.size(); i++) {
Cell cell = header.createCell(i);
@@ -1729,24 +1823,17 @@ public class SimilarAsinTaskService {
if (resultRow == null) {
resultRow = findResultRowByAsin(parsedRow.getAsin(), resultMap);
}
String missingReason = "";
if (resultRow == null) {
missingReason = hasPromptFields(parsedRow) ? "未匹配到检测结果" : "未送检：缺少标题或图片";
}
Row row = sheet.createRow(rowIndex++);
int col = 0;
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getDisplayId(), parsedRow.getSourceId()));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getAsin(), ""));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getCountry(), ""));
row.createCell(col++).setCellValue(readValueByHeader(parsedRow, "价格", "price"));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getTitle(), "") : firstNonBlank(resultRow.getTitle(), parsedRow.getTitle()));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getUrl(), "") : imageUrlCellValue(resultRow, parsedRow.getUrl()));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : userFacingCozeCellValue(resultRow, resultRow.getTitleRisk()));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : userFacingCozeCellValue(resultRow, resultRow.getAppearanceRisk()));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : userFacingCozeCellValue(resultRow, resultRow.getPatentRisk()));
row.createCell(col).setCellValue(resultRow == null ? "未送检" : userFacingConclusion(resultRow));
row.createCell(col++).setCellValue(resultRow == null
? firstNonBlank(parsedRow.getPrice(), readValueByHeader(parsedRow, "价格", "price"))
: firstNonBlank(resultRow.getPrice(), firstNonBlank(parsedRow.getPrice(), readValueByHeader(parsedRow, "价格", "price"))));
row.createCell(col++).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getIsStock(), ""));
row.createCell(col).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getSimilarity(), ""));
}
writeReasonSheet(workbook, headerStyle, parsed, resultMap);
workbook.write(fos);
workbook.dispose();
} catch (Exception ex) {
@@ -1754,39 +1841,6 @@ public class SimilarAsinTaskService {
}
}
private void writeReasonSheet(SXSSFWorkbook workbook,
CellStyle headerStyle,
SimilarAsinParsedPayloadDto parsed,
Map<String, SimilarAsinResultRowDto> resultMap) {
Sheet sheet = workbook.createSheet("原因");
Row header = sheet.createRow(0);
List<String> headers = List.of("ASIN", "外观原因", "专利原因", "标题原因");
for (int i = 0; i < headers.size(); i++) {
Cell cell = header.createCell(i);
cell.setCellValue(headers.get(i));
cell.setCellStyle(headerStyle);
}
Set<String> writtenAsins = new LinkedHashSet<>();
int rowIndex = 1;
for (SimilarAsinParsedRowVo parsedRow : parsed.getAllItems()) {
String asin = normalize(parsedRow.getAsin()).toUpperCase(Locale.ROOT);
if (asin.isBlank() || !writtenAsins.add(asin)) {
continue;
}
SimilarAsinResultRowDto resultRow = findResultRow(parsedRow, resultMap);
if (resultRow == null) {
resultRow = findResultRowByAsin(asin, resultMap);
}
Row row = sheet.createRow(rowIndex++);
row.createCell(0).setCellValue(asin);
row.createCell(1).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getAppearanceReason(), ""));
row.createCell(2).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getPatentReason(), ""));
row.createCell(3).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getTitleReason(), ""));
rowIndex++;
}
}
private SimilarAsinResultRowDto findResultRowByAsin(String asin, Map<String, SimilarAsinResultRowDto> resultMap) {
String normalizedAsin = normalize(asin).toUpperCase(Locale.ROOT);
if (normalizedAsin.isBlank() || resultMap == null || resultMap.isEmpty()) {
@@ -1797,7 +1851,7 @@ public class SimilarAsinTaskService {
if (row == null || !normalizedAsin.equals(normalize(row.getAsin()).toUpperCase(Locale.ROOT))) {
continue;
}
if (hasResolvedCozeFields(row) || hasReasonFields(row)) {
if (hasResolvedCozeFields(row)) {
return row;
}
if (fallback == null) {
@@ -1807,15 +1861,6 @@ public class SimilarAsinTaskService {
return fallback;
}
private boolean hasReasonFields(SimilarAsinResultRowDto row) {
if (row == null) {
return false;
}
return !normalize(row.getAppearanceReason()).isBlank()
|| !normalize(row.getPatentReason()).isBlank()
|| !normalize(row.getTitleReason()).isBlank();
}
private String imageUrlCellValue(SimilarAsinResultRowDto resultRow, String fallbackUrl) {
if (resultRow == null) {
return firstNonBlank(fallbackUrl, "");
@@ -1840,6 +1885,7 @@ public class SimilarAsinTaskService {
int idCol = findRequiredHeader(headerMap, "id");
int asinCol = findRequiredHeader(headerMap, "asin");
int countryCol = findRequiredHeader(headerMap, "国家", "country");
int priceCol = findOptionalHeaderExact(headerMap, "价格", "price");
int urlCol = findOptionalHeaderExact(headerMap,
"url", "rul", "link", "image", "img", "pic", "picture",
"链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接");
@@ -1882,6 +1928,7 @@ public class SimilarAsinTaskService {
vo.setRowToken(buildRowToken(source.getFileKey(), vo.getRowIndex()));
vo.setAsin(asin);
vo.setCountry(country);
vo.setPrice(priceCol >= 0 ? cell(row, priceCol, formatter) : "");
vo.setUrl(urlCol >= 0 ? cell(row, urlCol, formatter) : "");
vo.setTitle(titleCol >= 0 ? cell(row, titleCol, formatter) : "");
vo.setValues(readRowValues(row, headers, formatter));
@@ -1938,7 +1985,9 @@ public class SimilarAsinTaskService {
return hasUsableCozeField(row.getTitleRisk())
|| hasUsableCozeField(row.getAppearanceRisk())
|| hasUsableCozeField(row.getPatentRisk())
|| hasUsableCozeField(row.getConclusion());
|| hasUsableCozeField(row.getConclusion())
|| hasUsableCozeField(row.getIsStock())
|| hasUsableCozeField(row.getSimilarity());
}
private boolean hasUsableCozeField(String value) {
@@ -2183,9 +2232,10 @@ public class SimilarAsinTaskService {
return val == null ? "" : val.replace(String.valueOf((char) 0xFEFF), "").replace((char) 0x3000, ' ').trim().replaceAll("\\s+", " ");
}
private String buildParsedPayloadJson(String aiPrompt, List<SimilarAsinSourceFileDto> sourceFiles, List<String> headers, List<SimilarAsinParsedGroupVo> groups, List<SimilarAsinParsedRowVo> allRows) {
private String buildParsedPayloadJson(String aiPrompt, String apiKey, List<SimilarAsinSourceFileDto> sourceFiles, List<String> headers, List<SimilarAsinParsedGroupVo> groups, List<SimilarAsinParsedRowVo> allRows) {
SimilarAsinParsedPayloadDto payload = new SimilarAsinParsedPayloadDto();
payload.setAiPrompt(normalize(aiPrompt));
payload.setApiKey(normalize(apiKey));
payload.setSourceFiles(sourceFiles == null ? List.of() : sourceFiles);
payload.setHeaders(headers == null ? List.of() : headers);
payload.setItems(List.of());
@@ -2194,14 +2244,16 @@ public class SimilarAsinTaskService {
return writeJson(payload, "保存解析结果失败");
}
private String buildTaskResultJson(String aiPrompt, List<SimilarAsinSourceFileDto> sourceFiles, String parsedPayloadPointer) {
private String buildTaskResultJson(String aiPrompt, String apiKey, List<SimilarAsinSourceFileDto> sourceFiles, String parsedPayloadPointer) {
Map<String, Object> payload = new LinkedHashMap<>();
payload.put("aiPrompt", normalize(aiPrompt));
payload.put("apiKey", normalize(apiKey));
payload.put("sourceFileKeys", sourceFiles == null ? List.of() : sourceFiles.stream()
.map(SimilarAsinSourceFileDto::getFileKey)
.filter(Objects::nonNull)
.toList());
payload.put("parsedPayloadRef", parsedPayloadPointer);
payload.put("ownerInstanceId", currentInstanceId());
return writeJson(payload, "保存任务结果索引失败");
}
@@ -2435,11 +2487,12 @@ public class SimilarAsinTaskService {
}
private record CozeBatchContext(Long jobId,
Long resultId,
String chunkScopeHash,
Integer chunkIndex,
Integer batchIndex,
Integer batchTotal) {
Long resultId,
String chunkScopeHash,
Integer chunkIndex,
Integer batchIndex,
Integer batchTotal,
String ownerInstanceId) {
}
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<SimilarAsinParsedRowVo> allRows) {

7
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobDispatchCoordinator.java
View File

@@ -33,9 +33,12 @@ public class TaskFileJobDispatchCoordinator {
if (published) {
return;
}
boolean dispatched = taskFileJobLocalDispatcher.dispatch(event.jobId(), event.taskId(), event.moduleType());
boolean dispatched = taskFileJobLocalDispatcher.dispatch(event.jobId(), event.taskId(), event.moduleType(), true);
if (dispatched) {
log.info("[task-file-job] local async dispatch scheduled jobId={} taskId={} moduleType={}",
log.info("[task-file-job] forced local async dispatch scheduled jobId={} taskId={} moduleType={}",
event.jobId(), event.taskId(), event.moduleType());
} else {
log.warn("[task-file-job] local dispatch unavailable after mq publish fallback jobId={} taskId={} moduleType={}",
event.jobId(), event.taskId(), event.moduleType());
}
}

57
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java
View File

@@ -27,35 +27,50 @@ public class TaskFileJobLocalDispatcher {
private boolean localDispatchEnabled;
public boolean dispatch(Long jobId, Long taskId, String moduleType) {
if (!localDispatchEnabled || jobId == null || jobId <= 0) {
return dispatch(jobId, taskId, moduleType, false);
}
public boolean dispatch(Long jobId, Long taskId, String moduleType, boolean force) {
if (jobId == null || jobId <= 0) {
return false;
}
if (!force && !localDispatchEnabled) {
log.info("[task-file-job] local dispatch disabled jobId={} taskId={} moduleType={}",
jobId, taskId, moduleType);
return false;
}
try {
taskFileJobDispatchExecutor.execute(() -> {
try {
TaskFileJobEntity job = taskFileJobMapper.selectById(jobId);
if (job == null) {
log.warn("[task-file-job] local dispatch ignored, job not found jobId={} taskId={} moduleType={}",
jobId, taskId, moduleType);
return;
}
TaskResultFileJobWorker taskResultFileJobWorker = taskResultFileJobWorkerProvider.getIfAvailable();
if (taskResultFileJobWorker == null) {
log.warn("[task-file-job] local dispatch ignored, worker unavailable jobId={} taskId={} moduleType={}",
jobId, taskId, moduleType);
return;
}
taskResultFileJobWorker.process(job);
} catch (Exception ex) {
log.warn("[task-file-job] local dispatch failed jobId={} taskId={} moduleType={} msg={}",
jobId, taskId, moduleType, ex.getMessage(), ex);
}
});
taskFileJobDispatchExecutor.execute(() -> processLocally(jobId, taskId, moduleType));
return true;
} catch (RuntimeException ex) {
log.warn("[task-file-job] local dispatch executor rejected jobId={} taskId={} moduleType={} msg={}",
jobId, taskId, moduleType, ex.getMessage(), ex);
if (force) {
processLocally(jobId, taskId, moduleType);
return true;
}
return false;
}
}
private void processLocally(Long jobId, Long taskId, String moduleType) {
try {
TaskFileJobEntity job = taskFileJobMapper.selectById(jobId);
if (job == null) {
log.warn("[task-file-job] local dispatch ignored, job not found jobId={} taskId={} moduleType={}",
jobId, taskId, moduleType);
return;
}
TaskResultFileJobWorker taskResultFileJobWorker = taskResultFileJobWorkerProvider.getIfAvailable();
if (taskResultFileJobWorker == null) {
log.warn("[task-file-job] local dispatch ignored, worker unavailable jobId={} taskId={} moduleType={}",
jobId, taskId, moduleType);
return;
}
taskResultFileJobWorker.process(job);
} catch (Exception ex) {
log.warn("[task-file-job] local dispatch failed jobId={} taskId={} moduleType={} msg={}",
jobId, taskId, moduleType, ex.getMessage(), ex);
}
}
}

37
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java
View File

@@ -1,6 +1,7 @@
package com.nanri.aiimage.modules.task.service;
import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskService;
import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.modules.brand.service.BrandTaskService;
import com.nanri.aiimage.modules.deletebrand.service.DeleteBrandRunService;
import com.nanri.aiimage.modules.patroldelete.service.PatrolDeleteTaskService;
@@ -33,6 +34,7 @@ public class TaskResultFileJobWorker {
private final TaskResultPayloadService taskResultPayloadService;
private final FileResultMapper fileResultMapper;
private final TaskFileJobLocalDispatcher taskFileJobLocalDispatcher;
private final InstanceMetadata instanceMetadata;
private final ShopMatchTaskService shopMatchTaskService;
private final PriceTrackTaskService priceTrackTaskService;
private final ProductRiskTaskService productRiskTaskService;
@@ -86,6 +88,11 @@ public class TaskResultFileJobWorker {
}
public void process(TaskFileJobEntity job) {
if (isOwnerScopedCozeJob(job) && !isOwnedByCurrentInstance(job)) {
log.info("[task-file-job] skip owner-scoped job because owner is another instance jobId={} taskId={} moduleType={} owner={} current={}",
job.getId(), job.getTaskId(), job.getModuleType(), ownerFromScopeKey(job.getScopeKey()), currentInstanceId());
return;
}
if (job == null || job.getId() == null || !taskFileJobService.markRunning(job.getId())) {
return;
}
@@ -147,6 +154,36 @@ public class TaskResultFileJobWorker {
return result == null ? null : result.getResultFileUrl();
}
private boolean isOwnerScopedCozeJob(TaskFileJobEntity job) {
if (job == null || job.getModuleType() == null) {
return false;
}
return "APPEARANCE_PATENT".equals(job.getModuleType()) || "SIMILAR_ASIN".equals(job.getModuleType());
}
private boolean isOwnedByCurrentInstance(TaskFileJobEntity job) {
String owner = ownerFromScopeKey(job == null ? null : job.getScopeKey());
return owner == null || owner.isBlank() || owner.equals(currentInstanceId());
}
private String currentInstanceId() {
String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId();
return instanceId == null || instanceId.isBlank() ? "unknown-instance" : instanceId;
}
private String ownerFromScopeKey(String scopeKey) {
if (scopeKey == null || scopeKey.isBlank()) {
return null;
}
String marker = ":owner:";
int index = scopeKey.lastIndexOf(marker);
if (index < 0) {
return null;
}
String owner = scopeKey.substring(index + marker.length()).trim();
return owner.isBlank() ? null : owner;
}
private boolean dispatch(TaskFileJobEntity job) {
String moduleType = job.getModuleType();
if ("SHOP_MATCH".equals(moduleType)) {

4
backend-java/src/main/resources/application.yml
View File

@@ -150,6 +150,8 @@ aiimage:
coze-batch-size: ${AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE:10}
coze-connect-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_CONNECT_TIMEOUT_MILLIS:10000}
coze-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_READ_TIMEOUT_MILLIS:60000}
coze-poll-interval-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_INTERVAL_MILLIS:5000}
coze-poll-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_TIMEOUT_MILLIS:600000}
stale-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES:20}
stale-finalize-cron: ${AIIMAGE_APPEARANCE_PATENT_STALE_FINALIZE_CRON:0 */2 * * * *}
similar-asin:
@@ -160,6 +162,8 @@ aiimage:
coze-batch-size: ${AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE:10}
coze-connect-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_CONNECT_TIMEOUT_MILLIS:10000}
coze-read-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_READ_TIMEOUT_MILLIS:60000}
coze-poll-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_INTERVAL_MILLIS:5000}
coze-poll-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_TIMEOUT_MILLIS:600000}
stale-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_STALE_TIMEOUT_MINUTES:20}
stale-finalize-cron: ${AIIMAGE_SIMILAR_ASIN_STALE_FINALIZE_CRON:0 */2 * * * *}
security: