Merge branch 'bzq/knowledge_add_viewpoint'

.gitignore

@@ -16,3 +16,4 @@ eta_task
 /eta_task.exe~
 /*.mod
 /test/
+/.vscode/

models/classify.go

@@ -3,15 +3,17 @@ package models
 import (
 	"eta_gn/eta_task/global"
 	"fmt"
-	"github.com/rdlucklib/rdluck_tools/paging"
 	"strings"
 	"time"
+
+	"github.com/rdlucklib/rdluck_tools/paging"
 )
 
 type Classify struct {
 	Id             int       `gorm:"column:id;primaryKey"` // `orm:"column(id);pk"`
 	ClassifyName   string    `description:"分类名称"`
 	Sort           int       `json:"-"`
+	Level          int       `gorm:"column:level"` //`description:"分类级别"`
 	ParentId       int       `description:"父级分类id"`
 	CreateTime     time.Time `description:"创建时间"`
 	ModifyTime     time.Time `description:"修改时间"`
@@ -273,3 +275,18 @@ func (m *Classify) GetItemsByCondition(condition string, pars []interface{}, fie
 	err = global.DmSQL["rddp"].Raw(sql, pars...).Find(&items).Error
 	return
 }
+
+func (m *Classify) GetItemById(id int) (item *Classify, err error) {
+	sql := fmt.Sprintf(`SELECT * FROM %s WHERE %s = ? `, m.TableName(), "id")
+	err = global.DmSQL["rddp"].Raw(sql, id).First(&item).Error
+	return
+}
+
+func (m *Classify) GetItemsByIds(ids []int) (items []*Classify, err error) {
+	if len(ids) == 0 {
+		return
+	}
+	sql := fmt.Sprintf(`SELECT * FROM %s WHERE %s IN (?) ORDER BY level ASC`, m.TableName(), "id")
+	err = global.DEFAULT_DmSQL.Raw(sql, ids).Find(&items).Error
+	return
+}

models/knowledge_resource.go

@@ -0,0 +1,209 @@
+package models
+
+import (
+	"eta_gn/eta_task/global"
+	"eta_gn/eta_task/utils"
+	"time"
+)
+
+const (
+	// 事件类型-事件库
+	KnowledgeResourceTypeEvent = iota
+	// 事件-政策库
+	KnowledgeResourceTypePolicy
+	// 事件-报告库
+	KnowledgeResourceTypeReport
+	// 事件类型-知识库
+	KnowledgeResourceTypeKnow
+	// 事件-观点库
+	KnowledgeResourceTypeOpinion
+)
+
+const (
+	// 未发布
+	KnowledgeResourceStateUnpublished = iota
+	// 已发布
+	KnowledgeResourceStatePublished
+	// 待审批
+	KnowledgeResourceStatePending
+	// 已驳回
+	KnowledgeResourceStateRejected
+	// 已通过
+	KnowledgeResourceStateApproved
+)
+
+type KnowledgeResourceFile struct {
+	KnowledgeResourceFileId int    `gorm:"column:knowledge_resource_file_id;primaryKey;autoIncrement"`
+	KnowledgeResourceId     int    `gorm:"column:knowledge_resource_id"`
+	FileUrl                 string `gorm:"column:file_url"`
+}
+
+func (k KnowledgeResourceFile) TableName() string {
+	return "knowledge_resource_file"
+}
+
+type KnowledgeResource struct {
+	KnowledgeResourceId int        `gorm:"column:knowledge_resource_id;;primaryKey;autoIncrement"`
+	ResourceType        int        `gorm:"column:resource_type;"`
+	ClassifyId          int        `gorm:"column:classify_id"`
+	Title               string     `gorm:"column:title;"`
+	CreateTime          time.Time  `gorm:"column:create_time" description:"创建时间"`
+	ModifyTime          time.Time  `gorm:"column:modify_time;autoUpdateTime" description:"修改时间"`
+	State               int        `gorm:"column:state" description:"0:未发布；1:已发布；2:待审批；3:已驳回；4:已通过"`
+	Content             string     `gorm:"column:content"`
+	ResourceCode        string     `gorm:"column:resource_code"`
+	AdminId             int        `gorm:"column:admin_id" description:"创建者账号"`
+	AdminRealName       string     `gorm:"column:admin_real_name" description:"创建者姓名"`
+	SourceFrom          string     `gorm:"column:source_from"`
+	TagId               int        `gorm:"column:tag_id;default:0;NOT NULL"`
+	StartTime           *time.Time `gorm:"column:start_time"`
+	EndTime             *time.Time `gorm:"column:end_time"`
+	IsFile              int        `gorm:"column:is_file;default:0;NOT NULL"`
+	FileUrl             string     `gorm:"column:file_url"`
+	OutId               int        `gorm:"column:out_id" description:"外部系统ID"`
+	IsDelete            int        `gorm:"column:is_delete;default:0;NOT NULL"`
+}
+
+func (k KnowledgeResource) TableName() string {
+	return "knowledge_resource"
+}
+
+func (k *KnowledgeResource) BatchCreate(item []*KnowledgeResource) (err error) {
+	return global.DmSQL["rddp"].CreateInBatches(item, utils.MultiAddNum).Error
+}
+
+func (k *KnowledgeResource) Create(fileUrl []string) (err error) {
+	tx := global.DmSQL["rddp"].Begin()
+	defer func() {
+		if err != nil {
+			tx.Rollback()
+		} else {
+			tx.Commit()
+		}
+	}()
+	err = tx.Create(k).Error
+	if len(fileUrl) > 0 {
+		addFile := make([]*KnowledgeResourceFile, 0, len(fileUrl))
+		for _, url := range fileUrl {
+			file := new(KnowledgeResourceFile)
+			file.KnowledgeResourceId = k.KnowledgeResourceId
+			file.FileUrl = url
+			addFile = append(addFile, file)
+		}
+		err = tx.CreateInBatches(addFile, utils.MultiAddNum).Error
+	}
+	return
+}
+
+func (k *KnowledgeResource) Update(cols []string) error {
+	return global.DmSQL["rddp"].Model(k).Select(cols).Updates(k).Error
+}
+
+func (k *KnowledgeResource) GetBatchKnowledgeResourceByOutIds(id []int, resourceType int) (item []*KnowledgeResource, err error) {
+	if len(id) == 0 {
+		return
+	}
+	sql := "SELECT * FROM knowledge_resource WHERE out_id IN (?) AND resource_type = ?"
+	err = global.DmSQL["rddp"].Raw(sql, id, resourceType).Find(&item).Error
+	return
+}
+
+// GetKnowledgeResourceByOutId 根据外部系统ID获取知识资源
+func (k *KnowledgeResource) GetKnowledgeResourceByOutId(outId int) (item *KnowledgeResource, err error) {
+	sql := "SELECT * FROM knowledge_resource WHERE out_id = ?"
+	err = global.DmSQL["rddp"].Raw(sql, outId).First(&item).Error
+	return
+}
+
+func (k *KnowledgeResource) GetMaxTimeKnowledgeByCondition(condition string, pars []interface{}) (item *KnowledgeResource, err error) {
+	sql := "SELECT * FROM knowledge_resource WHERE 1=1 "
+	if condition != "" {
+		sql += condition
+	}
+	sql += " ORDER BY modify_time DESC "
+	err = global.DmSQL["rddp"].Raw(sql, pars...).First(&item).Error
+	return
+}
+
+type KnowledgeClassify struct {
+	ClassifyId   int       `gorm:"primaryKey;column:classify_id"`
+	ClassifyName string    `gorm:"column:classify_name;default:'';type:varchar(125);not null"` // 注意：varchar的默认长度可能需要根据实际情况调整
+	Sort         int       `gorm:"column:sort;default:0;type:tinyint"`
+	ParentId     int       `gorm:"column:parent_id;default:0;type:int"`
+	CreateTime   time.Time `gorm:"column:create_time;default:CURRENT_TIMESTAMP"`
+	ModifyTime   time.Time `gorm:"column:modify_time;default:CURRENT_TIMESTAMP"`
+	Enabled      int       `gorm:"column:enabled;default:1;type:tinyint"`
+	Level        int       `gorm:"column:level;default:0;type:bigint"`
+	ResourceType int       `gorm:"column:resource_type;default:0;not null;type:tinyint"`
+}
+
+func (k KnowledgeClassify) TableName() string {
+	return "knowledge_classify"
+}
+
+func (k *KnowledgeClassify) Create() error {
+	return global.DmSQL["rddp"].Create(k).Error
+}
+
+func (k *KnowledgeClassify) GetClassifyByNameTypeAndParentId(classifyName string, resourceType, parentId int) (items *KnowledgeClassify, err error) {
+	sql := "SELECT * FROM knowledge_classify WHERE classify_name = ? AND resource_type = ? AND parent_id = ?"
+	err = global.DmSQL["rddp"].Raw(sql, classifyName, resourceType, parentId).First(&items).Error
+	return
+}
+
+func (k *KnowledgeClassify) GetClassifysByNameAndType(classifyName string, resourceType int) (item *KnowledgeClassify, err error) {
+	sql := "SELECT * FROM knowledge_classify WHERE classify_name = ? AND resource_type = ?"
+	err = global.DmSQL["rddp"].Raw(sql, classifyName, resourceType).Find(&item).Error
+	return
+}
+
+func (k *KnowledgeClassify) GetChildClassifyIdByNamePath(classifyNameFirst string, classifyNameSecond string, classifyNameThird string) (id int, err error) {
+	sql := `SELECT 
+    c3.classify_id AS id
+	FROM knowledge_classify c1 
+	JOIN knowledge_classify c2 ON c2.parent_id = c1.classify_id 
+					AND c2.classify_name = ? 
+					AND c2.level = 2
+	JOIN knowledge_classify c3 ON c3.parent_id = c2.classify_id 
+					AND c3.classify_name = ?
+					AND c3.level = 3
+	WHERE c1.classify_name = ?
+	AND c1.level = 1 `
+	err = global.DmSQL["rddp"].Raw(sql, classifyNameSecond, classifyNameThird, classifyNameFirst).Scan(&id).Error
+	return
+}
+
+type EtaBridgeKnowledgeResourceResp struct {
+	Code   int                `json:"code" description:"状态码"`
+	Msg    string             `json:"msg" description:"提示信息"`
+	Data   MarketOverviewResp `json:"data" description:"返回数据"`
+	ErrMsg string             `json:"-" description:"错误信息"`
+}
+
+type MarketOverviewResp struct {
+	Code string `json:"code"`
+	Mesg string `json:"mesg"`
+	Time string `json:"time"`
+	Data struct {
+		Result  bool                   `json:"result"`
+		Total   int                    `json:"total"`
+		Records []MarketOverviewRecord `json:"records"`
+	} `json:"data"`
+}
+
+type MarketOverviewRecord struct {
+	Category         string `json:"CATEGORY"`
+	Content          string `json:"CONTENT"`
+	CreateOriginTime string `json:"CREATE_ORIGIN_TIME"`
+	CreateTime       string `json:"CREATE_TIME"`
+	DataDate         string `json:"DATA_DATE"`
+	DataSource       string `json:"DATA_SOURCE"`
+	DataSourceCode   string `json:"DATA_SOURCE_CODE"`
+	Des              string `json:"DES"`
+	Id               int    `json:"ID"`
+	IsValidData      int    `json:"IS_VALID_DATA"` // 1:未删除；0:已删除
+	Person           string `json:"PERSON"`
+	Title            string `json:"TITLE"`
+	UpdateOriginTime string `json:"UPDATE_ORIGIN_TIME"`
+	UpdateTime       string `json:"UPDATE_TIME"`
+}

services/eta_bridge/es.go

@@ -0,0 +1,146 @@
+package eta_bridge
+
+import (
+	"context"
+	"encoding/json"
+	"eta_gn/eta_task/models"
+	"eta_gn/eta_task/utils"
+	"fmt"
+	"html"
+	"strconv"
+	"strings"
+
+	"github.com/PuerkitoBio/goquery"
+	"github.com/olivere/elastic/v7"
+)
+
+func NewClient() (client *elastic.Client, err error) {
+	client, err = elastic.NewClient(
+		elastic.SetURL(utils.ES_URL),
+		elastic.SetBasicAuth(utils.ES_USERNAME, utils.ES_PASSWORD),
+		elastic.SetSniff(false))
+
+	return
+}
+
+// EsAddOrEditKnowledgeResource 新增/修改es中的知识资源数据
+func EsAddOrEditKnowledgeResource(item *models.KnowledgeResource) (err error) {
+	defer func() {
+		if err != nil {
+			fmt.Println("EsAddOrEditData Err:", err.Error())
+			utils.FileLog.Info("EsAddOrEditKnowledgeResource err:", err)
+		}
+	}()
+	client, err := NewClient()
+	if err != nil {
+		return
+	}
+	indexName := utils.EsKnowledgeResourceIndexName
+	if item.IsFile == 0 {
+		content := ExtractTextFromResourceContent(item.Content)
+		contentRunes := []rune(content)
+		if len(contentRunes) > 60 {
+			item.Content = string(contentRunes[:60])
+		} else {
+			item.Content = content
+		}
+	}
+	request := client.Index().Index(indexName).Id(strconv.Itoa(item.KnowledgeResourceId)).BodyJson(item)
+	response, err := request.Do(context.Background())
+	if err != nil {
+		jsonBytes, _ := json.Marshal(item)
+		utils.FileLog.Info("add json:%s,EsAddOrEditKnowledgeResource err:%s", string(jsonBytes), err.Error())
+		return
+	}
+	if response.Status == 0 {
+		err = nil
+	} else {
+		fmt.Println("EsAddOrEditKnowledgeResource:", response.Status, response.Result)
+	}
+	return
+}
+
+func ExtractTextFromResourceContent(content string) (text string) {
+	content = html.UnescapeString(content)
+	doc, err := goquery.NewDocumentFromReader(strings.NewReader(content))
+	if err != nil {
+		return
+	}
+	text = doc.Text()
+	text = strings.ReplaceAll(text, "\n", "")
+	return
+}
+
+// UpdateEsKnowledgeResource updates a document in the Elasticsearch knowledge resource index.
+//
+// Parameters:
+//   - docId: The ID of the document to update.
+//   - docFields: A map containing the fields to update in the document.
+//
+// Returns:
+//   - err: An error object if an error occurred, otherwise nil.
+func UpdateEsKnowledgeResource(docId int, docFields map[string]interface{}) (err error) {
+	defer func() {
+		if err != nil {
+			fmt.Println("EsAddOrEditData Err:", err.Error())
+			utils.FileLog.Info("EsAddOrEditKnowledgeResource err:", err)
+		}
+	}()
+	client, err := NewClient()
+	if err != nil {
+		return
+	}
+	indexName := utils.EsKnowledgeResourceIndexName
+	res, err := client.Update().
+		Index(indexName).
+		Id(strconv.Itoa(docId)).
+		Doc(docFields).
+		Do(context.Background())
+	if err != nil {
+		utils.FileLog.Info("update docId:%d,UpdateEsKnowledgeResource err:%s", docId, err.Error())
+		return
+	}
+	if res.Result == "updated" {
+		err = nil
+	} else {
+		fmt.Println("UpdateEsKnowledgeResource:", res.Result)
+	}
+	return
+}
+
+func EsBatchAddOrEditKnowledgeResource(list []*models.KnowledgeResource) (err error) {
+	if len(list) == 0 {
+		return
+	}
+	defer func() {
+		if err != nil {
+			fmt.Println("EsBatchAddOrEditData Err:", err.Error())
+			utils.FileLog.Info("EsBatchAddOrEditKnowledgeResource err:", err)
+		}
+	}()
+	indexName := utils.EsKnowledgeResourceIndexName
+	client, err := NewClient()
+	if err != nil {
+		return
+	}
+
+	actions := make([]elastic.BulkableRequest, len(list))
+	for i, item := range list {
+		request := elastic.NewBulkIndexRequest().Index(indexName).Id(strconv.Itoa(item.KnowledgeResourceId)).Doc(item)
+		actions[i] = request
+	}
+	response, err := client.Bulk().Add(actions...).Do(context.Background())
+	if err != nil {
+		jsonBytes, _ := json.Marshal(list)
+		fmt.Println("add json:", string(jsonBytes))
+		fmt.Println("EsBatchAddOrEditKnowledgeResource err:", err)
+		return
+	}
+	if response.Errors {
+		err = fmt.Errorf("add knowledge resource to es failed, response result is %+v", response.Items)
+		fmt.Println("EsBatchAddOrEditKnowledgeResource:", response.Errors)
+	} else {
+		fmt.Printf("EsBatchAddOrEditKnowledgeResource:%+v\n", response)
+	}
+	return
+}

services/eta_bridge/knowledge.go

@@ -0,0 +1,227 @@
+package eta_bridge
+
+import (
+	"context"
+	"encoding/json"
+	"eta_gn/eta_task/models"
+	"eta_gn/eta_task/utils"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+var lockSyncDataNode sync.Mutex
+
+func SyncDataNode(cont context.Context) (err error) {
+	lockSyncDataNode.Lock()
+	defer lockSyncDataNode.Unlock()
+	syncDataNodeExecute()
+	return
+}
+
+func syncDataNodeExecute() (err error) {
+	fmt.Println("准备同步数据节点")
+	utils.FileLog.Info("准备同步数据节点")
+	errMsgList := make([]string, 0)
+	defer func() {
+		fmt.Println("同步数据节点结束")
+		if err != nil {
+			tips := "SyncDataNode-同步数据节点到ETA失败, ErrMsg:\n" + err.Error()
+			utils.FileLog.Info(tips)
+		}
+		if len(errMsgList) > 0 {
+			tips := "SyncDataNode-同步数据节点到ETA失败, ErrMsg:\n" + strings.Join(errMsgList, "\n")
+			utils.FileLog.Info(tips)
+		}
+	}()
+	obj := new(models.KnowledgeResource)
+	condition := " AND out_id > ? AND resource_type = ?"
+	pars := []interface{}{0, models.KnowledgeResourceTypeOpinion}
+	knowledge, err := obj.GetMaxTimeKnowledgeByCondition(condition, pars)
+	if err != nil && !utils.IsErrNoRow(err) {
+		errMsgList = append(errMsgList, err.Error())
+		return
+	}
+	param := make(map[string]interface{})
+	param["pageSize"] = 1000
+	if !utils.IsErrNoRow(err) {
+		now := time.Date(time.Now().Year(), time.Now().Month(), time.Now().Day(), 0, 0, 0, 0, time.Local)
+		if knowledge.ModifyTime.Before(now.AddDate(0, 0, -1)) {
+			param["UPDATE_TIME"] = knowledge.ModifyTime.Format(utils.FormatDateTime)
+		} else {
+			param["UPDATE_TIME"] = now.AddDate(0, 0, -1).Format(utils.FormatDateTime)
+		}
+	}
+
+	bBody, err, errMsg := HttpEtaBridgePost("/knowledge/viewpoint/acquire", param)
+	if err != nil {
+		errMsgList = append(errMsgList, errMsg)
+		errMsgList = append(errMsgList, err.Error())
+		return
+	}
+	var result models.EtaBridgeKnowledgeResourceResp
+	if er := json.Unmarshal(bBody, &result); er != nil {
+		errMsgList = append(errMsgList, er.Error())
+		err = er
+		return
+	}
+	if result.Code != 200 {
+		fmt.Println("eta_bridge同步数据节点失败")
+		errMsgList = append(errMsgList, result.Msg)
+		return
+	}
+	if result.Data.Code != "000000" {
+		fmt.Println("eta_bridge请求数据中心-市场简况接口同步数据节点失败")
+		errMsgList = append(errMsgList, result.Data.Mesg)
+		return
+	}
+	err = ViewPointBatchSave(result.Data.Data.Records)
+	if err != nil {
+		errMsgList = append(errMsgList, err.Error())
+	}
+	return
+}
+
+func ViewPointKnowledgeResourceClassifyCheckAndSave(classifyName string, resourceType int) (classifyId int, err error) {
+	obj := new(models.KnowledgeClassify)
+	obj, err = obj.GetClassifyByNameTypeAndParentId(classifyName, resourceType, 0)
+	if err != nil {
+		if !utils.IsErrNoRow(err) {
+			return
+		}
+		// 创建新的分类
+		err = nil
+		obj.ClassifyName = classifyName
+		obj.ResourceType = resourceType
+		obj.CreateTime = time.Now()
+		obj.ModifyTime = time.Now()
+		obj.Enabled = 1
+		obj.Level = 1
+		obj.Sort = 1
+		err = obj.Create()
+		if err != nil {
+			return
+		}
+		classifyId = obj.ClassifyId
+	} else {
+		classifyId = obj.ClassifyId
+	}
+	return
+}
+
+func ViewPointBatchSave(records []models.MarketOverviewRecord) (err error) {
+	if len(records) == 0 {
+		return
+	}
+	var outIds []int
+	var outMap = make(map[int]models.MarketOverviewRecord)
+	for _, record := range records {
+		outIds = append(outIds, record.Id)
+		outMap[record.Id] = record
+	}
+	obj := new(models.KnowledgeResource)
+	resourceList, err := obj.GetBatchKnowledgeResourceByOutIds(outIds, models.KnowledgeResourceTypeOpinion)
+	if err != nil {
+		return
+	}
+	for _, resource := range resourceList {
+		record, ok := outMap[resource.OutId]
+		if !ok {
+			continue
+		}
+		var updateCols []string
+		if resource.Title != record.Title {
+			resource.Title = record.Title
+			updateCols = append(updateCols, "title")
+		}
+		if resource.Content != record.Content {
+			resource.Content = record.Content
+			updateCols = append(updateCols, "content")
+		}
+		recordTime, er := time.Parse(utils.FormatDateTime, record.DataDate)
+		if er != nil {
+			fmt.Println("ViewPointBatchSave-时间转换失败,err :" + er.Error())
+			utils.FileLog.Info("ViewPointBatchSave-时间转换失败, record:%v ,err:%s", record, er.Error())
+			err = er
+			continue
+		}
+		if !resource.StartTime.Equal(recordTime) {
+			resource.StartTime = &recordTime
+			updateCols = append(updateCols, "start_time")
+		}
+		if resource.SourceFrom != record.DataSource {
+			resource.SourceFrom = record.DataSource
+			updateCols = append(updateCols, "source_from")
+		}
+		var isDelete int
+		if record.IsValidData == 0 {
+			isDelete = 1
+		}
+		if resource.IsDelete != isDelete {
+			resource.IsDelete = isDelete
+			updateCols = append(updateCols, "is_delete")
+		}
+		if len(updateCols) > 0 {
+			resource.ModifyTime = time.Now()
+			updateCols = append(updateCols, "modify_time")
+			err = resource.Update(updateCols)
+			if err != nil {
+				return
+			}
+			go func() {
+				err = EsAddOrEditKnowledgeResource(resource)
+				if err != nil {
+					utils.FileLog.Info("EsAddOrEditKnowledgeResource-同步es失败, err:%s", err.Error())
+				}
+			}()
+		}
+		delete(outMap, resource.OutId)
+	}
+	var addList []*models.KnowledgeResource
+	classifyId, er := ViewPointKnowledgeResourceClassifyCheckAndSave(records[0].Category, models.KnowledgeResourceTypeOpinion)
+	if er != nil {
+		err = er
+		return
+	}
+	for _, record := range outMap {
+		recordTime, er := time.Parse(utils.FormatDateTime, record.DataDate)
+		if er != nil {
+			fmt.Println("ViewPointBatchSave-时间转换失败,err :" + er.Error())
+			utils.FileLog.Info("ViewPointBatchSave-时间转换失败, record:%v ,err:%s", record, er.Error())
+			continue
+		}
+		timestamp := strconv.FormatInt(time.Now().UnixNano(), 10)
+		var isDelete int
+		if record.IsValidData == 0 {
+			isDelete = 1
+		}
+		addList = append(addList, &models.KnowledgeResource{
+			OutId:         record.Id,
+			Title:         record.Title,
+			Content:       record.Content,
+			State:         models.KnowledgeResourceStateApproved,
+			AdminRealName: "无",
+			StartTime:     &recordTime,
+			SourceFrom:    record.DataSource,
+			ClassifyId:    classifyId,
+			IsDelete:      isDelete,
+			ResourceCode:  utils.MD5(utils.CHART_PREFIX + "_" + timestamp),
+			ResourceType:  models.KnowledgeResourceTypeOpinion,
+			CreateTime:    time.Now(),
+			ModifyTime:    time.Now(),
+		})
+	}
+	err = obj.BatchCreate(addList)
+	if err != nil {
+		return
+	}
+	go func() {
+		e := EsBatchAddOrEditKnowledgeResource(addList)
+		if e != nil {
+			utils.FileLog.Info("ViewPointBatchSave-同步es失败, err:%s", e.Error())
+		}
+	}()
+	return
+}

services/task_gn.go

@@ -3,6 +3,7 @@ package services
 import (
 	"eta_gn/eta_task/services/eta_bridge"
 	"eta_gn/eta_task/utils"
+
 	"github.com/beego/beego/v2/task"
 )
 
@@ -31,6 +32,14 @@ func GnTask() {
 	//syncOaUser := task.NewTask("syncGnUser", syncOaUserTime, eta_bridge.SyncGnUser)
 	//task.AddTask("定时同步OA用户信息", syncOaUser)
 
+	// 每天6点，12点，18点，半点开始每10分钟同步一次
+	syncDataNodeTime := utils.SyncDataNodeTime
+	if syncDataNodeTime == `` {
+		syncDataNodeTime = "0 30/10 6,12,18 * * *"
+	}
+	syncDataNode := task.NewTask("syncDataNode", syncDataNodeTime, eta_bridge.SyncDataNode)
+	task.AddTask("定时同步数据节点观点数据", syncDataNode)
+
 	// 每分钟检查一下是否需要发送报告撰写提醒
 	sendReportWriteRemind := task.NewTask("sendReportWriteRemind", "0 */1 * * * *", ReportWriteRemind)
 	task.AddTask("定时发送报告撰写提醒", sendReportWriteRemind)

utils/common.go

@@ -21,6 +21,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"gorm.io/gorm"
 )
 
 func GetRandString(size int) string {
@@ -922,3 +924,10 @@ func TimeTransferString(format string, t time.Time) string {
 	}
 	return str
 }
+
+func IsErrNoRow(err error) bool {
+	if err == nil {
+		return false
+	}
+	return errors.Is(err, gorm.ErrRecordNotFound)
+}

utils/config.go

@@ -2,9 +2,10 @@ package utils
 
 import (
 	"fmt"
+	"strconv"
+
 	beego "github.com/beego/beego/v2/adapter"
 	"github.com/beego/beego/v2/server/web"
-	"strconv"
 )
 
 var (
@@ -82,9 +83,10 @@ var (
 
 // ES索引配置
 var (
-	EsReportIndexName        string //研报ES索引
-	EsEnglishReportIndexName string //英文研报ES索引
-	SmartReportIndexName     string //智能研报ES索引
+	EsReportIndexName            string //研报ES索引
+	EsEnglishReportIndexName     string //英文研报ES索引
+	SmartReportIndexName         string //智能研报ES索引
+	EsKnowledgeResourceIndexName string //知识资源库ES索引
 )
 
 // 科大讯飞--语音合成
@@ -123,6 +125,7 @@ var (
 	SyncCrmDataSourceType string // 桥接服务-crm内部来源系统参数配置
 	SyncCrmIndexNum       int    // 桥接服务-每次同步crm数据的数量
 	SyncOaUserTime        string // 桥接服务-OA用户同步时间
+	SyncDataNodeTime      string // 桥接服务-数据节点同步时间
 	SyncOaTodoPath        string // 桥接服务-OA待办接口
 )
 
@@ -230,6 +233,7 @@ func init() {
 		EsReportIndexName = config["es_report_index_name"]
 		EsEnglishReportIndexName = config["es_english_report_index_name"]
 		SmartReportIndexName = config["es_smart_report_index_name"]
+		EsKnowledgeResourceIndexName = config["es_knowledge_resource_index_name"]
 	}
 
 	// 科大讯飞
@@ -274,6 +278,7 @@ func init() {
 		SyncCrmAssetPkgCd = config["sync_crm_asset_pkg_cd"]         // 桥接服务-crm资产包配置
 		SyncCrmDataSourceType = config["sync_crm_data_source_type"] // 桥接服务-crm内部来源系统参数配置
 		SyncOaUserTime = config["sync_oa_user_time"]                // 桥接服务-OA用户同步时间
+		SyncDataNodeTime = config["sync_data_node_time"]            // 桥接服务-数据节点同步时间
 		SyncOaTodoPath = config["sync_oa_todo_path"]                // 桥接服务-OA待办接口
 
 		syncCrmIndexNumStr := config["sync_crm_index_num"] // 桥接服务-每次同步crm数据的数量