eta_api/services/task.go

package services

import (
	"encoding/json"
	"eta/eta_api/cache"
	"eta/eta_api/models"
	"eta/eta_api/models/rag"
	"eta/eta_api/services/binlog"
	"eta/eta_api/services/data"
	edbmonitor "eta/eta_api/services/edb_monitor"
	"eta/eta_api/services/llm"
	"eta/eta_api/utils"
	"fmt"
	"strings"
	"time"
)

func Task() {
	fmt.Println("task start")

	{
		// 修复客户试用数据
		//FixCompanyUpdateData()
		//FixCompanyTryDay()
		//FixCompanyTryDay()
		//StaticCompanyTryDay()
		//return
	}

	//FixPermissionStatus()
	//GetCompanyInfo()
	//ReportCount()
	//windSourceUrl:=`http://47.100.166.55:7002/edbInfo/wind/?EdbCode=M0001427&StartDate=2020-11-01&EndDate=2021-03-01`
	//data.AddAllArticle()

	//data.GetSmmIndex()
	//data.GetSmmIndexData()
	go AutoInsertLogToDB()

	//手工数据表格导入后的指标库刷新
	go ImportManualDataRefresh()

	// 加入上海钢联指标数据之后的刷新
	go MysteelChemicalDataAdd()

	//修复用户关注标识
	//GetWxUsersSubscribe()

	go AutoInsertAdminOperateRecordToDB()

	// 指标刷新
	go data.HandleEdbRefreshQueue()

	// 进行指标替换操作
	go DealReplaceEdbCache()

	// TODO:监听这里需要找下达梦的方案，主流程先跑起来再处理
	// 监听binlog
	if utils.MYSQL_DATA_BINLOG_URL != "" && utils.DbDriverName == `mysql` {
		go binlog.ListenMysql()

		go edbmonitor.HandleEdbMonitorEdbInfo()

		// 监听数据源binlog写入es
		go binlog.HandleDataSourceChange2Es()
	}
	go StartSessionManager()

	go llm.SaveAllChatRecordsToDB()

	// 定时任务进行微信文章操作
	// 定时任务进行微信公众号添加、文章刷新操作（暂不处理了，改成其他服务处理）

	//go HandleWechatArticleOp()

	// 定时任务进行微信文章LLM操作
	go HandleWechatArticleLLmOp()

	// 队列任务将eta报告同步到知识库操作
	go HandleEtaReportKnowledgeLLmOp()

	// 权益报告监听入库
	go AutoInsertRaiReport()

	// TODO:数据修复
	//FixNewEs()
	fmt.Println("task end")
}

//// 每日发布晨报
//func AutoPublishDayReport() {
//	defer func() {
//		if err := recover(); err != nil {
//			fmt.Println("[AutoPublishDayReport]", err)
//		}
//	}()
//
//	// 每日8:42发布晨报
//	ticker := time.Tick(50 * time.Second)
//	for range ticker {
//		nowTime := time.Now()
//		clock := nowTime.Format("1504")
//		if clock == "0842" {
//			if err := PublishTodayDayReport(); err != nil {
//				go alarm_msg.SendAlarmMsg(fmt.Sprint("每日晨报自动发送 AutoPublishDayReport ERR:", err), 3)
//				//utils.SendEmail(utils.APPNAME+" "+utils.RunMode+" 失败提醒", fmt.Sprint("AutoPublishDayReport ERR:", err), utils.EmailSendToUsers)
//			}
//		}
//	}
//}

// ImportManualDataRefresh 导入手工数据后的刷新
func ImportManualDataRefresh() {
	defer func() {
		if err := recover(); err != nil {
			fmt.Println("[ImportManualDataRefresh]", err)
		}
	}()
	for {
		utils.Rc.Brpop(utils.CACHE_IMPORT_MANUAL_DATA, func(b []byte) {
			edbCode := string(b)
			edbCode = strings.TrimPrefix(edbCode, `"`)
			edbCode = strings.TrimSuffix(edbCode, `"`)

			data.RefreshManualData(edbCode)
		})
	}
}

// MysteelChemicalDataAdd 加入上海钢联指标数据之后的刷新
func MysteelChemicalDataAdd() {
	defer func() {
		if err := recover(); err != nil {
			fmt.Println("[ImportManualDataRefresh]", err)
		}
	}()
	for {
		utils.Rc.Brpop(utils.CACHE_MYSTEEL_CHEMICAL_ADD_DATA, func(b []byte) {
			edbCode := string(b)
			edbCode = strings.TrimPrefix(edbCode, `"`)
			edbCode = strings.TrimSuffix(edbCode, `"`)

			data.RefreshMysteelChemicalData(edbCode)
		})
	}
}

//func init()  {
//	fmt.Println("start task init")
//	UpdateEnglishEmailLogErrMsg()
//	fmt.Println("end task init")
//}
//
//// UpdateEnglishEmailLogErrMsg 更新英文邮件日志的ErrMsg(研报后台4.2上线后执行, 仅一次)
//func UpdateEnglishEmailLogErrMsg() {
//	var cond string
//	var pars []interface{}
//	list, e := models.GetEnglishReportEmailLogList(cond, pars)
//	if e != nil {
//		fmt.Println("获取日志列表失败")
//		return
//	}
//	for _, v := range list {
//		if v.SendStatus != 0 || v.Source != 1 || v.Result == "" || v.ErrMsg != "" {
//			continue
//		}
//		// 取出错误信息
//		fmt.Printf("正在更新%d\n", v.Id)
//		r := new(AliyunEmailResult)
//		if e = json.Unmarshal([]byte(v.Result), &r); e != nil {
//			fmt.Println("JSON解析报错了1" + e.Error())
//			continue
//		}
//		rd := new(AliyunEmailResultData)
//		res := strings.Replace(r.Data, `\`, ``, -1)
//		if e = json.Unmarshal([]byte(res), &rd); e != nil {
//			fmt.Println("JSON解析报错了2" + e.Error())
//			continue
//		}
//		v.ErrMsg = rd.Message
//		if e = v.Update([]string{"ErrMsg"}); e != nil {
//			fmt.Println("更新失败了" + e.Error())
//			continue
//		}
//	}
//	fmt.Println("更新成功")
//}

//func FixCompanyUpdateData1() {
//	list, err := company.GetCompanyProductUpdateLogList()
//	if err != nil {
//		fmt.Println("获取客户变更数据失败：", err)
//		return
//	}
//
//	nowTime := time.Now()
//	for _, v := range list {
//		//item, tmpErr := company.GetCompanyProductLogItem(v.CompanyId, v.ProductId, v.CreateTime)
//		//if tmpErr != nil {
//		//	fmt.Println(v.Id, "找数据，", tmpErr)
//		//	continue
//		//}
//
//		//permissionList := make([]*company.CompanyReportPermission, 0)
//		//switch v.Source {
//		//case "add", "receive", "thaw", "delay", "apply_receive":
//		//
//		//}
//		permissionList, err := company.GetCompanyReportPermission(v.CompanyId, v.ProductId)
//		if err != nil {
//			fmt.Println("err:", err)
//			continue
//		}
//
//		startDate := v.CreateTime
//		endDate := v.CreateTime.AddDate(0, 2, 0)
//		isStop := 1
//		realEndDate := endDate
//		if realEndDate.After(nowTime) {
//			realEndDate = nowTime
//			isStop = 0
//		}
//		for _, permission := range permissionList {
//			permission.StartDate = startDate.Format(utils.FormatDate)
//			permission.EndDate = v.CreateTime.Format(utils.FormatDate)
//		}
//		companyProductTryOutUpdateLogItem := &company.CompanyProductTryOutUpdateLog{
//			Id:          0,
//			CompanyId:   v.CompanyId,
//			ProductId:   v.ProductId,
//			SellerId:    v.SellerId,
//			SellerName:  v.SellerName,
//			Source:      v.Source,
//			StartDate:   startDate,
//			EndDate:     endDate,
//			RealEndDate: realEndDate,
//			IsStop:      isStop,
//			CreateTime:  v.CreateTime,
//		}
//		err = company.AddCompanyProductTryOutUpdateLog(companyProductTryOutUpdateLogItem, permissionList)
//	}
//	//fmt.Println("结束")
//}

//func FixCompanyUpdateData2() {
//	list, err := company.GetTryOutCompanyOperationRecordList()
//	if err != nil {
//		fmt.Println("获取客户变更数据失败：", err)
//		return
//	}
//
//	nowTime := time.Now()
//	for _, v := range list {
//		//item, tmpErr := company.GetCompanyProductLogItem(v.CompanyId, v.ProductId, v.CreateTime)
//		//if tmpErr != nil {
//		//	fmt.Println(v.Id, "找数据，", tmpErr)
//		//	continue
//		//}
//
//		//permissionList := make([]*company.CompanyReportPermission, 0)
//		//switch v.Source {
//		//case "add", "receive", "thaw", "delay", "apply_receive":
//		//
//		//}
//		permissionList, err := company.GetCompanyReportPermission(v.CompanyId, v.ProductId)
//		if err != nil {
//			fmt.Println("err:", err)
//			continue
//		}
//
//		startDate := v.CreateTime
//		endDate := v.CreateTime.AddDate(0, 2, 0)
//		isStop := 1
//		realEndDate := endDate
//		if realEndDate.After(nowTime) {
//			realEndDate = nowTime
//			isStop = 0
//		}
//		for _, permission := range permissionList {
//			permission.StartDate = startDate.Format(utils.FormatDate)
//			permission.EndDate = v.CreateTime.Format(utils.FormatDate)
//		}
//		sellerName := ``
//		{
//			sysUser, _ := system.GetSysAdminById(v.SellerId)
//			if sysUser != nil {
//				sellerName = sysUser.RealName
//			}
//		}
//		companyProductTryOutUpdateLogItem := &company.CompanyProductTryOutUpdateLog{
//			Id:          0,
//			CompanyId:   v.CompanyId,
//			ProductId:   v.ProductId,
//			SellerId:    v.SellerId,
//			SellerName:  sellerName,
//			Source:      "formal_to_try_out",
//			StartDate:   startDate,
//			EndDate:     endDate,
//			RealEndDate: realEndDate,
//			IsStop:      isStop,
//			CreateTime:  v.CreateTime,
//		}
//		err = company.AddCompanyProductTryOutUpdateLog(companyProductTryOutUpdateLogItem, permissionList)
//	}
//	fmt.Println("结束")
//}

// FixCompanyTryDay 修复试用天数
//func FixCompanyTryDay() {
//	list, err := company.GetCompanyProductTryOutUpdateGroup()
//	if err != nil {
//		fmt.Println("获取客户变更数据失败：", err)
//		return
//	}
//
//	lenList := len(list)
//	for k, v := range list {
//		fmt.Println("剩余", lenList-k-1, "条数据修复")
//		companyProduct, tmpErr := company.GetCompanyProductByCompanyIdAndProductId(v.CompanyId, v.ProductId)
//		if tmpErr != nil {
//			fmt.Println("company_id:", v.CompanyId, ";product_id:", v.ProductId, "；找不到对应的客户，Err:", tmpErr)
//			continue
//		}
//
//		logList, err := company.GetCompanyProductTryOutUpdateList(v.CompanyId, v.ProductId)
//		if err != nil {
//			fmt.Println("查找客户日志失败，err:", err)
//			continue
//		}
//
//		//lenLog := len(logList)
//		var day int //实际试用天数
//		var endDate time.Time
//		for _, log := range logList {
//			startDate := log.StartDate
//			if endDate.IsZero() {
//				endDate = log.RealEndDate
//				day = utils.GetTimeSubDay(startDate, log.RealEndDate) + 1
//			} else {
//				if log.RealEndDate.After(endDate) {
//					if endDate.After(startDate) {
//						startDate = endDate
//					}
//					if startDate.Equal(log.EndDate) {
//						day += utils.GetTimeSubDay(startDate, log.RealEndDate)
//					} else {
//						day += utils.GetTimeSubDay(startDate, log.RealEndDate) + 1
//					}
//					endDate = log.RealEndDate
//				}
//			}
//		}
//		companyProduct.TryOutDayTotal = day
//		companyProduct.Update([]string{"TryOutDayTotal"})
//	}
//	fmt.Println("结束")
//}

// StaticCompanyTryDay 定时任务每天更新试用天数
//func StaticCompanyTryDay() {
//	list, err := company.GetCompanyProductTryOutUpdateNoStopGroup()
//	if err != nil {
//		fmt.Println("获取客户变更数据失败：", err)
//		return
//	}
//
//	lenList := len(list)
//	for k, v := range list {
//		isAdd := false //是否要增加一天，默认不加
//		fmt.Println("剩余", lenList-k-1, "条数据修复")
//		permissionList, tmpErr := company.GetCompanyReportPermissionByStatus(v.CompanyId, v.ProductId, "试用")
//		if tmpErr == nil {
//			fmt.Println("company_id:", v.CompanyId, ";product_id:", v.ProductId, "；寻找对应的试用品种失败，Err:", tmpErr)
//			continue
//		}
//		currPermissionIdList := make([]int, 0) //当前试用的品种
//		for _, permission := range permissionList {
//			currPermissionIdList = append(currPermissionIdList, permission.ChartPermissionId)
//		}
//
//		//获取所有未停止的记录列表
//		logList, err := company.GetCompanyProductTryOutUpdateNoStopListByEndDate(v.CompanyId, v.ProductId)
//		if err != nil {
//			fmt.Println("查找客户日志失败，err:", err)
//			continue
//		}
//
//		logPermissionIdList := make([]int, 0) //当前日志中的试用的品种
//		for _, log := range logList {
//			//获取所有未停止的品种试用记录列表
//			logPermissionList, tmpErr := company.GetCompanyProductTryOutPermissionUpdateNoStopListByEndDate(log.Id)
//			if tmpErr != nil {
//				fmt.Println("查找客户品种变更日志失败，err:", tmpErr)
//				continue
//			}
//			lenLogPermissionList := len(logPermissionList) //当前日志存在试用的品种数量
//			stopPermission := 0                            //当前日志需要停止的品种数量
//
//			currTime := time.Now() //当前时间
//			for _, logPermission := range logPermissionList {
//				if utils.InArrayByInt(logPermissionIdList, logPermission.ChartPermissionId) {
//					// 如果已经被其他记录使用了，那么就将当前记录给标记停止
//					logPermission.IsStop = 1
//
//					// 如果当前时间晚于结束日期，那么实际结束日期就是今天之前一天，否则就是当天
//					if currTime.After(logPermission.EndDate) {
//						logPermission.RealEndDate = currTime.AddDate(0, 0, -1)
//					} else {
//						logPermission.RealEndDate = time.Now()
//					}
//					logPermission.Update([]string{"IsStop", "RealEndDate"})
//					stopPermission++
//					continue
//				} else if !utils.InArrayByInt(currPermissionIdList, logPermission.ChartPermissionId) {
//					// 如果该品种不在当前客户的品种里面，那么也要将当前记录给标记停止
//					logPermission.IsStop = 1
//					// 如果当前时间晚于结束日期，那么实际结束日期就是今天之前一天，否则就是当天
//					if currTime.After(logPermission.EndDate) {
//						logPermission.RealEndDate = currTime.AddDate(0, 0, -1)
//					} else {
//						logPermission.RealEndDate = time.Now()
//					}
//					logPermission.Update([]string{"IsStop", "RealEndDate"})
//					stopPermission++
//					continue
//				}
//
//				// 剩下的说明还处于试用状态，需要添加1天试用期，且需要把该品种加入到当前日志中的试用的品种列表
//				isAdd = true
//				logPermissionIdList = append(logPermissionIdList, logPermission.ChartPermissionId)
//			}
//
//			//如果当前日志存在试用的品种数量 == 当前日志需要停止的品种数量
//			// 那么当前日志也是处于停用状态
//			if lenLogPermissionList == stopPermission {
//				log.IsStop = 1
//				// 如果当前时间晚于结束日期，那么实际结束日期就是今天之前一天，否则就是当天
//				if currTime.After(log.EndDate) {
//					log.RealEndDate = currTime.AddDate(0, 0, -1)
//				} else {
//					log.RealEndDate = time.Now()
//				}
//				log.Update([]string{"IsStop", "RealEndDate"})
//			}
//
//		}
//
//		// 如果需要添加，那么将该客户品种添加1天
//		if isAdd {
//
//		}
//		// 更新客户产品的试用天数
//		tmpErr = company.AddCompanyProductTryOutDayTotal(v.CompanyId, v.ProductId)
//		if tmpErr != nil {
//			fmt.Println("company_id:", v.CompanyId, ";product_id:", v.ProductId, "；更新客户产品的试用天数，Err:", tmpErr)
//			continue
//		}
//	}
//	fmt.Println("结束")
//}

//func Task2() {
//	fmt.Println("task start")
//
//	//_, _ = maycur.GetPublicOfferingSaleLeader()
//	//_ = maycur.TestSyncCompanyProfile()
//	//_ = maycur.SyncCompanyProfile()
//	//_ = maycur.ImportExcelEmployeeId()
//	//_ = maycur.ApiTest()
//	//_ = maycur.ApiTest2()
//	fmt.Println("task end")
//}

// FixEnCompanyPermission 英文权限上线时修复英文客户拥有所有权限(一次性)
func FixEnCompanyPermission() {
	var err error
	defer func() {
		if err != nil {
			fmt.Println("FixEnCompanyPermission Err: ", err.Error())
		}
	}()

	// 获取正式客户
	companies := make([]*models.EnglishCompany, 0)
	{
		cond := ` AND status = ?`
		pars := make([]interface{}, 0)
		pars = append(pars, 1)
		list, e := models.GetEnglishCompanyList(cond, pars, "")
		if e != nil {
			err = fmt.Errorf("GetEnglishCompanyList err: %s", e.Error())
			return
		}
		companies = list
	}

	// 获取所有权限
	permissions := make([]*models.EnPermission, 0)
	{
		cond := ` AND parent_id > ?`
		pars := make([]interface{}, 0)
		pars = append(pars, 0)
		ob := new(models.EnPermission)
		list, e := ob.GetItemsByCondition(cond, pars, []string{}, "")
		if e != nil {
			err = fmt.Errorf("GetPermissionItemsByCondition err: %s", e.Error())
			return
		}
		permissions = list
	}

	for _, c := range companies {
		ps := make([]*models.EnCompanyPermission, 0)
		for _, p := range permissions {
			ps = append(ps, &models.EnCompanyPermission{
				EnCompanyId:    c.CompanyId,
				EnPermissionId: p.EnPermissionId,
				CreateTime:     time.Now().Local(),
			})
		}
		if e := models.ClearAndCreateEnCompanyPermissions(c.CompanyId, ps); e != nil {
			err = fmt.Errorf("ClearAndCreateEnCompanyPermissions err: %s", e.Error())
			return
		}
	}
	fmt.Println("修复完成")
}

// ModifyEsEnglishReport 批量修改es里的英文研报信息和线上路演信息
func ModifyEsEnglishReport() {
	fmt.Println("开始")
	err := ModifyAllEsEnglishReportVideo()
	if err != nil {
		err = fmt.Errorf("重置es中的英文研报信息失败：ModifyAllEnglishReportInEs err: %s", err.Error())
		return
	}
	fmt.Println("结束")
}

//func InsertBloombergIndex() {
//	fmt.Println("开始写入")
//
//	start := 100000
//	now := time.Now()
//	for i := 1; i <= 100; i++ {
//		fmt.Printf("写入第%d个\n", i)
//
//		start += 1
//		index := new(data_manage.BaseFromBloombergIndex)
//		index.IndexCode = fmt.Sprintf("BLID%d", start)
//		index.IndexName = fmt.Sprintf("模拟Bloomberg-%s", index.IndexCode)
//		index.Unit = "无"
//		index.Source = utils.DATA_SOURCE_BLOOMBERG
//		index.Frequency = "日度"
//		index.StartDate = now.AddDate(0, 0, -i)
//		index.EndDate = now
//		index.CreateTime = time.Now().Local()
//		index.ModifyTime = time.Now().Local()
//		if e := index.Create(); e != nil {
//			fmt.Printf("新增指标失败, IndexCode: %s, err: %s", index.IndexCode, e.Error())
//			return
//		}
//
//		insertData := make([]*data_manage.BaseFromBloombergData, 0)
//		for ii := 0; ii <= 50; ii++ {
//			indexData := new(data_manage.BaseFromBloombergData)
//			indexData.BaseFromBloombergIndexId = index.BaseFromBloombergIndexId
//			indexData.IndexCode = index.IndexCode
//			indexData.DataTime = now.AddDate(0, 0, -ii)
//			va := GenerateRandomFloat64InRange()
//			va += float64(ii)
//			indexData.Value = va
//			indexData.CreateTime = time.Now().Local()
//			indexData.ModifyTime = time.Now().Local()
//			indexData.DataTimestamp = int(indexData.DataTime.UnixNano() / 1e6)
//			insertData = append(insertData, indexData)
//		}
//		ob := new(data_manage.BaseFromBloombergData)
//		if e := ob.CreateMulti(insertData); e != nil {
//			fmt.Printf("新增指标数据失败, IndexCode: %s, err: %s", index.IndexCode, e.Error())
//			return
//		}
//	}
//
//	fmt.Println("结束写入")
//}
//
//func GenerateRandomFloat64InRange() float64 {
//	var rnd = rand.New(rand.NewSource(time.Now().UnixNano())) // 设置随机数种子
//
//	return rnd.Float64()*11000 - 1000
//}

// HandleSearchByWechatOp
// @Description: 处理微信爬虫
func HandleWechatArticleOp() {
	defer func() {
		if err := recover(); err != nil {
			fmt.Println("[HandleWechatArticleOp]", err)
		}
	}()
	obj := rag.WechatPlatform{}
	for {
		utils.Rc.Brpop(utils.CACHE_WECHAT_PLATFORM_ARTICLE, func(b []byte) {
			wechatArticleOp := new(cache.WechatArticleOp)
			if err := json.Unmarshal(b, &wechatArticleOp); err != nil {
				fmt.Println("json unmarshal wrong!")
				return
			}
			item, tmpErr := obj.GetById(wechatArticleOp.WechatPlatformId)
			if tmpErr != nil {
				// 找不到就处理失败
				return
			}

			switch wechatArticleOp.Source {
			case `add`:
				AddWechatPlatform(item)
			case `refresh`:
				BeachAddWechatArticle(item, 2)

			}
		})
	}
}

// HandleWechatArticleLLmOp
// @Description: 处理微信文章加入知识库
func HandleWechatArticleLLmOp() {
	defer func() {
		if err := recover(); err != nil {
			fmt.Println("[HandleWechatArticleLLmOp]", err)
		}
	}()
	obj := rag.WechatArticle{}
	for {
		utils.Rc.Brpop(utils.CACHE_WECHAT_PLATFORM_ARTICLE_KNOWLEDGE, func(b []byte) {
			wechatArticleOp := new(cache.WechatArticleOp)
			if err := json.Unmarshal(b, &wechatArticleOp); err != nil {
				fmt.Println("json unmarshal wrong!")
				return
			}
			item, tmpErr := obj.GetById(wechatArticleOp.WechatPlatformId)
			if tmpErr != nil {
				// 找不到就处理失败
				return
			}

			// 文章加入到知识库
			ArticleToKnowledge(item)
			// 生成摘要
			//GenerateArticleAbstract(item)
		})
	}
}

// HandleEtaReportKnowledgeLLmOp
// @Description: 处理eta报告加入知识库操作
func HandleEtaReportKnowledgeLLmOp() {
	defer func() {
		if err := recover(); err != nil {
			fmt.Println("[HandleEtaReportKnowledgeLLmOp]", err)
		}
	}()
	for {
		utils.Rc.Brpop(utils.CACHE_ETA_REPORT_KNOWLEDGE, func(b []byte) {
			ragEtaReportOpOp := new(cache.RagEtaReportOpOp)
			if err := json.Unmarshal(b, &ragEtaReportOpOp); err != nil {
				fmt.Println("json unmarshal wrong!")
				return
			}
			switch ragEtaReportOpOp.Source {
			case `publish`:
				ReportAddOrModifyKnowledgeByReportId(ragEtaReportOpOp.ReportId)
			case `un_publish`:
				ReportUnPublishedKnowledgeByReportId(ragEtaReportOpOp.ReportId)
			}

		})
	}
}