增量

customerdata/src/task.go

@@ -27,8 +27,8 @@ import (
 )
 
 func TimeTask() {
-	// StartTask()
-	GetCustomerDatas()
+	StartTask()
+	// GetCustomerDatas()
 	c := cron.New()
 	//cronstr := "0 */" + fmt.Sprint(TaskTime) + " * * * ?"
 	cronstr := "0 5 */" + fmt.Sprint(TaskTime) + " * * ?" //每TaskTime小时执行一次
@@ -105,7 +105,7 @@ func GetCustomerData() {
 		cus.RemoveRepeatData()    //数据去重
 		cus.AssembelAndSaveData() //组装、保存数据
 	}
-	Sysconfig.LatestId = LatestId
+	// Sysconfig.LatestId = LatestId
 	Sysconfig.LatestTime = endTime
 	qu.WriteSysConfig(Sysconfig)
 	log.Println("定时任务结束-endId-Sysconfig.LatestTime ", endTime)
@@ -168,7 +168,7 @@ func GetCustomerDatas() {
 		cus.RemoveRepeatData()    //数据去重
 		cus.AssembelAndSaveData() //组装、保存数据
 	}
-	Sysconfig.LatestId = LatestId
+	// Sysconfig.LatestId = LatestId
 	Sysconfig.LatestTimes = endTime
 	qu.WriteSysConfig(Sysconfig)
 	log.Println("全量数据任务结束-endId-Sysconfig.LatestTimes ", endTime)
@@ -386,7 +386,7 @@ func (c *Customer) EsConGetDataV7(stype string, esCon *esv.EsV7) {
 			escount := int64(0)
 			qstr := sr.EsQuery
 			if stype == "all" {
-				esindex = "bidding_all"
+				esindex = "biddingall"
 				escount = Es2.Count(esindex, Itype, qstr)
 			} else {
 				escount = Es.Count(esindex, Itype, qstr)
@@ -447,6 +447,7 @@ func (c *Customer) EsConGetDataV7(stype string, esCon *esv.EsV7) {
 								}
 							}
 							id := qu.ObjToString(tmp["id"])
+							log.Println("当前数据 ", id)
 							//亚信
 							if CheckBidOpenAppidMap[c.AppId] {
 								if tmp["bidopentime"] != nil {

customerdata/src/util.go

@@ -216,27 +216,9 @@ func GetNotkeyAndKeyAddWord(list primitive.A, fieldMap map[string]interface{}, h
 // 根据时间获取起始和终止ID范围
 func GetIdRange() (bson.M, bson.M, bool, int64) {
 	defer qu.Catch()
-	// now := time.Now().Unix()
-	// for { //当前时间一直向前推半小时，直到取到数据
-	// now = now - 600 //10分钟前
-	// endTime := time.Unix(now, 0)
-	// endId := bson.NewObjectIdWithTime(endTime).Hex()
-	// if (now1 - now) > 3*600 {
-	// 	return bson.M{"range": bson.M{
-	// 		"id": bson.M{
-	// 			"lte": endId,
-	// 			"gt":  LatestId,
-	// 		},
-	// 	}}, true
-	// }
 	idQuery, endTime := "", int64(0)
-	esquery := `{"query":{"bool":{"must":{"range":{"pici":{"gt":"%d"}}}}},"_source":["id","pici"],"sort":{"pici":"desc"},"from":0,"size":1}`
-	if LatestId == "" {
-		idQuery = strings.Replace(fmt.Sprintf(esquery, LatestTime), `"gt"`, `"gte"`, -1)
-	} else {
-		esquerys := `{"query":{"bool":{"must":{"range":{"id":{"gt":"%s"}}}}},"_source":["id","pici"],"sort":{"pici":"desc"},"from":0,"size":1}`
-		idQuery = fmt.Sprintf(esquerys, LatestId)
-	}
+	esquery := `{"query":{"bool":{"must":{"range":{"pici":{"gt":"%d"}}}}},"_source":["pici"],"sort":{"pici":"desc"},"from":0,"size":1}`
+	idQuery = fmt.Sprintf(esquery, LatestTime)
 	resId := Es.Get(Index, Itype, idQuery)
 	if resId != nil && *resId != nil && len(*resId) == 1 {
 		endTime = qu.Int64All((*resId)[0]["pici"]) - 600
@@ -248,57 +230,20 @@ func GetIdRange() (bson.M, bson.M, bool, int64) {
 		"range": bson.M{
 			"pici": bson.M{
 				"lte": endTime,
-				"gt":  LatestTime,
-			},
-		},
-	}
-	tmpRange2 := bson.M{
-		"range": bson.M{
-			"pici": bson.M{
-				"lte": endTime,
-			},
-		},
-	}
-	tmpRange3 := bson.M{
-		"range": bson.M{
-			"id": bson.M{
-				"gt": LatestId,
+				"gt":  LatestTime - 7200,
 			},
 		},
 	}
 	LatestTime = endTime
-	if LatestId == "" {
-		return tmpRange1, bson.M{}, true, endTime
-	} else {
-		LatestId = ""
-		return tmpRange2, tmpRange3, true, endTime
-	}
+	return tmpRange1, bson.M{}, true, endTime
 }
 
 func GetIdRanges() (bson.M, bson.M, bool, int64) {
 	defer qu.Catch()
-	// now := time.Now().Unix()
-	// for { //当前时间一直向前推半小时，直到取到数据
-	// now = now - 600 //10分钟前
-	// endTime := time.Unix(now, 0)
-	// endId := bson.NewObjectIdWithTime(endTime).Hex()
-	// if (now1 - now) > 3*600 {
-	// 	return bson.M{"range": bson.M{
-	// 		"id": bson.M{
-	// 			"lte": endId,
-	// 			"gt":  LatestId,
-	// 		},
-	// 	}}, true
-	// }
 	idQuery, endTime := "", int64(0)
-	esquery := `{"query":{"bool":{"must":{"range":{"pici":{"gt":"%d"}}}}},"_source":["id","pici"],"sort":{"pici":"desc"},"from":0,"size":1}`
-	if LatestId == "" {
-		idQuery = strings.Replace(fmt.Sprintf(esquery, LatestTimes), `"gt"`, `"gte"`, -1)
-	} else {
-		esquerys := `{"query":{"bool":{"must":{"range":{"id":{"gt":"%s"}}}}},"_source":["id","pici"],"sort":{"pici":"desc"},"from":0,"size":1}`
-		idQuery = fmt.Sprintf(esquerys, LatestId)
-	}
-	resId := Es.Get(Index, Itype, idQuery)
+	esquery := `{"query":{"bool":{"must":{"range":{"pici":{"gt":"%d"}}}}},"_source":["pici"],"sort":{"pici":"desc"},"from":0,"size":1}`
+	idQuery = fmt.Sprintf(esquery, LatestTimes)
+	resId := Es2.Get("biddingall", "", idQuery)
 	if resId != nil && *resId != nil && len(*resId) == 1 {
 		endTime = qu.Int64All((*resId)[0]["pici"]) - 600
 	} else {
@@ -313,27 +258,8 @@ func GetIdRanges() (bson.M, bson.M, bool, int64) {
 			},
 		},
 	}
-	tmpRange2 := bson.M{
-		"range": bson.M{
-			"pici": bson.M{
-				"lte": endTime,
-			},
-		},
-	}
-	tmpRange3 := bson.M{
-		"range": bson.M{
-			"id": bson.M{
-				"gt": LatestId,
-			},
-		},
-	}
-	LatestTimes = endTime
-	if LatestId == "" {
-		return tmpRange1, bson.M{}, true, endTime
-	} else {
-		LatestId = ""
-		return tmpRange2, tmpRange3, true, endTime
-	}
+	LatestTime = endTime
+	return tmpRange1, bson.M{}, true, endTime
 }
 
 // 处理文本
@@ -580,9 +506,11 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, a
 	}
 	id := qu.ObjToString(tmp["id"])
 	ruleid := qu.ObjToString(tmp["ruleid"])
-	redis.Put("datag", appid+"_"+id, 1, 3*24*60*60)
+	redis.Put("datag", appid+"_"+id, 1, 5*24*60*60)
 	tmp["createtime"] = time.Now().Unix()
+	title := qu.ObjToString(tmp["title"])
 	subType := qu.ObjToString(tmp["subtype"])
+	buyer := qu.ObjToString(tmp["buyer"])
 	subTypeStr := "拟建,采购意向,预告,预审,预审结果,论证意见,需求公示"
 	if strings.Contains(subTypeStr, subType) {
 		tmp["jybxhref"] = `https://www.jianyu360.cn/article/bdprivate/` + qu.CommonEncodeArticle("bdprivate", id) + `.html`
@@ -597,9 +525,11 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, a
 		tmp["bidopentime"] = qu.Int64All(bidopentime)
 	}
 	//项目id去重
-	projectId, zbtime, yitime := GetProjectId(id)
+	projectId, buyers, zbtime, yitime := GetProjectId(id)
 	if projectId != "" {
-		projectIds := strings.Split(projectId, ",")
+		if buyer == "" && buyers != "" {
+			tmp["buyer"] = buyers
+		}
 		tmp["projectId"] = projectId
 		if zbtime > 0 {
 			tmp["zbtime"] = zbtime
@@ -609,12 +539,10 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, a
 		}
 		if projectAppidMap[appid] {
 			isOk := true
-			for _, v := range projectIds {
-				if _, ok := projectIdMap.Load(v); ok {
-					isOk = false
-				} else {
-					projectIdMap.Store(v, true)
-				}
+			if _, ok := projectIdMap.Load(projectId); ok {
+				isOk = false
+			} else {
+				projectIdMap.Store(projectId, true)
 			}
 			if !isOk {
 				MgoSave.Save(SaveCollProject, tmp)
@@ -807,8 +735,14 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, a
 		}
 		if appid == "jyFhJXQAYDAANZTEVPFD1U" {
 			if ruleid == "67e64794756e2b3b3b6d5062" {
-				info["id"] = id
-				tmp = info
+				tmp["agencyperson"] = info["agencyperson"]
+				tmp["agencytel"] = info["agencytel"]
+				tmp["buyerperson"] = info["buyerperson"]
+				tmp["buyertel"] = info["buyertel"]
+				tmp["bidopentime"] = info["bidopentime"]
+				tmp["winnerperson"] = info["winnerperson"]
+				tmp["winnertel"] = info["winnertel"]
+				tmp["bidendtime"] = info["bidendtime"]
 			}
 			site = strings.ReplaceAll(site, "军队", "JD")
 			site = strings.ReplaceAll(site, "兵团", "BT")
@@ -818,11 +752,26 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, a
 			site = strings.ReplaceAll(site, "中核", "ZH")
 			tmp["site"] = site
 		}
+		repeat_id := qu.ObjToString(info["repeat_id"])
+		if repeat_id != "" {
+			isExists, _ := redis.Exists("datag", appid+"_"+repeat_id)
+			if isExists {
+				log.Println("优选信息id重复 ", repeat_id, id)
+				return false
+			} else {
+				udata := MgoSave.FindOne("usermail", map[string]interface{}{"appid": appid, "id": repeat_id})
+				if udata != nil && len(udata) > 0 {
+					log.Println("优选信息id重复1 ", repeat_id, id)
+					return false
+				}
+			}
+		}
 	}
 	if FilterOwnerMap[appid] {
 		SearchOwnerInfo(tmp)
 		isConOk := false
 		owner := qu.ObjToString(tmp["owner"])
+		owner_company_phone := qu.ObjToString(tmp["owner_company_phone"])
 		if owner != "" {
 			contactData := MgoEnps.FindOne("buyer_enterprise", bson.M{"buyer_name": owner})
 			if contactData != nil && len(contactData) > 0 {
@@ -856,13 +805,24 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, a
 				}
 			}
 		} else {
-			return false
+			if appid == "jyHwBXQQoDCQBcTUdMJRJx" {
+				if !strings.Contains(title, "公司") {
+					return false
+				}
+			} else {
+				return false
+			}
 		}
 		if !isConOk {
 			if appid == "jyIypXQAEDAA9SQENJEAdY" {
 				return false
 			}
 		}
+		if appid == "jyGBhXQAAHAwNZQUhPPi98" {
+			if owner_company_phone == "" {
+				return false
+			}
+		}
 	}
 	//祥和不锈钢
 	detail := qu.ObjToString(tmp["detail"])
@@ -1351,31 +1311,29 @@ func SkipData(tmp map[string]interface{}) bool {
 	return false
 }
 
-func GetProjectId(id string) (string, int64, int64) {
+func GetProjectId(id string) (string, string, int64, int64) {
 	query := `{"query": {"bool": {"must": [{"term": {"ids": "%s"}}],"must_not": [],"should": []}}}`
 	querys := fmt.Sprintf(query, id)
-	projectId := ""
+	projectId, buyer := "", ""
 	zbtime, yitime := int64(0), int64(0)
-	projectIdArr := []string{}
 	data := Es.Get("projectset", "projectset", querys)
 	if data != nil && *data != nil && len(*data) > 0 {
-		for _, v := range *data {
-			projectIdArr = append(projectIdArr, qu.ObjToString(v["_id"]))
-			zbtime = qu.Int64All(v["zbtime"])
-			if listArr, ok := v["list"].([]interface{}); ok {
-				for _, l := range listArr {
-					if vs, oks := l.(map[string]interface{}); oks {
-						if qu.ObjToString(vs["subrtype"]) == "采购意向" {
-							yitime = qu.Int64All(vs["publishtime"])
-						}
+		datas := (*data)[0]
+		zbtime = qu.Int64All(datas["zbtime"])
+		buyer = qu.ObjToString(datas["buyer"])
+		if listArr, ok := datas["list"].([]interface{}); ok {
+			for _, l := range listArr {
+				if vs, oks := l.(map[string]interface{}); oks {
+					if qu.ObjToString(vs["subrtype"]) == "采购意向" {
+						yitime = qu.Int64All(vs["publishtime"])
 					}
 				}
 			}
 		}
-		projectId = strings.Join(projectIdArr, ",")
+		projectId = qu.ObjToString(datas["_id"])
 		log.Println("projectId", projectId)
 	}
-	return projectId, zbtime, yitime
+	return projectId, buyer, zbtime, yitime
 }
 
 func InitProjectId(appid string) {