feat(kingdee): 实现金蝶连接管理器及上下文支持- 新增 AbstractConnectionManager 抽象类，用于统一连接管理器接口

- 新增 KingDeeConnectionManager 实现类，负责创建金蝶 API 连接- 新增 ConnectionContext 类，用于注册和管理各类连接管理器- 修改 Sender 接口及其子类，将连接对象作为参数传入 send 方法- 更新 ProxyProcessor，通过上下文获取连接并传递给 Sender - 移除废弃的 ConnectionIdentity 接口及相关实现 - 重构 TenantPipeline 类，优化配置映射结构 - 调整 MessageContext 中 tenantId 类型为 Long，并迁移 Context 至 domain 包 - 更新 ProcessorChain 和相关处理器方法签名以适配 MessageContext- 在 TenantStorageContext 中增加事件发布逻辑以支持租户数据加载- 添加 NO_MATCHING_SET_OF_ACCOUNTS_FOUND 错误码- 修改日志配置中 springProperty 的 scope 属性值为 messageContext

feat(kingdee): 实现金蝶连接管理器及上下文支持- 新增 AbstractConnectionManager 抽象类，用于统一连接管理器接口
- 新增 KingDeeConnectionManager 实现类，负责创建金蝶 API 连接- 新增 ConnectionContext 类，用于注册和管理各类连接管理器- 修改 Sender 接口及其子类，将连接对象作为参数传入 send 方法- 更新 ProxyProcessor，通过上下文获取连接并传递给 Sender - 移除废弃的 ConnectionIdentity 接口及相关实现 - 重构 TenantPipeline 类，优化配置映射结构 - 调整 MessageContext 中 tenantId 类型为 Long，并迁移 Context 至 domain 包 - 更新 ProcessorChain 和相关处理器方法签名以适配 MessageContext- 在 TenantStorageContext 中增加事件发布逻辑以支持租户数据加载- 添加 NO_MATCHING_SET_OF_ACCOUNTS_FOUND 错误码- 修改日志配置中 springProperty 的 scope 属性值为 messageContext
zhangmeiyang
1 parent c29bc71d
Showing 22 changed files with 167 additions and 127 deletions
tax-boot/src/main/java/com/diligrp/tax/boot/receiver/TaxReceiver.java
tax-boot/src/main/resources/logback-spring.xml
tax-central/src/main/java/com/diligrp/tax/central/context/ConnectionContext.java
tax-central/src/main/java/com/diligrp/tax/central/context/TenantStorageContext.java
tax-central/src/main/java/com/diligrp/tax/central/domain/BaseMapping.java
tax-central/src/main/java/com/diligrp/tax/central/context/Context.java → tax-central/src/main/java/com/diligrp/tax/central/domain/MessageContext.java
tax-central/src/main/java/com/diligrp/tax/central/interfaces/ConnectionIdentity.java
tax-central/src/main/java/com/diligrp/tax/central/manager/AbstractConnectionManager.java
tax-central/src/main/java/com/diligrp/tax/central/manager/impl/KingDeeConnectionManager.java
tax-central/src/main/java/com/diligrp/tax/central/model/TenantPipeline.java
tax-central/src/main/java/com/diligrp/tax/central/process/Processor.java
tax-central/src/main/java/com/diligrp/tax/central/process/ProcessorChain.java
tax-central/src/main/java/com/diligrp/tax/central/type/TaxSystemType.java
tax-doc/src/main/java/com/diligrp/tax/doc/process/kingdee/InitializeProcessor.java
tax-map/src/main/java/com/diligrp/tax/mapping/process/kingdee/MappingProcessor.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/demarcate/Sender.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/demarcate/kingdee/CustomerSender.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/demarcate/kingdee/ReceiptSender.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/demarcate/kingdee/ReceivableSender.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/demarcate/kingdee/RefundSender.java
 package com.diligrp.tax.boot.receiver;
-import com.diligrp.tax.central.context.Context;
+import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.process.ProcessorChain;
 import com.diligrp.tax.central.type.StatusType;
 import com.diligrp.tax.central.type.SystemType;
@@ -20,7 +20,6 @@ import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
-import java.util.Optional;
 import static com.diligrp.tax.boot.queue.TaxAutoPush.*;
@@ -47,7 +46,7 @@ public class TaxReceiver {
     public void receiveMessage(Channel channel, Message message) throws IOException {
         var content = new String(message.getBody(), StandardCharsets.UTF_8);
         log.info("tax-agent收到消息:{}", content);
-        Context ctx = JsonUtils.fromJsonString(content, Context.class);
+        MessageContext ctx = JsonUtils.fromJsonString(content, MessageContext.class);
         try {
             handle(ctx);
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
@@ -59,13 +58,13 @@ public class TaxReceiver {
         }
     }
-    private void handle(Context ctx) {
-        Context context = processorChainMap.get(SystemType.from(ctx.getSystemType())).startProcess(ctx);
-        context.setStatus(StatusType.SUCCESS.code);
+    private void handle(MessageContext ctx) {
+        MessageContext messageContext = processorChainMap.get(SystemType.from(ctx.getSystemType())).startProcess(ctx);
+        messageContext.setStatus(StatusType.SUCCESS.code);
         //TODO write to db
     }
-    private void recordError(Exception e, Context ctx) {
+    private void recordError(Exception e, MessageContext ctx) {
         StringWriter sw = new StringWriter();
         PrintWriter pw = new PrintWriter(sw);
         e.printStackTrace(pw);
@@ -5,7 +5,7 @@
     <!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径 -->
     <property name="LOG_HOME" value="logs"/>
     <!-- springProperty读取springboot配置属性 -->
-    <springProperty scope="context" name="build.profile.id" source="spring.profiles.active"/>
+    <springProperty scope="messageContext" name="build.profile.id" source="spring.profiles.active"/>
     <statusListener class="ch.qos.logback.core.status.NopStatusListener"/>
     <!-- 日志控制台输出 -->
     <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+package com.diligrp.tax.central.context;
+
+import com.diligrp.tax.central.manager.AbstractConnectionManager;
+import com.diligrp.tax.central.type.SystemType;
+import jakarta.annotation.Resource;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-06 15:02
+ * @Version: todo
+ */
+@Component
+public class ConnectionContext implements InitializingBean, DisposableBean {
+
+    public static final Map<SystemType, AbstractConnectionManager<?>> CONNECTION_MAP = new ConcurrentHashMap<>();
+
+    @Resource
+    private List<AbstractConnectionManager<?>> connectionManagers;
+
+    @Override
+    public void destroy() throws Exception {
+
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        connectionManagers.forEach(c -> CONNECTION_MAP.put(c.markSystemType(), c));
+    }
+}
 package com.diligrp.tax.central.context;
+import com.diligrp.tax.central.event.RestoreTenantEvent;
 import com.diligrp.tax.central.model.TenantPipeline;
 import com.diligrp.tax.central.model.TenantPipelineConfig;
 import com.diligrp.tax.central.service.ITenantTaxService;
+import com.diligrp.tax.central.type.DocumentType;
 import com.diligrp.tax.central.type.SystemType;
 import jakarta.annotation.Resource;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Component;
 import java.util.List;
@@ -20,10 +25,12 @@ import java.util.stream.Collectors;
  * @Version: todo
  */
 @Component
-public class TenantStorageContext {
+public class TenantStorageContext implements InitializingBean, DisposableBean {
     @Resource
     private ITenantTaxService tenantTaxService;
+    @Resource
+    private ApplicationEventPublisher publisher;
     private static final ConcurrentHashMap<Long, Map<SystemType, Map<String, TenantPipeline>>> TENANT_PIPELINE_MAP = new ConcurrentHashMap<>();
@@ -32,14 +39,11 @@ public class TenantStorageContext {
         Map<SystemType, Map<String, TenantPipeline>> tenantPipelineMap = TENANT_PIPELINE_MAP.get(tenantId);
         Optional.ofNullable(tenantPipelineMap).ifPresent(map -> {
             Map<String, TenantPipeline> pipelineMap = map.get(systemType);
-            Optional.ofNullable(pipelineMap).ifPresent(e -> {
-                pipeline.set(pipelineMap.get(pipelineCode));
-            });
+            Optional.ofNullable(pipelineMap).ifPresent(e -> pipeline.set(pipelineMap.get(pipelineCode)));
         });
         return Optional.of(pipeline.get());
     }
-
     public void loadTenantPipeline() {
         List<TenantPipeline> taxPipelineVOS = tenantTaxService.listAllEnablePipeline();
         Optional.ofNullable(taxPipelineVOS).ifPresent(pipelines -> {
@@ -60,10 +64,19 @@ public class TenantStorageContext {
         });
     }
-
     private TenantPipeline loadConfig(TenantPipeline tenantPipeline) {
         List<TenantPipelineConfig> list = tenantTaxService.listByPipelineId(tenantPipeline.getId());
-        tenantPipeline.setTenantPipelineConfigs(list);
+        tenantPipeline.setTenantPipelineConfigs(list.stream().collect(Collectors.toMap(k -> DocumentType.from(k.getDocumentType()), v -> v)));
         return tenantPipeline;
     }
+
+    @Override
+    public void destroy() throws Exception {
+
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        publisher.publishEvent(new RestoreTenantEvent(this));
+    }
 }
 package com.diligrp.tax.central.domain;
+import com.diligrp.tax.central.model.TenantPipeline;
 import lombok.Getter;
 import lombok.Setter;
-import java.util.Set;
-
 /**
  * @Author: zhangmeiyang
  * @CreateTime: 2025-10-30 17:25
@@ -15,5 +14,4 @@ import java.util.Set;
 public abstract class BaseMapping {
     private String verifyInformation;
     private String returnKeys;
-    private Set<String> accountSet;
 }
-package com.diligrp.tax.central.context;
+package com.diligrp.tax.central.domain;
-import com.diligrp.tax.central.domain.BaseDocument;
-import com.diligrp.tax.central.domain.BaseMapping;
-import com.diligrp.tax.central.domain.BaseProxy;
 import lombok.Getter;
 import lombok.Setter;
 @Setter
 @Getter
-public class Context {
+public class MessageContext {
     /**
      * 开放 ID
      */
@@ -16,7 +13,7 @@ public class Context {
     /**
      * 租户 ID
      */
-    private String tenantId;
+    private Long tenantId;
     /**
      * 消息信息
      */
-package com.diligrp.tax.central.interfaces;
-
-public interface ConnectionIdentity<T> {
-
-    T getIdentity();
-}
+package com.diligrp.tax.central.manager;
+
+import com.diligrp.tax.central.model.TenantPipeline;
+import com.diligrp.tax.central.type.SystemType;
+
+public abstract class AbstractConnectionManager<T> {
+
+    public abstract SystemType markSystemType();
+
+    public abstract T getConnection(TenantPipeline tenantPipeline);
+}
+package com.diligrp.tax.central.manager.impl;
+
+import com.diligrp.tax.central.manager.AbstractConnectionManager;
+import com.diligrp.tax.central.model.TenantPipeline;
+import com.diligrp.tax.central.type.SystemType;
+import com.diligrp.tax.central.utils.JsonUtils;
+import com.kingdee.bos.webapi.entity.IdentifyInfo;
+import com.kingdee.bos.webapi.sdk.K3CloudApi;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-06 14:59
+ * @Version: todo
+ */
+@Component
+public class KingDeeConnectionManager extends AbstractConnectionManager<K3CloudApi> {
+    @Override
+    public SystemType markSystemType() {
+        return SystemType.KING_DEE;
+    }
+
+    @Override
+    public K3CloudApi getConnection(TenantPipeline tenantPipeline) {
+        IdentifyInfo identifyInfo = JsonUtils.convertValue(tenantPipeline.getParams(), IdentifyInfo.class);
+        return new K3CloudApi(identifyInfo);
+    }
+}
 package com.diligrp.tax.central.model;
-import com.diligrp.tax.central.interfaces.ConnectionIdentity;
+import com.diligrp.tax.central.type.DocumentType;
 import com.diligrp.tax.central.type.SystemType;
-import com.diligrp.tax.central.utils.JsonUtils;
-import com.kingdee.bos.webapi.entity.IdentifyInfo;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.Setter;
-import java.util.List;
 import java.util.Map;
 /**
@@ -19,17 +16,12 @@ import java.util.Map;
 @Setter
 @Getter
 @Builder
-public class TenantPipeline implements ConnectionIdentity<IdentifyInfo> {
+public class TenantPipeline {
     private Long id;
     private Long tenantId;
     private SystemType systemType;
     private String name;
     private String code;
     private Map<String, Object> params;
-    private List<TenantPipelineConfig> tenantPipelineConfigs;
-
-    @Override
-    public IdentifyInfo getIdentity() {
-        return JsonUtils.convertValue(params, IdentifyInfo.class);
-    }
+    private Map<DocumentType, TenantPipelineConfig> tenantPipelineConfigs;
 }
 package com.diligrp.tax.central.process;
-import com.diligrp.tax.central.context.Context;
+import com.diligrp.tax.central.domain.MessageContext;
 /**
@@ -14,10 +14,10 @@ public interface Processor {
     /**
      * 处理请求
      *
-     * @param context 输入上下文
+     * @param messageContext 输入上下文
      * @return 输出上下文
      */
-    Context process(Context context);
+    MessageContext process(MessageContext messageContext);
     /**
 package com.diligrp.tax.central.process;
-import com.diligrp.tax.central.context.Context;
+import com.diligrp.tax.central.domain.MessageContext;
 import java.util.Comparator;
 import java.util.List;
@@ -22,7 +22,7 @@ public class ProcessorChain {
     /**
      * 类型安全的流式处理
      */
-    public Context startProcess(Context context) {
-        return processors.stream().reduce(context, (ctx, processor) -> processor.process(ctx), (ctx1, ctx2) -> ctx2);
+    public MessageContext startProcess(MessageContext messageContext) {
+        return processors.stream().reduce(messageContext, (ctx, processor) -> processor.process(ctx), (ctx1, ctx2) -> ctx2);
     }
 }
@@ -16,6 +16,7 @@ public enum TaxSystemType {
     BUSINESS_MATCHES_ARE_INCORRECT(5005, "业务匹配不正确"),
     REMOTE_SERVICE_CALLS_ARE_EXCEPTIONAL(5006, "远程服务调用异常"),
     INVALID_HTTP_REQUEST_PARAMS(5007, "无效的http请求参数"),
+    NO_MATCHING_SET_OF_ACCOUNTS_FOUND(5008, "未找到匹配账套"),
     ;
     public final int code;
     public final String message;
 package com.diligrp.tax.doc.process.kingdee;
-import com.diligrp.tax.central.context.Context;
+import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.process.AbstractProcessor;
 import com.diligrp.tax.central.type.DocumentType;
@@ -25,14 +25,14 @@ import org.springframework.stereotype.Component;
 public class InitializeProcessor extends AbstractProcessor {
     @Override
-    public Context process(Context context) {
-        DocumentType from = DocumentType.from(context.getDocumentType());
-        SystemType system = SystemType.from(context.getSystemType());
+    public MessageContext process(MessageContext messageContext) {
+        DocumentType from = DocumentType.from(messageContext.getDocumentType());
+        SystemType system = SystemType.from(messageContext.getSystemType());
         if (!system.documentTypes.contains(from)) {
             throw new TaxAgentServiceException(TaxSystemType.BUSINESS_MATCHES_ARE_INCORRECT);
         }
         Builder<?> builder = DocumentContext.CONTEXT.get(from);
-        context.setDocumentObject(builder.build(context.getMsgBody()));
-        return context;
+        messageContext.setDocumentObject(builder.build(messageContext.getMsgBody()));
+        return messageContext;
     }
 }
 package com.diligrp.tax.mapping.process.kingdee;
-import com.diligrp.tax.central.context.Context;
+import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.domain.BaseMapping;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.process.AbstractProcessor;
@@ -27,17 +27,17 @@ import java.util.Map;
 public class MappingProcessor extends AbstractProcessor {
     @Override
-    public Context process(Context context) {
-        DocumentType from = DocumentType.from(context.getDocumentType());
-        SystemType system = SystemType.from(context.getSystemType());
+    public MessageContext process(MessageContext messageContext) {
+        DocumentType from = DocumentType.from(messageContext.getDocumentType());
+        SystemType system = SystemType.from(messageContext.getSystemType());
         if (!system.documentTypes.contains(from)) {
             throw new TaxAgentServiceException(TaxSystemType.BUSINESS_MATCHES_ARE_INCORRECT);
         }
         Transformer<?> transformer = MappingContext.CONTEXT.get(from);
-        BaseMapping transform = transformer.transform(context.getDocumentObject());
-        context.setMappingObject(transform);
+        BaseMapping transform = transformer.transform(messageContext.getDocumentObject());
+        messageContext.setMappingObject(transform);
         Map<String,Object> map = JsonUtils.convertValue(transform, new TypeReference<>() {});
         log.info("转换结果,{}", JsonUtils.toJsonString(map));
-        return context;
+        return messageContext;
     }
 }
@@ -3,6 +3,7 @@ package com.diligrp.tax.proxy.demarcate;
 import com.diligrp.tax.central.domain.BaseMapping;
 import com.diligrp.tax.central.domain.BaseProxy;
 import com.diligrp.tax.central.type.DocumentType;
+import com.kingdee.bos.webapi.sdk.K3CloudApi;
 /**
  * @Author: zhangmeiyang
@@ -21,7 +22,8 @@ public abstract class Sender&lt;T extends BaseProxy&gt; {
      * 变换
      *
      * @param mappingObject 映射对象
+     * @param connection
      * @return {@link T }
      */
-    public abstract T send(BaseMapping mappingObject);
+    public abstract T send(BaseMapping mappingObject, K3CloudApi connection);
 }
@@ -35,30 +35,22 @@ public class CustomerSender extends Sender&lt;CustomerProxy&gt; {
     }
     @Override
-    public CustomerProxy send(BaseMapping mappingObject) {
-        var identifyInfo = new IdentifyInfo();
-        identifyInfo.setAppId("293965_6+dDWztPVvgY1X1vX67BzbyEyLSc2tqL");
-        identifyInfo.setUserName("市场");
-        identifyInfo.setServerUrl("http://120.46.151.190/k3cloud/");
-        identifyInfo.setAppSecret("a384eaabf599461aa05d89ce4c366d27");
-        identifyInfo.setdCID("66435fe720f1d3");
-        K3CloudApi kingDeeApi = KingDeeHelper.getKingDeeApi(identifyInfo);
-        //TODO 账套获取
+    public CustomerProxy send(BaseMapping mappingObject, K3CloudApi connection) {
         CustomerMapping mapping = (CustomerMapping) mappingObject;
         List<FT_BD_CUSTLOCATION> contacts = mapping.getFT_BD_CUSTLOCATION();
         contacts.forEach(contact -> {
             Map<String, Object> model = JsonUtils.convertValue(contact, new TypeReference<>() {
             });
-            RepoRet<?> repoRet = KingDeeHelper.nonAuditSend(model, kingDeeApi, "BD_CommonContact");
+            RepoRet<?> repoRet = KingDeeHelper.nonAuditSend(model, connection, "BD_CommonContact");
             SuccessEntity first = repoRet.getResult().getResponseStatus().getSuccessEntitys().getFirst();
             contact.getFContactId().setFNUMBER(first.getNumber());
         });
         // 联系人新增
         Optional.ofNullable(mapping.getFCUSTID())
                 .filter(e -> !e.isEmpty())
-                .ifPresent(e -> KingDeeHelper.unAuditSend(e, kingDeeApi, markDocument().value));
+                .ifPresent(e -> KingDeeHelper.unAuditSend(e, connection, markDocument().value));
         RepoRet<?> repoRet = KingDeeHelper.auditSend(JsonUtils.convertValue(mapping, new TypeReference<>() {
-        }), kingDeeApi, markDocument().value);
+        }), connection, markDocument().value);
         //联系人新增和更新完成
         ArrayList<SuccessEntity> successEntity = repoRet.getResult().getResponseStatus().getSuccessEntitys();
         if (successEntity.isEmpty()) {
@@ -2,9 +2,7 @@ package com.diligrp.tax.proxy.demarcate.kingdee;
 import com.diligrp.tax.central.domain.BaseMapping;
 import com.diligrp.tax.central.domain.mapping.kingdee.ReceiptMapping;
-import com.diligrp.tax.central.domain.mapping.kingdee.ReceivableMapping;
 import com.diligrp.tax.central.domain.proxy.kingdee.ReceiptProxy;
-import com.diligrp.tax.central.domain.proxy.kingdee.ReceivableProxy;
 import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.type.DocumentType;
 import com.diligrp.tax.central.type.TaxSystemType;
@@ -36,24 +34,13 @@ public class ReceiptSender extends Sender&lt;ReceiptProxy&gt; {
     }
     @Override
-    public ReceiptProxy send(BaseMapping mappingObject) {
+    public ReceiptProxy send(BaseMapping mappingObject, K3CloudApi connection) {
         ReceiptMapping mapping = (ReceiptMapping) mappingObject;
-//        Set<String> accountSet = mapping.getAccountSet();
-        //TODO 获取账套信息并找到发送的IdentityInfo
-        var identifyInfo = new IdentifyInfo();
-        identifyInfo.setAppId("293965_6+dDWztPVvgY1X1vX67BzbyEyLSc2tqL");
-        identifyInfo.setUserName("市场");
-        identifyInfo.setServerUrl("http://120.46.151.190/k3cloud/");
-        identifyInfo.setAppSecret("a384eaabf599461aa05d89ce4c366d27");
-        identifyInfo.setdCID("66435fe720f1d3");
-        K3CloudApi api = KingDeeHelper.getKingDeeApi(identifyInfo);
-        List<Object> infos = KingDeeHelper.querySend(mapping, markDocument().value, identifyInfo);
+        List<Object> infos = KingDeeHelper.querySend(mapping, markDocument().value, connection);
         if (!infos.isEmpty()) {
             throw new TaxAgentServiceException(TaxSystemType.REPEAT_SENDING);
         }
-        Map<String, Object> model = JsonUtils.convertValue(mapping, new TypeReference<>() {
-        });
-        RepoRet<?> repoRet = KingDeeHelper.auditSend(model, api, markDocument().value);
+        RepoRet<?> repoRet = KingDeeHelper.auditSend(JsonUtils.convertValue(mapping, new TypeReference<>() {}), connection, markDocument().value);
         ArrayList<SuccessEntity> successEntity = repoRet.getResult().getResponseStatus().getSuccessEntitys();
         if (successEntity.isEmpty()) {
             throw new TaxAgentServiceException(TaxSystemType.REMOTE_SERVICE_CALLS_ARE_EXCEPTIONAL);
@@ -32,24 +32,15 @@ public class ReceivableSender extends Sender&lt;ReceivableProxy&gt; {
     }
     @Override
-    public ReceivableProxy send(BaseMapping mappingObject) {
+    public ReceivableProxy send(BaseMapping mappingObject, K3CloudApi connection) {
         ReceivableMapping mapping = (ReceivableMapping) mappingObject;
-//        Set<String> accountSet = mapping.getAccountSet();
-        //TODO 获取账套信息并找到发送的IdentityInfo
-        var identifyInfo = new IdentifyInfo();
-        identifyInfo.setAppId("293965_6+dDWztPVvgY1X1vX67BzbyEyLSc2tqL");
-        identifyInfo.setUserName("市场");
-        identifyInfo.setServerUrl("http://120.46.151.190/k3cloud/");
-        identifyInfo.setAppSecret("a384eaabf599461aa05d89ce4c366d27");
-        identifyInfo.setdCID("66435fe720f1d3");
-        K3CloudApi api = KingDeeHelper.getKingDeeApi(identifyInfo);
-        List<Object> infos = KingDeeHelper.querySend(mapping, markDocument().value, identifyInfo);
+        List<Object> infos = KingDeeHelper.querySend(mapping, markDocument().value, connection);
         if (!infos.isEmpty()) {
             throw new TaxAgentServiceException(TaxSystemType.REPEAT_SENDING);
         }
         Map<String, Object> model = JsonUtils.convertValue(mapping, new TypeReference<>() {
         });
-        RepoRet<?> repoRet = KingDeeHelper.auditSend(model, api, markDocument().value);
+        RepoRet<?> repoRet = KingDeeHelper.auditSend(model, connection, markDocument().value);
         ArrayList<SuccessEntity> successEntity = repoRet.getResult().getResponseStatus().getSuccessEntitys();
         if (successEntity.isEmpty()) {
             throw new TaxAgentServiceException(TaxSystemType.REMOTE_SERVICE_CALLS_ARE_EXCEPTIONAL);
@@ -34,24 +34,13 @@ public class RefundSender extends Sender&lt;RefundProxy&gt; {
     }
     @Override
-    public RefundProxy send(BaseMapping mappingObject) {
+    public RefundProxy send(BaseMapping mappingObject, K3CloudApi connection) {
         RefundMapping mapping = (RefundMapping) mappingObject;
-//        Set<String> accountSet = mapping.getAccountSet();
-        //TODO 获取账套信息并找到发送的IdentityInfo
-        var identifyInfo = new IdentifyInfo();
-        identifyInfo.setAppId("293965_6+dDWztPVvgY1X1vX67BzbyEyLSc2tqL");
-        identifyInfo.setUserName("市场");
-        identifyInfo.setServerUrl("http://120.46.151.190/k3cloud/");
-        identifyInfo.setAppSecret("a384eaabf599461aa05d89ce4c366d27");
-        identifyInfo.setdCID("66435fe720f1d3");
-        K3CloudApi api = KingDeeHelper.getKingDeeApi(identifyInfo);
-        List<Object> infos = KingDeeHelper.querySend(mapping, markDocument().value, identifyInfo);
+        List<Object> infos = KingDeeHelper.querySend(mapping, markDocument().value, connection);
         if (!infos.isEmpty()) {
             throw new TaxAgentServiceException(TaxSystemType.REPEAT_SENDING);
         }
-        Map<String, Object> model = JsonUtils.convertValue(mapping, new TypeReference<>() {
-        });
-        RepoRet<?> repoRet = KingDeeHelper.auditSend(model, api, markDocument().value);
+        RepoRet<?> repoRet = KingDeeHelper.auditSend(JsonUtils.convertValue(mapping, new TypeReference<>() {}), connection, markDocument().value);
         ArrayList<SuccessEntity> successEntity = repoRet.getResult().getResponseStatus().getSuccessEntitys();
         if (successEntity.isEmpty()) {
             throw new TaxAgentServiceException(TaxSystemType.REMOTE_SERVICE_CALLS_ARE_EXCEPTIONAL);
@@ -124,10 +124,9 @@ public class KingDeeHelper {
         }
     }
-    public static <T extends BaseMapping> List<Object> querySend(T t, String formId, IdentifyInfo identifyInfo) {
+    public static <T extends BaseMapping> List<Object> querySend(T t, String formId, K3CloudApi kingDeeApi) {
         var filterString = t.getVerifyInformation();
         var fieldKeys = t.getReturnKeys();
-        K3CloudApi kingDeeApi = getKingDeeApi(identifyInfo);
         return querySend(formId, filterString, fieldKeys, kingDeeApi);
     }
 package com.diligrp.tax.proxy.process.kingdee;
-import com.diligrp.tax.central.context.Context;
+import com.diligrp.tax.central.context.ConnectionContext;
+import com.diligrp.tax.central.context.TenantStorageContext;
 import com.diligrp.tax.central.domain.BaseProxy;
+import com.diligrp.tax.central.domain.MessageContext;
+import com.diligrp.tax.central.exception.TaxAgentServiceException;
+import com.diligrp.tax.central.manager.AbstractConnectionManager;
+import com.diligrp.tax.central.model.TenantPipeline;
 import com.diligrp.tax.central.process.AbstractProcessor;
 import com.diligrp.tax.central.type.DocumentType;
-import com.diligrp.tax.central.utils.JsonUtils;
+import com.diligrp.tax.central.type.SystemType;
+import com.diligrp.tax.central.type.TaxSystemType;
 import com.diligrp.tax.proxy.context.ProxyContext;
+import com.kingdee.bos.webapi.sdk.K3CloudApi;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
+import java.util.Optional;
+
 /**
  * 代理发送处理器
  * 将Map<String, Object>发送到目标系统
@@ -20,13 +29,15 @@ import org.springframework.stereotype.Component;
 public class ProxyProcessor extends AbstractProcessor {
     @Override
-    public Context process(Context context) {
-        log.info("开始发送代理请求");
-        log.info("base-document,{}", JsonUtils.toJsonString(context.getDocumentObject()));
-        log.info("base-mapping,{}", JsonUtils.toJsonString(context.getMappingObject()));
-        DocumentType from = DocumentType.from(context.getDocumentType());
-        BaseProxy send = ProxyContext.CONTEXT.get(from).send(context.getMappingObject());
-        context.setProxyObject(send);
-        return context;
+    public MessageContext process(MessageContext messageContext) {
+        DocumentType from = DocumentType.from(messageContext.getDocumentType());
+        SystemType systemType = SystemType.from(messageContext.getSystemType());
+        Optional<TenantPipeline> option = TenantStorageContext.getTenantPipeline(messageContext.getTenantId(), systemType, from.value);
+        TenantPipeline pipeline = option.orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.NO_MATCHING_SET_OF_ACCOUNTS_FOUND));
+        AbstractConnectionManager<?> abstractConnectionManager = ConnectionContext.CONNECTION_MAP.get(systemType);
+        K3CloudApi connection = (K3CloudApi) abstractConnectionManager.getConnection(pipeline);
+        BaseProxy send = ProxyContext.CONTEXT.get(from).send(messageContext.getMappingObject(), connection);
+        messageContext.setProxyObject(send);
+        return messageContext;
     }
 }