feat: add send many account mode

2025-12-14 12:43:09 +08:00
parent 2125ddc853
commit 5e20533aa9
26 changed files with 1547 additions and 793 deletions
@@ -1,6 +1,7 @@
 package send_message_service

 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"message-nest/models"
@@ -48,6 +49,9 @@ type SendMessageService struct {
 	AtUserIds []string
 	AtAll     bool

+	// 动态接收者（用于邮箱、微信公众号等支持群发的渠道）
+	Recipients []string
+
 	Status    int
 	LogOutput []string

@@ -253,12 +257,35 @@ func (sm *SendMessageService) Send(task models.TaskIns) (string, error) {
 			}
 		}

-		// 使用 SendUnified 方法（自动格式转换和@功能支持）
-		res, errMsg := channel.SendUnified(msgObj, ins.SendTasksIns, unifiedContent)
-		if res != "" {
-			sm.LogsAndStatusMark(fmt.Sprintf("返回内容：%s\n", res), sm.Status)
+		// 处理动态接收者（邮箱、微信公众号等支持群发的渠道）
+		isDynamicMode := sm.isDynamicRecipientMode(ins.SendTasksIns)
+
+		if isDynamicMode && sm.supportsDynamicRecipient(way.Type) && len(sm.Recipients) > 0 {
+			// 动态接收模式：使用API传入的Recipients列表（群发）
+			sm.LogsAndStatusMark(fmt.Sprintf("动态接收模式，共 %d 个接收者", len(sm.Recipients)), sm.Status)
+			for recipientIdx, recipient := range sm.Recipients {
+				sm.LogsAndStatusMark(fmt.Sprintf(">>> 接收者 %d/%d: %s", recipientIdx+1, len(sm.Recipients), recipient), sm.Status)
+
+				// 临时修改实例配置中的接收者
+				modifiedIns := sm.modifyInsRecipient(ins.SendTasksIns, recipient, way.Type)
+
+				// 使用 SendUnified 方法发送
+				res, errMsg := channel.SendUnified(msgObj, modifiedIns, unifiedContent)
+				if res != "" {
+					sm.LogsAndStatusMark(fmt.Sprintf("返回内容：%s", res), sm.Status)
+				} else {
+					sm.LogsAndStatusMark(sm.TransError(errMsg), errStrIsSuccess(errMsg))
+				}
+			}
 		} else {
-			sm.LogsAndStatusMark(sm.TransError(errMsg), errStrIsSuccess(errMsg))
+			// 固定接收模式：使用实例配置的to_account
+			sm.LogsAndStatusMark("固定接收模式，使用实例配置的接收者", sm.Status)
+			res, errMsg := channel.SendUnified(msgObj, ins.SendTasksIns, unifiedContent)
+			if res != "" {
+				sm.LogsAndStatusMark(fmt.Sprintf("返回内容：%s\n", res), sm.Status)
+			} else {
+				sm.LogsAndStatusMark(sm.TransError(errMsg), errStrIsSuccess(errMsg))
+			}
 		}

 	}
@@ -366,6 +393,64 @@ func (sm *SendMessageService) BuildTemplateContent(ins models.SendTasksIns) *uni
 	return content
 }

+// supportsDynamicRecipient 判断渠道是否支持动态接收者
+func (sm *SendMessageService) supportsDynamicRecipient(wayType string) bool {
+	// 支持动态接收者的渠道类型
+	supportedTypes := map[string]bool{
+		unified.MessageTypeEmail:           true,
+		unified.MessageTypeWeChatOFAccount: true,
+		// 可以继续添加其他支持动态接收者的渠道
+	}
+	return supportedTypes[wayType]
+}
+
+// isDynamicRecipientMode 判断实例是否配置为动态接收模式
+func (sm *SendMessageService) isDynamicRecipientMode(ins models.SendTasksIns) bool {
+	// 解析实例配置
+	var config map[string]interface{}
+	if err := json.Unmarshal([]byte(ins.Config), &config); err != nil {
+		return false
+	}
+
+	// 检查allowMultiRecip字段
+	// allowMultiRecip=true: 动态模式
+	// allowMultiRecip=false或不存在: 固定模式
+	if allowMultiRecip, ok := config["allowMultiRecip"]; ok {
+		if allow, ok := allowMultiRecip.(bool); ok {
+			return allow
+		}
+	}
+
+	// 默认为固定模式（兼容历史数据）
+	return false
+}
+
+// modifyInsRecipient 临时修改实例配置中的接收者
+func (sm *SendMessageService) modifyInsRecipient(ins models.SendTasksIns, recipient string, wayType string) models.SendTasksIns {
+	// 创建副本
+	modifiedIns := ins
+
+	// 根据渠道类型修改配置
+	var config map[string]interface{}
+	if err := json.Unmarshal([]byte(ins.Config), &config); err != nil {
+		sm.LogsAndStatusMark(fmt.Sprintf("解析实例配置失败: %s", err.Error()), SendFail)
+		return ins
+	}
+
+	// 修改接收者字段
+	config["to_account"] = recipient
+
+	// 序列化回JSON
+	modifiedConfigBytes, err := json.Marshal(config)
+	if err != nil {
+		sm.LogsAndStatusMark(fmt.Sprintf("序列化实例配置失败: %s", err.Error()), SendFail)
+		return ins
+	}
+
+	modifiedIns.Config = string(modifiedConfigBytes)
+	return modifiedIns
+}
+
 // GetSendMsg 获取对应消息内容（任务模式使用）
 // 先根据实例设置的类型取，取不到或者取到的是空，则使用text发送
 func (sm *SendMessageService) GetSendMsg(ins models.SendTasksIns) (string, string) {
@@ -24,6 +24,7 @@ const (
 	MessageTypeCustom          = channels.MessageTypeCustom
 	MessageTypeWeChatOFAccount = channels.MessageTypeWeChatOFAccount
 	MessageTypeMessageNest     = channels.MessageTypeMessageNest
+	MessageTypeAliyunSMS       = channels.MessageTypeAliyunSMS
 )

 // ChannelRegistry 渠道注册表
@@ -95,6 +96,7 @@ func GetGlobalChannelRegistry() *ChannelRegistry {
 		globalChannelRegistry.Register(channels.NewCustomChannel())
 		globalChannelRegistry.Register(channels.NewWeChatOFAccountChannel())
 		globalChannelRegistry.Register(channels.NewMessageNestChannel())
+		globalChannelRegistry.Register(channels.NewAliyunSMSChannel())
 	})
 	return globalChannelRegistry
 }
@@ -0,0 +1,106 @@
+package channels
+
+import (
+	"encoding/json"
+	"fmt"
+	"message-nest/models"
+	"message-nest/service/send_ins_service"
+	"message-nest/service/send_way_service"
+
+	openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client"
+	dysmsapi "github.com/alibabacloud-go/dysmsapi-20170525/v4/client"
+	util "github.com/alibabacloud-go/tea-utils/v2/service"
+	"github.com/alibabacloud-go/tea/tea"
+)
+
+type AliyunSMSChannel struct{ *BaseChannel }
+
+func NewAliyunSMSChannel() *AliyunSMSChannel {
+	return &AliyunSMSChannel{BaseChannel: NewBaseChannel(MessageTypeAliyunSMS, []string{FormatTypeText})}
+}
+
+func (c *AliyunSMSChannel) SendUnified(msgObj interface{}, ins models.SendTasksIns, content *UnifiedMessageContent) (string, string) {
+	auth, ok := msgObj.(send_way_service.WayDetailAliyunSMS)
+	if !ok {
+		return "", "类型转换失败"
+	}
+
+	insService := send_ins_service.SendTaskInsService{}
+	errStr, configInterface := insService.ValidateDiffIns(ins)
+	if errStr != "" {
+		return "", errStr
+	}
+
+	config, ok := configInterface.(models.InsAliyunSMSConfig)
+	if !ok {
+		return "", "阿里云短信config校验失败"
+	}
+
+	// 格式化内容
+	_, formattedContent, err := c.FormatContent(content)
+	if err != nil {
+		return "", err.Error()
+	}
+
+	// 创建阿里云短信客户端
+	client, err := c.createClient(auth.AccessKeyId, auth.AccessKeySecret)
+	if err != nil {
+		return "", fmt.Sprintf("创建阿里云短信客户端失败: %s", err.Error())
+	}
+
+	// 准备模板参数
+	templateParam := map[string]interface{}{
+		"content": formattedContent,
+	}
+	if content.Extra != nil {
+		for k, v := range content.Extra {
+			templateParam[k] = v
+		}
+	}
+
+	templateParamJSON, _ := json.Marshal(templateParam)
+
+	// 发送短信
+	sendSmsRequest := &dysmsapi.SendSmsRequest{
+		PhoneNumbers:  tea.String(config.PhoneNumber),
+		SignName:      tea.String(auth.SignName),
+		TemplateCode:  tea.String(config.TemplateCode),
+		TemplateParam: tea.String(string(templateParamJSON)),
+	}
+
+	runtime := &util.RuntimeOptions{}
+	tryErr := func() error {
+		defer func() {
+			if r := tea.Recover(recover()); r != nil {
+				// 处理panic
+			}
+		}()
+
+		response, _err := client.SendSmsWithOptions(sendSmsRequest, runtime)
+		if _err != nil {
+			return _err
+		}
+
+		if response.Body.Code != nil && *response.Body.Code != "OK" {
+			return fmt.Errorf("发送失败: %s - %s", tea.StringValue(response.Body.Code), tea.StringValue(response.Body.Message))
+		}
+
+		return nil
+	}()
+
+	if tryErr != nil {
+		return "", tryErr.Error()
+	}
+
+	return "", ""
+}
+
+// createClient 创建阿里云短信客户端
+func (c *AliyunSMSChannel) createClient(accessKeyId, accessKeySecret string) (*dysmsapi.Client, error) {
+	config := &openapi.Config{
+		AccessKeyId:     tea.String(accessKeyId),
+		AccessKeySecret: tea.String(accessKeySecret),
+		Endpoint:        tea.String("dysmsapi.aliyuncs.com"),
+	}
+	return dysmsapi.NewClient(config)
+}
@@ -15,6 +15,7 @@ const (
 	MessageTypeCustom          = "Custom"
 	MessageTypeWeChatOFAccount = "WeChatOFAccount"
 	MessageTypeMessageNest     = "MessageNest"
+	MessageTypeAliyunSMS       = "AliyunSMS"
 )

 // UnifiedMessageContent 统一的消息内容结构