eta_mini_crm/services/task.go

package services

import (
	"eta/eta_mini_crm/models"
	"eta/eta_mini_crm/services/elastic"
	"eta/eta_mini_crm/utils"
	"fmt"
	"time"
)

func InitTask() {
	fmt.Println("start task")
	CreateIndex()
	InitReportPushStatus()
	fmt.Println("end task!")
}

func CreateIndex() {
	var mappingJson = `{
  "mappings": {
    "properties": {
      "ReportPdfId": {"type": "integer"},
      "PdfUrl": {"type": "text"},
      "PdfName": {"type": "text"},
      "Title": {"type": "text"},
      "Author": {"type": "text"},
      "Abstract": {"type": "text"},
      "ClassifyIdFirst": {"type": "integer"},
      "ClassifyNameFirst": {"type": "text"},
      "ClassifyIdSecond": {"type": "integer"},
      "ClassifyNameSecond": {"type": "text"},
      "Stage": {"type": "integer"},
      "PublishTime": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "ModifyTime": {"type": "date"},
      "Pv": {"type": "integer"},
      "Uv": {"type": "integer"},
      "SysUserId": {"type": "integer"},
      "SysRealName": {"type": "text"},
      "State": {"type": "integer"}
	}
   }
}`
	err := elastic.EsCreateIndex(utils.MINI_REPORT_INDEX_NAME, mappingJson)
	if err != nil {
		fmt.Println(err)
	}
}

func InitReportPushStatus() {
	for {
		maxId, err := models.GetMaxSyncIdReportPush(1)
		fmt.Println("同步研报开始, maxId:", maxId)
		if err != nil {
			fmt.Println("同步研报失败, Err:", err)
		}
		reportList, err := models.GetBatchReport(maxId, 100)
		if err != nil {
			fmt.Println("同步研报失败, Err:", err)
		}
		var reportIds []int
		for _, v := range reportList {
			reportIds = append(reportIds, v.Id)
		}
		reportPushList, err := models.GetReportPushStatusByReportIds(reportIds)
		if err != nil {
			return
		}
		reportPushMap := make(map[int]struct{})
		for _, v := range reportPushList {
			reportPushMap[v.ReportId] = struct{}{}
		}
		var insertReportPushList []*models.ReportPushStatus
		for _, v := range reportList {
			if _, ok := reportPushMap[v.Id]; !ok {
				insertReportPushList = append(insertReportPushList, &models.ReportPushStatus{
					ReportId:           v.Id,
					State:              0,
					Title:              v.Title,
					Abstract:           v.Abstract,
					Stage:              v.Stage,
					ClassifyIdFirst:    v.ClassifyIdFirst,
					ClassifyNameFirst:  v.ClassifyNameFirst,
					ClassifyIdSecond:   v.ClassifyIdSecond,
					ClassifyNameSecond: v.ClassifyNameSecond,
					ClassifyIdThird:    v.ClassifyIdThird,
					ClassifyNameThird:  v.ClassifyNameThird,
					Author:             v.Author,
					ReportType:         1,
					PublishTime:        v.PublishTime,
					CreateTime:         time.Now(),
					ModifyTime:         time.Now(),
				})
			}
		}
		err = models.BatchAddReportPushStatus(insertReportPushList)
		if err != nil {
			return
		}
		if len(reportList) != 100 {
			fmt.Println("同步研报完成")
			return
		}
	}
}