diff --git a/backend-java/pom.xml b/backend-java/pom.xml
index ea9047b4..40f05716 100644
--- a/backend-java/pom.xml
+++ b/backend-java/pom.xml
@@ -21,7 +21,6 @@
3.5.7
1.5.5.Final
4.0.3
- 3.17.4
5.8.36
2.3.5
8.5.17
@@ -80,11 +79,6 @@
poi-ooxml
5.2.5
-
- com.aliyun.oss
- aliyun-sdk-oss
- ${aliyun.oss.version}
-
com.github.xiaoymin
knife4j-openapi3-jakarta-spring-boot-starter
diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/OssProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/OssProperties.java
index c47026f7..a8af0109 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/config/OssProperties.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/config/OssProperties.java
@@ -11,6 +11,7 @@ public class OssProperties {
private String publicEndpoint;
private String bucket;
private String imageVideoBucket;
+ private String digitalHumanBucket;
private String accessKeyId;
private String accessKeySecret;
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentHistoryItemVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentHistoryItemVo.java
index 25587c02..c5f497ce 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentHistoryItemVo.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentHistoryItemVo.java
@@ -14,7 +14,7 @@ public class AppearancePatentHistoryItemVo {
private String sourceFilename;
@Schema(description = "最终结果文件名。任务完成并生成 xlsx 后返回。", example = "17-result.xlsx")
private String resultFilename;
- @Schema(description = "最终结果文件下载地址。后端基于 OSS objectKey 生成的公开直链 URL。", example = "https://bucket.oss-cn-hangzhou.aliyuncs.com/result/appearance_patent/xxx/17-result.xlsx")
+ @Schema(description = "最终结果文件下载地址。后端基于 MinIO objectKey 生成的公开直链 URL。", example = "http://47.110.241.161:9000/nanri-ai-images/result/appearance_patent/xxx/17-result.xlsx")
private String downloadUrl;
private Long fileJobId;
private String fileStatus;
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandSourceFileDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandSourceFileDto.java
index 908b9b07..f313423e 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandSourceFileDto.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandSourceFileDto.java
@@ -9,7 +9,7 @@ import lombok.Data;
public class BrandSourceFileDto {
@NotBlank(message = "fileUrl 不能为空")
- @Schema(description = "OSS 文件下载链接，Java 将先下载文件再读取内容。", example = "https://example.oss-cn-hangzhou.aliyuncs.com/source/brand.xlsx")
+ @Schema(description = "对象存储文件下载链接，Java 将先下载文件再读取内容。", example = "http://47.110.241.161:9000/nanri-ai-images/source/brand.xlsx")
private String fileUrl;
@Schema(description = "原始文件名，用于生成任务描述和结果文件名。", example = "品牌样例.xlsx")
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java
index 6c2a53ed..b600e04a 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java
@@ -1,11 +1,8 @@
package com.nanri.aiimage.modules.digitalhuman.service;
-import com.aliyun.oss.OSS;
-import com.aliyun.oss.OSSClientBuilder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.common.exception.BusinessException;
-import com.nanri.aiimage.config.OssProperties;
import com.nanri.aiimage.modules.digitalhuman.mapper.DigitalHumanVersionMapper;
import com.nanri.aiimage.modules.digitalhuman.model.entity.DigitalHumanVersionEntity;
import com.nanri.aiimage.modules.digitalhuman.model.vo.DigitalHumanVersionVo;
@@ -33,11 +30,10 @@ public class DigitalHumanVersionService {
private static final String STATUS_DRAFT = "DRAFT";
private static final String STATUS_RELEASED = "RELEASED";
- private static final String OSS_PATH_PREFIX = "digital-human/versions/";
+ private static final String STORAGE_PATH_PREFIX = "digital-human/versions/";
private final DigitalHumanVersionMapper versionMapper;
private final OssStorageService ossStorageService;
- private final OssProperties ossProperties;
@Transactional
public DigitalHumanVersionVo uploadVersion(String version, MultipartFile file, String changelog,
@@ -67,23 +63,16 @@ public class DigitalHumanVersionService {
log.info("[digital-human-version] md5 calculated version={} md5={} elapsedMs={}",
version, md5, elapsedMs(startedAt));
- // 构建 OSS 路径
- String ossObjectKey = OSS_PATH_PREFIX + "v" + version + "/ShuFuDigitalHuman.zip";
+ String ossObjectKey = STORAGE_PATH_PREFIX + "v" + version + "/ShuFuDigitalHuman.zip";
- // 上传到 OSS
- OSS ossClient = buildOssClient();
- try {
- log.info("[digital-human-version] oss upload start version={} objectKey={} bytes={} elapsedMs={}",
- version, ossObjectKey, tempFile.length(), elapsedMs(startedAt));
- ossClient.putObject(ossProperties.getBucket(), ossObjectKey, tempFile);
- if (!ossClient.doesObjectExist(ossProperties.getBucket(), ossObjectKey)) {
- throw new BusinessException("数字人版本文件上传后在 OSS 中不可见，请重试");
- }
- log.info("[digital-human-version] oss uploaded version={} objectKey={} bytes={} elapsedMs={}",
- version, ossObjectKey, tempFile.length(), elapsedMs(startedAt));
- } finally {
- ossClient.shutdown();
+ log.info("[digital-human-version] minio upload start version={} objectKey={} bytes={} elapsedMs={}",
+ version, ossObjectKey, tempFile.length(), elapsedMs(startedAt));
+ ossStorageService.uploadDigitalHumanVersion(tempFile, ossObjectKey);
+ if (!ossStorageService.objectExists(ossObjectKey)) {
+ throw new BusinessException("数字人版本文件上传后在 MinIO 中不可见，请重试");
}
+ log.info("[digital-human-version] minio uploaded version={} objectKey={} bytes={} elapsedMs={}",
+ version, ossObjectKey, tempFile.length(), elapsedMs(startedAt));
// 保存数据库记录
DigitalHumanVersionEntity entity = new DigitalHumanVersionEntity();
@@ -161,7 +150,7 @@ public class DigitalHumanVersionService {
}
if (!ossStorageService.objectExists(entity.getOssObjectKey())) {
- throw new BusinessException("数字人版本文件不存在于 OSS，请重新上传该版本：" + entity.getOssObjectKey());
+ throw new BusinessException("数字人版本文件不存在于 MinIO，请重新上传该版本：" + entity.getOssObjectKey());
}
entity.setStatus(STATUS_RELEASED);
@@ -184,7 +173,7 @@ public class DigitalHumanVersionService {
// 清除其他版本的 is_latest 标记
if (!ossStorageService.objectExists(entity.getOssObjectKey())) {
- throw new BusinessException("数字人版本文件不存在于 OSS，请重新上传该版本：" + entity.getOssObjectKey());
+ throw new BusinessException("数字人版本文件不存在于 MinIO，请重新上传该版本：" + entity.getOssObjectKey());
}
versionMapper.update(null, new LambdaUpdateWrapper()
@@ -209,11 +198,11 @@ public class DigitalHumanVersionService {
throw new BusinessException("最新版本不能删除");
}
- // 删除 OSS 文件
+ // 删除 MinIO 文件
try {
ossStorageService.deleteObject(entity.getOssObjectKey());
} catch (Exception e) {
- log.warn("删除 OSS 文件失败：{}", entity.getOssObjectKey(), e);
+ log.warn("删除 MinIO 文件失败：{}", entity.getOssObjectKey(), e);
}
// 删除数据库记录
@@ -227,7 +216,7 @@ public class DigitalHumanVersionService {
throw new BusinessException("版本不存在：" + version);
}
if (!ossStorageService.objectExists(entity.getOssObjectKey())) {
- throw new BusinessException("数字人版本文件不存在于 OSS，请重新上传该版本：" + entity.getOssObjectKey());
+ throw new BusinessException("数字人版本文件不存在于 MinIO，请重新上传该版本：" + entity.getOssObjectKey());
}
return ossStorageService.generateDownloadUrl(entity.getOssObjectKey());
}
@@ -265,11 +254,4 @@ public class DigitalHumanVersionService {
}
}
- private OSS buildOssClient() {
- return new OSSClientBuilder().build(
- "https://" + ossProperties.getEndpoint(),
- ossProperties.getAccessKeyId(),
- ossProperties.getAccessKeySecret()
- );
- }
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/oss/OssStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/oss/OssStorageService.java
index a85d2cf7..b9ba7d51 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/oss/OssStorageService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/oss/OssStorageService.java
@@ -1,10 +1,14 @@
package com.nanri.aiimage.modules.file.service.oss;
-import com.aliyun.oss.OSS;
-import com.aliyun.oss.OSSClientBuilder;
-import com.aliyun.oss.model.OSSObject;
import com.nanri.aiimage.config.OssProperties;
-import lombok.RequiredArgsConstructor;
+import io.minio.GetObjectArgs;
+import io.minio.MinioClient;
+import io.minio.PutObjectArgs;
+import io.minio.RemoveObjectArgs;
+import io.minio.StatObjectArgs;
+import io.minio.UploadObjectArgs;
+import io.minio.errors.ErrorResponseException;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.ByteArrayInputStream;
@@ -12,232 +16,300 @@ import java.io.File;
import java.net.URI;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Locale;
import java.util.Objects;
import java.util.UUID;
+import java.util.stream.Stream;
@Service
-@RequiredArgsConstructor
public class OssStorageService {
- private final OssProperties ossProperties;
+ private static final String IMAGE_VIDEO_MODULE = "IMAGE_VIDEO";
+ private static final String DIGITAL_HUMAN_PREFIX = "digital-human/versions/";
+
+ private final OssProperties ossProperties;
+ private final MinioClient minioClient;
+
+ @Autowired
+ public OssStorageService(OssProperties ossProperties) {
+ this(ossProperties, null);
+ }
+
+ OssStorageService(OssProperties ossProperties, MinioClient minioClient) {
+ this.ossProperties = ossProperties;
+ this.minioClient = minioClient == null ? createClient(ossProperties) : minioClient;
+ }
- /**
- * 上传结果文件到 OSS，返回 objectKey。
- * 调用方应存储 objectKey，下载时通过 generateFreshDownloadUrl 生成公开直链。
- */
public String uploadResultFile(File file, String moduleType) {
- String objectKey = String.format("result/%s/%s/%s", moduleType.toLowerCase(), UUID.randomUUID(), file.getName());
- OSS ossClient = buildClient();
- try {
- ossClient.putObject(ossProperties.getBucket(), objectKey, file);
- return objectKey;
- } finally {
- ossClient.shutdown();
- }
+ String objectKey = resultObjectKey(file, moduleType);
+ uploadFile(file, resolveBucket(moduleType), objectKey);
+ return objectKey;
}
public UploadedResult uploadResultFileWithFreshDownloadUrl(File file, String moduleType) {
- String objectKey = String.format("result/%s/%s/%s", moduleType.toLowerCase(), UUID.randomUUID(), file.getName());
+ String objectKey = resultObjectKey(file, moduleType);
String bucket = resolveBucket(moduleType);
- OSS ossClient = buildClient();
- try {
- ossClient.putObject(bucket, objectKey, file);
- return new UploadedResult(objectKey, getPublicUrl(objectKey, bucket));
- } finally {
- ossClient.shutdown();
- }
+ uploadFile(file, bucket, objectKey);
+ return new UploadedResult(objectKey, getPublicUrl(objectKey, bucket));
}
public UploadedResult uploadPublicFileWithFreshDownloadUrl(File file, String moduleType, String originalFilename) {
- String normalizedModuleType = moduleType == null || moduleType.isBlank()
- ? "common"
- : moduleType.trim().toLowerCase();
+ String normalizedModuleType = normalizeModuleType(moduleType, "common");
String objectName = sanitizeObjectName(originalFilename);
if (objectName.isBlank()) {
objectName = file.getName();
}
String objectKey = String.format("upload/%s/%s/%s", normalizedModuleType, UUID.randomUUID(), objectName);
String bucket = resolveBucket(moduleType);
- OSS ossClient = buildClient();
- try {
- ossClient.putObject(bucket, objectKey, file);
- return new UploadedResult(objectKey, getPublicUrl(objectKey, bucket));
- } finally {
- ossClient.shutdown();
+ uploadFile(file, bucket, objectKey);
+ return new UploadedResult(objectKey, getPublicUrl(objectKey, bucket));
+ }
+
+ public String uploadDigitalHumanVersion(File file, String objectKey) {
+ if (objectKey == null || !objectKey.startsWith(DIGITAL_HUMAN_PREFIX)) {
+ throw new IllegalArgumentException("digital human objectKey must start with " + DIGITAL_HUMAN_PREFIX);
}
+ uploadFile(file, digitalHumanBucket(), objectKey);
+ return objectKey;
}
public String uploadText(String objectKey, String content) {
if (objectKey == null || objectKey.isBlank()) {
throw new IllegalArgumentException("objectKey must not be blank");
}
- OSS ossClient = buildClient();
- try (ByteArrayInputStream stream = new ByteArrayInputStream(
- Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8))) {
- ossClient.putObject(ossProperties.getBucket(), objectKey, stream);
+ byte[] bytes = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8);
+ try (ByteArrayInputStream stream = new ByteArrayInputStream(bytes)) {
+ buildClient().putObject(PutObjectArgs.builder()
+ .bucket(ossProperties.getBucket())
+ .object(objectKey)
+ .stream(stream, bytes.length, -1)
+ .contentType("application/json; charset=utf-8")
+ .build());
return objectKey;
} catch (Exception ex) {
- throw new IllegalStateException("failed to upload text to oss", ex);
- } finally {
- ossClient.shutdown();
+ throw storageFailure("upload text", objectKey, ex);
}
}
public String uploadTaskScopePayload(String moduleType, Long taskId, String scopeHash, String content) {
- String normalizedModuleType = moduleType == null || moduleType.isBlank()
- ? "unknown"
- : moduleType.trim().toLowerCase();
String normalizedScopeHash = scopeHash == null || scopeHash.isBlank() ? UUID.randomUUID().toString() : scopeHash;
- String objectKey = String.format("task-scope/%s/%s/%s.json", normalizedModuleType, taskId, normalizedScopeHash);
+ String objectKey = String.format("task-scope/%s/%s/%s.json", normalizeModuleType(moduleType), taskId, normalizedScopeHash);
return uploadText(objectKey, content);
}
public String uploadTaskParsedPayload(String moduleType, Long taskId, String scopeHash, String content) {
- String normalizedModuleType = moduleType == null || moduleType.isBlank()
- ? "unknown"
- : moduleType.trim().toLowerCase();
String normalizedScopeHash = scopeHash == null || scopeHash.isBlank() ? UUID.randomUUID().toString() : scopeHash;
- String objectKey = String.format("task-parsed/%s/%s/%s.json", normalizedModuleType, taskId, normalizedScopeHash);
+ String objectKey = String.format("task-parsed/%s/%s/%s.json", normalizeModuleType(moduleType), taskId, normalizedScopeHash);
return uploadText(objectKey, content);
}
public String readObjectAsString(String value) {
- String objectKey = resolveObjectKey(value);
- if (objectKey == null || objectKey.isBlank()) {
+ StorageLocation location = resolveStorageLocation(value);
+ if (location == null) {
return null;
}
- OSS ossClient = buildClient();
- try (OSSObject ossObject = ossClient.getObject(ossProperties.getBucket(), objectKey)) {
- return new String(ossObject.getObjectContent().readAllBytes(), StandardCharsets.UTF_8);
+ try (var stream = buildClient().getObject(GetObjectArgs.builder()
+ .bucket(location.bucket())
+ .object(location.objectKey())
+ .build())) {
+ return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
} catch (Exception ex) {
- throw new IllegalStateException("failed to read object from oss", ex);
- } finally {
- ossClient.shutdown();
+ throw storageFailure("read", location.objectKey(), ex);
}
}
public void deleteObject(String value) {
- String objectKey = resolveObjectKey(value);
- if (objectKey == null || objectKey.isBlank()) {
+ StorageLocation location = resolveStorageLocation(value);
+ if (location == null) {
return;
}
- OSS ossClient = buildClient();
try {
- ossClient.deleteObject(ossProperties.getBucket(), objectKey);
- } finally {
- ossClient.shutdown();
+ buildClient().removeObject(RemoveObjectArgs.builder()
+ .bucket(location.bucket())
+ .object(location.objectKey())
+ .build());
+ } catch (Exception ex) {
+ throw storageFailure("delete", location.objectKey(), ex);
}
}
public boolean objectExists(String value) {
- String objectKey = resolveObjectKey(value);
- if (objectKey == null || objectKey.isBlank()) {
+ StorageLocation location = resolveStorageLocation(value);
+ if (location == null) {
return false;
}
- OSS ossClient = buildClient();
try {
- return ossClient.doesObjectExist(ossProperties.getBucket(), objectKey);
- } finally {
- ossClient.shutdown();
+ buildClient().statObject(StatObjectArgs.builder()
+ .bucket(location.bucket())
+ .object(location.objectKey())
+ .build());
+ return true;
+ } catch (ErrorResponseException ex) {
+ if (isNotFound(ex)) {
+ return false;
+ }
+ throw storageFailure("stat", location.objectKey(), ex);
+ } catch (Exception ex) {
+ throw storageFailure("stat", location.objectKey(), ex);
}
}
- /**
- * 根据 objectKey 生成公开直链下载 URL。
- */
- public String generateDownloadUrl(String objectKey) {
- return getPublicUrl(objectKey);
+ public String generateDownloadUrl(String value) {
+ return getPublicUrl(value);
}
- /**
- * 获取公开（无签名）URL，格式：https://{bucket}.{endpoint}/{objectKey}
- */
public String getPublicUrl(String value) {
- return getPublicUrl(value, ossProperties.getBucket());
+ StorageLocation location = resolveStorageLocation(value);
+ return location == null ? null : getPublicUrl(location.objectKey(), location.bucket());
}
- private String getPublicUrl(String value, String bucket) {
- if (value == null || value.isBlank()) {
- return null;
- }
- String objectKey = resolveObjectKey(value);
- String normalizedKey = objectKey.startsWith("/") ? objectKey.substring(1) : objectKey;
- String host = String.format("%s.%s", bucket, publicEndpoint());
- try {
- return new URI("https", host, "/" + normalizedKey, null).toASCIIString();
- } catch (Exception ignored) {
- return String.format("https://%s/%s", host, normalizedKey);
- }
- }
-
- /**
- * 从存储值中解析出 objectKey，兼容两种格式：
- * - 旧格式：完整 URL（https://bucket.endpoint/objectKey?Expires=...）
- * - 新格式：直接是 objectKey（如 result/dedupe/uuid/file.xlsx）
- */
public String resolveObjectKey(String value) {
- if (value == null || value.isBlank()) {
- return value;
- }
- try {
- if (value.startsWith("http://") || value.startsWith("https://")) {
- String path = URI.create(value).getPath();
- return path.startsWith("/") ? path.substring(1) : path;
- }
- } catch (Exception ignored) {
- int schemeEnd = value.indexOf("://");
- int pathStart = schemeEnd < 0 ? -1 : value.indexOf('/', schemeEnd + 3);
- if (pathStart >= 0 && pathStart + 1 < value.length()) {
- String path = value.substring(pathStart + 1);
- int queryStart = path.indexOf('?');
- String objectKey = queryStart >= 0 ? path.substring(0, queryStart) : path;
- return URLDecoder.decode(objectKey, StandardCharsets.UTF_8);
- }
- }
- return value;
+ StorageLocation location = resolveStorageLocation(value);
+ return location == null ? value : location.objectKey();
}
- /**
- * 根据存储值（objectKey 或旧格式完整 URL）生成公开直链下载 URL。
- * 供各模块 listHistory 和下载接口使用，避免签名过期。
- */
public String generateFreshDownloadUrl(String value) {
+ return getPublicUrl(value);
+ }
+
+ private void uploadFile(File file, String bucket, String objectKey) {
+ if (file == null || !file.isFile()) {
+ throw new IllegalArgumentException("upload file must exist");
+ }
+ try {
+ buildClient().uploadObject(UploadObjectArgs.builder()
+ .bucket(bucket)
+ .object(objectKey)
+ .filename(file.getAbsolutePath())
+ .build());
+ } catch (Exception ex) {
+ throw storageFailure("upload", objectKey, ex);
+ }
+ }
+
+ private String getPublicUrl(String objectKey, String bucket) {
+ String encodedPath;
+ try {
+ encodedPath = new URI(null, null, "/" + bucket + "/" + trimLeadingSlash(objectKey), null)
+ .toASCIIString();
+ } catch (Exception ex) {
+ encodedPath = "/" + bucket + "/" + trimLeadingSlash(objectKey).replace(" ", "%20");
+ }
+ return trimTrailingSlash(publicEndpoint()) + encodedPath;
+ }
+
+ private StorageLocation resolveStorageLocation(String value) {
if (value == null || value.isBlank()) {
return null;
}
- return generateDownloadUrl(resolveObjectKey(value));
+ String normalizedValue = value.trim();
+ String host = null;
+ String objectKey = normalizedValue;
+ if (normalizedValue.startsWith("http://") || normalizedValue.startsWith("https://")) {
+ try {
+ URI uri = URI.create(normalizedValue);
+ host = uri.getHost();
+ objectKey = decodePath(uri.getRawPath());
+ } catch (Exception ignored) {
+ objectKey = decodePathFromMalformedUrl(normalizedValue);
+ }
+ }
+
+ objectKey = trimLeadingSlash(objectKey);
+ String bucket = bucketFromVirtualHost(host);
+ String pathBucket = bucketPrefix(objectKey);
+ if (pathBucket != null) {
+ bucket = pathBucket;
+ objectKey = objectKey.substring(pathBucket.length() + 1);
+ }
+ if (bucket == null) {
+ bucket = bucketForObjectKey(objectKey);
+ }
+ return objectKey.isBlank() ? null : new StorageLocation(bucket, objectKey);
}
- private OSS buildClient() {
- return new OSSClientBuilder().build(
- "https://" + ossProperties.getEndpoint(),
- ossProperties.getAccessKeyId(),
- ossProperties.getAccessKeySecret()
- );
+ private String bucketFromVirtualHost(String host) {
+ if (host == null) {
+ return null;
+ }
+ String normalizedHost = host.toLowerCase(Locale.ROOT);
+ for (String bucket : configuredBuckets()) {
+ if (normalizedHost.equals(bucket.toLowerCase(Locale.ROOT))
+ || normalizedHost.startsWith(bucket.toLowerCase(Locale.ROOT) + ".")) {
+ return bucket;
+ }
+ }
+ return null;
}
- private String resolveBucket(String moduleType) {
- if (moduleType != null
- && "IMAGE_VIDEO".equalsIgnoreCase(moduleType.trim())
- && ossProperties.getImageVideoBucket() != null
- && !ossProperties.getImageVideoBucket().isBlank()) {
- return ossProperties.getImageVideoBucket().trim();
+ private String bucketPrefix(String objectKey) {
+ for (String bucket : configuredBuckets()) {
+ if (objectKey.equals(bucket) || objectKey.startsWith(bucket + "/")) {
+ return bucket;
+ }
+ }
+ return null;
+ }
+
+ private String bucketForObjectKey(String objectKey) {
+ if (objectKey.startsWith(DIGITAL_HUMAN_PREFIX)) {
+ return digitalHumanBucket();
+ }
+ if (objectKey.startsWith("result/image_video/") || objectKey.startsWith("upload/image_video/")) {
+ return imageVideoBucket();
}
return ossProperties.getBucket();
}
+ private List configuredBuckets() {
+ return Stream.of(ossProperties.getBucket(), imageVideoBucket(), digitalHumanBucket())
+ .filter(Objects::nonNull)
+ .map(String::trim)
+ .filter(bucket -> !bucket.isBlank())
+ .distinct()
+ .toList();
+ }
+
+ private MinioClient buildClient() {
+ return minioClient;
+ }
+
+ private static MinioClient createClient(OssProperties properties) {
+ return MinioClient.builder()
+ .endpoint(withScheme(properties.getEndpoint()))
+ .credentials(properties.getAccessKeyId(), properties.getAccessKeySecret())
+ .build();
+ }
+
+ private String resolveBucket(String moduleType) {
+ return IMAGE_VIDEO_MODULE.equalsIgnoreCase(Objects.requireNonNullElse(moduleType, "").trim())
+ ? imageVideoBucket()
+ : ossProperties.getBucket();
+ }
+
+ private String imageVideoBucket() {
+ return firstNonBlank(ossProperties.getImageVideoBucket(), ossProperties.getBucket());
+ }
+
+ private String digitalHumanBucket() {
+ return firstNonBlank(ossProperties.getDigitalHumanBucket(), ossProperties.getBucket());
+ }
+
private String publicEndpoint() {
- String endpoint = ossProperties.getPublicEndpoint();
- if (endpoint == null || endpoint.isBlank()) {
- endpoint = ossProperties.getEndpoint();
- }
- endpoint = endpoint.trim();
- if (endpoint.startsWith("http://")) {
- endpoint = endpoint.substring("http://".length());
- } else if (endpoint.startsWith("https://")) {
- endpoint = endpoint.substring("https://".length());
- }
- return endpoint.endsWith("/") ? endpoint.substring(0, endpoint.length() - 1) : endpoint;
+ return withScheme(firstNonBlank(ossProperties.getPublicEndpoint(), ossProperties.getEndpoint()));
+ }
+
+ private String resultObjectKey(File file, String moduleType) {
+ return String.format("result/%s/%s/%s", normalizeModuleType(moduleType), UUID.randomUUID(), file.getName());
+ }
+
+ private String normalizeModuleType(String moduleType) {
+ return normalizeModuleType(moduleType, "unknown");
+ }
+
+ private String normalizeModuleType(String moduleType, String defaultValue) {
+ return moduleType == null || moduleType.isBlank() ? defaultValue : moduleType.trim().toLowerCase(Locale.ROOT);
}
private String sanitizeObjectName(String filename) {
@@ -252,6 +324,63 @@ public class OssStorageService {
return normalized.replaceAll("[\\r\\n]", "_");
}
+ private String decodePath(String rawPath) {
+ if (rawPath == null) {
+ return "";
+ }
+ return URLDecoder.decode(rawPath.replace("+", "%2B"), StandardCharsets.UTF_8);
+ }
+
+ private String decodePathFromMalformedUrl(String value) {
+ int schemeEnd = value.indexOf("://");
+ int pathStart = schemeEnd < 0 ? -1 : value.indexOf('/', schemeEnd + 3);
+ if (pathStart < 0 || pathStart + 1 >= value.length()) {
+ return "";
+ }
+ String path = value.substring(pathStart + 1);
+ int queryStart = path.indexOf('?');
+ return URLDecoder.decode(queryStart >= 0 ? path.substring(0, queryStart) : path, StandardCharsets.UTF_8);
+ }
+
+ private boolean isNotFound(ErrorResponseException ex) {
+ String code = ex.errorResponse() == null ? null : ex.errorResponse().code();
+ return "NoSuchKey".equals(code) || "NoSuchObject".equals(code) || "NoSuchBucket".equals(code);
+ }
+
+ private IllegalStateException storageFailure(String operation, String objectKey, Exception cause) {
+ return new IllegalStateException("failed to " + operation + " object in MinIO: " + objectKey, cause);
+ }
+
+ private static String withScheme(String endpoint) {
+ String normalized = Objects.requireNonNull(endpoint, "MinIO endpoint must not be null").trim();
+ return normalized.startsWith("http://") || normalized.startsWith("https://")
+ ? normalized
+ : "http://" + normalized;
+ }
+
+ private String trimLeadingSlash(String value) {
+ String result = Objects.requireNonNullElse(value, "");
+ while (result.startsWith("/")) {
+ result = result.substring(1);
+ }
+ return result;
+ }
+
+ private String trimTrailingSlash(String value) {
+ String result = value;
+ while (result.endsWith("/")) {
+ result = result.substring(0, result.length() - 1);
+ }
+ return result;
+ }
+
+ private String firstNonBlank(String preferred, String fallback) {
+ return preferred == null || preferred.isBlank() ? fallback : preferred.trim();
+ }
+
+ private record StorageLocation(String bucket, String objectKey) {
+ }
+
public record UploadedResult(String objectKey, String downloadUrl) {
}
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/mapper/ImageVideoAsyncTaskMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/mapper/ImageVideoAsyncTaskMapper.java
index 4cfdf53f..e8827450 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/mapper/ImageVideoAsyncTaskMapper.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/mapper/ImageVideoAsyncTaskMapper.java
@@ -3,18 +3,30 @@ package com.nanri.aiimage.modules.imagevideo.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity;
import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Update;
@Mapper
public interface ImageVideoAsyncTaskMapper extends BaseMapper {
@Update("UPDATE biz_image_video_async_task "
- + "SET status = 'RUNNING', updated_at = NOW(), attempt_count = attempt_count + 1 "
- + "WHERE id = #{taskId} AND status = 'PENDING'")
- int claimPending(Long taskId);
+ + "SET status = 'RUNNING', owner_instance_id = CASE "
+ + "WHEN owner_instance_id IS NULL OR owner_instance_id = '' THEN #{ownerInstanceId} "
+ + "ELSE owner_instance_id END, updated_at = NOW(), attempt_count = attempt_count + 1 "
+ + "WHERE id = #{taskId} AND status = 'PENDING' "
+ + "AND (owner_instance_id = #{ownerInstanceId} OR owner_instance_id IS NULL OR owner_instance_id = '')")
+ int claimPending(@Param("taskId") Long taskId, @Param("ownerInstanceId") String ownerInstanceId);
@Update("UPDATE biz_image_video_async_task "
- + "SET status = 'POLLING', updated_at = NOW(), attempt_count = attempt_count + 1 "
- + "WHERE id = #{taskId} AND status = 'WAITING'")
- int claimWaiting(Long taskId);
+ + "SET status = 'POLLING', owner_instance_id = CASE "
+ + "WHEN owner_instance_id IS NULL OR owner_instance_id = '' THEN #{ownerInstanceId} "
+ + "ELSE owner_instance_id END, updated_at = NOW(), attempt_count = attempt_count + 1 "
+ + "WHERE id = #{taskId} AND status = 'WAITING' "
+ + "AND (owner_instance_id = #{ownerInstanceId} OR owner_instance_id IS NULL OR owner_instance_id = '')")
+ int claimWaiting(@Param("taskId") Long taskId, @Param("ownerInstanceId") String ownerInstanceId);
+
+ @Update("UPDATE biz_image_video_async_task SET status = 'WAITING', updated_at = NOW() "
+ + "WHERE status = 'POLLING' "
+ + "AND (owner_instance_id = #{ownerInstanceId} OR owner_instance_id IS NULL OR owner_instance_id = '')")
+ int requeuePollingTasks(@Param("ownerInstanceId") String ownerInstanceId);
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/model/entity/ImageVideoAsyncTaskEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/model/entity/ImageVideoAsyncTaskEntity.java
index 78d35205..825ad95d 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/model/entity/ImageVideoAsyncTaskEntity.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/model/entity/ImageVideoAsyncTaskEntity.java
@@ -14,6 +14,7 @@ public class ImageVideoAsyncTaskEntity {
@TableId(type = IdType.AUTO)
private Long id;
private Long userId;
+ private String ownerInstanceId;
private String taskType;
private String status;
private String requestJson;
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 cd4712bb..817299ed 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
@@ -4,6 +4,8 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
+import com.nanri.aiimage.common.exception.TaskOwnerMismatchException;
+import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.modules.imagevideo.mapper.ImageVideoAsyncTaskMapper;
import com.nanri.aiimage.modules.imagevideo.model.dto.DouyinCopyRequest;
import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoVoiceCloneRequest;
@@ -15,7 +17,9 @@ import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowRunReque
import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity;
import com.nanri.aiimage.modules.imagevideo.model.vo.ImageVideoAsyncTaskVo;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.event.EventListener;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
@@ -24,6 +28,7 @@ import java.time.LocalDateTime;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
@Slf4j
@@ -51,6 +56,7 @@ public class ImageVideoAsyncTaskService {
private final ImageVideoArchiveService archiveService;
private final ObjectMapper objectMapper;
private final TaskExecutor cozeTaskExecutor;
+ private final InstanceMetadata instanceMetadata;
public ImageVideoAsyncTaskService(
ImageVideoAsyncTaskMapper taskMapper,
@@ -58,13 +64,15 @@ public class ImageVideoAsyncTaskService {
ImageVideoWorkflowConfigService workflowConfigService,
ImageVideoArchiveService archiveService,
ObjectMapper objectMapper,
- @Qualifier("cozeTaskExecutor") TaskExecutor cozeTaskExecutor) {
+ @Qualifier("cozeTaskExecutor") TaskExecutor cozeTaskExecutor,
+ InstanceMetadata instanceMetadata) {
this.taskMapper = taskMapper;
this.cozeService = cozeService;
this.workflowConfigService = workflowConfigService;
this.archiveService = archiveService;
this.objectMapper = objectMapper;
this.cozeTaskExecutor = cozeTaskExecutor;
+ this.instanceMetadata = instanceMetadata;
}
public ImageVideoAsyncTaskVo submitDouyinCopy(DouyinCopyRequest request) {
@@ -102,6 +110,7 @@ public class ImageVideoAsyncTaskService {
if (task == null) {
throw new BusinessException("Image video task not found");
}
+ ensureTaskOwnedByCurrentInstance(task, "query async task");
task = recoverFalseFailedTask(task);
ImageVideoAsyncTaskVo result = toVo(task);
if (TaskStatus.FAILED.name().equals(task.getStatus())) {
@@ -114,6 +123,9 @@ public class ImageVideoAsyncTaskService {
public void dispatchPendingTasks() {
List tasks = taskMapper.selectList(new LambdaQueryWrapper()
.eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name())
+ .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
+ .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId)
+ .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
.orderByAsc(ImageVideoAsyncTaskEntity::getId)
.last("LIMIT " + DISPATCH_BATCH_SIZE));
tasks.forEach(task -> cozeTaskExecutor.execute(() -> executeTask(task.getId())));
@@ -123,11 +135,23 @@ public class ImageVideoAsyncTaskService {
public void pollWaitingTasks() {
List tasks = taskMapper.selectList(new LambdaQueryWrapper()
.eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name())
+ .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
+ .or().isNull(ImageVideoAsyncTaskEntity::getOwnerInstanceId)
+ .or().eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, ""))
.orderByAsc(ImageVideoAsyncTaskEntity::getUpdatedAt)
.last("LIMIT " + POLL_BATCH_SIZE));
tasks.forEach(task -> cozeTaskExecutor.execute(() -> pollTask(task.getId())));
}
+ @EventListener(ApplicationReadyEvent.class)
+ public void recoverInterruptedPollingTasks() {
+ int recovered = taskMapper.requeuePollingTasks(currentInstanceId());
+ if (recovered > 0) {
+ log.warn("[image-video] recovered interrupted polling tasks count={} owner={}",
+ recovered, currentInstanceId());
+ }
+ }
+
@Scheduled(fixedDelayString = "${aiimage.image-video.failed-task-cleanup-delay-ms:60000}")
public void cleanupFailedTasks() {
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(FAILED_TASK_RETENTION_MINUTES);
@@ -146,6 +170,7 @@ public class ImageVideoAsyncTaskService {
LocalDateTime now = LocalDateTime.now();
ImageVideoAsyncTaskEntity task = new ImageVideoAsyncTaskEntity();
task.setUserId(userId);
+ task.setOwnerInstanceId(currentInstanceId());
task.setTaskType(type.name());
task.setStatus(TaskStatus.PENDING.name());
task.setRequestJson(writeJson(payload));
@@ -154,12 +179,14 @@ public class ImageVideoAsyncTaskService {
task.setCreatedAt(now);
task.setUpdatedAt(now);
taskMapper.insert(task);
+ log.info("[image-video] async task submitted taskId={} type={} owner={}",
+ task.getId(), task.getTaskType(), task.getOwnerInstanceId());
cozeTaskExecutor.execute(() -> executeTask(task.getId()));
return toVo(task);
}
private void executeTask(Long taskId) {
- if (taskMapper.claimPending(taskId) != 1) {
+ if (taskMapper.claimPending(taskId, currentInstanceId()) != 1) {
return;
}
ImageVideoAsyncTaskEntity task = taskMapper.selectById(taskId);
@@ -189,17 +216,13 @@ public class ImageVideoAsyncTaskService {
}
private void pollTask(Long taskId) {
- if (taskMapper.claimWaiting(taskId) != 1) {
+ if (taskMapper.claimWaiting(taskId, currentInstanceId()) != 1) {
return;
}
ImageVideoAsyncTaskEntity task = taskMapper.selectById(taskId);
if (task == null) {
return;
}
- if (task.getSubmittedAt() != null && task.getSubmittedAt().plusHours(1).isBefore(LocalDateTime.now())) {
- failTask(task, new BusinessException("Coze task exceeded the one-hour polling limit"));
- return;
- }
try {
TaskType type = TaskType.valueOf(task.getTaskType());
String workflowId = workflowIdFor(type);
@@ -216,12 +239,20 @@ public class ImageVideoAsyncTaskService {
}
return;
}
+ if (pollingDeadlineExceeded(task)) {
+ failTask(task, new BusinessException("Coze task exceeded the one-hour polling limit"));
+ return;
+ }
task.setStatus(TaskStatus.WAITING.name());
task.setCozeStatus(cozeStatus);
task.setUpdatedAt(LocalDateTime.now());
taskMapper.updateById(task);
} catch (Exception ex) {
// Coze history calls are retried by the next polling cycle until the task's overall deadline.
+ if (pollingDeadlineExceeded(task)) {
+ failTask(task, new BusinessException("Coze task exceeded the one-hour polling limit", ex));
+ return;
+ }
task.setStatus(TaskStatus.WAITING.name());
task.setErrorMessage(truncate(messageOf(ex)));
task.setUpdatedAt(LocalDateTime.now());
@@ -493,6 +524,26 @@ public class ImageVideoAsyncTaskService {
return value.length() <= 1000 ? value : value.substring(0, 1000);
}
+ private boolean pollingDeadlineExceeded(ImageVideoAsyncTaskEntity task) {
+ return task.getSubmittedAt() != null
+ && task.getSubmittedAt().plusHours(1).isBefore(LocalDateTime.now());
+ }
+
+ private void ensureTaskOwnedByCurrentInstance(ImageVideoAsyncTaskEntity task, String operation) {
+ String owner = task == null ? null : task.getOwnerInstanceId();
+ if (owner == null || owner.isBlank() || Objects.equals(owner, currentInstanceId())) {
+ return;
+ }
+ log.info("[image-video] route task operation to owner taskId={} operation={} owner={} current={}",
+ task.getId(), operation, owner, currentInstanceId());
+ throw new TaskOwnerMismatchException(task.getId(), operation, owner, currentInstanceId());
+ }
+
+ private String currentInstanceId() {
+ String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId();
+ return instanceId == null || instanceId.isBlank() ? "unknown-instance" : instanceId;
+ }
+
private enum TaskStatus {
PENDING, RUNNING, WAITING, POLLING, SUCCESS, FAILED
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/controller/PublishController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/controller/PublishController.java
index cfac2f58..f228a79f 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/controller/PublishController.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/controller/PublishController.java
@@ -36,7 +36,7 @@ public class PublishController {
@PostMapping("/parse")
@Operation(
summary = "匹配店铺、解析 Excel 并创建多文件批次任务",
- description = "按文件名（去扩展名）先匹配已管理店铺，再查询紫鸟索引；解析每个非空 Sheet。表头必须依次为 id、ASIN、国家、品牌、价格、状态、同步状态、同步国家。匹配并解析成功的文件为 PENDING，失败文件为 FAILED；只要存在可处理文件，任务为 PENDING，否则任务为 FAILED。")
+ description = "按文件名（去扩展名）先匹配已管理店铺，再查询紫鸟索引；解析每个非空 Sheet。源文件前五列表头必须依次为 id、ASIN、国家、品牌、价格，第六列及以后不解析；状态、同步状态、同步国家由 Python 回传并写入最终结果文件。匹配并解析成功的文件为 PENDING，失败文件为 FAILED；只要存在可处理文件，任务为 PENDING，否则任务为 FAILED。")
public ApiResponse parse(@Valid @RequestBody PublishParseRequest request) {
return ApiResponse.success(publishTaskService.parseAndCreateTask(request));
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/model/vo/PublishItemsPageVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/model/vo/PublishItemsPageVo.java
index 6ac17683..40b75aee 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/model/vo/PublishItemsPageVo.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/model/vo/PublishItemsPageVo.java
@@ -40,6 +40,6 @@ public class PublishItemsPageVo {
private Long total;
@Schema(description = "总页数", example = "14")
private Integer totalPages;
- @Schema(description = "按原 Excel 行序返回的八列数据")
+ @Schema(description = "按原 Excel 行序返回前五列源数据；状态、同步状态、同步国家等待 Python 处理后回传")
private List items = new ArrayList<>();
}
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java
index 5c8cbb72..5c2432e7 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java
@@ -7,6 +7,8 @@ import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
+import com.nanri.aiimage.common.exception.TaskOwnerMismatchException;
+import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.publish.mapper.PublishFileMapper;
@@ -87,6 +89,7 @@ public class PublishTaskService {
private final OssStorageService ossStorageService;
private final ObjectMapper objectMapper;
private final TransactionTemplate transactionTemplate;
+ private final InstanceMetadata instanceMetadata;
@Value("${aiimage.publish.stale-timeout-minutes:30}")
private int staleTimeoutMinutes;
@@ -349,6 +352,12 @@ public class PublishTaskService {
List staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
+ .and(owner -> owner
+ .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL")
+ .or()
+ .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''")
+ .or()
+ .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId()))
.lt(FileTaskEntity::getUpdatedAt, threshold)
.orderByAsc(FileTaskEntity::getUpdatedAt)
.last("limit 100"));
@@ -377,6 +386,7 @@ public class PublishTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("task not found");
}
+ ensureTaskOwnedByCurrentInstance(task, "process publish result file");
FileResultEntity result = fileResultMapper.selectById(job.getResultId());
if (result == null || !MODULE_TYPE.equals(result.getModuleType())
|| !task.getId().equals(result.getTaskId())) {
@@ -522,6 +532,7 @@ public class PublishTaskService {
|| task.getUpdatedAt() == null || !task.getUpdatedAt().isBefore(threshold)) {
return;
}
+ ensureTaskOwnedByCurrentInstance(task, "cleanup stale publish task");
if (taskFileJobService.countUnfinishedAssembleJobs(taskId, MODULE_TYPE) > 0L) {
return;
}
@@ -551,7 +562,7 @@ public class PublishTaskService {
result.setSuccess(0);
result.setErrorMessage(null);
fileResultMapper.updateById(result);
- taskFileJobService.enqueueAssembleResult(taskId, MODULE_TYPE, result.getId(), "task:" + taskId);
+ taskFileJobService.enqueueAssembleResult(taskId, MODULE_TYPE, result.getId(), ownerScopeKey(taskId));
}
private PersistedTask persistTask(PublishParseRequest request, List preparedFiles) {
@@ -568,7 +579,8 @@ public class PublishTaskService {
task.setSuccessFileCount(0);
task.setFailedFileCount(failedFiles);
task.setRequestJson(writeJson(request, "序列化上架任务失败"));
- task.setResultJson("{}");
+ task.setResultJson(writeJson(Map.of("ownerInstanceId", currentInstanceId()),
+ "Failed to save publish task instance owner"));
task.setErrorMessage(processableFiles > 0 ? null : "全部文件解析或店铺匹配失败");
task.setCreatedBy("user:" + request.getUserId());
task.setUserId(request.getUserId());
@@ -665,7 +677,7 @@ public class PublishTaskService {
task.setErrorMessage(null);
task.setFinishedAt(null);
fileTaskMapper.updateById(task);
- taskFileJobService.enqueueAssembleResult(taskId, MODULE_TYPE, result.getId(), "task:" + taskId);
+ taskFileJobService.enqueueAssembleResult(taskId, MODULE_TYPE, result.getId(), ownerScopeKey(taskId));
}
private List loadTaskDetails(List tasks) {
@@ -801,9 +813,47 @@ public class PublishTaskService {
|| (userId != null && !userId.equals(task.getUserId()))) {
throw new BusinessException("任务不存在");
}
+ ensureTaskOwnedByCurrentInstance(task, "access publish task");
return task;
}
+ public void ensureTaskOwnedByCurrentInstance(FileTaskEntity task, String operation) {
+ String owner = ownerFromTask(task);
+ if (owner == null || owner.isBlank() || Objects.equals(owner, currentInstanceId())) {
+ return;
+ }
+ log.warn("[publish] reject task operation because owner is another instance taskId={} operation={} owner={} current={}",
+ task == null ? null : task.getId(), operation, owner, currentInstanceId());
+ throw new TaskOwnerMismatchException(
+ task == null ? null : task.getId(), operation, owner, currentInstanceId());
+ }
+
+ private String ownerFromTask(FileTaskEntity task) {
+ if (task == null || task.getResultJson() == null || task.getResultJson().isBlank()) {
+ return null;
+ }
+ try {
+ JsonNode root = objectMapper.readTree(task.getResultJson());
+ if (root == null) {
+ return null;
+ }
+ String owner = root.path("ownerInstanceId").asText("");
+ return owner.isBlank() ? null : owner;
+ } catch (Exception ex) {
+ log.warn("[publish] read task owner failed taskId={} msg={}", task.getId(), safeMessage(ex));
+ return null;
+ }
+ }
+
+ private String currentInstanceId() {
+ String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId();
+ return firstNonBlank(instanceId, "unknown-instance");
+ }
+
+ private String ownerScopeKey(Long taskId) {
+ return "task:" + taskId + ":owner:" + currentInstanceId();
+ }
+
private PublishFileEntity requireFile(Long taskId, Long fileId) {
if (fileId == null || fileId <= 0) {
throw new BusinessException("file_id 不合法");
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookService.java
index 88d600c3..a2330690 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookService.java
@@ -30,7 +30,9 @@ import java.util.zip.ZipOutputStream;
@Service
public class PublishWorkbookService {
- public static final List HEADERS = List.of(
+ public static final List SOURCE_HEADERS = List.of(
+ "id", "ASIN", "国家", "品牌", "价格");
+ public static final List RESULT_HEADERS = List.of(
"id", "ASIN", "国家", "品牌", "价格", "状态", "同步状态", "同步国家");
public static final String XLSX_CONTENT_TYPE =
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
@@ -66,9 +68,9 @@ public class PublishWorkbookService {
if (!validatedSheets.contains(sheetNo)) {
throw new BusinessException("工作表 " + sheetName + " 缺少严格的上架表头");
}
- List values = new ArrayList<>(HEADERS.size());
+ List values = new ArrayList<>(SOURCE_HEADERS.size());
boolean nonEmpty = false;
- for (int columnIndex = 0; columnIndex < HEADERS.size(); columnIndex++) {
+ for (int columnIndex = 0; columnIndex < SOURCE_HEADERS.size(); columnIndex++) {
String value = normalize(rowMap.get(columnIndex));
values.add(value);
nonEmpty = nonEmpty || !value.isBlank();
@@ -163,19 +165,14 @@ public class PublishWorkbookService {
if (headerMap == null || headerMap.isEmpty()) {
throw new BusinessException("Excel 表头为空");
}
- for (int index = 0; index < HEADERS.size(); index++) {
+ for (int index = 0; index < SOURCE_HEADERS.size(); index++) {
String actual = normalize(headerMap.get(index));
- String expected = HEADERS.get(index);
+ String expected = SOURCE_HEADERS.get(index);
if (!expected.equals(actual)) {
throw new BusinessException("Excel 表头不匹配，第 " + (index + 1)
+ " 列应为 " + expected + "，实际为 " + actual);
}
}
- for (Map.Entry entry : headerMap.entrySet()) {
- if (entry.getKey() != null && entry.getKey() >= HEADERS.size() && !normalize(entry.getValue()).isBlank()) {
- throw new BusinessException("Excel 表头必须严格为: " + String.join("/", HEADERS));
- }
- }
}
private PublishRowDto toRow(List values) {
@@ -185,9 +182,6 @@ public class PublishWorkbookService {
row.setCountry(values.get(2));
row.setBrand(values.get(3));
row.setPrice(values.get(4));
- row.setStatus(values.get(5));
- row.setSyncStatus(values.get(6));
- row.setSyncCountries(values.get(7));
return row;
}
@@ -210,9 +204,9 @@ public class PublishWorkbookService {
private void writeSheet(Sheet sheet, List rows, CellStyle headerStyle) {
Row header = sheet.createRow(0);
- for (int index = 0; index < HEADERS.size(); index++) {
+ for (int index = 0; index < RESULT_HEADERS.size(); index++) {
Cell cell = header.createCell(index);
- cell.setCellValue(HEADERS.get(index));
+ cell.setCellValue(RESULT_HEADERS.get(index));
cell.setCellStyle(headerStyle);
}
int rowIndex = 1;
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinHistoryItemVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinHistoryItemVo.java
index 1ccd3a20..d55210be 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinHistoryItemVo.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinHistoryItemVo.java
@@ -14,7 +14,7 @@ public class SimilarAsinHistoryItemVo {
private String sourceFilename;
@Schema(description = "最终结果文件名。任务完成并生成 xlsx 后返回。", example = "17-result.xlsx")
private String resultFilename;
- @Schema(description = "最终结果文件下载地址。后端基于 OSS objectKey 生成的公开直链 URL。", example = "https://bucket.oss-cn-hangzhou.aliyuncs.com/result/similar_asin/xxx/17-result.xlsx")
+ @Schema(description = "最终结果文件下载地址。后端基于 MinIO objectKey 生成的公开直链 URL。", example = "http://47.110.241.161:9000/nanri-ai-images/result/similar_asin/xxx/17-result.xlsx")
private String downloadUrl;
private Long fileJobId;
private String fileStatus;
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 fa21a7f2..d20044cb 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
@@ -206,6 +206,7 @@ public class SimilarAsinTaskService {
"id",
"asin",
"国家",
+ "价格",
"卖家名称",
"品牌",
"是否有货",
@@ -219,9 +220,9 @@ public class SimilarAsinTaskService {
"阿里巴巴图片2"
);
- private static final int IMG_COL_MAIN = 11;
- private static final int IMG_COL_PUZZLE1 = 12;
- private static final int IMG_COL_PUZZLE2 = 13;
+ private static final int IMG_COL_MAIN = 12;
+ private static final int IMG_COL_PUZZLE1 = 13;
+ private static final int IMG_COL_PUZZLE2 = 14;
private final LocalFileStorageService localFileStorageService;
private final OssStorageService ossStorageService;
@@ -4533,6 +4534,11 @@ public class SimilarAsinTaskService {
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getDisplayId(), parsedRow.getSourceId()));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getAsin(), ""));
row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getCountry(), ""));
+ row.createCell(col++).setCellValue(resultRow == null
+ ? firstNonBlank(parsedRow.getPrice(), readValueByHeader(parsedRow, "价格", "price"))
+ : firstNonBlank(
+ resultRow.getPrice(),
+ firstNonBlank(parsedRow.getPrice(), readValueByHeader(parsedRow, "价格", "price"))));
row.createCell(col++).setCellValue(readValueByHeader(parsedRow, "卖家名称", "卖家名", "卖家", "店铺名称", "店铺名", "seller name", "seller_name", "seller-name", "sellername", "store name", "shop name"));
row.createCell(col++).setCellValue(readValueByHeader(parsedRow, "品牌", "brand"));
String isStock = resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getIsStock());
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 b9593bba..eb5696b2 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
@@ -95,7 +95,7 @@ public class TaskFileJobService {
List ownerJobs = taskFileJobMapper.selectList(new LambdaQueryWrapper()
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
.lt(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)
- .in(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN"))
+ .in(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN", "PUBLISH"))
.like(TaskFileJobEntity::getScopeKey, ownerMarker)
.orderByAsc(TaskFileJobEntity::getUpdatedAt)
.last("limit " + safeLimit));
@@ -108,7 +108,7 @@ public class TaskFileJobService {
.in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED"))
.lt(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT)
.and(wrapper -> wrapper
- .notIn(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN"))
+ .notIn(TaskFileJobEntity::getModuleType, List.of("APPEARANCE_PATENT", "SIMILAR_ASIN", "PUBLISH"))
.or()
.isNull(TaskFileJobEntity::getScopeKey)
.or()
diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java
index 9715042d..397736ca 100644
--- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java
+++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java
@@ -103,6 +103,9 @@ public class TaskHeartbeatService {
return null;
}
String moduleType = task.getModuleType() == null ? "" : task.getModuleType();
+ if (PublishTaskService.MODULE_TYPE.equals(moduleType)) {
+ publishTaskService.ensureTaskOwnedByCurrentInstance(task, "publish task heartbeat");
+ }
String status = task.getStatus();
if (!STATUS_RUNNING.equals(status)) {
log.warn("[task-heartbeat] file task is not running taskId={} actualUserId={} moduleType={} status={}",
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 de7f2cb2..28c5e192 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
@@ -95,7 +95,7 @@ public class TaskResultFileJobWorker {
}
public void process(TaskFileJobEntity job) {
- if (isOwnerScopedCozeJob(job) && !isOwnedByCurrentInstance(job)) {
+ if (isOwnerScopedJob(job) && !isOwnedByCurrentInstance(job)) {
log.debug("[task-file-job] skip owner-scoped job because owner is another instance jobId={} taskId={} moduleType={} owner={} current={}",
job.getId(), job.getTaskId(), job.getModuleType(), ownerFromScopeKey(job.getScopeKey()), currentInstanceId());
return;
@@ -149,7 +149,7 @@ public class TaskResultFileJobWorker {
}
boolean completed = dispatch(job);
if (!completed) {
- if (isOwnerScopedCozeJob(job)) {
+ if (isOwnerScopedJob(job)) {
taskFileJobService.touchRunning(job.getId());
log.info("[task-file-job] process waiting for async coze result jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
@@ -219,11 +219,13 @@ public class TaskResultFileJobWorker {
return result == null ? null : result.getResultFileUrl();
}
- private boolean isOwnerScopedCozeJob(TaskFileJobEntity job) {
+ private boolean isOwnerScopedJob(TaskFileJobEntity job) {
if (job == null || job.getModuleType() == null) {
return false;
}
- return "APPEARANCE_PATENT".equals(job.getModuleType()) || "SIMILAR_ASIN".equals(job.getModuleType());
+ return "APPEARANCE_PATENT".equals(job.getModuleType())
+ || "SIMILAR_ASIN".equals(job.getModuleType())
+ || "PUBLISH".equals(job.getModuleType());
}
private boolean isOwnedByCurrentInstance(TaskFileJobEntity job) {
diff --git a/backend-java/src/main/resources/application-local.example.yml b/backend-java/src/main/resources/application-local.example.yml
index dd84ed0f..b278ef98 100644
--- a/backend-java/src/main/resources/application-local.example.yml
+++ b/backend-java/src/main/resources/application-local.example.yml
@@ -12,9 +12,12 @@ AIIMAGE_DB_URL=jdbc:mysql://127.0.0.1:3306/aiimage?useUnicode=true&characterEnco
AIIMAGE_DB_USERNAME=change-me
AIIMAGE_DB_PASSWORD=change-me
-AIIMAGE_OSS_REGION=cn-hangzhou
-AIIMAGE_OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com
-AIIMAGE_OSS_BUCKET=change-me
+AIIMAGE_OSS_REGION=us-east-1
+AIIMAGE_OSS_ENDPOINT=http://47.110.241.161:9000
+AIIMAGE_OSS_PUBLIC_ENDPOINT=http://47.110.241.161:9000
+AIIMAGE_OSS_BUCKET=nanri-ai-images
+AIIMAGE_IMAGE_VIDEO_OSS_BUCKET=shufu-video
+AIIMAGE_DIGITAL_HUMAN_OSS_BUCKET=nanri-ai-digital-human
AIIMAGE_OSS_ACCESS_KEY_ID=change-me
AIIMAGE_OSS_ACCESS_KEY_SECRET=change-me
diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml
index f2680585..a423c90a 100644
--- a/backend-java/src/main/resources/application.yml
+++ b/backend-java/src/main/resources/application.yml
@@ -84,13 +84,14 @@ aiimage:
read-timeout-millis: ${AIIMAGE_INSTANCE_ROUTE_READ_TIMEOUT_MILLIS:300000}
request-body-cache-limit-bytes: ${AIIMAGE_INSTANCE_ROUTE_REQUEST_BODY_CACHE_LIMIT_BYTES:104857600}
oss:
- region: ${AIIMAGE_OSS_REGION:cn-hangzhou}
- endpoint: ${AIIMAGE_OSS_ENDPOINT:oss-cn-hangzhou.aliyuncs.com}
- public-endpoint: ${AIIMAGE_OSS_PUBLIC_ENDPOINT:oss-cn-hangzhou.aliyuncs.com}
+ region: ${AIIMAGE_OSS_REGION:us-east-1}
+ endpoint: ${AIIMAGE_OSS_ENDPOINT:http://47.110.241.161:9000}
+ public-endpoint: ${AIIMAGE_OSS_PUBLIC_ENDPOINT:http://47.110.241.161:9000}
bucket: ${AIIMAGE_OSS_BUCKET:nanri-ai-images}
image-video-bucket: ${AIIMAGE_IMAGE_VIDEO_OSS_BUCKET:shufu-video}
- access-key-id: ${AIIMAGE_OSS_ACCESS_KEY_ID:LTAI5tNpyvzMNz9f2dHarsm8}
- access-key-secret: ${AIIMAGE_OSS_ACCESS_KEY_SECRET:bQSZnFH455i8tzyOgeahJmUzwmhynz}
+ digital-human-bucket: ${AIIMAGE_DIGITAL_HUMAN_OSS_BUCKET:nanri-ai-digital-human}
+ access-key-id: ${AIIMAGE_OSS_ACCESS_KEY_ID:appuser}
+ access-key-secret: ${AIIMAGE_OSS_ACCESS_KEY_SECRET:AppUser@2024SecureKey}
transient-storage:
enabled: ${AIIMAGE_TRANSIENT_STORAGE_ENABLED:true}
endpoint: ${AIIMAGE_TRANSIENT_STORAGE_ENDPOINT:http://121.196.149.225:9000}
diff --git a/backend-java/src/main/resources/db/V78__image_video_task_instance_routing.sql b/backend-java/src/main/resources/db/V78__image_video_task_instance_routing.sql
new file mode 100644
index 00000000..50f2c390
--- /dev/null
+++ b/backend-java/src/main/resources/db/V78__image_video_task_instance_routing.sql
@@ -0,0 +1,3 @@
+ALTER TABLE `biz_image_video_async_task`
+ ADD COLUMN `owner_instance_id` VARCHAR(128) NULL AFTER `user_id`,
+ ADD KEY `idx_image_video_owner_status` (`owner_instance_id`, `status`, `updated_at`);
diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskServiceTest.java
index 7155452a..ed973af7 100644
--- a/backend-java/src/test/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskServiceTest.java
+++ b/backend-java/src/test/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskServiceTest.java
@@ -2,12 +2,16 @@ package com.nanri.aiimage.modules.imagevideo.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.config.ImageVideoProperties;
+import com.nanri.aiimage.config.InstanceMetadata;
+import com.nanri.aiimage.common.exception.TaskOwnerMismatchException;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.imagevideo.mapper.ImageVideoAsyncTaskMapper;
+import com.nanri.aiimage.modules.imagevideo.model.dto.DouyinCopyRequest;
import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity;
import com.nanri.aiimage.modules.imagevideo.model.vo.DouyinCopyVo;
import org.junit.jupiter.api.Test;
import org.springframework.core.task.TaskExecutor;
+import org.mockito.ArgumentCaptor;
import java.time.LocalDateTime;
import java.util.List;
@@ -15,6 +19,7 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -55,7 +60,8 @@ class ImageVideoAsyncTaskServiceTest {
ObjectMapper objectMapper = new ObjectMapper();
TaskExecutor directExecutor = Runnable::run;
ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
- taskMapper, cozeService, workflowConfigService, mock(ImageVideoArchiveService.class), objectMapper, directExecutor);
+ taskMapper, cozeService, workflowConfigService, mock(ImageVideoArchiveService.class), objectMapper,
+ directExecutor, instanceMetadata("server-121"));
ImageVideoAsyncTaskEntity task = waitingDouyinTask();
Map cozeResult = Map.of("data", List.of(Map.of(
@@ -67,7 +73,7 @@ class ImageVideoAsyncTaskServiceTest {
copyResult.setScriptDraft("这条裙裤绝了，遮肉显腿长，快拍！");
when(taskMapper.selectList(any())).thenReturn(List.of(task));
- when(taskMapper.claimWaiting(79L)).thenReturn(1);
+ when(taskMapper.claimWaiting(79L, "server-121")).thenReturn(1);
when(taskMapper.selectById(79L)).thenReturn(task);
when(workflowConfigService.douyinCopyWorkflowId()).thenReturn("7652941112982798388");
when(cozeService.getWorkflowResult(1L, "7652941112982798388", "7662364426301964329"))
@@ -90,13 +96,14 @@ class ImageVideoAsyncTaskServiceTest {
ImageVideoCozeService cozeService = mock(ImageVideoCozeService.class);
ImageVideoWorkflowConfigService workflowConfigService = mock(ImageVideoWorkflowConfigService.class);
ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
- taskMapper, cozeService, workflowConfigService, mock(ImageVideoArchiveService.class), new ObjectMapper(), Runnable::run);
+ taskMapper, cozeService, workflowConfigService, mock(ImageVideoArchiveService.class), new ObjectMapper(),
+ Runnable::run, instanceMetadata("server-121"));
ImageVideoAsyncTaskEntity task = waitingDouyinTask();
Map cozeResult = Map.of("data", List.of(Map.of("execute_status", "Running")));
when(taskMapper.selectList(any())).thenReturn(List.of(task));
- when(taskMapper.claimWaiting(79L)).thenReturn(1);
+ when(taskMapper.claimWaiting(79L, "server-121")).thenReturn(1);
when(taskMapper.selectById(79L)).thenReturn(task);
when(workflowConfigService.douyinCopyWorkflowId()).thenReturn("7652941112982798388");
when(cozeService.getWorkflowResult(1L, "7652941112982798388", "7662364426301964329"))
@@ -117,7 +124,8 @@ class ImageVideoAsyncTaskServiceTest {
ImageVideoWorkflowConfigService workflowConfigService = mock(ImageVideoWorkflowConfigService.class);
ImageVideoArchiveService archiveService = mock(ImageVideoArchiveService.class);
ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
- taskMapper, cozeService, workflowConfigService, archiveService, new ObjectMapper(), Runnable::run);
+ taskMapper, cozeService, workflowConfigService, archiveService, new ObjectMapper(), Runnable::run,
+ instanceMetadata("server-121"));
ImageVideoAsyncTaskEntity task = waitingWorkflowTask();
Map cozeResult = Map.of("data", List.of(Map.of(
"execute_status", "Fail",
@@ -125,7 +133,7 @@ class ImageVideoAsyncTaskServiceTest {
"error_message", "model unavailable")));
when(taskMapper.selectList(any())).thenReturn(List.of(task));
- when(taskMapper.claimWaiting(98L)).thenReturn(1);
+ when(taskMapper.claimWaiting(98L, "server-121")).thenReturn(1);
when(taskMapper.selectById(98L)).thenReturn(task);
when(workflowConfigService.imageVideoWorkflowId()).thenReturn("workflow-1");
when(cozeService.getWorkflowResult(1L, "workflow-1", "exec-98")).thenReturn(cozeResult);
@@ -148,7 +156,8 @@ class ImageVideoAsyncTaskServiceTest {
mock(ImageVideoWorkflowConfigService.class),
mock(ImageVideoArchiveService.class),
new ObjectMapper(),
- Runnable::run);
+ Runnable::run,
+ instanceMetadata("server-121"));
ImageVideoAsyncTaskEntity task = waitingWorkflowTask();
task.setStatus("FAILED");
task.setCozeStatus("FAIL");
@@ -171,7 +180,8 @@ class ImageVideoAsyncTaskServiceTest {
mock(ImageVideoWorkflowConfigService.class),
mock(ImageVideoArchiveService.class),
new ObjectMapper(),
- Runnable::run);
+ Runnable::run,
+ instanceMetadata("server-121"));
ImageVideoAsyncTaskEntity task = waitingWorkflowTask();
task.setStatus("FAILED");
task.setCozeStatus("RUNNING");
@@ -194,7 +204,8 @@ class ImageVideoAsyncTaskServiceTest {
mock(ImageVideoWorkflowConfigService.class),
mock(ImageVideoArchiveService.class),
new ObjectMapper(),
- Runnable::run);
+ Runnable::run,
+ instanceMetadata("server-121"));
when(taskMapper.delete(any())).thenReturn(3);
service.cleanupFailedTasks();
@@ -202,10 +213,107 @@ class ImageVideoAsyncTaskServiceTest {
verify(taskMapper).delete(any());
}
+ @Test
+ void submittedTaskIsBoundToCurrentInstance() {
+ ImageVideoAsyncTaskMapper taskMapper = mock(ImageVideoAsyncTaskMapper.class);
+ ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
+ taskMapper,
+ mock(ImageVideoCozeService.class),
+ mock(ImageVideoWorkflowConfigService.class),
+ mock(ImageVideoArchiveService.class),
+ new ObjectMapper(),
+ Runnable::run,
+ instanceMetadata("server-121"));
+ DouyinCopyRequest request = new DouyinCopyRequest();
+ request.setUserId(1L);
+ request.setUrl("https://example.com/video");
+
+ service.submitDouyinCopy(request);
+
+ ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(ImageVideoAsyncTaskEntity.class);
+ verify(taskMapper).insert(taskCaptor.capture());
+ assertEquals("server-121", taskCaptor.getValue().getOwnerInstanceId());
+ }
+
+ @Test
+ void taskQueryOnAnotherInstanceTriggersOwnerForwarding() {
+ ImageVideoAsyncTaskMapper taskMapper = mock(ImageVideoAsyncTaskMapper.class);
+ ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
+ taskMapper,
+ mock(ImageVideoCozeService.class),
+ mock(ImageVideoWorkflowConfigService.class),
+ mock(ImageVideoArchiveService.class),
+ new ObjectMapper(),
+ Runnable::run,
+ instanceMetadata("server-121"));
+ ImageVideoAsyncTaskEntity task = waitingWorkflowTask();
+ task.setOwnerInstanceId("server-110");
+ when(taskMapper.selectOne(any())).thenReturn(task);
+
+ TaskOwnerMismatchException exception = assertThrows(
+ TaskOwnerMismatchException.class,
+ () -> service.getTask(98L, 1L));
+
+ assertEquals(98L, exception.getTaskId());
+ assertEquals("server-110", exception.getOwnerInstanceId());
+ assertEquals("server-121", exception.getCurrentInstanceId());
+ }
+
+ @Test
+ void applicationStartRequeuesInterruptedPollingTasksForOwner() {
+ ImageVideoAsyncTaskMapper taskMapper = mock(ImageVideoAsyncTaskMapper.class);
+ ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
+ taskMapper,
+ mock(ImageVideoCozeService.class),
+ mock(ImageVideoWorkflowConfigService.class),
+ mock(ImageVideoArchiveService.class),
+ new ObjectMapper(),
+ Runnable::run,
+ instanceMetadata("server-121"));
+ when(taskMapper.requeuePollingTasks("server-121")).thenReturn(2);
+
+ service.recoverInterruptedPollingTasks();
+
+ verify(taskMapper).requeuePollingTasks("server-121");
+ }
+
+ @Test
+ void overdueTaskStillConsumesTerminalCozeResultAfterRecovery() {
+ ImageVideoAsyncTaskMapper taskMapper = mock(ImageVideoAsyncTaskMapper.class);
+ ImageVideoCozeService cozeService = mock(ImageVideoCozeService.class);
+ ImageVideoWorkflowConfigService workflowConfigService = mock(ImageVideoWorkflowConfigService.class);
+ ImageVideoAsyncTaskService service = new ImageVideoAsyncTaskService(
+ taskMapper,
+ cozeService,
+ workflowConfigService,
+ mock(ImageVideoArchiveService.class),
+ new ObjectMapper(),
+ Runnable::run,
+ instanceMetadata("server-121"));
+ ImageVideoAsyncTaskEntity task = waitingWorkflowTask();
+ task.setSubmittedAt(LocalDateTime.now().minusHours(2));
+ Map cozeResult = Map.of("data", List.of(Map.of(
+ "execute_status", "Success",
+ "output", "{\"Output\":\"https://example.com/result.mp4\"}")));
+
+ when(taskMapper.selectList(any())).thenReturn(List.of(task));
+ when(taskMapper.claimWaiting(98L, "server-121")).thenReturn(1);
+ when(taskMapper.selectById(98L)).thenReturn(task);
+ when(workflowConfigService.imageVideoWorkflowId()).thenReturn("workflow-1");
+ when(cozeService.getWorkflowResult(1L, "workflow-1", "exec-98")).thenReturn(cozeResult);
+
+ service.pollWaitingTasks();
+
+ assertEquals("SUCCESS", task.getStatus());
+ assertEquals("SUCCESS", task.getCozeStatus());
+ assertNotNull(task.getCompletedAt());
+ }
+
private ImageVideoAsyncTaskEntity waitingDouyinTask() {
ImageVideoAsyncTaskEntity task = new ImageVideoAsyncTaskEntity();
task.setId(79L);
task.setUserId(1L);
+ task.setOwnerInstanceId("server-121");
task.setTaskType("DOUYIN_COPY");
task.setStatus("WAITING");
task.setCozeExecuteId("7662364426301964329");
@@ -222,4 +330,8 @@ class ImageVideoAsyncTaskServiceTest {
task.setCozeExecuteId("exec-98");
return task;
}
+
+ private InstanceMetadata instanceMetadata(String instanceId) {
+ return new InstanceMetadata(instanceId);
+ }
}
diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskServiceTest.java
index 04345bae..c3a91aaa 100644
--- a/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskServiceTest.java
+++ b/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskServiceTest.java
@@ -4,12 +4,16 @@ import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
+import com.nanri.aiimage.common.exception.TaskOwnerMismatchException;
+import com.nanri.aiimage.config.InstanceMetadata;
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.publish.mapper.PublishFileMapper;
import com.nanri.aiimage.modules.publish.mapper.PublishItemMapper;
+import com.nanri.aiimage.modules.publish.model.dto.PublishParseRequest;
import com.nanri.aiimage.modules.publish.model.dto.PublishResultFileDto;
import com.nanri.aiimage.modules.publish.model.dto.PublishRowDto;
+import com.nanri.aiimage.modules.publish.model.dto.PublishSourceFileDto;
import com.nanri.aiimage.modules.publish.model.dto.PublishSubmitResultRequest;
import com.nanri.aiimage.modules.publish.model.entity.PublishFileEntity;
import com.nanri.aiimage.modules.publish.model.entity.PublishItemEntity;
@@ -20,6 +24,7 @@ import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
+import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -29,9 +34,11 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import java.time.LocalDateTime;
+import java.io.File;
import java.util.List;
import java.util.function.Consumer;
@@ -42,6 +49,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -74,11 +82,17 @@ class PublishTaskServiceTest {
@Mock private OssStorageService ossStorageService;
@Mock private ObjectMapper objectMapper;
@Mock private TransactionTemplate transactionTemplate;
+ @Mock private InstanceMetadata instanceMetadata;
@InjectMocks private PublishTaskService service;
@BeforeEach
void executeTransactionsInline() {
+ lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
+ lenient().when(transactionTemplate.execute(any())).thenAnswer(invocation -> {
+ TransactionCallback callback = invocation.getArgument(0);
+ return callback.doInTransaction(null);
+ });
lenient().doAnswer(invocation -> {
Consumer callback = invocation.getArgument(0);
callback.accept(null);
@@ -86,6 +100,65 @@ class PublishTaskServiceTest {
}).when(transactionTemplate).executeWithoutResult(any());
}
+ @Test
+ void parseStoresCurrentInstanceAsTaskOwner() throws Exception {
+ PublishParseRequest request = new PublishParseRequest();
+ request.setUserId(7L);
+ request.setPublishCountry("DE");
+ PublishSourceFileDto source = new PublishSourceFileDto();
+ source.setFileKey("uploads/shop.xlsx");
+ source.setOriginalFilename("shop.xlsx");
+ request.setFiles(List.of(source));
+
+ File localFile = mock(File.class);
+ when(localFile.exists()).thenReturn(true);
+ when(localFileStorageService.findLocalSourceFile(source.getFileKey())).thenReturn(localFile);
+ when(ziniaoShopSwitchService.findIndexedStoreByName("shop", false)).thenReturn(
+ new ZiniaoShopMatchResultVo(true, "shop-1", "shop", "company", "Amazon",
+ 7L, null, "MATCHED", null));
+ when(workbookService.parse(localFile)).thenReturn(
+ new PublishWorkbookService.ParsedWorkbook(List.of(row("1"))));
+ ObjectMapper realObjectMapper = new ObjectMapper();
+ when(objectMapper.writeValueAsString(any())).thenAnswer(invocation ->
+ realObjectMapper.writeValueAsString(invocation.getArgument(0)));
+ doAnswer(invocation -> {
+ FileTaskEntity inserted = invocation.getArgument(0);
+ inserted.setId(501L);
+ return 1;
+ }).when(fileTaskMapper).insert(any(FileTaskEntity.class));
+ doAnswer(invocation -> {
+ PublishFileEntity inserted = invocation.getArgument(0);
+ inserted.setId(601L);
+ return 1;
+ }).when(publishFileMapper).insert(any(PublishFileEntity.class));
+
+ service.parseAndCreateTask(request);
+
+ org.mockito.ArgumentCaptor taskCaptor =
+ org.mockito.ArgumentCaptor.forClass(FileTaskEntity.class);
+ verify(fileTaskMapper).insert(taskCaptor.capture());
+ assertEquals("instance-a", realObjectMapper.readTree(
+ taskCaptor.getValue().getResultJson()).path("ownerInstanceId").asText());
+ }
+
+ @Test
+ void taskAccessRejectsAnotherInstanceForExistingRouterToForward() throws Exception {
+ long taskId = 109L;
+ FileTaskEntity task = task(taskId, 7L, "RUNNING");
+ task.setResultJson("{\"ownerInstanceId\":\"instance-b\"}");
+ when(fileTaskMapper.selectById(taskId)).thenReturn(task);
+ when(objectMapper.readTree(task.getResultJson())).thenReturn(
+ new ObjectMapper().readTree(task.getResultJson()));
+
+ TaskOwnerMismatchException error = assertThrows(TaskOwnerMismatchException.class,
+ () -> service.getTaskDetail(taskId, 7L));
+
+ assertEquals(taskId, error.getTaskId());
+ assertEquals("instance-b", error.getOwnerInstanceId());
+ assertEquals("instance-a", error.getCurrentInstanceId());
+ verifyNoInteractions(publishItemMapper);
+ }
+
@Test
void resultCallbackRejectsAnotherUserBeforeReadingOrReplacingRows() {
long taskId = 101L;
@@ -180,7 +253,8 @@ class PublishTaskServiceTest {
assertEquals("SUCCESS", file.getStatus());
verify(taskFileJobService).enqueueAssembleResult(
- taskId, PublishTaskService.MODULE_TYPE, resultId, "task:" + taskId);
+ taskId, PublishTaskService.MODULE_TYPE, resultId,
+ "task:" + taskId + ":owner:instance-a");
verify(lock).close();
}
@@ -265,7 +339,8 @@ class PublishTaskServiceTest {
verify(taskFileJobService, times(2))
.countUnfinishedAssembleJobs(taskId, PublishTaskService.MODULE_TYPE);
verify(taskFileJobService).enqueueAssembleResult(
- taskId, PublishTaskService.MODULE_TYPE, resultId, "task:" + taskId);
+ taskId, PublishTaskService.MODULE_TYPE, resultId,
+ "task:" + taskId + ":owner:instance-a");
verify(fileTaskMapper).updateById(task);
verify(fileResultMapper).updateById(result);
verify(lock).close();
diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookServiceTest.java
index f959e619..40cf3540 100644
--- a/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookServiceTest.java
+++ b/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookServiceTest.java
@@ -22,6 +22,7 @@ import java.util.zip.ZipFile;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -30,13 +31,22 @@ class PublishWorkbookServiceTest {
private final PublishWorkbookService service = new PublishWorkbookService();
@Test
- void parsesOnlyTheExactEightHeadersAcrossNonEmptySheets() throws Exception {
+ void parsesOnlyFirstFiveColumnsAcrossNonEmptySheets() throws Exception {
Path directory = Files.createTempDirectory("publish-parse-");
try {
File valid = directory.resolve("valid.xlsx").toFile();
try (Workbook workbook = new XSSFWorkbook();
FileOutputStream output = new FileOutputStream(valid)) {
- writeSourceSheet(workbook.createSheet("英国数据"), "英国", "B001");
+ Sheet uk = workbook.createSheet("英国数据");
+ writeSourceSheet(uk, "英国", "B001");
+ for (int index = PublishWorkbookService.SOURCE_HEADERS.size();
+ index < PublishWorkbookService.RESULT_HEADERS.size(); index++) {
+ uk.getRow(0).createCell(index)
+ .setCellValue(PublishWorkbookService.RESULT_HEADERS.get(index));
+ }
+ uk.getRow(1).createCell(5).setCellValue("旧状态");
+ uk.getRow(1).createCell(6).setCellValue("旧同步状态");
+ uk.getRow(1).createCell(7).setCellValue("旧同步国家");
workbook.createSheet("空白页");
writeSourceSheet(workbook.createSheet("德国数据"), "DE", "B002");
workbook.write(output);
@@ -45,6 +55,9 @@ class PublishWorkbookServiceTest {
PublishWorkbookService.ParsedWorkbook parsed = service.parse(valid);
assertEquals(2, parsed.rows().size());
assertEquals("B001", parsed.rows().get(0).getAsin());
+ assertNull(parsed.rows().get(0).getStatus());
+ assertNull(parsed.rows().get(0).getSyncStatus());
+ assertNull(parsed.rows().get(0).getSyncCountries());
assertEquals("DE", parsed.rows().get(1).getCountry());
File invalid = directory.resolve("invalid.xlsx").toFile();
@@ -52,7 +65,7 @@ class PublishWorkbookServiceTest {
FileOutputStream output = new FileOutputStream(invalid)) {
Sheet sheet = workbook.createSheet("错误表头");
Row header = sheet.createRow(0);
- List headers = new ArrayList<>(PublishWorkbookService.HEADERS);
+ List headers = new ArrayList<>(PublishWorkbookService.SOURCE_HEADERS);
headers.set(1, "Asin");
for (int index = 0; index < headers.size(); index++) {
header.createCell(index).setCellValue(headers.get(index));
@@ -82,8 +95,8 @@ class PublishWorkbookServiceTest {
Set.of(workbook.getSheetName(0), workbook.getSheetName(1)));
Sheet uk = workbook.getSheet("英国");
assertNotNull(uk);
- for (int index = 0; index < PublishWorkbookService.HEADERS.size(); index++) {
- assertEquals(PublishWorkbookService.HEADERS.get(index),
+ for (int index = 0; index < PublishWorkbookService.RESULT_HEADERS.size(); index++) {
+ assertEquals(PublishWorkbookService.RESULT_HEADERS.get(index),
uk.getRow(0).getCell(index).getStringCellValue());
}
assertEquals(CellType.NUMERIC, uk.getRow(1).getCell(4).getCellType());
@@ -124,8 +137,8 @@ class PublishWorkbookServiceTest {
private void writeSourceSheet(Sheet sheet, String country, String asin) {
Row header = sheet.createRow(0);
- for (int index = 0; index < PublishWorkbookService.HEADERS.size(); index++) {
- header.createCell(index).setCellValue(PublishWorkbookService.HEADERS.get(index));
+ for (int index = 0; index < PublishWorkbookService.SOURCE_HEADERS.size(); index++) {
+ header.createCell(index).setCellValue(PublishWorkbookService.SOURCE_HEADERS.get(index));
}
Row data = sheet.createRow(1);
data.createCell(0).setCellValue("1");
@@ -133,9 +146,6 @@ class PublishWorkbookServiceTest {
data.createCell(2).setCellValue(country);
data.createCell(3).setCellValue("Brand");
data.createCell(4).setCellValue(50);
- data.createCell(5).setCellValue("成功");
- data.createCell(6).setCellValue("成功");
- data.createCell(7).setCellValue("德国，法国");
}
private PublishRowDto row(String id, String asin, String country, String price) {
diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTest.java
index 3edd55e7..fb17f55b 100644
--- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTest.java
+++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTest.java
@@ -3,10 +3,29 @@ package com.nanri.aiimage.modules.similarasin.service;
import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Field;
+import java.util.List;
+
import static org.junit.jupiter.api.Assertions.assertEquals;
class SimilarAsinTaskServiceTest {
+ @Test
+ @SuppressWarnings("unchecked")
+ void resultWorkbookRestoresPriceAfterCountryAndShiftsImageColumns() throws Exception {
+ Field headersField = SimilarAsinTaskService.class.getDeclaredField("RESULT_HEADERS");
+ headersField.setAccessible(true);
+ List headers = (List) headersField.get(null);
+
+ assertEquals(List.of(
+ "id", "asin", "国家", "价格", "卖家名称", "品牌", "是否有货", "相似度",
+ "是否符合类目", "不符合理由", "产品类目", "状态", "主图", "阿里巴巴图片1", "阿里巴巴图片2"
+ ), headers);
+ assertEquals(12, staticIntField("IMG_COL_MAIN"));
+ assertEquals(13, staticIntField("IMG_COL_PUZZLE1"));
+ assertEquals(14, staticIntField("IMG_COL_PUZZLE2"));
+ }
+
@Test
void resultStatusUsesReturnedCozeDataAndImages() {
assertEquals("\u5931\u8d25", SimilarAsinTaskService.resolveResultStatus(null));
@@ -30,4 +49,10 @@ class SimilarAsinTaskServiceTest {
SimilarAsinResultRowDto withVisibleResultData = new SimilarAsinResultRowDto();
assertEquals("\u6210\u529f", SimilarAsinTaskService.resolveResultStatus(withVisibleResultData, "", "80%", "", "", ""));
}
+
+ private int staticIntField(String name) throws Exception {
+ Field field = SimilarAsinTaskService.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.getInt(null);
+ }
}
diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java
index 3724e4de..487a5d4c 100644
--- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java
+++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java
@@ -25,6 +25,7 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
+import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -33,6 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -83,7 +85,10 @@ class TaskHeartbeatServiceTest {
assertTrue(result.isAlive());
assertEquals(PublishTaskService.MODULE_TYPE, result.getModuleType());
- verify(fileTaskMapper).update(isNull(), any(LambdaUpdateWrapper.class));
+ InOrder routingBeforeUpdate = inOrder(publishTaskService, fileTaskMapper);
+ routingBeforeUpdate.verify(publishTaskService)
+ .ensureTaskOwnedByCurrentInstance(task, "publish task heartbeat");
+ routingBeforeUpdate.verify(fileTaskMapper).update(isNull(), any(LambdaUpdateWrapper.class));
verify(publishTaskService).touchHeartbeat(taskId, request);
}
}
diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorkerTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorkerTest.java
index d54b6e24..b0aedf03 100644
--- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorkerTest.java
+++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorkerTest.java
@@ -93,13 +93,14 @@ class TaskResultFileJobWorkerTest {
job.setTaskId(taskId);
job.setResultId(resultId);
job.setModuleType(PublishTaskService.MODULE_TYPE);
- job.setScopeKey("publish:20141");
+ job.setScopeKey("task:20141:owner:instance-a");
FileResultEntity result = new FileResultEntity();
result.setResultFileUrl("result/publish/20141.xlsx");
TaskDistributedLockService.LockHandle lock = mock(TaskDistributedLockService.LockHandle.class);
when(taskFileJobService.markRunning(jobId)).thenReturn(true);
+ when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
when(taskDistributedLockService.acquire(
PublishTaskService.MODULE_TYPE,
taskId,
@@ -112,4 +113,19 @@ class TaskResultFileJobWorkerTest {
verify(taskFileJobService).markSuccess(job, "result/publish/20141.xlsx");
verifyNoInteractions(taskResultPayloadService);
}
+
+ @Test
+ void publishFileJobOwnedByAnotherInstanceIsNotClaimed() {
+ TaskFileJobEntity job = new TaskFileJobEntity();
+ job.setId(13642L);
+ job.setTaskId(20142L);
+ job.setResultId(22930L);
+ job.setModuleType(PublishTaskService.MODULE_TYPE);
+ job.setScopeKey("task:20142:owner:instance-b");
+ when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
+
+ worker.process(job);
+
+ verifyNoInteractions(taskFileJobService, taskDistributedLockService, publishTaskService);
+ }
}
diff --git a/frontend-vue/src/pages/brand/components/BrandPublishTab.vue b/frontend-vue/src/pages/brand/components/BrandPublishTab.vue
index 7ff8314a..9190596c 100644
--- a/frontend-vue/src/pages/brand/components/BrandPublishTab.vue
+++ b/frontend-vue/src/pages/brand/components/BrandPublishTab.vue
@@ -40,7 +40,6 @@
name="publish-country"
class="country-check-input"
:value="row.code"
- :disabled="hasQueueWork"
/>
{{ row.label }} ({{ row.code }})
@@ -54,13 +53,13 @@
v-for="row in COUNTRY_OPTIONS"
:key="row.code"
class="country-check-row"
- :class="{ disabled: row.code === publishCountry || hasQueueWork }"
+ :class="{ disabled: row.code === publishCountry }"
>
{{ row.label }} ({{ row.code }})
@@ -75,10 +74,10 @@
{{ operationHint }}
@@ -231,7 +230,7 @@ interface PublishDispatchOptions {
ziniaoVersion: ZiniaoVersion
}
-interface StoredPublishQueue {
+interface PublishQueueBatch {
taskId: number
pendingFileIds: number[]
activeFileId: number | null
@@ -239,6 +238,11 @@ interface StoredPublishQueue {
options: PublishDispatchOptions
}
+interface StoredPublishQueueState {
+ active: PublishQueueBatch | null
+ pending: PublishQueueBatch[]
+}
+
const selectedPaths = ref([])
const uploadedFiles = ref([])
const uploading = ref(false)
@@ -253,6 +257,7 @@ const currentFiles = ref([])
const pendingFileIds = ref([])
const activeFileId = ref(null)
const dispatchOptions = ref(null)
+const queuedBatches = ref([])
const taskSnapshots = ref>({})
const historyItems = ref([])
const missingTaskIds = ref([])
@@ -355,8 +360,9 @@ const progressLoop = useTaskProgressLoop({
})
const displayPaths = computed(() => selectedPaths.value.slice(0, 8))
-const hasQueueWork = computed(() => pendingFileIds.value.length > 0 || activeFileId.value != null)
-const selectionDisabled = computed(() => uploading.value || parsing.value || hasQueueWork.value)
+const activeBatchHasWork = computed(() => pendingFileIds.value.length > 0 || activeFileId.value != null)
+const hasQueueWork = computed(() => activeBatchHasWork.value || queuedBatches.value.length > 0)
+const selectionDisabled = computed(() => uploading.value || parsing.value)
const currentDetail = computed(() => {
const taskId = currentTaskId.value
@@ -512,13 +518,12 @@ async function submitRun() {
})
if (!parsed.taskId) throw new Error('后端未返回有效任务标识')
- currentTaskId.value = parsed.taskId
- currentFiles.value = parsed.files || []
- dispatchOptions.value = {
+ const options: PublishDispatchOptions = {
publishCountry: publishCountry.value,
syncCountries: syncCountries.value.filter((country) => country !== publishCountry.value),
ziniaoVersion: ziniaoVersion.value,
}
+ const files = parsed.files || []
applyTaskSnapshot(parsed.taskId, {
task: {
id: parsed.taskId,
@@ -527,26 +532,44 @@ async function submitRun() {
sourceFileCount: parsed.sourceFileCount,
totalRows: parsed.totalRows,
},
- files: parsed.files || [],
+ files,
result: parsed.result,
})
- pendingFileIds.value = currentFiles.value
+ const batch: PublishQueueBatch = {
+ taskId: parsed.taskId,
+ pendingFileIds: files
.filter((file) => !isTerminalStatus(file.status) && isMatchedFile(file))
.map((file) => file.fileId)
- .filter((fileId) => Number.isFinite(fileId) && fileId > 0)
- activeFileId.value = null
- saveQueueState()
+ .filter((fileId) => Number.isFinite(fileId) && fileId > 0),
+ activeFileId: null,
+ files,
+ options,
+ }
progressLoop.add(parsed.taskId)
+ selectedPaths.value = []
+ uploadedFiles.value = []
await Promise.all([loadDashboard(), loadHistory()])
- if (!pendingFileIds.value.length) {
+ if (!batch.pendingFileIds.length) {
queueMessage.value = '解析完成，当前没有匹配成功且可执行的文件。'
ElMessage.warning(queueMessage.value)
return
}
- queueMessage.value = `批次 ${parsed.taskNo || parsed.taskId} 已创建，开始串行处理 ${pendingFileIds.value.length} 个文件。`
+ const shouldWait = queueWorkerRunning.value || activeBatchHasWork.value || queuedBatches.value.length > 0
+ if (shouldWait) {
+ queuedBatches.value.push(batch)
+ saveQueueState()
+ queueMessage.value = `批次 ${parsed.taskNo || parsed.taskId} 已加入等待队列，前面还有 ${queuedBatches.value.length} 个批次`
+ ElMessage.success(queueMessage.value)
+ if (!queueWorkerRunning.value) void processQueue()
+ return
+ }
+
+ setActiveBatch(batch)
+ saveQueueState()
+ queueMessage.value = `批次 ${parsed.taskNo || parsed.taskId} 已创建，开始串行处理 ${batch.pendingFileIds.length} 个文件。`
ElMessage.success(queueMessage.value)
void processQueue()
} catch (error) {
@@ -560,37 +583,85 @@ async function submitRun() {
function saveQueueState() {
if (typeof window === 'undefined') return
- if (!currentTaskId.value || !dispatchOptions.value || !hasQueueWork.value) {
+ const active: PublishQueueBatch | null = currentTaskId.value && dispatchOptions.value && activeBatchHasWork.value
+ ? {
+ taskId: currentTaskId.value,
+ pendingFileIds: pendingFileIds.value,
+ activeFileId: activeFileId.value,
+ files: currentFiles.value,
+ options: dispatchOptions.value,
+ }
+ : null
+ const pending = queuedBatches.value.filter(
+ (batch) => batch.pendingFileIds.length > 0 || batch.activeFileId != null,
+ )
+ if (!active && !pending.length) {
window.localStorage.removeItem(queueStorageKey())
return
}
- const state: StoredPublishQueue = {
- taskId: currentTaskId.value,
- pendingFileIds: pendingFileIds.value,
- activeFileId: activeFileId.value,
- files: currentFiles.value,
- options: dispatchOptions.value,
- }
+ const state: StoredPublishQueueState = { active, pending }
window.localStorage.setItem(queueStorageKey(), JSON.stringify(state))
}
+function setActiveBatch(batch: PublishQueueBatch, restoreForm = false) {
+ currentTaskId.value = batch.taskId
+ currentFiles.value = batch.files
+ pendingFileIds.value = [...batch.pendingFileIds]
+ activeFileId.value = batch.activeFileId
+ dispatchOptions.value = batch.options
+ if (restoreForm) {
+ publishCountry.value = batch.options.publishCountry
+ syncCountries.value = [...batch.options.syncCountries]
+ ziniaoVersion.value = batch.options.ziniaoVersion
+ }
+}
+
+function takeNextBatch() {
+ while (queuedBatches.value.length) {
+ const next = queuedBatches.value.shift()
+ if (!next) break
+ if (
+ missingTaskIds.value.includes(next.taskId)
+ || isTerminalStatus(taskSnapshots.value[next.taskId]?.task?.status)
+ ) {
+ continue
+ }
+ setActiveBatch(next)
+ saveQueueState()
+ return true
+ }
+ saveQueueState()
+ return false
+}
+
function loadQueueState() {
if (typeof window === 'undefined') return
try {
const raw = window.localStorage.getItem(queueStorageKey())
if (!raw) return
- const state = JSON.parse(raw) as StoredPublishQueue
- if (!Number.isFinite(state.taskId) || state.taskId <= 0 || !state.options) return
- currentTaskId.value = state.taskId
- pendingFileIds.value = (state.pendingFileIds || []).filter((id) => Number.isFinite(id) && id > 0)
- activeFileId.value = Number.isFinite(state.activeFileId) && Number(state.activeFileId) > 0
- ? Number(state.activeFileId)
- : null
- currentFiles.value = Array.isArray(state.files) ? state.files : []
- dispatchOptions.value = state.options
- publishCountry.value = state.options.publishCountry
- syncCountries.value = [...state.options.syncCountries]
- ziniaoVersion.value = state.options.ziniaoVersion
+ const state = JSON.parse(raw) as Partial & Partial
+ const active = state.active && Number.isFinite(state.active.taskId) && state.active.taskId > 0
+ ? state.active
+ : !state.active && Number.isFinite(state.taskId) && Number(state.taskId) > 0 && state.options
+ ? {
+ taskId: Number(state.taskId),
+ pendingFileIds: state.pendingFileIds || [],
+ activeFileId: state.activeFileId ?? null,
+ files: state.files || [],
+ options: state.options,
+ }
+ : null
+ if (active) setActiveBatch(active, true)
+ queuedBatches.value = (Array.isArray(state.pending) ? state.pending : [])
+ .filter((batch) => Number.isFinite(batch.taskId) && batch.taskId > 0 && batch.options)
+ .map((batch) => ({
+ ...batch,
+ pendingFileIds: (batch.pendingFileIds || []).filter((id) => Number.isFinite(id) && id > 0),
+ activeFileId: Number.isFinite(batch.activeFileId) && Number(batch.activeFileId) > 0
+ ? Number(batch.activeFileId)
+ : null,
+ files: Array.isArray(batch.files) ? batch.files : [],
+ }))
} catch {
window.localStorage.removeItem(queueStorageKey())
}
@@ -711,7 +782,7 @@ async function waitForFileTerminal(taskId: number, fileId: number) {
}
async function processQueue() {
- if (disposed || queueWorkerRunning.value || !currentTaskId.value || !dispatchOptions.value) return
+ if (disposed || queueWorkerRunning.value || (!activeBatchHasWork.value && !queuedBatches.value.length)) return
const api = getPywebviewApi()
if (!api?.enqueue_json) {
queueMessage.value = '当前客户端未提供 enqueue_json，无法派发上架任务。'
@@ -719,76 +790,88 @@ async function processQueue() {
return
}
- const taskId = currentTaskId.value
queueWorkerRunning.value = true
- progressLoop.add(taskId)
try {
- await activatePublishTask(taskId)
- while (!disposed && (activeFileId.value != null || pendingFileIds.value.length > 0)) {
- if (
- missingTaskIds.value.includes(taskId)
- || isTerminalStatus(taskSnapshots.value[taskId]?.task?.status)
- ) {
- pendingFileIds.value = []
- activeFileId.value = null
- saveQueueState()
- break
- }
- if (activeFileId.value != null) {
- const restoredFileId = activeFileId.value
- queueMessage.value = `正在等待文件 ${getCurrentFile(restoredFileId)?.sourceFilename || restoredFileId} 完成...`
- const restoredStatus = await waitForFileTerminal(taskId, restoredFileId)
- if (restoredStatus === 'STOPPED') return
- activeFileId.value = null
- saveQueueState()
- continue
+ while (!disposed) {
+ if (!currentTaskId.value || !dispatchOptions.value || !activeBatchHasWork.value) {
+ if (!takeNextBatch()) break
}
- const nextFileId = pendingFileIds.value.shift()
- if (!nextFileId) break
- const file = getCurrentFile(nextFileId)
- if (!file || isTerminalStatus(file.status)) {
- saveQueueState()
- continue
- }
+ const taskId = currentTaskId.value
+ if (!taskId || !dispatchOptions.value) continue
+ progressLoop.add(taskId)
+ await activatePublishTask(taskId)
- activeFileId.value = nextFileId
- saveQueueState()
- try {
- await activatePublishFile(taskId, nextFileId)
- updateCurrentFile(nextFileId, { status: 'RUNNING', progressMessage: '已派发到 Python 队列' })
- const payload = buildQueuePayload(taskId, file)
- const result = await api.enqueue_json(payload)
- if (!result?.success) throw new Error(result?.error || 'Python 队列拒绝接收任务')
- queueMessage.value = pendingFileIds.value.length
- ? `文件 ${file.sourceFilename || nextFileId} 已入队，完成后继续剩余 ${pendingFileIds.value.length} 个文件。`
- : `文件 ${file.sourceFilename || nextFileId} 已入队，等待执行完成。`
- const finalStatus = await waitForFileTerminal(taskId, nextFileId)
- if (finalStatus === 'STOPPED') return
- queueMessage.value = `文件 ${file.sourceFilename || nextFileId} ${['SUCCESS', 'COMPLETED'].includes(finalStatus) ? '已完成' : '执行失败'}，继续下一个文件。`
- activeFileId.value = null
- saveQueueState()
- } catch (error) {
- const reason = error instanceof Error ? error.message : '文件派发失败'
- try {
- await submitDispatchFailure(taskId, file, reason)
- } catch (compensationError) {
- const message = compensationError instanceof Error ? compensationError.message : '失败状态提交失败'
- queueMessage.value = `文件 ${file.sourceFilename || nextFileId} 派发失败，后端未确认失败状态：${message}。串行队列已暂停。`
+ while (!disposed && (activeFileId.value != null || pendingFileIds.value.length > 0)) {
+ if (
+ missingTaskIds.value.includes(taskId)
+ || isTerminalStatus(taskSnapshots.value[taskId]?.task?.status)
+ ) {
+ pendingFileIds.value = []
+ activeFileId.value = null
saveQueueState()
- throw new Error(queueMessage.value)
+ break
+ }
+ if (activeFileId.value != null) {
+ const restoredFileId = activeFileId.value
+ queueMessage.value = `正在等待文件 ${getCurrentFile(restoredFileId)?.sourceFilename || restoredFileId} 完成...`
+ const restoredStatus = await waitForFileTerminal(taskId, restoredFileId)
+ if (restoredStatus === 'STOPPED') return
+ activeFileId.value = null
+ saveQueueState()
+ continue
}
- queueMessage.value = `文件 ${file.sourceFilename || nextFileId} 派发失败，已记录并继续下一个文件。`
- activeFileId.value = null
- saveQueueState()
- }
- }
- if (!disposed) {
- queueMessage.value = '本批次文件已全部按顺序派发完成，正在生成结果文件。'
+ const nextFileId = pendingFileIds.value.shift()
+ if (!nextFileId) break
+ const file = getCurrentFile(nextFileId)
+ if (!file || isTerminalStatus(file.status)) {
+ saveQueueState()
+ continue
+ }
+
+ activeFileId.value = nextFileId
+ saveQueueState()
+ try {
+ await activatePublishFile(taskId, nextFileId)
+ updateCurrentFile(nextFileId, { status: 'RUNNING', progressMessage: '已派发到 Python 队列' })
+ const payload = buildQueuePayload(taskId, file)
+ const result = await api.enqueue_json(payload)
+ if (!result?.success) throw new Error(result?.error || 'Python 队列拒绝接收任务')
+ queueMessage.value = pendingFileIds.value.length
+ ? `文件 ${file.sourceFilename || nextFileId} 已入队，完成后继续剩余 ${pendingFileIds.value.length} 个文件。`
+ : `文件 ${file.sourceFilename || nextFileId} 已入队，等待执行完成。`
+ const finalStatus = await waitForFileTerminal(taskId, nextFileId)
+ if (finalStatus === 'STOPPED') return
+ queueMessage.value = `文件 ${file.sourceFilename || nextFileId} ${['SUCCESS', 'COMPLETED'].includes(finalStatus) ? '已完成' : '执行失败'}，继续下一个文件。`
+ activeFileId.value = null
+ saveQueueState()
+ } catch (error) {
+ const reason = error instanceof Error ? error.message : '文件派发失败'
+ try {
+ await submitDispatchFailure(taskId, file, reason)
+ } catch (compensationError) {
+ const message = compensationError instanceof Error ? compensationError.message : '失败状态提交失败'
+ queueMessage.value = `文件 ${file.sourceFilename || nextFileId} 派发失败，后端未确认失败状态：${message}。串行队列已暂停。`
+ saveQueueState()
+ throw new Error(queueMessage.value)
+ }
+ queueMessage.value = `文件 ${file.sourceFilename || nextFileId} 派发失败，已记录并继续下一个文件。`
+ activeFileId.value = null
+ saveQueueState()
+ }
+ }
+
+ if (disposed) return
await progressLoop.refreshOnce()
await Promise.all([loadDashboard(), loadHistory()])
- ElMessage.success('上架批次已完成串行派发')
+ if (queuedBatches.value.length) {
+ queueMessage.value = `当前批次已完成派发，继续处理后续 ${queuedBatches.value.length} 个等待批次。`
+ continue
+ }
+ queueMessage.value = '所有上架批次已按顺序派发完成，正在生成结果文件。'
+ ElMessage.success('上架等待队列已完成派发')
+ break
}
} catch (error) {
if (disposed) return
@@ -825,34 +908,57 @@ async function loadHistory() {
}
function reconcileStoredQueue() {
+ function reconcileBatch(batch: PublishQueueBatch) {
+ if (missingTaskIds.value.includes(batch.taskId)) return null
+ const snapshot = taskSnapshots.value[batch.taskId]
+ if (isTerminalStatus(snapshot?.task?.status)) return null
+ const files = mergeFiles(batch.files || [], snapshot?.files || [])
+ let pendingIds = (batch.pendingFileIds || []).filter((fileId) => {
+ const file = files.find((item) => item.fileId === fileId)
+ return file && !isTerminalStatus(file.status)
+ })
+ let activeId = batch.activeFileId
+ if (activeId != null) {
+ const activeStatus = normalizeStatus(files.find((file) => file.fileId === activeId)?.status)
+ if (!activeStatus || activeStatus === 'PENDING') {
+ pendingIds = [activeId, ...pendingIds.filter((fileId) => fileId !== activeId)]
+ activeId = null
+ } else if (isTerminalStatus(activeStatus)) {
+ activeId = null
+ }
+ }
+ if (!pendingIds.length && activeId == null) return null
+ return { ...batch, pendingFileIds: pendingIds, activeFileId: activeId, files }
+ }
+
const taskId = currentTaskId.value
- if (!taskId) return
- if (missingTaskIds.value.includes(taskId)) {
+ const active = taskId && dispatchOptions.value
+ ? reconcileBatch({
+ taskId,
+ pendingFileIds: pendingFileIds.value,
+ activeFileId: activeFileId.value,
+ files: currentFiles.value,
+ options: dispatchOptions.value,
+ })
+ : null
+ if (active) {
+ setActiveBatch(active)
+ } else {
+ currentTaskId.value = null
+ currentFiles.value = []
pendingFileIds.value = []
activeFileId.value = null
- saveQueueState()
- queueMessage.value = `任务 ${taskId} 已不存在，本地队列已清理。`
- return
+ dispatchOptions.value = null
}
- if (isTerminalStatus(taskSnapshots.value[taskId]?.task?.status)) {
- pendingFileIds.value = []
- activeFileId.value = null
- saveQueueState()
- return
- }
- pendingFileIds.value = pendingFileIds.value.filter((fileId) => {
- const file = getCurrentFile(fileId)
- return file && !isTerminalStatus(file.status)
- })
- if (activeFileId.value != null) {
- const activeId = activeFileId.value
- const activeStatus = normalizeStatus(getCurrentFile(activeId)?.status)
+ queuedBatches.value = queuedBatches.value
+ .map(reconcileBatch)
+ .filter((batch): batch is PublishQueueBatch => batch != null)
+
+ if (active?.activeFileId != null) {
+ const activeId = active.activeFileId
+ const activeStatus = normalizeStatus(active.files.find((file) => file.fileId === activeId)?.status)
if (!activeStatus || activeStatus === 'PENDING') {
- pendingFileIds.value = [activeId, ...pendingFileIds.value.filter((fileId) => fileId !== activeId)]
- activeFileId.value = null
queueMessage.value = `文件 ${activeId} 尚未激活，已恢复到待派发队列。`
- } else if (isTerminalStatus(activeStatus)) {
- activeFileId.value = null
}
}
saveQueueState()
@@ -935,13 +1041,14 @@ async function downloadResult(detail: PublishTaskDetailVo) {
onMounted(async () => {
loadQueueState()
if (currentTaskId.value) progressLoop.add(currentTaskId.value)
+ for (const batch of queuedBatches.value) progressLoop.add(batch.taskId)
await Promise.all([loadDashboard(), loadHistory()])
- if (currentTaskId.value) await progressLoop.refreshOnce()
+ if (currentTaskId.value || queuedBatches.value.length) await progressLoop.refreshOnce()
reconcileStoredQueue()
if (hasQueueWork.value) {
queueMessage.value = activeFileId.value != null
? `检测到未完成文件 ${activeFileId.value}，继续等待完成后衔接后续文件。`
- : `检测到未完成队列，继续处理剩余 ${pendingFileIds.value.length} 个文件。`
+ : `检测到未完成队列，继续处理当前文件及后续 ${queuedBatches.value.length} 个批次。`
void processQueue()
}
})