From 203937335df7722f4c712d83a9f73c03c33e1f82 Mon Sep 17 00:00:00 2001 From: super <2903208875@qq.com> Date: Tue, 5 May 2026 14:11:35 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=84=E7=90=86=E4=B8=80=E4=BA=9B=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E5=86=B2=E7=AA=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/amazon/detail_spider.py | 99 ++--- app/amazon/similar_asin.py | 14 +- .../com/nanri/aiimage/AiImageApplication.java | 6 + .../service/DistributedJobLockService.java | 25 ++ .../aiimage/config/SchedulingConfig.java | 6 + .../config/TaskOperationLockConfig.java | 78 ++++ .../client/AppearancePatentCozeClient.java | 9 +- .../service/AppearancePatentTaskService.java | 419 ++++++++++++++---- .../brand/service/BrandTaskService.java | 16 + .../service/DeleteBrandStaleTaskService.java | 50 +++ .../client/SimilarAsinCozeClient.java | 9 +- .../service/SimilarAsinTaskService.java | 38 +- .../service/TaskDistributedLockService.java | 160 +++++++ .../task/service/TaskFileJobService.java | 20 + .../task/service/TaskResultFileJobWorker.java | 35 +- .../TransientPayloadStorageService.java | 16 +- .../service/ZiniaoShopIndexService.java | 37 +- .../src/main/resources/application.yml | 2 + scripts/coze_diag.py | 234 ++++++++++ 19 files changed, 1075 insertions(+), 198 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java create mode 100644 scripts/coze_diag.py diff --git a/app/amazon/detail_spider.py b/app/amazon/detail_spider.py index 5273934..6134c91 100644 --- a/app/amazon/detail_spider.py +++ b/app/amazon/detail_spider.py @@ -111,31 +111,15 @@ class ChromeAmzone(ChromeAmzoneBase): -<<<<<<< HEAD -class SpiderTask(TaskBase): - task_name = "亚马逊采集" -======= -class SpiderTask: - mark_name = "亚马逊采集" - - def __init__(self, user_info: dict = None): - """初始化审批任务处理器 - - Args: - user_info: 用户信息字典，包含 company, username, password - """ - self.user_info = user_info or {} - self.running = True +class SpiderTask(TaskBase): + task_name = "亚马逊采集" + mark_name = "亚马逊采集" + + def __init__(self, user_info: dict = None): + super().__init__(user_info) self.result_api_module = "appearance-patent" - - def log(self, message: str, level: str = "INFO"): - timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - if level == "ERROR": - show_notification(message, "error") - print(f"[{timestamp}] [PriceTask] [{level}] {message}") ->>>>>>> 6d398b66bc24634b13f1de162e8f68836e64441d - - @staticmethod + + @staticmethod def group_by_id_prefix(data): """ 根据每条数据的 id 字段进行分组： @@ -180,25 +164,20 @@ class SpiderTask: }) return normalized - def process_task(self, task_data: dict): - """处理审批任务主入口 - - Args: - task_data: 任务数据 - """ - try: -<<<<<<< HEAD - data = task_data.get("data", {}) - task_id = data.get("taskId") - groups = self.normalize_groups(data) -======= + def process_task(self, task_data: dict): + """处理审批任务主入口 + + Args: + task_data: 任务数据 + """ + task_id = None + try: data = task_data.get("data", {}) - self.result_api_module = "similar-asin" if task_data.get("type") == "similar-asin-run" else "appearance-patent" task_id = data.get("taskId") groups = self.normalize_groups(data) ->>>>>>> 6d398b66bc24634b13f1de162e8f68836e64441d - - # 用于测试 + self.result_api_module = "similar-asin" if task_data.get("type") == "similar-asin-run" else "appearance-patent" + + # 用于测试 limit = data.get("limit", None) if not task_id: @@ -329,22 +308,22 @@ class SpiderTask: runing_task[task_id]["status"] = "failed" runing_task[task_id]["error"] = str(e) - def post_result(self, task_id: int, chunkIndex:int,chunkTotal: int, asin: str, error:str="", - item_data:dict={}, - is_done: bool = False): - """回传处理结果到API - """ + def post_result(self, task_id: int, chunkIndex:int,chunkTotal: int, asin: str, error:str="", + item_data=None, + is_done: bool = False): + """回传处理结果到API + """ url = f"{DELETE_BRAND_API_BASE}/api/{self.result_api_module}/tasks/{task_id}/result" - payload ={ - "submissionId": f"{task_id}", - "chunkIndex": chunkIndex, - "chunkTotal": chunkTotal, - "error": error, - "groups": item_data, - "done": is_done - } + payload ={ + "submissionId": f"{task_id}", + "chunkIndex": chunkIndex, + "chunkTotal": chunkTotal, + "error": error, + "groups": item_data or [], + "done": is_done + } max_retries = 3 for retry in range(max_retries): @@ -354,13 +333,13 @@ class SpiderTask: self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)") self.log(f"回传URL: {url}") self.log(f"回传数据: {payload}") - response = requests.post( - url, - json=payload, - headers={"Content-Type": "application/json"}, - timeout=request_timeout, - verify=False - ) + response = requests.post( + url, + data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + headers={"Content-Type": "application/json; charset=utf-8"}, + timeout=request_timeout, + verify=False + ) self.log(f"回传结果: {response.text}") data = response.json() if response.text else {} if response.status_code == 200 and isinstance(data, dict) and data.get("success"): diff --git a/app/amazon/similar_asin.py b/app/amazon/similar_asin.py index 4cc81eb..352d6c0 100644 --- a/app/amazon/similar_asin.py +++ b/app/amazon/similar_asin.py @@ -466,13 +466,13 @@ class SimilarAsinTask(TaskBase): self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)") self.log(f"回传URL: {url}") self.log(f"回传数据: {payload}") - response = requests.post( - url, - json=payload, - headers={"Content-Type": "application/json"}, - timeout=request_timeout, - verify=False - ) + response = requests.post( + url, + data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + headers={"Content-Type": "application/json; charset=utf-8"}, + timeout=request_timeout, + verify=False + ) self.log(f"回传结果: {response.text}") data = response.json() if response.text else {} if response.status_code == 200 and isinstance(data, dict) and data.get("success"): diff --git a/backend-java/src/main/java/com/nanri/aiimage/AiImageApplication.java b/backend-java/src/main/java/com/nanri/aiimage/AiImageApplication.java index f86b27e..e00444b 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/AiImageApplication.java +++ b/backend-java/src/main/java/com/nanri/aiimage/AiImageApplication.java @@ -4,11 +4,17 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableScheduling; +import java.time.ZoneId; +import java.util.TimeZone; + @SpringBootApplication @EnableScheduling public class AiImageApplication { + private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai"); + public static void main(String[] args) { + TimeZone.setDefault(TimeZone.getTimeZone(BUSINESS_ZONE)); SpringApplication.run(AiImageApplication.class, args); } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java b/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java index 40f1772..0821a3e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java @@ -20,6 +20,12 @@ public class DistributedJobLockService { + "else return 0 end", Long.class ); + private static final DefaultRedisScript RENEW_SCRIPT = new DefaultRedisScript<>( + "if redis.call('get', KEYS[1]) == ARGV[1] then " + + "return redis.call('pexpire', KEYS[1], ARGV[2]) " + + "else return 0 end", + Long.class + ); private final StringRedisTemplate stringRedisTemplate; private final String ownerPrefix; @@ -74,6 +80,25 @@ public class DistributedJobLockService { log.warn("[job-lock] release failed jobName={} key={} msg={}", jobName, key, ex.getMessage()); } } + + public boolean renew(Duration ttl) { + if (released) { + return false; + } + Duration actualTtl = (ttl == null || ttl.isNegative() || ttl.isZero()) ? Duration.ofMinutes(10) : ttl; + try { + Long renewed = stringRedisTemplate.execute( + RENEW_SCRIPT, + Collections.singletonList(key), + token, + String.valueOf(actualTtl.toMillis()) + ); + return Long.valueOf(1L).equals(renewed); + } catch (Exception ex) { + log.warn("[job-lock] renew failed jobName={} key={} msg={}", jobName, key, ex.getMessage()); + return false; + } + } } private String buildLockKey(String jobName) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java index f93fc47..daf0116 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java @@ -6,15 +6,21 @@ import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import java.time.Clock; +import java.time.ZoneId; + @Configuration @Slf4j public class SchedulingConfig { + private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai"); + @Bean public TaskScheduler taskScheduler() { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.setPoolSize(4); scheduler.setThreadNamePrefix("aiimage-scheduling-"); + scheduler.setClock(Clock.system(BUSINESS_ZONE)); scheduler.setWaitForTasksToCompleteOnShutdown(true); scheduler.setAwaitTerminationSeconds(30); scheduler.setErrorHandler(ex -> log.warn("[scheduling] task execution failed: {}", ex.getMessage(), ex)); 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 new file mode 100644 index 0000000..25e8365 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java @@ -0,0 +1,78 @@ +package com.nanri.aiimage.config; + +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.servlet.HandlerInterceptor; +import org.springframework.web.servlet.config.annotation.InterceptorRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; + +import java.util.Locale; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +@Configuration +@RequiredArgsConstructor +public class TaskOperationLockConfig implements WebMvcConfigurer { + + private final TaskDistributedLockService taskDistributedLockService; + + @Override + public void addInterceptors(InterceptorRegistry registry) { + registry.addInterceptor(new TaskOperationLockInterceptor(taskDistributedLockService)) + .addPathPatterns("/api/**"); + } + + @Slf4j + private static final class TaskOperationLockInterceptor implements HandlerInterceptor { + + private static final Pattern TASK_PATH = Pattern.compile(".*/api/([^/]+)/tasks/(\\d+)(?:/.*)?$"); + private static final Set MUTATING_METHODS = Set.of("POST", "PUT", "PATCH", "DELETE"); + private static final String LOCK_ATTRIBUTE = TaskOperationLockInterceptor.class.getName() + ".LOCK"; + + private final TaskDistributedLockService taskDistributedLockService; + + private TaskOperationLockInterceptor(TaskDistributedLockService taskDistributedLockService) { + this.taskDistributedLockService = taskDistributedLockService; + } + + @Override + public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) { + if (!MUTATING_METHODS.contains(request.getMethod())) { + return true; + } + String uri = request.getRequestURI(); + Matcher matcher = TASK_PATH.matcher(uri == null ? "" : uri); + if (!matcher.matches()) { + return true; + } + String moduleType = matcher.group(1).trim().toUpperCase(Locale.ROOT).replace('-', '_'); + Long taskId = Long.valueOf(matcher.group(2)); + TaskDistributedLockService.LockHandle lockHandle = + taskDistributedLockService.acquire(moduleType, taskId, TaskDistributedLockService.DEFAULT_WAIT_MILLIS); + if (lockHandle == null) { + log.info("[task-operation-lock] rejected busy task operation method={} uri={} moduleType={} taskId={}", + request.getMethod(), uri, moduleType, taskId); + throw new BusinessException(40902, "TASK_LOCK_BUSY"); + } + request.setAttribute(LOCK_ATTRIBUTE, lockHandle); + return true; + } + + @Override + public void afterCompletion(HttpServletRequest request, + HttpServletResponse response, + Object handler, + Exception ex) { + Object lockHandle = request.getAttribute(LOCK_ATTRIBUTE); + if (lockHandle instanceof TaskDistributedLockService.LockHandle handle) { + handle.close(); + } + } + } +} 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 f14f6a6..ce85200 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 @@ -6,6 +6,7 @@ import com.nanri.aiimage.config.AppearancePatentProperties; import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto; import lombok.RequiredArgsConstructor; 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; @@ -24,6 +25,8 @@ import java.util.Map; @Slf4j public class AppearancePatentCozeClient { + private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + private final AppearancePatentProperties properties; private final ObjectMapper objectMapper; @@ -216,7 +219,8 @@ public class AppearancePatentCozeClient { .uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath())) .headers(headers -> { headers.setBearerAuth(stripBearer(properties.getCozeToken())); - headers.setContentType(MediaType.APPLICATION_JSON); + headers.setContentType(APPLICATION_JSON_UTF8); + headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name()); }); request.body(body); return request.exchange((clientRequest, clientResponse) -> { @@ -237,7 +241,8 @@ public class AppearancePatentCozeClient { .uri(joinUrl(properties.getCozeBaseUrl(), path)) .headers(headers -> { headers.setBearerAuth(stripBearer(properties.getCozeToken())); - headers.setContentType(MediaType.APPLICATION_JSON); + headers.setContentType(APPLICATION_JSON_UTF8); + headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name()); }) .exchange((clientRequest, clientResponse) -> { byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody()); 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 1a8794e..5a3f5a4 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 @@ -38,6 +38,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity; import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskFileJobService; import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; @@ -71,6 +72,7 @@ import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; import java.util.ArrayList; +import java.util.Comparator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -98,6 +100,9 @@ public class AppearancePatentTaskService { 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 long TASK_LOCK_RETRY_DELAY_MILLIS = 200L; private static final List RESULT_HEADERS = List.of( "id", "asin", @@ -125,6 +130,7 @@ public class AppearancePatentTaskService { private final TransientPayloadStorageService transientPayloadStorageService; private final PlatformTransactionManager transactionManager; private final DistributedJobLockService distributedJobLockService; + private final TaskDistributedLockService taskDistributedLockService; @Autowired @Qualifier("cozeTaskExecutor") private TaskExecutor cozeTaskExecutor; @@ -296,24 +302,41 @@ public class AppearancePatentTaskService { .eq(FileResultEntity::getUserId, userId) .orderByDesc(FileResultEntity::getCreatedAt) .last("limit " + safeLimit)); - Map statusMap = new LinkedHashMap<>(); + Map taskMap = new LinkedHashMap<>(); List taskIds = rows.stream().map(FileResultEntity::getTaskId).filter(Objects::nonNull).distinct().toList(); if (!taskIds.isEmpty()) { for (FileTaskEntity task : fileTaskMapper.selectList(new LambdaQueryWrapper() - .select(FileTaskEntity::getId, FileTaskEntity::getStatus) + .select(FileTaskEntity::getId, + FileTaskEntity::getStatus, + FileTaskEntity::getCreatedAt, + FileTaskEntity::getUpdatedAt, + FileTaskEntity::getFinishedAt) .in(FileTaskEntity::getId, taskIds))) { - statusMap.put(task.getId(), task.getStatus()); + taskMap.put(task.getId(), task); } } Map jobMap = taskFileJobService.findAssembleJobsByResultIds(MODULE_TYPE, rows.stream() .map(FileResultEntity::getId) .filter(Objects::nonNull) .toList()); + List sortedRows = new ArrayList<>(); for (FileResultEntity row : rows) { - String taskStatus = statusMap.get(row.getTaskId()); + FileTaskEntity task = taskMap.get(row.getTaskId()); + String taskStatus = task == null ? null : task.getStatus(); if (STATUS_PENDING.equals(taskStatus)) { continue; } + sortedRows.add(row); + } + sortedRows.sort(Comparator + .comparingInt((FileResultEntity row) -> historyPriority(row, taskMap.get(row.getTaskId()), jobMap.get(row.getId()))) + .thenComparing((FileResultEntity row) -> historyActivityTime(row, taskMap.get(row.getTaskId()), jobMap.get(row.getId())), + Comparator.nullsLast(Comparator.reverseOrder())) + .thenComparing(FileResultEntity::getCreatedAt, Comparator.nullsLast(Comparator.reverseOrder())) + .thenComparing(FileResultEntity::getId, Comparator.nullsLast(Comparator.reverseOrder()))); + for (FileResultEntity row : sortedRows) { + FileTaskEntity task = taskMap.get(row.getTaskId()); + String taskStatus = task == null ? null : task.getStatus(); vo.getItems().add(toHistoryItem(row, taskStatus, jobMap.get(row.getId()))); } return vo; @@ -384,6 +407,16 @@ public class AppearancePatentTaskService { } public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) { + TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, TASK_LOCK_WAIT_MILLIS); + if (lockHandle == null) { + throw new BusinessException(40902, "任务正在处理上一批结果，请稍后重试"); + } + try (lockHandle) { + submitResultLocked(taskId, request); + } + } + + private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) { if (transactionManager != null) { SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(taskId, request)); inNewTransaction(() -> { @@ -423,7 +456,7 @@ public class AppearancePatentTaskService { chunk.setScopeHash(scopeHash); chunk.setChunkIndex(chunkIndex); chunk.setChunkTotal(chunkTotal); - String storedPayload = transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson); chunk.setPayloadJson(storedPayload); chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson)); chunk.setCreatedAt(LocalDateTime.now()); @@ -536,10 +569,16 @@ public class AppearancePatentTaskService { if (heartbeatMillis > thresholdMillis) { continue; } - inNewTransaction(() -> { - finalizeStaleTask(task.getId(), "Python interrupted before uploading final appearance patent result"); - return null; - }); + TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(task.getId(), 0L); + if (lockHandle == null) { + continue; + } + try (lockHandle) { + inNewTransaction(() -> { + finalizeStaleTask(task.getId(), "Python interrupted before uploading final appearance patent result"); + return null; + }); + } } return; } @@ -559,7 +598,16 @@ public class AppearancePatentTaskService { if (heartbeatMillis > thresholdMillis) { continue; } - finalizeTask(task, "Python interrupted before uploading final appearance patent result", allRowCount(task), true); + TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(task.getId(), 0L); + if (lockHandle == null) { + continue; + } + try (lockHandle) { + FileTaskEntity latestTask = fileTaskMapper.selectById(task.getId()); + if (latestTask != null) { + finalizeTask(latestTask, "Python interrupted before uploading final appearance patent result", allRowCount(latestTask), true); + } + } } } @@ -596,7 +644,7 @@ public class AppearancePatentTaskService { chunk.setScopeHash(scopeHash); chunk.setChunkIndex(chunkIndex); chunk.setChunkTotal(chunkTotal); - String storedPayload = transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson); chunk.setPayloadJson(storedPayload); chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson)); chunk.setCreatedAt(LocalDateTime.now()); @@ -784,7 +832,7 @@ public class AppearancePatentTaskService { String payloadJson = writeJson(rows == null ? List.of() : rows, "结果序列化失败"); payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "appearance patent chunk payload merge failed"); String oldPayload = chunk.getPayloadJson(); - String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + String storedPayload = storeSharedChunkPayloadVersioned(taskId, scopeHash, chunkIndex, payloadJson); chunk.setPayloadJson(storedPayload); chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson)); chunk.setUpdatedAt(LocalDateTime.now()); @@ -876,7 +924,7 @@ public class AppearancePatentTaskService { } List representatives = new ArrayList<>(); for (List siblings : groupedRows.values()) { - boolean alreadyResolved = siblings.stream().anyMatch(this::hasResolvedCozeFields); + boolean alreadyResolved = siblings.stream().anyMatch(this::hasCompleteCozeResult); if (alreadyResolved) { continue; } @@ -1064,6 +1112,18 @@ public class AppearancePatentTaskService { if (job == null || job.getTaskId() == null || job.getResultId() == null) { throw new BusinessException("result file job arguments are incomplete"); } + 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"); + touchJavaSideTaskActivity(job.getTaskId()); + return false; + } + try (lockHandle) { + return processResultFileJobLocked(job); + } + } + + private boolean processResultFileJobLocked(TaskFileJobEntity job) { FileTaskEntity task = fileTaskMapper.selectById(job.getTaskId()); if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { throw new BusinessException("task not found"); @@ -1076,9 +1136,15 @@ public class AppearancePatentTaskService { .eq(TaskChunkEntity::getTaskId, task.getId()) .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) .orderByAsc(TaskChunkEntity::getChunkIndex)); - Map> allRowsByBaseId = loadAllRowsByBaseId(task); int cozeWorkUnits = countCozeWorkUnits(chunks, Math.max(1, properties.getCozeBatchSize())); int totalProgressUnits = Math.max(3, cozeWorkUnits + 3); + if (countPendingCozeStates(task.getId()) > 0) { + taskFileJobService.touchRunning(job.getId()); + touchJavaSideTaskActivity(task.getId()); + saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze submitted, waiting for result"); + return false; + } + Map> allRowsByBaseId = loadAllRowsByBaseId(task); saveFileBuildProgress(task, job, totalProgressUnits, 0, "Submitting Coze"); boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId); if (pendingCoze) { @@ -1113,7 +1179,18 @@ public class AppearancePatentTaskService { if (state == null || state.getId() == null) { continue; } - cozeTaskExecutor.execute(() -> pollPendingCozeState(state.getId())); + 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); + } } } } @@ -1231,8 +1308,7 @@ public class AppearancePatentTaskService { batchTotal ); String batchPayload = writeJson(batchRows, "serialize coze batch payload failed"); - String storedBatchPayload = transientPayloadStorageService.storeParsedPayloadFast( - MODULE_TYPE, task.getId(), batchScopeHash, batchPayload, true); + String storedBatchPayload = storeSharedCozeBatchPayload(task.getId(), batchScopeHash, batchPayload); TaskScopeStateEntity state = new TaskScopeStateEntity(); state.setTaskId(task.getId()); state.setModuleType(MODULE_TYPE); @@ -1260,25 +1336,53 @@ public class AppearancePatentTaskService { } private void pollPendingCozeState(Long stateId) { - TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId); - if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) { + Long taskIdForLock = null; + TaskScopeStateEntity lockState = taskScopeStateMapper.selectById(stateId); + if (lockState != null) { + taskIdForLock = lockState.getTaskId(); + } + TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskIdForLock, 0L); + if (lockHandle == null) { + if (taskIdForLock != null) { + log.info("[appearance-patent] coze poll skipped because task is locked taskId={} stateId={}", + taskIdForLock, stateId); + } return; } - if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) { - return; + try (lockHandle) { + pollPendingCozeStateLocked(stateId); } - if (!tryClaimCozeStateForPoll(state)) { - return; - } - CozeBatchContext context = readCozeBatchContext(state); - if (context == null || context.jobId() == null || context.resultId() == null) { - markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing"); - return; - } - taskFileJobService.touchRunning(context.jobId()); + } + + private void pollPendingCozeStateLocked(Long stateId) { try { + TaskScopeStateEntity 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()))) { + return; + } + if (!tryClaimCozeStateForPoll(state)) { + log.info("[appearance-patent] coze poll skipped by claim guard taskId={} stateId={} executeId={} lastPolledAt={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), state.getCozeLastPolledAt()); + return; + } + CozeBatchContext context = readCozeBatchContext(state); + if (context == null || context.jobId() == null || context.resultId() == null) { + log.warn("[appearance-patent] coze poll aborted because batch context is missing taskId={} stateId={} executeId={} stateJson={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), abbreviate(state.getStateJson(), 300)); + markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing"); + return; + } + taskFileJobService.touchRunning(context.jobId()); + log.info("[appearance-patent] coze poll start taskId={} stateId={} executeId={} jobId={} chunk={} batch={}/{}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), + context.jobId(), context.chunkIndex(), context.batchIndex(), context.batchTotal()); AppearancePatentCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(state.getCozeExecuteId()); if (!poll.hasPayload() && !poll.isFinished() && !isCozeStateTimedOut(state)) { + log.info("[appearance-patent] coze poll pending taskId={} stateId={} executeId={} status={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), poll.status()); updateCozeStateRunning(state, null); return; } @@ -1291,6 +1395,9 @@ public class AppearancePatentTaskService { : "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); @@ -1302,27 +1409,38 @@ public class AppearancePatentTaskService { markCozeStateTerminal(state, failureMessage.isBlank() ? COZE_STATUS_DONE : COZE_STATUS_FAILED, failureMessage.isBlank() ? null : failureMessage); - maybeFinalizeCozeJob(state.getTaskId(), context); + log.info("[appearance-patent] coze poll completed taskId={} stateId={} executeId={} status={} batchRows={} mergedRows={} failure={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), + failureMessage.isBlank() ? COZE_STATUS_DONE : COZE_STATUS_FAILED, + batchRows.size(), cozeRows.size(), firstNonBlank(failureMessage, "-")); + maybeFinalizeCozeJobLocked(state.getTaskId(), context); } catch (Exception ex) { + TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId); String message = firstNonBlank(ex.getMessage(), "Coze poll failed"); - if (isCozeStateTimedOut(state)) { - List batchRows = readCozeBatchRows(state); - FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId()); - if (task != null) { - Map> allRowsByBaseId = loadAllRowsByBaseId(task); - mergeCozeRowsIntoChunk(task, - context.chunkScopeHash(), - context.chunkIndex(), - cozeClient.markRowsFailed(batchRows, message), - allRowsByBaseId); + if (state != null) { + CozeBatchContext context = readCozeBatchContext(state); + if (isCozeStateTimedOut(state) && context != null) { + List batchRows = readCozeBatchRows(state); + FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId()); + if (task != null) { + Map> allRowsByBaseId = loadAllRowsByBaseId(task); + mergeCozeRowsIntoChunk(task, + context.chunkScopeHash(), + context.chunkIndex(), + cozeClient.markRowsFailed(batchRows, message), + allRowsByBaseId); + } + markCozeStateTerminal(state, COZE_STATUS_FAILED, message); + maybeFinalizeCozeJobLocked(state.getTaskId(), context); + return; } - markCozeStateTerminal(state, COZE_STATUS_FAILED, message); - maybeFinalizeCozeJob(state.getTaskId(), context); + log.warn("[appearance-patent] coze poll failed taskId={} stateId={} executeId={} err={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), message); + updateCozeStateRunning(state, message); return; } - log.warn("[appearance-patent] coze poll failed taskId={} stateId={} executeId={} err={}", - state.getTaskId(), state.getId(), state.getCozeExecuteId(), message); - updateCozeStateRunning(state, message); + log.warn("[appearance-patent] coze poll crashed before state refresh stateId={} err={}", + stateId, message, ex); } } @@ -1376,6 +1494,19 @@ public class AppearancePatentTaskService { } private void maybeFinalizeCozeJob(Long taskId, CozeBatchContext context) { + if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) { + return; + } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(taskId, 0L); + if (taskLockHandle == null) { + return; + } + try (taskLockHandle) { + maybeFinalizeCozeJobLocked(taskId, context); + } + } + + private void maybeFinalizeCozeJobLocked(Long taskId, CozeBatchContext context) { if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) { return; } @@ -1388,19 +1519,15 @@ public class AppearancePatentTaskService { if (countPendingCozeStates(taskId) > 0) { return; } - FileTaskEntity task = fileTaskMapper.selectById(taskId); - FileResultEntity result = fileResultMapper.selectById(context.resultId()); TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId()); - if (task == null || result == null || job == null || "SUCCESS".equals(job.getStatus())) { + if (job == null || "SUCCESS".equals(job.getStatus())) { return; } - int cozeWorkUnits = countCompletedCozeStates(taskId); - int totalProgressUnits = Math.max(3, cozeWorkUnits + 3); - completeCozeFileJob(task, result, job, totalProgressUnits, cozeWorkUnits); - taskFileJobService.markSuccess(job, result.getResultFileUrl()); - cleanupResultFileJob(job); - log.info("[appearance-patent] coze async job finalized taskId={} jobId={} resultId={} resultFileUrl={}", - taskId, job.getId(), result.getId(), result.getResultFileUrl()); + boolean requeued = taskFileJobService.requeue(job.getId(), "Coze results ready, assembling xlsx"); + if (requeued) { + log.info("[appearance-patent] coze async results ready, result file job requeued taskId={} jobId={} resultId={}", + taskId, job.getId(), context.resultId()); + } } catch (Exception ex) { TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId()); if (job != null) { @@ -1416,6 +1543,9 @@ public class AppearancePatentTaskService { TaskFileJobEntity job, int totalProgressUnits, int cozeWorkUnits) { + if (countPendingCozeStates(task.getId()) > 0) { + throw new BusinessException("Coze 结果仍在处理中，暂不能生成结果文件"); + } int assembleProgress = Math.max(1, Math.min(totalProgressUnits - 2, cozeWorkUnits)); saveFileBuildProgress(task, job, totalProgressUnits, assembleProgress, "Assembling xlsx"); assembleResultWorkbook(task, result); @@ -1534,6 +1664,20 @@ public class AppearancePatentTaskService { + ":batch:" + batchIndex; } + private String abbreviate(String value, int maxLength) { + if (value == null) { + return ""; + } + int normalizedMaxLength = Math.max(0, maxLength); + if (value.length() <= normalizedMaxLength) { + return value; + } + if (normalizedMaxLength <= 3) { + return value.substring(0, normalizedMaxLength); + } + return value.substring(0, normalizedMaxLength - 3) + "..."; + } + private int countCozeWorkUnits(List chunks, int batchSize) { if (chunks == null || chunks.isEmpty()) { return 0; @@ -1579,6 +1723,11 @@ public class AppearancePatentTaskService { if (job == null || job.getTaskId() == null) { return; } + if (countPendingCozeStates(job.getTaskId()) > 0) { + log.warn("[appearance-patent] skip cleanup because coze is still pending taskId={} jobId={}", + job.getTaskId(), job.getId()); + return; + } deleteTransientTaskPayloads(job.getTaskId()); taskChunkMapper.delete(new LambdaQueryWrapper() .eq(TaskChunkEntity::getTaskId, job.getTaskId()) @@ -1591,17 +1740,19 @@ public class AppearancePatentTaskService { private void assembleResultWorkbook(FileTaskEntity task, FileResultEntity result) { AppearancePatentParsedPayloadDto parsed = readParsedPayload(task); Map resultMap = loadPersistedResultRowsWithRetry(task.getId(), parsed.getAllItems().size()); - long resolvedRows = parsed.getAllItems().stream() - .filter(row -> findResultRow(row, resultMap) != null || findResultRowByAsin(row.getAsin(), resultMap) != null) + List receivedRows = filterReceivedParsedRows(parsed.getAllItems(), resultMap); + long resolvedRows = receivedRows.stream() + .filter(row -> findResultRow(row, resultMap) != null) .count(); long reasonRows = resultMap.values().stream() .filter(this::hasReasonFields) .count(); - log.info("[appearance-patent] assemble workbook taskId={} parsedRows={} resultRows={} resolvedRows={} reasonRows={}", - task.getId(), parsed.getAllItems().size(), resultMap.size(), resolvedRows, reasonRows); - if (!parsed.getAllItems().isEmpty() && resultMap.isEmpty()) { + log.info("[appearance-patent] assemble workbook taskId={} parsedRows={} receivedRows={} resultRows={} resolvedRows={} reasonRows={}", + task.getId(), parsed.getAllItems().size(), receivedRows.size(), resultMap.size(), resolvedRows, reasonRows); + if (!parsed.getAllItems().isEmpty() && receivedRows.isEmpty()) { throw new BusinessException("外观专利检测结果为空，请稍后重试生成结果文件"); } + validateCompleteCozeCoverage(task.getId(), receivedRows, resultMap); File outputDir = new File(storageProperties.getLocalTempDir(), "appearance-patent-result"); if (!outputDir.exists() && !outputDir.mkdirs()) { throw new BusinessException("创建结果目录失败"); @@ -1614,13 +1765,13 @@ public class AppearancePatentTaskService { + "-result.xlsx"; File xlsx = new File(outputDir, tempFilename); try { - writeResultWorkbook(xlsx, parsed, resultMap); + writeResultWorkbook(xlsx, parsed, receivedRows, resultMap); String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE); result.setResultFilename(filename); result.setResultFileUrl(objectKey); result.setResultFileSize(xlsx.length()); result.setResultContentType(CONTENT_TYPE_XLSX); - result.setRowCount(parsed.getAllItems().size()); + result.setRowCount(receivedRows.size()); } finally { if (xlsx.exists() && !xlsx.delete()) { log.warn("[appearance-patent] delete temp xlsx failed file={}", xlsx); @@ -1628,6 +1779,56 @@ public class AppearancePatentTaskService { } } + private List filterReceivedParsedRows(List parsedRows, + Map resultMap) { + if (parsedRows == null || parsedRows.isEmpty() || resultMap == null || resultMap.isEmpty()) { + return List.of(); + } + List receivedRows = new ArrayList<>(); + for (AppearancePatentParsedRowVo parsedRow : parsedRows) { + if (findResultRow(parsedRow, resultMap) != null) { + receivedRows.add(parsedRow); + } + } + return receivedRows; + } + + private void validateCompleteCozeCoverage(Long taskId, + List receivedRows, + Map resultMap) { + if (receivedRows == null || receivedRows.isEmpty()) { + return; + } + int expectedRows = 0; + int missingRows = 0; + List sampleAsins = new ArrayList<>(); + for (AppearancePatentParsedRowVo parsedRow : receivedRows) { + if (!hasPromptFields(parsedRow)) { + continue; + } + expectedRows++; + AppearancePatentResultRowDto resultRow = findResultRow(parsedRow, resultMap); + if (resultRow == null || !hasCompleteCozeResult(resultRow)) { + missingRows++; + if (sampleAsins.size() < 5) { + sampleAsins.add(firstNonBlank(parsedRow.getAsin(), firstNonBlank(parsedRow.getDisplayId(), ""))); + } + } + } + if (expectedRows > 0 && missingRows > 0) { + log.warn("[appearance-patent] incomplete coze coverage taskId={} expectedRows={} missingRows={} samples={}", + taskId, expectedRows, missingRows, sampleAsins); + throw new BusinessException("Coze 结果不完整：缺少 " + missingRows + "/" + expectedRows + " 条检测结果，请等待重试或重新运行任务"); + } + } + + private boolean hasCompleteCozeResult(AppearancePatentResultRowDto row) { + return hasUsableCozeField(row.getTitleRisk()) + && hasUsableCozeField(row.getAppearanceRisk()) + && hasUsableCozeField(row.getPatentRisk()) + && hasUsableCozeField(row.getConclusion()); + } + private Map loadPersistedResultRows(Long taskId) { Map result = new LinkedHashMap<>(); List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() @@ -1667,7 +1868,10 @@ public class AppearancePatentTaskService { } } - private void writeResultWorkbook(File xlsx, AppearancePatentParsedPayloadDto parsed, Map resultMap) { + private void writeResultWorkbook(File xlsx, + AppearancePatentParsedPayloadDto parsed, + List receivedRows, + Map resultMap) { try (SXSSFWorkbook workbook = new SXSSFWorkbook(200); FileOutputStream fos = new FileOutputStream(xlsx)) { Sheet sheet = workbook.createSheet("外观专利检测结果"); CellStyle headerStyle = workbook.createCellStyle(); @@ -1686,11 +1890,9 @@ public class AppearancePatentTaskService { } int rowIndex = 1; - for (AppearancePatentParsedRowVo parsedRow : parsed.getAllItems()) { + List rowsToWrite = receivedRows == null ? List.of() : receivedRows; + for (AppearancePatentParsedRowVo parsedRow : rowsToWrite) { AppearancePatentResultRowDto resultRow = findResultRow(parsedRow, resultMap); - if (resultRow == null) { - resultRow = findResultRowByAsin(parsedRow.getAsin(), resultMap); - } String missingReason = ""; if (resultRow == null) { missingReason = hasPromptFields(parsedRow) ? "未匹配到检测结果" : "未送检：缺少标题或图片"; @@ -1708,7 +1910,7 @@ public class AppearancePatentTaskService { row.createCell(col++).setCellValue(resultRow == null ? missingReason : userFacingCozeCellValue(resultRow, resultRow.getPatentRisk())); row.createCell(col).setCellValue(resultRow == null ? "未送检" : userFacingConclusion(resultRow)); } - writeReasonSheet(workbook, headerStyle, parsed, resultMap); + writeReasonSheet(workbook, headerStyle, rowsToWrite, resultMap); workbook.write(fos); workbook.dispose(); } catch (Exception ex) { @@ -1718,7 +1920,7 @@ public class AppearancePatentTaskService { private void writeReasonSheet(SXSSFWorkbook workbook, CellStyle headerStyle, - AppearancePatentParsedPayloadDto parsed, + List rowsToWrite, Map resultMap) { Sheet sheet = workbook.createSheet("原因"); Row header = sheet.createRow(0); @@ -1731,15 +1933,12 @@ public class AppearancePatentTaskService { Set writtenAsins = new LinkedHashSet<>(); int rowIndex = 1; - for (AppearancePatentParsedRowVo parsedRow : parsed.getAllItems()) { + for (AppearancePatentParsedRowVo parsedRow : rowsToWrite == null ? List.of() : rowsToWrite) { String asin = normalize(parsedRow.getAsin()).toUpperCase(Locale.ROOT); if (asin.isBlank() || !writtenAsins.add(asin)) { continue; } AppearancePatentResultRowDto 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(), "")); @@ -2072,6 +2271,36 @@ public class AppearancePatentTaskService { return vo; } + private int historyPriority(FileResultEntity row, FileTaskEntity task, TaskFileJobEntity job) { + String taskStatus = task == null ? null : task.getStatus(); + if (STATUS_RUNNING.equals(taskStatus) || isHistoryFileBuilding(row, taskStatus, job)) { + return 0; + } + return 1; + } + + private LocalDateTime historyActivityTime(FileResultEntity row, FileTaskEntity task, TaskFileJobEntity job) { + LocalDateTime latest = latestTime( + task == null ? null : task.getUpdatedAt(), + job == null ? null : job.getUpdatedAt(), + task == null ? null : task.getFinishedAt(), + row == null ? null : row.getCreatedAt(), + task == null ? null : task.getCreatedAt()); + return latest == null && row != null ? row.getCreatedAt() : latest; + } + + private boolean isHistoryFileBuilding(FileResultEntity row, String taskStatus, TaskFileJobEntity job) { + if (!STATUS_SUCCESS.equals(taskStatus)) { + return false; + } + boolean fileReady = row != null && row.getResultFileUrl() != null && !row.getResultFileUrl().isBlank(); + if (fileReady) { + return false; + } + String fileStatus = job == null ? null : job.getStatus(); + return !STATUS_SUCCESS.equals(fileStatus) && !STATUS_FAILED.equals(fileStatus); + } + private void attachFileJobState(AppearancePatentHistoryItemVo vo, FileResultEntity row, TaskFileJobEntity job) { vo.setFileReady(row.getResultFileUrl() != null && !row.getResultFileUrl().isBlank()); if (job == null) { @@ -2126,6 +2355,22 @@ public class AppearancePatentTaskService { return t == null ? null : t.toString(); } + private LocalDateTime latestTime(LocalDateTime... values) { + LocalDateTime latest = null; + if (values == null) { + return null; + } + for (LocalDateTime value : values) { + if (value == null) { + continue; + } + if (latest == null || value.isAfter(latest)) { + latest = value; + } + } + return latest; + } + private String firstNonBlank(String preferred, String fallback) { return preferred == null || preferred.isBlank() ? fallback : preferred.trim(); } @@ -2157,9 +2402,35 @@ public class AppearancePatentTaskService { } private String storeParsedPayload(Long taskId, String scopeHash, String parsedPayloadJson) { + requireSharedTransientPayloadStorage("parsed payload"); return transientPayloadStorageService.storeParsedPayloadFast(MODULE_TYPE, taskId, scopeHash, parsedPayloadJson, false); } + private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) { + return taskDistributedLockService.acquire(MODULE_TYPE, taskId, TASK_LOCK_TTL, waitMillis); + } + + private String storeSharedChunkPayload(Long taskId, String scopeHash, Integer chunkIndex, String payloadJson) { + requireSharedTransientPayloadStorage("chunk payload"); + return transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + } + + private String storeSharedChunkPayloadVersioned(Long taskId, String scopeHash, Integer chunkIndex, String payloadJson) { + requireSharedTransientPayloadStorage("merged chunk payload"); + return transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + } + + private String storeSharedCozeBatchPayload(Long taskId, String scopeHash, String payloadJson) { + requireSharedTransientPayloadStorage("coze batch payload"); + return transientPayloadStorageService.storeParsedPayloadFast(MODULE_TYPE, taskId, scopeHash, payloadJson, true); + } + + private void requireSharedTransientPayloadStorage(String payloadType) { + if (!transientPayloadStorageService.isSharedWriteEnabled()) { + throw new BusinessException("外观专利" + payloadType + "必须写入共享临时存储 RustFS，请检查 aiimage.transient-storage 配置"); + } + } + private long elapsedMs(long start, long end) { return (end - start) / 1_000_000L; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index bb5a2d7..7ca724e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -32,6 +32,7 @@ import com.nanri.aiimage.modules.brand.model.vo.LegacyBrandTaskItemVo; import com.nanri.aiimage.modules.brand.model.vo.LegacyBrandTaskListVo; import com.nanri.aiimage.modules.file.service.oss.OssStorageService; import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskFileJobService; import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; import lombok.RequiredArgsConstructor; @@ -89,6 +90,7 @@ public class BrandTaskService { private final BrandTaskProgressCacheService brandTaskProgressCacheService; private final BrandTaskStorageService brandTaskStorageService; private final DistributedJobLockService distributedJobLockService; + private final TaskDistributedLockService taskDistributedLockService; private final ObjectMapper objectMapper; private final TaskFileJobService taskFileJobService; private final TaskProgressSnapshotService taskProgressSnapshotService; @@ -1211,6 +1213,11 @@ public class BrandTaskService { .eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING) .lt(BrandCrawlTaskEntity::getUpdatedAt, threshold)); for (BrandCrawlTaskEntity task : runningTasks) { + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { if (tryRecoverCompletedRunningTask(task)) { continue; } @@ -1236,6 +1243,15 @@ public class BrandTaskService { } } } + } + + private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) { + TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis); + if (lockHandle == null) { + log.info("[brand-task-lock] skip task because lock is busy taskId={}", taskId); + } + return lockHandle; + } private boolean tryRecoverCompletedRunningTask(BrandCrawlTaskEntity task) { if (task == null || task.getId() == null) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java index e743487..d1f8e5a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java @@ -14,6 +14,7 @@ import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskCacheService; import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; @@ -58,6 +59,7 @@ public class DeleteBrandStaleTaskService { private final PatrolDeleteTaskCacheService patrolDeleteTaskCacheService; private final DeleteBrandProgressProperties deleteBrandProgressProperties; private final DistributedJobLockService distributedJobLockService; + private final TaskDistributedLockService taskDistributedLockService; @Value("${aiimage.temp-dir.retention-hours:24}") private long tempDirRetentionHours; @@ -148,6 +150,11 @@ public class DeleteBrandStaleTaskService { continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_DELETE_BRAND, task.getId()); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { try { deleteBrandRunService.tryFinalizeTask(task.getId(), true); FileTaskEntity refreshed = fileTaskMapper.selectById(task.getId()); @@ -184,6 +191,7 @@ public class DeleteBrandStaleTaskService { deleteBrandTaskCacheService.delete(task.getId()); log.warn("[stale-check] delete-brand failed taskId={} reason=timeout", task.getId()); } + } } } @@ -229,6 +237,12 @@ public class DeleteBrandStaleTaskService { task.getId(), task.getCreatedAt(), initialThreshold); continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PRODUCT_RISK, task.getId()); + if (taskLockHandle == null) { + stats.skippedTaskCount++; + continue; + } + try (taskLockHandle) { try { if (productRiskTaskService.tryFinalizeTask(task.getId(), true)) { stats.finalizedTaskCount++; @@ -252,6 +266,7 @@ public class DeleteBrandStaleTaskService { } else { stats.skippedTaskCount++; } + } } return stats; } @@ -294,6 +309,12 @@ public class DeleteBrandStaleTaskService { stats.skippedTaskCount++; continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PRICE_TRACK, task.getId()); + if (taskLockHandle == null) { + stats.skippedTaskCount++; + continue; + } + try (taskLockHandle) { try { if (priceTrackTaskService.tryFinalizeTask(task.getId(), true)) { stats.finalizedTaskCount++; @@ -318,6 +339,7 @@ public class DeleteBrandStaleTaskService { } else { stats.skippedTaskCount++; } + } } return stats; } @@ -364,6 +386,12 @@ public class DeleteBrandStaleTaskService { task.getId(), task.getCreatedAt(), initialThreshold); continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_SHOP_MATCH, task.getId()); + if (taskLockHandle == null) { + stats.skippedTaskCount++; + continue; + } + try (taskLockHandle) { try { if (shopMatchTaskService.tryFinalizeTask(task.getId(), true)) { stats.finalizedTaskCount++; @@ -388,6 +416,7 @@ public class DeleteBrandStaleTaskService { } else { stats.skippedTaskCount++; } + } } return stats; } @@ -428,6 +457,12 @@ public class DeleteBrandStaleTaskService { stats.skippedTaskCount++; continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PATROL_DELETE, task.getId()); + if (taskLockHandle == null) { + stats.skippedTaskCount++; + continue; + } + try (taskLockHandle) { try { if (patrolDeleteTaskService.tryFinalizeTask(task.getId(), true)) { stats.finalizedTaskCount++; @@ -451,6 +486,7 @@ public class DeleteBrandStaleTaskService { } else { stats.skippedTaskCount++; } + } } return stats; } @@ -463,6 +499,14 @@ public class DeleteBrandStaleTaskService { return baseline != null && baseline.isAfter(initialThreshold); } + private TaskDistributedLockService.LockHandle acquireTaskLock(String moduleType, Long taskId) { + TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(moduleType, taskId, 0L); + if (lockHandle == null) { + log.info("[task-lock] skip stale task because task lock is busy moduleType={} taskId={}", moduleType, taskId); + } + return lockHandle; + } + @Scheduled(cron = "${aiimage.delete-brand-progress.finalize-check-cron:30 */2 * * * *}") public void finalizeCompletedRunningTasks() { DistributedJobLockService.LockHandle lockHandle = @@ -483,11 +527,17 @@ public class DeleteBrandStaleTaskService { .last("limit 100")); for (FileTaskEntity task : runningTasks) { + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_DELETE_BRAND, task.getId()); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { try { deleteBrandRunService.tryFinalizeTask(task.getId(), true); } catch (Exception ignored) { // Keep compensation best-effort; the next schedule can retry. } + } } } } 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 0b9bdc3..ec4da81 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 @@ -6,6 +6,7 @@ import com.nanri.aiimage.config.SimilarAsinProperties; import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto; import lombok.RequiredArgsConstructor; 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; @@ -24,6 +25,8 @@ import java.util.Map; @Slf4j public class SimilarAsinCozeClient { + private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + private final SimilarAsinProperties properties; private final ObjectMapper objectMapper; @@ -216,7 +219,8 @@ public class SimilarAsinCozeClient { .uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath())) .headers(headers -> { headers.setBearerAuth(stripBearer(properties.getCozeToken())); - headers.setContentType(MediaType.APPLICATION_JSON); + headers.setContentType(APPLICATION_JSON_UTF8); + headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name()); }); request.body(body); return request.exchange((clientRequest, clientResponse) -> { @@ -237,7 +241,8 @@ public class SimilarAsinCozeClient { .uri(joinUrl(properties.getCozeBaseUrl(), path)) .headers(headers -> { headers.setBearerAuth(stripBearer(properties.getCozeToken())); - headers.setContentType(MediaType.APPLICATION_JSON); + headers.setContentType(APPLICATION_JSON_UTF8); + headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name()); }) .exchange((clientRequest, clientResponse) -> { byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody()); 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 1705920..5ee9461 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 @@ -39,6 +39,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity; import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; import lombok.RequiredArgsConstructor; @@ -121,6 +122,7 @@ public class SimilarAsinTaskService { private final SimilarAsinTaskCacheService taskCacheService; private final SimilarAsinProperties properties; private final TaskFileJobService taskFileJobService; + private final TaskDistributedLockService taskDistributedLockService; private final TaskProgressSnapshotService taskProgressSnapshotService; private final TransientPayloadStorageService transientPayloadStorageService; private final PlatformTransactionManager transactionManager; @@ -536,10 +538,16 @@ public class SimilarAsinTaskService { if (heartbeatMillis > thresholdMillis) { continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { inNewTransaction(() -> { finalizeStaleTask(task.getId(), "Python interrupted before uploading final similar ASIN result"); return null; }); + } } return; } @@ -559,7 +567,13 @@ public class SimilarAsinTaskService { if (heartbeatMillis > thresholdMillis) { continue; } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { finalizeTask(task, "Python interrupted before uploading final similar ASIN result", allRowCount(task), true); + } } } @@ -1271,6 +1285,18 @@ 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()))) { + return; + } if (!tryClaimCozeStateForPoll(state)) { return; } @@ -1328,6 +1354,7 @@ public class SimilarAsinTaskService { state.getTaskId(), state.getId(), state.getCozeExecuteId(), message); updateCozeStateRunning(state, message); } + } } private void updateCozeStateRunning(TaskScopeStateEntity state, String error) { @@ -1383,8 +1410,7 @@ public class SimilarAsinTaskService { if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) { return; } - DistributedJobLockService.LockHandle lockHandle = - distributedJobLockService.tryLock("similar-asin:coze-finalize:" + taskId, Duration.ofMinutes(5)); + TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, 0L); if (lockHandle == null) { return; } @@ -1415,6 +1441,14 @@ public class SimilarAsinTaskService { } } + private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) { + TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis); + if (lockHandle == null) { + log.info("[similar-asin] skip task operation because task lock is busy taskId={}", taskId); + } + return lockHandle; + } + private void completeCozeFileJob(FileTaskEntity task, FileResultEntity result, TaskFileJobEntity job, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java new file mode 100644 index 0000000..ac967f4 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java @@ -0,0 +1,160 @@ +package com.nanri.aiimage.modules.task.service; + +import com.nanri.aiimage.common.service.DistributedJobLockService; +import jakarta.annotation.PreDestroy; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +@Service +@RequiredArgsConstructor +@Slf4j +public class TaskDistributedLockService { + + public static final Duration DEFAULT_LOCK_TTL = Duration.ofMinutes(5); + public static final long DEFAULT_WAIT_MILLIS = 10000L; + private static final long RETRY_DELAY_MILLIS = 200L; + + private final DistributedJobLockService distributedJobLockService; + private final ThreadLocal> localLocks = ThreadLocal.withInitial(LinkedHashMap::new); + private final ScheduledExecutorService renewalExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "task-lock-renewal"); + thread.setDaemon(true); + return thread; + }); + + public LockHandle acquire(String moduleType, Long taskId) { + return acquire(moduleType, taskId, DEFAULT_WAIT_MILLIS); + } + + public LockHandle acquire(String moduleType, Long taskId, long waitMillis) { + return acquire(moduleType, taskId, DEFAULT_LOCK_TTL, waitMillis); + } + + public LockHandle acquire(String moduleType, Long taskId, Duration ttl, long waitMillis) { + String lockName = taskLockName(moduleType, taskId); + if (lockName == null) { + return null; + } + Map heldLocks = localLocks.get(); + HeldLock heldLock = heldLocks.get(lockName); + if (heldLock != null) { + heldLock.depth++; + return new LockHandle(lockName); + } + + long deadline = System.nanoTime() + Math.max(0L, waitMillis) * 1_000_000L; + do { + DistributedJobLockService.LockHandle delegate = distributedJobLockService.tryLock(lockName, ttl); + if (delegate != null) { + HeldLock newHeldLock = new HeldLock(delegate, ttl); + scheduleRenewal(lockName, newHeldLock, ttl); + heldLocks.put(lockName, newHeldLock); + return new LockHandle(lockName); + } + if (waitMillis <= 0L || System.nanoTime() >= deadline) { + return null; + } + try { + Thread.sleep(RETRY_DELAY_MILLIS); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return null; + } + } while (true); + } + + public String taskLockName(String moduleType, Long taskId) { + if (moduleType == null || moduleType.isBlank() || taskId == null || taskId <= 0) { + return null; + } + String normalizedModule = moduleType.trim().toLowerCase(Locale.ROOT).replace('_', '-'); + return "task:" + normalizedModule + ":" + taskId; + } + + private void scheduleRenewal(String lockName, HeldLock heldLock, Duration ttl) { + long ttlMillis = Math.max(1000L, ttl.toMillis()); + long renewalDelayMillis = Math.max(1000L, ttlMillis / 3L); + heldLock.renewalFuture = renewalExecutor.scheduleWithFixedDelay(() -> { + if (heldLock.closed) { + return; + } + boolean renewed = heldLock.delegate.renew(heldLock.ttl); + if (!renewed) { + heldLock.closed = true; + ScheduledFuture future = heldLock.renewalFuture; + if (future != null) { + future.cancel(false); + } + log.warn("[task-lock] renewal failed lockName={}", lockName); + } + }, renewalDelayMillis, renewalDelayMillis, TimeUnit.MILLISECONDS); + } + + @PreDestroy + public void shutdownRenewalExecutor() { + renewalExecutor.shutdownNow(); + } + + private static final class HeldLock { + + private final DistributedJobLockService.LockHandle delegate; + private final Duration ttl; + private ScheduledFuture renewalFuture; + private int depth = 1; + private volatile boolean closed; + + private HeldLock(DistributedJobLockService.LockHandle delegate, Duration ttl) { + this.delegate = delegate; + this.ttl = ttl; + } + + private void close() { + closed = true; + if (renewalFuture != null) { + renewalFuture.cancel(false); + } + delegate.close(); + } + } + + public final class LockHandle implements AutoCloseable { + + private final String lockName; + private boolean closed; + + private LockHandle(String lockName) { + this.lockName = lockName; + } + + @Override + public void close() { + if (closed) { + return; + } + closed = true; + Map heldLocks = localLocks.get(); + HeldLock heldLock = heldLocks.get(lockName); + if (heldLock == null) { + return; + } + heldLock.depth--; + if (heldLock.depth <= 0) { + heldLocks.remove(lockName); + if (heldLocks.isEmpty()) { + localLocks.remove(); + } + heldLock.close(); + } + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java index 0b210e3..e1cbdb1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java @@ -168,6 +168,26 @@ public class TaskFileJobService { .set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())); } + @Transactional + public boolean requeue(Long jobId, String message) { + if (jobId == null || jobId <= 0) { + return false; + } + int updated = taskFileJobMapper.update(null, new LambdaUpdateWrapper() + .eq(TaskFileJobEntity::getId, jobId) + .ne(TaskFileJobEntity::getStatus, "SUCCESS") + .set(TaskFileJobEntity::getStatus, "PENDING") + .set(TaskFileJobEntity::getErrorMessage, message) + .set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()) + .set(TaskFileJobEntity::getFinishedAt, null)); + if (updated <= 0) { + return false; + } + TaskFileJobEntity refreshed = taskFileJobMapper.selectById(jobId); + publishDispatchEvent(refreshed); + return true; + } + public void markSuccess(TaskFileJobEntity job, String resultFileUrl) { taskFileJobMapper.update(null, new LambdaUpdateWrapper() .eq(TaskFileJobEntity::getId, job.getId()) 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 1ebd668..a21ec42 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 @@ -29,6 +29,7 @@ import java.util.List; public class TaskResultFileJobWorker { private final TaskFileJobService taskFileJobService; + private final TaskDistributedLockService taskDistributedLockService; private final TaskResultPayloadService taskResultPayloadService; private final FileResultMapper fileResultMapper; private final TaskFileJobLocalDispatcher taskFileJobLocalDispatcher; @@ -103,20 +104,30 @@ public class TaskResultFileJobWorker { private void processInternal(TaskFileJobEntity job) { long startedAt = System.currentTimeMillis(); try { - boolean completed = dispatch(job); - if (!completed) { - taskFileJobService.touchRunning(job.getId()); - log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={}", - job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), - System.currentTimeMillis() - startedAt); + TaskDistributedLockService.LockHandle lockHandle = + taskDistributedLockService.acquire(job.getModuleType(), job.getTaskId(), TaskDistributedLockService.DEFAULT_WAIT_MILLIS); + if (lockHandle == null) { + taskFileJobService.requeue(job.getId(), "Task is busy, waiting for previous task operation"); + log.info("[task-file-job] process requeued because task lock is busy jobId={} taskId={} moduleType={} resultId={}", + job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId()); return; } - String resultFileUrl = resolveResultFileUrl(job); - taskFileJobService.markSuccess(job, resultFileUrl); - cleanupAfterSuccess(job); - log.info("[task-file-job] process success jobId={} taskId={} moduleType={} resultId={} elapsedMs={} resultFileUrl={}", - job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), - System.currentTimeMillis() - startedAt, resultFileUrl); + try (lockHandle) { + boolean completed = dispatch(job); + if (!completed) { + taskFileJobService.touchRunning(job.getId()); + log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={}", + job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), + System.currentTimeMillis() - startedAt); + return; + } + String resultFileUrl = resolveResultFileUrl(job); + taskFileJobService.markSuccess(job, resultFileUrl); + cleanupAfterSuccess(job); + log.info("[task-file-job] process success jobId={} taskId={} moduleType={} resultId={} elapsedMs={} resultFileUrl={}", + job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), + System.currentTimeMillis() - startedAt, resultFileUrl); + } } catch (Exception ex) { String message = ex.getMessage() == null ? "结果文件生成失败" : ex.getMessage(); log.warn("[task-file-job] process failed jobId={} taskId={} moduleType={} resultId={} msg={}", diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java index 297869b..032eec8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java @@ -33,7 +33,12 @@ public class TransientPayloadStorageService { private final ObjectMapper objectMapper; public boolean isWriteEnabled() { - return properties.isEnabled(); + return properties.isEnabled() + && (rustfsObjectStorageService.isConfigured() || storageProperties.getLocalTempDir() != null); + } + + public boolean isSharedWriteEnabled() { + return properties.isEnabled() && rustfsObjectStorageService.isConfigured(); } public String storeScopePayload(String moduleType, Long taskId, String scopeHash, String content, boolean encodeAsJsonString) { @@ -170,16 +175,19 @@ public class TransientPayloadStorageService { return content; } String objectKey = buildObjectKey(category, moduleType, taskId, scopeHash, entryKey); - String pointer = storeLocal(objectKey, content); - if (pointer == null && rustfsObjectStorageService.isConfigured()) { + String pointer = null; + if (rustfsObjectStorageService.isConfigured()) { try { pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, content, verifyAfterUpload); } catch (Exception ex) { - log.warn("[transient-payload] rustfs upload failed and local fallback unavailable objectKey={} err={}", + log.warn("[transient-payload] rustfs upload failed objectKey={} err={}", objectKey, ex.getMessage()); throw ex; } } + if (pointer == null) { + pointer = storeLocal(objectKey, content); + } if (pointer == null) { return content; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java index 7e15824..6345388 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java @@ -341,7 +341,8 @@ public class ZiniaoShopIndexService { cursor.setMessage(ex.getMessage()); cursor.setLastFinishedAt(now); if (isIpWhitelistRefreshFailure(ex)) { - markExistingEntriesRefreshBlocked(now, REFRESH_BLOCKED_REASON_IP_WHITELIST, ex.getMessage()); + log.warn("[ziniao-index] refresh blocked reason={} existing shop index rows left untouched", + REFRESH_BLOCKED_REASON_IP_WHITELIST); } ziniaoTransientCacheService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL); log.warn("[ziniao-index] refresh failed status=FAILED msg={}", ex.getMessage()); @@ -589,40 +590,6 @@ public class ZiniaoShopIndexService { return message != null && message.contains("白名单"); } - private void markExistingEntriesRefreshBlocked(long now, String reason, String message) { - List entities = ziniaoMemoryStoreService.listAliveEntitiesByType( - ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, - SHOP_INDEX_LIST_LIMIT - ); - if (entities.isEmpty()) { - return; - } - int updated = 0; - for (ZiniaoMemoryStoreEntity entity : entities) { - String rowKey = entity.getCacheKey(); - ZiniaoShopIndexEntryDto existingEntry; - try { - existingEntry = objectMapper.readValue(entity.getPayloadJson(), ZiniaoShopIndexEntryDto.class); - } catch (Exception ex) { - log.warn("[ziniao-index] skip refresh-block mark corrupt shop_index row cacheKey={}", rowKey); - continue; - } - if (existingEntry == null || !STATUS_ACTIVE.equals(existingEntry.getStatus())) { - continue; - } - existingEntry.setLastRefreshBlockedAt(now); - existingEntry.setRefreshBlockedReason(reason); - if (message != null && !message.isBlank()) { - existingEntry.setMessage(message); - } - ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, rowKey, existingEntry, resolveEntryTtl()); - updated++; - } - if (updated > 0) { - log.warn("[ziniao-index] refresh blocked reason={} existing active rows kept usable count={}", reason, updated); - } - } - private Duration resolveEntryTtl() { Integer ttlHours = ziniaoProperties.getShopIndexEntryTtlHours(); if (ttlHours == null || ttlHours <= 0) { diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index b089525..9248ad9 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -10,6 +10,8 @@ spring: multipart: max-file-size: 200MB max-request-size: 500MB + jackson: + time-zone: Asia/Shanghai datasource: driver-class-name: com.mysql.cj.jdbc.Driver url: ${AIIMAGE_DB_URL:jdbc:mysql://127.0.0.1:3306/aiimage?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false&allowPublicKeyRetrieval=true} diff --git a/scripts/coze_diag.py b/scripts/coze_diag.py new file mode 100644 index 0000000..ce82740 --- /dev/null +++ b/scripts/coze_diag.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +import argparse +import json +import os +import sys +import time +from typing import Any, Dict, Optional + +import requests + + +DEFAULT_BASE_URL = "https://api.coze.cn" +DEFAULT_RUN_PATH = "/v1/workflow/run" +DEFAULT_HISTORY_PATH = "/v1/workflows/{workflow_id}/run_histories/{execute_id}" +DEFAULT_SAMPLE_PROMPT = ( + "请帮我排查以下亚马逊商品在英国及欧洲地区是否存在知识产权侵权风险。" + "请直接给出明确结论（有侵权风险 或 未发现明显侵权风险）。" +) + + +def build_headers(token: str) -> Dict[str, str]: + normalized = (token or "").strip() + if normalized.lower().startswith("bearer "): + normalized = normalized[7:].strip() + return { + "Authorization": f"Bearer {normalized}", + "Content-Type": "application/json; charset=utf-8", + "Accept-Charset": "utf-8", + } + + +def join_url(base_url: str, path: str) -> str: + return base_url.rstrip("/") + "/" + path.lstrip("/") + + +def print_block(title: str, payload: Any) -> None: + print(f"\n=== {title} ===") + if isinstance(payload, (dict, list)): + print(json.dumps(payload, ensure_ascii=False, indent=2)) + else: + print(payload) + + +def request_json(method: str, url: str, headers: Dict[str, str], **kwargs: Any) -> Dict[str, Any]: + started = time.time() + response = requests.request(method, url, headers=headers, timeout=kwargs.pop("timeout", 60), **kwargs) + elapsed_ms = int((time.time() - started) * 1000) + body_text = response.text + print_block( + f"{method.upper()} {url} -> HTTP {response.status_code} ({elapsed_ms} ms)", + body_text if len(body_text) < 5000 else body_text[:5000] + "\n......", + ) + response.raise_for_status() + try: + return response.json() + except Exception as exc: + raise RuntimeError(f"response is not valid JSON: {exc}") from exc + + +def extract_execute_id(payload: Dict[str, Any]) -> str: + for key in ("execute_id", "executeId"): + value = payload.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + raise RuntimeError("execute_id not found in Coze response") + + +def payload_data_text(payload: Dict[str, Any]) -> str: + data = payload.get("data") + if isinstance(data, str): + return data + if isinstance(data, dict): + inner = data.get("data") + if isinstance(inner, str): + return inner + return "" + + +def workflow_status(payload: Dict[str, Any]) -> str: + for key in ("status", "run_status", "workflow_status"): + value = payload.get(key) + if isinstance(value, str) and value.strip(): + return value.strip().upper() + data = payload.get("data") + if isinstance(data, dict): + for key in ("status", "run_status", "workflow_status"): + value = data.get(key) + if isinstance(value, str) and value.strip(): + return value.strip().upper() + return "" + + +def looks_finished(payload: Dict[str, Any]) -> bool: + status = workflow_status(payload) + return status in {"SUCCESS", "SUCCEEDED", "DONE", "FAILED", "ERROR", "CANCELLED"} + + +def build_sample_run_body(workflow_id: str, prompt: str) -> Dict[str, Any]: + return { + "workflow_id": workflow_id, + "parameters": { + "title_list": ["WIRESTER Green Aluminum Round Bike Bell"], + "url_list": ["https://m.media-amazon.com/images/I/51dETWpSQXL._AC_SY450_.jpg"], + "items": [ + { + "group_key": "diag::1", + "row_id": "1", + "asin": "B0D14L34S7", + "country": "英国", + "title": "WIRESTER Green Aluminum Round Bike Bell", + "url": "https://m.media-amazon.com/images/I/51dETWpSQXL._AC_SY450_.jpg", + } + ], + "prompt": prompt, + }, + "is_async": True, + } + + +def poll_history( + base_url: str, + workflow_id: str, + execute_id: str, + token: str, + history_path: str, + poll_interval: float, + max_polls: int, +) -> int: + headers = build_headers(token) + history_url = join_url( + base_url, + history_path.replace("{workflow_id}", workflow_id).replace("{execute_id}", execute_id), + ) + for idx in range(1, max_polls + 1): + print(f"\n--- poll #{idx} execute_id={execute_id} ---") + payload = request_json("GET", history_url, headers) + code = payload.get("code") + status = workflow_status(payload) + data_text = payload_data_text(payload) + print(f"code={code!r} status={status or '-'} has_data={bool(data_text)}") + if data_text: + print_block("Resolved Data", data_text) + return 0 + if looks_finished(payload): + print("workflow reached terminal status but no data was returned") + return 2 + time.sleep(max(0.2, poll_interval)) + print("history polling reached max attempts without terminal result") + return 3 + + +def main() -> int: + parser = argparse.ArgumentParser(description="Diagnose Coze run/history connectivity from server.") + parser.add_argument("--base-url", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_BASE_URL", DEFAULT_BASE_URL)) + parser.add_argument("--workflow-id", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID") or os.getenv("workflow_id")) + parser.add_argument("--token", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN") or os.getenv("coze_token")) + parser.add_argument("--run-path", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_PATH", DEFAULT_RUN_PATH)) + parser.add_argument( + "--history-path", + default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_HISTORY_PATH", DEFAULT_HISTORY_PATH), + ) + parser.add_argument("--execute-id", help="Only test history/poll for an existing execute_id.") + parser.add_argument("--poll-interval", type=float, default=2.0) + parser.add_argument("--max-polls", type=int, default=20) + parser.add_argument("--prompt", default=DEFAULT_SAMPLE_PROMPT) + parser.add_argument("--skip-run", action="store_true", help="Do not create a new run; only poll execute_id.") + args = parser.parse_args() + + if not args.workflow_id: + print("missing workflow id: pass --workflow-id or set AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID/workflow_id", file=sys.stderr) + return 4 + if not args.token: + print("missing token: pass --token or set AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN/coze_token", file=sys.stderr) + return 4 + + print_block( + "Config", + { + "base_url": args.base_url, + "workflow_id": args.workflow_id, + "run_path": args.run_path, + "history_path": args.history_path, + "has_token": bool(args.token), + "execute_id": args.execute_id, + "skip_run": args.skip_run, + }, + ) + + if args.execute_id: + return poll_history( + args.base_url, + args.workflow_id, + args.execute_id, + args.token, + args.history_path, + args.poll_interval, + args.max_polls, + ) + + if args.skip_run: + print("--skip-run requires --execute-id", file=sys.stderr) + return 4 + + headers = build_headers(args.token) + run_url = join_url(args.base_url, args.run_path) + body = build_sample_run_body(args.workflow_id, args.prompt) + print_block("Run Request Body", body) + payload = request_json("POST", run_url, headers, json=body) + execute_id = extract_execute_id(payload) + print(f"\nexecute_id={execute_id}") + return poll_history( + args.base_url, + args.workflow_id, + execute_id, + args.token, + args.history_path, + args.poll_interval, + args.max_polls, + ) + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except requests.HTTPError as exc: + response = exc.response + if response is not None: + print(f"HTTP error: {response.status_code} {response.text}", file=sys.stderr) + else: + print(f"HTTP error: {exc}", file=sys.stderr) + raise + except Exception as exc: + print(f"fatal: {exc}", file=sys.stderr) + raise