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" userDao "eta/eta_mini_ht_api/models/user" "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, latestId int64, err error) { 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, searchAll bool) (list []ReportDTO, err error) { offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize) reports, err := reportDao.GetReportPageByOrgIds(pageInfo.LatestId, pageInfo.PageSize, offset, orgIds, searchAll) 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, reportDao.StatusPublish) //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 etaStatus(status int) reportDao.ReportStatus { if status == etaDao.Passed || status == etaDao.Published { return reportDao.StatusPublish } else { return reportDao.StatusUnPublish } } func SyncETAReportList(list []eta.ETAReport) (err error) { logger.Info("同步研报数量%d", len(list)) var reports []reportDao.Report for _, etaRp := range list { 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] } } //authorNames := strings.Split(etaRp.Author, ",") //authorNamesWithOutEmpty := stringUtils.RemoveEmptyStrings(authorNames) //for _, authorName := range authorNamesWithOutEmpty { status := etaStatus(etaRp.State) destRp := convertEtaReport(etaRp, status) destRp.CoverSrc = coverSrc 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"` Status string `json:"status"` } 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 { var exist bool exist, err = elastic().Exist(htConfig.GetReportIndex(), report.ID) if err != nil { logger.Error("查询es失败,reportId::%d,err:%v", report.ID, err) } if exist { update := UpdateESReport{ Title: report.Title, Author: report.Author, PublishedTime: report.PublishedTime, Abstract: report.Abstract, Status: string(report.Status), } success := elastic().Update(htConfig.GetReportIndex(), report.ID, update) if !success { logger.Error("更新es失败,reportId::%d,err:%v", report.ID, err) } if report.Status == reportDao.StatusUnPublish { //隐藏热度搜索 err = userDao.HiddenFlows(report.ID, userDao.ReportSourceType) if err != nil { logger.Error("隐藏热度搜索失败,reportId::%d,err:%v", report.ID, err) } } } else { insert := ESReport{ ReportID: report.ID, OrgId: report.OrgID, Title: report.Title, Author: report.Author, Source: report.Source, Abstract: report.Abstract, CoverSrc: report.CoverSrc, Status: report.Status, PublishedTime: report.PublishedTime, } success := elastic().CreateDocument(htConfig.GetReportIndex(), report.ID, insert) 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, reportDao.StatusPublish) //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 htStatus(status int, isDelete int) reportDao.ReportStatus { if isDelete == 1 || status != ht.Publish { return reportDao.StatusUnPublish } return reportDao.StatusUnPublish } 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 { status := htStatus(htRp.Status, htRp.IsDelete) destRp := convertHTReport(htRp, status) //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) { htOrgIds, err := GetHTReportIdsByPermissionIds(permissionIds) if err != nil { logger.Error("品种筛选ht报告id失败:%v", err) htOrgIds = []int{} } etaOrgIds, err := GetETAReportIdsByPermissionIds(permissionIds) if err != nil { logger.Error("品种筛选eta报告id失败:%v", err) etaOrgIds = []int{} } totalCol := int64(len(etaOrgIds) + len(htOrgIds)) if totalCol == 0 { latestId = 0 return } ids = make(map[string][]int, 2) if len(etaOrgIds) == 0 { ids["ETA"] = []int{} } else { ids["ETA"] = etaOrgIds } if len(htOrgIds) == 0 { ids["HT"] = []int{} } else { ids["HT"] = htOrgIds } //ids = append(etaIds, htIds...) total, latestId, err = reportDao.GetMaxIdByPermissionIds(ids) if err != nil { logger.Error("获取筛选报告的最大记录和记录数失败:%v", err) return } return } func convertEtaReport(etaRp eta.ETAReport, status reportDao.ReportStatus) 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: status, } } func convertHTReport(etaRp ht.HTReport, status reportDao.ReportStatus) 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.MatchAllByCondition).ByCondition("status", "PUBLISH") } 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.MatchAllByCondition).ByCondition("status", "PUBLISH") } 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.RangeByCondition).Range(0, max, ESRangeColumn).ByCondition("status", "PUBLISH") }