wechat-api/internal/service/MessageHandlers/wecom/user_label_push_notice.go

package wecom

import (
	"context"
	"encoding/json"
	"entgo.io/ent/dialect/sql"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"strconv"
	"time"
	"wechat-api/ent"
	"wechat-api/ent/label"
	"wechat-api/ent/labellog"
	"wechat-api/ent/wx"
	"wechat-api/internal/lock"
	"wechat-api/internal/pkg/wechat_ws"
	"wechat-api/internal/svc"
	"wechat-api/workphone/wecom"
)

type UserLabelPushNotice struct {
	svcCtx *svc.ServiceContext
}

func NewUserLabelPushNotice(svcCtx *svc.ServiceContext) *UserLabelPushNotice {
	return &UserLabelPushNotice{
		svcCtx: svcCtx,
	}
}

// Handle 实现 MessageHandlerStrategy 接口
func (f *UserLabelPushNotice) Handle(ctx context.Context, msg *wechat_ws.MsgJsonObject, svcCtx *svc.ServiceContext) error {
	var message wecom.UserLabelPushNoticeMessage
	logx.Infof("msg.Message 的内容是：%s", msg.Message)
	if err := json.Unmarshal([]byte(msg.Message), &message); err != nil {
		return err
	}
	//WxWxId := strconv.FormatInt(message.WxId, 10) //属主微信id
	wxInfo, err := svcCtx.DB.Wx.Query().
		Where(wx.WxidEQ(message.WxId)).
		Only(ctx)
	if err != nil {
		return err
	}
	// 整理标签ID和名称列表
	labelIDs := make([]uint64, 0, len(message.LabelGroups))
	labelNameSet := make(map[string]struct{}) // 用于去重
	for _, labelInfoMessage := range message.LabelGroups {
		if len(labelInfoMessage.Labels) == 0 {
			continue
		}
		for _, LabelInfo := range labelInfoMessage.Labels {
			labelID, err := strconv.ParseUint(LabelInfo.Id, 10, 64)
			if err != nil {
				fmt.Println("labelId 类型转成失败：", err)
				continue
			}
			labelIDs = append(labelIDs, labelID)
			labelNameSet[labelInfoMessage.Name+"-"+LabelInfo.Name] = struct{}{}
		}
	}

	// 提前查询现有 LabelLog 和 Label
	existingLabelLogs, err := svcCtx.DB.LabelLog.Query().
		Where(
			labellog.LabelIDIn(labelIDs...),
			labellog.OrganizationID(wxInfo.OrganizationID),
			labellog.WxID(message.WxId),
		).
		Select(labellog.FieldLabelID).
		All(ctx)
	if err != nil {
		return fmt.Errorf("查询 LabelLog 失败: %w", err)
	}
	existingLabels, err := svcCtx.DB.Label.Query().
		Where(
			label.NameIn(keysFromMap(labelNameSet)...),
			label.OrganizationID(wxInfo.OrganizationID),
		).
		Select(label.FieldName).
		All(ctx)
	if err != nil {
		return fmt.Errorf("查询 Label 失败: %w", err)
	}

	existingLabelLogMap := make(map[uint64]struct{})
	for _, log := range existingLabelLogs {
		existingLabelLogMap[log.LabelID] = struct{}{}
	}
	existingLabelMap := make(map[string]struct{})
	for _, lab := range existingLabels {
		existingLabelMap[lab.Name] = struct{}{}
	}

	var bulkLabelLogs []*ent.LabelLogCreate
	var bulkLabels []*ent.LabelCreate

	for _, labelWx := range message.LabelGroups {

		if len(labelWx.Labels) == 0 {
			continue
		}
		for _, labelChildInfo := range labelWx.Labels {
			// 时间戳处理
			tsInt := int64(labelChildInfo.CreateTime)
			// 插入 LabelLog
			labelWecomId, err := StringToUint64(labelChildInfo.Id)
			if err != nil {
				logx.Errorf("时间戳转换失败: %v (Label ID: %d)", err, labelWx.Id)
				continue
			}
			s := labelWx.Name + "-" + labelChildInfo.Name
			if _, exists := existingLabelLogMap[labelWecomId]; !exists {
				bulkLabelLogs = append(bulkLabelLogs,
					svcCtx.DB.LabelLog.Create().
						SetLabelName(s).
						SetLabelID(labelWecomId).
						SetOrganizationID(wxInfo.OrganizationID).
						SetWxID(message.WxId).
						SetCreatedAt(time.Unix(tsInt, 0)),
				)
			}
			// 插入 Label
			if _, exists := existingLabelMap[s]; !exists {
				bulkLabels = append(bulkLabels,
					svcCtx.DB.Label.Create().
						SetName(s).
						SetType(4).
						SetStatus(1).
						SetOrganizationID(wxInfo.OrganizationID).
						SetFrom(3). // 标签来源：1后台创建 2个微同步 3企业微信同步
						SetMode(1).
						SetConditions(`{}`).
						SetCreatedAt(time.Now()).
						SetUpdatedAt(time.Now()),
				)
			}
		}

	}

	lock.LockWxId(message.WxId)
	defer lock.UnlockWxId(message.WxId)
	// 批量插入 LabelLog
	if len(bulkLabelLogs) > 0 {
		err := svcCtx.DB.LabelLog.CreateBulk(bulkLabelLogs...).
			OnConflict(sql.ConflictColumns(labellog.FieldLabelID, labellog.FieldWxID, labellog.FieldOrganizationID)).
			DoNothing().
			Exec(ctx)
		if err != nil {
			logx.Error("labelLog 批量插入失败", bulkLabelLogs)
		}
	}

	// 批量插入 Label
	if len(bulkLabels) > 0 {
		err := svcCtx.DB.Label.CreateBulk(bulkLabels...).
			OnConflict(sql.ConflictColumns(label.FieldName, label.FieldOrganizationID)).
			DoNothing().
			Exec(ctx)
		if err != nil {
			logx.Error("label 批量插入失败", bulkLabels)
			return err
		}
	}

	return nil
}

func keysFromMap(m map[string]struct{}) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	return keys
}

func StringToUint64(s string) (uint64, error) {
	return strconv.ParseUint(s, 10, 64)
}