针对 gtid lteid 传参，string 转换

createEsIndex/bidding_es.go

@@ -44,6 +44,9 @@ func biddingTask(mapInfo map[string]interface{}) {
 				"$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
 			},
 		}
+	} else {
+		//针对gte/lte,单独转换
+		q = convertToMongoID(q)
 	}
 
 	ch := make(chan bool, 10)
@@ -75,10 +78,13 @@ func biddingTask(mapInfo map[string]interface{}) {
 				return
 			}
 			//只针对增量数据处理；全量数据 需要用extracttype字段判断
+			//7:	重复数据
+			//8:	不重复
 			if util.IntAll(tmp["dataprocess"]) != 8 {
 				return
 			}
 			//// 增量数据使用上面判断；全量数据使用下面配置
+			//-1:重复 ,1：不重复 ,0：入库 9：分类
 			//if util.IntAll(tmp["extracttype"]) != 1 {
 			//	return
 			//}
@@ -149,6 +155,9 @@ func biddingAllTask(mapInfo map[string]interface{}) {
 				"$lte": mongodb.StringTOBsonId(mapInfo["lteid"].(string)),
 			},
 		}
+	} else {
+		//针对gte/lte,单独转换
+		q = convertToMongoID(q)
 	}
 
 	ch := make(chan bool, 20)

createEsIndex/project_es.go

@@ -39,6 +39,9 @@ func projectTask(data []byte, mapInfo map[string]interface{}) {
 				q["_id"] = tmpQ
 			}
 		}
+		if q["_id"] != nil {
+			q = convertToMongoID(q)
+		}
 	}
 
 	conn := MgoP.GetMgoConn()

createEsIndex/utils.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"app.yhyue.com/data_processing/common_utils/mongodb"
+)
+
+//convertToMongoID convertToMongoID
+func convertToMongoID(query map[string]interface{}) map[string]interface{} {
+	result := make(map[string]interface{})
+	if query == nil {
+		return result
+	}
+
+	idMap := query["_id"].(map[string]interface{})
+	if idMap != nil {
+		tmpQ := map[string]interface{}{}
+		for c, id := range idMap {
+			if idStr, ok := id.(string); ok && id != "" {
+				tmpQ[c] = mongodb.StringTOBsonId(idStr)
+			}
+		}
+		result["_id"] = tmpQ
+	}
+
+	return result
+}