feat(需求): 撞款重复检查改查数据库——①采集明细先落库再更新文件(V111 biz_shop_data_crawl_item行级明细,归档时countryResults展开,同店同日幂等替换) ②撞款扫描换源直接查明细表(不再解析OSS xlsx,修复旧文件表头差异致4家店静默丢弃) ③存量Excel数据自动迁移(日报快照row_payload优先+OSS宽松表头回退:ASIN码/小写兼容) ④移除84家店铺全量空态注入(重复检查基数=已采集店,超管口径8家) ⑤新增明细聚合/宽松解析单测6个
This commit is contained in:
+18
@@ -0,0 +1,18 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopdatacrawl.mapper;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlItemEntity;
|
||||||
|
import org.apache.ibatis.annotations.Delete;
|
||||||
|
import org.apache.ibatis.annotations.Mapper;
|
||||||
|
import org.apache.ibatis.annotations.Param;
|
||||||
|
|
||||||
|
@Mapper
|
||||||
|
public interface ShopDataCrawlItemMapper extends BaseMapper<ShopDataCrawlItemEntity> {
|
||||||
|
|
||||||
|
/** 删除某店某批次(每日快照替换语义:先删后插)。 */
|
||||||
|
@Delete("""
|
||||||
|
DELETE FROM biz_shop_data_crawl_item
|
||||||
|
WHERE shop_name = #{shopName} AND business_date = #{businessDate}
|
||||||
|
""")
|
||||||
|
int deleteBatch(@Param("shopName") String shopName, @Param("businessDate") java.time.LocalDate businessDate);
|
||||||
|
}
|
||||||
+32
@@ -0,0 +1,32 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopdatacrawl.model.entity;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.annotation.IdType;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableId;
|
||||||
|
import com.baomidou.mybatisplus.annotation.TableName;
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
|
||||||
|
@Data
|
||||||
|
@TableName("biz_shop_data_crawl_item")
|
||||||
|
public class ShopDataCrawlItemEntity {
|
||||||
|
|
||||||
|
@TableId(type = IdType.AUTO)
|
||||||
|
private Long id;
|
||||||
|
private Long shopId;
|
||||||
|
private String shopName;
|
||||||
|
private String groupName;
|
||||||
|
private LocalDate businessDate;
|
||||||
|
private String country;
|
||||||
|
private String asin;
|
||||||
|
private String brand;
|
||||||
|
private String price;
|
||||||
|
private String itemDate;
|
||||||
|
private String unitsSold;
|
||||||
|
private String itemJson;
|
||||||
|
private Long resultId;
|
||||||
|
private Long taskId;
|
||||||
|
private Long dailyFileId;
|
||||||
|
private LocalDateTime createdAt;
|
||||||
|
}
|
||||||
+129
@@ -0,0 +1,129 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopdatacrawl.service;
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.mapper.ShopDataCrawlItemMapper;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlItemEntity;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo;
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
import org.springframework.transaction.annotation.Transactional;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.LinkedHashMap;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 店铺数据采集行级明细落库:采集数据先落库、再更新 Excel 文件;
|
||||||
|
* 撞款重复检查(shopduplicatecheck)直接查明细表聚合,不再解析 OSS xlsx。
|
||||||
|
* 语义:每店×每日一份累计快照(先删同批再插,幂等可重放)。
|
||||||
|
*/
|
||||||
|
@Slf4j
|
||||||
|
@Service
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
public class ShopDataCrawlItemStoreService {
|
||||||
|
|
||||||
|
private final ShopDataCrawlItemMapper itemMapper;
|
||||||
|
private final ObjectMapper objectMapper;
|
||||||
|
|
||||||
|
/** 国家站点码(旧文件 sheet 未命中映射则保留原文参与撞款,国家序列为空)。 */
|
||||||
|
public static String normalizeCountry(String raw) {
|
||||||
|
return raw == null ? "" : raw.trim().toUpperCase();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 把当日累计快照(countryResults 展开)落为行级明细。
|
||||||
|
* 增量归档调用点在 Excel 组装/上传之前(先落库再更新文件);
|
||||||
|
* 存量迁移由 shopduplicatecheck 模块构造 Vo 后复用(幂等:先删同批再插)。
|
||||||
|
*/
|
||||||
|
@Transactional(rollbackFor = Exception.class)
|
||||||
|
public void saveShopBatchFromSnapshot(String shopName, LocalDate businessDate,
|
||||||
|
List<ShopDataCrawlResultItemVo> snapshots,
|
||||||
|
Long resultId, Long taskId, Long dailyFileId) {
|
||||||
|
List<ShopDataCrawlItemEntity> rows = new ArrayList<>();
|
||||||
|
for (ShopDataCrawlResultItemVo vo : snapshots) {
|
||||||
|
if (vo == null || Boolean.FALSE.equals(vo.getSuccess()) || vo.getCountryResults() == null) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
for (ShopDataCrawlCountryResultDto countryResult : vo.getCountryResults()) {
|
||||||
|
if (countryResult == null || countryResult.getItems() == null) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
String country = normalizeCountry(countryResult.getCountry());
|
||||||
|
for (ShopDataCrawlRowDto item : countryResult.getItems()) {
|
||||||
|
if (item == null || blank(item.getAsin())) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
rows.add(toEntity(shopName, businessDate, country, item, resultId, taskId, dailyFileId));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
replaceBatch(shopName, businessDate, rows, taskId);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void replaceBatch(String shopName, LocalDate businessDate,
|
||||||
|
List<ShopDataCrawlItemEntity> rows, Long taskId) {
|
||||||
|
int deleted = itemMapper.deleteBatch(shopName, businessDate);
|
||||||
|
int inserted = 0;
|
||||||
|
for (ShopDataCrawlItemEntity row : rows) {
|
||||||
|
itemMapper.insert(row);
|
||||||
|
inserted++;
|
||||||
|
}
|
||||||
|
log.info("[shop-data-crawl-item] 店铺明细批次落库 shop={} date={} 删除 {} 行,新插 {} 行(taskId={})",
|
||||||
|
shopName, businessDate, deleted, inserted, taskId);
|
||||||
|
}
|
||||||
|
|
||||||
|
private ShopDataCrawlItemEntity toEntity(String shopName, LocalDate businessDate, String country,
|
||||||
|
ShopDataCrawlRowDto item, Long resultId, Long taskId, Long dailyFileId) {
|
||||||
|
ShopDataCrawlItemEntity entity = new ShopDataCrawlItemEntity();
|
||||||
|
entity.setShopName(shopName);
|
||||||
|
entity.setBusinessDate(businessDate);
|
||||||
|
entity.setCountry(country);
|
||||||
|
entity.setAsin(item.getAsin().trim());
|
||||||
|
entity.setBrand(trimToNull(item.getBrand()));
|
||||||
|
entity.setPrice(trimToNull(item.getPrice()));
|
||||||
|
entity.setItemDate(trimToNull(item.getDate()));
|
||||||
|
entity.setUnitsSold(trimToNull(item.getUnitsSold()));
|
||||||
|
entity.setItemJson(extraColumnsJson(item));
|
||||||
|
entity.setResultId(resultId);
|
||||||
|
entity.setTaskId(taskId);
|
||||||
|
entity.setDailyFileId(dailyFileId);
|
||||||
|
return entity;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 其余展示列(库存销量/销售排名/页面浏览量/商品图)原文兜底为 JSON,明细查询不依赖。 */
|
||||||
|
private String extraColumnsJson(ShopDataCrawlRowDto item) {
|
||||||
|
try {
|
||||||
|
Map<String, Object> extra = new LinkedHashMap<>();
|
||||||
|
if (item.getInventorySales() != null) {
|
||||||
|
extra.put("inventorySales", item.getInventorySales());
|
||||||
|
}
|
||||||
|
if (item.getSalesRank() != null) {
|
||||||
|
extra.put("salesRank", item.getSalesRank());
|
||||||
|
}
|
||||||
|
if (item.getPageViews() != null) {
|
||||||
|
extra.put("pageViews", item.getPageViews());
|
||||||
|
}
|
||||||
|
if (item.getCommodityImage() != null) {
|
||||||
|
extra.put("commodityImage", item.getCommodityImage());
|
||||||
|
}
|
||||||
|
return extra.isEmpty() ? null : objectMapper.writeValueAsString(extra);
|
||||||
|
} catch (Exception ex) {
|
||||||
|
log.warn("[shop-data-crawl-item] 明细附加列序列化失败 shop={} asin={}",
|
||||||
|
item.getAsin(), ex.getMessage());
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static boolean blank(String value) {
|
||||||
|
return value == null || value.isBlank();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String trimToNull(String value) {
|
||||||
|
return value == null || value.isBlank() ? null : value.trim();
|
||||||
|
}
|
||||||
|
}
|
||||||
+9
@@ -118,6 +118,7 @@ public class ShopDataCrawlTaskService {
|
|||||||
private final TransientPayloadStorageService transientPayloadStorageService;
|
private final TransientPayloadStorageService transientPayloadStorageService;
|
||||||
private final InstanceMetadata instanceMetadata;
|
private final InstanceMetadata instanceMetadata;
|
||||||
private final ShopDataCrawlDailyFileService dailyFileService;
|
private final ShopDataCrawlDailyFileService dailyFileService;
|
||||||
|
private final ShopDataCrawlItemStoreService shopDataCrawlItemStoreService;
|
||||||
private final PlatformTransactionManager transactionManager;
|
private final PlatformTransactionManager transactionManager;
|
||||||
private final TaskProgressLightAssembler taskProgressLightAssembler;
|
private final TaskProgressLightAssembler taskProgressLightAssembler;
|
||||||
|
|
||||||
@@ -1931,6 +1932,14 @@ public class ShopDataCrawlTaskService {
|
|||||||
// 不再读回旧累计对象(readObjectBytes)并整表重写(replaceCountriesWorkbook);
|
// 不再读回旧累计对象(readObjectBytes)并整表重写(replaceCountriesWorkbook);
|
||||||
// 历史成员行无 payload 时按结果快照兜底,兼容旧归档数据。
|
// 历史成员行无 payload 时按结果快照兜底,兼容旧归档数据。
|
||||||
List<ShopDataCrawlResultItemVo> accumulatedItems = buildDailyFileFromData(baseDailyFile, List.of(snapshot));
|
List<ShopDataCrawlResultItemVo> accumulatedItems = buildDailyFileFromData(baseDailyFile, List.of(snapshot));
|
||||||
|
// 先落库再更新文件(用户约定):行级明细先于 Excel 组装/上传;
|
||||||
|
// 失败抛异常走现有失败/补偿路径,明细幂等(同店同日先删后插)。
|
||||||
|
LocalDate itemBatchDate = baseDailyFile != null && baseDailyFile.getBusinessDate() != null
|
||||||
|
? baseDailyFile.getBusinessDate()
|
||||||
|
: dailyFileService.currentBusinessDateTime().toLocalDate();
|
||||||
|
shopDataCrawlItemStoreService.saveShopBatchFromSnapshot(
|
||||||
|
snapshot.getShopName(), itemBatchDate, accumulatedItems,
|
||||||
|
snapshot.getResultId(), task.getId(), baseDailyFile == null ? null : baseDailyFile.getId());
|
||||||
int rowCount = excelAssemblyService.writeWorkbook(outputXlsx, accumulatedItems);
|
int rowCount = excelAssemblyService.writeWorkbook(outputXlsx, accumulatedItems);
|
||||||
String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE);
|
String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE);
|
||||||
if (blank(objectKey)) {
|
if (blank(objectKey)) {
|
||||||
|
|||||||
+75
@@ -0,0 +1,75 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopduplicatecheck.mapper;
|
||||||
|
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopDailyFileMigrationDto;
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopLatestItemRowDto;
|
||||||
|
import org.apache.ibatis.annotations.Mapper;
|
||||||
|
import org.apache.ibatis.annotations.Param;
|
||||||
|
import org.apache.ibatis.annotations.Select;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 撞款扫描的明细表数据源:biz_shop_data_crawl_item(采集先落库)。
|
||||||
|
* 不依赖 OSS xlsx 解析,每店取最新 business_date 批次行聚合。
|
||||||
|
*/
|
||||||
|
@Mapper
|
||||||
|
public interface ShopDuplicateCheckItemMapper {
|
||||||
|
|
||||||
|
/** 每店最新 business_date 批次的全部明细行(撞款统计直接数据源)。 */
|
||||||
|
@Select("""
|
||||||
|
SELECT i.shop_name AS shopName,
|
||||||
|
i.country AS country,
|
||||||
|
i.asin AS asin,
|
||||||
|
i.item_date AS itemDate,
|
||||||
|
i.price AS price,
|
||||||
|
i.brand AS brand
|
||||||
|
FROM biz_shop_data_crawl_item i
|
||||||
|
JOIN (
|
||||||
|
SELECT shop_name, MAX(business_date) AS max_date
|
||||||
|
FROM biz_shop_data_crawl_item
|
||||||
|
GROUP BY shop_name
|
||||||
|
) latest
|
||||||
|
ON latest.shop_name = i.shop_name AND latest.max_date = i.business_date
|
||||||
|
""")
|
||||||
|
List<ShopLatestItemRowDto> selectLatestShopItems();
|
||||||
|
|
||||||
|
/** 幂等判定:该店该日是否已有明细批次。 */
|
||||||
|
@Select("""
|
||||||
|
SELECT COUNT(1) FROM biz_shop_data_crawl_item
|
||||||
|
WHERE shop_name = #{shopName} AND business_date = #{businessDate}
|
||||||
|
""")
|
||||||
|
long countShopBatch(@Param("shopName") String shopName, @Param("businessDate") LocalDate businessDate);
|
||||||
|
|
||||||
|
/** 存量迁移数据源:每店最新已补偿 daily_file(join 最新结果行拿店名/文件地址)。 */
|
||||||
|
@Select("""
|
||||||
|
SELECT ranked.daily_file_id AS dailyFileId,
|
||||||
|
ranked.business_date AS businessDate,
|
||||||
|
ranked.latest_result_id AS latestResultId,
|
||||||
|
fr.source_filename AS shopName,
|
||||||
|
fr.result_file_url AS resultFileUrl,
|
||||||
|
fr.task_id AS taskId
|
||||||
|
FROM (
|
||||||
|
SELECT df.id AS daily_file_id,
|
||||||
|
df.business_date,
|
||||||
|
df.latest_result_id,
|
||||||
|
ROW_NUMBER() OVER (
|
||||||
|
PARTITION BY df.shop_key
|
||||||
|
ORDER BY df.business_date DESC, df.id DESC) AS row_no
|
||||||
|
FROM biz_shop_data_crawl_daily_file df
|
||||||
|
WHERE df.compensation_done = 1
|
||||||
|
) ranked
|
||||||
|
JOIN biz_file_result fr ON fr.id = ranked.latest_result_id
|
||||||
|
WHERE ranked.row_no = 1
|
||||||
|
AND fr.module_type = 'SHOP_DATA_CRAWL'
|
||||||
|
""")
|
||||||
|
List<ShopDailyFileMigrationDto> selectLegacyMigrationFiles();
|
||||||
|
|
||||||
|
/** 取每日成员行(row_payload 快照 JSON),无 payload 时返回 null。 */
|
||||||
|
@Select("""
|
||||||
|
SELECT row_payload FROM biz_shop_data_crawl_daily_member
|
||||||
|
WHERE daily_file_id = #{dailyFileId} AND result_id = #{resultId}
|
||||||
|
LIMIT 1
|
||||||
|
""")
|
||||||
|
String selectMemberPayload(@Param("dailyFileId") Long dailyFileId, @Param("resultId") Long resultId);
|
||||||
|
}
|
||||||
+13
@@ -0,0 +1,13 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopduplicatecheck.model.dto;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
|
|
||||||
|
/** 存量迁移数据源:每店最新已补偿每日文件(含店名/结果文件地址)。 */
|
||||||
|
public record ShopDailyFileMigrationDto(
|
||||||
|
Long dailyFileId,
|
||||||
|
LocalDate businessDate,
|
||||||
|
Long latestResultId,
|
||||||
|
String shopName,
|
||||||
|
String resultFileUrl,
|
||||||
|
Long taskId) {
|
||||||
|
}
|
||||||
+11
@@ -0,0 +1,11 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopduplicatecheck.model.dto;
|
||||||
|
|
||||||
|
/** 撞款扫描明细行(biz_shop_data_crawl_item 每店最新批次行)。 */
|
||||||
|
public record ShopLatestItemRowDto(
|
||||||
|
String shopName,
|
||||||
|
String country,
|
||||||
|
String asin,
|
||||||
|
String itemDate,
|
||||||
|
String price,
|
||||||
|
String brand) {
|
||||||
|
}
|
||||||
+175
-157
@@ -5,43 +5,47 @@ import com.fasterxml.jackson.databind.ObjectMapper;
|
|||||||
import com.nanri.aiimage.common.exception.BusinessException;
|
import com.nanri.aiimage.common.exception.BusinessException;
|
||||||
import com.nanri.aiimage.common.service.DistributedJobLockService;
|
import com.nanri.aiimage.common.service.DistributedJobLockService;
|
||||||
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo;
|
||||||
|
import com.nanri.aiimage.modules.shopdatacrawl.service.ShopDataCrawlItemStoreService;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDataDuplicateScanMapper;
|
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDataDuplicateScanMapper;
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDuplicateCheckItemMapper;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDuplicateCheckSourceMapper;
|
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDuplicateCheckSourceMapper;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ScanFullRowDto;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ScanFullRowDto;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ScanLightRowDto;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ScanLightRowDto;
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopDailyFileMigrationDto;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopGroupLabelDto;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopGroupLabelDto;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopSourceRowDto;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopLatestItemRowDto;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.entity.ShopDataDuplicateScanEntity;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.entity.ShopDataDuplicateScanEntity;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.payload.DuplicateScanPayload;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.payload.DuplicateScanPayload;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.model.payload.DuplicateScanSummary;
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.payload.DuplicateScanSummary;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckAggregator;
|
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckAggregator;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckWorkbookParser;
|
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.DuplicateCheckWorkbookParser;
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.RawRow;
|
||||||
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.ShopParsed;
|
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.ShopParsed;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.scheduling.annotation.Scheduled;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
import java.time.LocalDateTime;
|
import java.time.LocalDateTime;
|
||||||
import java.time.format.DateTimeFormatter;
|
import java.time.format.DateTimeFormatter;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.HashSet;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.LinkedHashSet;
|
import java.util.LinkedHashSet;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Locale;
|
import java.util.Locale;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.concurrent.ExecutionException;
|
|
||||||
import java.util.concurrent.ExecutorService;
|
|
||||||
import java.util.concurrent.Executors;
|
|
||||||
import java.util.concurrent.Future;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
|
||||||
import java.util.concurrent.atomic.AtomicLong;
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 撞款扫描:每日 00:00 定时 + force 同步全量重扫。逐店取最新结果文件、拉取并解析 xlsx、
|
* 撞款扫描:每日 00:00 定时 + force 同步全量重扫。
|
||||||
* 聚合成 shops/items/summary 落库 shop_data_duplicate_scan(格式与 Python 现行版本一致)。
|
* 数据源 = 采集明细表(biz_shop_data_crawl_item,采集先落库再更新文件),
|
||||||
|
* 直接查库聚合 shops/items/summary 落库 shop_data_duplicate_scan(输出契约不变)。
|
||||||
* 双实例通过 Redis 分布式锁防重;扫描失败落 FAILED 行并上抛,force 场景由端点转 409/500。
|
* 双实例通过 Redis 分布式锁防重;扫描失败落 FAILED 行并上抛,force 场景由端点转 409/500。
|
||||||
*/
|
*/
|
||||||
@Service
|
@Service
|
||||||
@@ -53,7 +57,6 @@ public class ShopDataDuplicateCheckScanService {
|
|||||||
|
|
||||||
private static final long MAX_RESULT_BYTES = 256L * 1024 * 1024;
|
private static final long MAX_RESULT_BYTES = 256L * 1024 * 1024;
|
||||||
private static final long MEM_CACHE_MAX_BYTES = 64L * 1024 * 1024;
|
private static final long MEM_CACHE_MAX_BYTES = 64L * 1024 * 1024;
|
||||||
private static final int PARSE_THREADS = 4;
|
|
||||||
private static final java.time.Duration SCAN_LOCK_TTL = java.time.Duration.ofHours(6);
|
private static final java.time.Duration SCAN_LOCK_TTL = java.time.Duration.ofHours(6);
|
||||||
|
|
||||||
private final ShopDataDuplicateScanMapper scanMapper;
|
private final ShopDataDuplicateScanMapper scanMapper;
|
||||||
@@ -61,6 +64,8 @@ public class ShopDataDuplicateCheckScanService {
|
|||||||
private final OssStorageService ossStorageService;
|
private final OssStorageService ossStorageService;
|
||||||
private final DistributedJobLockService distributedJobLockService;
|
private final DistributedJobLockService distributedJobLockService;
|
||||||
private final ObjectMapper objectMapper;
|
private final ObjectMapper objectMapper;
|
||||||
|
private final ShopDuplicateCheckItemMapper itemMapper;
|
||||||
|
private final ShopDataCrawlItemStoreService itemStoreService;
|
||||||
|
|
||||||
/** 进程内读缓存:轻量行 id 判活,行变化才全量读(对应 Python _duplicate_scan_memory)。 */
|
/** 进程内读缓存:轻量行 id 判活,行变化才全量读(对应 Python _duplicate_scan_memory)。 */
|
||||||
private final AtomicLong cachedRowId = new AtomicLong(-1L);
|
private final AtomicLong cachedRowId = new AtomicLong(-1L);
|
||||||
@@ -71,12 +76,16 @@ public class ShopDataDuplicateCheckScanService {
|
|||||||
ShopDuplicateCheckSourceMapper sourceMapper,
|
ShopDuplicateCheckSourceMapper sourceMapper,
|
||||||
OssStorageService ossStorageService,
|
OssStorageService ossStorageService,
|
||||||
DistributedJobLockService distributedJobLockService,
|
DistributedJobLockService distributedJobLockService,
|
||||||
ObjectMapper objectMapper) {
|
ObjectMapper objectMapper,
|
||||||
|
ShopDuplicateCheckItemMapper itemMapper,
|
||||||
|
ShopDataCrawlItemStoreService itemStoreService) {
|
||||||
this.scanMapper = scanMapper;
|
this.scanMapper = scanMapper;
|
||||||
this.sourceMapper = sourceMapper;
|
this.sourceMapper = sourceMapper;
|
||||||
this.ossStorageService = ossStorageService;
|
this.ossStorageService = ossStorageService;
|
||||||
this.distributedJobLockService = distributedJobLockService;
|
this.distributedJobLockService = distributedJobLockService;
|
||||||
this.objectMapper = objectMapper;
|
this.objectMapper = objectMapper;
|
||||||
|
this.itemMapper = itemMapper;
|
||||||
|
this.itemStoreService = itemStoreService;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** 读侧视图:scanned_at 为最新 SUCCESS 行 created_at(yyyy-MM-dd HH:mm:ss)。 */
|
/** 读侧视图:scanned_at 为最新 SUCCESS 行 created_at(yyyy-MM-dd HH:mm:ss)。 */
|
||||||
@@ -171,6 +180,8 @@ public class ShopDataDuplicateCheckScanService {
|
|||||||
|
|
||||||
private void runScanAndSave(DistributedJobLockService.LockHandle lock) {
|
private void runScanAndSave(DistributedJobLockService.LockHandle lock) {
|
||||||
try {
|
try {
|
||||||
|
// 存量 Excel 数据先同步进 MySQL 明细(幂等),之后扫描直接查库聚合。
|
||||||
|
migrateLegacyItemsOnce();
|
||||||
List<ShopParsed> parsedShops = collectShopParsed(lock);
|
List<ShopParsed> parsedShops = collectShopParsed(lock);
|
||||||
DuplicateCheckAggregator.Aggregate aggregate =
|
DuplicateCheckAggregator.Aggregate aggregate =
|
||||||
DuplicateCheckAggregator.aggregate(parsedShops, "job");
|
DuplicateCheckAggregator.aggregate(parsedShops, "job");
|
||||||
@@ -215,183 +226,190 @@ public class ShopDataDuplicateCheckScanService {
|
|||||||
return message == null || message.isBlank() ? ex.getClass().getSimpleName() : message;
|
return message == null || message.isBlank() ? ex.getClass().getSimpleName() : message;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** 逐店并发拉取/解析结果文件;单店失败仅告警跳过,不中断整体。
|
/**
|
||||||
* 扫描基线与店铺表全量对齐:无结果文件的店以空态(0 记录)注入,保证撞款控制台可见全部店铺。 */
|
* 撞款扫描数据源:直接查采集明细表(biz_shop_data_crawl_item,每店最新 business_date 批次)。
|
||||||
|
* 用户约定:不再解析 OSS xlsx(旧 4 家店因 xlsx 格式差异被静默丢弃的根因),
|
||||||
|
* 重复检查基数 = 已采集落库的店(超管应见 8 家)。
|
||||||
|
*/
|
||||||
private List<ShopParsed> collectShopParsed(DistributedJobLockService.LockHandle lock) {
|
private List<ShopParsed> collectShopParsed(DistributedJobLockService.LockHandle lock) {
|
||||||
List<ShopSourceRowDto> rows = sourceMapper.selectLatestResultRows();
|
List<ShopLatestItemRowDto> rows = itemMapper.selectLatestShopItems();
|
||||||
List<ShopParsed> parsedShops = new ArrayList<>();
|
|
||||||
if (rows.isEmpty()) {
|
if (rows.isEmpty()) {
|
||||||
|
log.info("[shop-duplicate-check] 明细表暂无采集数据,本次扫描不出库");
|
||||||
|
return List.of();
|
||||||
|
}
|
||||||
|
// 店铺名去重(分组标签来自店铺管理 biz_shop_manage,与页面口径一致)。
|
||||||
|
Set<String> shopNames = new LinkedHashSet<>();
|
||||||
|
for (ShopLatestItemRowDto row : rows) {
|
||||||
|
if (row != null) {
|
||||||
|
shopNames.add(normalizeShopName(row.shopName()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Map<String, String> groupLabels = new HashMap<>();
|
||||||
|
for (ShopGroupLabelDto dto : sourceMapper.selectGroupLabels(new ArrayList<>(shopNames))) {
|
||||||
|
groupLabels.put(shopKey(dto.getShopName()), dto.getGroupName() == null ? "" : dto.getGroupName());
|
||||||
|
}
|
||||||
|
List<ShopParsed> parsedShops = buildParsedShops(rows, groupLabels);
|
||||||
|
log.info("[shop-duplicate-check] 明细数据源扫描:{} 家店铺,{} 行明细", parsedShops.size(), rawRowsCount(parsedShops));
|
||||||
return parsedShops;
|
return parsedShops;
|
||||||
}
|
}
|
||||||
Map<String, String> groupLabels = loadGroupLabels(rows);
|
|
||||||
Set<String> parsedNames = new HashSet<>();
|
/** 明细行 → 按店 ShopParsed(纯函数,便于单测):空 ASIN 行过滤、国家去重大写、分组标签注入。 */
|
||||||
ExecutorService executor = Executors.newFixedThreadPool(Math.min(PARSE_THREADS, rows.size()));
|
static List<ShopParsed> buildParsedShops(List<ShopLatestItemRowDto> rows, Map<String, String> groupLabels) {
|
||||||
try {
|
Map<String, List<ShopLatestItemRowDto>> byShop = new LinkedHashMap<>();
|
||||||
List<Future<ShopParsed>> futures = new ArrayList<>(rows.size());
|
for (ShopLatestItemRowDto row : rows) {
|
||||||
for (ShopSourceRowDto row : rows) {
|
if (row == null) {
|
||||||
futures.add(executor.submit(() -> parseShop(row, groupLabels)));
|
|
||||||
}
|
|
||||||
for (Future<ShopParsed> future : futures) {
|
|
||||||
if (lock != null) {
|
|
||||||
lock.renew(SCAN_LOCK_TTL);
|
|
||||||
}
|
|
||||||
try {
|
|
||||||
ShopParsed item = future.get();
|
|
||||||
if (item != null) {
|
|
||||||
parsedShops.add(item);
|
|
||||||
parsedNames.add(normalizeShopName(item.shopName()));
|
|
||||||
}
|
|
||||||
} catch (ExecutionException ex) {
|
|
||||||
log.warn("[shop-duplicate-check] 并发解析店铺结果文件异常: {}", ex.getCause() == null
|
|
||||||
? ex.getMessage() : ex.getCause().getMessage());
|
|
||||||
} catch (InterruptedException ex) {
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
log.warn("[shop-duplicate-check] 解析线程被中断");
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} finally {
|
|
||||||
executor.shutdown();
|
|
||||||
try {
|
|
||||||
if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
|
|
||||||
executor.shutdownNow();
|
|
||||||
}
|
|
||||||
} catch (InterruptedException ex) {
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
executor.shutdownNow();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// 基线扩展:店铺表全量中尚未解析到结果文件行的店 → 空态注入(0 记录、分组标签来自店铺表)。
|
|
||||||
List<ShopGroupLabelDto> allShops = sourceMapper.selectAllShopLabels();
|
|
||||||
int injectedEmptyShops = 0;
|
|
||||||
for (ShopGroupLabelDto all : allShops) {
|
|
||||||
String name = normalizeShopName(all.getShopName());
|
|
||||||
if (name.isEmpty() || parsedNames.contains(name)) {
|
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
String groupName = all.getGroupName() == null ? "" : all.getGroupName();
|
String name = row.shopName() == null ? "未命名" : (row.shopName().trim().isEmpty() ? "未命名" : row.shopName().trim());
|
||||||
parsedShops.add(new ShopParsed(name, groupName, List.of(), List.of()));
|
byShop.computeIfAbsent(name, k -> new ArrayList<>()).add(row);
|
||||||
injectedEmptyShops++;
|
|
||||||
}
|
}
|
||||||
if (injectedEmptyShops > 0) {
|
Map<String, String> safeGroups = groupLabels == null ? Map.of() : groupLabels;
|
||||||
log.info("[shop-duplicate-check] 扫描基线含空态店铺 {} 家(无结果文件,以 0 记录参与撞款视图)", injectedEmptyShops);
|
List<ShopParsed> parsedShops = new ArrayList<>();
|
||||||
|
for (Map.Entry<String, List<ShopLatestItemRowDto>> entry : byShop.entrySet()) {
|
||||||
|
String shopName = entry.getKey();
|
||||||
|
List<RawRow> rawRows = new ArrayList<>();
|
||||||
|
Set<String> countries = new LinkedHashSet<>();
|
||||||
|
for (ShopLatestItemRowDto row : entry.getValue()) {
|
||||||
|
if (blank(row.asin())) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
String country = row.country() == null ? "" : row.country().trim().toUpperCase(Locale.ROOT);
|
||||||
|
rawRows.add(new RawRow(
|
||||||
|
row.asin().trim(),
|
||||||
|
nvl(row.itemDate()),
|
||||||
|
nvl(row.price()),
|
||||||
|
nvl(row.brand()),
|
||||||
|
country));
|
||||||
|
if (!country.isEmpty()) {
|
||||||
|
countries.add(country);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
parsedShops.add(new ShopParsed(shopName,
|
||||||
|
safeGroups.getOrDefault(shopKey(shopName), ""),
|
||||||
|
new ArrayList<>(countries),
|
||||||
|
rawRows));
|
||||||
}
|
}
|
||||||
return parsedShops;
|
return parsedShops;
|
||||||
}
|
}
|
||||||
|
|
||||||
private Map<String, String> loadGroupLabels(List<ShopSourceRowDto> rows) {
|
private static int rawRowsCount(List<ShopParsed> shops) {
|
||||||
Set<String> names = new LinkedHashSet<>();
|
int total = 0;
|
||||||
for (ShopSourceRowDto row : rows) {
|
for (ShopParsed shop : shops) {
|
||||||
String name = normalizeShopName(row.getShopName());
|
total += shop.rows().size();
|
||||||
if (!name.isEmpty()) {
|
|
||||||
names.add(name);
|
|
||||||
}
|
}
|
||||||
}
|
return total;
|
||||||
Map<String, String> labels = new HashMap<>();
|
|
||||||
if (names.isEmpty()) {
|
|
||||||
return labels;
|
|
||||||
}
|
|
||||||
for (ShopGroupLabelDto dto : sourceMapper.selectGroupLabels(new ArrayList<>(names))) {
|
|
||||||
labels.put(shopKey(dto.getShopName()), dto.getGroupName() == null ? "" : dto.getGroupName());
|
|
||||||
}
|
|
||||||
return labels;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private ShopParsed parseShop(ShopSourceRowDto row, Map<String, String> groupLabels) {
|
/**
|
||||||
long resultId = row.getResultId() == null ? 0L : row.getResultId();
|
* 存量数据迁移(幂等):每店最新每日文件的采集数据 → 行级明细表。
|
||||||
if (resultId <= 0) {
|
* 优先 daily_member.row_payload(快照 JSON,无需拉 OSS);老归档无 payload 时
|
||||||
return null;
|
* 回退解析 OSS xlsx(宽松表头),保证 8 家历史店全部入库。
|
||||||
|
*/
|
||||||
|
private void migrateLegacyItemsOnce() {
|
||||||
|
List<ShopDailyFileMigrationDto> files = itemMapper.selectLegacyMigrationFiles();
|
||||||
|
if (files.isEmpty()) {
|
||||||
|
log.info("[shop-data-crawl-item] 存量明细迁移:无待迁移每日文件");
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
String shopName = normalizeShopName(row.getShopName());
|
int migrated = 0;
|
||||||
if (row.getResultFileUrl() == null || row.getResultFileUrl().isBlank()) {
|
int skipped = 0;
|
||||||
return null;
|
int failed = 0;
|
||||||
|
for (ShopDailyFileMigrationDto file : files) {
|
||||||
|
LocalDate businessDate = file.businessDate();
|
||||||
|
String shopName = normalizeShopName(file.shopName());
|
||||||
|
if (businessDate == null || file.latestResultId() == null) {
|
||||||
|
log.warn("[shop-data-crawl-item] 存量迁移跳过(缺日期/结果行) shop={}", shopName);
|
||||||
|
failed++;
|
||||||
|
continue;
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
if (row.getResultFileSize() != null && row.getResultFileSize() > MAX_RESULT_BYTES) {
|
if (itemMapper.countShopBatch(shopName, businessDate) > 0) {
|
||||||
log.warn("[shop-data-crawl] 结果文件过大跳过 result_id={} size={}", resultId, row.getResultFileSize());
|
skipped++;
|
||||||
return null;
|
continue;
|
||||||
}
|
}
|
||||||
byte[] bytes;
|
List<ShopDataCrawlResultItemVo> snapshots = buildLegacySnapshot(file);
|
||||||
try {
|
if (snapshots == null) {
|
||||||
bytes = ossStorageService.readObjectBytesBounded(row.getResultFileUrl(), MAX_RESULT_BYTES);
|
log.warn("[shop-data-crawl-item] 存量迁移无数据源 shop={} date={}(无快照且文件不可解析)",
|
||||||
} catch (IllegalArgumentException ex) {
|
shopName, businessDate);
|
||||||
log.warn("[shop-data-crawl] 结果文件过大跳过 result_id={}: {}", resultId, ex.getMessage());
|
failed++;
|
||||||
return null;
|
continue;
|
||||||
}
|
}
|
||||||
List<com.nanri.aiimage.modules.shopduplicatecheck.service.support.RawRow> parsed;
|
itemStoreService.saveShopBatchFromSnapshot(shopName, businessDate, snapshots,
|
||||||
try {
|
file.latestResultId(), file.taskId(), file.dailyFileId());
|
||||||
parsed = DuplicateCheckWorkbookParser.parse(bytes);
|
migrated++;
|
||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
log.warn("[shop-data-crawl] 解析结果文件失败 result_id={}: {}", resultId,
|
log.error("[shop-data-crawl-item] 存量迁移失败 shop={} date={} msg={}",
|
||||||
ex.getMessage() == null ? ex.toString() : ex.getMessage());
|
shopName, businessDate, ex.getMessage());
|
||||||
return null;
|
failed++;
|
||||||
}
|
}
|
||||||
List<String> countryCodes = parseCountryCodesJson(row.getCountryCodesJson());
|
|
||||||
if (countryCodes.isEmpty()) {
|
|
||||||
countryCodes = parseCountryCodesRequest(row.getRequestJson());
|
|
||||||
}
|
}
|
||||||
String groupName = groupLabels.getOrDefault(shopKey(shopName), "");
|
log.info("[shop-data-crawl-item] 存量明细迁移完成:迁移 {} 家,跳过 {} 家(已有明细),失败 {} 家",
|
||||||
return new ShopParsed(shopName, groupName, countryCodes, parsed);
|
migrated, skipped, failed);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 构造迁移快照:row_payload 优先,回退 OSS xlsx 宽松解析;均不可用返回 null。 */
|
||||||
|
private List<ShopDataCrawlResultItemVo> buildLegacySnapshot(ShopDailyFileMigrationDto file) throws Exception {
|
||||||
|
String payload = itemMapper.selectMemberPayload(file.dailyFileId(), file.latestResultId());
|
||||||
|
if (payload != null && !payload.isBlank()) {
|
||||||
|
try {
|
||||||
|
return List.of(objectMapper.readValue(payload, ShopDataCrawlResultItemVo.class));
|
||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
log.warn("[shop-data-crawl] 拉取结果文件失败 result_id={}: {}", resultId,
|
log.warn("[shop-data-crawl-item] 迁移快照 JSON 解析失败,回退文件解析 shop={} date={} msg={}",
|
||||||
ex.getMessage() == null ? ex.toString() : ex.getMessage());
|
file.shopName(), file.businessDate(), ex.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (file.resultFileUrl() == null || file.resultFileUrl().isBlank()) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
byte[] bytes = ossStorageService.readObjectBytesBounded(file.resultFileUrl(), MAX_RESULT_BYTES);
|
||||||
|
List<RawRow> rawRows = DuplicateCheckWorkbookParser.parse(bytes);
|
||||||
|
return List.of(rawRowsToSnapshot(file, rawRows));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** RawRow(宽松解析结果)→ 快照 VO(countryResults 按国家聚合)。 */
|
||||||
|
private ShopDataCrawlResultItemVo rawRowsToSnapshot(ShopDailyFileMigrationDto file, List<RawRow> rawRows) {
|
||||||
|
ShopDataCrawlResultItemVo vo = new ShopDataCrawlResultItemVo();
|
||||||
|
vo.setShopName(normalizeShopName(file.shopName()));
|
||||||
|
vo.setSuccess(true);
|
||||||
|
vo.setResultId(file.latestResultId());
|
||||||
|
vo.setTaskId(file.taskId());
|
||||||
|
Map<String, List<ShopDataCrawlRowDto>> byCountry = new LinkedHashMap<>();
|
||||||
|
for (RawRow row : rawRows) {
|
||||||
|
if (row == null || blank(row.asin())) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
ShopDataCrawlRowDto dto = new ShopDataCrawlRowDto();
|
||||||
|
dto.setAsin(row.asin().trim());
|
||||||
|
dto.setDate(nvl(row.date()));
|
||||||
|
dto.setPrice(nvl(row.price()));
|
||||||
|
dto.setBrand(nvl(row.brand()));
|
||||||
|
String country = ShopDataCrawlItemStoreService.normalizeCountry(row.country());
|
||||||
|
byCountry.computeIfAbsent(country, k -> new ArrayList<>()).add(dto);
|
||||||
|
}
|
||||||
|
List<ShopDataCrawlCountryResultDto> countryResults = new ArrayList<>();
|
||||||
|
for (Map.Entry<String, List<ShopDataCrawlRowDto>> entry : byCountry.entrySet()) {
|
||||||
|
ShopDataCrawlCountryResultDto cr = new ShopDataCrawlCountryResultDto();
|
||||||
|
cr.setCountry(entry.getKey());
|
||||||
|
cr.setItems(entry.getValue());
|
||||||
|
countryResults.add(cr);
|
||||||
|
}
|
||||||
|
vo.setCountryResults(countryResults);
|
||||||
|
return vo;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static boolean blank(String value) {
|
||||||
|
return value == null || value.isBlank();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String nvl(String value) {
|
||||||
|
return value == null ? "" : value.trim();
|
||||||
}
|
}
|
||||||
|
|
||||||
private String normalizeShopName(String raw) {
|
private String normalizeShopName(String raw) {
|
||||||
return raw == null ? "未命名" : (raw.trim().isEmpty() ? "未命名" : raw.trim());
|
return raw == null ? "未命名" : (raw.trim().isEmpty() ? "未命名" : raw.trim());
|
||||||
}
|
}
|
||||||
|
|
||||||
private String shopKey(String name) {
|
private static String shopKey(String name) {
|
||||||
return (name == null ? "" : name.trim()).toLowerCase(Locale.ROOT);
|
return (name == null ? "" : name.trim()).toLowerCase(Locale.ROOT);
|
||||||
}
|
}
|
||||||
|
|
||||||
private List<String> parseCountryCodesJson(String jsonValue) {
|
|
||||||
if (jsonValue == null || jsonValue.isBlank()) {
|
|
||||||
return List.of();
|
|
||||||
}
|
|
||||||
try {
|
|
||||||
JsonNode node = objectMapper.readTree(jsonValue);
|
|
||||||
if (node != null && node.isArray()) {
|
|
||||||
return normalizeCodes(node);
|
|
||||||
}
|
|
||||||
} catch (Exception ex) {
|
|
||||||
log.warn("[shop-data-crawl] 解析 country_codes_json 失败: {}", ex.getMessage());
|
|
||||||
}
|
|
||||||
return List.of();
|
|
||||||
}
|
|
||||||
|
|
||||||
private List<String> parseCountryCodesRequest(String requestJson) {
|
|
||||||
if (requestJson == null || requestJson.isBlank()) {
|
|
||||||
return List.of();
|
|
||||||
}
|
|
||||||
try {
|
|
||||||
JsonNode node = objectMapper.readTree(requestJson);
|
|
||||||
if (node == null || !node.isObject()) {
|
|
||||||
return List.of();
|
|
||||||
}
|
|
||||||
JsonNode raw = node.get("countryCodes");
|
|
||||||
if (raw == null) {
|
|
||||||
raw = node.get("country_codes");
|
|
||||||
}
|
|
||||||
if (raw != null && raw.isArray()) {
|
|
||||||
return normalizeCodes(raw);
|
|
||||||
}
|
|
||||||
} catch (Exception ex) {
|
|
||||||
log.warn("[shop-data-crawl] 解析 request_json 国家代码失败: {}", ex.getMessage());
|
|
||||||
}
|
|
||||||
return List.of();
|
|
||||||
}
|
|
||||||
|
|
||||||
private List<String> normalizeCodes(JsonNode array) {
|
|
||||||
List<String> codes = new ArrayList<>();
|
|
||||||
for (JsonNode item : array) {
|
|
||||||
String value = item.asText();
|
|
||||||
if (value != null && !value.trim().isEmpty()) {
|
|
||||||
codes.add(value.trim().toUpperCase(Locale.ROOT));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return codes;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
+7
-1
@@ -68,8 +68,9 @@ public final class DuplicateCheckWorkbookParser {
|
|||||||
}
|
}
|
||||||
for (Map.Entry<Integer, String> entry : headerMap.entrySet()) {
|
for (Map.Entry<Integer, String> entry : headerMap.entrySet()) {
|
||||||
String header = cell(entry.getValue());
|
String header = cell(entry.getValue());
|
||||||
|
String normalized = header.replaceAll("\\s+", "").toUpperCase();
|
||||||
int index = entry.getKey();
|
int index = entry.getKey();
|
||||||
if ("ASIN".equals(header)) {
|
if (isAsinHeader(normalized)) {
|
||||||
asinCol = index;
|
asinCol = index;
|
||||||
} else if ("日期".equals(header)) {
|
} else if ("日期".equals(header)) {
|
||||||
dateCol = index;
|
dateCol = index;
|
||||||
@@ -81,6 +82,11 @@ public final class DuplicateCheckWorkbookParser {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** 表头宽松匹配(兼容旧版/大小写/「ASIN码」类变体):含 ASIN 即视为 ASIN 列。 */
|
||||||
|
private static boolean isAsinHeader(String normalized) {
|
||||||
|
return normalized.contains("ASIN");
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void onRow(String sheetName, Integer sheetNo, int rowIndex,
|
public void onRow(String sheetName, Integer sheetNo, int rowIndex,
|
||||||
Map<Integer, String> headerMap, Map<Integer, String> rowMap) {
|
Map<Integer, String> headerMap, Map<Integer, String> rowMap) {
|
||||||
|
|||||||
@@ -0,0 +1,25 @@
|
|||||||
|
-- 店铺数据采集行级明细表:采集数据先落库再更新文件,撞款重复检查直接查库(不再解析 Excel)。
|
||||||
|
-- 语义:每店×每日一份快照(countryResults 展开为行);撞款扫描取每家店最新 business_date 的明细。
|
||||||
|
CREATE TABLE IF NOT EXISTS biz_shop_data_crawl_item (
|
||||||
|
id BIGINT NOT NULL AUTO_INCREMENT,
|
||||||
|
shop_id BIGINT NULL COMMENT '店铺 ID(biz_shop_manage.id)',
|
||||||
|
shop_name VARCHAR(255) NOT NULL COMMENT '规范化店铺名(与结果文件 source_filename 一致)',
|
||||||
|
group_name VARCHAR(255) NULL COMMENT '归档时店铺分组名快照(多分组「、」连接)',
|
||||||
|
business_date DATE NOT NULL COMMENT '采集批次日期(每日文件日期)',
|
||||||
|
country VARCHAR(8) NOT NULL COMMENT '国家站点码(UK/DE/FR/ES/IT,展示层统一中文)',
|
||||||
|
asin VARCHAR(32) NOT NULL COMMENT 'ASIN',
|
||||||
|
brand VARCHAR(512) NULL COMMENT '品牌',
|
||||||
|
price VARCHAR(64) NULL COMMENT '价格(原始文本)',
|
||||||
|
item_date VARCHAR(64) NULL COMMENT '原「日期」列原文(如 2026年8月15日 上午10:59)',
|
||||||
|
units_sold VARCHAR(64) NULL COMMENT '售出件数(原始文本)',
|
||||||
|
item_json JSON NULL COMMENT '其余列原文兜底(库存销量/销售排名/浏览量/商品图等)',
|
||||||
|
result_id BIGINT NULL COMMENT '来源结果文件行 id(biz_file_result.id)',
|
||||||
|
task_id BIGINT NULL COMMENT '来源任务 id(biz_file_task.id)',
|
||||||
|
daily_file_id BIGINT NULL COMMENT '来源每日文件行 id(biz_shop_data_crawl_daily_file.id)',
|
||||||
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
UNIQUE KEY uk_shop_data_crawl_item_batch (shop_name, business_date, country, asin),
|
||||||
|
KEY idx_shop_data_crawl_item_asin (asin),
|
||||||
|
KEY idx_shop_data_crawl_item_shop_date (shop_name, business_date),
|
||||||
|
KEY idx_shop_data_crawl_item_biz_date (business_date)
|
||||||
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='店铺数据采集行级明细(快照批次)';
|
||||||
+1
@@ -136,6 +136,7 @@ class ShopDataCrawlChunkUpsertTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -172,6 +172,7 @@ class ShopDataCrawlCleanupTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 30L);
|
ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 30L);
|
||||||
|
|||||||
+1
@@ -145,6 +145,7 @@ class ShopDataCrawlDailyFileIncrementalTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -151,6 +151,7 @@ class ShopDataCrawlDailyFileJobSplitTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -151,6 +151,7 @@ class ShopDataCrawlDailyFileLockTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -133,6 +133,7 @@ class ShopDataCrawlLightweightProgressTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -132,6 +132,7 @@ class ShopDataCrawlOwnerColumnTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -145,6 +145,7 @@ class ShopDataCrawlProgressQueryTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -132,6 +132,7 @@ class ShopDataCrawlRowDedupKeyTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -133,6 +133,7 @@ class ShopDataCrawlScopeCounterTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -134,6 +134,7 @@ class ShopDataCrawlScopeMergeTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+1
@@ -128,6 +128,7 @@ class ShopDataCrawlTaskServiceChunkTest {
|
|||||||
transientPayloadStorageService,
|
transientPayloadStorageService,
|
||||||
instanceMetadata,
|
instanceMetadata,
|
||||||
dailyFileService,
|
dailyFileService,
|
||||||
|
mock(ShopDataCrawlItemStoreService.class),
|
||||||
null,
|
null,
|
||||||
mock(TaskProgressLightAssembler.class));
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
|
||||||
|
|||||||
+52
@@ -0,0 +1,52 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopduplicatecheck.service;
|
||||||
|
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ShopLatestItemRowDto;
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.RawRow;
|
||||||
|
import com.nanri.aiimage.modules.shopduplicatecheck.service.support.ShopParsed;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 撞款明细数据源(biz_shop_data_crawl_item)→ ShopParsed 转换:
|
||||||
|
* 按店分组、空 ASIN 行过滤、国家去重大写、分组标签注入。
|
||||||
|
*/
|
||||||
|
class ShopDataDuplicateCheckScanServiceTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void buildParsedShops_groupsByShopAndNormalizesCountry() {
|
||||||
|
List<ShopLatestItemRowDto> rows = List.of(
|
||||||
|
new ShopLatestItemRowDto("店A", "uk", "B0001", "2026-09-01 10:00", "9.90", "品牌甲"),
|
||||||
|
new ShopLatestItemRowDto("店A", "DE", "B0001", "2026-09-01 11:00", "9.90", "品牌甲"),
|
||||||
|
new ShopLatestItemRowDto("店B", "FR", "B0002", "2026-09-02 09:00", "19.9", "品牌乙"),
|
||||||
|
new ShopLatestItemRowDto("店B", "FR", "", "2026-09-02 09:01", "1.00", "空ASIN行"),
|
||||||
|
new ShopLatestItemRowDto(null, "UK", "B0003", "", "", ""));
|
||||||
|
List<ShopParsed> parsed = ShopDataDuplicateCheckScanService.buildParsedShops(
|
||||||
|
rows, Map.of("店b", "华东组"));
|
||||||
|
|
||||||
|
assertEquals(3, parsed.size(), "按店分组(含未命名兜底)");
|
||||||
|
ShopParsed shopA = parsed.stream().filter(s -> s.shopName().equals("店A")).findFirst().orElseThrow();
|
||||||
|
assertEquals(List.of("UK", "DE"), shopA.countryCodes(), "国家码去重大写");
|
||||||
|
assertEquals(2, shopA.rows().size());
|
||||||
|
RawRow ukRow = shopA.rows().get(0);
|
||||||
|
assertEquals("B0001", ukRow.asin());
|
||||||
|
assertEquals("UK", ukRow.country());
|
||||||
|
|
||||||
|
ShopParsed shopB = parsed.stream().filter(s -> s.shopName().equals("店B")).findFirst().orElseThrow();
|
||||||
|
assertEquals(1, shopB.rows().size(), "空 ASIN 行过滤");
|
||||||
|
assertEquals("华东组", shopB.groupName(), "分组标签注入(shopKey 大小写不敏感)");
|
||||||
|
|
||||||
|
ShopParsed unnamed = parsed.stream().filter(s -> s.shopName().equals("未命名")).findFirst().orElseThrow();
|
||||||
|
assertEquals(1, unnamed.rows().size(), "无店铺名兜底未命名");
|
||||||
|
assertEquals("", unnamed.rows().get(0).brand(), "缺省字段空串兜底");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void buildParsedShops_emptyInputReturnsEmpty() {
|
||||||
|
assertTrue(ShopDataDuplicateCheckScanService.buildParsedShops(List.of(), Map.of()).isEmpty());
|
||||||
|
}
|
||||||
|
}
|
||||||
+78
@@ -0,0 +1,78 @@
|
|||||||
|
package com.nanri.aiimage.modules.shopduplicatecheck.service.support;
|
||||||
|
|
||||||
|
import org.apache.poi.xssf.streaming.SXSSFWorkbook;
|
||||||
|
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import java.io.ByteArrayOutputStream;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 宽松表头解析(兼容旧版文件):ASIN 列大小写/空白/「ASIN码」类变体均可识别,
|
||||||
|
* 是存量 Excel 数据迁移入库的前提(旧版 4 家店因表头差异被整店丢弃的根因)。
|
||||||
|
*/
|
||||||
|
class DuplicateCheckWorkbookParserTest {
|
||||||
|
|
||||||
|
private static byte[] workbookWith(boolean stream, String sheetName, String[] headers, String[] values) throws Exception {
|
||||||
|
try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
|
||||||
|
if (stream) {
|
||||||
|
SXSSFWorkbook workbook = new SXSSFWorkbook();
|
||||||
|
write(workbook, sheetName, headers, values);
|
||||||
|
workbook.write(out);
|
||||||
|
workbook.dispose();
|
||||||
|
} else {
|
||||||
|
XSSFWorkbook workbook = new XSSFWorkbook();
|
||||||
|
write(workbook, sheetName, headers, values);
|
||||||
|
workbook.write(out);
|
||||||
|
workbook.close();
|
||||||
|
}
|
||||||
|
return out.toByteArray();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void write(org.apache.poi.ss.usermodel.Workbook workbook, String sheetName,
|
||||||
|
String[] headers, String[] values) {
|
||||||
|
org.apache.poi.ss.usermodel.Sheet sheet = workbook.createSheet(sheetName);
|
||||||
|
org.apache.poi.ss.usermodel.Row header = sheet.createRow(0);
|
||||||
|
for (int i = 0; i < headers.length; i++) {
|
||||||
|
header.createCell(i).setCellValue(headers[i]);
|
||||||
|
}
|
||||||
|
org.apache.poi.ss.usermodel.Row row = sheet.createRow(1);
|
||||||
|
for (int i = 0; i < values.length; i++) {
|
||||||
|
row.createCell(i).setCellValue(values[i]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void parsesLegacyAsinCodeHeader() throws Exception {
|
||||||
|
byte[] bytes = workbookWith(false, "英国",
|
||||||
|
new String[]{"ASIN码", "日期", "价格", "品牌"},
|
||||||
|
new String[]{"B0OLD001", "2026年8月15日 上午10:59", "£34.46", "DOLGABA"});
|
||||||
|
var rows = DuplicateCheckWorkbookParser.parse(bytes);
|
||||||
|
assertEquals(1, rows.size(), "「ASIN码」表头变体可识别");
|
||||||
|
assertEquals("B0OLD001", rows.get(0).asin());
|
||||||
|
assertEquals("UK", rows.get(0).country(), "中文 sheet 名映射 UK");
|
||||||
|
assertEquals("2026年8月15日 上午10:59", rows.get(0).date());
|
||||||
|
assertEquals("DOLGABA", rows.get(0).brand());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void parsesLowercaseAsinHeader() throws Exception {
|
||||||
|
byte[] bytes = workbookWith(true, "asin",
|
||||||
|
new String[]{"asin", "date", "price", "brand"},
|
||||||
|
new String[]{"B0LOW001", "2026-08-15 10:59", "9.99", "X"});
|
||||||
|
var rows = DuplicateCheckWorkbookParser.parse(bytes);
|
||||||
|
assertEquals(1, rows.size(), "小写 asin 表头可识别");
|
||||||
|
assertEquals("B0LOW001", rows.get(0).asin());
|
||||||
|
assertEquals(rows.get(0).country(), "", "未知 sheet 名不丢行(国家为空)");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void skipsSheetWithoutAsinColumn() throws Exception {
|
||||||
|
byte[] bytes = workbookWith(false, "德国",
|
||||||
|
new String[]{"日期", "价格"},
|
||||||
|
new String[]{"2026-08-15", "9.99"});
|
||||||
|
assertEquals(0, DuplicateCheckWorkbookParser.parse(bytes).size(), "无 ASIN 列的 sheet 跳过");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user