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

task-35

Browse Source
This commit is contained in:
黄自达
2026-08-29 21:28:28 +08:00
parent add8b057d5
commit 2f89f795f6
6 changed files with 687 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyMemberEntity.java
+2
View File
@@ -16,5 +16,7 @@ public class ShopDataCrawlDailyMemberEntity {
private Long dailyFileId;
private Long taskId;
private Long resultId;
/** 结果快照 JSON（数据层增量模型：整表重建时按成员行累积，不再读回旧累计对象）。 */
private String rowPayload;
private LocalDateTime createdAt;
}
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java
+17 -1
View File
@@ -138,7 +138,7 @@ public class ShopDataCrawlDailyFileService {
.eq(ShopDataCrawlDailyMemberEntity::getResultId, resultId)) > 0;
}
public boolean addMember(Long dailyFileId, Long taskId, Long resultId) {
public boolean addMemberWithPayload(Long dailyFileId, Long taskId, Long resultId, String rowPayload) {
if (dailyFileId == null || dailyFileId <= 0 || taskId == null || taskId <= 0
|| resultId == null || resultId <= 0) {
return false;
@@ -147,6 +147,7 @@ public class ShopDataCrawlDailyFileService {
member.setDailyFileId(dailyFileId);
member.setTaskId(taskId);
member.setResultId(resultId);
member.setRowPayload(rowPayload);
member.setCreatedAt(currentBusinessDateTime());
try {
dailyMemberMapper.insert(member);
@@ -156,6 +157,10 @@ public class ShopDataCrawlDailyFileService {
}
}
public boolean addMember(Long dailyFileId, Long taskId, Long resultId) {
return addMemberWithPayload(dailyFileId, taskId, resultId, null);
}
public List<ShopDataCrawlDailyMemberEntity> listMembers(Long dailyFileId) {
if (dailyFileId == null || dailyFileId <= 0) {
return List.of();
@@ -183,6 +188,17 @@ public class ShopDataCrawlDailyFileService {
.in(ShopDataCrawlDailyMemberEntity::getResultId, resultIds));
}
/** 累计文件跨天滚动时把旧文件成员行（含 row_payload）迁移到新文件，保留跨天携带的行。 */
public void reassignMembers(Long fromDailyFileId, Long toDailyFileId) {
if (fromDailyFileId == null || fromDailyFileId <= 0 || toDailyFileId == null || toDailyFileId <= 0) {
return;
}
ShopDataCrawlDailyMemberEntity update = new ShopDataCrawlDailyMemberEntity();
update.setDailyFileId(toDailyFileId);
dailyMemberMapper.update(update, new LambdaQueryWrapper<ShopDataCrawlDailyMemberEntity>()
.eq(ShopDataCrawlDailyMemberEntity::getDailyFileId, fromDailyFileId));
}
public long countObjectReferences(String objectKey) {
if (objectKey == null || objectKey.isBlank()) {
return 0L;
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java
+91 -20
View File
@@ -1762,7 +1762,7 @@ public class ShopDataCrawlTaskService {
try {
persistedResult = executeShortTransaction(() -> persistDailyAggregation(
task, row, userId, shopKey, shopKeyHash, businessDate,
preparation, artifact));
preparation, artifact, snapshot));
if (persistedResult.discardUploadedObject() && artifact.uploaded()) {
deleteObjectQuietly(artifact.objectKey());
}
@@ -1832,35 +1832,103 @@ public class ShopDataCrawlTaskService {
"shop-data-crawl-result",
String.valueOf(task.getId()),
"daily-" + UUID.randomUUID()));
File baseXlsx = FileUtil.file(workRoot, "base.xlsx");
File outputXlsx = FileUtil.file(workRoot, filename);
try {
if (baseDailyFile != null && !blank(existingObjectKey)) {
try {
Files.write(baseXlsx.toPath(), ossStorageService.readObjectBytes(existingObjectKey));
} catch (Exception ex) {
throw new BusinessException("读取累计文件失败: " + safeMessage(ex));
}
int rowCount = excelAssemblyService.replaceCountriesWorkbook(baseXlsx, outputXlsx, List.of(snapshot));
String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE);
if (blank(objectKey)) {
throw new BusinessException("累计文件上传后未返回文件地址");
}
return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, rowCount);
}
excelAssemblyService.writeWorkbook(outputXlsx, List.of(snapshot));
// Task 35：数据层增量模型。整表从成员行（row_payload）累积重建，
// 不再读回旧累计对象（readObjectBytes）并整表重写（replaceCountriesWorkbook）；
// 历史成员行无 payload 时按结果快照兜底，兼容旧归档数据。
List<ShopDataCrawlResultItemVo> accumulatedItems = buildDailyFileFromData(baseDailyFile, List.of(snapshot));
int rowCount = excelAssemblyService.writeWorkbook(outputXlsx, accumulatedItems);
String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE);
if (blank(objectKey)) {
throw new BusinessException("累计文件上传后未返回文件地址");
}
return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, addedRowCount);
return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, rowCount);
} finally {
FileUtil.del(baseXlsx);
FileUtil.del(outputXlsx);
FileUtil.del(workRoot);
}
}
/**
* 从数据层累积重建每日累计文件的快照列表：既有成员行按 (createdAt, id) 升序读取，
* 优先用行级 payload；payload 缺失（历史数据）时按结果快照兜底；新结果追加在末尾。
*/
private List<ShopDataCrawlResultItemVo> buildDailyFileFromData(ShopDataCrawlDailyFileEntity baseDailyFile,
List<ShopDataCrawlResultItemVo> appended) {
List<ShopDataCrawlResultItemVo> accumulated = new ArrayList<>();
if (baseDailyFile != null && baseDailyFile.getId() != null) {
for (ShopDataCrawlDailyMemberEntity member : sortedDailyMembers(dailyFileService.listMembers(baseDailyFile.getId()))) {
ShopDataCrawlResultItemVo item = snapshotFromPayload(member);
if (item == null) {
FileResultEntity result = fileResultMapper.selectById(member.getResultId());
item = result == null ? null : loadSnapshotForDailyMember(result);
}
if (item == null) {
throw new BusinessException("无法读取累计文件中的结果数据，请重试文件任务");
}
accumulated.add(item);
}
}
for (ShopDataCrawlResultItemVo item : appended == null ? List.<ShopDataCrawlResultItemVo>of() : appended) {
if (item != null) {
accumulated.add(item);
}
}
return accumulated;
}
private ShopDataCrawlResultItemVo snapshotFromPayload(ShopDataCrawlDailyMemberEntity member) {
if (member == null || blank(member.getRowPayload())) {
return null;
}
try {
return objectMapper.readValue(member.getRowPayload(), ShopDataCrawlResultItemVo.class);
} catch (Exception ex) {
log.warn("[shop-data-crawl] daily member payload 解析失败 member={} msg={}",
member.getId(), safeMessage(ex));
return null;
}
}
private String snapshotPayload(ShopDataCrawlResultItemVo snapshot) {
if (snapshot == null) {
return null;
}
try {
return objectMapper.writeValueAsString(snapshot);
} catch (Exception ex) {
log.warn("[shop-data-crawl] 累计文件成员 payload 序列化失败 result={} msg={}",
snapshot.getResultId(), safeMessage(ex));
return null;
}
}
private List<ShopDataCrawlDailyMemberEntity> sortedDailyMembers(List<ShopDataCrawlDailyMemberEntity> memberRows) {
List<ShopDataCrawlDailyMemberEntity> members = new ArrayList<>(memberRows == null ? List.of() : memberRows);
members.removeIf(Objects::isNull);
members.sort(Comparator
.comparing(ShopDataCrawlDailyMemberEntity::getCreatedAt,
Comparator.nullsLast(Comparator.naturalOrder()))
.thenComparing(ShopDataCrawlDailyMemberEntity::getId,
Comparator.nullsLast(Comparator.naturalOrder())));
return members;
}
/** 累计文件跨天滚动时，把旧文件成员行（含 row_payload）原样迁到新文件，保留跨天携带的行。 */
private void reassignOlderMembers(ShopDataCrawlDailyFileEntity newDailyFile,
List<ShopDataCrawlDailyFileEntity> olderFiles) {
if (newDailyFile == null || newDailyFile.getId() == null) {
return;
}
for (ShopDataCrawlDailyFileEntity older : olderFiles == null ? List.<ShopDataCrawlDailyFileEntity>of() : olderFiles) {
if (older == null || older.getId() == null) {
continue;
}
dailyFileService.reassignMembers(older.getId(), newDailyFile.getId());
}
}
private DailyAggregationResult persistDailyAggregation(FileTaskEntity task,
FileResultEntity row,
Long userId,
@@ -1868,7 +1936,8 @@ public class ShopDataCrawlTaskService {
String shopKeyHash,
LocalDate businessDate,
DailyAggregationPreparation preparation,
DailyWorkbookArtifact artifact) {
DailyWorkbookArtifact artifact,
ShopDataCrawlResultItemVo snapshot) {
ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate(
userId, shopKeyHash, businessDate);
if (handleExistingDailyMembership(row, dailyFile)) {
@@ -1923,10 +1992,12 @@ public class ShopDataCrawlTaskService {
} else {
dailyFileService.update(dailyFile);
}
if (!dailyFileService.addMember(dailyFile.getId(), row.getTaskId(), row.getId())) {
if (!dailyFileService.addMemberWithPayload(dailyFile.getId(), row.getTaskId(), row.getId(),
snapshotPayload(snapshot))) {
throw new BusinessException("结果已归档，请重试文件任务");
}
reassignOlderMembers(dailyFile, olderFiles);
for (ShopDataCrawlDailyFileEntity older : olderFiles) {
dailyFileService.deleteDailyFile(older.getId());
}
backend-java/src/main/resources/db/V93__shop_data_crawl_daily_member_row_payload.sql
+3
View File
@@ -0,0 +1,3 @@
ALTER TABLE `biz_shop_data_crawl_daily_member`
ADD COLUMN `row_payload` MEDIUMTEXT NULL DEFAULT NULL COMMENT '结果快照 JSON（数据层增量模型：整表重建按成员行累积，旧数据为 NULL 时按结果快照兜底）' AFTER `result_id`;