workTasks/wxSign/wxSignGroup/group.go

package wxSignGroup

import (
	"context"
	"fmt"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"net/rpc"
	"regexp"
)

type (
	SignGroup struct {
		Id    int64  `json:"id"`
		Name  string `json:"name"`
		Count int64  `json:"count"`
	}
	AccessToken struct {
		AccessToken string `json:"access_token"`
		ExpiresIn   int64  `json:"expires_in"`
		LastTime    int64  `json:"last_time"`
	}
	GroupFull struct {
		Sign     *SignGroup
		Old, New []string
	}
)

type WxSignGroup struct {
	data map[int]*GroupFull
}

func NewWxSignManager(appId string) (*WxSignGroup, error) {
	//加载所有分组
	var (
		wsg  = &WxSignGroup{}
		data = map[int]*GroupFull{}
		ctx  = context.Background()

		re = regexp.MustCompile(`^关注第(\d+)天$`)
	)
	//加载分组信息
	list, err := wsg.allGroup(ctx, appId)
	if err != nil {
		return nil, err
	}
	for _, day := range g.Cfg().MustGet(ctx, "subDaySign").Ints() {
		var isExists = false
		for _, group := range list {
			match := re.FindStringSubmatch(group.Name)
			if len(match) == 0 {
				continue
			}
			s := gconv.Int(match[1])
			if s == day {
				data[day] = &GroupFull{
					Sign: group,
					Old:  []string{},
					New:  []string{},
				}
				isExists = true
				break
			}
		}
		if !isExists {
			newGroupName := fmt.Sprintf("关注第%d天", day)
			newGroup, err := wsg.greatGroup(ctx, appId, newGroupName)
			if err != nil {
				g.Log().Errorf(ctx, "创建%s异常 %v", newGroupName, err)
				continue
			}
			data[day] = &GroupFull{
				Sign: newGroup,
				Old:  []string{},
				New:  []string{},
			}
		}
	}
	wsg.data = data
	return wsg, nil
}

//func (wsg *WxSignGroup) getToken(appId string) string {
//	for {
//		ctx := context.Background()
//		gv, err := g.Redis().Get(ctx, fmt.Sprintf("WxToken_%s", g.Cfg().MustGet(ctx, "wxAppId").String()))
//		if err != nil {
//			g.Log().Error(ctx, "获取accessToken异常", err)
//			continue
//		}
//		var accessToken AccessToken
//		if err := gv.Struct(&accessToken); err != nil {
//			g.Log().Error(ctx, "获取反序列化异常", err)
//			continue
//		}
//		if time.Now().Unix() > accessToken.ExpiresIn {
//			g.Log().Error(ctx, "token已过期", err)
//			continue
//		}
//		return accessToken.AccessToken
//	}
//}

func (wsg *WxSignGroup) GetWxAccessToken(ctx context.Context, code string) string {
	var repl string
	client, err := rpc.DialHTTP("tcp", g.Config().MustGet(ctx, "wxTokenRpc").String())
	if err != nil {
		g.Log().Errorf(ctx, code, err)
		return repl
	}
	defer client.Close()
	err = client.Call("WxTokenRpc.GetAccessToken", code, &repl)
	if err != nil {
		g.Log().Errorf(ctx, code, err)
		return repl
	}
	if repl == "" {
		g.Log().Errorf(ctx, "未获取到accessToken")
	} else {
		g.Log().Errorf(ctx, "获取到accessToken %s", repl)
	}
	return repl
}

func (wsg *WxSignGroup) AddNewUsers(dayNum int, openids ...string) error {
	group, ok := wsg.data[dayNum]
	if !ok {
		return fmt.Errorf("未知标签")
	}
	group.New = append(group.New, openids...)
	return nil
}

// LoadGroupUser 加载标签下的用户
func (wsg *WxSignGroup) LoadGroupUser(ctx context.Context, appId string) {
	//加载标签对应用户
	for _, group := range wsg.data {
		var nextOpenid string
		group.Old = []string{}
		for {
			userIds, nextId, err := wsg.getGroupUsers(ctx, appId, group.Sign.Id, nextOpenid)
			if err != nil {
				g.Log().Errorf(ctx, "加载%s用户异常", group.Sign.Name)
				break
			}
			group.Old = append(group.Old, userIds...)
			if nextId == "" {
				break
			}
			nextOpenid = nextId
		}
		g.Log().Infof(ctx, "加载 %s 标签下 %d用户，实际加载%d个", group.Sign.Name, group.Sign.Count, len(group.Old))
	}
	g.Dump(wsg.data)
}

// ClearGroupUser 清除标签下的用户
func (wsg *WxSignGroup) ClearGroupUser(ctx context.Context, appId string) {
	for _, group := range wsg.data {
		for _, openids := range chunkArray(group.Old, 50) {
			if err := wsg.groupDelUser(ctx, appId, group.Sign.Id, openids...); err != nil {
				g.Log().Errorf(ctx, "%s 清除 %s标签下的用户异常 %v", appId, group.Sign.Name, err)
			}
		}
	}
}

// UpdateNewGroupUser 更新最新的标签用户
func (wsg *WxSignGroup) UpdateNewGroupUser(ctx context.Context, appId string) {
	for _, group := range wsg.data {
		for _, openids := range chunkArray(group.New, 50) {
			if err := wsg.groupAddUser(ctx, appId, group.Sign.Id, openids...); err != nil {
				g.Log().Errorf(ctx, "%s %s标签添加用户异常 %v", appId, group.Sign.Name, err)
			}
		}
	}
}

func (wsg *WxSignGroup) DelGroup(ctx context.Context, appId string, ids ...int) {
	for _, groupId := range ids {
		if err := wsg.delGroup(ctx, appId, gconv.Int64(groupId)); err != nil {
			g.Log().Errorf(ctx, "%s 删除群组异常 %v", appId, err)
		}
	}
}

// greatGroup 创建分组
func (wsg *WxSignGroup) greatGroup(ctx context.Context, appId, name string) (*SignGroup, error) {
	res, err := g.Client().Header(map[string]string{"Content-Type": "application/json"}).
		Post(ctx, fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/tags/create?access_token=%s", wsg.GetWxAccessToken(ctx, appId)), g.Map{
			"tag": g.Map{"name": name},
		})
	if err != nil {
		return nil, err
	}
	defer res.Close()

	var rData SignGroup

	cMap := gconv.Map(res.ReadAll())
	if errMsg := gconv.String(cMap["errmsg"]); errMsg != "" {
		return nil, fmt.Errorf(errMsg)
	}
	if err := gconv.Struct(cMap["tag"], &rData); err != nil {
		return nil, err
	}
	return &rData, nil
}

// delGroup 删除分组
func (wsg *WxSignGroup) delGroup(ctx context.Context, appId string, groupId int64) error {
	res, err := g.Client().Header(map[string]string{"Content-Type": "application/json"}).
		Post(ctx, fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/tags/delete?access_token=%s", wsg.GetWxAccessToken(ctx, appId)), g.Map{
			"tag": g.Map{"id": groupId},
		})
	if err != nil {
		return err
	}
	defer res.Close()

	rMap := gconv.Map(res.ReadAll())
	if gconv.String(rMap["errmsg"]) != "ok" {
		return fmt.Errorf("%s 删除分组异常:%s", appId, gconv.String(rMap["errmsg"]))
	}
	return nil
}

// groupAddUser 分组添加用户
func (wsg *WxSignGroup) groupAddUser(ctx context.Context, appId string, groupId int64, openids ...string) error {
	if len(openids) == 0 {
		return nil
	}
	if len(openids) > 50 {
		return fmt.Errorf("超出最大限制")
	}
	res, err := g.Client().Header(map[string]string{"Content-Type": "application/json"}).
		Post(ctx, fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/tags/members/batchtagging?access_token=%s", wsg.GetWxAccessToken(ctx, appId)), g.Map{
			"tagid":       groupId,
			"openid_list": openids,
		})
	if err != nil {
		return err
	}
	defer res.Close()

	rMap := gconv.Map(res.ReadAll())
	if gconv.String(rMap["errmsg"]) != "ok" {
		return fmt.Errorf("%d 添加用户异常:%s", appId, gconv.String(rMap["errmsg"]))
	}
	return nil
}

// groupDelUser 分组删除用户
func (wsg *WxSignGroup) groupDelUser(ctx context.Context, appId string, groupId int64, openids ...string) error {
	if len(openids) == 0 {
		return nil
	}
	if len(openids) > 50 {
		return fmt.Errorf("超出最大限制")
	}
	res, err := g.Client().Header(map[string]string{"Content-Type": "application/json"}).
		Post(ctx, fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/tags/members/batchuntagging?access_token=%s", wsg.GetWxAccessToken(ctx, appId)), g.Map{
			"tagid":       groupId,
			"openid_list": openids,
		})
	if err != nil {
		return err
	}
	defer res.Close()

	rMap := gconv.Map(res.ReadAll())
	if gconv.String(rMap["errmsg"]) != "ok" {
		return fmt.Errorf("%s 标签删除用户异常:%s", appId, gconv.String(rMap["errmsg"]))
	}
	return nil
}

// allGroup 获取所有分组
func (wsg *WxSignGroup) allGroup(ctx context.Context, appId string) ([]*SignGroup, error) {
	res, err := g.Client().Get(ctx, fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/tags/get?access_token=%s", wsg.GetWxAccessToken(ctx, appId)))
	if err != nil {
		return nil, err
	}
	defer res.Close()
	var rData []*SignGroup
	m := gconv.Map(res.ReadAll())
	if err := gconv.Struct(m["tags"], &rData); err != nil {
		return nil, err
	}
	return rData, nil
}

// getGroupUsers 获取所有分组
func (wsg *WxSignGroup) getGroupUsers(ctx context.Context, appId string, groupId int64, next_openid string) ([]string, string, error) {
	res, err := g.Client().Header(map[string]string{"Content-Type": "application/json"}).
		Post(ctx, fmt.Sprintf("https://api.weixin.qq.com/cgi-bin/user/tag/get?access_token=%s", wsg.GetWxAccessToken(ctx, appId)), g.Map{
			"tagid":       groupId,
			"next_openid": next_openid,
		})
	if err != nil {
		return nil, "", err
	}
	defer res.Close()

	var rData = struct {
		Count int `json:"count"`
		Data  struct {
			Openid []string `json:"openid"`
		} `json:"data"`
		NextOpenid string `json:"next_openid"`
	}{}

	if err := gconv.Struct(res.ReadAll(), &rData); err != nil {
		return nil, "", err
	}
	if rData.Data.Openid != nil && len(rData.Data.Openid) > 0 {
		return rData.Data.Openid, rData.NextOpenid, nil
	}
	return nil, "", nil
}

func chunkArray(arr []string, size int) [][]string {
	var result [][]string
	// 遍历一维数组，每次取 size 个元素
	for i := 0; i < len(arr); i += size {
		// 计算切片的结束位置，确保不超出数组的长度
		end := i + size
		if end > len(arr) {
			end = len(arr)
		}
		// 将切片添加到结果数组中
		result = append(result, arr[i:end])
	}
	return result
}