feat: 全票池同步补绑定回填步骤(39ffac1口径瘦身版)——同页票号批量定位未绑定存量登记补绑+页级事务批量落库+同步流水,绑定关系无变化的已绑定登记零写入(新架构登记无金蝶流水号/状态字段,不做状态跟随刷新)
This commit is contained in:
+98
-3
@@ -50,7 +50,7 @@ import org.springblade.thirdparty.bfi.pojo.vo.BfiInvoicePoolVO;
|
|||||||
import org.springblade.thirdparty.bfi.pojo.vo.BfiPageVO;
|
import org.springblade.thirdparty.bfi.pojo.vo.BfiPageVO;
|
||||||
import org.springblade.thirdparty.bfi.pojo.vo.BfiReceivedInvoiceVO;
|
import org.springblade.thirdparty.bfi.pojo.vo.BfiReceivedInvoiceVO;
|
||||||
import org.springblade.thirdparty.bfi.service.IBfiService;
|
import org.springblade.thirdparty.bfi.service.IBfiService;
|
||||||
import org.springblade.transport.constant.KingdeeAttachmentConstant;
|
import org.springblade.transport.constantq.KingdeeAttachmentConstant;
|
||||||
import org.springblade.transport.constant.KingdeeBfiConstant;
|
import org.springblade.transport.constant.KingdeeBfiConstant;
|
||||||
import org.springblade.transport.constant.TransportBusiConstant;
|
import org.springblade.transport.constant.TransportBusiConstant;
|
||||||
import org.springblade.transport.mapper.CustomerArchiveMapper;
|
import org.springblade.transport.mapper.CustomerArchiveMapper;
|
||||||
@@ -1181,7 +1181,8 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 同步主体:滚动增量窗口分页拉取 A801-1(按票池更新时间)+ 镜像批量 upsert(全列覆盖)。
|
* 同步主体(39ffac1 绑定回填口径瘦身版):滚动增量窗口分页拉取 A801-1(按票池更新时间)
|
||||||
|
* + 镜像批量 upsert(全列覆盖)+ 未绑定存量登记顺带补绑(同页发票号码批量定位登记,页级事务批量落库)。
|
||||||
* 同步窗口内所有可用态票全量重写,无指纹比对(ponytail: 万行级窗口全列重写成本可接受,
|
* 同步窗口内所有可用态票全量重写,无指纹比对(ponytail: 万行级窗口全列重写成本可接受,
|
||||||
* 人工字段清单式指纹有"新增列漏维护即静默不同步"坑;量级上来后升级为整行载荷规范化哈希)。
|
* 人工字段清单式指纹有"新增列漏维护即静默不同步"坑;量级上来后升级为整行载荷规范化哈希)。
|
||||||
* 拉取只认正常票(invoiceStatus=0),不推单据使用状态——在用/已用票随 updateTime 进窗口刷新
|
* 拉取只认正常票(invoiceStatus=0),不推单据使用状态——在用/已用票随 updateTime 进窗口刷新
|
||||||
@@ -1193,9 +1194,20 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
|||||||
long startMillis = System.currentTimeMillis();
|
long startMillis = System.currentTimeMillis();
|
||||||
LocalDateTime windowStart = LocalDateTime.now().minusDays(poolIncrementDays);
|
LocalDateTime windowStart = LocalDateTime.now().minusDays(poolIncrementDays);
|
||||||
LocalDateTime windowEnd = LocalDateTime.now();
|
LocalDateTime windowEnd = LocalDateTime.now();
|
||||||
|
// 以下为本轮统计口径,全部累计整轮并进入结束摘要
|
||||||
|
// A801-1 原始拉取张数
|
||||||
int pulledCount = 0;
|
int pulledCount = 0;
|
||||||
|
// 镜像新增条数(票池有、本地无)
|
||||||
int created = 0;
|
int created = 0;
|
||||||
|
// 镜像重写条数
|
||||||
int updated = 0;
|
int updated = 0;
|
||||||
|
// 登记首次绑定条数(罕见:存量手工数据未绑定时顺带补绑)
|
||||||
|
int bound = 0;
|
||||||
|
// 登记无变化跳过条数
|
||||||
|
int unchanged = 0;
|
||||||
|
// 页级批量落库失败条数(整页跳过,下轮幂等重试)
|
||||||
|
int failed = 0;
|
||||||
|
// 实际处理的翻页数,触达最大页数保护时配合 warn 定位截断
|
||||||
int pageCount = 0;
|
int pageCount = 0;
|
||||||
log.info("金蝶全票池同步开始,增量窗口 {} ~ {},单页 {} 条最多 {} 页",
|
log.info("金蝶全票池同步开始,增量窗口 {} ~ {},单页 {} 条最多 {} 页",
|
||||||
windowStart.format(POOL_TIME_FORMATTER), windowEnd.format(POOL_TIME_FORMATTER),
|
windowStart.format(POOL_TIME_FORMATTER), windowEnd.format(POOL_TIME_FORMATTER),
|
||||||
@@ -1210,6 +1222,13 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
|||||||
BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows);
|
BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows);
|
||||||
created += upsertResult.created();
|
created += upsertResult.created();
|
||||||
updated += upsertResult.updated();
|
updated += upsertResult.updated();
|
||||||
|
// 已登记票跟随刷新(39ffac1 口径):同页发票号码批量定位登记,内存比对变更后页级事务批量落库
|
||||||
|
if (!upsertResult.mirrors().isEmpty()) {
|
||||||
|
RefreshResult refreshResult = this.refreshBoundReceipts(pageNo, upsertResult.mirrors());
|
||||||
|
bound += refreshResult.bound();
|
||||||
|
unchanged += refreshResult.unchanged();
|
||||||
|
failed += refreshResult.failed();
|
||||||
|
}
|
||||||
if (rows.size() < POOL_PAGE_SIZE) {
|
if (rows.size() < POOL_PAGE_SIZE) {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
@@ -1220,11 +1239,87 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl<InvoiceReceiptMap
|
|||||||
MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE);
|
MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE);
|
||||||
}
|
}
|
||||||
String summary = "金蝶全票池同步:翻页" + pageCount + "页拉取" + pulledCount + "张发票,镜像新增" + created
|
String summary = "金蝶全票池同步:翻页" + pageCount + "页拉取" + pulledCount + "张发票,镜像新增" + created
|
||||||
+ "更新" + updated + ",耗时" + (System.currentTimeMillis() - startMillis) + "ms";
|
+ "更新" + updated + ",补绑" + bound + ",登记无变化跳过" + unchanged
|
||||||
|
+ ",失败" + failed + ",耗时" + (System.currentTimeMillis() - startMillis) + "ms";
|
||||||
log.info("金蝶全票池同步结束,{}", summary);
|
log.info("金蝶全票池同步结束,{}", summary);
|
||||||
return summary;
|
return summary;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** 已登记票跟随刷新结果统计(绑定/刷新/无变化/失败) */
|
||||||
|
private record RefreshResult(int bound, int unchanged, int failed) {
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 同页发票号码批量定位收票登记,绑定关系变更页级事务批量落库(39ffac1 口径瘦身版):
|
||||||
|
* 新架构下登记绑定在保存时完成且登记无金蝶流水号/状态字段(70d5d34 已删),
|
||||||
|
* 已绑定登记零写入,仅未绑定存量登记(罕见,人工数据)顺带补绑;批量失败整页跳过待下轮幂等重试
|
||||||
|
*
|
||||||
|
* @param pageNo 当前页码(日志定位)
|
||||||
|
* @param mirrorMap 本页 invoiceNo -> 刚 upsert 的镜像
|
||||||
|
* @return 刷新结果统计
|
||||||
|
*/
|
||||||
|
private RefreshResult refreshBoundReceipts(int pageNo, Map<String, KingdeeInvoicePool> mirrorMap) {
|
||||||
|
Map<String, List<InvoiceReceipt>> receiptMap = list(Wrappers.<InvoiceReceipt>lambdaQuery()
|
||||||
|
.in(InvoiceReceipt::getInvoiceNo, mirrorMap.keySet())
|
||||||
|
.ne(InvoiceReceipt::getApprovalStatus, TransportBusiConstant.VOIDED)).stream()
|
||||||
|
.filter(receipt -> !Objects.equals(receipt.getIsDeleted(), 1))
|
||||||
|
.collect(Collectors.groupingBy(InvoiceReceipt::getInvoiceNo));
|
||||||
|
// 镜像有了却没人登记:本页不产生任何刷新,属正常(票池范围大于我方登记范围)
|
||||||
|
if (receiptMap.isEmpty()) {
|
||||||
|
return new RefreshResult(0, 0, 0);
|
||||||
|
}
|
||||||
|
List<InvoiceReceipt> changedReceipts = new ArrayList<>();
|
||||||
|
List<InvoiceReceiptRecord> syncRecords = new ArrayList<>();
|
||||||
|
int bound = 0;
|
||||||
|
int unchanged = 0;
|
||||||
|
for (Map.Entry<String, List<InvoiceReceipt>> entry : receiptMap.entrySet()) {
|
||||||
|
KingdeeInvoicePool mirror = mirrorMap.get(entry.getKey());
|
||||||
|
for (InvoiceReceipt receipt : entry.getValue()) {
|
||||||
|
// 新架构登记无 kingdeeBillNo/kingdeeStatus 字段(70d5d34 已删),仅绑定关系可变:
|
||||||
|
// 已绑定且指向同一镜像的登记零写入;未绑定存量登记(罕见,人工数据)顺带补绑
|
||||||
|
boolean unbound = receipt.getKingdeeInvoicePoolId() == null
|
||||||
|
|| !Objects.equals(receipt.getKingdeeInvoicePoolId(), mirror.getId());
|
||||||
|
if (!unbound) {
|
||||||
|
unchanged++;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
receipt.setKingdeeInvoicePoolId(mirror.getId());
|
||||||
|
changedReceipts.add(receipt);
|
||||||
|
syncRecords.add(this.buildPoolSyncRecord(receipt, "定时绑定金蝶票池发票"));
|
||||||
|
bound++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (changedReceipts.isEmpty()) {
|
||||||
|
return new RefreshResult(0, unchanged, 0);
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
// 页级事务批量落库:绑定/状态批量更新 + 同步流水批量写入(JDBC batch,整页原子生效)
|
||||||
|
Db.updateBatchById(changedReceipts);
|
||||||
|
if (!syncRecords.isEmpty()) {
|
||||||
|
Db.saveBatch(syncRecords);
|
||||||
|
}
|
||||||
|
log.info("金蝶全票池同步 第{}页落库完成,补绑 {} 条,同步流水 {} 条",
|
||||||
|
pageNo, changedReceipts.size(), syncRecords.size());
|
||||||
|
return new RefreshResult(bound, unchanged, 0);
|
||||||
|
} catch (Exception batchException) {
|
||||||
|
// 批量落库失败:整页跳过计入失败,下轮任务幂等重试,不做逐单降级(保持全批量链路)
|
||||||
|
log.error("金蝶全票池同步登记刷新批量落库异常,第{}页 {} 条跳过待下轮重试", pageNo, changedReceipts.size(), batchException);
|
||||||
|
return new RefreshResult(0, unchanged, changedReceipts.size());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 构建金蝶票池同步流水(操作人以「定时任务」名义记录) */
|
||||||
|
private InvoiceReceiptRecord buildPoolSyncRecord(InvoiceReceipt receipt, String reason) {
|
||||||
|
InvoiceReceiptRecord syncRecord = new InvoiceReceiptRecord();
|
||||||
|
syncRecord.setInvoiceReceiptId(receipt.getId());
|
||||||
|
syncRecord.setActionType("sync");
|
||||||
|
syncRecord.setActionName(reason);
|
||||||
|
syncRecord.setFromStatus(receipt.getApprovalStatus());
|
||||||
|
syncRecord.setToStatus(receipt.getApprovalStatus());
|
||||||
|
syncRecord.setOperatorName(AuthUtil.getUserName() == null ? "定时任务" : AuthUtil.getUserName());
|
||||||
|
return syncRecord;
|
||||||
|
}
|
||||||
|
|
||||||
/** A801-1 按票池更新时间窗口分页拉取(只认正常票,不推单据使用状态,票面可用性由本地过滤) */
|
/** A801-1 按票池更新时间窗口分页拉取(只认正常票,不推单据使用状态,票面可用性由本地过滤) */
|
||||||
private List<BfiInvoicePoolVO> pullPoolWindowPage(LocalDateTime windowStart, LocalDateTime windowEnd, int pageNo) {
|
private List<BfiInvoicePoolVO> pullPoolWindowPage(LocalDateTime windowStart, LocalDateTime windowEnd, int pageNo) {
|
||||||
BfiInvoicePoolQueryDataDTO data = new BfiInvoicePoolQueryDataDTO();
|
BfiInvoicePoolQueryDataDTO data = new BfiInvoicePoolQueryDataDTO();
|
||||||
|
|||||||
Reference in New Issue
Block a user