jy-data-extract/udpdataclear/udpSensitiveWords/grpc_server/mgo.go

package main

import (
	"context"
	"fmt"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"gopkg.in/mgo.v2/bson"
	"log"
	"math/big"
	"strconv"
	"time"
)

type Mgo struct {
	Uri                string
	PoolSize           uint64
	mgoEn              *mongo.Client
}

type MgoSess struct {
	Db     string
	Coll   string
	Query  interface{}
	Sorts  []string
	fields interface{}
	limit  int64
	skip   int64
	M      *MongodbSim
}

type MgoIter struct {
	Cursor *mongo.Cursor
}

func (mt *MgoIter) Next(result interface{}) bool {
	if mt.Cursor != nil {
		if mt.Cursor.Next(nil) {
			err := mt.Cursor.Decode(result)
			if err != nil {
				log.Println("mgo cur err", err.Error())
				mt.Cursor.Close(nil)
				return false
			}
			return true
		} else {
			mt.Cursor.Close(nil)
			return false
		}
	} else {
		return false
	}

}

func (ms *MgoSess) DB(name string) *MgoSess {
	ms.Db = name
	return ms
}

func (ms *MgoSess) C(name string) *MgoSess {
	ms.Coll = name
	return ms
}

func (ms *MgoSess) Find(q interface{}) *MgoSess {
	ms.Query = q
	return ms
}

func (ms *MgoSess) Select(fields interface{}) *MgoSess {
	ms.fields = fields
	return ms
}

func (ms *MgoSess) Limit(limit int64) *MgoSess {
	ms.limit = limit
	return ms
}
func (ms *MgoSess) Skip(skip int64) *MgoSess {
	ms.skip = skip
	return ms
}

func (ms *MgoSess) Sort(sorts ...string) *MgoSess {
	ms.Sorts = sorts
	return ms
}

func (ms *MgoSess) Iter() *MgoIter {
	it := &MgoIter{}
	find := options.Find()
	if ms.skip > 0 {
		find.SetSkip(ms.skip)
	}
	if ms.limit > 0 {
		find.SetLimit(ms.limit)
	}
	find.SetBatchSize(100)
	if len(ms.Sorts) > 0 {
		sort := bson.M{}
		for _, k := range ms.Sorts {
			switch k[:1] {
			case "-":
				sort[k[1:]] = -1
			case "+":
				sort[k[1:]] = 1
			default:
				sort[k] = 1
			}
		}
		find.SetSort(sort)
	}
	if ms.fields != nil {
		find.SetProjection(ms.fields)
	}
	cur, err := ms.M.C.Database(ms.Db).Collection(ms.Coll).Find(ms.M.Ctx, ms.Query, find)
	if err != nil {
		log.Println("mgo find err", err.Error())
	} else {
		it.Cursor = cur
	}
	return it
}

type MongodbSim struct {
	MongodbAddr string
	Size        int
	//	MinSize     int
	DbName   string
	C        *mongo.Client
	Ctx      context.Context
	ShortCtx context.Context
	pool     chan bool
	UserName string
	Password string
}

func (m *MongodbSim) GetMgoConn() *MgoSess {
	//m.Open()
	ms := &MgoSess{}
	ms.M = m
	return ms
}

func (m *MongodbSim) DestoryMongoConn(ms *MgoSess) {
	//m.Close()
	ms.M = nil
	ms = nil
}

func (m *MongodbSim) InitPool() {
	opts := options.Client()
	opts.SetConnectTimeout(3 * time.Second)
	opts.ApplyURI("mongodb://" + m.MongodbAddr)
	opts.SetMaxPoolSize(uint64(m.Size))
	m.pool = make(chan bool, m.Size)

	if m.UserName !="" && m.Password !="" {
		cre := options.Credential{
			Username:m.UserName,
			Password:m.Password,
		}
		opts.SetAuth(cre)
	}



	opts.SetMaxConnIdleTime(2 * time.Hour)
	m.Ctx, _ = context.WithTimeout(context.Background(), 99999*time.Hour)
	m.ShortCtx, _ = context.WithTimeout(context.Background(), 1*time.Minute)
	client, err := mongo.Connect(m.ShortCtx, opts)
	if err != nil {
		log.Println("mgo init error:", err.Error())
	} else {
		m.C = client
		log.Println("init success")
	}
}

func (m *MongodbSim) Open() {
	m.pool <- true
}
func (m *MongodbSim) Close() {
	<-m.pool
}

//批量插入
func (m *MongodbSim) UpSertBulk(c string, doc ...[]map[string]interface{}) (map[int64]interface{}, bool) {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	var writes []mongo.WriteModel
	for _, d := range doc {
		write := mongo.NewUpdateOneModel()
		write.SetFilter(d[0])
		write.SetUpdate(d[1])
		write.SetUpsert(true)
		writes = append(writes, write)
	}
	r, e := coll.BulkWrite(m.Ctx, writes)
	if e != nil {
		log.Println("mgo upsert error:", e.Error())
		return nil, false
	}
	//	else {
	//		if r.UpsertedCount != int64(len(doc)) {
	//			log.Println("mgo upsert uncomplete:uc/dc", r.UpsertedCount, len(doc))
	//		}
	//		return true
	//	}
	return r.UpsertedIDs, true
}

//批量插入
func (m *MongodbSim) SaveBulk(c string, doc ...map[string]interface{}) bool {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	var writes []mongo.WriteModel
	for _, d := range doc {
		write := mongo.NewInsertOneModel()
		write.SetDocument(d)
		writes = append(writes, write)
	}
	_, e := coll.BulkWrite(m.Ctx, writes)
	if e != nil {
		log.Println("mgo savebulk error:", e.Error())
		return false
	}
	return true
}

//保存
func (m *MongodbSim) Save(c string, doc map[string]interface{}) interface{} {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	r, err := coll.InsertOne(m.Ctx, doc)
	if err != nil {
		return nil
	}
	return r.InsertedID
}

//更新by Id
func (m *MongodbSim) UpdateById(c, id string, doc map[string]interface{}) bool {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	_, err := coll.UpdateOne(m.Ctx, map[string]interface{}{"_id": StringTOBsonId(id)}, doc)
	if err != nil {
		return false
	}
	return true
}

//删除by id
func (m *MongodbSim) DeleteById(c, id string) int64 {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	r, err := coll.DeleteOne(m.Ctx, map[string]interface{}{"_id": StringTOBsonId(id)})
	if err != nil {
		return 0
	}
	return r.DeletedCount
}

//通过条件删除
func (m *MongodbSim) Delete(c string, query map[string]interface{}) int64 {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	r, err := coll.DeleteMany(m.Ctx, query)
	if err != nil {
		return 0
	}
	return r.DeletedCount
}

//findbyid
func (m *MongodbSim) FindById(c, id string) map[string]interface{} {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	r := coll.FindOne(m.Ctx, map[string]interface{}{"_id": StringTOBsonId(id)})
	v := map[string]interface{}{}
	r.Decode(&v)
	return v
}

//findone
func (m *MongodbSim) FindOne(c string, query map[string]interface{}) map[string]interface{} {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	r := coll.FindOne(m.Ctx, query)
	v := map[string]interface{}{}
	r.Decode(&v)
	return v
}

//find
func (m *MongodbSim) Find(c string, query map[string]interface{}, sort, fields interface{}) ([]map[string]interface{}, error) {
	m.Open()
	defer m.Close()
	coll := m.C.Database(m.DbName).Collection(c)
	op := options.Find()
	r, err := coll.Find(m.Ctx, query, op.SetSort(sort), op.SetProjection(fields))
	if err != nil {
		log.Fatal(err)
		return nil, err
	}
	var results []map[string]interface{}
	if err = r.All(m.Ctx, &results); err != nil {
		log.Fatal(err)
		return nil, err
	}
	return results, nil
}

//创建_id
func NewObjectId() primitive.ObjectID {
	return primitive.NewObjectID()
}



func StringTOBsonId(id string) primitive.ObjectID {
	objectId, _ := primitive.ObjectIDFromHex(id)
	return objectId
}

func BsonTOStringId(id interface{}) string {
	return id.(primitive.ObjectID).Hex()
}






func toMegaBytes(bytes uint64) float64 {
	return float64(bytes) / 1024 / 1024
}

func IntAll(num interface{}) int {
	return IntAllDef(num, 0)
}

func Int64All(num interface{}) int64 {
	if i, ok := num.(int64); ok {
		return int64(i)
	} else if i0, ok0 := num.(int32); ok0 {
		return int64(i0)
	} else if i1, ok1 := num.(float64); ok1 {
		return int64(i1)
	} else if i2, ok2 := num.(int); ok2 {
		return int64(i2)
	} else if i3, ok3 := num.(float32); ok3 {
		return int64(i3)
	} else if i4, ok4 := num.(string); ok4 {
		i64, _ := strconv.ParseInt(i4, 10, 64)
		//in, _ := strconv.Atoi(i4)
		return i64
	} else if i5, ok5 := num.(int16); ok5 {
		return int64(i5)
	} else if i6, ok6 := num.(int8); ok6 {
		return int64(i6)
	} else if i7, ok7 := num.(*big.Int); ok7 {
		in, _ := strconv.ParseInt(fmt.Sprint(i7), 10, 64)
		return int64(in)
	} else if i8, ok8 := num.(*big.Float); ok8 {
		in, _ := strconv.ParseInt(fmt.Sprint(i8), 10, 64)
		return int64(in)
	} else {
		return 0
	}
}

func Float64All(num interface{}) float64 {
	if i, ok := num.(float64); ok {
		return float64(i)
	} else if i0, ok0 := num.(int32); ok0 {
		return float64(i0)
	} else if i1, ok1 := num.(int64); ok1 {
		return float64(i1)
	} else if i2, ok2 := num.(int); ok2 {
		return float64(i2)
	} else if i3, ok3 := num.(float32); ok3 {
		return float64(i3)
	} else if i4, ok4 := num.(string); ok4 {
		in, _ := strconv.ParseFloat(i4, 64)
		return in
	} else if i5, ok5 := num.(int16); ok5 {
		return float64(i5)
	} else if i6, ok6 := num.(int8); ok6 {
		return float64(i6)
	} else if i6, ok6 := num.(uint); ok6 {
		return float64(i6)
	} else if i6, ok6 := num.(uint8); ok6 {
		return float64(i6)
	} else if i6, ok6 := num.(uint16); ok6 {
		return float64(i6)
	} else if i6, ok6 := num.(uint32); ok6 {
		return float64(i6)
	} else if i6, ok6 := num.(uint64); ok6 {
		return float64(i6)
	} else if i7, ok7 := num.(*big.Float); ok7 {
		in, _ := strconv.ParseFloat(fmt.Sprint(i7), 64)
		return float64(in)
	} else if i8, ok8 := num.(*big.Int); ok8 {
		in, _ := strconv.ParseFloat(fmt.Sprint(i8), 64)
		return float64(in)
	} else {
		return 0
	}
}

func IntAllDef(num interface{}, defaultNum int) int {
	if i, ok := num.(int); ok {
		return int(i)
	} else if i0, ok0 := num.(int32); ok0 {
		return int(i0)
	} else if i1, ok1 := num.(float64); ok1 {
		return int(i1)
	} else if i2, ok2 := num.(int64); ok2 {
		return int(i2)
	} else if i3, ok3 := num.(float32); ok3 {
		return int(i3)
	} else if i4, ok4 := num.(string); ok4 {
		in, _ := strconv.Atoi(i4)
		return int(in)
	} else if i5, ok5 := num.(int16); ok5 {
		return int(i5)
	} else if i6, ok6 := num.(int8); ok6 {
		return int(i6)
	} else if i7, ok7 := num.(*big.Int); ok7 {
		in, _ := strconv.Atoi(fmt.Sprint(i7))
		return int(in)
	} else if i8, ok8 := num.(*big.Float); ok8 {
		in, _ := strconv.Atoi(fmt.Sprint(i8))
		return int(in)
	} else {
		return defaultNum
	}
}

func ObjToString(old interface{}) string {
	if nil == old {
		return ""
	} else {
		r, _ := old.(string)
		return r
	}
}

func ObjToStringDef(old interface{}, defaultstr string) string {
	if nil == old {
		return defaultstr
	} else {
		r, _ := old.(string)
		if r == "" {
			return defaultstr
		}
		return r
	}
}

//对象数组转成string数组
func ObjArrToStringArr(old []interface{}) []string {
	if old != nil {
		new := make([]string, len(old))
		for i, v := range old {
			new[i] = v.(string)
		}
		return new
	} else {
		return nil
	}
}

//对象数组转成map数组
func ObjArrToMapArr(old []interface{}) []map[string]interface{} {
	if old != nil {
		new := make([]map[string]interface{}, len(old))
		for i, v := range old {
			new[i] = v.(map[string]interface{})
		}
		return new
	} else {
		return nil
	}
}

//map数组转成对象数组
func MapArrToObjArr(old []map[string]interface{}) []interface{} {
	if old != nil {
		new := make([]interface{}, len(old))
		for i, v := range old {
			new[i] = v
		}
		return new
	} else {
		return nil
	}
}










func InitMgoEn(uri string, poolSize uint64,username_password ... string) (*Mgo, error) {
	//fengweiqiang fwq@123123
	m := Mgo{}
	m.Uri = uri
	if poolSize == 0 {
		m.PoolSize = 100
	}
	//fengweiqiang/fwq@123123
	options_client := options.Client().
		ApplyURI(uri).SetMaxPoolSize(m.PoolSize)
	if len(username_password)==2{
		options_client = options_client.SetAuth(options.Credential{Username: username_password[0], Password: username_password[1]})
	}
	client, err := mongo.Connect(context.Background(), options_client)
	if err != nil {
		return nil, err
	} else {
		m.mgoEn = client
		return &m, nil
	}
}

func (m *Mgo) GetCon() *mongo.Client {
	if m.mgoEn != nil {
		return m.mgoEn
	} else {
		return nil
	}
}

//var zwjeReg *regexp.Regexp = regexp.MustCompile(`([〇零点壹贰叁肆伍陆柒捌玖拾百佰千仟万萬亿億元圆角分整正]{4,40})`)