hongze_open_api/services/day_new.go

package services

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/silenceper/wechat/v2/work/msgaudit"
	"hongze/hongze_open_api/models/tables/day_new"
	"hongze/hongze_open_api/services/alarm_msg"
	"hongze/hongze_open_api/services/wework"
	"hongze/hongze_open_api/utils"
	"strconv"
	"strings"
	"time"
)

func DayNewWeworkMsgRefresh() (err error)  {
	cacheKey := utils.CACHE_KEY_DAYNEW_REFRESH
	//设置redis 防止频繁操作
	if !utils.Rc.SetNX(cacheKey, 1, 1*time.Minute) {
		err = errors.New(fmt.Sprintf("系统处理中，请1分钟之后再试"))
		return
	}
	defer func() {
		if err != nil {
			go alarm_msg.SendAlarmMsg("企业微信 会议存档消息刷新失败, Err："+err.Error(), 1)
		}
	}()
	list := make([]msgaudit.TextMessage, 0)
	//获取最新的拉取日志，得到seq和limit的值
	lastReq, e := day_new.GetLasReqLog()
	if e != nil && e.Error() != utils.ErrNoRow() {
		err = errors.New(fmt.Sprintf("获取最新请求日志失败 err: %v", e))
		return
	}

	now := time.Now()
	seq := uint64(0)
	limit := uint64(100)
	timeout := 5
	seqRes := uint64(0)
	if e == nil {
		seq = lastReq.Seq
		limit = uint64(lastReq.Limit)
	}

	msgAuditClient := wework.NewWeWorkMsgAuditClient()
	list, seqRes, err = msgAuditClient.GetMsgAuditContent(seq, limit, timeout)
	if err != nil {
		err = errors.New(fmt.Sprintf("查询会话存档信息失败 err: %v", err))
		return
	}
	if len(list) == 0 { //本次无消息更新
		return
	}
	//查询5天内的消息，用来判断消息是否已存在
	before5Time := time.Now().Add(-120*time.Hour).Unix() * 1000
	cond := " and msg_time > ?"
	var pars []interface{}
	pars = append(pars, before5Time)
	existList, err := day_new.GetWeworkMsgByCondition(cond, pars)
	if err != nil {
		return
	}
	existMsgMap := make(map[string]struct{}, 0)
	for _, v := range existList {
		existMsgMap[v.MsgId] = struct{}{}
	}
	//如果成功获取到值，则新增一条新的请求日志，是否需要加入事务，防止请求成功后，数据没有正常插入引起
	userMap := make(map[string]struct{}, 0)  //企业内用户
	extendUserMap := make(map[string]struct{}, 0) //企业外联系人
	insertList := make([]*day_new.WeworkMsg, 0)

	for _, v := range list {
		if _, ok := existMsgMap[v.MsgID]; ok {
			continue
		}
		toList, _ := json.Marshal(v.ToList)
		v.ToList = append(v.ToList, v.From) //把发言者的userid也加入到用户map里
		for _, u := range v.ToList {
			if len(u) > 10 && strings.HasPrefix(u, "wm") {
				if _, ok := extendUserMap[u]; !ok {
					extendUserMap[u] = struct{}{}
				}
			}else{
				if _, ok := userMap[u]; !ok {
					userMap[u] = struct{}{}
				}
			}
		}
		tmp := &day_new.WeworkMsg{
			MsgId:      v.MsgID,
			Action:     v.Action,
			From:       v.From,
			ToList:     string(toList),
			RoomId:     v.RoomID,
			MsgTime:    v.MsgTime,
			MsgType:    v.MsgType,
			Content:    v.Text.Content,
			ContentEn:  "",
			ReportId:   0,
			CreateTime: now,
			IsAdd:      0,
			IsDelete:   0,
			ModifyTime: now,
		}
		insertList = append(insertList, tmp)
	}
	//批量插入数据库
	err = day_new.AddWeworkMsgMulti(insertList)
	if err != nil {
		err = errors.New(fmt.Sprintf("批量新增模版消息失败 err: %v", err))
		return
	}

	reqLog := &day_new.WeworkMsgLog{
		Seq:        seqRes,
		Limit:      int(limit),
		Total:      len(list),
		ReqResult:  int8(1),
		CreateTime: now,
		ModifyTime: now,
	}
	err = day_new.AddMsgReqLog(reqLog)
	if err != nil {
		err = errors.New(fmt.Sprintf("新增请求日志失败 err: %v", err))
		return
	}

	//批量插入用户信息
	//查询已有的用户信息，判断是否需要新增
	allUsers, err := day_new.GetAllWeworkUser()
	if err != nil {
		err = errors.New(fmt.Sprintf("查询所有企业微信用户失败 err: %v", err))
		return
	}
	insertUserList := make([]*day_new.WeworkUser, 0)
	userExistMap := make(map[string]struct{}, 0)  //企业内用户
	extendUseExistrMap := make(map[string]struct{}, 0) //企业外联系人
	for _, v := range allUsers {
		if v.WwExtendUserId != "" {
			extendUseExistrMap[v.WwExtendUserId] = struct{}{}
		}
		if v.WwUserId != "" {
			userExistMap[v.WwUserId] = struct{}{}
		}
	}
	for k := range extendUserMap {
		if _, ok := extendUseExistrMap[k]; !ok {
			tmp := &day_new.WeworkUser{
				WwUserId:       "",
				WwExtendUserId: k,
				WwNickName:     "",
				WwDeptId:       0,
				CreateTime:     now,
				ModifyTime:     now,
			}
			insertUserList = append(insertUserList, tmp)
		}
	}

	for k := range userMap {
		if _, ok := userExistMap[k]; !ok {
			tmp := &day_new.WeworkUser{
				WwUserId:       k,
				WwExtendUserId: "",
				WwNickName:     "",
				WwDeptId:       0,
				CreateTime:     now,
				ModifyTime:     now,
			}
			insertUserList = append(insertUserList, tmp)
		}
	}
	if len(insertUserList) > 0 {
		err = day_new.AddWeworkUserMulti(insertUserList)
		if err != nil {
			err = errors.New(fmt.Sprintf("新增企业微信用户失败 err: %v", err))
			return
		}
	}

	//批量翻译
	go DayNewTranslateContent()

	go GetWeWorkUsersNickName()

	return
}

func GetWeWorkUsersNickName() (err error)  {
	defer func() {
		if err != nil {
			go alarm_msg.SendAlarmMsg("企业微信 查询用户昵称失败, Err："+err.Error(), 1)
		}
	}()
	//查询没有昵称的用户库，循环调用查询昵称接口
	users, err := day_new.GetEmptyNickNameWeworkUser()
	if err != nil {
		return
	}
	if len(users) == 0{
		return
	}
	multi := ""
	needChangeIds := ""

	client := wework.NewWeWorkUserNameClient()
	ExtendClient := wework.NewWeWorkExternalContactClient()
	//设置批量更新
	for _, v := range users {
		if v.WwUserId != "" {
			ret, e := client.GetUser(v.WwUserId)
			if e != nil {
				err = e
				return
			}
			if ret.Name != "" {
				needChangeIds += strconv.Itoa(int(v.Id)) + ","
				multi += ` WHEN `+strconv.Itoa(int(v.Id))+` THEN "`+ ret.Name +`"`
			}
		}else if v.WwExtendUserId != "" {
			// todo 获取外部联系人姓名
			ret1, e := ExtendClient.GetExternalContact(v.WwExtendUserId)
			if e != nil {
				err = e
				return
			}
			if ret1.ExternalContact.Name != "" {
				needChangeIds += strconv.Itoa(int(v.Id)) + ","
				multi += ` WHEN `+strconv.Itoa(int(v.Id))+` THEN "`+ ret1.ExternalContact.Name +`"`
			}
		}
	}

	if needChangeIds != "" {
		needChangeIds = strings.Trim(needChangeIds, ",")
		err = day_new.MultiUpdateWeworkUserName(multi, needChangeIds)
		if err != nil {
			err = errors.New(fmt.Sprintf("更新企业微信成员姓名失败 err: %v", err))
			return
		}
	}

	return
}

func DayNewTranslateContent() (err error) {
	cacheKey := utils.CACHE_KEY_DAYNEW_TRANSLATE
	//设置redis 防止频繁操作，控制每分钟执行1次
	if !utils.Rc.SetNX(cacheKey, 1, 1*time.Minute) {
		err = errors.New(fmt.Sprintf("系统处理中，请1分钟之后再试"))
		return
	}
	defer func() {
		if err != nil {
			go alarm_msg.SendAlarmMsg("企业微信 中翻英操作失败, Err："+err.Error(), 1)
		}
	}()
	//查询待翻译的内容列表
	condition := ` and (content_en = "" or content_en is null)`
	var pars []interface{}
	list, err := day_new.GetWeworkMsgByConditionLimit(condition, pars, 2000) //默认最多查询2000条
	if err != nil {
		err = errors.New(fmt.Sprintf("查询未翻译的聊天记录失败 err: %v", err))
		return
	}
	multi := ""
	needChangeIds := ""
	count := 0
	contentMap := make(map[string]string, 0)
	contentEnMap := make(map[string]string)
	for _, v := range list {
		//如果单条翻译的字符数超过1000，则直接翻译，否则批量翻译
		if len(v.Content) > 1000 {
			en, e := AliTranslate(v.Content)
			if e != nil {
				err = e
				return
			}
			needChangeIds += strconv.Itoa(int(v.Id)) + ","
			multi += ` WHEN `+strconv.Itoa(int(v.Id))+` THEN "`+ en +`"`
		}else{
			if count >= 50 {  //待翻译的条数不能超过50; 单条翻译字符数不能超过1000字符
				contentEnMap, err = batchTranslateHandler(contentMap)
				if err != nil {
					return
				}
				// 拼接更新sql
				for rk, rv := range contentEnMap {
					needChangeIds += rk + ","
					multi += ` WHEN `+rk+` THEN "`+ rv +`"`
				}
				contentMap = make(map[string]string, 0)
				count = 0
			}
			contentMap[strconv.Itoa(int(v.Id))] = v.Content
			count += 1
		}
	}
	//剩余不满50条的content
	if count > 0 {
		contentEnMap, err = batchTranslateHandler(contentMap)
		if err != nil {
			return
		}
		// 拼接更新sql
		for rk, rv := range contentEnMap {
			needChangeIds += rk + ","
			multi += ` WHEN `+rk+` THEN "`+ rv +`"`
		}
	}

	if needChangeIds != "" {
		needChangeIds = strings.Trim(needChangeIds, ",")
		err = day_new.MultiUpdateContentEn(multi, needChangeIds)
		if err != nil {
			err = errors.New(fmt.Sprintf("更新翻译后的内容失败 err: %v", err))
			return
		}
	}
	return
}

func batchTranslateHandler(contentMap map[string]string) (contentEnMap map[string]string, err error)  {
	bytes,_ := json.Marshal(contentMap)
	content := string(bytes)
	en, err := AliTranslate(content)
	if err != nil {
		return
	}
	//json转为map数据结构
	err = json.Unmarshal([]byte(en), &contentEnMap)
	return
}