diff --git a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/IInvoiceReceiptService.java b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/IInvoiceReceiptService.java index 0a45bca0..cf5a877d 100644 --- a/blade-service/blade-transport/src/main/java/org/springblade/transport/service/IInvoiceReceiptService.java +++ b/blade-service/blade-transport/src/main/java/org/springblade/transport/service/IInvoiceReceiptService.java @@ -70,6 +70,9 @@ public interface IInvoiceReceiptService extends BaseService { /** * 定时查询金蝶全票池发票（A801-1）并绑定收票登记 + *

+ * 按租户加分布式锁做 single-flight（需开启 {@code blade.lock.enabled}）：多副本并发触发、 + * 或手动触发撞上进行中的轮次时，抢不到锁的一轮直接跳过并返回提示，不重复拉取与写库。 * * @return 执行摘要，含拉取页数、镜像新增/更新、绑定与状态刷新统计 */ @@ -78,8 +81,8 @@ public interface IInvoiceReceiptService extends BaseService { /** * 批量应用金蝶票池绑定结果（页级事务）：登记绑定批量更新 + 同步流水批量写入，整页原子生效 * - * @param receipts 已回填绑定字段的收票登记 - * @param records 同步流水 + * @param receipts 变更比对后需更新、且已回填绑定字段的收票登记 + * @param records 同步流水，绑定关系变化或金蝶状态迁移时才写，可为 receipts 的子集 */ void applyInvoicePoolSyncBatch(List receipts, List records); 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 f43dd196..9112fd15 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 @@ -33,6 +33,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springblade.core.log.exception.ServiceException; import org.springblade.core.mp.base.BaseServiceImpl; +import org.springblade.core.redis.lock.LockType; +import org.springblade.core.redis.lock.RedisLockClient; import org.springblade.core.secure.utils.AuthUtil; import org.springblade.core.tool.jackson.JsonUtil; import org.springblade.core.tool.utils.Func; @@ -65,6 +67,7 @@ import org.springblade.transport.pojo.entity.KingdeeInvoicePool; import org.springblade.transport.pojo.vo.InvoiceReceiptVO; import org.springblade.transport.service.IInvoiceReceiptService; import org.springblade.transport.wrapper.InvoiceReceiptWrapper; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -83,6 +86,7 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.stream.Collectors; @@ -121,6 +125,30 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl> POOL_FINGERPRINT_FIELDS = List.of( + KingdeeInvoicePool::getInvoiceNo, + KingdeeInvoicePool::getSerialNo, + KingdeeInvoicePool::getKingdeeBillNo, + KingdeeInvoicePool::getKingdeeStatus, + KingdeeInvoicePool::getInvoiceDate, + KingdeeInvoicePool::getInvoiceType, + KingdeeInvoicePool::getTaxRate, + KingdeeInvoicePool::getInvoiceAmount, + KingdeeInvoicePool::getTaxAmount, + KingdeeInvoicePool::getReceiverName, + KingdeeInvoicePool::getBuyerTaxNo, + KingdeeInvoicePool::getIssuerName, + KingdeeInvoicePool::getSalerTaxNo, + KingdeeInvoicePool::getCheckStatus, + KingdeeInvoicePool::getAttachmentsJson, + KingdeeInvoicePool::getSourceUpdatedTime); + + /** 定时绑定 single-flight 锁键前缀，按租户区分，同租户同一时刻只允许一轮全票池同步 */ + private static final String POOL_SYNC_LOCK_PREFIX = "transport:invoice-receipt:kingdee-pool-sync:"; + private final KingdeeInvoicePoolMapper invoicePoolMapper; private final InvoiceReceiptSettlementMapper settlementRelationMapper; private final InvoiceReceiptRecordMapper recordMapper; @@ -128,6 +156,8 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl redisLockClientProvider; /** A801-1 增量窗口天数（按票池更新时间），首轮/兜底全量比对窗口 */ @Value("${thirdParty.bfi.invoiceReceipt.incrementDays:30}") @@ -141,6 +171,13 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl selectPage(IPage page, InvoiceReceiptVO query) { LambdaQueryWrapper wrapper = Wrappers.lambdaQuery() @@ -412,58 +449,137 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl customerTaxNos = customerTaxNoSet(); - log.info("定时绑定金蝶全票池开始，增量窗口 {} ~ {}，发票种类：{}，客商税号白名单 {} 个", + log.info("定时绑定金蝶全票池开始，增量窗口 {} ~ {}，发票种类：{}，客商税号白名单 {} 个，单页 {} 条最多 {} 页", windowStart.format(POOL_TIME_FORMATTER), windowEnd.format(POOL_TIME_FORMATTER), Func.isEmpty(invoiceTypes) ? "不限制" : invoiceTypes, - customerTaxNos.isEmpty() ? "无（不限制）" : String.valueOf(customerTaxNos.size())); - for (int pageNo = 1; pageNo <= MAX_POOL_PAGES; pageNo++) { + customerTaxNos.isEmpty() ? "无（不限制）" : String.valueOf(customerTaxNos.size()), + POOL_PAGE_SIZE, MAX_POOL_PAGES); + int pageNo = 1; + for (; pageNo <= MAX_POOL_PAGES; pageNo++) { List rows = queryPoolPage(windowStart, windowEnd, pageNo); if (rows.isEmpty()) { break; } + pageCount++; pulledCount += rows.size(); - // 发票种类白名单 + 客商税号白名单（购、销双方）过滤（翻页终止仍按原始页条数判断） + // 发票种类白名单 + 客商税号白名单（购、销双方税号均须命中）过滤（翻页终止仍按原始页条数判断） List typeAllowedRows = filterByInvoiceTypes(rows); - filteredCount += rows.size() - typeAllowedRows.size(); + int pageTypeFiltered = rows.size() - typeAllowedRows.size(); + filteredCount += pageTypeFiltered; List allowedRows = filterByArchiveTaxNos(customerTaxNos, typeAllowedRows); - salerFilteredCount += typeAllowedRows.size() - allowedRows.size(); - // ② 镜像批量 upsert：票池为唯一事实源，无论是否已登记均刷新镜像（批量查改，避免逐条 N+1） + int pageTaxFiltered = typeAllowedRows.size() - allowedRows.size(); + taxNoFilteredCount += pageTaxFiltered; + allowedCount += allowedRows.size(); + // 每页过滤漏斗：定位「拉了几千张却一张都不入库」到底卡在发票种类还是客商税号白名单 + log.info("定时绑定金蝶全票池 第{}页过滤明细：原始 {} 张，发票种类白名单外 {} 张，客商税号白名单外 {} 张，通过 {} 张", + pageNo, rows.size(), pageTypeFiltered, pageTaxFiltered, allowedRows.size()); + // ② 镜像批量 upsert：票池为唯一事实源，无论是否已登记均以票池为准，票面字段有变化才重写（批量查改避免逐条 N+1） BatchUpsertResult upsertResult = batchUpsertInvoicePool(allowedRows); Map mirrorMap = upsertResult.mirrors(); created += upsertResult.created(); updated += upsertResult.updated(); - // ③ 同页发票号码批量定位收票登记，内存回填绑定后页级批量落库（批量失败整页跳过待下轮） + mirrorUnchanged += upsertResult.unchanged(); + // ③ 同页发票号码批量定位收票登记，内存比对变更后页级批量落库（无变化不写库不刷流水；批量失败整页跳过待下轮） if (!mirrorMap.isEmpty()) { Map> receiptMap = list(Wrappers.lambdaQuery() .in(InvoiceReceipt::getInvoiceNo, mirrorMap.keySet()) .ne(InvoiceReceipt::getApprovalStatus, VOIDED)).stream() .filter(receipt -> !Objects.equals(receipt.getIsDeleted(), 1)) .collect(Collectors.groupingBy(InvoiceReceipt::getInvoiceNo)); + // 镜像有了却没人登记：本页不产生任何绑定，属正常（票池范围大于我方登记范围） + if (receiptMap.isEmpty()) { + log.info("定时绑定金蝶全票池 第{}页 {} 张镜像未匹配到收票登记（发票号码未登记或登记已作废）", + pageNo, mirrorMap.size()); + } List targets = new ArrayList<>(); for (Map.Entry entry : mirrorMap.entrySet()) { for (InvoiceReceipt receipt : receiptMap.getOrDefault(entry.getKey(), List.of())) { + KingdeeInvoicePool mirror = entry.getValue(); + // A801-1 未返回流水号时保持登记原值，与镜像非空字段更新口径一致，避免空值每轮判为变更 + String kingdeeBillNo = Func.isNotEmpty(mirror.getKingdeeBillNo()) + ? mirror.getKingdeeBillNo() : receipt.getKingdeeBillNo(); boolean unbound = receipt.getKingdeeInvoicePoolId() == null - || !Objects.equals(receipt.getKingdeeInvoicePoolId(), entry.getValue().getId()); + || !Objects.equals(receipt.getKingdeeInvoicePoolId(), mirror.getId()); + boolean statusChanged = !Objects.equals(receipt.getKingdeeStatus(), mirror.getKingdeeStatus()); + boolean billNoChanged = !Objects.equals(receipt.getKingdeeBillNo(), kingdeeBillNo); + // 变更检测：绑定关系、金蝶流水号、同步状态三项均无变化则本轮跳过，避免整窗每轮重写与流水刷屏 + if (!unbound && !statusChanged && !billNoChanged) { + unchanged++; + continue; + } // 内存回填：绑定关系对齐当前发票号码对应镜像（同租户同发票号镜像唯一），票面字段不覆盖人工登记信息 - receipt.setKingdeeInvoicePoolId(entry.getValue().getId()); - receipt.setKingdeeBillNo(entry.getValue().getKingdeeBillNo()); - receipt.setKingdeeStatus(entry.getValue().getKingdeeStatus()); - targets.add(new BindTarget(receipt, entry.getValue(), unbound)); + receipt.setKingdeeInvoicePoolId(mirror.getId()); + receipt.setKingdeeBillNo(kingdeeBillNo); + receipt.setKingdeeStatus(mirror.getKingdeeStatus()); + // 流水只在绑定关系变化或金蝶状态迁移时写，纯流水号回填不刷流水 + InvoiceReceiptRecord syncRecord = unbound || statusChanged ? buildSyncRecord(receipt, + unbound ? "定时绑定金蝶票池发票" : "定时刷新金蝶票池状态") : null; + targets.add(new BindTarget(receipt, mirror, unbound, syncRecord)); } } if (!targets.isEmpty()) { @@ -471,20 +587,28 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl buildSyncRecord(target.receipt(), "定时绑定金蝶票池发票")).toList()); + targets.stream().map(BindTarget::syncRecord).filter(Objects::nonNull).toList()); + int pageBound = 0; + int pageRecorded = 0; for (BindTarget target : targets) { + if (target.syncRecord() != null) { + pageRecorded++; + } if (target.unbound()) { bound++; - log.info("定时绑定金蝶票池发票成功，发票号码={}，登记id={}，金蝶流水号={}", - target.receipt().getInvoiceNo(), target.receipt().getId(), target.mirror().getSerialNo()); + pageBound++; + log.info("定时绑定金蝶票池发票成功，第{}页，发票号码={}，登记id={}，金蝶流水号={}", + pageNo, target.receipt().getInvoiceNo(), target.receipt().getId(), target.mirror().getSerialNo()); } else { refreshed++; } } + log.info("定时绑定金蝶票池发票第{}页落库完成，变更 {} 条（绑定 {} 条、状态刷新 {} 条），同步流水 {} 条", + pageNo, targets.size(), pageBound, targets.size() - pageBound, pageRecorded); } catch (Exception batchException) { // 批量落库失败：整页跳过计入失败，下轮任务幂等重试，不做逐单降级（保持全批量链路） failed += targets.size(); - log.error("定时绑定金蝶票池发票批量落库异常，本页 {} 条跳过待下轮重试", targets.size(), batchException); + log.error("定时绑定金蝶票池发票批量落库异常，第{}页 {} 条跳过待下轮重试", pageNo, targets.size(), batchException); } } } @@ -492,12 +616,19 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl MAX_POOL_PAGES) { + // 末轮仍满页说明增量窗口数据量已超过翻页保护，尾部数据本轮未处理（下轮窗口未变仍会漏，须调整配置） + log.warn("金蝶全票池增量窗口翻页达最大保护 {} 页（{} 条）仍未到底，窗口尾部本轮未处理，请缩小 incrementDays 或调大保护页数", + MAX_POOL_PAGES, MAX_POOL_PAGES * POOL_PAGE_SIZE); + } // ④ 补偿绑定：仍为未绑定状态的登记按发票号码精确补查（不受增量窗口限制）， // 兜底增量窗口滑动或批量落库持续失败导致的漏绑——发票在金蝶票池可随时按号补查，直到绑定成功为止 int compensated = compensateUnboundReceipts(self); - String summary = "金蝶全票池同步：拉取" + pulledCount + "张发票，类型过滤" + filteredCount + "张，客商过滤" - + salerFilteredCount + "张，镜像新增" + created + "更新" + updated + "，绑定" + bound + "，状态刷新" - + refreshed + "，失败" + failed + "，补偿绑定" + compensated; + String summary = "金蝶全票池同步：翻页" + pageCount + "页拉取" + pulledCount + "张发票，类型过滤" + filteredCount + + "张，客商过滤" + taxNoFilteredCount + "张，过滤后剩" + allowedCount + "张，镜像新增" + created + "更新" + + updated + "无变化" + mirrorUnchanged + "，绑定" + bound + "，状态刷新" + + refreshed + "，登记无变化跳过" + unchanged + "，失败" + failed + "，补偿绑定" + compensated + "，耗时" + + (System.currentTimeMillis() - startMillis) + "ms"; log.info("定时绑定金蝶全票池结束，{}", summary); return summary; } @@ -520,6 +651,7 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl invoiceNos = unboundReceipts.stream() @@ -531,17 +663,23 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl targets = new ArrayList<>(); List newMirrors = new ArrayList<>(); + List poolMissInvoiceNos = new ArrayList<>(); + int localHitCount = mirrorMap.size(); + int remoteQueryCount = 0; for (InvoiceReceipt receipt : unboundReceipts) { if (Func.isEmpty(receipt.getInvoiceNo())) { + log.warn("补偿绑定跳过，登记发票号码为空，登记id={}", receipt.getId()); continue; } String invoiceNo = receipt.getInvoiceNo().trim(); KingdeeInvoicePool mirror = mirrorMap.get(invoiceNo); if (mirror == null) { + // rim 契约无批量号码参数，本地镜像缺失只能逐号远程补查，是本轮最主要的远程调用开销来源 + remoteQueryCount++; List poolInvoices = filterByInvoiceTypes(queryPoolByInvoiceNo(invoiceNo)); if (poolInvoices.isEmpty()) { // 金蝶票池暂无该发票：跳过不写库，待后续轮次补查，避免流水刷屏 - log.warn("补偿绑定跳过，金蝶票池未查询到发票，登记id={}，发票号码={}", receipt.getId(), invoiceNo); + poolMissInvoiceNos.add(invoiceNo); continue; } mirror = fillInvoicePool(null, poolInvoices.get(0)); @@ -551,9 +689,16 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl buildSyncRecord(target.receipt(), "补偿绑定金蝶票池发票")).toList()); + targets.stream().map(BindTarget::syncRecord).filter(Objects::nonNull).toList()); targets.forEach(target -> log.info("补偿绑定金蝶票池发票成功，发票号码={}，登记id={}，金蝶流水号={}", target.receipt().getInvoiceNo(), target.receipt().getId(), target.mirror().getSerialNo())); + log.info("定时绑定金蝶全票池补偿结束，补偿绑定 {} 条，本地镜像命中 {} 条，远程按号补查 {} 次（新增镜像 {} 条），票池无票 {} 张，耗时 {}ms", + targets.size(), localHitCount, remoteQueryCount, newMirrors.size(), poolMissInvoiceNos.size(), + System.currentTimeMillis() - startMillis); return targets.size(); } @@ -659,15 +807,16 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl rows) { Map mirrorMap = new LinkedHashMap<>(); if (rows.isEmpty()) { - return new BatchUpsertResult(mirrorMap, 0, 0); + return new BatchUpsertResult(mirrorMap, 0, 0, 0); } List invoiceNos = rows.stream() .map(row -> row.getInvoiceNo().trim()) @@ -680,14 +829,23 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl first)); List toInsert = new ArrayList<>(); List toUpdate = new ArrayList<>(); + int unchangedCount = 0; for (BfiInvoicePoolVO row : rows) { if (Func.isEmpty(row.getInvoiceNo())) { continue; } KingdeeInvoicePool existing = existingMap.get(row.getInvoiceNo().trim()); + // fillInvoicePool 原地改写实体，故先取填充前的票面指纹用于变更比对 + String[] fingerprintBefore = invoicePoolFingerprint(existing); KingdeeInvoicePool mirror = fillInvoicePool(existing, row); - (existing == null ? toInsert : toUpdate).add(mirror); mirrorMap.put(mirror.getInvoiceNo(), mirror); + if (existing == null) { + toInsert.add(mirror); + } else if (invoicePoolFingerprintChanged(fingerprintBefore, mirror)) { + toUpdate.add(mirror); + } else { + unchangedCount++; + } } // 批量写库：失败时中断本轮，由任务层记录异常、下轮重试（upsert 幂等） if (!toInsert.isEmpty()) { @@ -696,8 +854,9 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl mirrors, int created, int updated) { + private record BatchUpsertResult(Map mirrors, int created, int updated, + int unchanged) { } /** - * 定时绑定目标：登记实体 + 对应镜像 + 是否首次绑定（区分绑定与状态刷新统计） - * - * @param receipt 已回填绑定字段的收票登记 - * @param mirror 对应票据池镜像 - * @param unbound true 表示首次绑定，false 表示仅状态刷新 + * 镜像票面指纹：A801-1 回填字段的规范化取值快照，用于判断本轮是否需要重写镜像；null 镜像（新增）返回 null */ - private record BindTarget(InvoiceReceipt receipt, KingdeeInvoicePool mirror, boolean unbound) { + private String[] invoicePoolFingerprint(KingdeeInvoicePool mirror) { + if (mirror == null) { + return null; + } + String[] fingerprint = new String[POOL_FINGERPRINT_FIELDS.size()]; + for (int index = 0; index < fingerprint.length; index++) { + fingerprint[index] = fingerprintValue(POOL_FINGERPRINT_FIELDS.get(index).apply(mirror)); + } + return fingerprint; + } + + /** + * 镜像票面是否变化：填充后为空视为维持库中原值（MyBatis-Plus 非空字段更新策略不写空值）， + * 与填充前一致则不算变化，避免源侧空字段导致每轮判为变更而无效重写 + */ + private boolean invoicePoolFingerprintChanged(String[] fingerprintBefore, KingdeeInvoicePool mirror) { + if (fingerprintBefore == null) { + return true; + } + for (int index = 0; index < fingerprintBefore.length; index++) { + String fingerprintAfter = fingerprintValue(POOL_FINGERPRINT_FIELDS.get(index).apply(mirror)); + if (!fingerprintAfter.isEmpty() && !fingerprintAfter.equals(fingerprintBefore[index])) { + return true; + } + } + return false; + } + + /** 指纹取值规范化：空值统一为空串，金额去尾零使 13.0000 与 13 等价，其余按去首尾空白后的文本比对 */ + private String fingerprintValue(Object value) { + if (value == null) { + return ""; + } + if (value instanceof BigDecimal decimal) { + return decimal.stripTrailingZeros().toPlainString(); + } + return value.toString().trim(); + } + + /** + * 定时绑定目标：登记实体 + 对应镜像 + 是否绑定关系变化 + 同步流水 + * + * @param receipt 已回填绑定字段的收票登记 + * @param mirror 对应票据池镜像 + * @param unbound true 表示绑定关系发生变化（含首次绑定），false 表示仅金蝶侧信息刷新 + * @param syncRecord 同步流水，null 表示本轮变更无需记录流水 + */ + private record BindTarget(InvoiceReceipt receipt, KingdeeInvoicePool mirror, boolean unbound, + InvoiceReceiptRecord syncRecord) { } /** @@ -801,19 +1006,24 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl filterByArchiveTaxNos(Set customerTaxNos, List rows) { if (customerTaxNos.isEmpty()) { return rows; } return rows.stream() - .filter(row -> row.getBuyerTaxNo() != null && row.getSalerTaxNo() != null - && customerTaxNos.contains(row.getBuyerTaxNo().trim()) - && customerTaxNos.contains(row.getSalerTaxNo().trim())) + .filter(row -> taxNoArchived(customerTaxNos, row.getBuyerTaxNo()) + && taxNoArchived(customerTaxNos, row.getSalerTaxNo())) .toList(); } + /** 税号是否命中客商档案集合（空税号不命中） */ + private boolean taxNoArchived(Set customerTaxNos, String taxNo) { + return Func.isNotEmpty(taxNo) && customerTaxNos.contains(taxNo.trim()); + } + /** * 加载客商档案税号集合（购、销方白名单来源）：status=1 有效客商的统一社会信用代码（纳税人识别号） */ @@ -848,61 +1058,63 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl 13），顶层为空回退首条明细，均无效时置 0（镜像列非空约束） */ private BigDecimal resolveTaxRate(BfiInvoicePoolVO row) { - BigDecimal taxRate = parsePercent(row.getTaxRate()); + String invoiceNo = row.getInvoiceNo(); + BigDecimal taxRate = parsePercent(invoiceNo, row.getTaxRate()); if (taxRate == null) { taxRate = Optional.ofNullable(row.getItems()).orElse(List.of()).stream() .map(BfiInvoicePoolItemVO::getTaxRate) .filter(StringUtil::isNotBlank) .findFirst() - .map(this::parsePercent) + .map(value -> parsePercent(invoiceNo, value)) .orElse(null); } return taxRate == null ? BigDecimal.ZERO : taxRate; } - private BigDecimal parsePercent(String value) { + private BigDecimal parsePercent(String invoiceNo, String value) { if (StringUtil.isBlank(value)) { return null; } try { return new BigDecimal(value.trim()).multiply(BigDecimal.valueOf(100)); } catch (NumberFormatException exception) { - log.warn("金蝶票池发票税率解析失败：{}", value); + log.warn("金蝶票池发票税率解析失败，按 0 处理，发票号码={}，税率={}", invoiceNo, value); return null; } } - private BigDecimal parseAmount(String value) { + private BigDecimal parseAmount(String invoiceNo, String value) { if (StringUtil.isBlank(value)) { return BigDecimal.ZERO; } try { return new BigDecimal(value.trim()).setScale(2, RoundingMode.HALF_UP); } catch (NumberFormatException exception) { - log.warn("金蝶票池发票金额解析失败：{}", value); + log.warn("金蝶票池发票金额解析失败，按 0 处理，发票号码={}，金额={}", invoiceNo, value); return BigDecimal.ZERO; } } - private LocalDate parseDate(String value) { + private LocalDate parseDate(String invoiceNo, String value) { if (StringUtil.isBlank(value)) { return null; } try { return LocalDate.parse(value.trim().substring(0, 10)); } catch (Exception exception) { - log.warn("金蝶票池发票开票日期解析失败：{}", value); + log.warn("金蝶票池发票开票日期解析失败，置空处理，发票号码={}，日期={}", invoiceNo, value); return null; } } - private LocalDateTime parseDateTime(String value) { + private LocalDateTime parseDateTime(String invoiceNo, String value) { if (StringUtil.isBlank(value)) { return null; } try { return LocalDateTime.parse(value.trim().replace(" ", "T").substring(0, 19)); } catch (Exception exception) { + log.warn("金蝶票池发票源侧时间解析失败，置空处理，发票号码={}，时间={}", invoiceNo, value); return null; } } @@ -915,6 +1127,12 @@ public class InvoiceReceiptServiceImpl extends BaseServiceImpl length ? trimmed.substring(0, length) : trimmed; } + /** 租户标识：定时任务上下文无登录用户时以 default 加锁，锁粒度与数据隔离边界一致 */ + private String currentTenant() { + String tenantId = AuthUtil.getTenantId(); + return Func.isEmpty(tenantId) ? "default" : tenantId; + } + /** 操作人：定时任务上下文无登录用户时以「定时任务」名义记录流水 */ private String currentOperator() { String userName = AuthUtil.getUserName(); diff --git a/blade-service/blade-transport/src/test/java/org/springblade/transport/SyncKingdeeTest.java b/blade-service/blade-transport/src/test/java/org/springblade/transport/SyncKingdeeTest.java new file mode 100644 index 00000000..1a14bd46 --- /dev/null +++ b/blade-service/blade-transport/src/test/java/org/springblade/transport/SyncKingdeeTest.java @@ -0,0 +1,57 @@ +/** + * BladeX Commercial License Agreement + * Copyright (c) 2018-2099, https://bladex.cn. All rights reserved. + *

+ * Use of this software is governed by the Commercial License Agreement + * obtained after purchasing a license from BladeX. + *

+ * 1. This software is for development use only under a valid license + * from BladeX. + *

+ * 2. Redistribution of this software's source code to any third party + * without a commercial license is strictly prohibited. + *

+ * 3. Licensees may copyright their own code but cannot use segments + * from this software for such purposes. Copyright of this software + * remains with BladeX. + *

+ * Using this software signifies agreement to this License, and the software + * must not be used for illegal purposes. + *

+ * THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is + * not liable for any claims arising from secondary or illegal development. + *

+ * Author: Chill Zhuang (bladejava@qq.com) + */ +package org.springblade.transport; + +import org.junit.jupiter.api.Test; +import org.springblade.transport.service.IInvoiceReceiptService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; + +/** + * 金蝶数据定时同步测试：手动触发收票管理金蝶全票池发票绑定（A801-1），打印执行摘要便于联调抽查。 + *

+ * 测试用属性显式关闭 {@code thirdParty.bfi.invoiceReceipt.syncEnabled}：{@code @SpringBootTest} 会连同 + * {@code @EnableScheduling} 一起启动上下文，不关闭则 cron 会在同一 JVM 内并发再跑一轮， + * 与手动这一轮互相抢数据（未开启 blade.lock.enabled 时没有 single-flight 保护），执行摘要会对不上。 + * + * @author Chill + */ +@SpringBootTest(classes = TransportApplication.class, properties = { + "spring.profiles.active=dev", + "spring.main.allow-bean-definition-overriding=true", + "thirdParty.bfi.invoiceReceipt.syncEnabled=false" +}) +class SyncKingdeeTest { + + @Autowired + private IInvoiceReceiptService invoiceReceiptService; + + @Test + void syncKingdeeInvoicePool() { + String summary = invoiceReceiptService.syncKingdeeInvoicePool(); + System.out.println("金蝶数据定时同步执行结果：" + summary); + } +} diff --git a/doc/docs/业务流程/2026-09-20-收票管理同步金蝶全票池发票（A801-1查询+批量同步+定时绑定关联）业务流程.md b/doc/docs/业务流程/2026-09-20-收票管理同步金蝶全票池发票（A801-1查询+批量同步+定时绑定关联）业务流程.md index 9b6fe56b..d0181cb2 100644 --- a/doc/docs/业务流程/2026-09-20-收票管理同步金蝶全票池发票（A801-1查询+批量同步+定时绑定关联）业务流程.md +++ b/doc/docs/业务流程/2026-09-20-收票管理同步金蝶全票池发票（A801-1查询+批量同步+定时绑定关联）业务流程.md @@ -88,7 +88,7 @@ InvoiceReceiptKingdeeSyncTask InvoiceReceiptServiceImpl - 任务类仿 `PaymentKingdeeResultTask` / `CustomerKingdeeSyncTask`：业务逻辑收敛在 `IInvoiceReceiptService.syncKingdeeInvoicePool()`，任务类只做开关与调度，便于手动触发联调。 - 增量窗口约定：**每轮固定滚动近 N 天**（可配，建议 30 天），不依赖"上次成功时间点"——单轮失败不会漏，同一批数据会被后续每轮反复幂等覆盖。 -- **客商税号白名单（购、销双方）**：每轮加载客商档案有效税号（`unifiedCreditCode`，status=1），窗口拉回的发票须**购方税号与销方税号同时命中**该集合才入镜像、参与绑定，白名单外发票不入镜像不绑定；**集合为空不限制**（档案未维护时兼容全量）。`buyerTaxNo`/`salerTaxNo` 均为 rim 契约单值参数，无法按税号集合远程查询，故采用窗口拉取 + 内存过滤的等价实现。 +- **客商税号白名单（购、销双方均须命中）**：每轮加载客商档案有效税号（`unifiedCreditCode`，status=1），窗口拉回的发票须**购方税号与销方税号同时命中**该集合才入镜像、参与绑定；我方主体税号同样维护在客商档案中，故进项票要求「购方=我方、销方=建档客商」同时成立。白名单外发票不入镜像不绑定；**集合为空不限制**（档案未维护时兼容全量）。由此推论：**供应商未在客商档案维护统一社会信用代码，其进项票不会进票据池镜像，前端选票也就选不到**——档案维护是同步的前置条件，可用每页过滤明细日志的「客商税号白名单外 N 张」定位漏档。`buyerTaxNo`/`salerTaxNo` 均为 rim 契约单值参数，无法按税号集合远程查询，故采用窗口拉取 + 内存过滤的等价实现。 - **防漏绑补偿**：滚动窗口唯一的漏绑缝隙是「持续失败超过 N 天（窗口滑出 updateTime）」，第 ⑤ 步兜底——仍为未绑定（`kingdeeInvoicePoolId` 为空）的登记按发票号码精确补查 A801-1（优先复用本地镜像，缺失才远程），单轮限量（`compensateLimit`，默认 200，按创建时间先进先出），绑定成功即移出补查队列；金蝶票池暂无该发票时跳过不写库、待后续轮次。**手动同步与补偿绑定不受税号白名单约束**（票由登记侧发起，属我方票）。 - 提供手动触发端点 `POST /invoice-receipt/sync-invoice-pool`（权限码复用 `invoice_receipt_sync`），返回执行摘要，供联调与补偿。 @@ -98,9 +98,9 @@ InvoiceReceiptKingdeeSyncTask InvoiceReceiptServiceImpl | --- | --- | | 匹配键 | `invoiceNo`（发票号码）。收票登记侧 `assertInvoiceUnused` 已保证一张发票仅可登记一次，镜像侧按 `invoiceNo` upsert 天然幂等 | | 发票种类白名单 | `thirdParty.bfi.invoiceReceipt.invoiceTypes`（逗号分隔票易通编码，如 `26,27`）控制可同步的发票种类：白名单外发票**不落镜像不绑定**（手动同步记「发票类型不在同步范围」流水，定时拉取在类型过滤统计中体现）；配置为空不限制。定时拉取翻页终止仍按原始页条数判断，不受过滤影响 | -| 客商税号白名单 | 客商档案有效税号（`unifiedCreditCode`，status=1）为白名单，窗口拉回的发票须**购方税号与销方税号同时命中**才入镜像、参与绑定；**集合为空不限制**。与发票种类白名单叠加生效；手动同步与补偿绑定不受此约束 | -| 镜像批量写库 | 同页镜像**一次批量查询**既有记录（`invoiceNo IN 同页号码`）+ `Db.saveBatch` / `Db.updateBatchById` **批量写入**（避免逐条查改 N+1）；批量写失败中断本轮，由任务层记录异常、下轮重试（upsert 幂等） | -| 绑定批量落库 | 绑定回填先在**内存**完成（登记实体按页批量查出，票面字段不覆盖），再以**页级事务**批量落库：`updateBatchById` 登记绑定 + `Db.saveBatch` 同步流水（`applyInvoicePoolSyncBatch`，JDBC batch）。批量失败**整页跳过**计入失败统计，下轮任务幂等重试，**无逐单降级路径**（`applyInvoicePoolSyncResult` 仅手动单条同步使用） | +| 客商税号白名单 | 客商档案有效税号（`unifiedCreditCode`，status=1）为白名单，窗口拉回的发票须**购方税号与销方税号同时命中**才入镜像、参与绑定（我方主体税号亦维护在客商档案中）；**集合为空不限制**。供应商未建档/未维护统一社会信用代码 → 其进项票不进镜像、前端选票选不到。与发票种类白名单叠加生效；手动同步与补偿绑定不受此约束 | +| 镜像批量写库 | 同页镜像**一次批量查询**既有记录（`invoiceNo IN 同页号码`）+ `Db.saveBatch` / `Db.updateBatchById` **批量写入**（避免逐条查改 N+1）；写入前按**票面字段指纹**比对（`attachments_json` 等大字段在内），**无变化的镜像不重写**，避免每轮整窗重写大字段；指纹中空值按「维持库中原值」处理，与 MyBatis-Plus 非空字段更新策略一致，防止源侧空字段每轮误判为变更。批量写失败中断本轮，由任务层记录异常、下轮重试（upsert 幂等） | +| 绑定批量落库 | 绑定回填先在**内存**完成（登记实体按页批量查出，票面字段不覆盖），再逐项比对 `kingdeeInvoicePoolId`/`kingdeeBillNo`/`kingdeeStatus`，**三项均无变化的登记不更新、不计入统计**；有变化的以**页级事务**批量落库：`updateBatchById` 登记绑定 + `Db.saveBatch` 同步流水（`applyInvoicePoolSyncBatch`，JDBC batch）。**流水只在绑定关系变化（原因「定时绑定金蝶票池发票」）或金蝶状态迁移（原因「定时刷新金蝶票池状态」）时写入**，纯流水号回填不刷流水。批量失败**整页跳过**计入失败统计，下轮任务幂等重试，**无逐单降级路径**（`applyInvoicePoolSyncResult` 仅手动单条同步使用） | | 防漏绑补偿 | 增量窗口为滚动近 N 天，单轮失败不漏；唯一缝隙是「持续失败超 N 天窗口滑出」。补偿绑定步骤对未绑定登记（`kingdeeInvoicePoolId` 为空、非作废）批量补绑：本地镜像**一次批量查询**复用（主流程本轮已落库镜像命中免远程），缺失才按发票号码**精确补查** A801-1（rim 契约无批量号码参数，仅此处逐号远程），新增镜像批量写入 + 绑定页级事务批量落库；单轮限量 `compensateLimit`（默认 200）先进先出，绑定成功即出队；票池暂无则跳过待下轮，不写库不刷流水 | | 手动同步查询 | A801-1 `invoiceNo` 精确单查（rim 契约单页，返回 0/1 条）；多条理论上不存在，取第一条并 warn 日志 | | 定时绑定范围 | `approvalStatus IN (draft, reviewing, approved, returned)` 且（`kingdeeInvoicePoolId` 为空 或 指向镜像不存在的记录）；`voided` 不参与绑定 | @@ -109,7 +109,7 @@ InvoiceReceiptKingdeeSyncTask InvoiceReceiptServiceImpl | 金额校验 | 绑定时若票池价税合计与登记 `invoiceAmount` 不一致仅记 warn 日志（跨系统开票金额可能变更），不阻断绑定 | | 已绑定状态刷新 | 跟随镜像刷新 `kingdeeStatus`；票池 `delete=1`（作废删除）置 `failed`，流水记录原因，**不自动作废**收票登记，人工确认处理 | | 逐单隔离 | 定时链路无逐单路径：镜像批量查改 + 绑定页级批量落库；仅批量落库异常时整页跳过、下轮幂等重试 | -| 关键日志 | 任务开始（增量窗口 + 发票种类 + 客商税号白名单个数）、每页拉取条数、镜像批量写入条数（新增/更新）、逐条绑定成功（发票号码 + 登记id + 金蝶流水号）、绑定异常（含堆栈）、结束汇总摘要（拉取/类型过滤/客商过滤/镜像新增更新/绑定/状态刷新/失败/补偿绑定） | +| 关键日志 | 任务开始（增量窗口 + 发票种类 + 客商税号白名单个数 + 单页/最大页数）、每页拉取条数（含 BFI 侧远程耗时）、镜像批量写入条数（新增/更新/无变化）、每页落库小结（变更/绑定/状态刷新/流水条数）、绑定成功逐条（发票号码 + 登记id + 金蝶流水号）、落库异常（含页码与堆栈）、翻页触顶 warn（窗口尾部本轮未处理）、补偿小结（补偿绑定数 + 本地镜像命中 + 远程按号补查次数 + 票池无票清单 + 耗时）、锁跳过 warn、未启用分布式锁 warn、结束汇总摘要（翻页/拉取/类型过滤/客商过滤/镜像新增更新/绑定/状态刷新/登记无变化跳过/失败/补偿绑定/耗时） | ### 3.5 状态映射（kingdeeStatus 语义） @@ -170,6 +170,11 @@ InvoiceReceiptKingdeeSyncTask InvoiceReceiptServiceImpl | thirdParty.bfi.invoiceReceipt.incrementDays | BFI_INVOICE_RECEIPT_INCREMENT_DAYS | 30 | 首轮/兜底增量窗口天数 | | thirdParty.bfi.invoiceReceipt.invoiceTypes | BFI_INVOICE_RECEIPT_INVOICE_TYPES | 空 | 同步发票种类白名单（逗号分隔票易通编码，如 26,27 数电普票/专票），为空不限制 | | thirdParty.bfi.invoiceReceipt.compensateLimit | BFI_INVOICE_RECEIPT_COMPENSATE_LIMIT | 200 | 补偿绑定单轮限量（未绑定登记按发票号码精确补查条数上限） | +| thirdParty.bfi.invoiceReceipt.lockLeaseSeconds | BFI_INVOICE_RECEIPT_LOCK_LEASE_SECONDS | 1800 | 全票池同步 single-flight 锁租约秒数，须大于最坏一轮耗时；进程异常退出时锁在租约到期后自动释放 | + +**数据源前置条件**：`doc/nacos/blade-dev.yaml`、`blade-test.yaml`、`blade-prod.yaml` 的 MySQL 连接串须带 `rewriteBatchedStatements=true`，否则 `Db.saveBatch` / `updateBatchById` 仍是逐条往返、不构成真正的 JDBC 批量（该参数为 MySQL 专有，切换其他数据库需移除）。配置变更后需重启服务生效。 + +**分布式锁前置条件**：`RedisLockClient` 仅在 `blade.lock.enabled=true` 时装配（该开关无缺省值，不配即关闭）。dev/test 模板已开启，且 `blade.lock.address` 与 `spring.data.redis` 指向同一实例；prod 模板保持关闭，开启前须先补 `blade.lock.password`（与 `spring.data.redis` 一致），否则 Redisson 建连被拒会拖垮依赖该模板的所有服务。锁未启用时全票池同步照常执行，只是失去多副本 single-flight 保护并打 warn 日志。 ## 7. 异常处理 @@ -179,7 +184,7 @@ InvoiceReceiptKingdeeSyncTask InvoiceReceiptServiceImpl | 手动同步按发票号未查到 | 登记 `kingdeeStatus=unsynced`，流水记录「金蝶票池未查询到」，不抛异常中断批量 | | 绑定时金额不一致 | warn 日志（发票号 + 双方金额），仍按 3.4 口径绑定 | | 单条绑定/状态刷新异常 | 记录 error 日志（含发票号与堆栈），计数后继续下一单 | -| 定时任务与手动同步并发 | 镜像按 invoiceNo upsert、登记按主键 `updateById`、绑定判重幂等，最终一致 | +| 定时任务与手动同步并发 | 链路二入口按租户加 Redis 分布式锁做 **single-flight**（`RedisLockClient.tryLock(key, REENTRANT, 0, lockLeaseSeconds)`，零等待）：多副本调度、手动端点撞上进行中的轮次时，抢不到锁的一侧直接跳过并返回「同步进行中，本轮跳过」，不重复拉取与写库；未开启 `blade.lock.enabled` 时降级为无锁执行并 warn。锁之外仍靠镜像按 invoiceNo upsert、登记按主键更新、绑定判重幂等保证最终一致 | | A801-1 翻页异常（rim 无总数） | 以「本页条数 < pageSize 或空页」终止，最大 50 页保护，防御死循环 | ## 8. 涉及代码位置 @@ -188,14 +193,14 @@ InvoiceReceiptKingdeeSyncTask InvoiceReceiptServiceImpl | --- | --- | --- | | 前端页面 | `tms-erp-web/src/views/payment/invoice-receipt.vue`（handleBatchSync）、`src/api/payment/invoiceReceipt.js`（syncKingdee） | 无需改动（复用现有单条同步接口与按钮） | | 同步入口改造 | `blade-service/blade-transport/.../service/impl/InvoiceReceiptServiceImpl.java`（syncKingdee 接入 A801-1 查询 + 镜像 upsert + 绑定回填） | 修改 | -| 定时绑定编排 | 同上 `syncKingdeeInvoicePool()`（增量窗口分页拉取 + upsert + 绑定）、`applyInvoicePoolSyncResult()`（逐单独立事务：绑定回填 + 同步流水） | 新增 | +| 定时绑定编排 | 同上 `syncKingdeeInvoicePool()`（RedisLockClient single-flight + 租户锁键）、`executeInvoicePoolSync()`（增量窗口分页拉取 + 票面指纹比对 upsert + 变更比对绑定）、`applyInvoicePoolSyncBatch()`（页级事务：绑定批量更新 + 同步流水批量写入）；`applyInvoicePoolSyncResult()` 仅手动单条同步使用 | 新增 | | Service 接口 | `.../service/IInvoiceReceiptService.java`（syncKingdeeInvoicePool） | 新增 | | 定时任务 | `blade-service/blade-transport/.../task/InvoiceReceiptKingdeeSyncTask.java` | 新增（仿 PaymentKingdeeResultTask） | | 手动触发端点 | `.../controller/InvoiceReceiptController.java`（POST /invoice-receipt/sync-invoice-pool） | 新增（可选） | | 镜像实体 | `blade-transport-api/.../pojo/entity/KingdeeInvoicePool.java`（serialNo/salerTaxNo/buyerTaxNo/checkStatus） | 修改 | | BFI 客户端 | `blade-third-party-api/blade-bfi-api/.../service/IBfiService.java`（queryInvoicePool）、`pojo/dto/BfiInvoicePoolQueryDataDTO.java`、`pojo/vo/BfiInvoicePoolVO.java` | 无改动复用 | | SQL | `doc/sql/transport/`（镜像表新增列 + 索引幂等脚本） | 新增 | -| Nacos 配置模板 | `doc/nacos/blade-dev.yaml`、`blade-prod.yaml`、`third-party-api.yaml`（thirdParty.bfi.invoiceReceipt.*） | 新增 | +| Nacos 配置模板 | `doc/nacos/blade-dev.yaml`、`blade-test.yaml`、`blade-prod.yaml`（thirdParty.bfi.invoiceReceipt.*、MySQL `rewriteBatchedStatements`、`blade.lock.enabled`）、`third-party-api.yaml` | 新增 | ## 9. 遗留事项与联调确认 diff --git a/doc/nacos/blade-dev.yaml b/doc/nacos/blade-dev.yaml index e142c598..7de03d27 100644 --- a/doc/nacos/blade-dev.yaml +++ b/doc/nacos/blade-dev.yaml @@ -32,7 +32,8 @@ spring: blade: #分布式锁配置 lock: - enabled: false + # blade.lock.enabled=true 才会装配 RedisLockClient；收票全票池同步等定时任务依赖它做 single-flight + enabled: true address: redis://127.0.0.1:6379 #多团队协作服务配置 loadbalancer: @@ -46,7 +47,8 @@ blade: datasource: dev: # MySql - url: jdbc:mysql://localhost:3306/transport?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&tinyInt1isBit=false&allowMultiQueries=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true + # rewriteBatchedStatements=true：MyBatis-Plus saveBatch/updateBatchById 依赖该参数才会合并为多值语句，否则仍是逐条往返（仅 MySQL 支持，切换其他库需移除） + url: jdbc:mysql://localhost:3306/transport?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&tinyInt1isBit=false&allowMultiQueries=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true&rewriteBatchedStatements=true username: root password: root # PostgreSQL diff --git a/doc/nacos/blade-prod.yaml b/doc/nacos/blade-prod.yaml index f706c11a..db12adcd 100644 --- a/doc/nacos/blade-prod.yaml +++ b/doc/nacos/blade-prod.yaml @@ -35,14 +35,17 @@ blade: #分布式锁配置 lock: ##是否启用分布式锁 + ##开启前必须先补 password(与 spring.data.redis 一致)，否则 Redisson 无密码连接会被拒绝， + ##影响的是全部依赖该模板的服务；收票全票池等同步定时任务依赖它做多副本 single-flight enabled: false + #password: ##将docker脚本部署的redis服务映射为宿主机ip ##生产环境推荐使用阿里云高可用redis服务并设置密码 address: redis://172.16.203.228:6379 #通用开发生产环境数据库地址(特殊情况可在对应的子工程里配置覆盖) datasource: prod: - url: jdbc:mysql://172.16.203.228:3306/transport?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&tinyInt1isBit=false&allowMultiQueries=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true + url: jdbc:mysql://172.16.203.228:3306/transport?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&tinyInt1isBit=false&allowMultiQueries=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true&rewriteBatchedStatements=true username: transport password: ab3acJYSC8A4ySpk diff --git a/doc/nacos/blade-test.yaml b/doc/nacos/blade-test.yaml index 2839eb5a..dbb91f12 100644 --- a/doc/nacos/blade-test.yaml +++ b/doc/nacos/blade-test.yaml @@ -33,14 +33,14 @@ spring: blade: #分布式锁配置 lock: - ##是否启用分布式锁 - enabled: false + ##是否启用分布式锁(收票全票池同步等定时任务依赖 RedisLockClient 做 single-flight，多副本部署必须开启) + enabled: true ##将docker脚本部署的redis服务映射为宿主机ip address: redis://192.168.0.188:3379 #通用开发生产环境数据库地址(特殊情况可在对应的子工程里配置覆盖) datasource: test: - url: jdbc:mysql://192.168.0.188:3306/bladex?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&tinyInt1isBit=false&allowMultiQueries=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true + url: jdbc:mysql://192.168.0.188:3306/bladex?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&transformedBitIsBoolean=true&tinyInt1isBit=false&allowMultiQueries=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true&rewriteBatchedStatements=true username: root password: root