eta_mini_ht_api/domian/report/report_service.go

package report

import (
	"encoding/json"
	"eta/eta_mini_ht_api/common/component/config"
	"eta/eta_mini_ht_api/common/component/es"
	logger "eta/eta_mini_ht_api/common/component/log"
	"eta/eta_mini_ht_api/common/contants"
	"eta/eta_mini_ht_api/common/utils/page"
	stringUtils "eta/eta_mini_ht_api/common/utils/string"
	analystService "eta/eta_mini_ht_api/domian/financial_analyst"
	userService "eta/eta_mini_ht_api/domian/user"
	"eta/eta_mini_ht_api/models"
	"eta/eta_mini_ht_api/models/eta"
	etaDao "eta/eta_mini_ht_api/models/eta"
	"eta/eta_mini_ht_api/models/ht"
	mediaDao "eta/eta_mini_ht_api/models/media"
	reportDao "eta/eta_mini_ht_api/models/report"
	"github.com/google/uuid"
	"math/rand"
	"strconv"
	"strings"
	"time"
)

const (
	SourceETA = "ETA"
	SourceHT  = "HT"

	DESC models.Order = "desc"
	ASC  models.Order = "asc"

	ESColumn      = "title"
	ESRangeColumn = "reportId"
)

var (
	sortField = []string{"_score:desc"}
	htConfig  = config.GetConfig(contants.HT).(*config.HTBizConfig)
)

func elastic() *es.ESClient {
	return es.GetInstance()
}

// ESReport Report ES研报mapping
type ESReport struct {
	ReportID      int                    `json:"reportId"`
	OrgId         int                    `json:"orgId"`
	Title         string                 `json:"title"`
	Author        string                 `json:"author"`
	Source        reportDao.ReportSource `json:"source"`
	Abstract      string                 `json:"abstract"`
	CoverSrc      int                    `json:"coverSrc"`
	Status        reportDao.ReportStatus `json:"status"`
	PublishedTime string                 `json:"publishedTime"`
}

type ReportDTO struct {
	ReportID         int             `json:"reportId"`
	OrgId            int             `json:"orgId"`
	Title            string          `json:"title"`
	Author           string          `json:"author"`
	Source           string          `json:"source"`
	Abstract         string          `json:"abstract"`
	PublishedTime    string          `json:"publishedTime"`
	SecondPermission map[int]string  `json:"-"`
	Permissions      map[int]string  `json:"-"`
	PermissionNames  interface{}     `json:"permissionNames,omitempty"`
	Highlight        []string        `json:"highlight,omitempty"`
	Detail           json.RawMessage `json:"detail,omitempty"`
	PdfUrl           string          `json:"pdfUrl,omitempty"`
	CoverSrc         int             `json:"coverSrc"`
	CoverUrl         string          `json:"coverUrl"`
}

type Detail struct {
}

type PermissionDTO struct {
	ID       int
	Name     string
	ParentID int
}

func GetGetReportById(reportId int) (ReportDTO ReportDTO, err error) {
	report, err := reportDao.GetReportById(reportId)
	if err != nil {
		return
	}
	//orgId := report.OrgID
	//names, _ := reportDao.GetAuthorByOrgId(orgId, string(report.Source))
	//if names != nil && len(names) > 1 {
	//	names = stringUtils.UniqueItems(names)
	//	report.Author = strings.Join(names, ",")
	//}
	//report.PublishedTime = report.PublishedTime
	ReportDTO = convertReportDTO(report, true)
	return
}

func GetTotalPageCount() (total int64) {
	return reportDao.GetTotalPageCount()
}
func GetTotalPageCountByAnalyst(analyst string) (total int64, latestId int64) {
	return reportDao.GetTotalPageCountByAnalyst(analyst)
}
func SearchMaxReportId(key string) (total int64, reportId int64) {
	sort := []string{"reportId:desc"}
	request := matchAll(sort, key)
	//同步es
	re, err := elastic().Count(request)
	if err != nil {
		logger.Error("es搜索异常:%v", err)
	}
	count := re.Count
	total = int64(count)
	if total > 0 {
		request = match(key, 0, count, sort)
		re, err = elastic().Search(request)
		if err != nil {
			logger.Error("es搜索异常:%v", err)
		}
		hits := elastic().GetSource(re.Hits)
		data := hits[0].Source
		report := ReportDTO{}
		err = json.Unmarshal(data, &report)
		if err != nil {
			logger.Error("获取当前最大研报id失败:%v", err)
			return
		}
		reportId = int64(report.ReportID)
	}
	return
}

func SearchReportList(key string, from int, size int, max int64) (reports []ReportDTO, err error) {
	//同步es
	sorts := append(sortField, "publishedTime:desc")
	request := matchRange(key, from, size, max, sorts)
	re, err := elastic().Search(request)
	if err != nil {
		logger.Error("es搜索异常:%v", err)
	}
	hits := elastic().GetSource(re.Hits)
	if len(hits) == 0 {
		reports = []ReportDTO{}
		return
	}
	for _, hit := range hits {
		var content map[string][]string
		err = json.Unmarshal(hit.Highlight, &content)
		report := ReportDTO{}
		err = json.Unmarshal(hit.Source, &report)
		if err != nil {
			logger.Error("解析研报数据失败:%v", err)
			continue
		}
		report.Highlight = content[ESColumn]
		report.Title = report.Highlight[0]
		report.PublishedTime = report.PublishedTime[:10]
		reports = append(reports, report)
	}
	return
}
func GetReportPageByAnalyst(pageInfo page.PageInfo, analyst string) (list []ReportDTO, err error) {
	offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize)
	reports, err := reportDao.GetReportPageByAnalyst(pageInfo.LatestId, pageInfo.PageSize, offset, analyst)
	if err != nil {
		logger.Error("分页查询报告列表失败:%v", err)
		return
	}
	list = make([]ReportDTO, 0)
	if reports != nil {
		for _, report := range reports {
			dto := convertReportDTO(report, false)
			list = append(list, dto)
		}
	}
	return
}
func GetReportPageByOrgIds(pageInfo page.PageInfo, orgIds map[string][]int) (list []ReportDTO, err error) {
	offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize)
	reports, err := reportDao.GetReportPageByOrgIds(pageInfo.LatestId, pageInfo.PageSize, offset, orgIds)
	if err != nil {
		logger.Error("分页查询报告列表失败:%v", err)
		return
	}
	list = make([]ReportDTO, 0)
	if reports != nil {
		for _, report := range reports {
			dto := convertReportDTO(report, false)
			list = append(list, dto)
		}
	}
	return
}

func GetNewReportByPublishTime(time time.Time) (reports []ReportDTO) {
	list := reportDao.GetNewReportByPublishTime(time)
	if list != nil {
		for _, report := range list {
			dto := convertReportDTO(report, false)
			reports = append(reports, dto)
		}
	}
	return
}
func GetReportPage(pageInfo page.PageInfo) (list []ReportDTO, err error) {
	offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize)
	reports, err := reportDao.GetReportPage(pageInfo.LatestId, pageInfo.PageSize, offset)
	if err != nil {
		logger.Error("分页查询报告列表失败:%v", err)
		return
	}
	list = make([]ReportDTO, 0)
	if reports != nil {
		for _, report := range reports {
			dto := convertReportDTO(report, false)
			list = append(list, dto)
		}
	}
	return
}
func getETAReportFirstPermissions(id int) (permissionDTOs []PermissionDTO) {
	classifyId, err := etaDao.GetReportClassifyById(id)
	if err != nil || classifyId == 0 {
		logger.Error("获取研报分类信息失败:%v", err)
		return
	}
	permissions, err := etaDao.GetFirstPermissionsByClassifyID(classifyId)
	if err != nil {
		logger.Error("获取研报一级品种信息失败:%v", err)
		return
	}
	for _, permission := range permissions {
		permissionDTOs = append(permissionDTOs, convertPermissionDTO(permission))
	}
	return
}

func getETAReportSecondPermissions(id int) (permissionDTOs []PermissionDTO) {
	classifyId, err := etaDao.GetReportClassifyById(id)
	if err != nil || classifyId == 0 {
		logger.Error("获取研报分类信息失败:%v", err)
		return
	}
	permissions, err := etaDao.GetSecondPermissionsByClassifyID(classifyId)
	if err != nil {
		logger.Error("获取研报二级品种信息失败:%v", err)
		return
	}
	for _, permission := range permissions {
		permissionDTOs = append(permissionDTOs, convertPermissionDTO(permission))
	}
	return
}
func (es ESReport) GetId() string {
	return strconv.Itoa(es.ReportID)
}
func GetETALatestReportId() (id int, err error) {
	return reportDao.GetLatestReportIdBySource(reportDao.SourceETA)
}

func GetHTLatestReportId() (id int, err error) {
	return reportDao.GetLatestReportIdBySource(reportDao.SourceHT)
}
func InitETAReportList(list []eta.ETAReport) (err error) {
	logger.Info("同步研报数量%d", len(list))
	var reports []reportDao.Report
	for _, etaRp := range list {
		//authorNames := strings.Split(etaRp.Author, ",")
		//authorNamesWithOutEmpty := stringUtils.RemoveEmptyStrings(authorNames)
		//for _, authorName := range authorNamesWithOutEmpty {
		var coverSrc int
		var permissions []etaDao.ChartPermission
		permissions, err = etaDao.GetSecondPermissionsByClassifyID(etaRp.ClassifyID)
		if err != nil || len(permissions) == 0 {
			logger.Error("获取研报二级品种信息失败:%v", err)
			coverSrc = 0
		} else {
			permissionsId := permissions[0].ChartPermissionID
			var ids []int
			ids, err = mediaDao.GetIdsByPermissionId(permissionsId)
			if err != nil {
				logger.Error("获取图片资源失败:%v", err)
			}
			if ids == nil || len(ids) == 0 {
				coverSrc = 0
			} else {
				src := rand.NewSource(time.Now().UnixNano())
				r := rand.New(src)
				// 从切片中随机选择一个元素
				randomIndex := r.Intn(len(ids))
				coverSrc = ids[randomIndex]
			}
		}
		destRp := convertEtaReport(etaRp)
		//destRp.Author = authorName
		destRp.CoverSrc = coverSrc
		reports = append(reports, destRp)
		//}
	}
	err = reportDao.BatchInsertReport(&reports)
	if err != nil {
		logger.Error("同步ETA研报失败:%v", err)
		return
	}

	return initES(reports)

}
func SyncETAReportList(list []eta.ETAReport) (err error) {
	logger.Info("同步研报数量%d", len(list))
	var reports []reportDao.Report
	for _, etaRp := range list {
		//authorNames := strings.Split(etaRp.Author, ",")
		//authorNamesWithOutEmpty := stringUtils.RemoveEmptyStrings(authorNames)
		//for _, authorName := range authorNamesWithOutEmpty {
		destRp := convertEtaReport(etaRp)
		//destRp.Author = authorName
		reports = append(reports, destRp)
	}
	//}
	esList, err := reportDao.InsertOrUpdateReport(reports, SourceETA)
	if esList == nil {
		return
	}
	return syncESAndSendMessage(esList)
}

type UpdateESReport struct {
	Title         string `json:"title"`
	Author        string `json:"author"`
	Abstract      string `json:"abstract"`
	PublishedTime string `json:"publishedTime"`
}

func syncESAndSendMessage(reports []reportDao.Report) (err error) {
	var esReports []es.ESBase
	for _, etaRp := range reports {
		esRp := convertEsReport(etaRp)
		esReports = append(esReports, esRp)
	}
	//同步es
	for _, report := range reports {
		update := UpdateESReport{
			Title:         report.Title,
			Author:        report.Author,
			PublishedTime: report.PublishedTime,
			Abstract:      report.Abstract,
		}
		success := elastic().Update(htConfig.GetReportIndex(), report.ID, update)
		if !success {
			logger.Error("更新es失败,reportId::%d,err:%v", report.ID, err)
		}
	}

	//err = elastic().BulkInsert(htConfig.GetReportIndex(), esReports)
	if err != nil {
		logger.Error("同步ETA研报到es失败:%v", err)
		return
	}
	//生产meta信息
	logger.Info("生成推送META信息")
	for _, report := range reports {
		userIds := userService.GetPostUser(report.Author, report.PublishedTime)
		var author analystService.FinancialAnalystDTO
		author, err = analystService.GetAnalystByName(report.Author)
		if err != nil {
			logger.Error("获取研报作者失败:%v", err)
			continue
		}
		if len(userIds) > 0 {
			usersStr := stringUtils.IntToStringSlice(userIds)
			Meta := userService.MetaData{
				AuthorName:    report.Author,
				AuthorId:      author.Id,
				SourceId:      report.ID,
				PublishedTime: report.PublishedTime,
			}
			metaStr, _ := json.Marshal(Meta)
			toStr := strings.Join(usersStr, ",")
			UUID := uuid.New()
			uuidStr := UUID.String()
			metaContent := userService.MetaInfoDTO{
				From:       "HT",
				Uid:        "report:" + uuidStr,
				Meta:       string(metaStr),
				MetaType:   "USER_NOTICE",
				SourceType: "REPORT",
				To:         toStr,
			}
			err = userService.CreateMetaInfo(metaContent)
			if err != nil {
				logger.Error("创建Meta信息失败:%v", err)
				return err
			}
		}
	}
	return
}

func initES(reports []reportDao.Report) (err error) {
	var esReports []es.ESBase
	for _, etaRp := range reports {
		esRp := convertEsReport(etaRp)
		esReports = append(esReports, esRp)
	}
	//同步es
	err = elastic().BulkInsert(htConfig.GetReportIndex(), esReports)
	if err != nil {
		logger.Error("同步ETA研报到es失败:%v", err)
		return
	}
	return
}
func InitHTReportList(list []ht.HTReport) (noRecord bool, err error) {

	var reports []reportDao.Report
	//获取系统中ht品种名
	permissions, err := reportDao.GetGLAuthorNames()
	if err != nil {
		logger.Error("获取钢联研报作者失败:%v", err)
		return
	}
	for _, htRp := range list {
		for _, permission := range permissions {
			if htRp.PermissionName == permission.Permission {
				if permission.AuthorNames != "" {
					htRp.PublishUserName = permission.AuthorNames
				}
				//authorNames := strings.Split(htRp.PublishUserName, ",")
				//authorNamesWithOutEmpty := stringUtils.RemoveEmptyStrings(authorNames)
				//for _, authorName := range authorNamesWithOutEmpty {
				destRp := convertHTReport(htRp)
				//destRp.Author = authorName
				var coverSrc int
				permissionId, err := etaDao.GetPermissionIdByName(htRp.PermissionName)
				if err != nil {
					logger.Error("HT获取eta品种id失败:%v", err)
					coverSrc = 0
				}
				ids, err := mediaDao.GetIdsByPermissionId(permissionId)
				if err != nil {
					logger.Error("获取图片资源失败:%v", err)
					coverSrc = 0
				}
				if ids == nil || len(ids) == 0 {
					coverSrc = 0
				} else {
					src := rand.NewSource(time.Now().UnixNano())
					r := rand.New(src)
					// 从切片中随机选择一个元素
					randomIndex := r.Intn(len(ids))
					coverSrc = ids[randomIndex]
				}
				destRp.CoverSrc = coverSrc
				destRp.PlateName = htRp.PermissionName
				reports = append(reports, destRp)
			}
			//}
		}
	}
	if len(reports) == 0 {
		return true, nil
	} else {
		logger.Info("同步研报数量%d", len(reports))
		err = reportDao.BatchInsertReport(&reports)
		if err != nil {
			logger.Error("同步HT研报失败:%v", err)
			return false, err
		}
		return false, initES(reports)
	}
}
func SyncHTReportList(list []ht.HTReport) (noRecord bool, err error) {
	var reports []reportDao.Report
	permissions, err := reportDao.GetGLAuthorNames()
	if err != nil {
		logger.Error("获取钢联研报作者失败:%v", err)
		return
	}
	for _, htRp := range list {
		for _, permission := range permissions {
			if htRp.PermissionName == permission.Permission {
				if permission.AuthorNames != "" {
					htRp.PublishUserName = permission.AuthorNames
				}
				//authorNames := strings.Split(htRp.PublishUserName, ",")
				//authorNamesWithOutEmpty := stringUtils.RemoveEmptyStrings(authorNames)
				//for _, authorName := range authorNamesWithOutEmpty {
				destRp := convertHTReport(htRp)
				//destRp.Author = authorName
				var coverSrc int
				permissionId, err := etaDao.GetPermissionIdByName(htRp.PermissionName)
				if err != nil {
					logger.Error("HT获取eta品种id失败:%v", err)
					coverSrc = 0
				}
				ids, err := mediaDao.GetIdsByPermissionId(permissionId)
				if err != nil {
					logger.Error("获取图片资源失败:%v", err)
					coverSrc = 0
				}
				if ids == nil || len(ids) == 0 {
					coverSrc = 0
				} else {
					src := rand.NewSource(time.Now().UnixNano())
					r := rand.New(src)
					// 从切片中随机选择一个元素
					randomIndex := r.Intn(len(ids))
					coverSrc = ids[randomIndex]
				}
				destRp.CoverSrc = coverSrc
				destRp.PlateName = htRp.PermissionName
				reports = append(reports, destRp)
			}
			//}
		}
	}
	if len(reports) == 0 {
		return true, nil
	} else {
		logger.Info("同步研报数量%d", len(list))
	}
	esList, err := reportDao.InsertOrUpdateReport(reports, SourceHT)
	if esList == nil {
		return false, err
	}
	return false, syncESAndSendMessage(reports)
}
func GetListOrderByConditionWeekly(week bool, column string, limit int, order models.Order) (dtoList []ReportDTO, err error) {
	reports, err := reportDao.GetListOrderByCondition(week, column, limit, order)
	if err != nil {
		logger.Error("获取研报失败:%v", err)
		return
	}
	for _, reportItem := range reports {
		dto := convertReportDTO(reportItem, false)
		dtoList = append(dtoList, dto)
	}
	return
}

func GetListByCondition[T any](column string, ids []T) (dtoList []ReportDTO, err error) {
	var values []interface{}
	for _, id := range ids {
		values = append(values, id)
	}
	reports, err := reportDao.GetListByCondition(column, ids)
	if err != nil {
		logger.Error("获取研报失败:%v", err)
		return
	}
	for _, reportItem := range reports {
		dto := convertReportDTO(reportItem, false)
		dtoList = append(dtoList, dto)
	}
	return
}

func GetTotalPageCountByPermissionIds(permissionIds []int) (total int64, latestId int64, ids map[string][]int) {
	htIds, err := GetHTReportIdsByPermissionIds(permissionIds)
	if err != nil {
		logger.Error("品种筛选ht报告id失败:%v", err)
		htIds = []int{}
	}
	etaIds, err := GetETAReportIdsByPermissionIds(permissionIds)
	if err != nil {
		logger.Error("品种筛选eta报告id失败:%v", err)
		etaIds = []int{}
	}
	total = int64(len(etaIds) + len(htIds))
	ids = make(map[string][]int, 2)
	if len(etaIds) == 0 {
		ids["ETA"] = []int{}
	} else {
		ids["ETA"] = etaIds
	}
	if len(htIds) == 0 {
		ids["HT"] = []int{}
	} else {
		ids["HT"] = htIds
	}
	//ids = append(etaIds, htIds...)
	latestId = reportDao.GetMaxIdByPermissionIds(ids)
	return
}
func convertEtaReport(etaRp eta.ETAReport) reportDao.Report {
	return reportDao.Report{
		OrgID:         etaRp.ID,
		Title:         etaRp.Title,
		Abstract:      etaRp.Abstract,
		Author:        etaRp.Author,
		CoverSrc:      0,
		PublishedTime: etaRp.PublishTime.Format(time.DateTime),
		Source:        reportDao.SourceETA,
		SendStatus:    reportDao.UNSEND,
		Status:        reportDao.StatusPublish,
	}
}

func convertHTReport(etaRp ht.HTReport) reportDao.Report {
	return reportDao.Report{
		OrgID:         etaRp.Id,
		Title:         etaRp.ReportName,
		Author:        etaRp.PublishUserName,
		PublishedTime: etaRp.PublishedTime,
		CoverSrc:      0,
		Source:        reportDao.SourceHT,
		SendStatus:    reportDao.UNSEND,
		Status:        reportDao.StatusPublish,
	}
}
func convertEsReport(report reportDao.Report) ESReport {
	return ESReport{
		ReportID:      report.ID,
		Title:         report.Title,
		OrgId:         report.OrgID,
		Author:        report.Author,
		Source:        report.Source,
		Abstract:      report.Abstract,
		Status:        report.Status,
		CoverSrc:      report.CoverSrc,
		PublishedTime: report.PublishedTime,
	}
}

func convertReportDTO(report reportDao.Report, fullTime bool) (reportDTO ReportDTO) {
	reportDTO = ReportDTO{
		ReportID:      report.ID,
		Title:         report.Title,
		OrgId:         report.OrgID,
		Author:        report.Author,
		Source:        string(report.Source),
		CoverSrc:      report.CoverSrc,
		Abstract:      report.Abstract,
		PublishedTime: report.PublishedTime,
	}
	publishDate, err := time.Parse(time.DateTime, reportDTO.PublishedTime)
	if err == nil && !fullTime {
		reportDTO.PublishedTime = publishDate.Format(time.DateOnly)
	}
	return
}

func matchAll(sorts []string, key string) (request *es.ESQueryRequest) {
	req := new(es.ESQueryRequest)
	return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, 0, 1, sorts, es.MatchAll)
}
func match(key string, from int, to int, sorts []string) (request *es.ESQueryRequest) {
	req := new(es.ESQueryRequest)
	return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, from, to, sorts, es.Match)
}
func matchRange(key string, from int, to int, max int64, sorts []string) (request *es.ESQueryRequest) {
	req := new(es.ESQueryRequest)
	return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, from, to, sorts, es.Range).Range(0, max, ESRangeColumn)
}