Es存储用户阅读记录优化

controllers/article.go

@@ -579,6 +579,7 @@ func (this *ArticleCommonController) Detail() {
 			}
 		}
 	}
+	detail.SellerAndMobile = "" //业务需要强制处理为空
 	resp.HasPermission = 1
 	detail.Abstract, _ = services.GetReportContentTextSub(detail.Abstract)
 	detail.SellerList = sellerList

services/elastic.go

@@ -102,14 +102,17 @@ func CreateIndexNameArticleHistory() {
 }
 
 //func UpdateWxUserLabel(cont context.Context) (err error) {
-func AddAllArticleHistory() (err error) {
+func AddAllArticleHistory(cont context.Context) (err error) {
 	defer func() {
 		if err != nil {
 			go utils.SendAlarmMsg("同步阅读记录到es失败;Err:"+err.Error(), 2)
-			go utils.SendEmail("更新用户标签失败"+"【"+utils.APPNAME+"】"+time.Now().Format(utils.FormatDateTime), ";Err:"+err.Error(), utils.EmailSendToUsers)
-			utils.FileLog.Info("更新用户标签失败,Err:%s", err.Error())
+			go utils.SendEmail("同步阅读记录到es失败"+"【"+utils.APPNAME+"】"+time.Now().Format(utils.FormatDateTime), ";Err:"+err.Error(), utils.EmailSendToUsers)
+			utils.FileLog.Info("同步阅读记录到es失败,Err:%s", err.Error())
 		}
 	}()
+
+	var updateUserIds string //更改过的用户ID
+	userIdMap := make(map[int]int)
 	condition := `  AND r.create_time < '` + time.Now().Format(utils.FormatDate) + `' 	AND r.company_id IN (
 					SELECT a.company_id 
 					FROM  company AS a INNER JOIN company_product AS b ON a.company_id = b.company_id 
@@ -126,7 +129,7 @@ func AddAllArticleHistory() (err error) {
 		return err
 	}
 	//处理前一天新增的数据
-	//condition += `  AND r.create_time >='` + time.Now().AddDate(0, 0, -1).Format(utils.FormatDate) + `'`
+	condition += `  AND r.create_time >='` + time.Now().AddDate(0, 0, -1).Format(utils.FormatDate) + `'`
 	total, err := models.GetCygxArticleHistoryCountByCompany(condition)
 	//fmt.Println(total)
 	//return
@@ -148,76 +151,113 @@ func AddAllArticleHistory() (err error) {
 				fmt.Println("EsAddOrEditData Err:", err.Error())
 				return err
 			}
+			if _, ok := userIdMap[v.UserId]; !ok {
+				updateUserIds += strconv.Itoa(v.UserId) + ","
+				userIdMap[v.UserId] = v.UserId
+			}
 		}
 	}
 	//处理前一天新增的数据 end
 
 	//处理前一天被移动的用户
-	//{
-	//	startDate := time.Now().AddDate(0, 0, -1).Format(utils.FormatDate)
-	//	endDate := time.Now().Format(utils.FormatDate)
-	//	listUpdateUser, err := models.GetWxUserOpLogList(startDate, endDate)
-	//	if err != nil && err.Error() != utils.ErrNoRow() {
-	//		return err
-	//	}
-	//	var mobiles string
-	//	if len(listUpdateUser) > 0 {
-	//		for _, v := range listUpdateUser {
-	//			mobiles += "'" + v.Mobile + "',"
-	//		}
-	//	}
-	//	mobiles = strings.TrimRight(mobiles, ",")
-	//	if mobiles != "" {
-	//		condition = ` AND r.mobile IN (` + mobiles + `)`
-	//		//修改用户的阅读记录(es 自动判断，如果有他会修改数据)
-	//		listUpdatePv, err := models.GetCygxArticleHistoryRecordByCompanyList(condition, 0, 0)
-	//		if err != nil {
-	//			fmt.Println("GetArticleHistoryRecordAllByMobileList ，Err" + err.Error())
-	//			return err
-	//		}
-	//
-	//		for _, v := range listUpdatePv {
-	//			err := EsAddArticleHistoryData(v)
-	//			if err != nil {
-	//				fmt.Println("EsAddOrEditData Err:", err.Error())
-	//				return err
-	//			}
-	//		}
-	//	}
-	//}
+
+	startDate := time.Now().AddDate(0, 0, -1).Format(utils.FormatDate)
+	endDate := time.Now().Format(utils.FormatDate)
+	var mobiles string
+	listUpdateUser, err := models.GetWxUserOpLogList(startDate, endDate)
+	if err != nil && err.Error() != utils.ErrNoRow() {
+		return err
+	}
+	if len(listUpdateUser) > 0 {
+		for _, v := range listUpdateUser {
+			mobiles += "'" + v.Mobile + "',"
+		}
+	}
+	mobiles = strings.TrimRight(mobiles, ",")
+	if mobiles != "" {
+		condition = ` AND r.mobile IN (` + mobiles + `)`
+		//修改用户的阅读记录(es 自动判断，如果有他会修改数据)
+		listUpdatePv, err := models.GetCygxArticleHistoryRecordByCompanyList(condition, 0, 0)
+		if err != nil {
+			fmt.Println("GetArticleHistoryRecordAllByMobileList ，Err" + err.Error())
+			return err
+		}
+
+		for _, v := range listUpdatePv {
+			err := EsAddArticleHistoryData(v)
+			if err != nil {
+				fmt.Println("EsAddOrEditData Err:", err.Error())
+				return err
+			}
+		}
+	}
+
 	//处理前一天被移动的用户 end
 
 	//处理前一天被删除的用户
-	//{
-	//	listDeleteUser, err := models.GetWxUserOpLogDeleteList(startDate, endDate)
-	//	if err != nil && err.Error() != utils.ErrNoRow() {
-	//		return err
-	//	}
-	//	mobiles = ""
-	//	if len(listUpdateUser) > 0 {
-	//		for _, v := range listDeleteUser {
-	//			mobiles += "'" + v.Mobile + "',"
-	//		}
-	//	}
-	//	mobiles = strings.TrimRight(mobiles, ",")
-	//	if mobiles != "" {
-	//		condition = ` AND r.mobile IN (` + mobiles + `)`
-	//		listDeletePv, err := models.GetCygxArticleHistoryRecordByCompanyList(condition, 0, 0)
-	//		if err != nil {
-	//			fmt.Println("GetArticleHistoryRecordAllByMobileList ，Err" + err.Error())
-	//			return err
-	//		}
-	//		//fmt.Println("Es 删除")
-	//		for _, v := range listDeletePv {
-	//			err := EsDeleteData(utils.IndexNameArticleHistory, strconv.Itoa(v.Id))
-	//			if err != nil {
-	//				fmt.Println("EsAddOrEditData Err:", err.Error())
-	//				return err
-	//			}
-	//		}
-	//	}
-	//}
+	{
+		listDeleteUser, err := models.GetWxUserOpLogDeleteList(startDate, endDate)
+		if err != nil && err.Error() != utils.ErrNoRow() {
+			return err
+		}
+		mobiles = ""
+		if len(listUpdateUser) > 0 {
+			for _, v := range listDeleteUser {
+				mobiles += "'" + v.Mobile + "',"
+			}
+		}
+		mobiles = strings.TrimRight(mobiles, ",")
+		if mobiles != "" {
+			condition = ` AND r.mobile IN (` + mobiles + `)`
+			listDeletePv, err := models.GetCygxArticleHistoryRecordByCompanyList(condition, 0, 0)
+			if err != nil {
+				fmt.Println("GetArticleHistoryRecordAllByMobileList ，Err" + err.Error())
+				return err
+			}
+			//fmt.Println("Es 删除")
+			for _, v := range listDeletePv {
+				err := EsDeleteData(utils.IndexNameArticleHistory, strconv.Itoa(v.Id))
+				if err != nil {
+					fmt.Println("EsAddOrEditData Err:", err.Error())
+					return err
+				}
+			}
+		}
+	}
 	//处理前一天被删除的用户 end
+
+	//处理新增的阅读记录的用户阅读数量、机构阅读数量(暂未找到批量修改的方法，后期优化处理 2022.7.11)
+	updateUserIds = strings.TrimRight(updateUserIds, ",")
+	if updateUserIds != "" {
+		condition = `  AND r.create_time >='` + time.Now().AddDate(0, 0, -1).Format(utils.FormatDate) + `' AND r.user_id IN (` + updateUserIds + `)`
+		total, err := models.GetCygxArticleHistoryCountByCompany(condition)
+		//fmt.Println(total)
+		//return
+		fmt.Println(total)
+		if err != nil {
+			fmt.Println("GetCygxArticleHistoryCountByCompany Err:", err.Error())
+			return err
+		}
+		for i := 0; i <= total/1000; i++ {
+			allList, err := models.GetCygxArticleHistoryRecordByCompanyList(condition, 1000*i, 1000)
+			if err != nil {
+				fmt.Println("GetCygxArticleHistoryRecordByCompanyList Err:", err.Error())
+				return err
+			}
+			for k, v := range allList {
+				fmt.Println(v.Id, "___", k)
+				err := EsAddArticleHistoryData(v)
+				if err != nil {
+					fmt.Println("EsAddOrEditData Err:", err.Error())
+					return err
+				}
+				if _, ok := userIdMap[v.UserId]; !ok {
+					updateUserIds += strconv.Itoa(v.UserId) + ","
+					userIdMap[v.UserId] = v.UserId
+				}
+			}
+		}
+	}
 	return
 }
 

services/task.go

@@ -69,6 +69,9 @@ func Task() {
 		updateWxUserLabel := task.NewTask("updateWxUserLabel", "0 01 0 * * *", UpdateWxUserLabel) //更新用户的标签
 		task.AddTask("updateWxUserLabel", updateWxUserLabel)
 
+		addAllArticleHistory := task.NewTask("addAllArticleHistory", "0 30 2 * * *", AddAllArticleHistory) //把前一天的用户阅读记录同步到ES
+		task.AddTask("addAllArticleHistory", addAllArticleHistory)
+
 		getCeLueArticlePv := task.NewTask("getCeLueArticlePv", "0 */10 * * * *", GetCeLueArticlePv) //通过三方接口获取策略平台上的阅读记录
 		task.AddTask("getCeLueArticlePv", getCeLueArticlePv)
 	}
@@ -77,14 +80,14 @@ func Task() {
 		task.AddTask("getArticleListByApi", getArticleListByApi)
 	}
 	//CreateIndexNameArticleHistory()
-	AddAllArticleHistory()
+	//AddAllArticleHistory()
 	//DoCompany()
 	//ActivityAttendanceDetail()
 	//SynchronizationArthistory()//同步原有的阅读记录
 	//GetAddpArticle() //同步日度点评数据
 	//SendEmailAllUserWithRAI()     //发送当前所有权益用户到沈大爷的邮箱
 	//SendEmailAllUserWithCompany() //发送这些公司下的用户到沈大爷的邮箱
-	//task.StartTask()
+	task.StartTask()
 	//StatisticalReport()//导出报表
 	fmt.Println("end")
 }