feat(tax-doc): 支持系统数据ID传递并优化税务映射逻辑

- 修改 Builder 接口及其实现类，增加 systemDataId 参数支持 - 引入 ITaxPipelineMappingService服务用于查询现有映射关系- 在 CustomerBuilder 中根据 systemDataId 查询并设置客户ID - 扩展 TaxPipelineMappingCreate 模型以支持更新操作 - 实现 DynamicTaxPipelineMappingService 的 update 方法 - 调整 TaxPipelineMappingRepository 及其 XML 配置以区分更新状态与完整更新 -重构 TaxReceiver 和 TaxReceiveService以适配新的处理流程 - 在 recordMapping 方法中实现插入或更新税务映射记录的逻辑

feat(tax-doc): 支持系统数据ID传递并优化税务映射逻辑
- 修改 Builder 接口及其实现类，增加 systemDataId 参数支持 - 引入 ITaxPipelineMappingService服务用于查询现有映射关系- 在 CustomerBuilder 中根据 systemDataId 查询并设置客户ID - 扩展 TaxPipelineMappingCreate 模型以支持更新操作 - 实现 DynamicTaxPipelineMappingService 的 update 方法 - 调整 TaxPipelineMappingRepository 及其 XML 配置以区分更新状态与完整更新 -重构 TaxReceiver 和 TaxReceiveService以适配新的处理流程 - 在 recordMapping 方法中实现插入或更新税务映射记录的逻辑
zhangmeiyang
1 parent 9975a942
Showing 13 changed files with 91 additions and 34 deletions
tax-boot/src/main/java/com/diligrp/tax/boot/receiver/TaxReceiver.java
tax-boot/src/main/java/com/diligrp/tax/boot/service/TaxReceiveService.java
tax-central/src/main/java/com/diligrp/tax/central/model/TaxPipelineMappingCreate.java
tax-central/src/main/java/com/diligrp/tax/central/service/ITaxPipelineMappingService.java
tax-doc/src/main/java/com/diligrp/tax/doc/demarcate/Builder.java
tax-doc/src/main/java/com/diligrp/tax/doc/demarcate/kingdee/CustomerBuilder.java
tax-doc/src/main/java/com/diligrp/tax/doc/demarcate/kingdee/ReceiptBuilder.java
tax-doc/src/main/java/com/diligrp/tax/doc/demarcate/kingdee/ReceivableBuilder.java
tax-doc/src/main/java/com/diligrp/tax/doc/demarcate/kingdee/RefundBuilder.java
tax-doc/src/main/java/com/diligrp/tax/doc/process/kingdee/InitializeProcessor.java
tax-storage/src/main/java/com/diligrp/tax/storage/repo/TaxPipelineMappingRepository.java
tax-storage/src/main/java/com/diligrp/tax/storage/service/DynamicTaxPipelineMappingService.java
tax-storage/src/main/resources/com/diligrp/tax/storage/repo/TaxPipelineMappingRepository.xml
@@ -2,6 +2,8 @@ package com.diligrp.tax.boot.receiver;
 import com.diligrp.tax.boot.service.TaxReceiveService;
 import com.diligrp.tax.central.domain.MessageContext;
+import com.diligrp.tax.central.process.ProcessorChain;
+import com.diligrp.tax.central.type.SystemType;
 import com.rabbitmq.client.Channel;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
@@ -15,6 +17,7 @@ import org.springframework.transaction.annotation.Transactional;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Map;
 import java.util.Optional;
 import static com.diligrp.tax.boot.queue.TaxAutoPush.*;
@@ -32,6 +35,9 @@ public class TaxReceiver {
     @Resource
     private TaxReceiveService taxReceiveService;
+    @Resource
+    private Map<SystemType, ProcessorChain> processorChainMap;
+
     @RabbitListener(bindings =
     @QueueBinding(
             value = @Queue(value = NORMAL_QUEUE, autoDelete = "false"),
@@ -45,8 +51,9 @@ public class TaxReceiver {
         log.info("tax-agent收到消息:{}", content);
         MessageContext ctx = null;
         try {
-            ctx = taxReceiveService.messageHandle(content);
-            taxReceiveService.recordMapping(ctx);
+            ctx = taxReceiveService.messagePreHandle(content);
+            var trans = processorChainMap.get(SystemType.from(ctx.getSystemType())).startProcess(ctx);
+            taxReceiveService.recordMapping(trans);
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
             log.info("tax-agent消息处理成功:{}", content);
         } catch (Exception e) {
@@ -5,7 +5,7 @@ import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.model.TaxPipelineMappingCreate;
 import com.diligrp.tax.central.model.TenantPipeline;
-import com.diligrp.tax.central.process.ProcessorChain;
+import com.diligrp.tax.central.model.TenantTaxPipelineMapping;
 import com.diligrp.tax.central.service.ITaxPipelineMappingService;
 import com.diligrp.tax.central.service.ITenantTaxService;
 import com.diligrp.tax.central.type.DocumentType;
@@ -19,7 +19,6 @@ import org.springframework.transaction.annotation.Transactional;
 import java.io.PrintWriter;
 import java.io.StringWriter;
-import java.util.Map;
 import java.util.Optional;
 /**
@@ -31,17 +30,16 @@ import java.util.Optional;
 public class TaxReceiveService {
     @Resource
-    private Map<SystemType, ProcessorChain> processorChainMap;
-
-    @Resource
     private ITenantTaxService tenantTaxService;
     @Resource
     private ITaxPipelineMappingService taxPipelineMappingService;
-    public MessageContext messageHandle(String content) {
+    public MessageContext messagePreHandle(String content) {
         MessageContext ctx = JsonUtils.fromJsonString(content, MessageContext.class);
+        Long tenantId = tenantTaxService.getTenantId(ctx.getGroup(), ctx.getEntity());
+        ctx.setTenantId(tenantId);
         DocumentType from = DocumentType.from(ctx.getDocumentType());
         SystemType system = SystemType.from(ctx.getSystemType());
         Optional.of(system).filter(s -> s.documentTypes.contains(from)).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.BUSINESS_MATCHES_ARE_INCORRECT));
@@ -54,19 +52,31 @@ public class TaxReceiveService {
         ctx.setTenantPipeline(pipeline);
         //获取租户id
         ctx.setTenantId(tenantTaxService.getTenantId(ctx.getGroup(), ctx.getEntity()));
-        return processorChainMap.get(SystemType.from(ctx.getSystemType())).startProcess(ctx);
+        return ctx;
     }
     @Transactional
     public void recordMapping(MessageContext messageContext) {
-        TaxPipelineMappingCreate create = new TaxPipelineMappingCreate();
-        create.setTenantId(messageContext.getTenantId());
-        create.setPipelineId(messageContext.getTenantPipeline().getId());
-        create.setDocumentType(messageContext.getDocumentType());
-        create.setSystemDataId(messageContext.getSystemDataId());
-        create.setState(MappingStateType.SYNCED.value);
-        create.setOriginData(messageContext.getMsgBody());
-        taxPipelineMappingService.insert(create);
+        Optional<TenantTaxPipelineMapping> dbOptions = taxPipelineMappingService.findByPipelineIdAndDocumentTypeAndSystemDataId(messageContext.getTenantPipeline().getTenantId(), messageContext.getTenantPipeline().getId(), messageContext.getDocumentType(), messageContext.getSystemDataId());
+        dbOptions.ifPresentOrElse(
+                e -> {
+                    TaxPipelineMappingCreate create = new TaxPipelineMappingCreate();
+                    create.setId(e.getId());
+                    create.setTenantId(messageContext.getTenantId());
+                    create.setOriginData(messageContext.getMsgBody());
+                    taxPipelineMappingService.update(create);
+                }, () -> {
+                    TaxPipelineMappingCreate create = new TaxPipelineMappingCreate();
+                    create.setTenantId(messageContext.getTenantId());
+                    create.setPipelineId(messageContext.getTenantPipeline().getId());
+                    create.setDocumentType(messageContext.getDocumentType());
+                    create.setSystemDataId(messageContext.getSystemDataId());
+                    create.setPipelineDataId(messageContext.getPipelineDataId());
+                    create.setState(MappingStateType.SYNCED.value);
+                    create.setOriginData(messageContext.getMsgBody());
+                    taxPipelineMappingService.insert(create);
+                });
+
     }
     @Transactional
@@ -16,6 +16,10 @@ import lombok.Setter;
 @Valid
 public class TaxPipelineMappingCreate {
     /**
+     * id
+     */
+    private Long id;
+    /**
      * 租户 ID
      */
     @NotNull(message = "租户ID不能为空")
@@ -3,6 +3,7 @@ package com.diligrp.tax.central.service;
 import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.model.TaxPipelineMappingCreate;
 import com.diligrp.tax.central.model.TenantTaxPipelineMapping;
+import org.springframework.validation.annotation.Validated;
 import java.util.Optional;
@@ -25,4 +26,11 @@ public interface ITaxPipelineMappingService {
      * @param taxPipelineMappingCreate
      */
     void insert(TaxPipelineMappingCreate taxPipelineMappingCreate);
+
+    /**
+     * 更新
+     *
+     * @param taxPipelineMappingCreate 税务管道映射创建
+     */
+    void update(TaxPipelineMappingCreate taxPipelineMappingCreate);
 }
@@ -21,9 +21,10 @@ public abstract class Builder&lt;T extends BaseDocument&gt; {
     /**
      * 建
      *
-     * @param body     身体
+     * @param body         身体
      * @param pipeline
+     * @param systemDataId
      * @return {@link T }
      */
-    public abstract T build(String body, TenantPipeline pipeline);
+    public abstract T build(String body, TenantPipeline pipeline, String systemDataId);
 }
@@ -5,10 +5,13 @@ import com.diligrp.tax.central.domain.document.kingdee.basic.item.Contact;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.model.TenantPipeline;
 import com.diligrp.tax.central.model.TenantPipelineConfig;
+import com.diligrp.tax.central.model.TenantTaxPipelineMapping;
+import com.diligrp.tax.central.service.ITaxPipelineMappingService;
 import com.diligrp.tax.central.type.DocumentType;
 import com.diligrp.tax.central.type.TaxSystemType;
 import com.diligrp.tax.central.utils.JsonUtils;
 import com.diligrp.tax.doc.demarcate.Builder;
+import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -29,19 +32,25 @@ import java.util.stream.Stream;
 @Slf4j
 public class CustomerBuilder extends Builder<StandardCustomer> {
+    @Resource
+    private ITaxPipelineMappingService taxPipelineMappingService;
+
+
     @Override
     public DocumentType markDocument() {
         return DocumentType.CUSTOMER_INFO;
     }
     @Override
-    public StandardCustomer build(String body, TenantPipeline pipeline) {
+    public StandardCustomer build(String body, TenantPipeline pipeline, String systemDataId) {
         StandardCustomer customer = JsonUtils.fromJsonString(body, StandardCustomer.class);
         Optional.ofNullable(customer.getContacts()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请完善联系人信息"));
-        Optional.ofNullable(customer.getSystemDataId()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写系统数据ID"));
+        Optional.ofNullable(systemDataId).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写系统数据ID"));
         List<TenantPipelineConfig> list = pipeline.getTenantPipelineConfigs().get(markDocument());
         Optional.ofNullable(list).ifPresent(ts -> configureDefaultData(customer, ts));
-        //TODO 查询数据库的客户信息 如果存在 赋值id
+        //查询数据库的客户信息 如果存在 赋值id
+        Optional<TenantTaxPipelineMapping> dbOptions = taxPipelineMappingService.findByPipelineIdAndDocumentTypeAndSystemDataId(pipeline.getTenantId(), pipeline.getId(), markDocument().value, systemDataId);
+        dbOptions.ifPresent(e -> customer.setCustomerId(e.getPipelineDataId()));
         return customer;
     }
@@ -33,7 +33,7 @@ public class ReceiptBuilder extends Builder&lt;ReceiptBill&gt; {
     }
     @Override
-    public ReceiptBill build(String body, TenantPipeline pipeline) {
+    public ReceiptBill build(String body, TenantPipeline pipeline, String systemDataId) {
         ReceiptBill bill = JsonUtils.fromJsonString(body, ReceiptBill.class);
         Optional.ofNullable(bill.getReceiptItems()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写收款项目"));
         bill.setCreateTime(Optional.ofNullable(bill.getCreateTime()).orElse(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));
 package com.diligrp.tax.doc.demarcate.kingdee;
-import com.diligrp.tax.central.domain.document.kingdee.bill.ReceiptBill;
 import com.diligrp.tax.central.domain.document.kingdee.bill.ReceivableBill;
-import com.diligrp.tax.central.domain.document.kingdee.bill.item.ReceiptItem;
 import com.diligrp.tax.central.domain.document.kingdee.bill.item.ReceivableItem;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.model.TenantPipeline;
@@ -10,7 +8,6 @@ import com.diligrp.tax.central.model.TenantPipelineConfig;
 import com.diligrp.tax.central.type.DocumentType;
 import com.diligrp.tax.central.type.TaxSystemType;
 import com.diligrp.tax.central.utils.JsonUtils;
-import com.diligrp.tax.central.utils.MappingUtils;
 import com.diligrp.tax.doc.demarcate.Builder;
 import org.springframework.stereotype.Component;
@@ -37,7 +34,7 @@ public class ReceivableBuilder extends Builder&lt;ReceivableBill&gt; {
     }
     @Override
-    public ReceivableBill build(String body, TenantPipeline pipeline) {
+    public ReceivableBill build(String body, TenantPipeline pipeline, String systemDataId) {
         ReceivableBill bill = JsonUtils.fromJsonString(body, ReceivableBill.class);
         Optional.ofNullable(bill.getReceivableItems()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写收款项目"));
         bill.setCreateTime(Optional.ofNullable(bill.getCreateTime()).orElse(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));
 package com.diligrp.tax.doc.demarcate.kingdee;
-import com.diligrp.tax.central.domain.document.kingdee.bill.ReceivableBill;
 import com.diligrp.tax.central.domain.document.kingdee.bill.RefundBill;
-import com.diligrp.tax.central.domain.document.kingdee.bill.item.ReceivableItem;
 import com.diligrp.tax.central.domain.document.kingdee.bill.item.RefundItem;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.model.TenantPipeline;
@@ -10,7 +8,6 @@ import com.diligrp.tax.central.model.TenantPipelineConfig;
 import com.diligrp.tax.central.type.DocumentType;
 import com.diligrp.tax.central.type.TaxSystemType;
 import com.diligrp.tax.central.utils.JsonUtils;
-import com.diligrp.tax.central.utils.MappingUtils;
 import com.diligrp.tax.doc.demarcate.Builder;
 import org.springframework.stereotype.Component;
@@ -37,7 +34,7 @@ public class RefundBuilder extends Builder&lt;RefundBill&gt; {
     }
     @Override
-    public RefundBill build(String body, TenantPipeline pipeline) {
+    public RefundBill build(String body, TenantPipeline pipeline, String systemDataId) {
         RefundBill bill = JsonUtils.fromJsonString(body, RefundBill.class);
         Optional.ofNullable(bill.getRefundItems()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写收款项目"));
         bill.setCreateTime(Optional.ofNullable(bill.getCreateTime()).orElse(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));
 package com.diligrp.tax.doc.process.kingdee;
+import com.diligrp.tax.central.domain.BaseDocument;
 import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.process.AbstractProcessor;
 import com.diligrp.tax.doc.context.DocumentContext;
@@ -23,7 +24,8 @@ public class InitializeProcessor extends AbstractProcessor {
     @Override
     public MessageContext process(MessageContext messageContext) {
         Builder<?> builder = DocumentContext.CONTEXT.get(messageContext.getDocumentTypeEnum());
-        messageContext.setDocumentObject(builder.build(messageContext.getMsgBody(), messageContext.getTenantPipeline()));
+        BaseDocument build = builder.build(messageContext.getMsgBody(), messageContext.getTenantPipeline(), messageContext.getSystemDataId());
+        messageContext.setDocumentObject(build);
         return messageContext;
     }
 }
@@ -25,7 +25,7 @@ public interface TaxPipelineMappingRepository {
      *
      * @param mapping 税务管道映射
      */
-    void update(TaxPipelineMapping mapping);
+    void updateStatus(TaxPipelineMapping mapping);
     /**
      * 按管道 ID 查找
@@ -56,4 +56,11 @@ public interface TaxPipelineMappingRepository {
      * @param tableName 表名
      */
     void dropTenantMappingTable(@Param("tableName") String tableName);
+
+    /**
+     * 更新
+     *
+     * @param taxPipelineMapping 税务管道映射
+     */
+    void update(TaxPipelineMapping taxPipelineMapping);
 }
@@ -88,4 +88,10 @@ public class DynamicTaxPipelineMappingService implements ITaxPipelineMappingServ
         TaxPipelineMapping taxPipelineMapping = JsonUtils.convertValue(taxPipelineMappingCreate, TaxPipelineMapping.class);
         taxPipelineMappingRepository.insert(taxPipelineMapping);
     }
+
+    @Override
+    public void update(TaxPipelineMappingCreate taxPipelineMappingCreate) {
+        TaxPipelineMapping taxPipelineMapping = JsonUtils.convertValue(taxPipelineMappingCreate, TaxPipelineMapping.class);
+        taxPipelineMappingRepository.update(taxPipelineMapping);
+    }
 }
@@ -19,7 +19,7 @@
             , #{state}
             , #{errorMessage})
     </insert>
-    <update id="update">
+    <update id="updateStatus">
         UPDATE
             tax_pipeline_mapping_${tenantId}
         SET
@@ -96,5 +96,14 @@
     <update id="dropTenantMappingTable">
         DROP TABLE IF EXISTS ${tableName}
     </update>
+    <update id="update">
+        UPDATE
+            tax_pipeline_mapping_${tenantId}
+        SET
+            origin_data   = #{originData}
+          , error_message = #{errorMessage}
+        WHERE
+            id = #{id}
+    </update>
 </mapper>