This commit is contained in:
+56
-7
@@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.brand.client;
|
|||||||
|
|
||||||
import com.fasterxml.jackson.annotation.JsonAlias;
|
import com.fasterxml.jackson.annotation.JsonAlias;
|
||||||
import com.nanri.aiimage.config.BrandCheckProperties;
|
import com.nanri.aiimage.config.BrandCheckProperties;
|
||||||
|
import jakarta.annotation.PreDestroy;
|
||||||
import lombok.Data;
|
import lombok.Data;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
@@ -16,6 +17,12 @@ import java.util.ArrayList;
|
|||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.CompletableFuture;
|
||||||
|
import java.util.concurrent.CompletionException;
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
import java.util.concurrent.ThreadFactory;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
@Component
|
@Component
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
@@ -24,9 +31,29 @@ public class BrandCheckClient {
|
|||||||
|
|
||||||
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
|
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
|
||||||
|
|
||||||
|
/** 品牌检查并发度:避免一次批量检查把第三方服务打满。 */
|
||||||
|
private static final int BRAND_CHECK_CONCURRENCY = 3;
|
||||||
|
|
||||||
private final BrandCheckProperties properties;
|
private final BrandCheckProperties properties;
|
||||||
private volatile RestClient sharedRestClient;
|
private volatile RestClient sharedRestClient;
|
||||||
|
|
||||||
|
private final ExecutorService checkExecutor = Executors.newFixedThreadPool(
|
||||||
|
BRAND_CHECK_CONCURRENCY, namedThreadFactory("brand-check"));
|
||||||
|
|
||||||
|
@PreDestroy
|
||||||
|
void shutdownCheckExecutor() {
|
||||||
|
checkExecutor.shutdownNow();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static ThreadFactory namedThreadFactory(String prefix) {
|
||||||
|
AtomicInteger counter = new AtomicInteger();
|
||||||
|
return r -> {
|
||||||
|
Thread t = new Thread(r, prefix + "-" + counter.incrementAndGet());
|
||||||
|
t.setDaemon(true);
|
||||||
|
return t;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
public BrandCheckResponse check(String brand) {
|
public BrandCheckResponse check(String brand) {
|
||||||
return check(brand, properties.getDefaultStrategy());
|
return check(brand, properties.getDefaultStrategy());
|
||||||
}
|
}
|
||||||
@@ -55,23 +82,45 @@ public class BrandCheckClient {
|
|||||||
|
|
||||||
public BrandCheckBatchResult checkAll(List<String> brands, String strategy) {
|
public BrandCheckBatchResult checkAll(List<String> brands, String strategy) {
|
||||||
List<String> distinctBrands = distinctNonBlank(brands);
|
List<String> distinctBrands = distinctNonBlank(brands);
|
||||||
|
List<CompletableFuture<BrandCheckOutcome>> futures = new ArrayList<>(distinctBrands.size());
|
||||||
|
for (String brand : distinctBrands) {
|
||||||
|
futures.add(CompletableFuture.supplyAsync(
|
||||||
|
() -> checkOneBrand(brand, strategy), checkExecutor));
|
||||||
|
}
|
||||||
List<Object> failedData = new ArrayList<>();
|
List<Object> failedData = new ArrayList<>();
|
||||||
List<Object> queryFailedData = new ArrayList<>();
|
List<Object> queryFailedData = new ArrayList<>();
|
||||||
for (String brand : distinctBrands) {
|
for (CompletableFuture<BrandCheckOutcome> future : futures) {
|
||||||
|
BrandCheckOutcome outcome = awaitOutcome(future);
|
||||||
|
failedData.addAll(outcome.failedData());
|
||||||
|
queryFailedData.addAll(outcome.queryFailedData());
|
||||||
|
}
|
||||||
|
return new BrandCheckBatchResult(distinctBrands, failedData, queryFailedData);
|
||||||
|
}
|
||||||
|
|
||||||
|
private BrandCheckOutcome checkOneBrand(String brand, String strategy) {
|
||||||
try {
|
try {
|
||||||
BrandCheckResponse response = check(brand, strategy);
|
BrandCheckResponse response = check(brand, strategy);
|
||||||
if (response == null) {
|
if (response == null) {
|
||||||
queryFailedData.add(brand);
|
return new BrandCheckOutcome(List.of(), List.of(brand));
|
||||||
continue;
|
|
||||||
}
|
}
|
||||||
failedData.addAll(nullToEmpty(response.getFaildData()));
|
return new BrandCheckOutcome(nullToEmpty(response.getFaildData()), nullToEmpty(response.getQueryFaildData()));
|
||||||
queryFailedData.addAll(nullToEmpty(response.getQueryFaildData()));
|
|
||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
log.warn("[brand-check] request failed brand={} strategy={} err={}", brand, strategy, ex.getMessage());
|
log.warn("[brand-check] request failed brand={} strategy={} err={}", brand, strategy, ex.getMessage());
|
||||||
queryFailedData.add(brand);
|
return new BrandCheckOutcome(List.of(), List.of(brand));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return new BrandCheckBatchResult(distinctBrands, failedData, queryFailedData);
|
|
||||||
|
private BrandCheckOutcome awaitOutcome(CompletableFuture<BrandCheckOutcome> future) {
|
||||||
|
try {
|
||||||
|
return future.join();
|
||||||
|
} catch (CompletionException ex) {
|
||||||
|
log.warn("[brand-check] unexpected batch failure err={}",
|
||||||
|
ex.getCause() == null ? ex.getMessage() : String.valueOf(ex.getCause().getMessage()));
|
||||||
|
return new BrandCheckOutcome(List.of(), List.of());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private record BrandCheckOutcome(List<Object> failedData, List<Object> queryFailedData) {
|
||||||
}
|
}
|
||||||
|
|
||||||
public BrandCheckBatchResult checkTitleText(String titleText) {
|
public BrandCheckBatchResult checkTitleText(String titleText) {
|
||||||
|
|||||||
+44
-1
@@ -35,6 +35,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
|
|||||||
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||||
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||||
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||||
|
import jakarta.annotation.PreDestroy;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.apache.poi.ss.usermodel.Cell;
|
import org.apache.poi.ss.usermodel.Cell;
|
||||||
@@ -65,6 +66,12 @@ import java.util.List;
|
|||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Objects;
|
import java.util.Objects;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.CompletableFuture;
|
||||||
|
import java.util.concurrent.CompletionException;
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
import java.util.concurrent.ThreadFactory;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
import java.util.zip.ZipEntry;
|
import java.util.zip.ZipEntry;
|
||||||
import java.util.zip.ZipOutputStream;
|
import java.util.zip.ZipOutputStream;
|
||||||
|
|
||||||
@@ -82,6 +89,25 @@ public class BrandTaskService {
|
|||||||
private static final String STATUS_FAILED = "failed";
|
private static final String STATUS_FAILED = "failed";
|
||||||
private static final String STATUS_CANCELLED = "cancelled";
|
private static final String STATUS_CANCELLED = "cancelled";
|
||||||
private static final String MODULE_TYPE = "BRAND";
|
private static final String MODULE_TYPE = "BRAND";
|
||||||
|
/** 结果文件并发上传数:OSS/MinIO 上传互不依赖,3 并发平衡收益与内存占用。 */
|
||||||
|
private static final int RESULT_UPLOAD_CONCURRENCY = 3;
|
||||||
|
|
||||||
|
private final ExecutorService resultUploadExecutor = Executors.newFixedThreadPool(
|
||||||
|
RESULT_UPLOAD_CONCURRENCY, namedThreadFactory("brand-result-upload"));
|
||||||
|
|
||||||
|
@PreDestroy
|
||||||
|
void shutdownResultUploadExecutor() {
|
||||||
|
resultUploadExecutor.shutdownNow();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static ThreadFactory namedThreadFactory(String prefix) {
|
||||||
|
AtomicInteger counter = new AtomicInteger();
|
||||||
|
return r -> {
|
||||||
|
Thread t = new Thread(r, prefix + "-" + counter.incrementAndGet());
|
||||||
|
t.setDaemon(true);
|
||||||
|
return t;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
private final BrandCrawlTaskMapper brandCrawlTaskMapper;
|
private final BrandCrawlTaskMapper brandCrawlTaskMapper;
|
||||||
private final OssStorageService ossStorageService;
|
private final OssStorageService ossStorageService;
|
||||||
@@ -1036,10 +1062,27 @@ public class BrandTaskService {
|
|||||||
throw new BusinessException("没有可上传的结果文件");
|
throw new BusinessException("没有可上传的结果文件");
|
||||||
}
|
}
|
||||||
List<String> fullUrls = new ArrayList<>();
|
List<String> fullUrls = new ArrayList<>();
|
||||||
|
List<CompletableFuture<String>> uploadFutures = new ArrayList<>(entries.size());
|
||||||
for (OutputEntry entry : entries) {
|
for (OutputEntry entry : entries) {
|
||||||
|
uploadFutures.add(CompletableFuture.supplyAsync(() -> {
|
||||||
// 存储完整公开地址,下载时会自动通过 resolveObjectKey 提取 objectKey(见 OssStorageService.resolveObjectKey)
|
// 存储完整公开地址,下载时会自动通过 resolveObjectKey 提取 objectKey(见 OssStorageService.resolveObjectKey)
|
||||||
String objectKey = ossStorageService.uploadResultFile(entry.resultFile(), MODULE_TYPE);
|
String objectKey = ossStorageService.uploadResultFile(entry.resultFile(), MODULE_TYPE);
|
||||||
fullUrls.add(ossStorageService.getPublicUrl(objectKey));
|
return ossStorageService.getPublicUrl(objectKey);
|
||||||
|
}, resultUploadExecutor));
|
||||||
|
}
|
||||||
|
for (CompletableFuture<String> future : uploadFutures) {
|
||||||
|
try {
|
||||||
|
fullUrls.add(future.join());
|
||||||
|
} catch (CompletionException ex) {
|
||||||
|
Throwable cause = ex.getCause() == null ? ex : ex.getCause();
|
||||||
|
if (cause instanceof IOException ioe) {
|
||||||
|
throw ioe;
|
||||||
|
}
|
||||||
|
if (cause instanceof RuntimeException re) {
|
||||||
|
throw re;
|
||||||
|
}
|
||||||
|
throw new RuntimeException(cause);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Map<String, Object> result = new LinkedHashMap<>();
|
Map<String, Object> result = new LinkedHashMap<>();
|
||||||
result.put("urls", fullUrls);
|
result.put("urls", fullUrls);
|
||||||
|
|||||||
+12
-4
@@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.productcategory.service;
|
|||||||
|
|
||||||
import cn.hutool.crypto.digest.DigestUtil;
|
import cn.hutool.crypto.digest.DigestUtil;
|
||||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||||
|
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
||||||
import com.nanri.aiimage.common.exception.BusinessException;
|
import com.nanri.aiimage.common.exception.BusinessException;
|
||||||
import com.nanri.aiimage.modules.productcategory.mapper.ProductCategoryMapper;
|
import com.nanri.aiimage.modules.productcategory.mapper.ProductCategoryMapper;
|
||||||
import com.nanri.aiimage.modules.productcategory.model.dto.ProductCategorySaveRequest;
|
import com.nanri.aiimage.modules.productcategory.model.dto.ProductCategorySaveRequest;
|
||||||
@@ -309,10 +310,17 @@ public class ProductCategoryService {
|
|||||||
if (rows.isEmpty()) {
|
if (rows.isEmpty()) {
|
||||||
return childCountById;
|
return childCountById;
|
||||||
}
|
}
|
||||||
for (ProductCategoryEntity row : rows) {
|
List<Map<String, Object>> countRows = productCategoryMapper.selectMaps(new QueryWrapper<ProductCategoryEntity>()
|
||||||
Long childCount = productCategoryMapper.selectCount(new LambdaQueryWrapper<ProductCategoryEntity>()
|
.select("parent_id", "count(*) AS child_count")
|
||||||
.eq(ProductCategoryEntity::getParentId, row.getId()));
|
.isNotNull("parent_id")
|
||||||
childCountById.put(row.getId(), childCount == null ? 0 : childCount.intValue());
|
.groupBy("parent_id"));
|
||||||
|
for (Map<String, Object> countRow : countRows) {
|
||||||
|
Object parentId = countRow.get("parent_id");
|
||||||
|
Object childCount = countRow.get("child_count");
|
||||||
|
if (parentId == null || childCount == null) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
childCountById.put(((Number) parentId).longValue(), ((Number) childCount).intValue());
|
||||||
}
|
}
|
||||||
return childCountById;
|
return childCountById;
|
||||||
}
|
}
|
||||||
|
|||||||
+16
-6
@@ -69,6 +69,7 @@ import java.util.Objects;
|
|||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
import java.util.stream.Collectors;
|
import java.util.stream.Collectors;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
@Slf4j
|
@Slf4j
|
||||||
@Service
|
@Service
|
||||||
@@ -449,15 +450,24 @@ public class PublishTaskService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
TaskOptions options = readTaskOptions(task);
|
TaskOptions options = readTaskOptions(task);
|
||||||
|
List<Long> successfulFileIds = successfulFiles.stream()
|
||||||
|
.map(PublishFileEntity::getId)
|
||||||
|
.toList();
|
||||||
|
Map<Long, List<PublishItemEntity>> itemsByFileId = successfulFileIds.isEmpty()
|
||||||
|
? Map.of()
|
||||||
|
: publishItemMapper.selectList(
|
||||||
|
new LambdaQueryWrapper<PublishItemEntity>()
|
||||||
|
.eq(PublishItemEntity::getTaskId, task.getId())
|
||||||
|
.in(PublishItemEntity::getFileId, successfulFileIds)
|
||||||
|
.orderByAsc(PublishItemEntity::getRowIndex))
|
||||||
|
.stream()
|
||||||
|
.collect(Collectors.groupingBy(PublishItemEntity::getFileId,
|
||||||
|
LinkedHashMap::new, Collectors.toList()));
|
||||||
List<PublishWorkbookService.WorkbookInput> inputs = new ArrayList<>();
|
List<PublishWorkbookService.WorkbookInput> inputs = new ArrayList<>();
|
||||||
int rowCount = 0;
|
int rowCount = 0;
|
||||||
for (PublishFileEntity file : successfulFiles) {
|
for (PublishFileEntity file : successfulFiles) {
|
||||||
List<PublishItemEntity> items = publishItemMapper.selectList(
|
List<PublishRowDto> rows = itemsByFileId.getOrDefault(file.getId(), List.of())
|
||||||
new LambdaQueryWrapper<PublishItemEntity>()
|
.stream().map(this::toRowDto).toList();
|
||||||
.eq(PublishItemEntity::getTaskId, task.getId())
|
|
||||||
.eq(PublishItemEntity::getFileId, file.getId())
|
|
||||||
.orderByAsc(PublishItemEntity::getRowIndex));
|
|
||||||
List<PublishRowDto> rows = items.stream().map(this::toRowDto).toList();
|
|
||||||
rowCount += rows.size();
|
rowCount += rows.size();
|
||||||
inputs.add(new PublishWorkbookService.WorkbookInput(
|
inputs.add(new PublishWorkbookService.WorkbookInput(
|
||||||
file.getSourceFilename(), file.getShopName(), options.publishCountry(), rows));
|
file.getSourceFilename(), file.getShopName(), options.publishCountry(), rows));
|
||||||
|
|||||||
+7
-3
@@ -136,9 +136,12 @@ public class QueryAsinResolveService {
|
|||||||
if (ordered.isEmpty()) {
|
if (ordered.isEmpty()) {
|
||||||
throw new BusinessException("shop_names 无有效店铺名");
|
throw new BusinessException("shop_names 无有效店铺名");
|
||||||
}
|
}
|
||||||
|
List<QueryAsinEntity> allAsins = queryAsinMapper.selectList(
|
||||||
|
new LambdaQueryWrapper<QueryAsinEntity>().orderByAsc(QueryAsinEntity::getId));
|
||||||
|
List<QueryAsinCountryAsinsDto> preloadedQueryAsins = toCountryAsins(allAsins);
|
||||||
ProductRiskMatchShopsVo vo = new ProductRiskMatchShopsVo();
|
ProductRiskMatchShopsVo vo = new ProductRiskMatchShopsVo();
|
||||||
for (String shopName : ordered) {
|
for (String shopName : ordered) {
|
||||||
vo.getItems().add(matchOneShop(request.getUserId(), shopName));
|
vo.getItems().add(matchOneShop(shopName, preloadedQueryAsins));
|
||||||
}
|
}
|
||||||
return vo;
|
return vo;
|
||||||
}
|
}
|
||||||
@@ -161,7 +164,8 @@ public class QueryAsinResolveService {
|
|||||||
return count == null ? 0L : count;
|
return count == null ? 0L : count;
|
||||||
}
|
}
|
||||||
|
|
||||||
private ProductRiskShopQueueItemVo matchOneShop(Long userId, String shopName) {
|
private ProductRiskShopQueueItemVo matchOneShop(String shopName,
|
||||||
|
List<QueryAsinCountryAsinsDto> preloadedQueryAsins) {
|
||||||
ProductRiskShopQueueItemVo item = new ProductRiskShopQueueItemVo();
|
ProductRiskShopQueueItemVo item = new ProductRiskShopQueueItemVo();
|
||||||
item.setShopName(shopName);
|
item.setShopName(shopName);
|
||||||
try {
|
try {
|
||||||
@@ -180,7 +184,7 @@ public class QueryAsinResolveService {
|
|||||||
item.setMatchStatus(matched.getMatchStatus());
|
item.setMatchStatus(matched.getMatchStatus());
|
||||||
item.setMatchMessage(matched.getMatchMessage());
|
item.setMatchMessage(matched.getMatchMessage());
|
||||||
item.setOpenStoreUrl(matched.getOpenStoreUrl());
|
item.setOpenStoreUrl(matched.getOpenStoreUrl());
|
||||||
item.setQueryAsins(loadQueryAsinsForShop(userId, shopName));
|
item.setQueryAsins(preloadedQueryAsins);
|
||||||
if (item.isMatched() && item.getQueryAsins().isEmpty()) {
|
if (item.isMatched() && item.getQueryAsins().isEmpty()) {
|
||||||
item.setMatchMessage("紫鸟已匹配,但后台查询 ASIN 表暂无数据");
|
item.setMatchMessage("紫鸟已匹配,但后台查询 ASIN 表暂无数据");
|
||||||
}
|
}
|
||||||
|
|||||||
+24
-1
@@ -60,6 +60,7 @@ import java.time.Duration;
|
|||||||
import java.time.LocalDate;
|
import java.time.LocalDate;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Comparator;
|
import java.util.Comparator;
|
||||||
|
import java.util.HashMap;
|
||||||
import java.util.HashSet;
|
import java.util.HashSet;
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
@@ -80,6 +81,8 @@ public class ShopDataCrawlTaskService {
|
|||||||
private static final int RESULT_PENDING = -1;
|
private static final int RESULT_PENDING = -1;
|
||||||
private static final int RESULT_FAILED = 0;
|
private static final int RESULT_FAILED = 0;
|
||||||
private static final int RESULT_SUCCESS = 1;
|
private static final int RESULT_SUCCESS = 1;
|
||||||
|
/** 批量 IN 查询单批上限。 */
|
||||||
|
private static final int ID_BATCH_SIZE = 500;
|
||||||
private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断";
|
private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断";
|
||||||
private static final String PARTIAL_RESULT_MESSAGE = "Python 中断,已保留已回传的部分数据";
|
private static final String PARTIAL_RESULT_MESSAGE = "Python 中断,已保留已回传的部分数据";
|
||||||
private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:";
|
private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:";
|
||||||
@@ -800,12 +803,24 @@ public class ShopDataCrawlTaskService {
|
|||||||
Comparator.nullsLast(Comparator.naturalOrder()))
|
Comparator.nullsLast(Comparator.naturalOrder()))
|
||||||
.thenComparing(ShopDataCrawlDailyMemberEntity::getId,
|
.thenComparing(ShopDataCrawlDailyMemberEntity::getId,
|
||||||
Comparator.nullsLast(Comparator.naturalOrder())));
|
Comparator.nullsLast(Comparator.naturalOrder())));
|
||||||
|
Map<Long, FileResultEntity> resultById = new HashMap<>();
|
||||||
|
List<Long> keptResultIds = new ArrayList<>();
|
||||||
|
for (ShopDataCrawlDailyMemberEntity member : members) {
|
||||||
|
if (!removedResultIds.contains(member.getResultId())) {
|
||||||
|
keptResultIds.add(member.getResultId());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (List<Long> batch : toIdBatches(keptResultIds)) {
|
||||||
|
for (FileResultEntity result : fileResultMapper.selectBatchIds(batch)) {
|
||||||
|
resultById.put(result.getId(), result);
|
||||||
|
}
|
||||||
|
}
|
||||||
List<DailyMemberData> survivors = new ArrayList<>();
|
List<DailyMemberData> survivors = new ArrayList<>();
|
||||||
for (ShopDataCrawlDailyMemberEntity member : members) {
|
for (ShopDataCrawlDailyMemberEntity member : members) {
|
||||||
if (removedResultIds.contains(member.getResultId())) {
|
if (removedResultIds.contains(member.getResultId())) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
FileResultEntity result = fileResultMapper.selectById(member.getResultId());
|
FileResultEntity result = resultById.get(member.getResultId());
|
||||||
if (result == null || !Integer.valueOf(RESULT_SUCCESS).equals(result.getSuccess())) {
|
if (result == null || !Integer.valueOf(RESULT_SUCCESS).equals(result.getSuccess())) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
@@ -818,6 +833,14 @@ public class ShopDataCrawlTaskService {
|
|||||||
return survivors;
|
return survivors;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private List<List<Long>> toIdBatches(List<Long> ids) {
|
||||||
|
List<List<Long>> batches = new ArrayList<>();
|
||||||
|
for (int start = 0; start < ids.size(); start += ID_BATCH_SIZE) {
|
||||||
|
batches.add(ids.subList(start, Math.min(start + ID_BATCH_SIZE, ids.size())));
|
||||||
|
}
|
||||||
|
return batches;
|
||||||
|
}
|
||||||
|
|
||||||
private ShopDataCrawlResultItemVo loadSnapshotForDailyMember(FileResultEntity result) {
|
private ShopDataCrawlResultItemVo loadSnapshotForDailyMember(FileResultEntity result) {
|
||||||
ShopDataCrawlResultItemVo snapshot = taskResultItemService.getResultSnapshot(
|
ShopDataCrawlResultItemVo snapshot = taskResultItemService.getResultSnapshot(
|
||||||
result.getTaskId(), MODULE_TYPE, result.getId(), ShopDataCrawlResultItemVo.class);
|
result.getTaskId(), MODULE_TYPE, result.getId(), ShopDataCrawlResultItemVo.class);
|
||||||
|
|||||||
+15
-8
@@ -280,9 +280,22 @@ public class ShopManageGroupService {
|
|||||||
for (Long memberUserId : memberUserIds) {
|
for (Long memberUserId : memberUserIds) {
|
||||||
uniqueIds.add(normalizePositiveId(memberUserId, "组员用户不能为空"));
|
uniqueIds.add(normalizePositiveId(memberUserId, "组员用户不能为空"));
|
||||||
}
|
}
|
||||||
|
List<AdminUserEntity> rawUsers = adminUserMapper.selectBatchIds(uniqueIds);
|
||||||
|
Map<Long, AdminUserEntity> userById = new LinkedHashMap<>();
|
||||||
|
for (AdminUserEntity user : rawUsers) {
|
||||||
|
userById.put(user.getId(), user);
|
||||||
|
}
|
||||||
|
Set<Long> boundUserIds = groupMemberMapper.selectList(
|
||||||
|
new LambdaQueryWrapper<ShopManageGroupMemberEntity>()
|
||||||
|
.in(ShopManageGroupMemberEntity::getUserId, uniqueIds)
|
||||||
|
.ne(currentGroupId != null,
|
||||||
|
ShopManageGroupMemberEntity::getGroupId, currentGroupId))
|
||||||
|
.stream()
|
||||||
|
.map(ShopManageGroupMemberEntity::getUserId)
|
||||||
|
.collect(Collectors.toSet());
|
||||||
List<AdminUserEntity> members = new ArrayList<>();
|
List<AdminUserEntity> members = new ArrayList<>();
|
||||||
for (Long memberUserId : uniqueIds) {
|
for (Long memberUserId : uniqueIds) {
|
||||||
AdminUserEntity user = adminUserMapper.selectById(memberUserId);
|
AdminUserEntity user = userById.get(memberUserId);
|
||||||
if (user == null) {
|
if (user == null) {
|
||||||
throw new BusinessException("组员用户不存在");
|
throw new BusinessException("组员用户不存在");
|
||||||
}
|
}
|
||||||
@@ -295,13 +308,7 @@ public class ShopManageGroupService {
|
|||||||
if (!operatorCanBindAllNormals && (user.getCreatedById() == null || !user.getCreatedById().equals(leaderId))) {
|
if (!operatorCanBindAllNormals && (user.getCreatedById() == null || !user.getCreatedById().equals(leaderId))) {
|
||||||
throw new BusinessException("只能添加组长自己创建的普通员工账号");
|
throw new BusinessException("只能添加组长自己创建的普通员工账号");
|
||||||
}
|
}
|
||||||
LambdaQueryWrapper<ShopManageGroupMemberEntity> query = new LambdaQueryWrapper<ShopManageGroupMemberEntity>()
|
if (boundUserIds.contains(user.getId())) {
|
||||||
.eq(ShopManageGroupMemberEntity::getUserId, user.getId());
|
|
||||||
if (currentGroupId != null) {
|
|
||||||
query.ne(ShopManageGroupMemberEntity::getGroupId, currentGroupId);
|
|
||||||
}
|
|
||||||
Long count = groupMemberMapper.selectCount(query);
|
|
||||||
if (count != null && count > 0) {
|
|
||||||
throw new BusinessException("该普通员工账号已加入其他分组");
|
throw new BusinessException("该普通员工账号已加入其他分组");
|
||||||
}
|
}
|
||||||
members.add(user);
|
members.add(user);
|
||||||
|
|||||||
+30
-6
@@ -17,8 +17,10 @@ import java.security.MessageDigest;
|
|||||||
import java.security.NoSuchAlgorithmException;
|
import java.security.NoSuchAlgorithmException;
|
||||||
import java.time.LocalDateTime;
|
import java.time.LocalDateTime;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
|
import java.util.LinkedHashMap;
|
||||||
import java.util.LinkedHashSet;
|
import java.util.LinkedHashSet;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.ExecutorService;
|
||||||
@@ -50,6 +52,9 @@ public class SimilarAsinImagePrefetchService {
|
|||||||
/** P2-11:预热线程池容量。与图片处理 CPU 槽位对齐,避免预热路径绕过主池放大并发。 */
|
/** P2-11:预热线程池容量。与图片处理 CPU 槽位对齐,避免预热路径绕过主池放大并发。 */
|
||||||
private static final int PREFETCH_POOL_SIZE = 2;
|
private static final int PREFETCH_POOL_SIZE = 2;
|
||||||
|
|
||||||
|
/** 批量命中检查/批量 touch 单次 IN 上限,避免超长 SQL 参数列表。 */
|
||||||
|
private static final int CACHE_LOOKUP_BATCH_SIZE = 500;
|
||||||
|
|
||||||
/** 排队等待上一个 task future 时的最大等待时间(秒),避免被 hung future 永久卡住。 */
|
/** 排队等待上一个 task future 时的最大等待时间(秒),避免被 hung future 永久卡住。 */
|
||||||
private static final long INFLIGHT_WAIT_SECONDS = 60L;
|
private static final long INFLIGHT_WAIT_SECONDS = 60L;
|
||||||
|
|
||||||
@@ -118,21 +123,40 @@ public class SimilarAsinImagePrefetchService {
|
|||||||
int hit = 0;
|
int hit = 0;
|
||||||
int miss = 0;
|
int miss = 0;
|
||||||
int fail = 0;
|
int fail = 0;
|
||||||
|
Map<String, String> hashByUrl = new LinkedHashMap<>();
|
||||||
|
Set<String> cachedHashes = new LinkedHashSet<>();
|
||||||
for (String url : urls) {
|
for (String url : urls) {
|
||||||
try {
|
|
||||||
String urlHash = sha256Hex(url);
|
String urlHash = sha256Hex(url);
|
||||||
if (urlHash == null) {
|
if (urlHash == null) {
|
||||||
fail++;
|
fail++;
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
hashByUrl.put(url, urlHash);
|
||||||
|
}
|
||||||
|
if (!hashByUrl.isEmpty()) {
|
||||||
|
List<String> allHashes = new ArrayList<>(hashByUrl.values());
|
||||||
|
for (int start = 0; start < allHashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) {
|
||||||
|
List<String> batch = allHashes.subList(start,
|
||||||
|
Math.min(start + CACHE_LOOKUP_BATCH_SIZE, allHashes.size()));
|
||||||
// 命中 DB cache:bumping last_used_at 即可,不再下载。
|
// 命中 DB cache:bumping last_used_at 即可,不再下载。
|
||||||
Long existing = taskImageCacheMapper.selectCount(new LambdaQueryWrapper<TaskImageCacheEntity>()
|
List<String> found = taskImageCacheMapper.selectUrlHashes(batch);
|
||||||
.eq(TaskImageCacheEntity::getUrlHash, urlHash));
|
if (found != null) {
|
||||||
if (existing != null && existing > 0L) {
|
cachedHashes.addAll(found);
|
||||||
taskImageCacheMapper.touchLastUsed(urlHash);
|
}
|
||||||
hit++;
|
}
|
||||||
|
for (int start = 0; start < allHashes.size(); start += CACHE_LOOKUP_BATCH_SIZE) {
|
||||||
|
taskImageCacheMapper.touchLastUsedBatch(allHashes.subList(start,
|
||||||
|
Math.min(start + CACHE_LOOKUP_BATCH_SIZE, allHashes.size())));
|
||||||
|
}
|
||||||
|
hit = cachedHashes.size();
|
||||||
|
}
|
||||||
|
for (Map.Entry<String, String> entry : hashByUrl.entrySet()) {
|
||||||
|
String url = entry.getKey();
|
||||||
|
String urlHash = entry.getValue();
|
||||||
|
if (cachedHashes.contains(urlHash)) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
try {
|
||||||
ResizedImage thumb = imageEmbedder.fetchAndResizeForCache(url);
|
ResizedImage thumb = imageEmbedder.fetchAndResizeForCache(url);
|
||||||
if (thumb == null || thumb.bytes() == null || thumb.bytes().length == 0) {
|
if (thumb == null || thumb.bytes() == null || thumb.bytes().length == 0) {
|
||||||
fail++;
|
fail++;
|
||||||
|
|||||||
+6
@@ -3751,7 +3751,13 @@ public class SimilarAsinTaskService {
|
|||||||
if (taskId == null || taskId <= 0) {
|
if (taskId == null || taskId <= 0) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
// 只取 flush 需要的列:parsed_payload_json/coze_* 等大字段不拉,避免跨库大结果集传输(每任务可达数百行大 JSON)
|
||||||
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||||
|
.select(TaskScopeStateEntity::getId,
|
||||||
|
TaskScopeStateEntity::getTaskId,
|
||||||
|
TaskScopeStateEntity::getModuleType,
|
||||||
|
TaskScopeStateEntity::getCozeStatus,
|
||||||
|
TaskScopeStateEntity::getStateJson)
|
||||||
.eq(TaskScopeStateEntity::getTaskId, taskId)
|
.eq(TaskScopeStateEntity::getTaskId, taskId)
|
||||||
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
|
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
|
||||||
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_DONE, COZE_STATUS_FAILED)));
|
.in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_DONE, COZE_STATUS_FAILED)));
|
||||||
|
|||||||
+15
@@ -9,6 +9,7 @@ import org.apache.ibatis.annotations.Select;
|
|||||||
import org.apache.ibatis.annotations.Update;
|
import org.apache.ibatis.annotations.Update;
|
||||||
|
|
||||||
import java.time.LocalDateTime;
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* P2-12:图片缩略图缓存 mapper。提供 LRU 命中刷新、按 url_hash 直读字节两个轻量入口,
|
* P2-12:图片缩略图缓存 mapper。提供 LRU 命中刷新、按 url_hash 直读字节两个轻量入口,
|
||||||
@@ -23,6 +24,20 @@ public interface TaskImageCacheMapper extends BaseMapper<TaskImageCacheEntity> {
|
|||||||
@Update("UPDATE biz_task_image_cache SET last_used_at = NOW(3) WHERE url_hash = #{urlHash}")
|
@Update("UPDATE biz_task_image_cache SET last_used_at = NOW(3) WHERE url_hash = #{urlHash}")
|
||||||
int touchLastUsed(@Param("urlHash") String urlHash);
|
int touchLastUsed(@Param("urlHash") String urlHash);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 批量命中检查:仅取 url_hash 列(避免把 image_bytes BLOB 拉回 JVM)。
|
||||||
|
*/
|
||||||
|
@Select("<script>SELECT url_hash FROM biz_task_image_cache WHERE url_hash IN " +
|
||||||
|
"<foreach collection='urlHashes' item='hash' open='(' separator=',' close=')'>#{hash}</foreach></script>")
|
||||||
|
List<String> selectUrlHashes(@Param("urlHashes") List<String> urlHashes);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 批量 bumping last_used_at。
|
||||||
|
*/
|
||||||
|
@Update("<script>UPDATE biz_task_image_cache SET last_used_at = NOW(3) WHERE url_hash IN " +
|
||||||
|
"<foreach collection='urlHashes' item='hash' open='(' separator=',' close=')'>#{hash}</foreach></script>")
|
||||||
|
int touchLastUsedBatch(@Param("urlHashes") List<String> urlHashes);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 按 url_hash 直读缩略图字节,命中时返回 BLOB;未命中返回 null。
|
* 按 url_hash 直读缩略图字节,命中时返回 BLOB;未命中返回 null。
|
||||||
* 选择只 select image_bytes 一列,避免把整行 entity(含 url 字符串)拉回 JVM。
|
* 选择只 select image_bytes 一列,避免把整行 entity(含 url 字符串)拉回 JVM。
|
||||||
|
|||||||
+51
@@ -1,6 +1,7 @@
|
|||||||
package com.nanri.aiimage.modules.ziniao.memory.service;
|
package com.nanri.aiimage.modules.ziniao.memory.service;
|
||||||
|
|
||||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
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.common.exception.BusinessException;
|
||||||
import com.nanri.aiimage.modules.ziniao.memory.mapper.ZiniaoMemoryStoreMapper;
|
import com.nanri.aiimage.modules.ziniao.memory.mapper.ZiniaoMemoryStoreMapper;
|
||||||
import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEntity;
|
import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEntity;
|
||||||
@@ -14,6 +15,7 @@ import org.springframework.transaction.annotation.Transactional;
|
|||||||
|
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.time.LocalDateTime;
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Objects;
|
import java.util.Objects;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
@@ -181,6 +183,55 @@ public class ZiniaoMemoryStoreService {
|
|||||||
return deleted;
|
return deleted;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 批量标 STALE 回写(索引巡检):调用方已持有完整行(含 id),不再逐条 findOne,
|
||||||
|
* 避免每轮刷新把数百行按"每个事务一条 SELECT + 一条 UPDATE"的方式打回数据库。
|
||||||
|
* 仅 payload 内容发生变化的行走这里。
|
||||||
|
*/
|
||||||
|
@Transactional
|
||||||
|
public int updateStaleMarks(List<ZiniaoMemoryStoreEntity> rows) {
|
||||||
|
if (rows == null || rows.isEmpty()) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
int updated = 0;
|
||||||
|
for (ZiniaoMemoryStoreEntity row : rows) {
|
||||||
|
if (row == null || row.getId() == null) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
updated += ziniaoMemoryStoreMapper.updateById(row);
|
||||||
|
if (CACHE_TYPE_SHOP_INDEX_ENTRY.equals(row.getCacheType())) {
|
||||||
|
log.info("[ziniao-shop-index-db] mark-stale cacheKey={} id={} companyNameSample={} expiresAt={} bytes={}",
|
||||||
|
row.getCacheKey(), row.getId(), sampleCompanyNameForLog(row.getCompanyName()),
|
||||||
|
row.getExpiresAt(), row.getPayloadJson() == null ? 0 : row.getPayloadJson().length());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return updated;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 批量刷新过期时间(索引巡检):payload 未变化的行不重写 JSON,仅用一条 UPDATE 延长 TTL,
|
||||||
|
* 保持"未过期行常驻表内"的原有语义。
|
||||||
|
*/
|
||||||
|
@Transactional
|
||||||
|
public void touchExpiryBatch(List<ZiniaoMemoryStoreEntity> rows, Duration ttl) {
|
||||||
|
if (rows == null || rows.isEmpty() || ttl == null || ttl.isZero() || ttl.isNegative()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
List<Long> ids = new ArrayList<>();
|
||||||
|
for (ZiniaoMemoryStoreEntity row : rows) {
|
||||||
|
if (row != null && row.getId() != null) {
|
||||||
|
ids.add(row.getId());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (ids.isEmpty()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
ziniaoMemoryStoreMapper.update(null, new LambdaUpdateWrapper<ZiniaoMemoryStoreEntity>()
|
||||||
|
.in(ZiniaoMemoryStoreEntity::getId, ids)
|
||||||
|
.set(ZiniaoMemoryStoreEntity::getExpiresAt, LocalDateTime.now().plusSeconds(ttl.getSeconds()))
|
||||||
|
.set(ZiniaoMemoryStoreEntity::getUpdatedAt, LocalDateTime.now()));
|
||||||
|
}
|
||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
public int deleteAllByType(String cacheType) {
|
public int deleteAllByType(String cacheType) {
|
||||||
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
|
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
|
||||||
|
|||||||
+27
-2
@@ -19,6 +19,7 @@ import java.nio.charset.StandardCharsets;
|
|||||||
import java.security.MessageDigest;
|
import java.security.MessageDigest;
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.time.Instant;
|
import java.time.Instant;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.HashSet;
|
import java.util.HashSet;
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
@@ -590,10 +591,13 @@ public class ZiniaoShopIndexService {
|
|||||||
if (entities.isEmpty()) {
|
if (entities.isEmpty()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
List<ZiniaoMemoryStoreEntity> staleUpdates = new ArrayList<>();
|
||||||
|
List<ZiniaoMemoryStoreEntity> expiryTouches = new ArrayList<>();
|
||||||
int staleCount = 0;
|
int staleCount = 0;
|
||||||
for (ZiniaoMemoryStoreEntity entity : entities) {
|
for (ZiniaoMemoryStoreEntity entity : entities) {
|
||||||
String rowKey = entity.getCacheKey();
|
String rowKey = entity.getCacheKey();
|
||||||
if (activeCacheKeys.contains(rowKey)) {
|
if (activeCacheKeys.contains(rowKey)) {
|
||||||
|
// 活跃行已由本轮 refresh 的 put() 刷新过 TTL,无需再碰
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
ZiniaoShopIndexEntryDto existingEntry;
|
ZiniaoShopIndexEntryDto existingEntry;
|
||||||
@@ -629,10 +633,31 @@ public class ZiniaoShopIndexService {
|
|||||||
}
|
}
|
||||||
existingEntry.setStatus(STATUS_STALE);
|
existingEntry.setStatus(STATUS_STALE);
|
||||||
existingEntry.setMessage("店铺索引已过期,请等待后台刷新");
|
existingEntry.setMessage("店铺索引已过期,请等待后台刷新");
|
||||||
existingEntry.setLastRefreshedAt(now);
|
// 不再更新 lastRefreshedAt:STALE 行查询路径不读该字段(仅 ACTIVE 行参与保鲜判断),
|
||||||
ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, rowKey, existingEntry, resolveEntryTtl());
|
// 跳过可让已标 STALE 且内容不变的行不再被逐轮全量重写。
|
||||||
|
String newPayload;
|
||||||
|
try {
|
||||||
|
newPayload = objectMapper.writeValueAsString(existingEntry);
|
||||||
|
} catch (Exception ex) {
|
||||||
|
throw new BusinessException("写入紫鸟记忆存储失败", ex);
|
||||||
|
}
|
||||||
|
if (Objects.equals(newPayload, entity.getPayloadJson())) {
|
||||||
|
// 内容未变化:仅延长 TTL,保持"未过期行常驻"语义,不重写 JSON
|
||||||
|
expiryTouches.add(entity);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
entity.setPayloadJson(newPayload);
|
||||||
|
entity.setExpiresAt(LocalDateTime.now().plus(resolveEntryTtl()));
|
||||||
|
entity.setUpdatedAt(LocalDateTime.now());
|
||||||
|
staleUpdates.add(entity);
|
||||||
staleCount++;
|
staleCount++;
|
||||||
}
|
}
|
||||||
|
if (!staleUpdates.isEmpty()) {
|
||||||
|
ziniaoMemoryStoreService.updateStaleMarks(staleUpdates);
|
||||||
|
}
|
||||||
|
if (!expiryTouches.isEmpty()) {
|
||||||
|
ziniaoMemoryStoreService.touchExpiryBatch(expiryTouches, resolveEntryTtl());
|
||||||
|
}
|
||||||
if (staleCount > 0) {
|
if (staleCount > 0) {
|
||||||
log.info("[ziniao-index] marked stale shop_index rows count={}", staleCount);
|
log.info("[ziniao-index] marked stale shop_index rows count={}", staleCount);
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -388,7 +388,7 @@ class ShopDataCrawlTaskServiceRetentionTest {
|
|||||||
when(taskDistributedLockService.acquire(MODULE_TYPE, TASK_ID))
|
when(taskDistributedLockService.acquire(MODULE_TYPE, TASK_ID))
|
||||||
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
||||||
when(fileResultMapper.selectById(RESULT_ID)).thenReturn(currentRow);
|
when(fileResultMapper.selectById(RESULT_ID)).thenReturn(currentRow);
|
||||||
when(fileResultMapper.selectById(200L)).thenReturn(previous);
|
when(fileResultMapper.selectBatchIds(List.of(200L))).thenReturn(List.of(previous));
|
||||||
when(fileResultMapper.selectList(any())).thenReturn(List.of());
|
when(fileResultMapper.selectList(any())).thenReturn(List.of());
|
||||||
when(dailyFileService.findByLatestResultId(RESULT_ID)).thenReturn(List.of(daily));
|
when(dailyFileService.findByLatestResultId(RESULT_ID)).thenReturn(List.of(daily));
|
||||||
when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of(removedMember));
|
when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of(removedMember));
|
||||||
|
|||||||
+107
@@ -3,6 +3,7 @@ package com.nanri.aiimage.modules.ziniao.service;
|
|||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
import com.nanri.aiimage.common.exception.BusinessException;
|
import com.nanri.aiimage.common.exception.BusinessException;
|
||||||
import com.nanri.aiimage.config.ZiniaoProperties;
|
import com.nanri.aiimage.config.ZiniaoProperties;
|
||||||
|
import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEntity;
|
||||||
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoMemoryStoreService;
|
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoMemoryStoreService;
|
||||||
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoTransientCacheService;
|
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoTransientCacheService;
|
||||||
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopIndexEntryDto;
|
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopIndexEntryDto;
|
||||||
@@ -17,6 +18,7 @@ import org.mockito.Mock;
|
|||||||
import org.mockito.junit.jupiter.MockitoExtension;
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
|
|
||||||
@@ -24,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
|
|||||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
import static org.mockito.ArgumentMatchers.anyInt;
|
import static org.mockito.ArgumentMatchers.anyInt;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyList;
|
||||||
import static org.mockito.ArgumentMatchers.argThat;
|
import static org.mockito.ArgumentMatchers.argThat;
|
||||||
import static org.mockito.ArgumentMatchers.eq;
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
import static org.mockito.Mockito.never;
|
import static org.mockito.Mockito.never;
|
||||||
@@ -209,6 +212,110 @@ class ZiniaoShopIndexServiceTest {
|
|||||||
assertEquals(Integer.valueOf(0), cursor.getNextApiKeyOffset());
|
assertEquals(Integer.valueOf(0), cursor.getNextApiKeyOffset());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void staleRowsWithoutPayloadChangeSkipRewriteAndOnlyTouchExpiry() throws Exception {
|
||||||
|
ZiniaoApiKeyProvider.ApiKeyAccount key = new ZiniaoApiKeyProvider.ApiKeyAccount("key", "acct");
|
||||||
|
when(ziniaoApiKeyProvider.listApiKeyAccounts()).thenReturn(List.of(key));
|
||||||
|
when(ziniaoAuthService.resolveCompanyIdForIndex("key")).thenReturn(1L);
|
||||||
|
when(ziniaoAuthService.getOrLoadStaffForIndex("key", 1L)).thenReturn(List.of(staff(11L)));
|
||||||
|
when(ziniaoAuthService.getOrLoadUserStoresForIndex("key", 1L, 11L)).thenReturn(List.of());
|
||||||
|
|
||||||
|
ZiniaoShopIndexEntryDto dto = new ZiniaoShopIndexEntryDto();
|
||||||
|
dto.setNormalizedShopName("shop-a");
|
||||||
|
dto.setCompanyName("acct");
|
||||||
|
dto.setStatus("STALE");
|
||||||
|
dto.setMessage("店铺索引已过期,请等待后台刷新");
|
||||||
|
dto.setLastRefreshedAt(1000L);
|
||||||
|
String payload = new ObjectMapper().writeValueAsString(dto);
|
||||||
|
ZiniaoMemoryStoreEntity row = new ZiniaoMemoryStoreEntity();
|
||||||
|
row.setId(1L);
|
||||||
|
row.setCacheType("SHOP_INDEX_ENTRY");
|
||||||
|
row.setCacheKey("n:shop-a");
|
||||||
|
row.setCompanyName("acct");
|
||||||
|
row.setPayloadJson(payload);
|
||||||
|
row.setExpiresAt(LocalDateTime.now().plusHours(12));
|
||||||
|
when(ziniaoMemoryStoreService.listAliveEntitiesByType("SHOP_INDEX_ENTRY", 10000)).thenReturn(List.of(row));
|
||||||
|
|
||||||
|
service.refreshAllShopIndex();
|
||||||
|
|
||||||
|
verify(ziniaoMemoryStoreService, never()).updateStaleMarks(anyList());
|
||||||
|
verify(ziniaoMemoryStoreService).touchExpiryBatch(anyList(), any(Duration.class));
|
||||||
|
verify(ziniaoMemoryStoreService, never()).put(any(), any(), any(), any());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void activeEntryMissingFromThisRoundGetsRewrittenAsStale() throws Exception {
|
||||||
|
ZiniaoApiKeyProvider.ApiKeyAccount key = new ZiniaoApiKeyProvider.ApiKeyAccount("key", "acct");
|
||||||
|
when(ziniaoApiKeyProvider.listApiKeyAccounts()).thenReturn(List.of(key));
|
||||||
|
when(ziniaoAuthService.resolveCompanyIdForIndex("key")).thenReturn(1L);
|
||||||
|
when(ziniaoAuthService.getOrLoadStaffForIndex("key", 1L)).thenReturn(List.of(staff(11L)));
|
||||||
|
when(ziniaoAuthService.getOrLoadUserStoresForIndex("key", 1L, 11L)).thenReturn(List.of());
|
||||||
|
|
||||||
|
ZiniaoShopIndexEntryDto dto = new ZiniaoShopIndexEntryDto();
|
||||||
|
dto.setNormalizedShopName("shop-a");
|
||||||
|
dto.setStatus("ACTIVE");
|
||||||
|
dto.setLastRefreshedAt(1000L);
|
||||||
|
String payload = new ObjectMapper().writeValueAsString(dto);
|
||||||
|
ZiniaoMemoryStoreEntity row = new ZiniaoMemoryStoreEntity();
|
||||||
|
row.setId(1L);
|
||||||
|
row.setCacheType("SHOP_INDEX_ENTRY");
|
||||||
|
row.setCacheKey("n:shop-a");
|
||||||
|
row.setCompanyName("acct");
|
||||||
|
row.setPayloadJson(payload);
|
||||||
|
row.setExpiresAt(LocalDateTime.now().plusHours(12));
|
||||||
|
when(ziniaoMemoryStoreService.listAliveEntitiesByType("SHOP_INDEX_ENTRY", 10000)).thenReturn(List.of(row));
|
||||||
|
|
||||||
|
service.refreshAllShopIndex();
|
||||||
|
|
||||||
|
ArgumentCaptor<List<ZiniaoMemoryStoreEntity>> captor = ArgumentCaptor.forClass(List.class);
|
||||||
|
verify(ziniaoMemoryStoreService).updateStaleMarks(captor.capture());
|
||||||
|
assertEquals(1, captor.getValue().size());
|
||||||
|
ZiniaoMemoryStoreEntity updated = captor.getValue().get(0);
|
||||||
|
assertEquals(1L, updated.getId());
|
||||||
|
assertTrue(updated.getPayloadJson().contains("\"STALE\""));
|
||||||
|
assertEquals("店铺索引已过期,请等待后台刷新",
|
||||||
|
new ObjectMapper().readTree(updated.getPayloadJson()).get("message").asText());
|
||||||
|
verify(ziniaoMemoryStoreService, never()).touchExpiryBatch(anyList(), any(Duration.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void activeKeysFromThisRoundAreSkippedInStaleMarking() throws Exception {
|
||||||
|
ZiniaoApiKeyProvider.ApiKeyAccount key = new ZiniaoApiKeyProvider.ApiKeyAccount("key", "acct");
|
||||||
|
when(ziniaoApiKeyProvider.listApiKeyAccounts()).thenReturn(List.of(key));
|
||||||
|
when(ziniaoAuthService.resolveCompanyIdForIndex("key")).thenReturn(1L);
|
||||||
|
when(ziniaoAuthService.getOrLoadStaffForIndex("key", 1L)).thenReturn(List.of(staff(11L)));
|
||||||
|
when(ziniaoAuthService.getOrLoadUserStoresForIndex("key", 1L, 11L)).thenReturn(List.of(shop("s-1", "shop-1")));
|
||||||
|
|
||||||
|
ZiniaoShopIndexEntryDto activeDto = new ZiniaoShopIndexEntryDto();
|
||||||
|
activeDto.setNormalizedShopName("shop-1");
|
||||||
|
activeDto.setStatus("ACTIVE");
|
||||||
|
ZiniaoMemoryStoreEntity activeRow = new ZiniaoMemoryStoreEntity();
|
||||||
|
activeRow.setId(1L);
|
||||||
|
activeRow.setCacheType("SHOP_INDEX_ENTRY");
|
||||||
|
activeRow.setCacheKey("s:s-1");
|
||||||
|
activeRow.setPayloadJson(new ObjectMapper().writeValueAsString(activeDto));
|
||||||
|
activeRow.setExpiresAt(LocalDateTime.now().plusHours(12));
|
||||||
|
|
||||||
|
ZiniaoShopIndexEntryDto staleDto = new ZiniaoShopIndexEntryDto();
|
||||||
|
staleDto.setNormalizedShopName("shop-2");
|
||||||
|
staleDto.setStatus("ACTIVE");
|
||||||
|
ZiniaoMemoryStoreEntity staleRow = new ZiniaoMemoryStoreEntity();
|
||||||
|
staleRow.setId(2L);
|
||||||
|
staleRow.setCacheType("SHOP_INDEX_ENTRY");
|
||||||
|
staleRow.setCacheKey("n:shop-2");
|
||||||
|
staleRow.setPayloadJson(new ObjectMapper().writeValueAsString(staleDto));
|
||||||
|
staleRow.setExpiresAt(LocalDateTime.now().plusHours(12));
|
||||||
|
when(ziniaoMemoryStoreService.listAliveEntitiesByType("SHOP_INDEX_ENTRY", 10000))
|
||||||
|
.thenReturn(List.of(activeRow, staleRow));
|
||||||
|
|
||||||
|
service.refreshAllShopIndex();
|
||||||
|
|
||||||
|
ArgumentCaptor<List<ZiniaoMemoryStoreEntity>> captor = ArgumentCaptor.forClass(List.class);
|
||||||
|
verify(ziniaoMemoryStoreService).updateStaleMarks(captor.capture());
|
||||||
|
assertEquals(1, captor.getValue().size());
|
||||||
|
assertEquals("n:shop-2", captor.getValue().get(0).getCacheKey());
|
||||||
|
}
|
||||||
|
|
||||||
private ZiniaoShopIndexRefreshCursorDto capturedCursor() {
|
private ZiniaoShopIndexRefreshCursorDto capturedCursor() {
|
||||||
ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
|
ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
|
||||||
verify(ziniaoTransientCacheService, times(2)).put(
|
verify(ziniaoTransientCacheService, times(2)).put(
|
||||||
|
|||||||
Reference in New Issue
Block a user