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

完成上架相关优化、oss迁移

Browse Source
This commit is contained in:
supernijia
2026-07-25 17:57:34 +08:00
parent 8ab36b1aaf
commit 4ddb8b47b0
28 changed files with 964 additions and 382 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/config/OssProperties.java
+1
View File
@@ -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;
}
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentHistoryItemVo.java
+1 -1
View File
@@ -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;
backend-java/src/main/java/com/nanri/aiimage/modules/brand/model/dto/BrandSourceFileDto.java
+1 -1
View File
@@ -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")
backend-java/src/main/java/com/nanri/aiimage/modules/digitalhuman/service/DigitalHumanVersionService.java
+14 -32
View File
@@ -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<DigitalHumanVersionEntity>()
@@ -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()
);
}
}
backend-java/src/main/java/com/nanri/aiimage/modules/file/service/oss/OssStorageService.java
+274 -145
View File
@@ -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<String> 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) {
}
}
backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/mapper/ImageVideoAsyncTaskMapper.java
+18 -6
View File
@@ -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<ImageVideoAsyncTaskEntity> {
@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);
}
backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/model/entity/ImageVideoAsyncTaskEntity.java
+1
View File
@@ -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;
backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoAsyncTaskService.java
+58 -7
View File
@@ -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<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.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<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.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
}
backend-java/src/main/java/com/nanri/aiimage/modules/publish/controller/PublishController.java
+1 -1
View File
@@ -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<PublishParseVo> parse(@Valid @RequestBody PublishParseRequest request) {
return ApiResponse.success(publishTaskService.parseAndCreateTask(request));
}
backend-java/src/main/java/com/nanri/aiimage/modules/publish/model/vo/PublishItemsPageVo.java
+1 -1
View File
@@ -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<PublishRowDto> items = new ArrayList<>();
}
backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java
+53 -3
View File
@@ -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<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<PreparedFile> 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<PublishTaskDetailVo> loadTaskDetails(List<FileTaskEntity> 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 不合法");
backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishWorkbookService.java
+9 -15
View File
@@ -30,7 +30,9 @@ import java.util.zip.ZipOutputStream;
@Service
public class PublishWorkbookService {
public static final List<String> HEADERS = List.of(
public static final List<String> SOURCE_HEADERS = List.of(
"id", "ASIN", "国家", "品牌", "价格");
public static final List<String> 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<String> values = new ArrayList<>(HEADERS.size());
List<String> 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<Integer, String> 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<String> 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<PublishRowDto> 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;
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/vo/SimilarAsinHistoryItemVo.java
+1 -1
View File
@@ -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;
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
+9 -3
View File
@@ -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());
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java
+2 -2
View File
@@ -95,7 +95,7 @@ public class TaskFileJobService {
List<TaskFileJobEntity> ownerJobs = taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
.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()
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java
+3
View File
@@ -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={}",
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java
+6 -4
View File
@@ -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) {