package report import ( "encoding/json" "eta/eta_mini_ht_api/common/component/es" logger "eta/eta_mini_ht_api/common/component/log" "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" reportDao "eta/eta_mini_ht_api/models/report" "github.com/google/uuid" "strconv" "strings" "time" ) const ( SourceETA = "ETA" SourceHT = "HT" DESC models.Order = "desc" ASC models.Order = "asc" ESIndex = "report_index" ESColumn = "title" ESRangeColumn = "reportId" ) var ( sortField = []string{"_score:desc"} ) 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 string `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"` Permissions map[int]string `json:"-"` PermissionNames interface{} `json:"permissionNames,omitempty"` Highlight []string `json:"highlight,omitempty"` Detail json.RawMessage `json:"detail,omitempty"` CoverSrc string `json:"coverSrc"` } 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 { ReportDTO = convertReportDTO(report) } 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) list = append(list, dto) } } return } func GetReportPageByOrgIds(pageInfo page.PageInfo, orgIds []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) 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) 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) 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 (es ESReport) GetId() string { return strconv.Itoa(es.ReportID) } func GetETALatestReportId() (id int, err error) { return reportDao.GetETALatestReportId() } func SyncETAReportList(list []eta.ETAReport) (err error) { logger.Info("同步研报数量%d", len(list)) var reports []reportDao.Report var esReports []es.ESBase 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) } } err = reportDao.BatchInsertReport(&reports) if err != nil { logger.Error("同步ETA研报失败:%v", err) return } for _, etaRp := range reports { esRp := convertEsReport(etaRp) esReports = append(esReports, esRp) } //同步es err = elastic().BulkInsert(ESIndex, 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) author, _ := analystService.GetAnalystByName(report.Author) 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: "ETA", 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 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 _, report := range reports { dto := convertReportDTO(report) 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 _, report := range reports { dto := convertReportDTO(report) dtoList = append(dtoList, dto) } return } func GetTotalPageCountByPermissionIds(permissionIds []int) (total int64, latestId int64, ids []int) { //TODO 一期品种筛选reportIds htIds := []int{} etaIds, err := GetETAReportIdsByPermissionIds(permissionIds) if err != nil { logger.Error("品种筛选eta报告id失败:%v", err) etaIds = []int{} } total = int64(len(etaIds) + len(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, PublishedTime: etaRp.PublishTime, Source: reportDao.SourceETA, Status: reportDao.StatusInit, } } 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) (reportDTO ReportDTO) { reportDTO = ReportDTO{ ReportID: report.ID, Title: report.Title, OrgId: report.OrgID, Author: report.Author, Source: string(report.Source), Abstract: report.Abstract, PublishedTime: report.PublishedTime, } publishDate, err := time.Parse(time.DateTime, reportDTO.PublishedTime) if err == nil { reportDTO.PublishedTime = publishDate.Format(time.DateOnly) } return } func matchAll(sorts []string, key string) (request *es.ESQueryRequest) { req := new(es.ESQueryRequest) return req.CreateESQueryRequest(ESIndex, 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(ESIndex, 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(ESIndex, ESColumn, key, from, to, sorts, es.Range).Range(0, max, ESRangeColumn) }