人工干预，修改线上go任务文件

product/suggest-task.go

@@ -44,12 +44,18 @@ type Word struct {
 	WordRank       float64 `json:"wordRank"`
 	KeywordVersion string  `json:"keywordVersion"`
 }
+const BRAND_TYPE = "1"
+const CATEGORY_TYPE = "2"
+const MANUAL_TYPE = "3"
 
 var wordMap sync.Map
 var brandMap = make(map[string]int)
 var categoryMap = make(map[string]int)
 var manualMap = make(map[string]int32)
 var sensitiveMap = make(map[string]bool)
+var usedMap sync.Map
+var now = time.Now()
+var dateStr = fmt.Sprintf("%d-%d-%d",now.Year(),now.Month(),now.Day())
 
 var t2s, _ = gocc.New("t2s")
 var prefixFilterArr = []string{"https://", "http://", "dg", "d & g", "dolce&gabbana",
@@ -88,16 +94,56 @@ func main() {
 			go queryIndex(i*10000, db, bulkProcessor, &wg, wordMap)
 		}
 
-		err = bulkProcessor.Flush()
-		if err != nil { log.Print(err.Error()) }
+
 	}
 
 
 	wg.Wait()
+	fmt.Println("all thread has read maps")
+	checkUnusedData(bulkProcessor)
+
+	err = bulkProcessor.Flush()
+	if err != nil { log.Print(err.Error()) }
 	defer db.Close()
 	fmt.Printf("Cost %d ms\n", time.Since(startTime).Nanoseconds()/1e6)
 }
 
+func checkUnusedData(bulkProcessor *elastic.BulkProcessor) {
+	for brand := range brandMap {
+		if _, exist := usedMap.Load(brand + BRAND_TYPE); !exist {
+			addWord(brand,bulkProcessor,BRAND_TYPE)
+		}
+	}
+	for category := range categoryMap {
+		if _, exist := usedMap.Load(category + CATEGORY_TYPE); !exist {
+			addWord(category,bulkProcessor,CATEGORY_TYPE)
+		}
+	}
+	for manual := range manualMap {
+		if _, exist := usedMap.Load(manual + MANUAL_TYPE); !exist {
+			addWord(manual,bulkProcessor,MANUAL_TYPE)
+		}
+	}
+}
+
+func addWord(keyword string, processor *elastic.BulkProcessor, wordType string) {
+	var w = new(Word)
+	switch wordType {
+	case BRAND_TYPE: w.IsBrand = true
+	case CATEGORY_TYPE: w.IsCategory = true
+	case MANUAL_TYPE: { w.IsManual = true; w.ManualValue = manualMap[keyword]}
+	}
+	w.Keyword = keyword
+	w.KeywordVersion = dateStr
+	processWord(w)
+	if !isFilterWord(w) {
+		id := fmt.Sprintf("%x", md5.Sum([]byte(w.Keyword)))
+		req := elastic.NewBulkIndexRequest().
+			Index("search_suggest_index").
+			Type("search_suggest_type").Id(id).Doc(w)
+		processor.Add(req)
+	}
+}
 func loadErpDB() {
 	db, err := sql.Open("mysql", "so_Erp_R:5RgzudyyFlApTmve@tcp(192.168.50.40:3306)/secooErpDB")
 	if err != nil { log.Print(err.Error()) }
@@ -193,7 +239,7 @@ func queryInfo(db *sql.DB) []int  {
 func queryIndex(idFlag int, db *sql.DB, bulkProcessor *elastic.BulkProcessor, wg *sync.WaitGroup, m sync.Map)  {
 	wg.Add(1)
 	fmt.Printf("%d doing.\n", idFlag)
-	var sqlStr = fmt.Sprintf("select * from app_search_keyword_year_week_p_day where id >= %d limit 10000", idFlag)
+	var sqlStr = fmt.Sprintf("select * from app_search_keyword_year_week_p_day where id >= %d and id < %d", idFlag, idFlag + 10000)
 	results, err := db.Query(sqlStr)
 	if err != nil { log.Print(err.Error()) }
 
@@ -276,15 +322,18 @@ func processWord(w *Word) {
 
 	if _, isExist := brandMap[w.Keyword]; isExist {
 		w.IsBrand = true
+		usedMap.Store(w.Keyword + BRAND_TYPE, true)
 	}
 
 	if _, isExist := manualMap[w.Keyword]; isExist {
 		w.IsManual = true
 		w.ManualValue = manualMap[w.Keyword]
+		usedMap.Store(w.Keyword + MANUAL_TYPE,true)
 	}
 
 	if _, isExist := categoryMap[w.Keyword]; isExist {
 		w.IsCategory = true
+		usedMap.Store(w.Keyword + CATEGORY_TYPE,  true)
 	}
 
 	if _, isExist := sensitiveMap[w.Keyword]; isExist {
@@ -419,4 +468,4 @@ func calculateCountFactor(count int32, rank int32) float64 {
 	//根据搜索次数，转换为热度因子
 	count = count * rank + 10
 	return math.Log10(math.Sqrt(float64(count)))
-}
+}
\ No newline at end of file