workTasks/urlStatic/job/job.go

package job

import (
	uuid2 "app.yhyue.com/moapp/jybase/go-xweb/uuid"
	"context"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/gogf/gf/v2/util/gutil"
	"github.com/pkg/errors"
	"regexp"
	"strings"
	"time"
	"workTasks/common/match"
	"workTasks/urlStatic/clickIterSource"
)

type (
	UserId   string
	UUid     string
	MatchObj struct {
		urlMapping      map[string][]*matchItem
		result          map[UUid]map[UserId]int64 //统计计算次数
		uuidCodeMapping map[UUid]string
		matchObj        *match.TrieNode
	}
	matchItem struct {
		UUid UUid
		Name string
		Code string
		Rule map[string]interface{}
	}
	dbStruct struct {
		UserId string    `json:"userId"`
		Code   string    `json:"code"`
		Num    int64     `json:"num"`
		Date   time.Time `json:"date"`
	}
)

var re = regexp.MustCompile(`\[[^\]]*\]`)

func NewMatchObj(ctx context.Context) *MatchObj {
	var (
		urlMapping      = map[string][]*matchItem{}
		treeMatch       = &match.TrieNode{}
		uuidCodeMapping = map[UUid]string{}
		result          = map[UUid]map[UserId]int64{}
	)
	for _, mObj := range g.Cfg().MustGet(ctx, "match").Maps() {
		var (
			name = gconv.String(mObj["name"])
			code = gconv.String(mObj["code"])
			rule = gconv.Map(mObj["rule"])

			urlArr       []string
			newMatchItem []*matchItem
		)
		for _, s := range gconv.Strings(rule["url"]) {
			urlArr = append(urlArr, s)
		}

		arr := re.FindAllStringSubmatch(code, -1)
		if len(arr) > 0 {
			for _, str := range arr {
				var (
					rep    = str[0]
					key    = rep[1 : len(rep)-1]
					repVal = gconv.Strings(rule[key])
				)
				for _, val := range repVal {
					ruleItem := gconv.Map(gutil.Copy(rule))
					ruleItem[key] = val
					var (
						uuid      = UUid(uuid2.New())
						finalCode = strings.ReplaceAll(code, rep, val)
					)

					uuidCodeMapping[uuid] = finalCode
					newMatchItem = append(newMatchItem, &matchItem{
						UUid: uuid,
						Name: name,
						Code: strings.ReplaceAll(code, rep, val),
						Rule: ruleItem,
					})
				}
			}
		} else {
			uuid := UUid(uuid2.New())
			uuidCodeMapping[uuid] = code
			newMatchItem = append(newMatchItem, &matchItem{
				UUid: uuid,
				Name: name,
				Code: code,
				Rule: rule,
			})
		}

		if len(urlArr) > 0 {
			for _, uStr := range gconv.Strings(rule["url"]) {
				if uStr != "" {
					treeMatch.Insert(uStr)
				}
				urlMapping[uStr] = newMatchItem
			}
		} else {
			urlMapping[""] = newMatchItem
		}

	}

	for uid, _ := range uuidCodeMapping {
		result[uid] = make(map[UserId]int64)
	}
	return &MatchObj{
		matchObj:        treeMatch,
		urlMapping:      urlMapping,
		result:          result,
		uuidCodeMapping: uuidCodeMapping,
	}
}

func (o *MatchObj) Match(ctx context.Context, start, end time.Time) error {
	rows, err := clickIterSource.Ch_analysis.Query(ctx, "SELECT date,url,user_id,mini_program_code,action_id,breaker_name,page_id FROM dwd_f_personnel_behavior WHERE user_id!='' AND date>=? and date<? ", start.Format(time.DateTime), end.Format(time.DateTime))
	if err != nil {
		return errors.Wrap(err, "加载日志数据异常")
	}
	defer rows.Close()
	for rows.Next() {
		var (
			date                                                              time.Time
			user_id, url, mini_program_code, action_id, breaker_name, page_id string
		)
		if err := rows.Scan(
			&date, &url, &user_id, &mini_program_code, &action_id, &breaker_name, &page_id,
		); err != nil {
			g.Log().Errorf(ctx, "字段映射异常 %v", err)
		}
		for _, str := range append(o.matchObj.FindWords(url), "") {
			for _, item := range o.urlMapping[str] {
				var isValid = func() bool {
					for key, val := range item.Rule {
						if key == "url" {
							continue
						}
						switch key {
						case "mini_program_code":
							if !hasPrefix(mini_program_code, val) {
								return false
							}
						case "action_id":
							if !compareValue(action_id, val) {
								return false
							}
						case "breaker_name":
							if !compareValue(breaker_name, val) {
								return false
							}
						case "page_id":
							if !compareValue(page_id, val) {
								return false
							}
						}
					}
					return true
				}()
				if isValid {
					if _, ok := o.result[item.UUid][UserId(user_id)]; ok {
						o.result[item.UUid][UserId(user_id)]++
					} else {
						o.result[item.UUid][UserId(user_id)] = 1
					}
				}
			}
		}
	}
	return nil
}

func compareValue(val string, setting interface{}) bool {
	if v, ok := setting.(string); ok {
		return v == val
	} else if v, ok := setting.([]string); ok {
		for _, s := range v {
			if s == val {
				return true
			}
		}
	}
	return false
}

func hasPrefix(val string, setting interface{}) bool {
	if v, ok := setting.(string); ok {
		return strings.HasPrefix(val, v)
	} else if v, ok := setting.([]string); ok {
		for _, s := range v {
			return strings.HasPrefix(val, s)
		}
	}
	return false
}

func (o *MatchObj) Save(ctx context.Context, data time.Time) error {
	var (
		r     []*dbStruct
		total int
	)
	for uuid, mData := range o.result {
		for userId, num := range mData {
			r = append(r, &dbStruct{
				UserId: string(userId),
				Code:   o.uuidCodeMapping[uuid],
				Num:    num,
				Date:   data,
			})
		}
		if len(r) >= 2000 {
			var t = r
			if _, err := g.DB().Save(ctx, "dwd_d_visit", t, 500); err != nil {
				g.Log().Errorf(ctx, "save errr %v", err)
			}
			total += 500
			r = []*dbStruct{}
		}
	}
	if len(r) > 0 {
		if _, err := g.DB().Save(ctx, "dwd_d_visit", r, 500); err != nil {
			g.Log().Errorf(ctx, "save errr %v", err)
		}
		total += len(r)
	}
	if total > 0 {
		g.Log().Infof(ctx, "任务[%s] 共%d条数据", data, total)
	}
	return nil
}