package batch_msg import ( "context" "encoding/json" "errors" "github.com/suyuan32/simple-admin-common/msg/errormsg" "github.com/suyuan32/simple-admin-common/utils/uuidx" "github.com/zeromicro/go-zero/core/errorx" "strings" "time" "wechat-api/ent" "wechat-api/ent/contact" "wechat-api/ent/custom_types" "wechat-api/ent/label" "wechat-api/ent/labelrelationship" "wechat-api/hook/aliyun" "wechat-api/internal/utils/dberrorhandler" "wechat-api/internal/svc" "wechat-api/internal/types" "github.com/zeromicro/go-zero/core/logx" ) type CreateWhatcappBatchMsgLogic struct { logx.Logger ctx context.Context svcCtx *svc.ServiceContext } func NewCreateWhatcappBatchMsgLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateWhatcappBatchMsgLogic { return &CreateWhatcappBatchMsgLogic{ Logger: logx.WithContext(ctx), ctx: ctx, svcCtx: svcCtx} } func (l *CreateWhatcappBatchMsgLogic) CreateWhatcappBatchMsg(req *types.BatchMsgInfo) (*types.BaseMsgResp, error) { organizationId := l.ctx.Value("organizationId").(uint64) var err error var tagstring, tag string allContact := false tagIdArray := make([]uint64, 0) for _, labelId := range req.Labels { tagIdArray = append(tagIdArray, labelId) if labelId == uint64(0) { //全部 allContact = true } } startTime := time.Now() sendNow := true // 把 req.Msg 字符串的内容 json_decode 到 msgArray // 然后再把每一条信息都给所有指定的用户记录到 message_records 表 var msgArray []custom_types.Action if req.Msg != nil && *req.Msg != "" { err = json.Unmarshal([]byte(*req.Msg), &msgArray) if err != nil { return nil, errors.New("解析JSON失败") } } var msgActionList []custom_types.Action if len(msgArray) > 0 { msgActionList = make([]custom_types.Action, len(msgArray)) for i, msg := range msgArray { if msg.Type == 1 { msgActionList[i] = custom_types.Action{ Type: msg.Type, Content: msg.Content, } } else { msgActionList[i] = custom_types.Action{ Type: msg.Type, Content: msg.Content, Meta: &custom_types.Meta{ Filename: msg.Meta.Filename, }, } } } } userList := make([]*ent.Contact, 0) tagMap := make(map[string][]uint64) if allContact { // 获取 contact 表中 cc+phone 匹配的联系人 userList, err = l.svcCtx.DB.Contact.Query().Where( contact.OrganizationID(organizationId), contact.Ctype(2), ).All(l.ctx) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, req) } tagids := make([]uint64, 0, 1) tagids = append(tagids, uint64(0)) tagMap["contact_tag"] = tagids tagByte, err := json.Marshal(tagMap) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, req) } tagstring = string(tagByte) tag = "全部" } else { if allContact { // 所有联系人 // 获取 contact 表中 wx_wxid 等于 req.Fromwxid 的 type 为1的数据 userList, err = l.svcCtx.DB.Contact.Query().Where( contact.OrganizationID(organizationId), contact.Ctype(2), ).All(l.ctx) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, req) } tag = "全部联系人" } else { userList, err = l.getContactList(req.Labels, *req.Cc, *req.Phone) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, req) } } tagMap["contact_tag"] = req.Labels tagByte, err := json.Marshal(tagMap) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, req) } tagstring = string(tagByte) tagArray := l.getLabelListByIds(req.Labels, req.GroupLabels) if tag != "" { tagArray = append(tagArray, tag) } tag = strings.Join(tagArray, ",") } // 这里是根据userlist 和 批量消息数 获得最终待发送消息总数 total := int32(len(userList)) * int32(len(msgActionList)) l.Logger.Infof("userList=%v user_total=%v\n", userList, total) if total == 0 { return &types.BaseMsgResp{Msg: "未查询到收信人,请重新选择", Code: 3}, nil } //TODO 发送批量消息 sto := make([]string, 0) for _, v := range userList { sto = append(sto, v.Cc+v.Phone) } fullPhone := *req.Cc + *req.Phone result, err := aliyun.SendBatchChatappMessage(*req.TemplateCode, *req.Lang, fullPhone, sto) if err != nil { return nil, errorx.NewInvalidArgumentError(err.Error()) } l.Logger.Infof("send_batch_chatapp_message result=%v\n", result) batchNo := uuidx.NewUUID().String() var sendTime *time.Time if !sendNow { sendTime = &startTime } tx, err := l.svcCtx.DB.Tx(context.Background()) _, err = tx.BatchMsg.Create(). SetNotNilBatchNo(&batchNo). SetStatus(0). //SetNotNilFromwxid(req.Fromwxid). SetNotNilMsg(req.Msg). SetNotNilTag(&tag). SetNotNilPhone(req.Phone). SetNotNilCc(req.Cc). SetNotNilTagids(&tagstring). SetTotal(total). SetNotNilTaskName(req.TaskName). SetNotNilStartTime(&startTime). SetNillableSendTime(sendTime). SetType(3). SetNotNilOrganizationID(&organizationId). SetCtype(2). Save(l.ctx) if err != nil { _ = tx.Rollback() l.Logger.Errorf("insert whatsapp batch_msg err=%v\n", err) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, req) } } msgs := make([]*ent.MsgCreate, 100) for idx, user := range userList { phone := user.Cc + user.Phone msgRow := tx.Msg.Create(). SetNotNilCc(req.Cc). SetNotNilPhone(req.Phone). SetNotNilToid(&phone). SetMsgtype(int32(1)). SetNotNilMsg(req.Msg). SetStatus(1). SetNotNilBatchNo(&batchNo) msgs = append(msgs, msgRow) // 100条插入一次 if idx == 100 { _, err = tx.Msg.CreateBulk(msgs...).Save(l.ctx) if err != nil { _ = tx.Rollback() l.Logger.Errorf("msg CreateBulk err: %v", err) } } msgs = make([]*ent.MsgCreate, 100) } // 不足100条插入一次 if len(msgs) > 0 { _, err = tx.Msg.CreateBulk(msgs...).Save(l.ctx) if err != nil { _ = tx.Rollback() l.Logger.Errorf("msg CreateBulk err: %v", err) } } _ = tx.Commit() return &types.BaseMsgResp{Msg: errormsg.CreateSuccess}, nil } func (l *CreateWhatcappBatchMsgLogic) getContactList(labels []uint64, cc, phone string) ([]*ent.Contact, error) { // 获取 label_relationship 表中,label_id 等于 labids 的 contact_id labelrelationships, err := l.svcCtx.DB.LabelRelationship.Query().Where(labelrelationship.LabelIDIn(labels...)).All(l.ctx) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, nil) } contact_ids := make([]uint64, 0, len(labelrelationships)) for _, labelrelationship := range labelrelationships { contact_ids = append(contact_ids, labelrelationship.ContactID) } userList := make([]*ent.Contact, 0) if len(contact_ids) > 0 { // 获取 contact 表中 wx_wxid 等于 req.Fromwxid 并且 id 等于 contact_ids 并且 type 为1或2 的数据 userList, err = l.svcCtx.DB.Contact.Query().Where( contact.Cc(cc), contact.Phone(phone), contact.IDIn(contact_ids...), contact.Ctype(2), ).All(l.ctx) if err != nil { return nil, dberrorhandler.DefaultEntError(l.Logger, err, nil) } } return userList, nil } func (l *CreateWhatcappBatchMsgLogic) getLabelListByIds(labels, groupLabels []uint64) []string { result := make([]string, 0) labels = append(labels, groupLabels...) if len(labels) > 0 { contacts, err := l.svcCtx.DB.Label.Query().Where( label.IDIn(labels...), ).Select("name").All(l.ctx) l.Logger.Infof("contacts=%v", contacts) if err != nil { return result } for _, val := range contacts { result = append(result, val.Name) } } return result }