feat(storage): 新增税务管道映射功能及相关服务

- 添加 DynamicTaxPipelineMappingController 控制器，提供分页查询接口 - 实现 DynamicTaxPipelineMappingService 服务类，支持创建租户映射表、查询管道数据等功能 - 新增 TaxPipelineMappingCO 和 TaxPipelineMappingCreateCO 数据传输对象 - 定义 MappingStateType 枚举，表示映射状态（已同步、同步失败、同步重试） - 增加 GlobalExceptionHandler 统一异常处理机制- 扩展 ITaxPipelineMappingService 接口定义- 在 TaxPipelineConfigController 和 TaxPipelineController 中添加 @RebuildIndex 注解 - 修改 TaxPipelineCO、TaxPipelineConfigCO、TaxTenantCO 验证组配置 - 更新 TaxPipelineMappingRepository 接口及 XML 映射文件，支持动态表操作 - 添加 RebuildIndexAspect 切面类，用于发布 RestoreTenantEvent 事件 - 引入 spring-boot-starter-aop依赖以支持切面编程 - 调整 PageQuery 类，增加分页参数验证规则 -优化 TaxTenantService，在保存租户时自动创建映射表- 添加 TenantTaxPipelineMapping 模型类用于中央模块使用

feat(storage): 新增税务管道映射功能及相关服务
- 添加 DynamicTaxPipelineMappingController 控制器，提供分页查询接口 - 实现 DynamicTaxPipelineMappingService 服务类，支持创建租户映射表、查询管道数据等功能 - 新增 TaxPipelineMappingCO 和 TaxPipelineMappingCreateCO 数据传输对象 - 定义 MappingStateType 枚举，表示映射状态（已同步、同步失败、同步重试） - 增加 GlobalExceptionHandler 统一异常处理机制- 扩展 ITaxPipelineMappingService 接口定义- 在 TaxPipelineConfigController 和 TaxPipelineController 中添加 @RebuildIndex 注解 - 修改 TaxPipelineCO、TaxPipelineConfigCO、TaxTenantCO 验证组配置 - 更新 TaxPipelineMappingRepository 接口及 XML 映射文件，支持动态表操作 - 添加 RebuildIndexAspect 切面类，用于发布 RestoreTenantEvent 事件 - 引入 spring-boot-starter-aop依赖以支持切面编程 - 调整 PageQuery 类，增加分页参数验证规则 -优化 TaxTenantService，在保存租户时自动创建映射表- 添加 TenantTaxPipelineMapping 模型类用于中央模块使用
zhangmeiyang
1 parent c880414c
Showing 26 changed files with 802 additions and 45 deletions
tax-central/src/main/java/com/diligrp/tax/central/model/TenantTaxPipelineMapping.java
tax-central/src/main/java/com/diligrp/tax/central/service/ITaxPipelineMappingService.java
tax-storage/pom.xml
tax-storage/src/main/java/com/diligrp/tax/storage/GlobalExceptionHandler.java
tax-storage/src/main/java/com/diligrp/tax/storage/Valid.java
tax-storage/src/main/java/com/diligrp/tax/storage/anno/RebuildIndex.java
tax-storage/src/main/java/com/diligrp/tax/storage/anno/RebuildIndexAspect.java
tax-storage/src/main/java/com/diligrp/tax/storage/controller/DynamicTaxPipelineMappingController.java
tax-storage/src/main/java/com/diligrp/tax/storage/controller/TaxPipelineConfigController.java
tax-storage/src/main/java/com/diligrp/tax/storage/controller/TaxPipelineController.java
tax-storage/src/main/java/com/diligrp/tax/storage/controller/TaxTenantController.java
tax-storage/src/main/java/com/diligrp/tax/storage/domain/TaxPipelineMapping.java
tax-storage/src/main/java/com/diligrp/tax/storage/message/PageQuery.java
tax-storage/src/main/java/com/diligrp/tax/storage/model/co/TaxPipelineCO.java
tax-storage/src/main/java/com/diligrp/tax/storage/model/co/TaxPipelineConfigCO.java
tax-storage/src/main/java/com/diligrp/tax/storage/model/co/TaxPipelineMappingCO.java
tax-storage/src/main/java/com/diligrp/tax/storage/model/co/TaxPipelineMappingCreateCO.java
tax-storage/src/main/java/com/diligrp/tax/storage/model/co/TaxTenantCO.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
+package com.diligrp.tax.central.model;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDateTime;
+
+@Getter
+@Setter
+public class TenantTaxPipelineMapping {
+    /**
+     * id
+     */
+    private Long id;
+    /**
+     * 管道 ID
+     */
+    private Long pipelineId;
+    /**
+     * 文档类型
+     */
+    private String documentType;
+    /**
+     * 第三方系统数据 ID
+     */
+    private String systemDataId;
+    /**
+     * 所在账套数据 ID
+     */
+    private String pipelineDataId;
+    /**
+     * 原产地数据
+     */
+    private String originData;
+    /**
+     * 州
+     */
+    private Integer state;
+    /**
+     * 创建时间
+     */
+    private LocalDateTime createdTime;
+    /**
+     * 修改时间
+     */
+    private LocalDateTime modifiedTime;
+}
+package com.diligrp.tax.central.service;
+
+import com.diligrp.tax.central.model.TenantTaxPipelineMapping;
+
+import java.util.Optional;
+
+public interface ITaxPipelineMappingService {
+
+    /**
+     * 按管道 ID 和文档类型以及系统数据 ID 查找
+     *
+     * @param tenantId     租户 ID
+     * @param pipelineId   管道 ID
+     * @param documentType 文档类型
+     * @param systemDataId 系统数据 ID
+     * @return {@link Optional }<{@link TenantTaxPipelineMapping }>
+     */
+    Optional<TenantTaxPipelineMapping> findByPipelineIdAndDocumentTypeAndSystemDataId(Long tenantId,Long pipelineId,String documentType,String systemDataId);
+}
@@ -25,5 +25,9 @@
             <groupId>com.baomidou</groupId>
             <artifactId>mybatis-plus-spring-boot3-starter</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-aop</artifactId>
+        </dependency>
     </dependencies>
 </project>
+package com.diligrp.tax.storage;
+
+import com.diligrp.tax.central.exception.TaxAgentServiceException;
+import com.diligrp.tax.central.type.TaxSystemType;
+import com.diligrp.tax.storage.message.Message;
+import com.fasterxml.jackson.databind.exc.InvalidFormatException;
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.ConstraintViolationException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.support.DefaultMessageSourceResolvable;
+import org.springframework.http.converter.HttpMessageNotReadableException;
+import org.springframework.validation.BindException;
+import org.springframework.validation.BindingResult;
+import org.springframework.validation.ObjectError;
+import org.springframework.web.bind.MethodArgumentNotValidException;
+import org.springframework.web.bind.MissingServletRequestParameterException;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * 统一异常处理器
+ */
+@ControllerAdvice
+@Slf4j
+public class GlobalExceptionHandler {
+    /**
+     * 处理业务异常
+     */
+    @ExceptionHandler(TaxAgentServiceException.class)
+    public Message<?> handleBusinessException(TaxAgentServiceException e) {
+        log.error("Method throw TaxAgentServiceException", e);
+        return Message.failure(e.getCode(), e.getMessage());
+    }
+
+    /**
+     * 统一拦截参数转换异常 JSON -> Object
+     *
+     * @param ex InvalidFormatException
+     * @return 响应消息对象
+     */
+    @ExceptionHandler({HttpMessageNotReadableException.class, InvalidFormatException.class})
+    public Message<?> invalidFormatExceptionHandler(Exception ex) {
+        return Message.failure(TaxSystemType.ABNORMAL_PARAMETERS.code, ex.getMessage());
+    }
+
+    /**
+     * 参数验证
+     *
+     * @param exs 参数异常
+     * @return BaseOutput
+     */
+    @ExceptionHandler(value = ConstraintViolationException.class)
+    public Message<?> constraintViolationException(ConstraintViolationException exs) {
+        Set<ConstraintViolation<?>> violations = exs.getConstraintViolations();
+        String msg = violations.stream().map(ConstraintViolation::getMessage)
+                .collect(Collectors.joining(","));
+        return Message.failure(TaxSystemType.ABNORMAL_PARAMETERS.code, msg);
+    }
+
+    /**
+     * 参数验证
+     *
+     * @param e MethodArgumentNotValidException @RequestBody
+     *          BindException 普通form表单
+     * @return BaseOutput
+     */
+    @ExceptionHandler(value = {MethodArgumentNotValidException.class, BindException.class})
+    public Message<?> methodArgumentNotValidException(Exception e) {
+        BindingResult bindingResult;
+        if (e instanceof MethodArgumentNotValidException) {
+            bindingResult = ((MethodArgumentNotValidException) e).getBindingResult();
+        } else {
+            bindingResult = ((BindException) e).getBindingResult();
+        }
+        List<ObjectError> allErrors = bindingResult.getAllErrors();
+        String msg = allErrors.stream().map(DefaultMessageSourceResolvable::getDefaultMessage)
+                .collect(Collectors.joining("、"));
+
+        return Message.failure(TaxSystemType.ABNORMAL_PARAMETERS.code, msg);
+    }
+
+    /**
+     * 缺少请求参数异常
+     *
+     * @param ex 请求参数异常
+     * @return 异常信息
+     */
+    @ExceptionHandler(MissingServletRequestParameterException.class)
+    public Message<?> missingServletRequestParameterExceptionHandler(MissingServletRequestParameterException ex) {
+        return Message.failure(TaxSystemType.ABNORMAL_PARAMETERS.code, String.format("缺少参数：%s", ex.getParameterName()));
+    }
+
+    /**
+     * 断言验证
+     *
+     * @param ex IllegalArgumentException
+     * @return BaseOutput
+     */
+    @ExceptionHandler(value = IllegalArgumentException.class)
+    public Message<?> illegalArgumentException(IllegalArgumentException ex) {
+        return Message.failure(TaxSystemType.ABNORMAL_PARAMETERS.code, ex.getMessage());
+    }
+
+    /**
+     * 统一拦截系统其它异常
+     *
+     * @param ex 系统异常
+     * @return 响应消息对象
+     */
+    @ExceptionHandler(Exception.class)
+    public Message<?> exceptionHandler(Exception ex) {
+        log.error("unknown error :", ex);
+        return Message.failure(TaxSystemType.UNKNOWN_ERROR.code, TaxSystemType.UNKNOWN_ERROR.message);
+    }
+}
+package com.diligrp.tax.storage;
+
+/**
+ * 标记接口
+ *
+ * @author zhangmeiyang
+ * @date 2025/11/07
+ */
+public interface Valid {
+    interface Create {
+    }
+
+    interface Read {
+    }
+
+    interface Update {
+    }
+
+    interface Delete {
+    }
+}
+package com.diligrp.tax.storage.anno;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface RebuildIndex {
+}
+package com.diligrp.tax.storage.anno;
+
+import com.diligrp.tax.central.event.RestoreTenantEvent;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.aspectj.lang.annotation.After;
+import org.aspectj.lang.annotation.Aspect;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-07 14:06
+ * @Version: todo
+ */
+@Aspect
+@Component
+@Slf4j
+public class RebuildIndexAspect {
+
+    @Resource
+    private ApplicationEventPublisher publisher;
+
+    /**
+     * 防重
+     *
+     * @return
+     * @date 2020/8/12
+     */
+    @After("@annotation(com.diligrp.tax.storage.anno.RebuildIndex)")
+    public void rebuildIndex() {
+        log.info("开始重建缓存");
+        publisher.publishEvent(new RestoreTenantEvent(this));
+    }
+}
+package com.diligrp.tax.storage.controller;
+
+import com.diligrp.tax.storage.Valid;
+import com.diligrp.tax.storage.message.Message;
+import com.diligrp.tax.storage.model.co.TaxPipelineMappingCO;
+import com.diligrp.tax.storage.service.DynamicTaxPipelineMappingService;
+import jakarta.annotation.Resource;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-07 11:14
+ * @Version: todo
+ */
+@RestController
+@RequestMapping("/tax/tenant/pipeline/mapping")
+public class DynamicTaxPipelineMappingController {
+
+    @Resource
+    private DynamicTaxPipelineMappingService dynamicTaxPipelineMappingService;
+
+    @RequestMapping("page")
+    public Message<?> page(@RequestBody @Validated(value = {Valid.Read.class}) TaxPipelineMappingCO taxPipelineMappingCO) {
+        return Message.success(dynamicTaxPipelineMappingService.findTenantPipelineData(taxPipelineMappingCO));
+    }
+
+}
 package com.diligrp.tax.storage.controller;
  
+import com.diligrp.tax.storage.Valid;
+import com.diligrp.tax.storage.anno.RebuildIndex;
 import com.diligrp.tax.storage.message.Message;
 import com.diligrp.tax.storage.model.co.TaxPipelineConfigCO;
 import com.diligrp.tax.storage.service.TaxPipelineConfigService;
@@ -54,18 +56,33 @@ public class TaxPipelineConfigController {
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("/save")
-    public Message<?> save(@RequestBody @Validated TaxPipelineConfigCO taxPipelineConfig) {
+    @RebuildIndex
+    public Message<?> save(@RequestBody @Validated(value = Valid.Create.class) TaxPipelineConfigCO taxPipelineConfig) {
         taxPipelineConfigService.save(taxPipelineConfig);
         return Message.success();
     }
  
     /**
+     * 更新
+     *
+     * @param taxPipelineConfig 税务管道配置
+     * @return {@link Message }<{@link ? }>
+     */
+    @RequestMapping("/update")
+    @RebuildIndex
+    public Message<?> update(@RequestBody @Validated(value = {Valid.Update.class}) TaxPipelineConfigCO taxPipelineConfig) {
+        taxPipelineConfigService.update(taxPipelineConfig);
+        return Message.success();
+    }
+
+    /**
      * 删除
      *
      * @param id id
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("/delete/{id}")
+    @RebuildIndex
     public Message<?> delete(@PathVariable("id") Long id) {
         taxPipelineConfigService.delete(id);
         return Message.success();
 package com.diligrp.tax.storage.controller;
  
+import com.diligrp.tax.storage.Valid;
+import com.diligrp.tax.storage.anno.RebuildIndex;
 import com.diligrp.tax.storage.message.Message;
 import com.diligrp.tax.storage.model.co.TaxPipelineCO;
 import com.diligrp.tax.storage.service.TaxPipelineService;
@@ -30,12 +32,26 @@ public class TaxPipelineController {
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("/save")
-    public Message<?> save(@RequestBody @Validated TaxPipelineCO taxPipeline) {
+    @RebuildIndex
+    public Message<?> save(@RequestBody @Validated(value = Valid.Create.class) TaxPipelineCO taxPipeline) {
         taxPipelineService.save(taxPipeline);
         return Message.success();
     }
  
     /**
+     * 保存账套
+     *
+     * @param taxPipeline 税务管道
+     * @return {@link Message }<{@link ? }>
+     */
+    @RequestMapping("/update")
+    @RebuildIndex
+    public Message<?> update(@RequestBody @Validated(value = Valid.Update.class) TaxPipelineCO taxPipeline) {
+        taxPipelineService.update(taxPipeline);
+        return Message.success();
+    }
+
+    /**
      * 按租户列出
      *
      * @param group  群
@@ -66,6 +82,7 @@ public class TaxPipelineController {
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("/disable/{id}")
+    @RebuildIndex
     public Message<?> disable(@PathVariable("id") Long id) {
         taxPipelineService.updateState(id, StateType.DISABLE);
         return Message.success();
@@ -78,6 +95,7 @@ public class TaxPipelineController {
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("/enable/{id}")
+    @RebuildIndex
     public Message<?> enable(@PathVariable("id") Long id) {
         taxPipelineService.updateState(id, StateType.ENABLE);
         return Message.success();
 package com.diligrp.tax.storage.controller;
  
+import com.diligrp.tax.storage.Valid;
+import com.diligrp.tax.storage.anno.RebuildIndex;
 import com.diligrp.tax.storage.message.Message;
 import com.diligrp.tax.storage.model.co.TaxTenantCO;
 import com.diligrp.tax.storage.service.TaxTenantService;
@@ -42,19 +44,33 @@ public class TaxTenantController {
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("/save")
-    public Message<?> save(@RequestBody @Validated TaxTenantCO taxTenantCO) {
+    @RebuildIndex
+    public Message<?> save(@RequestBody @Validated(value = {Valid.Create.class}) TaxTenantCO taxTenantCO) {
         taxTenantService.saveTenant(taxTenantCO);
         return Message.success();
     }
  
     /**
+     * 保存
+     *
+     * @param taxTenantCO 税务租户做
+     * @return {@link Message }<{@link ? }>
+     */
+    @RequestMapping("/update")
+    @RebuildIndex
+    public Message<?> update(@RequestBody @Validated(value = {Valid.Update.class}) TaxTenantCO taxTenantCO) {
+        taxTenantService.updateTenant(taxTenantCO);
+        return Message.success();
+    }
+
+    /**
      * 分页查询
      *
      * @param taxTenantCO 税务租户做
      * @return {@link Message }<{@link ? }>
      */
     @RequestMapping("page")
-    public Message<?> page(@RequestBody TaxTenantCO taxTenantCO) {
+    public Message<?> page(@RequestBody @Validated(value = {Valid.Read.class}) TaxTenantCO taxTenantCO) {
         return Message.success(taxTenantService.findByNameAndState(taxTenantCO));
     }
 }
 package com.diligrp.tax.storage.domain;
  
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableField;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
 import lombok.Getter;
 import lombok.Setter;
  
 import java.time.LocalDateTime;
-import java.util.Map;
  
 @Getter
 @Setter
-@TableName(value = "tax_pipeline_mapping")
 public class TaxPipelineMapping {
  
-    @TableId(type = IdType.AUTO)
+    /**
+     * 租户 ID
+     */
+    private Long tenantId;
+
+    /**
+     * id
+     */
     private Long id;
  
-    @TableField(value = "pipeline_id")
+    /**
+     * 管道 ID
+     */
     private Long pipelineId;
  
-    @TableField(value = "document_type")
+    /**
+     * 文档类型
+     */
     private String documentType;
  
-    @TableField(value = "system_data_id")
+    /**
+     * 第三方系统数据 ID
+     */
     private String systemDataId;
  
-    @TableField(value = "pipeline_data_id")
+    /**
+     * 所在账套数据 ID
+     */
     private String pipelineDataId;
  
-    @TableField(value = "origin_data", typeHandler = JacksonTypeHandler.class)
-    private Map<String, Object> originData;
+    /**
+     * 原产地数据
+     */
+    private String originData;
  
-    @TableField(value = "state")
+    /**
+     * 州
+     */
     private Integer state;
  
-    @TableField(value = "created_time")
+    /**
+     * 创建时间
+     */
     private LocalDateTime createdTime;
  
-    @TableField(value = "modified_time")
+    /**
+     * 修改时间
+     */
     private LocalDateTime modifiedTime;
-
 }
 package com.diligrp.tax.storage.message;
  
+import com.diligrp.tax.storage.Valid;
+import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
 import lombok.Setter;
  
@@ -14,9 +16,11 @@ public abstract class PageQuery {
     /**
      * 页码
      */
+    @NotNull(groups = {Valid.Read.class})
     protected Integer pageNumber;
     /**
      * 页面大小
      */
+    @NotNull(groups = {Valid.Read.class})
     protected Integer pageSize;
 }
 package com.diligrp.tax.storage.model.co;
  
+import com.diligrp.tax.storage.Valid;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
 import lombok.Setter;
  
@@ -17,6 +19,7 @@ public class TaxPipelineCO {
     /**
      * id
      */
+    @NotNull(groups = Valid.Update.class)
     private Long id;
     /**
      * 租户 ID
@@ -25,32 +28,32 @@ public class TaxPipelineCO {
     /**
      * 群
      */
-    @NotEmpty
+    @NotEmpty(groups = Valid.Create.class)
     private String group;
     /**
      * 实体
      */
-    @NotEmpty
+    @NotEmpty(groups = Valid.Create.class)
     private String entity;
     /**
      * 账套类型
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class})
     private String systemCode;
     /**
      * 账套名字
      */
-    @NotEmpty
+    @NotEmpty(groups = Valid.Create.class)
     private String name;
     /**
      * 账套编码
      */
-    @NotEmpty
+    @NotEmpty(groups = Valid.Create.class)
     private String code;
     /**
      * 参数
      */
-    @NotEmpty
+    @NotEmpty(groups = Valid.Create.class)
     private Map<String, Object> params;
     /**
      * 状态
 package com.diligrp.tax.storage.model.co;
  
+import com.diligrp.tax.storage.Valid;
 import jakarta.validation.constraints.NotEmpty;
 import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
@@ -12,30 +13,31 @@ public class TaxPipelineConfigCO {
     /**
      * id
      */
+    @NotNull(groups = {Valid.Update.class, Valid.Delete.class})
     private Long id;
  
     /**
      * 管道 ID
      */
-    @NotNull
+    @NotNull(groups = {Valid.Create.class})
     private Long pipelineId;
  
     /**
      * 文档类型
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class})
     private String documentType;
  
     /**
      * 配置键
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class})
     private String configKey;
  
     /**
      * 配置值
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class})
     private String configValue;
  
 }
+package com.diligrp.tax.storage.model.co;
+
+import com.diligrp.tax.storage.Valid;
+import com.diligrp.tax.storage.message.PageQuery;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public class TaxPipelineMappingCO extends PageQuery {
+
+    /**
+     * 分组
+     */
+    @NotEmpty(groups = {Valid.Read.class})
+    private String group;
+
+    /**
+     * 实体
+     */
+    @NotEmpty(groups = {Valid.Read.class})
+    private String entity;
+
+    /**
+     * 租户 ID
+     */
+    private Long tenantId;
+    /**
+     * 管道 ID
+     */
+    @NotNull(groups = {Valid.Read.class})
+    private Long pipelineId;
+    /**
+     * 文档类型
+     */
+    private String documentType;
+    /**
+     * 第三方系统数据 ID
+     */
+    private String systemDataId;
+    /**
+     * 州
+     */
+    private Integer state;
+}
+package com.diligrp.tax.storage.model.co;
+
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-07 11:06
+ * @Version: todo
+ */
+@Getter
+@Setter
+@Valid
+public class TaxPipelineMappingCreateCO {
+    /**
+     * 租户 ID
+     */
+    @NotNull(message = "租户ID不能为空")
+    private Long tenantId;
+    /**
+     * 管道 ID
+     */
+    @NotNull(message = "管道ID不能为空")
+    private Long pipelineId;
+    /**
+     * 文档类型
+     */
+    @NotEmpty(message = "文档类型不能为空")
+    private String documentType;
+    /**
+     * 同步系统数据唯一键
+     */
+    @NotEmpty(message = "同步系统数据唯一键不能为空")
+    private String systemDataId;
+    /**
+     * 账套数据唯一键
+     */
+    @NotEmpty(message = "账套数据唯一键不能为空")
+    private String pipelineDataId;
+    /**
+     * 原始发送数据
+     */
+    @NotEmpty(message = "原始发送数据不能为空")
+    private String originData;
+    /**
+     * 同步状态
+     */
+    @NotNull(message = "同步状态不能为空")
+    private Integer state;
+}
 package com.diligrp.tax.storage.model.co;
  
+import com.diligrp.tax.storage.Valid;
 import com.diligrp.tax.storage.message.PageQuery;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
 import lombok.Setter;
  
@@ -16,21 +18,22 @@ public class TaxTenantCO extends PageQuery {
     /**
      * id
      */
+    @NotNull(groups = {Valid.Update.class, Valid.Delete.class, Valid.Read.class})
     private Long id;
     /**
      * 名字
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class,Valid.Update.class})
     private String name;
     /**
      * 群
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class,Valid.Update.class})
     private String group;
     /**
      * 实体
      */
-    @NotEmpty
+    @NotEmpty(groups = {Valid.Create.class,Valid.Update.class})
     private String entity;
     /**
      * 状态
 package com.diligrp.tax.storage.repo;
  
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.diligrp.tax.storage.domain.TaxPipelineMapping;
+import com.diligrp.tax.central.model.TenantTaxPipelineMapping;
+import org.apache.ibatis.annotations.Param;
 import org.springframework.stereotype.Repository;
  
+import java.util.List;
+import java.util.Optional;
+
 @Repository
-public interface TaxPipelineMappingRepository extends BaseMapper<TaxPipelineMapping> {
+public interface TaxPipelineMappingRepository {
+
+    /**
+     * 插入
+     *
+     * @param mapping 税务管道映射
+     */
+    void insert(TaxPipelineMapping mapping);
+
+    /**
+     * 更新
+     *
+     * @param mapping 税务管道映射
+     */
+    void update(TaxPipelineMapping mapping);
+
+    /**
+     * 按管道 ID 查找
+     *
+     * @param mapping 税务管道映射
+     * @return {@link List }<{@link TaxPipelineMapping }>
+     */
+    Page<TenantTaxPipelineMapping> findByPipelineIdAndState(TaxPipelineMapping mapping, IPage<?> page);
+
+    /**
+     * 按管道 ID 和系统数据 ID 查找
+     *
+     * @param mapping 映射
+     * @return {@link Optional }<{@link TaxPipelineMapping }>
+     */
+    Optional<TenantTaxPipelineMapping> findByPipelineIdAndDocumentTypeAndSystemDataId(TaxPipelineMapping mapping);
+
+    /**
+     * 创建租户映射表
+     *
+     * @param tableName 表名
+     */
+    void createTenantMappingTable(@Param("tableName") String tableName);
+
+    /**
+     * 删除租户映射表
+     *
+     * @param tableName 表名
+     */
+    void dropTenantMappingTable(@Param("tableName") String tableName);
 }
+package com.diligrp.tax.storage.service;
+
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.diligrp.tax.central.exception.TaxAgentServiceException;
+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.storage.domain.TaxPipelineMapping;
+import com.diligrp.tax.storage.model.co.TaxPipelineMappingCO;
+import com.diligrp.tax.storage.model.co.TaxPipelineMappingCreateCO;
+import com.diligrp.tax.storage.repo.TaxPipelineMappingRepository;
+import com.diligrp.tax.storage.type.MappingStateType;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.validation.annotation.Validated;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-07 10:41
+ * @Version: todo
+ */
+@Component
+public class DynamicTaxPipelineMappingService implements ITaxPipelineMappingService {
+
+    @Resource
+    private TaxPipelineMappingRepository taxPipelineMappingRepository;
+
+    private static final String DEFAULT_TAX_PIPELINE_MAPPING_TABLE_NAME = "tax_pipeline_mapping_";
+
+    /**
+     * 创建租户映射表
+     *
+     * @param tenantId 租户 ID
+     */
+    @Transactional
+    public void createTenantMappingTable(String tenantId) {
+        String tenantIdValid = Optional.ofNullable(tenantId).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "租户ID不能为空"));
+        String tableName = DEFAULT_TAX_PIPELINE_MAPPING_TABLE_NAME + tenantIdValid;
+        taxPipelineMappingRepository.createTenantMappingTable(tableName);
+    }
+
+    /**
+     * 查询租户管道数据
+     *
+     * @param taxPipelineMappingCO
+     * @return {@link List }<{@link TenantTaxPipelineMapping }>
+     */
+    public Page<TenantTaxPipelineMapping> findTenantPipelineData(TaxPipelineMappingCO taxPipelineMappingCO) {
+        Optional.ofNullable(taxPipelineMappingCO.getTenantId()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "租户ID不能为空"));
+        Optional.ofNullable(taxPipelineMappingCO.getPipelineId()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "管道ID不能为空"));
+        Page<TenantTaxPipelineMapping> page = new Page<>(taxPipelineMappingCO.getPageNumber(), taxPipelineMappingCO.getPageSize());
+        TaxPipelineMapping taxPipelineMapping = JsonUtils.convertValue(taxPipelineMappingCO, TaxPipelineMapping.class);
+        return taxPipelineMappingRepository.findByPipelineIdAndState(taxPipelineMapping, page);
+    }
+
+    /**
+     * 按管道 ID 和文档类型以及系统数据 ID 查找
+     *
+     * @param pipelineId   管道 ID
+     * @param documentType 文档类型
+     * @param systemDataId 系统数据 ID
+     * @return {@link Optional }<{@link TenantTaxPipelineMapping }>
+     */
+    @Override
+    public Optional<TenantTaxPipelineMapping> findByPipelineIdAndDocumentTypeAndSystemDataId(Long tenantId, Long pipelineId, String documentType, String systemDataId) {
+        Optional.ofNullable(tenantId).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "租户ID不能为空"));
+        Optional.ofNullable(pipelineId).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "管道ID不能为空"));
+        Optional.ofNullable(documentType).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "文档类型不能为空"));
+        Optional.ofNullable(systemDataId).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "系统数据不能为空"));
+        TaxPipelineMapping taxPipelineMapping = new TaxPipelineMapping();
+        taxPipelineMapping.setTenantId(tenantId);
+        taxPipelineMapping.setPipelineId(pipelineId);
+        taxPipelineMapping.setDocumentType(documentType);
+        taxPipelineMapping.setSystemDataId(systemDataId);
+        taxPipelineMapping.setState(MappingStateType.SYNCED.value);
+        return taxPipelineMappingRepository.findByPipelineIdAndDocumentTypeAndSystemDataId(taxPipelineMapping);
+    }
+
+    /**
+     * 插入
+     *
+     * @param taxPipelineMappingCO 税务管道测绘公司
+     */
+    public void insert(@Validated TaxPipelineMappingCreateCO taxPipelineMappingCO) {
+        DocumentType.validateDocumentType(taxPipelineMappingCO.getDocumentType());
+        TaxPipelineMapping taxPipelineMapping = JsonUtils.convertValue(taxPipelineMappingCO, TaxPipelineMapping.class);
+        taxPipelineMapping.setState(MappingStateType.SYNCED.value);
+        taxPipelineMappingRepository.insert(taxPipelineMapping);
+    }
+}
@@ -63,7 +63,21 @@ public class TaxPipelineConfigService {
     @Transactional
     public void save(TaxPipelineConfigCO taxPipelineConfig) {
         DocumentType.validateDocumentType(taxPipelineConfig.getDocumentType());
-        taxPipelineConfigRepository.insertOrUpdate(JsonUtils.convertValue(taxPipelineConfig, TaxPipelineConfig.class));
+        TaxPipelineConfig config = JsonUtils.convertValue(taxPipelineConfig, TaxPipelineConfig.class);
+        taxPipelineConfigRepository.insert(config);
+    }
+
+
+    /**
+     * 更新
+     *
+     * @param taxPipelineConfig 税务管道配置
+     */
+    @Transactional
+    public void update(TaxPipelineConfigCO taxPipelineConfig) {
+        DocumentType.validateDocumentType(taxPipelineConfig.getDocumentType());
+        TaxPipelineConfig config = JsonUtils.convertValue(taxPipelineConfig, TaxPipelineConfig.class);
+        taxPipelineConfigRepository.updateById(config);
     }
  
     /**
+package com.diligrp.tax.storage.service;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-07 10:14
+ * @Version: todo
+ */
+public class TaxPipelineMappingService {
+}
@@ -44,15 +44,16 @@ public class TaxPipelineService {
         SystemType.validateSystemCode(taxPipelineCO.getSystemCode());
         TaxPipeline taxPipeline = JsonUtils.convertValue(taxPipelineCO, TaxPipeline.class);
         taxPipeline.setState(StateType.ENABLE.value);
-        taxPipelineRepository.insertOrUpdate(taxPipeline);
+        taxPipelineRepository.insert(taxPipeline);
     }
  
  
-    public List<TaxPipelineVO> listAllEnablePipeline(){
-        LambdaQueryWrapper<TaxPipeline> queryWrapper = new LambdaQueryWrapper<>();
-        queryWrapper.eq(TaxPipeline::getState, StateType.ENABLE.value);
-        List<TaxPipeline> taxPipelines = taxPipelineRepository.selectList(queryWrapper);
-        return taxPipelines.stream().map(taxPipeline -> JsonUtils.convertValue(taxPipeline, TaxPipelineVO.class)).toList();
+
+    @Transactional
+    public void update(TaxPipelineCO taxPipelineCO) {
+        SystemType.validateSystemCode(taxPipelineCO.getSystemCode());
+        TaxPipeline taxPipeline = JsonUtils.convertValue(taxPipelineCO, TaxPipeline.class);
+        taxPipelineRepository.updateById(taxPipeline);
     }
  
     /**
@@ -27,6 +27,8 @@ public class TaxTenantService {
  
     @Resource
     private TaxTenantRepository taxTenantRepository;
+    @Resource
+    private DynamicTaxPipelineMappingService dynamicTaxPipelineMappingService;
  
     /**
      * 获取租户
@@ -52,7 +54,19 @@ public class TaxTenantService {
     public void saveTenant(TaxTenantCO taxTenantCO) {
         TaxTenant taxTenant = JsonUtils.convertValue(taxTenantCO, TaxTenant.class);
         taxTenant.setState(StateType.ENABLE.value);
-        taxTenantRepository.insertOrUpdate(taxTenant);
+        taxTenantRepository.insert(taxTenant);
+        dynamicTaxPipelineMappingService.createTenantMappingTable(String.valueOf(taxTenant.getId()));
+    }
+
+    /**
+     * 更新租户
+     *
+     * @param taxTenantCO 税务租户公司
+     */
+    @Transactional
+    public void updateTenant(TaxTenantCO taxTenantCO) {
+        TaxTenant taxTenant = JsonUtils.convertValue(taxTenantCO, TaxTenant.class);
+        taxTenantRepository.updateById(taxTenant);
     }
  
     /**
+package com.diligrp.tax.storage.type;
+
+public enum MappingStateType {
+    SYNCED(0, "已同步"),
+    SYNC_FAILED(1, "同步失败"),
+    SYNC_RETRY(2, "同步重试");
+    public final int value;
+    public final String desc;
+
+    MappingStateType(int value, String desc) {
+        this.value = value;
+        this.desc = desc;
+    }
+}
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.diligrp.tax.storage.repo.TaxPipelineMappingRepository">
+    <insert id="insert">
+        INSERT INTO
+            tax_pipeline_mapping_${tenantId}( pipeline_id
+                                            , document_type
+                                            , system_data_id
+                                            , pipeline_data_id
+                                            , origin_data
+                                            , state)
+        VALUES
+            ( #{pipelineId}
+            , #{documentType}
+            , #{systemDataId}
+            , #{pipelineDataId}
+            , #{originData}
+            , #{state})
+    </insert>
+    <update id="update">
+        UPDATE
+            tax_pipeline_mapping_${tenantId}
+        SET
+            state = #{state}
+        WHERE
+            id = #{id}
+    </update>
+    <select id="findByPipelineIdAndState" resultType="com.diligrp.tax.central.model.TenantTaxPipelineMapping">
+        SELECT
+        id
+        , pipeline_id
+        , document_type
+        , system_data_id
+        , pipeline_data_id
+        , origin_data
+        , state
+        , created_Time
+        , modified_Time
+        FROM
+        tax_pipeline_mapping_${tenantId}
+        WHERE
+        1=1
+        AND pipeline_id = #{pipelineId}
+        <if test="state != null">
+            AND state = #{state}
+        </if>
+    </select>
+    <select id="findByPipelineIdAndDocumentTypeAndSystemDataId"
+            resultType="com.diligrp.tax.central.model.TenantTaxPipelineMapping">
+        SELECT
+            id
+          , pipeline_id
+          , document_type
+          , system_data_id
+          , pipeline_data_id
+          , origin_data
+          , state
+          , created_Time
+          , modified_Time
+        FROM
+            tax_pipeline_mapping_${tenantId}
+        WHERE
+              1 = 1
+          AND pipeline_id = #{pipelineId}
+          AND document_type = #{documentType}
+          AND system_data_id = #{systemDataId}
+    </select>
+
+    <!-- 建表语句 -->
+    <update id="createTenantMappingTable">
+        CREATE TABLE ${tableName}
+        (
+            `id`               bigint unsigned                                              NOT NULL AUTO_INCREMENT,
+            `pipeline_id`      bigint unsigned                                              NOT NULL COMMENT '账套ID',
+            `document_type`    varchar(40)                                                  NOT NULL COMMENT '业务单据类型',
+            `system_data_id`   varchar(50)                                                  NOT NULL COMMENT '系统数据ID',
+            `pipeline_data_id` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '账套数据ID',
+            `origin_data`      json                                                         NOT NULL COMMENT '原始数据',
+            `state`            tinyint                                                      NOT NULL COMMENT '状态 1=同步成功，2=同步失败',
+            `created_time`     datetime                                                     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+            `modified_time`    datetime                                                     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+            PRIMARY KEY (`id`) USING BTREE,
+            UNIQUE KEY `uk_pipeline_id_document_type_system_data_id` (`pipeline_id`, `document_type`, `system_data_id`) USING BTREE,
+            KEY `idx_pipeline_id_state` (`pipeline_id`, `state`) USING BTREE
+        ) ENGINE = InnoDB
+          DEFAULT CHARSET = utf8mb4
+          COLLATE = utf8mb4_0900_ai_ci COMMENT ='账套同步数据';
+    </update>
+
+    <!-- 删除表语句 -->
+    <update id="dropTenantMappingTable">
+        DROP TABLE IF EXISTS ${tableName}
+    </update>
+
+</mapper>