银行信息

data_export_bank/src/config.json

@@ -0,0 +1,23 @@
+{
+  "bid_mgodb": {
+    "addr": "127.0.0.1:27017",
+    "db": "zhengkun",
+    "coll": "extract_test",
+    "pool": 5
+  },
+  "qy_mgodb": {
+    "addr": "127.0.0.1:27017",
+    "db": "zhengkun",
+    "coll": "extract_test",
+    "pool": 5
+  }
+
+
+
+
+
+
+
+
+
+}

data_export_bank/src/main.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"log"
+	qu "qfw/util"
+)
+
+var (
+	sysconfig    			map[string]interface{}
+	bid_mgo,qy_mgo        	*MongodbSim
+	bid_coll,qy_coll		string
+)
+func initMgo()  {
+
+	bidconf := sysconfig["bid_mgodb"].(map[string]interface{})
+	bid_coll = qu.ObjToString(bidconf["coll"])
+	bid_mgo = &MongodbSim{
+		MongodbAddr: bidconf["addr"].(string),
+		DbName:      bidconf["db"].(string),
+		Size:        qu.IntAllDef(bidconf["pool"], 5),
+	}
+	bid_mgo.InitPool()
+
+	qyconf := sysconfig["qy_mgodb"].(map[string]interface{})
+	qy_coll = qu.ObjToString(qyconf["coll"])
+	bid_mgo = &MongodbSim{
+		MongodbAddr: qyconf["addr"].(string),
+		DbName:      qyconf["db"].(string),
+		Size:        qu.IntAllDef(qyconf["pool"], 5),
+	}
+	bid_mgo.InitPool()
+
+}
+
+
+func init()  {
+	qu.ReadConfig(&sysconfig)
+	initMgo()
+}
+func main()  {
+	log.Println("测试...")
+
+
+
+
+
+
+}

data_export_bank/src/mgo.go

@@ -0,0 +1,329 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+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()
+}

listen_data/src/main.go

@@ -99,34 +99,34 @@ func main()  {
 	//}
 	//save_mgo.InitPool()
 
+	save_mgo = &MongodbSim{
+		MongodbAddr: "127.0.0.1:27017",
+		DbName:      "zhengkun",
+		Size:        10,
+		UserName: "",
+		Password: "",
+	}
+	save_mgo.InitPool()
+
 	//save_mgo = &MongodbSim{
-	//	MongodbAddr: "127.0.0.1:27017",
-	//	DbName:      "zhengkun",
+	//	MongodbAddr: "172.17.4.85:27080",
+	//	DbName:      "qfw",
 	//	Size:        10,
 	//	UserName: "",
 	//	Password: "",
 	//}
 	//save_mgo.InitPool()
 
+
 	//save_mgo = &MongodbSim{
-	//	MongodbAddr: "172.17.4.85:27080",
+	//	MongodbAddr: "172.17.145.163:27083,172.17.4.187:27082",
 	//	DbName:      "qfw",
 	//	Size:        10,
-	//	UserName: "",
-	//	Password: "",
+	//	UserName: "zhengkun",
+	//	Password: "zk@123123",
 	//}
 	//save_mgo.InitPool()
 
-
-	save_mgo = &MongodbSim{
-		MongodbAddr: "172.17.145.163:27083,172.17.4.187:27082",
-		DbName:      "qfw",
-		Size:        10,
-		UserName: "zhengkun",
-		Password: "zk@123123",
-	}
-	save_mgo.InitPool()
-
 	/*
 		repairYiXieChaData() 		//医械查
 		dealWithClassData()			//AB 匹配
@@ -134,6 +134,11 @@ func main()  {
 		repairEqClassData()			//同分类-补充-0123
 	*/
 
+
+
+
+
+	decodeJyUrl()
 	return
 
 

listen_data/src/zkmethod.go

@@ -40,7 +40,7 @@ var MysqlTool	*Mysql
 
 //解密
 func decodeJyUrl()  {
-	test := "ALPY1wIdDIvUC06AlFzcE8zJicCCj1mdmB0Py8nLz0wc35zcFpUCgQ%3D"
+	test := "ABCY1wAdTwFLy0vNFV5cAdaJSc%2FCj1mcVVhKCg7LS43a2BzdCNUCXg%3D"
 	var Decode  = qu.CommonDecodeArticle("content", test)
 	log.Debug(Decode[0])
 	return
@@ -52,49 +52,16 @@ func encodeJyUrl()  {
 	var Encode  = fmt.Sprintf(Url, qu.CommonEncodeArticle("content", "5fd2f37bf0f9d716c171c021"))
 	log.Debug(Encode)
 }
-
-
-func aaaaaa(){
-	sess := save_mgo.GetMgoConn()
-	defer save_mgo.DestoryMongoConn(sess)
-	q,total := map[string]interface{}{},0
-
-	it := sess.DB("qfw").C("zktest_cgyx_data").Find(&q).Sort("_id").Iter()
-	for tmp := make(map[string]interface{}); it.Next(&tmp);total++{
-		if total%1000==0 {
-			log.Debug("curent index ",total,tmp["_id"])
-		}
-		tttid := qu.ObjToString(tmp["id"])
-		data := save_mgo.FindById("bidding",tttid)
-
-		if data==nil ||len(data)<2 {
-			log.Debug("异常",tttid)
-		}else {
-			updateid := BsonTOStringId(tmp["_id"])
-			save_mgo.UpdateById("zktest_cgyx_data",updateid, map[string]interface{}{
-				"$set": map[string]interface{}{
-					"purchasinglist":data["purchasinglist"],
-				},
-			})
-		}
-		tmp = make(map[string]interface{})
-	}
-
-	log.Debug("is set bdw over ",total)
-}
-
-
 //copy数据 2022-02-19之后
 func repairCopyNextData() {
-	//暂时截止 2-23日 621508800000000000000000
-	log.Debug("620fd0900000000000000000~621508800000000000000000")
+	//621508800000000000000000 2-23 之后的数据
 	log.Debug("copy 之后的数据")
 	sess := save_mgo.GetMgoConn()
 	defer save_mgo.DestoryMongoConn(sess)
 	q := map[string]interface{}{
 		"_id": map[string]interface{}{
-			"$gt": StringTOBsonId("620fd0900000000000000000"),
-			"$lte": StringTOBsonId("621508800000000000000000"),
+			"$gt": StringTOBsonId("121508800000000000000000"),
+			"$lte": StringTOBsonId("921c1c8032770a446e8b2732"),
 		},
 	}
 	log.Debug("查询条件~",q)
@@ -102,7 +69,7 @@ func repairCopyNextData() {
 	//适当增加线程数
 	repair_pool := make(chan bool, 5)
 	repair_wg := &sync.WaitGroup{}
-	it := sess.DB("qfw").C("result_20210108").Find(&q).Sort("_id").Iter()
+	it := sess.DB("qfw").C("result_20210109").Find(&q).Sort("_id").Iter()
 	for tmp := make(map[string]interface{}); it.Next(&tmp);total++{
 		if total%10000==0 {
 			log.Debug("curent index ",total,tmp["_id"])
@@ -114,7 +81,7 @@ func repairCopyNextData() {
 				<-repair_pool
 				repair_wg.Done()
 			}()
-			save_mgo.Save("result_20220218",tmp)
+			save_mgo.Save("result_20220219",tmp)
 
 
 		}(tmp)
@@ -125,6 +92,37 @@ func repairCopyNextData() {
 
 }
 
+
+//补-标的物数据
+func consummatePurlist(){
+	sess := save_mgo.GetMgoConn()
+	defer save_mgo.DestoryMongoConn(sess)
+	q,total := map[string]interface{}{},0
+
+	it := sess.DB("qfw").C("zktest_cgyx_data").Find(&q).Sort("_id").Iter()
+	for tmp := make(map[string]interface{}); it.Next(&tmp);total++{
+		if total%1000==0 {
+			log.Debug("curent index ",total,tmp["_id"])
+		}
+		tttid := qu.ObjToString(tmp["id"])
+		data := save_mgo.FindById("bidding",tttid)
+
+		if data==nil ||len(data)<2 {
+			log.Debug("异常",tttid)
+		}else {
+			updateid := BsonTOStringId(tmp["_id"])
+			save_mgo.UpdateById("zktest_cgyx_data",updateid, map[string]interface{}{
+				"$set": map[string]interface{}{
+					"purchasinglist":data["purchasinglist"],
+				},
+			})
+		}
+		tmp = make(map[string]interface{})
+	}
+
+	log.Debug("is set bdw over ",total)
+}
+
 //敏感词修复标记
 func repairSensitiveTag() {