diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java index 33a64fbd..de5b6cbe 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/impl/InvoiceReceiptServiceImpl.java @@ -50,6 +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.BfiReceivedInvoiceVO; import org.springblade.thirdparty.bfi.service.IBfiService; +import org.springblade.thirdparty.bfi.util.BfiUtil; import org.springblade.transport.constant.HcsSyncConstant; import org.springblade.transport.constant.KingdeeAttachmentConstant; import org.springblade.transport.config.BfiProperties; @@ -1304,11 +1305,11 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl2.5 万，从中点二分为两片递归续扫（递归深度受时间粒度自然收敛， - * 单片 500 条以内 1 页即短页终止） + * 同步单个票种的时间窗口：翻页与触顶二分续扫统一交给 {@link BfiUtil#forEachWindowSlice}， + * 页级客商单位过滤、镜像批量 upsert、已登记票跟随补绑在页回调中完成 * - * @param sliceStart 切片起点（含） - * @param sliceEnd 切片终点（不含，避免与下一片边界重复拉取） - * @param customerTaxNos 客商发票信息税号集，null 表示客商单位过滤关闭（整轮共享） - * @param stats 整轮统计累加器 + * @param windowStart 窗口起点（含） + * @param windowEnd 窗口终点（含） + * @param invoiceType 发票种类编码，null 不限 + * @param customerTaxNos 客商发票信息税号集，null 表示客商单位过滤关闭（整轮共享） + * @param stats 整轮统计累加器 */ - private void syncWindowSlice(LocalDateTime sliceStart, LocalDateTime sliceEnd, String invoiceType, + private void syncPoolWindow(LocalDateTime windowStart, LocalDateTime windowEnd, String invoiceType, Set customerTaxNos, SyncStats stats) { - // 二分收敛保护：切片粒度细到 1 秒仍触顶说明远端异常（同秒 >2.5 万条不可能），不再二分直接记漏 - if (!sliceStart.isBefore(sliceEnd)) { - log.error("金蝶全票池同步切片区间为空（{} ~ {}），远端 updateTime 粒度异常，放弃该切片", sliceStart, sliceEnd); - return; - } - boolean truncated = false; - for (int pageNo = 1; pageNo <= MAX_POOL_PAGES; pageNo++) { - List rows = this.pullPoolWindowPage(sliceStart, sliceEnd, invoiceType, pageNo); - if (rows.isEmpty()) { - break; - } - stats.pageCount++; - stats.pulledCount += rows.size(); - // 客商单位过滤：销/购方税号未同时登记在客商发票信息中的发票不入镜像； - // 短页判定用原始条数，过滤后的空页不能提前终止翻页 + stats.sliceCount += BfiUtil.forEachWindowSlice("金蝶全票池同步", windowStart, windowEnd, + POOL_PAGE_SIZE, MAX_POOL_PAGES, + (sliceStart, sliceEnd, pageNo) -> this.pullPoolWindowPage(sliceStart, sliceEnd, invoiceType, pageNo), + rows -> this.handlePoolPage(rows, customerTaxNos, stats)); + } + + /** + * 票池同步单页处理：客商单位过滤 + 镜像批量 upsert + 已登记票跟随补绑。 + * 客商过滤只影响入镜像行数，翻页节奏由 BfiUtil 按原始行数判断，不受过滤影响 + * + * @param rows 本页拉取的票池发票（原始行） + * @param customerTaxNos 客商发票信息税号集，null 表示客商单位过滤关闭 + * @param stats 整轮统计累加器 + */ + private void handlePoolPage(List rows, Set customerTaxNos, SyncStats stats) { + stats.pageCount++; + stats.pulledCount += rows.size(); + // 客商单位过滤：销/购方税号未同时登记在客商发票信息中的发票不入镜像 + if (customerTaxNos != null) { int rawPageSize = rows.size(); - if (customerTaxNos != null) { - rows = rows.stream().filter(row -> this.bothUnitsRegistered(row, customerTaxNos)).toList(); - stats.customerFiltered += rawPageSize - rows.size(); - } - BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows); - stats.created += upsertResult.created(); - stats.updated += upsertResult.updated(); - stats.mirrorUnchanged += upsertResult.unchanged(); - // 已登记票跟随补绑：同页发票号码批量定位登记，绑定关系变更页级事务批量落库 - if (!upsertResult.mirrors().isEmpty()) { - RefreshResult refreshResult = this.refreshBoundReceipts(stats.pageCount, upsertResult.mirrors()); - stats.bound += refreshResult.bound(); - stats.unchanged += refreshResult.unchanged(); - stats.failed += refreshResult.failed(); - } - // 远程短页：该片到底，正常终止（按原始拉取条数判断，客商过滤不改变翻页节奏） - if (rawPageSize < POOL_PAGE_SIZE) { - stats.sliceCount++; - return; - } - truncated = true; + rows = rows.stream().filter(row -> this.bothUnitsRegistered(row, customerTaxNos)).toList(); + stats.customerFiltered += rawPageSize - rows.size(); } - // 触顶：该切片数据超过单查询窗口上限，从中点二分续扫（updateTime 落在中点的行归左片 [start, mid)， - // 右片 [mid, end) 不含 mid，同 updateTime 行整体归左片，不重不漏） - if (truncated) { - LocalDateTime mid = sliceStart.plus(java.time.Duration.between(sliceStart, sliceEnd).dividedBy(2)); - if (!mid.isAfter(sliceStart)) { - // 切片已细到不足 1 秒仍满页：理论不可能，防御死递归 - log.error("金蝶全票池同步切片 {} ~ {} 已达最小粒度仍满页（>{} 条），该时间片数据异常，跳过", - sliceStart, sliceEnd, MAX_POOL_PAGES * POOL_PAGE_SIZE); - return; - } - log.warn("金蝶全票池同步切片 {} ~ {} 触达翻页保护 {} 页（{} 条），二分续扫 [{} ~ {}) + [{} ~ {})", - sliceStart, sliceEnd, MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE, sliceStart, mid, mid, sliceEnd); - this.syncWindowSlice(sliceStart, mid, invoiceType, customerTaxNos, stats); - this.syncWindowSlice(mid, sliceEnd, invoiceType, customerTaxNos, stats); - return; + BatchUpsertResult upsertResult = this.batchUpsertInvoicePool(rows); + stats.created += upsertResult.created(); + stats.updated += upsertResult.updated(); + stats.mirrorUnchanged += upsertResult.unchanged(); + // 已登记票跟随补绑：同页发票号码批量定位登记，绑定关系变更页级事务批量落库 + if (!upsertResult.mirrors().isEmpty()) { + RefreshResult refreshResult = this.refreshBoundReceipts(stats.pageCount, upsertResult.mirrors()); + stats.bound += refreshResult.bound(); + stats.unchanged += refreshResult.unchanged(); + stats.failed += refreshResult.failed(); } - stats.sliceCount++; } /** @@ -1523,7 +1503,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl pullPoolWindowPage(LocalDateTime windowStart, LocalDateTime windowEnd, + private BfiPageVO pullPoolWindowPage(LocalDateTime windowStart, LocalDateTime windowEnd, String invoiceType, int pageNo) { BfiInvoicePoolQueryDataDTO data = new BfiInvoicePoolQueryDataDTO(); data.setPageNo(pageNo); @@ -1536,7 +1516,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl poolPage = bfiService.queryInvoicePool(param); log.info("金蝶全票池同步 第{}页拉取 {} 条", pageNo, poolPage.getRows() == null ? 0 : poolPage.getRows().size()); - return poolPage.getRows() == null ? List.of() : poolPage.getRows(); + return poolPage; } /**