wip:芝麻上报数据提交

common/db.go

@@ -0,0 +1,47 @@
+package common
+
+import (
+	elastic "app.yhyue.com/moapp/jybase/es"
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/os/gctx"
+)
+
+type (
+	esConf struct {
+		Address  string
+		Size     int
+		Version  string
+		UserName string
+		Password string
+	}
+)
+
+var (
+	elasticConf, otherElasticConf esConf
+	RedisTime                     int64
+	PortraitCount                 int
+	OtherElastic                  elastic.Es
+)
+
+func init() {
+	initCtx := gctx.New()
+	if cfg := g.Cfg().MustGet(initCtx, "elasticsearch.default"); !cfg.IsNil() {
+		if err := cfg.Struct(&elasticConf); err == nil {
+			elastic.NewEs(elasticConf.Version, elasticConf.Address, elasticConf.Size, elasticConf.UserName, elasticConf.Password)
+		}
+		g.Log().Info(initCtx, "初始化 elastic.default 完成")
+	}
+	if cfg := g.Cfg().MustGet(initCtx, "elasticsearch.other"); !cfg.IsNil() {
+		if err := cfg.Struct(&otherElasticConf); err == nil {
+			OtherElastic = &elastic.EsV7{
+				Address:  otherElasticConf.Address,
+				UserName: otherElasticConf.UserName,
+				Password: otherElasticConf.Password,
+				Size:     otherElasticConf.Size,
+			}
+			OtherElastic.Init()
+		}
+		g.Log().Info(initCtx, "初始化 elastic.other 完成")
+	}
+
+}

common/mg.go

@@ -0,0 +1,63 @@
+package common
+
+import (
+	"app.yhyue.com/moapp/jybase/mongodb"
+	"context"
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/util/gconv"
+)
+
+type (
+	mgConf struct {
+		Address  string `json:"address"`
+		Size     int    `json:"size"`
+		DbName   string `json:"dbName"`
+		ReplSet  string `json:"replSet"`
+		UserName string `json:"userName"`
+		Password string `json:"password"`
+	}
+	mgObj struct {
+		mgMap map[string]*mongodb.MongodbSim
+	}
+)
+
+var (
+	MG = &mgObj{
+		mgMap: map[string]*mongodb.MongodbSim{},
+	}
+)
+
+func init() {
+	var (
+		ctx = context.Background()
+	)
+	for key, val := range g.Cfg().MustGet(ctx, "mongodb").Map() {
+		var cf mgConf
+		if err := gconv.Struct(val, &cf); err != nil {
+			g.Log().Errorf(ctx, "初始化mongodb %s出错", key)
+			continue
+		}
+		mgo := &mongodb.MongodbSim{
+			MongodbAddr: cf.Address,
+			Size:        cf.Size,
+			DbName:      cf.DbName,
+			ReplSet:     cf.ReplSet,
+			UserName:    cf.UserName,
+			Password:    cf.Password,
+		}
+		mgo.InitPool()
+		MG.mgMap[key] = mgo
+		g.Log().Infof(ctx, "初始化 mongodb %s 成功", key)
+	}
+}
+
+func (t *mgObj) DB(alias ...string) *mongodb.MongodbSim {
+	if len(alias) != 0 {
+		for _, s := range alias {
+			if esObj, ok := t.mgMap[s]; ok {
+				return esObj
+			}
+		}
+	}
+	return t.mgMap["default"]
+}

zhimaPush/README.MD

@@ -0,0 +1,22 @@
+# P602企业画像增加芝麻-曝光数据上报
+
+## 1.对接文档地址
+
+    支付宝接口文档地址：https://opendocs.alipay.com/pre-open/c680c2d7_zhima.credit.ep.dossier.label.query?pathHash=5ed0dc58
+
+### 
+
+
+## 2.曝光日志数据存储
+
+ 日志库 qfw.jy_zhima_logs
+```json
+
+{
+    "_id" : ObjectId("66dfdd32c2fbfed580d3a6d0"),
+    "s_userId" : "1719179967973031936",
+    "l_date" : NumberLong(1725947183),
+    "s_name" : "中移建设有限公司",
+    "credit_no" : "911305007183212099"
+}
+```

zhimaPush/config.yaml

@@ -0,0 +1,17 @@
+mongodb:
+  log:
+    address: "192.168.3.149:27190"
+    size: 5
+    dbName: "qfw"
+    replSet: ""
+    userName: "admin"
+    password: "123456"
+
+# 每天凌晨2点推送前一天访问数据
+runCron: "# 0 2 * * *"
+
+ali:
+  appid: "2021003111633206"
+  sceneCode: "JIANYU_LABEL"
+  apiPath: "https://openapi.alipay.com/gateway.do"
+  privateKey: "-----BEGIN RSA PRIVATE KEY-----\nMIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCaJnmz3N1Z43jxtmZJmPkInl3FbfudxjicThHj+Hkndp+muzZXlAPcWoYTFLAfVa6xg9xAc58fWTPUGOpouh0ZDEP8TGvD784AljDNwRXPI9+afjcxN8gzsNPlqA6KVkAu5u5uEI1HHQgtdyRoCEnXgQRXj6DqAs9olbx8HrTxHjZzjPd+HXv+v4na3zAvZsvPdhLB2ggpVW6uQN7ur8C+hVpB3ZO+gGVFHy3m1t08XgscljyOA9Vpk4T8TnKZqDvnI6ehWFCGeD11yoe3ipHJJdP/+u9/tXiKhm29mJvcM9C6gab6BOQABByQPb2nPFFyfPglbRYrYxgKp/eXUQTzAgMBAAECggEAO6BCVyZJoKlycNzC+Bjpy0Gmt+Vw/HheEWfJV8sW4gLqgk3MGVgxnhyN2/eWoKg96IA+w9d1Z2TTvWjd0yCafMPX0FGnJaUuo990hUH4s6UGqvXDV5BiKkxtylmVNkFilkHRXFWesDE0ppF9TZhXJ7rJ+Syfa6UmoSe6PrCKAPG4FD8MH1ODj5b42l/TRIFM948nTtTLFAFgd77P5ULBCKVfBB5V9XdKhLRdwG9RxugWwkxuCiMOtmk45DzA/PQgvkIzEViD/2NJX67/S88B0G63zdQizJrvxsLp9fThmjhlo/+CaF+a+nfIzg81TfQHlRuTjiYc7iE9n5K0sU82wQKBgQDydI9pEjkPH+ZEixEymdx8D4workBfa+GXgobJ2UwatqCJOCYqZJR0MhjTVeDJiiTGFiemNyZHT0WFQedAQ/O9mkRf0MRhzR6ZOiMnCG5pSSOKdmQXoqjOqcMzdgWBQRol0eLB/vto8M3bpt1VautR0qQOR7BrFsj57K3asofikQKBgQCiwwie5gyCAYB1gkp4QkLmQDSTJdZaTlzzx2oa2/ZB4Pg6wWWNJIxEfSwtuGVH4jGJl0Eer2YpHzaY7f2NfOWJCpNI/aRhM/RilQsFXTox/5bqlbAfspPMqSHnry2OgbZb9CRF3jH7kt65390SKde4A4JNE+jJ0Xb07Tzcr7+pQwKBgCDJ8KcWBs9r0flNEVtVgYLpm1vZDO2aqlySE2Zy0579OcK07Dxpc8YvDDpwb8DxxL1hO6V36d5U8yLtYbfIOMtYdIy8sAJF+jBgs2ZQ9pWYhYzxK+IiZk/kl3h7Ko1rX3zj4JW5xQhRIJXnACHQHtjm8rbXrR0ebLmX738XC9dxAoGAQfUDQ1E6dFyjb/syrnU7ow8VvyOwL7xd0rRW/TdyHdRLA5Rn0MCkzwRiqMx2zqCNLpLK4uJ1ZeiuOV8IRxP7+IL+mK+gs3tYZ7K6Wpwi/yvyPnIWg2Hcny1S/g5xSbSq76yKHn4ihy5Hb8+WSMwIDRvu68lLpq1ppWuWJGBcLP0CgYAnGtHlFY887zSI0aRdzV3EYm/sPnmYy0TWIUVLuGL8kkprcSIW2yRxx1GtpB+3A7cPu61WwRbiswJA6VmQHvn8f9ds81+aiO5KQLMk4MbrTwWhWLDpP4BxjWANVzPr19ZwmDJcYkh7YUI2iLqH7dXho5k62OYCM1T53xlql6HDcg==\n-----END RSA PRIVATE KEY-----"

zhimaPush/main.go

@@ -0,0 +1,237 @@
+package main
+
+import (
+	"context"
+	"crypto"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/sha256"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/os/gctx"
+	"github.com/gogf/gf/v2/os/glog"
+	"github.com/gogf/gf/v2/util/gconv"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"hash"
+	"net/url"
+	"strings"
+	"time"
+	"workTasks/common"
+)
+
+var (
+	jyPushManager *pushManager = initPush(gctx.New())
+)
+
+type (
+	pushManager struct {
+		Appid      string
+		SceneCode  string
+		ApiPath    string
+		PrivateKey *rsa.PrivateKey
+	}
+
+	bizContent struct {
+		FeedbackData []feedBackData `json:"feedback_data"`
+		SceneCode    string         `json:"scene_code"` //JIANYU_LABEL
+	}
+
+	feedBackData struct {
+		EpCertNo      string   `json:"ep_cert_no"`
+		ActionType    string   `json:"action_type"`
+		CountType     string   `json:"count_type"`
+		ActionContent []string `json:"action_content"`
+		ActionDate    string   `json:"action_date"`
+		ActionCount   string   `json:"action_count"`
+	}
+)
+
+func initPush(ctx context.Context) *pushManager {
+	privateKey, err := parsePrivateKey(g.Cfg().MustGet(ctx, "ali.privateKey").String())
+	if err != nil {
+		g.Log().Panic(ctx, err)
+	}
+	return &pushManager{
+		Appid:      g.Cfg().MustGet(ctx, "ali.appid").String(),
+		ApiPath:    g.Cfg().MustGet(ctx, "ali.apiPath").String(),
+		SceneCode:  g.Cfg().MustGet(ctx, "ali.sceneCode").String(),
+		PrivateKey: privateKey,
+	}
+}
+
+func main() {
+	ctx := gctx.New()
+	//c, e := gcron.Add(ctx, g.Cfg().MustGet(ctx, "runCron2", "5 * * * * *").String(), jyPushManager.RunJob)
+	//if e != nil {
+	//	g.Log().Errorf(ctx, "initAndUpdateDeptManager cron任务异常 %v", e)
+	//}
+	//c.Status()
+	if err := jyPushManager.DoPush(ctx, 2, 1, "91320214579517104U", "2024-09-10"); err != nil {
+		g.Log().Print(ctx, err)
+	}
+	//jyPushManager.RunJob(ctx)
+	select {}
+}
+
+func (m *pushManager) RunJob(ctx context.Context) {
+	sess := common.MG.DB("log").GetMgoConn()
+	defer common.MG.DB().DestoryMongoConn(sess)
+
+	yesterday := time.Now().AddDate(0, 0, -1)
+	start := time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, yesterday.Location())
+
+	pipeline := mongo.Pipeline{
+		bson.D{{"$match", g.Map{"l_date": g.Map{"$gte": start.Unix(), "$lt": start.Unix() + 60*60*24}}}},
+		bson.D{
+			{"$group", bson.D{
+				{"_id", "$credit_no"},
+				{"unique_users", bson.D{
+					{"$addToSet", bson.D{
+						{"$cond", bson.D{
+							{"if", bson.D{
+								{"$eq", []interface{}{"$s_userId", ""}},
+							}},
+							{"then", "empty"},
+							{"else", "$s_userId"},
+						}},
+					}},
+				}},
+				{"count", bson.D{
+					{"$sum", 1},
+				}},
+			}},
+		},
+		bson.D{
+			{"$project", bson.D{
+				{"_id", 1},
+				{"count", 1},
+				{"unique_user_count", bson.D{
+					{"$size", "$unique_users"},
+				}},
+			}},
+		},
+	}
+	// 执行聚合查询
+	cursor, err := sess.M.C.Database("qfw").Collection("jy_zhima_logs").Aggregate(context.TODO(), pipeline)
+	if err != nil {
+		glog.Panic(ctx, err)
+	}
+
+	defer cursor.Close(context.TODO())
+
+	// 遍历结果
+	for cursor.Next(context.TODO()) {
+		var result bson.M
+		if err := cursor.Decode(&result); err != nil {
+			g.Log().Errorf(ctx, "获取行数据异常 %v", err)
+			continue
+		}
+		var (
+			ent = gconv.String(result["_id"])
+			pv  = gconv.Int(result["count"])
+			uv  = gconv.Int(result["unique_user_count"])
+		)
+		if ent != "" && pv > 0 {
+			if err := m.DoPush(ctx, pv, uv, ent, start.Format(time.DateOnly)); err != nil {
+				g.Log().Errorf(ctx, "上报数据异常\nparam ent:%v pv:%v uv:%v\nerror: %v", ent, pv, uv, err)
+			}
+		}
+	}
+}
+
+func (m *pushManager) DoPush(ctx context.Context, pv, uv int, ent, dateStr string) error {
+	//g.Log().Printf(ctx, "模拟上报数据\nparam ent:%v pv:%v uv:%v", ent, pv, uv)
+	//return nil
+	bizC := bizContent{
+		FeedbackData: []feedBackData{
+			{
+				EpCertNo:      ent,
+				ActionType:    "PV",
+				CountType:     "accurate",
+				ActionContent: []string{"shortvideo_sales_top", "shortvideo_sales_top_num"},
+				ActionDate:    dateStr,
+				ActionCount:   gconv.String(pv),
+			},
+			{
+				EpCertNo:      ent,
+				ActionType:    "UV",
+				CountType:     "accurate",
+				ActionContent: []string{"label_definition", "shortvideo_sales_top_num"},
+				ActionDate:    dateStr,
+				ActionCount:   gconv.String(uv),
+			},
+		},
+		SceneCode: m.SceneCode,
+	}
+	dataStr, err := json.Marshal(bizC)
+	if err != nil {
+		return err
+	}
+	var data = url.Values{}
+	data.Add("app_id", g.Cfg().MustGet(ctx, "ali.appid").String())
+	data.Add("method", "zhima.credit.ep.acceptance.label.use")
+	data.Add("format", "json")
+	data.Add("charset", "UTF-8")
+	data.Add("sign_type", "RSA2")
+	data.Add("timestamp", time.Now().Format("2006-01-02 15:04:05"))
+	data.Add("version", "1.0")
+	data.Add("biz_content", string(dataStr))
+
+	signContentBytes, _ := url.QueryUnescape(data.Encode())
+	signature, err := m.getSign([]byte(signContentBytes))
+	if err != nil {
+		return err
+	}
+
+	data.Add("sign", signature)
+
+	fmt.Println(fmt.Sprintf("%s?%s", m.ApiPath, data.Encode()))
+
+	res, err := g.Client().Post(ctx, fmt.Sprintf("%s?%s", m.ApiPath, data.Encode()))
+	if err != nil {
+		return err
+	}
+	defer res.Close()
+
+	respByte := res.ReadAll()
+	g.Log().Print(ctx, "---", string(respByte))
+	return nil
+}
+
+func (m *pushManager) getSign(data []byte) (signature string, err error) {
+	var h hash.Hash
+	var hType crypto.Hash
+	h = sha256.New()
+	hType = crypto.SHA256
+
+	h.Write(data)
+	d := h.Sum(nil)
+	bs, err := rsa.SignPKCS1v15(rand.Reader, m.PrivateKey, hType, d)
+	if err != nil {
+		return "", err
+	}
+	signature = base64.StdEncoding.EncodeToString(bs)
+	return
+}
+
+// 加载秘钥
+func parsePrivateKey(key string) (*rsa.PrivateKey, error) {
+	block, _ := pem.Decode([]byte(key))
+	if block == nil {
+		return nil, errors.New("私钥格式不正确")
+	}
+	if strings.ToUpper(block.Type) != "RSA PRIVATE KEY" {
+		return nil, errors.New("私钥类型不正确" + block.Type)
+	}
+	rsaPrivateKey, err := x509.ParsePKCS8PrivateKey(block.Bytes)
+	if err != nil {
+		return nil, err
+	}
+	return rsaPrivateKey.(*rsa.PrivateKey), nil
+}