From a00ff1804c4f629dff83610ac127f09858b2512c Mon Sep 17 00:00:00 2001 From: super <2903208875@qq.com> Date: Wed, 6 May 2026 00:01:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/AppearancePatentProperties.java | 4 +- .../aiimage/config/SimilarAsinProperties.java | 4 +- .../config/TaskOperationLockConfig.java | 16 +- .../client/AppearancePatentCozeClient.java | 71 ++-- .../dto/AppearancePatentParseRequest.java | 7 + .../dto/AppearancePatentParsedPayloadDto.java | 3 + .../service/AppearancePatentTaskService.java | 174 +++++++--- .../client/SimilarAsinCozeClient.java | 123 +++++-- .../model/dto/SimilarAsinParseRequest.java | 7 + .../dto/SimilarAsinParsedPayloadDto.java | 3 + .../model/dto/SimilarAsinResultRowDto.java | 21 +- .../model/vo/SimilarAsinParsedRowVo.java | 3 + .../service/SimilarAsinTaskService.java | 307 ++++++++++-------- .../TaskFileJobDispatchCoordinator.java | 7 +- .../service/TaskFileJobLocalDispatcher.java | 57 ++-- .../task/service/TaskResultFileJobWorker.java | 37 +++ .../src/main/resources/application.yml | 4 + .../components/BrandAppearancePatentTab.vue | 70 +++- .../brand/components/BrandSimilarAsinTab.vue | 127 ++++++-- frontend-vue/src/shared/api/java-modules.ts | 11 +- new_web_source/appearance-patent.html | 30 +- new_web_source/similar-asin.html | 18 + 22 files changed, 796 insertions(+), 308 deletions(-) create mode 100644 new_web_source/similar-asin.html diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java index 6ab2db0..ad6b46b 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java @@ -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 * * * *"; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java index 711fca0..c4c8adc 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java @@ -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 * * * *"; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java index 25e8365..b52ab42 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java @@ -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 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, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java index ce85200..abd4107 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java @@ -30,7 +30,7 @@ public class AppearancePatentCozeClient { private final AppearancePatentProperties properties; private final ObjectMapper objectMapper; - public List inspect(List rows, String prompt) { + public List inspect(List 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 rows, String prompt) throws Exception { - JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt)); + public CozeSubmitResponse submitWorkflow(List 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 inspectWithFallback(List rows, String prompt) { + private List inspectWithFallback(List 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 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 inspectPartitionWithFailureFallback(List rows, String prompt) { + private List inspectPartitionWithFailureFallback(List 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 inspectSingleRowWithRetry(List rows, String prompt) throws Exception { + private List inspectSingleRowWithRetry(List 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 rows, String prompt) throws Exception { - String raw = runWorkflowAsyncAndWait(rows, prompt); + private InspectAttempt inspectOnce(List rows, String prompt, String apiKey) throws Exception { + String raw = runWorkflowAsyncAndWait(rows, prompt, apiKey); List 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 rows, String prompt) throws Exception { - JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt)); + private String runWorkflowAsyncAndWait(List 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 rows, String prompt) { - Map parameters = buildParameters(rows, prompt); + private String postWorkflow(List rows, String prompt, String apiKey) { + Map parameters = buildParameters(rows, prompt, apiKey); Map 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 buildParameters(List rows, String prompt) { + private Map buildParameters(List rows, String prompt, String apiKey) { List groupKeys = rows.stream().map(row -> nonBlank(row.getGroupKey(), rowKey(row))).toList(); List rowIds = rows.stream().map(row -> nonBlank(row.getId(), "")).toList(); List 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 maskCozeRequestBody(Map body) { + Map masked = new LinkedHashMap<>(body); + Object parametersObj = masked.get("parameters"); + if (parametersObj instanceof Map parameters) { + Map maskedParameters = new LinkedHashMap<>((Map) 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> buildItemObjects(List rows, List groupKeys, List rowIds, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParseRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParseRequest.java index 71fd5a1..9cc5b1f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParseRequest.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParseRequest.java @@ -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; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedPayloadDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedPayloadDto.java index 95abf6f..ef63cdd 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedPayloadDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentParsedPayloadDto.java @@ -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 sourceFiles = new ArrayList<>(); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index 5a3f5a4..015a7da 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -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 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 states = taskScopeStateMapper.selectList(new LambdaQueryWrapper() + .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 states = taskScopeStateMapper.selectList(new LambdaQueryWrapper() - .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 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> 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 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 sourceFiles, List headers, List groups, List allRows) { + private String buildParsedPayloadJson(String aiPrompt, String apiKey, List sourceFiles, List headers, List groups, List 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 sourceFiles, String parsedPayloadPointer) { + private String buildTaskResultJson(String aiPrompt, String apiKey, List sourceFiles, String parsedPayloadPointer) { Map 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 headers, List allRows) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java index ec4da81..ed70cd3 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java @@ -30,7 +30,7 @@ public class SimilarAsinCozeClient { private final SimilarAsinProperties properties; private final ObjectMapper objectMapper; - public List inspect(List rows, String prompt) { + public List inspect(List 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 rows, String prompt) throws Exception { - JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt)); + public CozeSubmitResponse submitWorkflow(List 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 inspectWithFallback(List rows, String prompt) { + private List inspectWithFallback(List 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 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 inspectPartitionWithFailureFallback(List rows, String prompt) { + private List inspectPartitionWithFailureFallback(List 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 inspectSingleRowWithRetry(List rows, String prompt) throws Exception { + private List inspectSingleRowWithRetry(List 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 rows, String prompt) throws Exception { - String raw = runWorkflowAsyncAndWait(rows, prompt); + private InspectAttempt inspectOnce(List rows, String prompt, String apiKey) throws Exception { + String raw = runWorkflowAsyncAndWait(rows, prompt, apiKey); List 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 rows, String prompt) throws Exception { - JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt)); + private String runWorkflowAsyncAndWait(List 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 rows, String prompt) { - Map parameters = buildParameters(rows, prompt); + private String postWorkflow(List rows, String prompt, String apiKey) { + Map parameters = buildParameters(rows, prompt, apiKey); Map 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 buildParameters(List rows, String prompt) { + private Map buildParameters(List rows, String prompt, String apiKey) { List groupKeys = rows.stream().map(row -> nonBlank(row.getGroupKey(), rowKey(row))).toList(); List rowIds = rows.stream().map(row -> nonBlank(row.getId(), "")).toList(); List asins = rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList(); List countries = rows.stream().map(row -> nonBlank(row.getCountry(), "")).toList(); + List prices = rows.stream().map(row -> nonBlank(row.getPrice(), "")).toList(); List titles = rows.stream().map(row -> nonBlank(row.getTitle(), row.getAsin())).toList(); List urls = rows.stream().map(row -> nonBlank(row.getUrl(), "")).toList(); List> 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 maskCozeRequestBody(Map body) { + Map masked = new LinkedHashMap<>(body); + Object parametersObj = masked.get("parameters"); + if (parametersObj instanceof Map parameters) { + Map maskedParameters = new LinkedHashMap<>((Map) 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> buildItemObjects(List rows, List groupKeys, List rowIds, List asins, List countries, + List prices, List titles, + List urls, List> urlLists) { List> 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, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParseRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParseRequest.java index 229e663..b8e354f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParseRequest.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParseRequest.java @@ -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; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParsedPayloadDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParsedPayloadDto.java index 83216af..471eda1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParsedPayloadDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinParsedPayloadDto.java @@ -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 sourceFiles = new ArrayList<>(); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java index e228702..a061356 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java @@ -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) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinParsedRowVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinParsedRowVo.java index 848485c..162aea0 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinParsedRowVo.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinParsedRowVo.java @@ -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; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 5ee9461..fb268c1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -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 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 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 result = new ArrayList<>(); Set emittedKeys = new LinkedHashSet<>(); for (SimilarAsinResultRowDto representative : representatives) { - List 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 pickGroupRepresentativesForCoze(java.util.Collection rows) { - Map> 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 representatives = new ArrayList<>(); - for (List 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 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> 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 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 batchRows = readCozeBatchRows(state); + if (batchRows.isEmpty() && failureMessage.isBlank()) { + failureMessage = "Coze batch payload missing"; + } List 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 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 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 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 resultMap) { - Sheet sheet = workbook.createSheet("原因"); - Row header = sheet.createRow(0); - List 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 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 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 sourceFiles, List headers, List groups, List allRows) { + private String buildParsedPayloadJson(String aiPrompt, String apiKey, List sourceFiles, List headers, List groups, List 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 sourceFiles, String parsedPayloadPointer) { + private String buildTaskResultJson(String aiPrompt, String apiKey, List sourceFiles, String parsedPayloadPointer) { Map 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 headers, List allRows) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobDispatchCoordinator.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobDispatchCoordinator.java index f797853..51e4a2c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobDispatchCoordinator.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobDispatchCoordinator.java @@ -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()); } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java index e6b38b4..46cdc7e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobLocalDispatcher.java @@ -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); + } + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java index a21ec42..4c06512 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java @@ -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)) { diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 9248ad9..3c95300 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -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: diff --git a/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue b/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue index e76a3c9..f09b551 100644 --- a/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue @@ -24,6 +24,11 @@
{{ defaultAiPrompt }}
+
+
密钥
+ +
+

Python 回传字段按 asin、国家、url、标题提交；后端接收分片后按 10 条一批调用 Coze。

-
-
任务 ID：{{ parseResult.taskId }}
-
总行数：{{ parseResult.totalRows }}，有效：{{ parseResult.acceptedRows }}，分组：{{ parseResult.groupCount || 0 }}，过滤：{{ parseResult.droppedRows }}
+
+
任务 ID：{{ visibleTaskSummary.taskId }}
+
总行数：{{ visibleTaskSummary.totalRows }}，有效：{{ visibleTaskSummary.acceptedRows }}，分组：{{ visibleTaskSummary.groupCount || 0 }}，过滤：{{ visibleTaskSummary.droppedRows }}
{{ queuePayloadText }}
@@ -171,6 +176,8 @@ const selectedFileNames = ref([]) const uploadedFiles = ref([]) const parseResult = ref(null) const parsedGroups = ref([]) +type TaskSummary = Pick +const queuedTaskSummary = ref(null) const defaultAiPrompt = `请帮我排查以下亚马逊商品在英国及欧洲地区是否存在知识产权侵权风险。 请直接给出明确结论（有侵权风险 或 未发现明显侵权风险），并严格按照以下三个维度提供精简的排查理由： 外观维度： 评估产品外形、图案设计是否与欧洲/英国常见外观专利雷同。 @@ -178,6 +185,7 @@ const defaultAiPrompt = `请帮我排查以下亚马逊商品在英国及欧洲 标题维度： 排查标题中是否包含大牌商标、敏感词或版权保护词汇。 结论：XX` const aiPrompt = ref('') +const cozeApiKey = ref('') const parsing = ref(false) const pushing = ref(false) const queuePayloadText = ref('') @@ -203,6 +211,7 @@ const parsedRows = computed(() => parsedGroups.value.flatMap((group) => group.items || []), ) const previewRows = computed(() => parsedGroups.value[0]?.items || []) +const visibleTaskSummary = computed(() => parseResult.value || queuedTaskSummary.value) const currentItems = computed(() => historyItems.value.filter((i) => i.taskId != null && pollingTaskIds.value.includes(i.taskId))) const historyOnlyItems = computed(() => historyItems.value.filter((i) => { @@ -215,6 +224,28 @@ function effectiveAiPrompt() { return aiPrompt.value.trim() || defaultAiPrompt } +function effectiveCozeApiKey() { + return cozeApiKey.value.trim() +} + +function maskSecret(secret: string) { + if (!secret) return '' + if (secret.length <= 10) return '***' + return `${secret.slice(0, 6)}***${secret.slice(-4)}` +} + +function payloadForDisplay }>(payload: T) { + return { + ...payload, + data: payload.data + ? { + ...payload.data, + api_key: maskSecret(String(payload.data.api_key || '')), + } + : payload.data, + } +} + function uidForStorage() { return typeof window !== 'undefined' ? window.localStorage.getItem('uid') || '0' : '0' } @@ -223,6 +254,16 @@ function pollingKey() { return `appearance-patent:tasks:${uidForStorage()}` } +function toTaskSummary(result: AppearancePatentParseVo): TaskSummary { + return { + taskId: result.taskId, + totalRows: result.totalRows, + acceptedRows: result.acceptedRows, + groupCount: result.groupCount, + droppedRows: result.droppedRows, + } +} + function savePollingIds() { if (typeof window === 'undefined') return const ids = Array.from(new Set([...pollingTaskIds.value, ...pendingFileTaskIds.value])) @@ -274,6 +315,7 @@ async function selectFiles() { selectedFileNames.value = files.map((f) => f.relativePath || f.originalFilename || f.fileKey) parseResult.value = null parsedGroups.value = [] + queuedTaskSummary.value = null queuePayloadText.value = '' } @@ -296,6 +338,7 @@ async function selectFolder() { selectedFileNames.value = result.items.map((item) => item.relativePath || item.absolutePath) parseResult.value = null parsedGroups.value = [] + queuedTaskSummary.value = null queuePayloadText.value = '' ElMessage.success(`已选择文件夹内 ${result.items.length} 个 Excel 文件`) } catch (error) { @@ -308,6 +351,10 @@ async function parseFiles() { ElMessage.warning('请先选择 Excel') return } + if (!effectiveCozeApiKey()) { + ElMessage.warning('请先填写密钥') + return + } parsing.value = true try { const files: UploadedFileRef[] = uploadedFiles.value.map((f) => ({ @@ -315,8 +362,9 @@ async function parseFiles() { originalFilename: f.originalFilename, relativePath: f.relativePath, })) - const res = await parseAppearancePatent(files, effectiveAiPrompt()) + const res = await parseAppearancePatent(files, effectiveAiPrompt(), effectiveCozeApiKey()) parseResult.value = res + queuedTaskSummary.value = null parsedGroups.value = res.groups?.length ? res.groups : (res.items?.length @@ -341,7 +389,8 @@ async function parseFiles() { async function pushToPythonQueue() { const api = getPywebviewApi() - const taskId = parseResult.value?.taskId + const currentParseResult = parseResult.value + const taskId = currentParseResult?.taskId if (!taskId || !parsedRows.value.length) { ElMessage.warning('请先解析文件') return @@ -357,7 +406,8 @@ async function pushToPythonQueue() { ts: Date.now(), data: { taskId, - prompt: parseResult.value?.aiPrompt || effectiveAiPrompt(), + prompt: currentParseResult.aiPrompt || effectiveAiPrompt(), + api_key: effectiveCozeApiKey(), groups: parsedGroups.value.map((group) => ({ sourceFileKey: group.sourceFileKey || '', sourceFilename: group.sourceFilename || '', @@ -389,13 +439,14 @@ async function pushToPythonQueue() { })), }, } - queuePayloadText.value = JSON.stringify(payload, null, 2) + queuePayloadText.value = JSON.stringify(payloadForDisplay(payload), null, 2) await activateAppearancePatentTask(taskId) const result = await api.enqueue_json(payload) if (!result?.success) { ElMessage.error(result?.error || '推送失败') return } + queuedTaskSummary.value = toTaskSummary(currentParseResult) addPollingTask(taskId) clearParsedTask() await loadDashboard() @@ -721,9 +772,10 @@ onUnmounted(() => { .btn-run:disabled { opacity: .55; cursor: not-allowed; } .selected-files { margin-top: 14px; color: #999; font-size: 12px; word-break: break-all; } .selected-files span { display: block; margin: 4px 0; } -.prompt-card { margin-bottom: 18px; } +.prompt-card, .secret-card { margin-bottom: 18px; } .prompt-input { width: 100%; box-sizing: border-box; resize: vertical; min-height: 180px; padding: 10px 12px; border: 1px solid #333; border-radius: 8px; background: #202020; color: #d8d8d8; font-size: 12px; line-height: 1.6; outline: none; } -.prompt-input:focus { border-color: #3498db; } +.secret-input { width: 100%; box-sizing: border-box; height: 38px; padding: 0 12px; border: 1px solid #333; border-radius: 8px; background: #202020; color: #d8d8d8; font-size: 12px; outline: none; } +.prompt-input:focus, .secret-input:focus { border-color: #3498db; } .prompt-default-label { margin-top: 10px; color: #8d8d8d; font-size: 12px; } .prompt-preview { margin-top: 10px; padding: 12px; border: 1px solid #2a2a2a; border-radius: 8px; background: #202020; color: #9ea7b3; font-size: 12px; line-height: 1.6; white-space: pre-wrap; } .parse-card, .queue-payload { margin-top: 14px; padding: 12px; border: 1px solid #2a2a2a; border-radius: 8px; background: #202020; color: #b8c1cc; font-size: 12px; } diff --git a/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue b/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue index 22bdccf..12dd64a 100644 --- a/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue @@ -6,7 +6,7 @@ @@ -67,13 +72,14 @@
解析结果 - 显示首组 {{ previewRows.length }} 条 / 共 {{ parseResult?.groupCount || parsedGroups.length }} 组，{{ parsedRows.length }} 条 + 共 {{ parsedRows.length }} 条
暂无解析结果
+ @@ -171,13 +177,14 @@ const selectedFileNames = ref([]) const uploadedFiles = ref([]) const parseResult = ref(null) const parsedGroups = ref([]) -const defaultAiPrompt = `请帮我排查以下亚马逊商品在英国及欧洲地区是否存在知识产权侵权风险。 -请直接给出明确结论（有侵权风险 或 未发现明显侵权风险），并严格按照以下三个维度提供精简的排查理由： -外观维度： 评估产品外形、图案设计是否与欧洲/英国常见外观专利雷同。 -专利维度： 评估产品的核心技术或物理结构是否存在侵权可能。 -标题维度： 排查标题中是否包含大牌商标、敏感词或版权保护词汇。 -结论：XX` +type TaskSummary = Pick +const queuedTaskSummary = ref(null) +const defaultAiPrompt = `1. 图案与形象限制：不含各国国旗及类似图案，无各国领导人画像；不含圣诞老人形象；排除迪士尼、漫威、三丽鸥等所有指定动漫IP的图案与关联物品，且不涉及知名油画作品图案。 +2. 品类排除：非图书、刀具、服装；不含药品、带容器的粉末及液体；与吸烟行为无关，无医疗标志。 +3. 功能与配件要求：不可充电且不带电池，无磁吸功能；配件多的情况下需完全匹配。 +4. 规格与外观限制：不含体积大的家具类物品（如椅子、桌子、大镜子、柜子）；不带任何品牌logo；无多种款式可选。` const aiPrompt = ref('') +const cozeApiKey = ref('') const parsing = ref(false) const pushing = ref(false) const queuePayloadText = ref('') @@ -200,9 +207,12 @@ const dashboard = ref({ const historyItems = ref([]) const parsedRows = computed(() => - parsedGroups.value.flatMap((group) => group.items || []), + parseResult.value?.items?.length + ? parseResult.value.items + : parsedGroups.value.flatMap((group) => group.items || []), ) -const previewRows = computed(() => parsedGroups.value[0]?.items || []) +const previewRows = computed(() => parsedRows.value) +const visibleTaskSummary = computed(() => parseResult.value || queuedTaskSummary.value) const currentItems = computed(() => historyItems.value.filter((i) => i.taskId != null && pollingTaskIds.value.includes(i.taskId))) const historyOnlyItems = computed(() => historyItems.value.filter((i) => { @@ -215,6 +225,28 @@ function effectiveAiPrompt() { return aiPrompt.value.trim() || defaultAiPrompt } +function effectiveCozeApiKey() { + return cozeApiKey.value.trim() +} + +function maskSecret(secret: string) { + if (!secret) return '' + if (secret.length <= 10) return '***' + return `${secret.slice(0, 6)}***${secret.slice(-4)}` +} + +function payloadForDisplay }>(payload: T) { + return { + ...payload, + data: payload.data + ? { + ...payload.data, + api_key: maskSecret(String(payload.data.api_key || '')), + } + : payload.data, + } +} + function uidForStorage() { return typeof window !== 'undefined' ? window.localStorage.getItem('uid') || '0' : '0' } @@ -223,6 +255,16 @@ function pollingKey() { return `similar-asin:tasks:${uidForStorage()}` } +function toTaskSummary(result: SimilarAsinParseVo): TaskSummary { + return { + taskId: result.taskId, + totalRows: result.totalRows, + acceptedRows: result.acceptedRows, + groupCount: result.groupCount, + droppedRows: result.droppedRows, + } +} + function savePollingIds() { if (typeof window === 'undefined') return const ids = Array.from(new Set([...pollingTaskIds.value, ...pendingFileTaskIds.value])) @@ -274,6 +316,7 @@ async function selectFiles() { selectedFileNames.value = files.map((f) => f.relativePath || f.originalFilename || f.fileKey) parseResult.value = null parsedGroups.value = [] + queuedTaskSummary.value = null queuePayloadText.value = '' } @@ -296,6 +339,7 @@ async function selectFolder() { selectedFileNames.value = result.items.map((item) => item.relativePath || item.absolutePath) parseResult.value = null parsedGroups.value = [] + queuedTaskSummary.value = null queuePayloadText.value = '' ElMessage.success(`已选择文件夹内 ${result.items.length} 个 Excel 文件`) } catch (error) { @@ -308,6 +352,10 @@ async function parseFiles() { ElMessage.warning('请先选择 Excel') return } + if (!effectiveCozeApiKey()) { + ElMessage.warning('请先填写密钥') + return + } parsing.value = true try { const files: UploadedFileRef[] = uploadedFiles.value.map((f) => ({ @@ -315,23 +363,22 @@ async function parseFiles() { originalFilename: f.originalFilename, relativePath: f.relativePath, })) - const res = await parseSimilarAsin(files, effectiveAiPrompt()) + const res = await parseSimilarAsin(files, effectiveAiPrompt(), effectiveCozeApiKey()) parseResult.value = res - parsedGroups.value = res.groups?.length - ? res.groups - : (res.items?.length - ? [{ - sourceFileKey: res.items[0]?.sourceFileKey, - sourceFilename: res.items[0]?.sourceFilename, - groupKey: res.items[0]?.groupKey, - baseId: '', - displayId: res.items[0]?.displayId, - itemCount: res.items.length, - items: res.items, - }] - : []) + queuedTaskSummary.value = null + parsedGroups.value = res.items?.length + ? [{ + sourceFileKey: res.items[0]?.sourceFileKey, + sourceFilename: res.items[0]?.sourceFilename, + groupKey: `similar-asin:${res.taskId}`, + baseId: '', + displayId: res.items[0]?.displayId, + itemCount: res.items.length, + items: res.items, + }] + : [] queuePayloadText.value = '' - ElMessage.success(`解析完成，共 ${res.groupCount || 0} 组 / ${res.acceptedRows} 条`) + ElMessage.success(`解析完成，共 ${res.acceptedRows} 条`) } catch (e) { ElMessage.error(e instanceof Error ? e.message : '解析失败') } finally { @@ -341,7 +388,8 @@ async function parseFiles() { async function pushToPythonQueue() { const api = getPywebviewApi() - const taskId = parseResult.value?.taskId + const currentParseResult = parseResult.value + const taskId = currentParseResult?.taskId if (!taskId || !parsedRows.value.length) { ElMessage.warning('请先解析文件') return @@ -350,6 +398,10 @@ async function pushToPythonQueue() { ElMessage.error('当前环境未启用 pywebview enqueue_json') return } + if (!effectiveCozeApiKey()) { + ElMessage.warning('请先填写密钥') + return + } pushing.value = true try { const payload = { @@ -357,7 +409,8 @@ async function pushToPythonQueue() { ts: Date.now(), data: { taskId, - prompt: parseResult.value?.aiPrompt || effectiveAiPrompt(), + prompt: currentParseResult.aiPrompt || effectiveAiPrompt(), + api_key: effectiveCozeApiKey(), groups: parsedGroups.value.map((group) => ({ sourceFileKey: group.sourceFileKey || '', sourceFilename: group.sourceFilename || '', @@ -372,8 +425,10 @@ async function pushToPythonQueue() { id: row.displayId, asin: row.asin, country: row.country, + price: row.price || '', url: row.url || '', title: row.title || '', + target_urls: [], })), })), rows: parsedRows.value.map((row) => ({ @@ -384,18 +439,21 @@ async function pushToPythonQueue() { id: row.displayId, asin: row.asin, country: row.country, + price: row.price || '', url: row.url || '', title: row.title || '', + target_urls: [], })), }, } - queuePayloadText.value = JSON.stringify(payload, null, 2) + queuePayloadText.value = JSON.stringify(payloadForDisplay(payload), null, 2) await activateSimilarAsinTask(taskId) const result = await api.enqueue_json(payload) if (!result?.success) { ElMessage.error(result?.error || '推送失败') return } + queuedTaskSummary.value = toTaskSummary(currentParseResult) addPollingTask(taskId) clearParsedTask() await loadDashboard() @@ -721,9 +779,10 @@ onUnmounted(() => { .btn-run:disabled { opacity: .55; cursor: not-allowed; } .selected-files { margin-top: 14px; color: #999; font-size: 12px; word-break: break-all; } .selected-files span { display: block; margin: 4px 0; } -.prompt-card { margin-bottom: 18px; } +.prompt-card, .secret-card { margin-bottom: 18px; } .prompt-input { width: 100%; box-sizing: border-box; resize: vertical; min-height: 180px; padding: 10px 12px; border: 1px solid #333; border-radius: 8px; background: #202020; color: #d8d8d8; font-size: 12px; line-height: 1.6; outline: none; } -.prompt-input:focus { border-color: #3498db; } +.secret-input { width: 100%; box-sizing: border-box; height: 38px; padding: 0 12px; border: 1px solid #333; border-radius: 8px; background: #202020; color: #d8d8d8; font-size: 12px; outline: none; } +.prompt-input:focus, .secret-input:focus { border-color: #3498db; } .prompt-default-label { margin-top: 10px; color: #8d8d8d; font-size: 12px; } .prompt-preview { margin-top: 10px; padding: 12px; border: 1px solid #2a2a2a; border-radius: 8px; background: #202020; color: #9ea7b3; font-size: 12px; line-height: 1.6; white-space: pre-wrap; } .parse-card, .queue-payload { margin-top: 14px; padding: 12px; border: 1px solid #2a2a2a; border-radius: 8px; background: #202020; color: #b8c1cc; font-size: 12px; } diff --git a/frontend-vue/src/shared/api/java-modules.ts b/frontend-vue/src/shared/api/java-modules.ts index 07f685e..6ccd6f0 100644 --- a/frontend-vue/src/shared/api/java-modules.ts +++ b/frontend-vue/src/shared/api/java-modules.ts @@ -1531,15 +1531,16 @@ export interface AppearancePatentTaskBatchVo { missingTaskIds?: number[]; } -export function parseAppearancePatent(files: UploadedFileRef[], aiPrompt: string) { +export function parseAppearancePatent(files: UploadedFileRef[], aiPrompt: string, apiKey?: string) { return unwrapJavaResponse( post< JavaApiResponse, - { user_id: number; files: UploadedFileRef[]; ai_prompt: string } + { user_id: number; files: UploadedFileRef[]; ai_prompt: string; api_key?: string } >(`${JAVA_API_PREFIX}/appearance-patent/parse`, { user_id: getCurrentUserId(), files, ai_prompt: aiPrompt, + api_key: apiKey, }), ); } @@ -1615,6 +1616,7 @@ export interface SimilarAsinParsedRow { groupKey?: string; asin: string; country: string; + price?: string; url?: string; title?: string; } @@ -1694,15 +1696,16 @@ export interface SimilarAsinTaskBatchVo { missingTaskIds?: number[]; } -export function parseSimilarAsin(files: UploadedFileRef[], aiPrompt: string) { +export function parseSimilarAsin(files: UploadedFileRef[], aiPrompt: string, apiKey?: string) { return unwrapJavaResponse( post< JavaApiResponse, - { user_id: number; files: UploadedFileRef[]; ai_prompt: string } + { user_id: number; files: UploadedFileRef[]; ai_prompt: string; api_key?: string } >(`${JAVA_API_PREFIX}/similar-asin/parse`, { user_id: getCurrentUserId(), files, ai_prompt: aiPrompt, + api_key: apiKey, }), ); } diff --git a/new_web_source/appearance-patent.html b/new_web_source/appearance-patent.html index 351c1bb..1b11e12 100644 --- a/new_web_source/appearance-patent.html +++ b/new_web_source/appearance-patent.html @@ -1,18 +1,18 @@ - - - - - - 外观专利检测 + + + + + + 外观专利检测 - - + + - - + + - - -
- - + + +
+ + diff --git a/new_web_source/similar-asin.html b/new_web_source/similar-asin.html new file mode 100644 index 0000000..2e8f07c --- /dev/null +++ b/new_web_source/similar-asin.html @@ -0,0 +1,18 @@ + + + + + + 相似ASIN检测 + + + + + + + + + +
+ +