diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java b/backend-java/src/main/java/com/nanri/aiimage/common/util/GroupResultPropagator.java
similarity index 97%
rename from backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java
rename to backend-java/src/main/java/com/nanri/aiimage/common/util/GroupResultPropagator.java
index 0c76629b..5bea7ca8 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/common/util/GroupResultPropagator.java
@@ -10,7 +10,7 @@ import java.util.function.BiConsumer;
import java.util.function.Function;
/**
- * Coze 回流数据按 ID 分组传播工具。
+ * LLM 回流数据按 ID 分组传播工具。
*
*
业务背景:
* 解析行按 Excel 行顺序排列,ID 形如 "1"、"1_1"、"1_2"、"2"、"2_1"。
@@ -33,7 +33,7 @@ import java.util.function.Function;
*
使用方式:
*
* // 专利结论列:组内任一行结论命中 "已侵权" 或 "侵权",组内都改为该标准值
- * CozeGroupResultPropagator.propagateByGroup(
+ * GroupResultPropagator.propagateByGroup(
* receivedRows,
* AppearancePatentParsedRowVo::getDisplayId,
* row -> findResultRow(row, resultMap),
@@ -44,9 +44,9 @@ import java.util.function.Function;
* );
*
*/
-public final class CozeGroupResultPropagator {
+public final class GroupResultPropagator {
- private static final Logger log = LoggerFactory.getLogger(CozeGroupResultPropagator.class);
+ private static final Logger log = LoggerFactory.getLogger(GroupResultPropagator.class);
/**
* 否定前缀关键字。若当前值同时包含 standard 和这些关键字之一,则不视为命中。
@@ -54,7 +54,7 @@ public final class CozeGroupResultPropagator {
*/
private static final List NEGATIVE_KEYWORDS = Arrays.asList("没有", "无", "不", "未");
- private CozeGroupResultPropagator() {
+ private GroupResultPropagator() {
}
/**
diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/CapacityPlanProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/CapacityPlanProperties.java
index 9b6e443b..2e91a85d 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/config/CapacityPlanProperties.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/config/CapacityPlanProperties.java
@@ -19,7 +19,7 @@ public class CapacityPlanProperties {
/** 数据库连接池(Hikari maximum-pool-size)。 */
private int dbPoolMaxSize = 30;
- /** 外部 HTTP 客户端(Coze/品牌/紫鸟)连接池容量。 */
+ /** 外部 HTTP 客户端(LLM/品牌/紫鸟)连接池容量。 */
private int httpClientPoolMaxSize = 32;
/** RustFS/MinIO OkHttp 连接池容量。 */
diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java b/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java
index 45767368..8507b19f 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java
@@ -8,7 +8,7 @@ import java.time.Duration;
/**
* Task 77:外部 HTTP 客户端统一连接复用池。
- * Coze / 品牌检查 / 紫鸟三个外部客户端共用同一个 java.net.http.HttpClient
+ * LLM / 品牌检查 / 紫鸟三个外部客户端共用同一个 java.net.http.HttpClient
* (内置 keep-alive 连接池),避免各自新建短命客户端导致连接无法复用、
* 每次请求都重新建连。各客户端按自身超时创建独立的
* JdkClientHttpRequestFactory(共享底层连接池),RestClient 单例懒加载。
diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java b/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java
index ab022c9d..3b916461 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java
@@ -4,6 +4,7 @@ import jakarta.servlet.FilterChain;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
+import java.util.Locale;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
@@ -19,13 +20,15 @@ public class RequestTraceFilter extends OncePerRequestFilter {
private static final Logger log = LoggerFactory.getLogger(RequestTraceFilter.class);
+ private static final int MIN_REQUEST_BODY_CACHE_LIMIT_BYTES = 1024 * 1024;
+
private final InstanceMetadata instanceMetadata;
private final int requestBodyCacheLimitBytes;
public RequestTraceFilter(InstanceMetadata instanceMetadata,
- @Value("${aiimage.instance-routing.request-body-cache-limit-bytes:104857600}") int requestBodyCacheLimitBytes) {
+ @Value("${aiimage.instance-routing.request-body-cache-limit-bytes:1048576}") int requestBodyCacheLimitBytes) {
this.instanceMetadata = instanceMetadata;
- this.requestBodyCacheLimitBytes = Math.max(1024 * 1024, requestBodyCacheLimitBytes);
+ this.requestBodyCacheLimitBytes = Math.max(MIN_REQUEST_BODY_CACHE_LIMIT_BYTES, requestBodyCacheLimitBytes);
}
@Override
@@ -76,16 +79,23 @@ public class RequestTraceFilter extends OncePerRequestFilter {
}
}
- private static HttpServletRequest wrapRequestIfNeeded(HttpServletRequest request, int requestBodyCacheLimitBytes) {
+ static HttpServletRequest wrapRequestIfNeeded(HttpServletRequest request, int requestBodyCacheLimitBytes) {
if (request instanceof ContentCachingRequestWrapper) {
return request;
}
+ // multipart 不缓存:过滤器日志不读 body,缓存会整体复制上传流到内存
+ String contentType = request.getContentType();
+ if (contentType != null && contentType.trim().toLowerCase(Locale.ROOT).startsWith("multipart/")) {
+ return request;
+ }
String method = request.getMethod();
if ("POST".equalsIgnoreCase(method)
|| "PUT".equalsIgnoreCase(method)
|| "PATCH".equalsIgnoreCase(method)
|| "DELETE".equalsIgnoreCase(method)) {
- return new ContentCachingRequestWrapper(request, requestBodyCacheLimitBytes);
+ // 阈值统一钳到 1MB 下限:0/负数会令 ContentCachingRequestWrapper 构造抛错
+ return new ContentCachingRequestWrapper(
+ request, Math.max(MIN_REQUEST_BODY_CACHE_LIMIT_BYTES, requestBodyCacheLimitBytes));
}
return request;
}
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 186a7e19..6665e546 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
@@ -3,65 +3,38 @@ package com.nanri.aiimage.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
-import java.util.ArrayList;
-import java.util.List;
-
@Data
@ConfigurationProperties(prefix = "aiimage.similar-asin")
public class SimilarAsinProperties {
- private String cozeBaseUrl = "https://api.coze.cn";
- private String cozeWorkflowPath = "/v1/workflow/run";
- private String cozeWorkflowHistoryPath = "/v1/workflows/{workflow_id}/run_histories/{execute_id}";
- private String cozeWorkflowId = "7635328462404583478";
- private String cozeToken = "";
- private List cozeCredentials = new ArrayList<>();
- private int cozeCredentialStripeSize = 0;
/**
- * P0-1:单次提交 Coze 工作流的 row 数量。
+ * P0-1:单次提交 LLM 批次的 row 数量。
* 历史值 10,在含 puzzle 多图行的场景下频繁触发 720712008
* "node executed out of limit: 1000"。降到 3 以避免节点上限被打爆。
* 出现持续 720712008 时还会被 P1-1 滑窗自适应再降到 1。
- * 不影响 AppearancePatentProperties 的同名值。
*/
- private int cozeBatchSize = 3;
+ private int llmBatchSize = 3;
/**
- * img_switch=false 时单次提交 Coze 的 row 数。
- * 不走图片检测时工作流压力小,恢复到 10 行一批以提高吞吐;开启图片检测时仍使用 cozeBatchSize。
+ * img_switch=false 时单次提交 LLM 的 row 数。
+ * 不走图片检测时压力小,恢复到 10 行一批以提高吞吐;开启图片检测时仍使用 llmBatchSize。
*/
- private int cozeTextOnlyBatchSize = 10;
- private int cozeConnectTimeoutMillis = 10000;
- private int cozeReadTimeoutMillis = 60000;
- private int cozePollIntervalMillis = 30000;
- private int cozePollTimeoutMillis = 1800000;
+ private int llmTextOnlyBatchSize = 10;
private long dbTaskTouchIntervalMillis = 120000L;
private long dbJobTouchIntervalMillis = 60000L;
private int staleTimeoutMinutes = 30;
private String staleFinalizeCron = "0 */2 * * * *";
/**
- * 同一 credential 两次提交之间的最小间隔(毫秒)。
- * 历史值硬编码 30000(持锁 sleep),导致单凭证仅 2 batch/分钟。
- * 几千行任务场景下成为提交吞吐瓶颈,下调到 5000ms 并改为锁外冷却。
- * 出现 Coze 限流加重时可通过 AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MIN_INTERVAL_MILLIS 调高。
- */
- private long cozeSubmitMinIntervalMillis = 5000L;
-
- /**
- * 末尾零头 batch 的强制 flush 阈值(分钟):当不足 cozeBatchSize 的零头 row
+ * 末尾零头 batch 的强制 flush 阈值(分钟):当不足 llmBatchSize 的零头 row
* 长时间挂着(Python 慢回传)时触发提交。
* 任务级实测:345 行 / 4h 总耗时中,约 2-3 小时是 batch 永远凑不满 batchSize 在等下一波回传,
- * 把阈值从 15 调到 1:最多 60s 后 1-2 行也强制提交,让 Coze 提交侧持续进票,
- * 总耗时降到与 Python 回传节奏接近。配合 cozeBatchSize=3、cozeSubmitMinIntervalMillis=5000,
- * 实际不会触发 Coze 限流。出现限流加重再调回 5/10。
+ * 把阈值从 15 调到 1:最多 60s 后 1-2 行也强制提交,让 LLM 提交侧持续进票,
+ * 总耗时降到与 Python 回传节奏接近。
*/
- private int cozeFlushPendingMinutes = 1;
+ private int llmFlushPendingMinutes = 1;
/**
- * 同 batch retry + split retry 共享的最大重试次数。原硬编码 5。
+ * 图片下载、解码和缩放共享该池;4 核生产机默认 2,避免图片任务占满整机 CPU。
*/
- private int cozeSubmitMaxRetryCount = 5;
-
- /** 图片下载、解码和缩放共享该池;4 核生产机默认 2,避免图片任务占满整机 CPU。 */
private int imageDownloadPoolSize = 2;
/**
@@ -97,40 +70,11 @@ public class SimilarAsinProperties {
private boolean imageDbCacheEnabled = false;
/**
- * 是否在 Coze 请求 parameters 中附带 api_key 字段。
- * 默认 true:线上 Coze 工作流将该字段视为必填,缺失会得到 4000
- * "Missing required parameters";前端传入的 api_key 必须透传到 coze。
- * 仅在工作流明确不再需要 api_key 时,可通过环境变量
- * AIIMAGE_SIMILAR_ASIN_COZE_INCLUDE_LEGACY_API_KEY=false 关闭。
- */
- private boolean cozeIncludeLegacyApiKey = true;
-
- /**
- * 是否使用旧的 item 字段顺序 {asin, sku, url, target_urls, title}。
- * 默认 false:当前实现使用 {asin, url, target_urls, title, sku}。
- * 出现兼容问题时可通过 AIIMAGE_SIMILAR_ASIN_COZE_USE_LEGACY_ITEM_ORDER=true
- * 切回旧顺序进行回归对比。
- */
- private boolean cozeUseLegacyItemFieldOrder = false;
-
- /**
- * 是否启用 P0-3 merge 增量缓冲:每个 batch DONE 时仅缓冲 cozeRows,
+ * 是否启用 merge 增量缓冲:每个 batch DONE 时仅缓冲 llmRows,
* 不立即合并到 chunk;finalize 前一次性按 chunkScopeHash 分组合并,
* 把 OSS chunk 读写从 1000+ 次降到 chunk 数量级。
- * 仅作用于"正常 poll DONE"路径;失败 batch / 单 batch 任务 / 其他
- * 11 个 mergeCozeRowsIntoChunk 调用点保留原立即 merge 行为。
- * 出现问题时可通过 AIIMAGE_SIMILAR_ASIN_COZE_RESULT_BUFFER_ENABLED=false
- * 一键回滚到老路径。
*/
- private boolean cozeResultBufferEnabled = true;
-
- /**
- * P0-4:单 credential 抢 Coze 提交锁的最长等待时间(毫秒)。
- * 原硬编码 1000ms,在高并发 split retry 时大量抛 "Coze submit throttle lock timeout"
- * 并把整批行 markFailed。应与 cozeSubmitMinIntervalMillis(5000ms)保持 1.5-2 倍关系,
- * 默认 10000ms 给抢锁更多时间。
- */
- private long cozeSubmitLockWaitMillis = 10000L;
+ private boolean llmResultBufferEnabled = true;
/**
* 解析接口返回的预览行/预览组数量上限。
@@ -193,15 +137,9 @@ public class SimilarAsinProperties {
*/
private int imagePrefetchBudgetSeconds = 60;
- /**
- * P0-4:抢 Coze 提交锁失败后下次重试间隔(毫秒)。
- * 原硬编码 500ms,会在指数退避算法中作为基础值(500/1000/2000/4000ms 上限 4000)。
- */
- private long cozeSubmitLockRetryDelayMillis = 500L;
-
/**
* 货源查询直连 LLM 模式开关(默认 true:新任务与存量 PENDING 批次都走直连 LLM,
- * 不再经过 Coze)。false 时回退到原 Coze 工作流链路(轮询/重试状态机保留)。
+ * 不再经过工作流中转)。
*/
private boolean directLlmEnabled = true;
@@ -230,11 +168,4 @@ public class SimilarAsinProperties {
/** 拼接图/主图下载超时(秒),慢源图片较多时放大该值。 */
private int llmImageDownloadTimeoutSeconds = 10;
-
- @Data
- public static class CozeCredential {
- private String name;
- private String workflowId;
- private String token;
- }
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java
index 017e7189..dbd0acb0 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java
@@ -1,5 +1,34 @@
package com.nanri.aiimage.config;
+import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskService;
+import com.nanri.aiimage.modules.brand.service.BrandTaskService;
+import com.nanri.aiimage.modules.collectdata.service.CollectDataService;
+import com.nanri.aiimage.modules.deletebrand.service.DeleteBrandRunService;
+import com.nanri.aiimage.modules.patroldelete.service.PatrolDeleteTaskService;
+import com.nanri.aiimage.modules.pricetrack.service.PriceTrackTaskService;
+import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService;
+import com.nanri.aiimage.modules.publish.service.PublishTaskService;
+import com.nanri.aiimage.modules.queryasin.service.QueryAsinTaskService;
+import com.nanri.aiimage.modules.shopdatacrawl.service.ShopDataCrawlTaskService;
+import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService;
+import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskService;
+import com.nanri.aiimage.modules.task.service.AppearancePatentResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.BrandResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.CollectDataResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.DeleteBrandResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.PatrolDeleteResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.PriceTrackResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.ProductRiskResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.PublishResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.QueryAsinResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.ResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.ResultFileJobHandlerRegistry;
+import com.nanri.aiimage.modules.task.service.ShopDataCrawlResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.ShopMatchResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.SimilarAsinResultFileJobHandler;
+import com.nanri.aiimage.modules.task.service.TaskResultPayloadService;
+import com.nanri.aiimage.modules.task.service.WithdrawResultFileJobHandler;
+import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService;
import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
@@ -9,6 +38,8 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import java.util.List;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@@ -16,6 +47,13 @@ import java.util.concurrent.Semaphore;
@Configuration
public class TaskFileJobConfig {
+ /** 结果文件 Job 支持的全部 moduleType(启动校验枚举源,见 ResultFileJobHandlerRegistry.validateCoverage) */
+ public static final Set RESULT_FILE_JOB_MODULE_TYPES = Set.of(
+ "SHOP_MATCH", "PRICE_TRACK", "PRODUCT_RISK_RESOLVE",
+ "PUBLISH", "QUERY_ASIN", "SHOP_DATA_CRAWL", "WITHDRAW",
+ "PATROL_DELETE", "APPEARANCE_PATENT", "SIMILAR_ASIN",
+ "DELETE_BRAND", "BRAND", "COLLECT_DATA");
+
@Bean("taskFileJobDispatchExecutor")
public TaskExecutor taskFileJobDispatchExecutor(
@Value("${aiimage.result-file-job.local-dispatch-pool-size:2}") int poolSize,
@@ -33,24 +71,24 @@ public class TaskFileJobConfig {
}
@Bean(destroyMethod = "shutdown")
- public ExecutorService cozeVirtualThreadExecutor() {
+ public ExecutorService taskQueueVirtualThreadExecutor() {
return Executors.newThreadPerTaskExecutor(Thread.ofVirtual()
- .name("coze-task-", 0)
+ .name("task-queue-", 0)
.factory());
}
- @Bean("cozeTaskExecutor")
- public TaskExecutor cozeTaskExecutor(
- ExecutorService cozeVirtualThreadExecutor,
+ @Bean("taskQueueExecutor")
+ public TaskExecutor taskQueueExecutor(
+ ExecutorService taskQueueVirtualThreadExecutor,
@Value("${aiimage.coze-task.max-concurrent:12}") int maxConcurrent,
@Value("${aiimage.coze-task.max-waiting:1000}") int maxWaiting,
ObjectProvider meterRegistryProvider) {
Semaphore semaphore = new Semaphore(Math.max(1, maxConcurrent));
TaskExecutor semaphoreLimited = new ConcurrentTaskExecutor(command -> {
if (command == null) {
- throw new IllegalArgumentException("coze 任务不能为 null");
+ throw new IllegalArgumentException("task 不能为 null");
}
- cozeVirtualThreadExecutor.execute(() -> {
+ taskQueueVirtualThreadExecutor.execute(() -> {
boolean acquired = false;
try {
semaphore.acquire();
@@ -65,6 +103,87 @@ public class TaskFileJobConfig {
}
});
});
- return new CozeTaskQueueGate(semaphoreLimited, maxWaiting, meterRegistryProvider);
+ return new TaskQueueGate(semaphoreLimited, maxWaiting, meterRegistryProvider);
+ }
+
+ @Bean
+ public ResultFileJobHandlerRegistry resultFileJobHandlerRegistry(List handlers) {
+ ResultFileJobHandlerRegistry registry = new ResultFileJobHandlerRegistry(handlers);
+ registry.validateCoverage(RESULT_FILE_JOB_MODULE_TYPES);
+ return registry;
+ }
+
+ @Bean
+ public ResultFileJobHandler appearancePatentResultFileJobHandler(
+ AppearancePatentTaskService appearancePatentTaskService) {
+ return new AppearancePatentResultFileJobHandler(appearancePatentTaskService);
+ }
+
+ @Bean
+ public ResultFileJobHandler brandResultFileJobHandler(
+ BrandTaskService brandTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new BrandResultFileJobHandler(brandTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler collectDataResultFileJobHandler(CollectDataService collectDataService) {
+ return new CollectDataResultFileJobHandler(collectDataService);
+ }
+
+ @Bean
+ public ResultFileJobHandler deleteBrandResultFileJobHandler(DeleteBrandRunService deleteBrandRunService) {
+ return new DeleteBrandResultFileJobHandler(deleteBrandRunService);
+ }
+
+ @Bean
+ public ResultFileJobHandler patrolDeleteResultFileJobHandler(
+ PatrolDeleteTaskService patrolDeleteTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new PatrolDeleteResultFileJobHandler(patrolDeleteTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler priceTrackResultFileJobHandler(
+ PriceTrackTaskService priceTrackTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new PriceTrackResultFileJobHandler(priceTrackTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler productRiskResultFileJobHandler(
+ ProductRiskTaskService productRiskTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new ProductRiskResultFileJobHandler(productRiskTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler publishResultFileJobHandler(PublishTaskService publishTaskService) {
+ return new PublishResultFileJobHandler(publishTaskService);
+ }
+
+ @Bean
+ public ResultFileJobHandler queryAsinResultFileJobHandler(
+ QueryAsinTaskService queryAsinTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new QueryAsinResultFileJobHandler(queryAsinTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler shopDataCrawlResultFileJobHandler(
+ ShopDataCrawlTaskService shopDataCrawlTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new ShopDataCrawlResultFileJobHandler(shopDataCrawlTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler shopMatchResultFileJobHandler(
+ ShopMatchTaskService shopMatchTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new ShopMatchResultFileJobHandler(shopMatchTaskService, taskResultPayloadService);
+ }
+
+ @Bean
+ public ResultFileJobHandler similarAsinResultFileJobHandler(SimilarAsinTaskService similarAsinTaskService) {
+ return new SimilarAsinResultFileJobHandler(similarAsinTaskService);
+ }
+
+ @Bean
+ public ResultFileJobHandler withdrawResultFileJobHandler(
+ WithdrawTaskService withdrawTaskService, TaskResultPayloadService taskResultPayloadService) {
+ return new WithdrawResultFileJobHandler(withdrawTaskService, taskResultPayloadService);
}
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/CozeTaskQueueGate.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java
similarity index 87%
rename from backend-java/src/main/java/com/nanri/aiimage/config/CozeTaskQueueGate.java
rename to backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java
index 82bb7373..4342b1e0 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/config/CozeTaskQueueGate.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java
@@ -11,7 +11,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
- * Task 75:虚拟线程任务排队闸门。Coze 执行池的信号量只限制"正在执行"的
+ * Task 75:虚拟线程任务排队闸门。任务执行池的信号量只限制"正在执行"的
* 并发度,提交侧仍会在虚拟线程里无限排队。此闸门在提交时统计"已受理未启动"
* 的等待数,达到上限立即拒绝并记录指标,防止等待队列无界堆积:
*
@@ -21,14 +21,14 @@ import java.util.concurrent.atomic.AtomicInteger;
*
*/
@Slf4j
-public class CozeTaskQueueGate implements TaskExecutor {
+public class TaskQueueGate implements TaskExecutor {
private final TaskExecutor delegate;
private final int maxWaiting;
private final ObjectProvider meterRegistryProvider;
private final AtomicInteger waiting = new AtomicInteger();
- public CozeTaskQueueGate(TaskExecutor delegate, int maxWaiting,
+ public TaskQueueGate(TaskExecutor delegate, int maxWaiting,
ObjectProvider meterRegistryProvider) {
this.delegate = delegate;
this.maxWaiting = Math.max(1, maxWaiting);
@@ -43,13 +43,13 @@ public class CozeTaskQueueGate implements TaskExecutor {
public void execute(Runnable command) {
if (command == null) {
recordRejected("invalid-input");
- throw new IllegalArgumentException("coze 任务不能为 null");
+ throw new IllegalArgumentException("task 不能为 null");
}
if (waiting.get() >= maxWaiting) {
recordRejected("queue-full");
- log.warn("[coze-task][gate] waiting queue full, reject submit waiting={} limit={}",
+ log.warn("[task-queue][gate] waiting queue full, reject submit waiting={} limit={}",
waiting.get(), maxWaiting);
- throw new TaskRejectedException("coze 等待队列已满,limit=" + maxWaiting
+ throw new TaskRejectedException("task 等待队列已满,limit=" + maxWaiting
+ ", waiting=" + waiting.get());
}
waiting.incrementAndGet();
@@ -68,7 +68,7 @@ public class CozeTaskQueueGate implements TaskExecutor {
waiting.decrementAndGet();
recordQueueWait(System.nanoTime() - queuedAt);
recordRejected("delegate-rejected");
- log.warn("[coze-task][gate] delegate rejected submit waiting={} limit={} msg={}",
+ log.warn("[task-queue][gate] delegate rejected submit waiting={} limit={} msg={}",
waiting.get(), maxWaiting, ex.getMessage(), ex);
throw ex;
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/metrics/ExternalCallMetricsRecorder.java b/backend-java/src/main/java/com/nanri/aiimage/metrics/ExternalCallMetricsRecorder.java
index 22ca3e36..24fe0f3e 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/metrics/ExternalCallMetricsRecorder.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/metrics/ExternalCallMetricsRecorder.java
@@ -13,13 +13,12 @@ import java.util.concurrent.TimeUnit;
/**
* Task 78:外部调用统一指标记录器。
- * 所有外部 HTTP 客户端(Coze / 品牌检查 / 紫鸟)在构建 RestClient 时挂载
+ * 所有外部 HTTP 客户端(LLM / 品牌检查 / 紫鸟)在构建 RestClient 时挂载
* {@link #interceptor(String)} 拦截器,统一记录:
*
* - 耗时:{@code aiimage.external-call.duration}(client + result 标签);
* - 失败率:{@code aiimage.external-call.total}(result=success/failure,2xx 之外计失败);
- * - payload 字节:{@code aiimage.external-call.payload.bytes}(请求体字节数);
- * - 重试次数:{@code aiimage.external-call.retry.total}(客户端重试循环内调用)。
+ * - payload 字节:{@code aiimage.external-call.payload.bytes}(请求体字节数)。
*
* 指标注册表通过 ObjectProvider 懒获取,未配置 Micrometer 时全部静默跳过,
* 不改变既有调用语义。
@@ -77,14 +76,6 @@ public class ExternalCallMetricsRecorder {
};
}
- /** 重试循环内每次进入下一次尝试前调用。 */
- public void recordRetry(String client) {
- MeterRegistry registry = meterRegistry();
- if (registry != null) {
- registry.counter("aiimage.external-call.retry.total", "client", client).increment();
- }
- }
-
private void record(String client, String result, long startedAt, long payloadBytes) {
MeterRegistry registry = meterRegistry();
if (registry == null) {
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/AppearancePatentLlmClient.java
similarity index 96%
rename from backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java
rename to backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentLlmClient.java
index 9e458949..ce5ed5b4 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/AppearancePatentLlmClient.java
@@ -30,12 +30,12 @@ import java.util.regex.Pattern;
/**
* 外观专利检测:直连 LLM(OpenAI 兼容 /v1/chat/completions),
- * 每行并发跑"商标关键词提取"与"外观侵权检测"两个请求,结果语义与原 Coze 工作流对齐。
+ * 每行并发跑"商标关键词提取"与"外观侵权检测"两个请求,结果语义与原 LLM 工作流对齐。
*/
@Component
@RequiredArgsConstructor
@Slf4j
-public class AppearancePatentCozeClient {
+public class AppearancePatentLlmClient {
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
private static final String INFRINGEMENT = "侵权";
@@ -44,7 +44,7 @@ public class AppearancePatentCozeClient {
private static final String APPEARANCE_ANOMALY = "外观识别异常";
private static final String MISSING_ROW_DATA = "爬虫数据缺失";
- /** 商标关键词提取系统提示词(与原 Coze 工作流一致) */
+ /** 商标关键词提取系统提示词(与原 LLM 工作流一致) */
private static final String TITLE_SYSTEM_PROMPT =
"你是品牌词提取工具。从用户输入的商品标题中,提取实际出现的品牌、商标、企业、平台名称及违规关键词。\n\n"
+ "硬性约束:\n"
@@ -57,7 +57,7 @@ public class AppearancePatentCozeClient {
+ "* **兜底输出**:如果文本中确实没有任何商标、品牌或违规词,直接输出 `\"无\"`。\n"
+ "* **零干扰输出**:**绝对不要**包含任何引言、解释、前缀、多余的空格或标点符号。";
- /** 外观专利检测系统提示词(与原 Coze 工作流一致) */
+ /** 外观专利检测系统提示词(与原 LLM 工作流一致) */
private static final String APPEARANCE_SYSTEM_PROMPT =
"# 角色定义\n"
+ "你是一位极其严谨的跨境电商知识产权(IP)律师兼视觉侵权鉴定专家。你的任务是基于用户提供的产品图片和描述,评估该产品在亚马逊等平台销售时的**外观设计(Design Patent)侵权风险**。\n\n"
@@ -158,7 +158,7 @@ public class AppearancePatentCozeClient {
}
/**
- * 单行检测:与原 Coze 工作流语义对齐。
+ * 单行检测:与原 LLM 工作流语义对齐。
* 任一 LLM 失败或数据缺失时走工作流"默认值"分支:appearance=外观识别异常、
* title 保留原始标题、title_reason/appearance_reason 填充对应错误信息。
*/
@@ -197,7 +197,7 @@ public class AppearancePatentCozeClient {
String titleReason = titleFailed ? titleError : firstNonBlank(rawTitle, MISSING_ROW_DATA);
return applyRowFallback(resultRow, rawTitle, titleReason, appearanceFailed ? appearanceReason : "");
}
- CozeResult result = new CozeResult(
+ LlmResult result = new LlmResult(
row.getGroupKey(),
row.getId(),
row.getAsin(),
@@ -219,7 +219,7 @@ public class AppearancePatentCozeClient {
String title,
String titleReason,
String appearanceReason) {
- CozeResult result = new CozeResult(
+ LlmResult result = new LlmResult(
row.getGroupKey(),
row.getId(),
row.getAsin(),
@@ -475,7 +475,7 @@ public class AppearancePatentCozeClient {
}
}
- private List parseResults(String raw) throws Exception {
+ private List parseResults(String raw) throws Exception {
JsonNode root = objectMapper.readTree(raw);
String dataText = extractResultDataText(root);
if (dataText.isBlank()) {
@@ -483,11 +483,11 @@ public class AppearancePatentCozeClient {
}
JsonNode dataRoot = objectMapper.readTree(dataText);
JsonNode array = dataRoot.isArray() ? dataRoot : dataRoot.path("data");
- List results = new ArrayList<>();
+ List results = new ArrayList<>();
if (array.isArray()) {
for (JsonNode node : array) {
JsonNode itemNode = resultItemNode(node);
- results.add(new CozeResult(
+ results.add(new LlmResult(
text(firstNonNull(
firstNonNull(node.get("group_key"), node.get("groupKey")),
firstNonNull(itemNode.get("group_key"), itemNode.get("groupKey")))),
@@ -543,13 +543,13 @@ public class AppearancePatentCozeClient {
return item;
}
- private List mergeRows(List rows, List results) {
- Map resultByGroupKey = new LinkedHashMap<>();
- Map resultByCompositeKey = new LinkedHashMap<>();
- Map resultByAsinCountry = new LinkedHashMap<>();
- Map resultByAsin = new LinkedHashMap<>();
- Map resultByRowId = new LinkedHashMap<>();
- for (CozeResult result : results) {
+ private List mergeRows(List rows, List results) {
+ Map resultByGroupKey = new LinkedHashMap<>();
+ Map resultByCompositeKey = new LinkedHashMap<>();
+ Map resultByAsinCountry = new LinkedHashMap<>();
+ Map resultByAsin = new LinkedHashMap<>();
+ Map resultByRowId = new LinkedHashMap<>();
+ for (LlmResult result : results) {
String groupKey = normalize(result.groupKey());
if (!groupKey.isBlank()) {
resultByGroupKey.putIfAbsent(groupKey, result);
@@ -576,7 +576,7 @@ public class AppearancePatentCozeClient {
boolean allowIndexFallback = results.size() == rows.size() && results.stream().noneMatch(this::hasIdentity);
for (int i = 0; i < rows.size(); i++) {
AppearancePatentResultRowDto row = copy(rows.get(i));
- CozeResult result = resultByGroupKey.get(normalize(row.getGroupKey()));
+ LlmResult result = resultByGroupKey.get(normalize(row.getGroupKey()));
if (result == null) {
result = resultByCompositeKey.get(rowKey(row.getId(), row.getAsin(), row.getCountry()));
}
@@ -606,7 +606,7 @@ public class AppearancePatentCozeClient {
return normalizedAsin + "::" + normalize(country);
}
- private boolean hasIdentity(CozeResult result) {
+ private boolean hasIdentity(LlmResult result) {
if (result == null) {
return false;
}
@@ -615,7 +615,7 @@ public class AppearancePatentCozeClient {
|| !normalize(result.asin()).isBlank();
}
- void applyResult(AppearancePatentResultRowDto row, CozeResult result) {
+ void applyResult(AppearancePatentResultRowDto row, LlmResult result) {
if (row == null || result == null) {
return;
}
@@ -637,21 +637,21 @@ public class AppearancePatentCozeClient {
row.setScore(result.score());
}
- private boolean isNoTitleRisk(CozeResult result) {
+ private boolean isNoTitleRisk(LlmResult result) {
return result != null
&& "无".equals(normalize(result.title()));
}
- private boolean shouldCheckBrand(CozeResult result) {
+ private boolean shouldCheckBrand(LlmResult result) {
return result != null
&& !normalize(result.title()).isBlank()
&& !normalize(result.appearance()).isBlank();
}
- String buildTitleRisk(String cozeTitle, BrandCheckClient.BrandCheckBatchResult brandCheck) {
+ String buildTitleRisk(String llmTitle, BrandCheckClient.BrandCheckBatchResult brandCheck) {
List brands = brandCheck == null ? List.of() : brandCheck.brands();
if (brands == null || brands.isEmpty()) {
- return firstNonBlank(cozeTitle, "");
+ return firstNonBlank(llmTitle, "");
}
if (brandCheck.hasFailedData()) {
return INFRINGEMENT;
@@ -922,7 +922,7 @@ public class AppearancePatentCozeClient {
return node == null || node.isNull() ? null : node.asText();
}
- private record CozeResult(
+ private record LlmResult(
String groupKey,
String rowId,
String asin,
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentResultRowDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentResultRowDto.java
index 632a99cb..159f0dbc 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentResultRowDto.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/dto/AppearancePatentResultRowDto.java
@@ -20,7 +20,7 @@ public class AppearancePatentResultRowDto {
@Schema(description = "主数据分组 key。Python 应从解析结果原样透传,用于把同组子行补回。", example = "uploads/20260426/appearance_patent_17.xlsx::2@2")
private String groupKey;
- @Schema(description = "Excel 中的 id。代表行通常是整数 id 或 n_1,例如 2_1;最终生成 xlsx 时,2_2、2_3 会复用同组 2_1 的 Coze 检测结果。", example = "2_1")
+ @Schema(description = "Excel 中的 id。代表行通常是整数 id 或 n_1,例如 2_1;最终生成 xlsx 时,2_2、2_3 会复用同组 2_1 的 LLM 检测结果。", example = "2_1")
private String id;
@Schema(description = "亚马逊 ASIN。后端会统一按大写处理和匹配。", example = "B0CJ8SNXXV")
@@ -72,26 +72,26 @@ public class AppearancePatentResultRowDto {
private String appearanceRisk;
@JsonAlias({"patent ", "patent"})
- @Schema(description = "Java 调用 Coze 后生成的专利维度检测结果,对应最终 xlsx 的“专利维度(发明/实用新型专利)”列。兼容 Coze 返回字段 patent 和 patent 后带空格的情况;Python 回传请求中不要传该字段。", example = "未发现明显发明或实用新型专利侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
+ @Schema(description = "Java 调用 LLM 后生成的专利维度检测结果,对应最终 xlsx 的“专利维度(发明/实用新型专利)”列。兼容 LLM 返回字段 patent 和 patent 后带空格的情况;Python 回传请求中不要传该字段。", example = "未发现明显发明或实用新型专利侵权风险。", accessMode = Schema.AccessMode.READ_ONLY)
private String patentRisk;
- @Schema(description = "Java 调用 Coze 后生成的最终结论,对应最终 xlsx 的“结论”列。Python 回传请求中不要传该字段。", example = "未发现明显侵权风险", accessMode = Schema.AccessMode.READ_ONLY)
+ @Schema(description = "Java 调用 LLM 后生成的最终结论,对应最终 xlsx 的“结论”列。Python 回传请求中不要传该字段。", example = "未发现明显侵权风险", accessMode = Schema.AccessMode.READ_ONLY)
private String conclusion;
@JsonAlias({"title_reason", "titleReason"})
- @Schema(description = "Coze 返回的标题维度原因", accessMode = Schema.AccessMode.READ_ONLY)
+ @Schema(description = "LLM 返回的标题维度原因", accessMode = Schema.AccessMode.READ_ONLY)
private String titleReason;
@JsonAlias({"appearance_reason", "appearanceReason"})
- @Schema(description = "Coze 返回的外观维度原因", accessMode = Schema.AccessMode.READ_ONLY)
+ @Schema(description = "LLM 返回的外观维度原因", accessMode = Schema.AccessMode.READ_ONLY)
private String appearanceReason;
@JsonAlias({"patent_reason", "patentReason", "patent reason"})
- @Schema(description = "Coze 返回的专利维度原因", accessMode = Schema.AccessMode.READ_ONLY)
+ @Schema(description = "LLM 返回的专利维度原因", accessMode = Schema.AccessMode.READ_ONLY)
private String patentReason;
@JsonAlias({"score", "Score", "评分"})
- @Schema(description = "Coze 回流的评分(外观维度),仅保留回流值,不写入最终 xlsx。", accessMode = Schema.AccessMode.READ_ONLY)
+ @Schema(description = "LLM 回流的评分(外观维度),仅保留回流值,不写入最终 xlsx。", accessMode = Schema.AccessMode.READ_ONLY)
private String score;
/**
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentParsedRowVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentParsedRowVo.java
index 87e7c276..581f930a 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentParsedRowVo.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentParsedRowVo.java
@@ -39,10 +39,10 @@ public class AppearancePatentParsedRowVo {
@Schema(description = "价格。", example = "12.99")
private String price;
- @Schema(description = "商品 SKU。Java 调用 Coze 时会放入 items[].sku。", example = "SKU-001")
+ @Schema(description = "商品 SKU。Java 调用 LLM 时会放入 items[].sku。", example = "SKU-001")
private String sku;
- @Schema(description = "商品图片 URL 或商品 URL,供 Coze 检测使用。", example = "https://webstatic.aiproxy.vip/output/demo.jpg")
+ @Schema(description = "商品图片 URL 或商品 URL,供 LLM 检测使用。", example = "https://webstatic.aiproxy.vip/output/demo.jpg")
private String url;
@Schema(description = "商品标题。", example = "Women Floral Dress Summer Casual")
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskCacheService.java
index bfb77d50..b7a46aec 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskCacheService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskCacheService.java
@@ -146,7 +146,7 @@ public class AppearancePatentTaskCacheService {
return Boolean.TRUE.equals(first);
} catch (Exception ex) {
log.warn("[appearance-patent-cache] mark row processed degraded taskId={} rowKey={} msg={}", taskId, rowKey, ex.getMessage());
- // Redis 降级时放行,由持久化层的 hasResolvedCozeFields 判据兜底去重。
+ // Redis 降级时放行,由持久化层的 hasResolvedLlmFields 判据兜底去重。
return true;
}
}
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 014f0318..a34c7fa8 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,12 +9,12 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.exception.TaskOwnerMismatchException;
import com.nanri.aiimage.common.service.DistributedJobLockService;
-import com.nanri.aiimage.common.util.CozeGroupResultPropagator;
+import com.nanri.aiimage.common.util.GroupResultPropagator;
import com.nanri.aiimage.common.util.FailedStatusRowFilter;
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.client.AppearancePatentLlmClient;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParseRequest;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedGroupPageDto;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedPayloadDto;
@@ -128,7 +128,7 @@ public class AppearancePatentTaskService {
private final TaskScopeStateMapper taskScopeStateMapper;
private final TaskChunkMapper taskChunkMapper;
private final ObjectMapper objectMapper;
- private final AppearancePatentCozeClient cozeClient;
+ private final AppearancePatentLlmClient llmClient;
private final AppearancePatentTaskCacheService taskCacheService;
private final AppearancePatentProperties properties;
private final TaskFileJobService taskFileJobService;
@@ -413,7 +413,7 @@ public class AppearancePatentTaskService {
completeSubmittedChunk(context);
return null;
});
- scheduleCozePipelineForSubmittedChunk(context);
+ scheduleLlmPipelineForSubmittedChunk(context);
return;
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
@@ -484,7 +484,7 @@ public class AppearancePatentTaskService {
scope.setLastError(request.getError());
scope.setCompleted(Boolean.TRUE.equals(request.getDone()) ? 1 : 0);
scope.setUpdatedAt(LocalDateTime.now());
- scope.setStateJson("{\"phase\":\"RECEIVED\",\"coze\":\"PENDING\"}");
+ scope.setStateJson("{\"phase\":\"RECEIVED\",\"llm\":\"PENDING\"}");
if (scope.getId() == null) {
taskScopeStateMapper.insert(scope);
} else {
@@ -499,7 +499,7 @@ public class AppearancePatentTaskService {
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
}
- scheduleCozePipelineForSubmittedChunk(context);
+ scheduleLlmPipelineForSubmittedChunk(context);
}
@Transactional
@@ -758,12 +758,12 @@ public class AppearancePatentTaskService {
long startedAt = System.currentTimeMillis();
List llmRows;
try {
- llmRows = cozeClient.inspectRows(batchRows, prompt, apiKey);
+ llmRows = llmClient.inspectRows(batchRows, prompt, apiKey);
} catch (Exception ex) {
String message = firstNonBlank(ex.getMessage(), "LLM 检测失败");
log.warn("[appearance-patent] llm batch failed taskId={} jobId={} rows={} batch={}/{} err={}",
task.getId(), job.getId(), batchRows.size(), batchIndex, batchTotal, message);
- llmRows = cozeClient.markRowsFailed(batchRows, message);
+ llmRows = llmClient.markRowsFailed(batchRows, message);
}
mergeLlmRowsIntoChunks(task, allRowsByBaseId, llmRows, batchRows);
log.info("[appearance-patent] llm batch done taskId={} jobId={} rows={} batch={}/{} costMs={}",
@@ -794,11 +794,11 @@ public class AppearancePatentTaskService {
deduped.put(key, row);
}
}
- mergeCozeRowsIntoSubmittedChunks(task, new ArrayList<>(deduped.values()), allRowsByBaseId);
+ mergeLlmRowsIntoSubmittedChunks(task, new ArrayList<>(deduped.values()), allRowsByBaseId);
}
/**
- * 收集未检测的候选行(行级 Redis 去重 + 已解析字段过滤),与原 Coze 语义一致。
+ * 收集未检测的候选行(行级 Redis 去重 + 已解析字段过滤),与原 LLM 语义一致。
*/
private List loadSubmittedChunks(Long taskId) {
if (taskId == null || taskId <= 0) {
@@ -823,9 +823,9 @@ public class AppearancePatentTaskService {
if (persistedRows.isEmpty()) {
continue;
}
- for (AppearancePatentResultRowDto row : pickGroupRepresentativesForCoze(persistedRows.values())) {
+ for (AppearancePatentResultRowDto row : pickGroupRepresentativesForLlm(persistedRows.values())) {
String key = rowKey(row);
- if (key.isBlank() || hasResolvedCozeFields(row)) {
+ if (key.isBlank() || hasResolvedLlmFields(row)) {
continue;
}
if (taskCacheService.isRowProcessed(chunk.getTaskId(), key)) {
@@ -848,7 +848,7 @@ public class AppearancePatentTaskService {
if (persistedRows.isEmpty()) {
continue;
}
- int unresolved = pickGroupRepresentativesForCoze(persistedRows.values()).size();
+ int unresolved = pickGroupRepresentativesForLlm(persistedRows.values()).size();
if (unresolved > 0) {
total += Math.max(1, (unresolved + batchSize - 1) / batchSize);
}
@@ -860,7 +860,7 @@ public class AppearancePatentTaskService {
return Math.max(1, properties.getFlushPendingMinutes()) * 60_000L;
}
- private void submitCozeForSubmittedChunk(SubmitContext context) {
+ private void submitLlmForSubmittedChunk(SubmitContext context) {
if (context == null || context.task() == null || context.task().getId() == null) {
return;
}
@@ -898,7 +898,7 @@ public class AppearancePatentTaskService {
touchJavaSideTaskActivity(task.getId());
}
- private void scheduleCozePipelineForSubmittedChunk(SubmitContext context) {
+ private void scheduleLlmPipelineForSubmittedChunk(SubmitContext context) {
if (context == null || context.task() == null || context.task().getId() == null) {
return;
}
@@ -919,7 +919,7 @@ public class AppearancePatentTaskService {
if (job == null || "SUCCESS".equals(job.getStatus())) {
return;
}
- taskFileJobService.requeue(job.getId(), "Appearance patent result uploaded, scheduling Coze/file assembly");
+ taskFileJobService.requeue(job.getId(), "Appearance patent result uploaded, scheduling LLM/file assembly");
touchJavaSideTaskActivity(task.getId());
}
@@ -961,7 +961,7 @@ public class AppearancePatentTaskService {
log.warn("[appearance-patent] Python 超时恢复继续推进检测/文件收尾 taskId={} activeAssembleJobs={}",
taskId, activeAssembleJobs);
}
- submitCozeForSubmittedChunk(new SubmitContext(task, null, null, 0, true, null));
+ submitLlmForSubmittedChunk(new SubmitContext(task, null, null, 0, true, null));
touchJavaSideTaskActivity(taskId);
return true;
}
@@ -973,7 +973,7 @@ public class AppearancePatentTaskService {
List inputStates = taskScopeStateMapper.selectList(new LambdaQueryWrapper()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
- .isNull(TaskScopeStateEntity::getCozeStatus)
+ .isNull(TaskScopeStateEntity::getLlmStatus)
.orderByDesc(TaskScopeStateEntity::getUpdatedAt));
if (inputStates == null || inputStates.isEmpty()) {
TaskChunkEntity latestChunk = taskChunkMapper.selectOne(new LambdaQueryWrapper()
@@ -1007,7 +1007,7 @@ public class AppearancePatentTaskService {
state.setLastChunkAt(now);
}
state.setUpdatedAt(now);
- state.setStateJson("{\"phase\":\"RECEIVED\",\"coze\":\"PENDING\"}");
+ state.setStateJson("{\"phase\":\"RECEIVED\",\"llm\":\"PENDING\"}");
taskScopeStateMapper.updateById(state);
updated++;
}
@@ -1020,7 +1020,7 @@ public class AppearancePatentTaskService {
Integer chunkTotal,
String error,
boolean completed,
- boolean cozeDone) {
+ boolean llmDone) {
TaskScopeStateEntity scope = taskScopeStateMapper.selectOne(new LambdaQueryWrapper()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
@@ -1043,14 +1043,14 @@ public class AppearancePatentTaskService {
scope.setLastError(error);
scope.setCompleted(completed ? 1 : 0);
scope.setUpdatedAt(now);
- scope.setStateJson(cozeDone
- ? "{\"phase\":\"RECEIVED\",\"coze\":\"DONE\"}"
- : "{\"phase\":\"RECEIVED\",\"coze\":\"PENDING\"}");
+ scope.setStateJson(llmDone
+ ? "{\"phase\":\"RECEIVED\",\"llm\":\"DONE\"}"
+ : "{\"phase\":\"RECEIVED\",\"llm\":\"PENDING\"}");
if (scope.getId() == null) {
try {
taskScopeStateMapper.insert(scope);
- log.info("[appearance-patent] scope state inserted taskId={} scope={} scopeHash={} completed={} cozeDone={}",
- taskId, scopeKey, scopeHash, completed, cozeDone);
+ log.info("[appearance-patent] scope state inserted taskId={} scope={} scopeHash={} completed={} llmDone={}",
+ taskId, scopeKey, scopeHash, completed, llmDone);
return;
} catch (DuplicateKeyException ex) {
log.info("[appearance-patent] duplicate scope state inserted concurrently taskId={} scope={}", taskId, scopeKey);
@@ -1072,14 +1072,14 @@ public class AppearancePatentTaskService {
scope.setLastError(error);
scope.setCompleted(completed ? 1 : 0);
scope.setUpdatedAt(now);
- scope.setStateJson(cozeDone
- ? "{\"phase\":\"RECEIVED\",\"coze\":\"DONE\"}"
- : "{\"phase\":\"RECEIVED\",\"coze\":\"PENDING\"}");
+ scope.setStateJson(llmDone
+ ? "{\"phase\":\"RECEIVED\",\"llm\":\"DONE\"}"
+ : "{\"phase\":\"RECEIVED\",\"llm\":\"PENDING\"}");
}
}
taskScopeStateMapper.updateById(scope);
- log.info("[appearance-patent] scope state updated taskId={} scope={} scopeHash={} completed={} cozeDone={}",
- taskId, scopeKey, scopeHash, completed, cozeDone);
+ log.info("[appearance-patent] scope state updated taskId={} scope={} scopeHash={} completed={} llmDone={}",
+ taskId, scopeKey, scopeHash, completed, llmDone);
}
private T inNewTransaction(Supplier action) {
@@ -1209,7 +1209,7 @@ public class AppearancePatentTaskService {
}
}
- private List pickGroupRepresentativesForCoze(java.util.Collection rows) {
+ private List pickGroupRepresentativesForLlm(java.util.Collection rows) {
Map> groupedRows = new LinkedHashMap<>();
if (rows == null) {
return List.of();
@@ -1223,7 +1223,7 @@ public class AppearancePatentTaskService {
}
List representatives = new ArrayList<>();
for (List siblings : groupedRows.values()) {
- boolean alreadyResolved = siblings.stream().anyMatch(this::hasCompleteCozeResult);
+ boolean alreadyResolved = siblings.stream().anyMatch(this::hasCompleteLlmResult);
if (alreadyResolved) {
continue;
}
@@ -1525,11 +1525,11 @@ public class AppearancePatentTaskService {
saveFileBuildProgress(task, job, totalProgressUnits, Math.max(1, plannedLlmUnits), "等待 Python 继续回传数据");
return false;
}
- completeCozeFileJob(task, result, job, totalProgressUnits);
+ completeLlmFileJob(task, result, job, totalProgressUnits);
return true;
}
- private void completeCozeFileJob(FileTaskEntity task,
+ private void completeLlmFileJob(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,
int totalProgressUnits) {
@@ -1552,18 +1552,18 @@ public class AppearancePatentTaskService {
saveFileBuildProgress(task, job, totalProgressUnits, totalProgressUnits, "结果文件已生成");
}
- private void mergeCozeRowsIntoSubmittedChunks(FileTaskEntity task,
- List cozeRows,
+ private void mergeLlmRowsIntoSubmittedChunks(FileTaskEntity task,
+ List llmRows,
Map> allRowsByBaseId) {
- mergeCozeRowsIntoSubmittedChunks(task, cozeRows, allRowsByBaseId, null, null);
+ mergeLlmRowsIntoSubmittedChunks(task, llmRows, allRowsByBaseId, null, null);
}
- private void mergeCozeRowsIntoSubmittedChunks(FileTaskEntity task,
- List cozeRows,
+ private void mergeLlmRowsIntoSubmittedChunks(FileTaskEntity task,
+ List llmRows,
Map> allRowsByBaseId,
String fallbackScopeHash,
Integer fallbackChunkIndex) {
- if (task == null || cozeRows == null || cozeRows.isEmpty()) {
+ if (task == null || llmRows == null || llmRows.isEmpty()) {
return;
}
List chunks = loadSubmittedChunks(task.getId());
@@ -1578,7 +1578,7 @@ public class AppearancePatentTaskService {
chunkByKey.put(chunkKey, chunk);
}
Map> mergeRowsByChunk = new LinkedHashMap<>();
- for (AppearancePatentResultRowDto resultRow : cozeRows) {
+ for (AppearancePatentResultRowDto resultRow : llmRows) {
for (AppearancePatentResultRowDto expandedRow : expandRows(List.of(resultRow), allRowsByBaseId)) {
String rowKey = rowKey(expandedRow);
if (rowKey.isBlank()) {
@@ -1601,7 +1601,7 @@ public class AppearancePatentTaskService {
}
}
if (!matched) {
- log.warn("[appearance-patent] coze row has no submitted chunk taskId={} rowKey={}",
+ log.warn("[appearance-patent] llm row has no submitted chunk taskId={} rowKey={}",
task.getId(), rowKey);
}
}
@@ -1626,7 +1626,7 @@ public class AppearancePatentTaskService {
Long count = taskScopeStateMapper.selectCount(new LambdaQueryWrapper()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
- .isNull(TaskScopeStateEntity::getCozeStatus)
+ .isNull(TaskScopeStateEntity::getLlmStatus)
.isNotNull(TaskScopeStateEntity::getLastChunkAt)
.eq(TaskScopeStateEntity::getCompleted, 1));
return count != null && count > 0;
@@ -1839,8 +1839,8 @@ public class AppearancePatentTaskService {
reasonRows += resultMap.values().stream()
.filter(this::hasReasonFields)
.count();
- validateCompleteCozeCoverage(task.getId(), receivedRows, resultMap);
- conclusionPropagated += CozeGroupResultPropagator.propagateByGroup(
+ validateCompleteLlmCoverage(task.getId(), receivedRows, resultMap);
+ conclusionPropagated += GroupResultPropagator.propagateByGroup(
receivedRows,
AppearancePatentParsedRowVo::getDisplayId,
row -> findResultRow(row, resultMap),
@@ -2020,7 +2020,7 @@ public class AppearancePatentTaskService {
return receivedRows;
}
- private void validateCompleteCozeCoverage(Long taskId,
+ private void validateCompleteLlmCoverage(Long taskId,
List receivedRows,
Map resultMap) {
if (receivedRows == null || receivedRows.isEmpty()) {
@@ -2039,26 +2039,26 @@ public class AppearancePatentTaskService {
}
expectedRows++;
AppearancePatentResultRowDto resultRow = findResultRow(parsedRow, resultMap);
- if (resultRow == null || !hasCompleteCozeResult(resultRow)) {
+ if (resultRow == null || !hasCompleteLlmResult(resultRow)) {
missingRows++;
if (sampleAsins.size() < 5) {
sampleAsins.add(firstNonBlank(parsedRow.getAsin(), firstNonBlank(parsedRow.getDisplayId(), "")));
}
}
}
- boolean enforceCompleteCozeCoverage = false;
- if (enforceCompleteCozeCoverage && expectedRows > 0 && missingRows > 0) {
- log.warn("[appearance-patent] incomplete coze coverage taskId={} expectedRows={} missingRows={} samples={}",
+ boolean enforceCompleteLlmCoverage = false;
+ if (enforceCompleteLlmCoverage && expectedRows > 0 && missingRows > 0) {
+ log.warn("[appearance-patent] incomplete llm coverage taskId={} expectedRows={} missingRows={} samples={}",
taskId, expectedRows, missingRows, sampleAsins);
- throw new BusinessException("Coze 结果不完整:缺少 " + missingRows + "/" + expectedRows + " 条检测结果,请等待重试或重新运行任务");
+ throw new BusinessException("LLM 结果不完整:缺少 " + missingRows + "/" + expectedRows + " 条检测结果,请等待重试或重新运行任务");
}
}
- private boolean hasCompleteCozeResult(AppearancePatentResultRowDto row) {
- return hasUsableCozeField(row.getTitleRisk())
- && hasUsableCozeField(row.getAppearanceRisk())
- && hasUsableCozeField(row.getPatentRisk())
- && hasUsableCozeField(row.getConclusion());
+ private boolean hasCompleteLlmResult(AppearancePatentResultRowDto row) {
+ return hasUsableLlmField(row.getTitleRisk())
+ && hasUsableLlmField(row.getAppearanceRisk())
+ && hasUsableLlmField(row.getPatentRisk())
+ && hasUsableLlmField(row.getConclusion());
}
private boolean hasPersistedResultRows(Long taskId) {
@@ -2231,8 +2231,8 @@ public class AppearancePatentTaskService {
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getTitle(), "") : firstNonBlank(resultRow.getTitle(), parsedRow.getTitle()));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getUrl(), "") : firstNonBlank(resultRow.getUrl(), parsedRow.getUrl()));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getSku(), "") : firstNonBlank(resultRow.getSku(), parsedRow.getSku()));
- row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getTitleRisk()));
- row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getAppearanceRisk()));
+ row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingLlmCellValue(resultRow, resultRow.getTitleRisk()));
+ row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingLlmCellValue(resultRow, resultRow.getAppearanceRisk()));
row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingConclusion(resultRow));
row.createCell(col).setCellValue(resultRow == null ? "" : userFacingStatus(resultRow));
}
@@ -2284,7 +2284,7 @@ public class AppearancePatentTaskService {
if (row == null || !normalizedAsin.equals(normalize(row.getAsin()).toUpperCase(Locale.ROOT))) {
continue;
}
- if (hasResolvedCozeFields(row) || hasReasonFields(row)) {
+ if (hasResolvedLlmFields(row) || hasReasonFields(row)) {
return row;
}
if (fallback == null) {
@@ -2484,19 +2484,19 @@ public class AppearancePatentTaskService {
return row != null && (!normalize(row.getTitle()).isBlank() || !normalize(row.getUrl()).isBlank());
}
- private boolean hasResolvedCozeFields(AppearancePatentResultRowDto row) {
+ private boolean hasResolvedLlmFields(AppearancePatentResultRowDto row) {
if (row == null) {
return false;
}
- return hasUsableCozeField(row.getTitleRisk())
- || hasUsableCozeField(row.getAppearanceRisk())
- || hasUsableCozeField(row.getPatentRisk())
- || hasUsableCozeField(row.getConclusion());
+ return hasUsableLlmField(row.getTitleRisk())
+ || hasUsableLlmField(row.getAppearanceRisk())
+ || hasUsableLlmField(row.getPatentRisk())
+ || hasUsableLlmField(row.getConclusion());
}
- private boolean hasUsableCozeField(String value) {
+ private boolean hasUsableLlmField(String value) {
String normalized = normalize(value);
- return !normalized.isBlank() && !isTechnicalCozeFailure(normalized);
+ return !normalized.isBlank() && !isTechnicalLlmFailure(normalized);
}
private Map buildHeaderMap(Row header, DataFormatter formatter) {
@@ -2808,8 +2808,8 @@ public class AppearancePatentTaskService {
return transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
}
- private String storeSharedCozeBatchPayload(Long taskId, String scopeHash, String payloadJson) {
- requireSharedTransientPayloadStorage("coze batch payload");
+ private String storeSharedLlmBatchPayload(Long taskId, String scopeHash, String payloadJson) {
+ requireSharedTransientPayloadStorage("llm batch payload");
return transientPayloadStorageService.storeParsedPayloadFast(MODULE_TYPE, taskId, scopeHash, payloadJson, true);
}
@@ -3048,13 +3048,13 @@ public class AppearancePatentTaskService {
return executionFailed ? STATUS_FAILED : STATUS_SUCCESS;
}
- private String userFacingCozeCellValue(AppearancePatentResultRowDto row, String value) {
+ private String userFacingLlmCellValue(AppearancePatentResultRowDto row, String value) {
String normalizedValue = normalize(value);
- if (!normalizedValue.isBlank() && !isTechnicalCozeFailure(normalizedValue)) {
+ if (!normalizedValue.isBlank() && !isTechnicalLlmFailure(normalizedValue)) {
return value;
}
- // coze 技术性失败:有错误信息则放入错误信息,没有则留空
- if (row != null && isTechnicalCozeFailure(row.getError())) {
+ // llm 技术性失败:有错误信息则放入错误信息,没有则留空
+ if (row != null && isTechnicalLlmFailure(row.getError())) {
return firstNonBlank(row.getError(), "");
}
return firstNonBlank(value, "");
@@ -3065,11 +3065,11 @@ public class AppearancePatentTaskService {
return "";
}
String conclusion = normalize(row.getConclusion());
- if (!conclusion.isBlank() && !isTechnicalCozeFailure(conclusion)) {
+ if (!conclusion.isBlank() && !isTechnicalLlmFailure(conclusion)) {
return row.getConclusion();
}
- // coze 技术性失败:有错误信息则放入错误信息,没有则留空
- if (isTechnicalCozeFailure(row.getError())) {
+ // llm 技术性失败:有错误信息则放入错误信息,没有则留空
+ if (isTechnicalLlmFailure(row.getError())) {
return firstNonBlank(row.getError(), "");
}
return firstNonBlank(row.getConclusion(), "");
@@ -3089,9 +3089,9 @@ public class AppearancePatentTaskService {
return normalized.isBlank() ? "\u5931\u8d25" : "\u6210\u529f";
}
- private boolean isTechnicalCozeFailure(String value) {
+ private boolean isTechnicalLlmFailure(String value) {
String normalized = normalize(value).toLowerCase(Locale.ROOT);
- return normalized.contains("coze")
+ return normalized.contains("coze") || normalized.contains("llm")
|| normalized.contains("结果不完整")
|| normalized.contains("工作流节点执行超限")
|| normalized.contains("调用超时")
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/mapper/CozeCredentialMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/coze/mapper/CozeCredentialMapper.java
deleted file mode 100644
index 608fda32..00000000
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/mapper/CozeCredentialMapper.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package com.nanri.aiimage.modules.coze.mapper;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.nanri.aiimage.modules.coze.model.entity.CozeCredentialEntity;
-import org.apache.ibatis.annotations.Mapper;
-
-@Mapper
-public interface CozeCredentialMapper extends BaseMapper {
-}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/model/entity/CozeCredentialEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/coze/model/entity/CozeCredentialEntity.java
deleted file mode 100644
index 3752f897..00000000
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/model/entity/CozeCredentialEntity.java
+++ /dev/null
@@ -1,25 +0,0 @@
-package com.nanri.aiimage.modules.coze.model.entity;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.Data;
-
-import java.time.LocalDateTime;
-
-@Data
-@TableName("biz_coze_credential")
-public class CozeCredentialEntity {
-
- @TableId(type = IdType.AUTO)
- private Long id;
- private String moduleType;
- private String credentialName;
- private String workflowId;
- private String token;
- private Integer enabled;
- private Integer maxConcurrent;
- private Integer sortOrder;
- private LocalDateTime createdAt;
- private LocalDateTime updatedAt;
-}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java
deleted file mode 100644
index a9482a15..00000000
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java
+++ /dev/null
@@ -1,200 +0,0 @@
-package com.nanri.aiimage.modules.coze.service;
-
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
-import com.nanri.aiimage.modules.coze.mapper.CozeCredentialMapper;
-import com.nanri.aiimage.modules.coze.model.entity.CozeCredentialEntity;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.stereotype.Service;
-
-import java.time.Duration;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-
-@Slf4j
-@Service
-@RequiredArgsConstructor
-public class CozeCredentialPoolService {
-
- private static final Duration INFLIGHT_TTL = Duration.ofMinutes(30);
-
- /**
- * 每个 moduleType 只 WARN 一次,避免高频日志噪音。
- * key = moduleType,value = 仅作占位,仅用 putIfAbsent 语义判断"是否已经 WARN 过"。
- */
- private final ConcurrentHashMap stripeWarnedModules = new ConcurrentHashMap<>();
-
- private final CozeCredentialMapper cozeCredentialMapper;
- private final StringRedisTemplate stringRedisTemplate;
-
- public List listEnabled(String moduleType) {
- if (moduleType == null || moduleType.isBlank()) {
- return List.of();
- }
- try {
- List rows = cozeCredentialMapper.selectList(new LambdaQueryWrapper()
- .eq(CozeCredentialEntity::getModuleType, moduleType)
- .eq(CozeCredentialEntity::getEnabled, 1)
- .orderByAsc(CozeCredentialEntity::getSortOrder)
- .orderByAsc(CozeCredentialEntity::getId));
- if (rows == null || rows.isEmpty()) {
- return List.of();
- }
- return rows.stream()
- .filter(Objects::nonNull)
- .filter(row -> !blank(row.getCredentialName())
- && !blank(row.getWorkflowId())
- && !blank(row.getToken()))
- .map(row -> new CozeCredential(
- row.getCredentialName().trim(),
- row.getWorkflowId().trim(),
- row.getToken().trim(),
- row.getMaxConcurrent() == null || row.getMaxConcurrent() <= 0
- ? Integer.MAX_VALUE
- : row.getMaxConcurrent()))
- .toList();
- } catch (Exception ex) {
- log.warn("[coze-credential] list enabled failed moduleType={} err={}", moduleType, ex.getMessage());
- return List.of();
- }
- }
-
- public CozeCredential chooseLeastInflight(String moduleType, List credentials) {
- return chooseRoundRobin(moduleType, credentials, 1);
- }
-
- public CozeCredential chooseRoundRobin(String moduleType, List credentials, int stripeSize) {
- if (credentials == null || credentials.isEmpty()) {
- return null;
- }
- // stripeSize ≤ 0 时自动按 credentials.size() 适配;正数则按配置走。
- // 注意:stripe=1 与 stripe=credentials.size() 在轮换正确性上等价,差异只在"每个凭据连续选中次数"。
- int safeStripeSize = stripeSize <= 0 ? credentials.size() : stripeSize;
- // 仅当 stripe>1 且小于凭据数时 WARN:此时凭据被切到下一张前会连续选 stripe 次,但未覆盖所有凭据就回到首张,存在偏向。
- if (safeStripeSize > 1 && safeStripeSize < credentials.size() && moduleType != null && !moduleType.isBlank()) {
- if (stripeWarnedModules.putIfAbsent(moduleType, Boolean.TRUE) == null) {
- log.warn("[coze-credential] stripeSize({}) < credentials.size({}) for moduleType={}, "
- + "round-robin may be biased; consider setting stripeSize == credentials.size() "
- + "or leave it 0/negative to auto-adapt",
- safeStripeSize, credentials.size(), moduleType);
- }
- }
- long cursor = nextCursor(moduleType);
- int index = (int) ((Math.max(0L, cursor) / safeStripeSize) % credentials.size());
- return credentials.get(index);
- }
-
- public BorrowedCredential borrow(String moduleType, CozeCredential credential) {
- if (credential == null) {
- return null;
- }
- String key = inflightKey(moduleType, credential.name());
- try {
- Long value = stringRedisTemplate.opsForValue().increment(key);
- stringRedisTemplate.expire(key, INFLIGHT_TTL);
- long inflight = value == null ? 0L : value;
- if (inflight > credential.maxConcurrent()) {
- release(moduleType, credential.name());
- return null;
- }
- return new BorrowedCredential(this, moduleType, credential.name());
- } catch (Exception ex) {
- log.warn("[coze-credential] borrow failed moduleType={} credential={} err={}",
- moduleType, credential.name(), ex.getMessage());
- return BorrowedCredential.noop();
- }
- }
-
- public void release(String moduleType, String credentialName) {
- if (blank(moduleType) || blank(credentialName)) {
- return;
- }
- try {
- Long value = stringRedisTemplate.opsForValue().decrement(inflightKey(moduleType, credentialName));
- if (value != null && value <= 0L) {
- stringRedisTemplate.delete(inflightKey(moduleType, credentialName));
- }
- } catch (Exception ex) {
- log.warn("[coze-credential] release failed moduleType={} credential={} err={}",
- moduleType, credentialName, ex.getMessage());
- }
- }
-
- private long inflight(String moduleType, String credentialName) {
- try {
- String raw = stringRedisTemplate.opsForValue().get(inflightKey(moduleType, credentialName));
- return raw == null || raw.isBlank() ? 0L : Long.parseLong(raw);
- } catch (Exception ex) {
- return 0L;
- }
- }
-
- private long nextCursor(String moduleType) {
- if (blank(moduleType)) {
- return 0L;
- }
- try {
- Long value = stringRedisTemplate.opsForValue().increment(cursorKey(moduleType));
- stringRedisTemplate.expire(cursorKey(moduleType), Duration.ofDays(7));
- return value == null ? 0L : Math.max(0L, value - 1L);
- } catch (Exception ex) {
- log.warn("[coze-credential] cursor increment failed moduleType={} err={}", moduleType, ex.getMessage());
- return System.nanoTime();
- }
- }
-
- private String inflightKey(String moduleType, String credentialName) {
- return "coze:credential:inflight:" + moduleType + ":" + credentialName;
- }
-
- private String cursorKey(String moduleType) {
- return "coze:credential:cursor:" + moduleType;
- }
-
- private boolean blank(String value) {
- return value == null || value.isBlank();
- }
-
- public record CozeCredential(String name,
- String workflowId,
- String token,
- int maxConcurrent) {
- }
-
- public static final class BorrowedCredential implements AutoCloseable {
- private final CozeCredentialPoolService owner;
- private final String moduleType;
- private final String credentialName;
- private final boolean noop;
- private boolean released;
-
- private BorrowedCredential(CozeCredentialPoolService owner, String moduleType, String credentialName) {
- this.owner = owner;
- this.moduleType = moduleType;
- this.credentialName = credentialName;
- this.noop = false;
- }
-
- private BorrowedCredential() {
- this.owner = null;
- this.moduleType = null;
- this.credentialName = null;
- this.noop = true;
- }
-
- private static BorrowedCredential noop() {
- return new BorrowedCredential();
- }
-
- @Override
- public void close() {
- if (released || noop) {
- return;
- }
- released = true;
- owner.release(moduleType, credentialName);
- }
- }
-}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java
index f33ae5d3..7ce5147a 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java
@@ -64,7 +64,7 @@ public class ImageVideoAsyncTaskService {
private final ImageVideoWorkflowConfigService workflowConfigService;
private final ImageVideoArchiveService archiveService;
private final ObjectMapper objectMapper;
- private final TaskExecutor cozeTaskExecutor;
+ private final TaskExecutor taskQueueExecutor;
private final InstanceMetadata instanceMetadata;
public ImageVideoAsyncTaskService(
@@ -73,14 +73,14 @@ public class ImageVideoAsyncTaskService {
ImageVideoWorkflowConfigService workflowConfigService,
ImageVideoArchiveService archiveService,
ObjectMapper objectMapper,
- @Qualifier("cozeTaskExecutor") TaskExecutor cozeTaskExecutor,
+ @Qualifier("taskQueueExecutor") TaskExecutor taskQueueExecutor,
InstanceMetadata instanceMetadata) {
this.taskMapper = taskMapper;
this.cozeService = cozeService;
this.workflowConfigService = workflowConfigService;
this.archiveService = archiveService;
this.objectMapper = objectMapper;
- this.cozeTaskExecutor = cozeTaskExecutor;
+ this.taskQueueExecutor = taskQueueExecutor;
this.instanceMetadata = instanceMetadata;
}
@@ -137,7 +137,7 @@ public class ImageVideoAsyncTaskService {
.or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
.orderByAsc(ImageVideoAsyncTaskEntity::getId)
.last("LIMIT " + DISPATCH_BATCH_SIZE));
- tasks.forEach(task -> cozeTaskExecutor.execute(() -> executeTask(task.getId())));
+ tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId())));
}
@Scheduled(fixedDelayString = "${aiimage.image-video.async-task-poll-delay-ms:5000}")
@@ -149,7 +149,7 @@ public class ImageVideoAsyncTaskService {
.or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
.orderByAsc(ImageVideoAsyncTaskEntity::getUpdatedAt)
.last("LIMIT " + POLL_BATCH_SIZE));
- tasks.forEach(task -> cozeTaskExecutor.execute(() -> pollTask(task.getId())));
+ tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId())));
}
@EventListener(ApplicationReadyEvent.class)
@@ -190,7 +190,7 @@ public class ImageVideoAsyncTaskService {
taskMapper.insert(task);
log.info("[image-video] async task submitted taskId={} type={} owner={}",
task.getId(), task.getTaskType(), task.getOwnerInstanceId());
- cozeTaskExecutor.execute(() -> executeTask(task.getId()));
+ taskQueueExecutor.execute(() -> executeTask(task.getId()));
return toVo(task);
}
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
deleted file mode 100644
index 2a93d042..00000000
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java
+++ /dev/null
@@ -1,1536 +0,0 @@
-package com.nanri.aiimage.modules.similarasin.client;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.nanri.aiimage.config.SimilarAsinProperties;
-import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService;
-import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto;
-import com.nanri.aiimage.modules.similarasin.util.SimilarAsinLogSupport;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.MediaType;
-import org.springframework.http.client.SimpleClientHttpRequestFactory;
-import org.springframework.stereotype.Component;
-import org.springframework.util.StreamUtils;
-import org.springframework.web.client.RestClient;
-
-import com.nanri.aiimage.config.HttpClientPool;
-
-import java.math.BigDecimal;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-@Component
-@Slf4j
-public class SimilarAsinCozeClient {
-
- private static final String MODULE_TYPE = "SIMILAR_ASIN";
- private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
- /** Task 19:history 轮询响应正文采样频率(每 N 次记一次完整正文,其余只记状态)。 */
- static final long HISTORY_RESPONSE_LOG_EVERY_N = 20L;
-
- private final SimilarAsinProperties properties;
- private final ObjectMapper objectMapper;
- private final CozeCredentialPoolService cozeCredentialPoolService;
- private final com.nanri.aiimage.metrics.ExternalCallMetricsRecorder externalCallMetrics;
- private final AtomicLong credentialCursor = new AtomicLong();
- private final AtomicLong historyResponseLogCounter = new AtomicLong();
- /**
- * P1-7:单例 RestClient。原 restClient() 每次提交/poll 都新建 SimpleClientHttpRequestFactory + RestClient,
- * 几千行任务并发时会反复创建短命对象造成不必要 GC 压力。RestClient 与 SimpleClientHttpRequestFactory
- * 都是线程安全的,复用一份即可;timeout 变更需要重启服务生效(与历史行为一致)。
- */
- private volatile RestClient sharedRestClient;
-
- public List inspect(List rows, String prompt, String apiKey) {
- return inspect(rows, prompt, apiKey, false);
- }
-
- public List inspect(List rows, String prompt, String apiKey, boolean imgSwitch) {
- return inspect(rows, prompt, apiKey, imgSwitch, false);
- }
-
- public List inspect(List rows,
- String prompt,
- String apiKey,
- boolean imgSwitch,
- boolean categorySwitch) {
- if (rows == null || rows.isEmpty()) {
- return List.of();
- }
- if (!hasConfiguredCredential()) {
- log.warn("[similar-asin] coze token not configured, keep raw rows size={}", rows.size());
- return rows.stream().map(this::copy).toList();
- }
- try {
- return inspectWithFallback(rows, prompt, apiKey, imgSwitch, categorySwitch);
- } catch (Exception ex) {
- String failureMessage = failureMessage(ex);
- log.warn("[similar-asin] coze batch failed size={} err={}", rows.size(), failureMessage);
- return rows.stream().map(this::copy).map(row -> markFailed(row, failureMessage)).toList();
- }
- }
-
- public CozeSubmitResponse submitWorkflow(List rows, String prompt, String apiKey) throws Exception {
- return submitWorkflow(rows, prompt, apiKey, false, nextCredential());
- }
-
- public CozeSubmitResponse submitWorkflow(List rows, String prompt, String apiKey, boolean imgSwitch) throws Exception {
- return submitWorkflow(rows, prompt, apiKey, imgSwitch, nextCredential());
- }
-
- public CozeSubmitResponse submitWorkflow(List rows,
- String prompt,
- String apiKey,
- boolean imgSwitch,
- boolean categorySwitch,
- CozeCredentialRef credential) throws Exception {
- CozeCredentialRef resolvedCredential = resolveCredential(credential);
- JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey, imgSwitch, categorySwitch, resolvedCredential));
- ensureSuccess(submitRoot);
- return new CozeSubmitResponse(
- extractExecuteId(submitRoot),
- extractResultDataText(submitRoot),
- writeJson(submitRoot),
- resolvedCredential.name()
- );
- }
-
- public CozeSubmitResponse submitWorkflow(List rows,
- String prompt,
- String apiKey,
- boolean imgSwitch,
- CozeCredentialRef credential) throws Exception {
- return submitWorkflow(rows, prompt, apiKey, imgSwitch, false, credential);
- }
-
- public CozePollResponse pollWorkflow(String executeId) throws Exception {
- return pollWorkflow(executeId, null);
- }
-
- public CozePollResponse pollWorkflow(String executeId, CozeCredentialRef credential) throws Exception {
- CozeCredentialRef resolvedCredential = resolveCredential(credential);
- JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId, resolvedCredential));
- ensureSuccess(pollRoot);
- String status = normalize(resolveWorkflowStatus(pollRoot)).toUpperCase(Locale.ROOT);
- String dataText = extractResultDataText(pollRoot);
- String outputText = dataText.isBlank() ? extractWorkflowOutputText(pollRoot) : "";
- String failureMessage = isFailedWorkflowStatus(status)
- ? firstNonBlank(resolveFailureMessage(pollRoot), "Coze 异步工作流失败")
- : "";
- return new CozePollResponse(executeId, status, dataText, outputText, failureMessage, writeJson(pollRoot),
- resolvedCredential.name());
- }
-
- public List mergeRowsFromDataText(List rows, String dataText) throws Exception {
- if (dataText == null || dataText.isBlank()) {
- return rows == null ? List.of() : rows.stream().map(this::copy).toList();
- }
- return mergeRows(rows, parseResults(wrapDataPayload(dataText)));
- }
-
- public List markRowsFailed(List rows, String failureMessage) {
- if (rows == null || rows.isEmpty()) {
- return List.of();
- }
- return rows.stream().map(this::copy).map(row -> markFailed(row, failureMessage)).toList();
- }
-
- private List inspectWithFallback(List rows, String prompt, String apiKey, boolean imgSwitch, boolean categorySwitch) {
- try {
- if (rows.size() == 1) {
- return inspectSingleRowWithRetry(rows, prompt, apiKey, imgSwitch, categorySwitch);
- }
- InspectAttempt attempt = inspectOnce(rows, prompt, apiKey, imgSwitch, categorySwitch);
- if (attempt.resolvedCount() < rows.size()) {
- throw new PartialCozeResultException(attempt.resolvedCount(), rows.size(), attempt.rawResultCount());
- }
- return attempt.mergedRows();
- } catch (Exception ex) {
- if (shouldSplitBatch(rows, ex)) {
- int middle = rows.size() / 2;
- 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, apiKey, imgSwitch, categorySwitch));
- merged.addAll(inspectPartitionWithFailureFallback(rows.subList(middle, rows.size()), prompt, apiKey, imgSwitch, categorySwitch));
- return merged;
- }
- throw propagate(ex);
- }
- }
-
- private List inspectPartitionWithFailureFallback(List rows, String prompt, String apiKey, boolean imgSwitch, boolean categorySwitch) {
- try {
- return inspectWithFallback(rows, prompt, apiKey, imgSwitch, categorySwitch);
- } catch (Exception ex) {
- String failureMessage = failureMessage(ex);
- log.warn("[similar-asin] coze partition failed size={} err={}", rows.size(), failureMessage);
- return rows.stream().map(this::copy).map(row -> markFailed(row, failureMessage)).toList();
- }
- }
-
- private List inspectSingleRowWithRetry(List rows, String prompt, String apiKey, boolean imgSwitch, boolean categorySwitch) throws Exception {
- SimilarAsinResultRowDto row = rows.getFirst();
- PartialCozeResultException lastFailure = null;
- for (int attemptIndex = 1; attemptIndex <= 3; attemptIndex++) {
- try {
- InspectAttempt attempt = inspectOnce(rows, prompt, apiKey, imgSwitch, categorySwitch);
- if (attempt.resolvedCount() == rows.size()) {
- return attempt.mergedRows();
- }
- log.warn("[similar-asin] coze single unresolved attempt={} rowId={} asin={} country={} title={} url={} raw={}",
- attemptIndex,
- row.getId(),
- row.getAsin(),
- row.getCountry(),
- abbreviate(row.getTitle(), 120),
- abbreviate(row.getUrl(), 120),
- abbreviate(attempt.raw(), 500));
- lastFailure = new PartialCozeResultException(attempt.resolvedCount(), rows.size(), attempt.rawResultCount());
- } catch (Exception ex) {
- if (attemptIndex >= 3 || !isRetryableBatchFailure(ex)) {
- throw ex;
- }
- log.warn("[similar-asin] coze single retryable failure attempt={} rowId={} asin={} country={} err={}",
- attemptIndex,
- row.getId(),
- row.getAsin(),
- row.getCountry(),
- failureMessage(ex));
- }
- if (attemptIndex < 3) {
- if (externalCallMetrics != null) {
- externalCallMetrics.recordRetry("coze");
- }
- sleepBeforeRetry(attemptIndex);
- }
- }
- throw lastFailure == null ? new PartialCozeResultException(0, rows.size(), 0) : lastFailure;
- }
-
- private InspectAttempt inspectOnce(List rows, String prompt, String apiKey, boolean imgSwitch, boolean categorySwitch) throws Exception {
- String raw = runWorkflowAsyncAndWait(rows, prompt, apiKey, imgSwitch, categorySwitch);
- List results = parseResults(raw);
- List merged = mergeRows(rows, results);
- return new InspectAttempt(raw, merged, resolvedCount(merged), results.size());
- }
-
- private String runWorkflowAsyncAndWait(List rows, String prompt, String apiKey, boolean imgSwitch, boolean categorySwitch) throws Exception {
- CozeCredentialRef credential = nextCredential();
- JsonNode submitRoot = objectMapper.readTree(postWorkflow(rows, prompt, apiKey, imgSwitch, categorySwitch, credential));
- ensureSuccess(submitRoot);
-
- String immediateData = extractResultDataText(submitRoot);
- if (!immediateData.isBlank()) {
- return wrapDataPayload(immediateData);
- }
-
- String executeId = extractExecuteId(submitRoot);
- if (executeId == null || executeId.isBlank()) {
- throw new IllegalStateException("Coze async execute_id missing");
- }
-
- long deadline = System.currentTimeMillis() + Math.max(10000, properties.getCozePollTimeoutMillis());
- while (System.currentTimeMillis() < deadline) {
- ensureNotInterrupted();
- JsonNode pollRoot = objectMapper.readTree(getWorkflowHistory(executeId, credential));
- ensureSuccess(pollRoot);
-
- String dataText = extractResultDataText(pollRoot);
- if (!dataText.isBlank()) {
- return wrapDataPayload(dataText);
- }
-
- String status = normalize(resolveWorkflowStatus(pollRoot)).toUpperCase(Locale.ROOT);
- if (isFailedWorkflowStatus(status)) {
- throw new IllegalStateException(firstNonBlank(resolveFailureMessage(pollRoot), "Coze 异步工作流失败"));
- }
- if (isSuccessfulWorkflowStatus(status)) {
- String outputText = extractWorkflowOutputText(pollRoot);
- if (!outputText.isBlank()) {
- return wrapDataPayload(outputText);
- }
- throw new IllegalStateException("Coze 异步工作流已完成但没有输出结果");
- }
- sleepQuietly(Math.max(200, properties.getCozePollIntervalMillis()));
- }
-
- throw new IllegalStateException("Coze 异步工作流轮询超时");
- }
-
- private String postWorkflow(List rows,
- String prompt,
- String apiKey,
- boolean imgSwitch,
- boolean categorySwitch,
- CozeCredentialRef credential) {
- Map parameters = buildParameters(rows, prompt, apiKey, imgSwitch, categorySwitch);
- Map body = new LinkedHashMap<>();
- body.put("workflow_id", credential.workflowId());
- body.put("parameters", parameters);
- body.put("is_async", Boolean.TRUE);
- log.debug("[similar-asin] coze request credential={} url={} body={}",
- credential.name(),
- joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()),
- SimilarAsinLogSupport.truncate(writeJson(maskCozeRequestBody(body))));
-
- RestClient.RequestBodySpec request = restClient().post()
- .uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
- .headers(headers -> {
- headers.setBearerAuth(stripBearer(credential.token()));
- headers.setContentType(APPLICATION_JSON_UTF8);
- headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
- });
- request.body(body);
- return request.exchange((clientRequest, clientResponse) -> {
- byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());
- String responseText = responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8);
- log.debug("[similar-asin] coze submit response status={} body={}",
- clientResponse.getStatusCode(),
- SimilarAsinLogSupport.truncate(responseText));
- return responseText;
- });
- }
-
- private String getWorkflowHistory(String executeId, CozeCredentialRef credential) {
- String path = properties.getCozeWorkflowHistoryPath()
- .replace("{workflow_id}", credential.workflowId())
- .replace("{execute_id}", executeId);
- long historyLogCounter = historyResponseLogCounter.getAndIncrement();
- return restClient().get()
- .uri(joinUrl(properties.getCozeBaseUrl(), path))
- .headers(headers -> {
- headers.setBearerAuth(stripBearer(credential.token()));
- headers.setContentType(APPLICATION_JSON_UTF8);
- headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
- })
- .exchange((clientRequest, clientResponse) -> {
- byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());
- String responseText = responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8);
- log.debug("[similar-asin] coze history response credential={} executeId={} status={} body={}",
- credential.name(), executeId,
- clientResponse.getStatusCode(),
- SimilarAsinLogSupport.shouldLog(historyLogCounter, HISTORY_RESPONSE_LOG_EVERY_N)
- ? SimilarAsinLogSupport.truncate(responseText)
- : "[sampled out]");
- return responseText;
- });
- }
-
- public CozeCredentialRef nextCredential() {
- List pooledCredentials = cozeCredentialPoolService.listEnabled(MODULE_TYPE);
- CozeCredentialPoolService.CozeCredential pooledCredential =
- cozeCredentialPoolService.chooseRoundRobin(MODULE_TYPE, pooledCredentials, properties.getCozeCredentialStripeSize());
- if (pooledCredential != null) {
- return new CozeCredentialRef(pooledCredential.name(), pooledCredential.workflowId(), pooledCredential.token(),
- pooledCredential.maxConcurrent());
- }
- List credentials = configuredCredentials();
- int configuredStripe = properties.getCozeCredentialStripeSize();
- int stripeSize = configuredStripe <= 0 ? Math.max(1, credentials.size()) : configuredStripe;
- long cursor = Math.max(0L, credentialCursor.getAndIncrement());
- int index = (int) ((cursor / stripeSize) % credentials.size());
- return credentials.get(index);
- }
-
- public CozeCredentialRef credentialByName(String name) {
- if (name == null || name.isBlank()) {
- return nextCredential();
- }
- String normalizedName = normalize(name);
- for (CozeCredentialRef credential : configuredCredentials()) {
- if (normalize(credential.name()).equals(normalizedName)) {
- return credential;
- }
- }
- return nextCredential();
- }
-
- public boolean hasConfiguredCredential() {
- return !configuredCredentials().isEmpty();
- }
-
- public int configuredCredentialCount() {
- return configuredCredentials().size();
- }
-
- private CozeCredentialRef resolveCredential(CozeCredentialRef credential) {
- return credential == null ? nextCredential() : credential;
- }
-
- private List configuredCredentials() {
- List credentials = new ArrayList<>();
- for (CozeCredentialPoolService.CozeCredential credential : cozeCredentialPoolService.listEnabled(MODULE_TYPE)) {
- credentials.add(new CozeCredentialRef(credential.name(), credential.workflowId(), credential.token(),
- credential.maxConcurrent()));
- }
- if (!credentials.isEmpty()) {
- return credentials;
- }
- if (properties.getCozeCredentials() != null) {
- int index = 1;
- for (SimilarAsinProperties.CozeCredential credential : properties.getCozeCredentials()) {
- if (credential == null
- || normalize(credential.getWorkflowId()).isBlank()
- || normalize(credential.getToken()).isBlank()) {
- continue;
- }
- String name = firstNonBlank(credential.getName(), "credential-" + index);
- credentials.add(new CozeCredentialRef(name, credential.getWorkflowId(), credential.getToken(), Integer.MAX_VALUE));
- index++;
- }
- }
- if (credentials.isEmpty()
- && properties.getCozeWorkflowId() != null && !properties.getCozeWorkflowId().isBlank()
- && properties.getCozeToken() != null && !properties.getCozeToken().isBlank()) {
- credentials.add(new CozeCredentialRef("default", properties.getCozeWorkflowId(), properties.getCozeToken(), Integer.MAX_VALUE));
- }
- return credentials;
- }
-
- private Map buildParameters(List rows, String prompt, String apiKey, boolean imgSwitch) {
- return buildParameters(rows, prompt, apiKey, imgSwitch, false);
- }
-
- private Map buildParameters(List rows,
- String prompt,
- String apiKey,
- boolean imgSwitch,
- boolean categorySwitch) {
- List asins = rows.stream().map(row -> safeText(row.getAsin())).toList();
- List titles = rows.stream().map(row -> safeText(firstNonBlank(row.getTitle(), row.getAsin()))).toList();
- List skus = rows.stream().map(row -> safeText(row.getSku())).toList();
- List urls = rows.stream().map(this::primaryImageUrl).toList();
- List> urlLists = rows.stream().map(this::imageUrls).toList();
- List>> alibabaLists = rows.stream().map(this::alibabaItems).toList();
-
- List