report_service.go 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957
  1. package report
  2. import (
  3. "encoding/json"
  4. "eta/eta_mini_ht_api/common/component/config"
  5. "eta/eta_mini_ht_api/common/component/es"
  6. logger "eta/eta_mini_ht_api/common/component/log"
  7. "eta/eta_mini_ht_api/common/contants"
  8. "eta/eta_mini_ht_api/common/utils/page"
  9. stringUtils "eta/eta_mini_ht_api/common/utils/string"
  10. configService "eta/eta_mini_ht_api/domian/config"
  11. analystService "eta/eta_mini_ht_api/domian/financial_analyst"
  12. userService "eta/eta_mini_ht_api/domian/user"
  13. "eta/eta_mini_ht_api/models"
  14. permissionDao "eta/eta_mini_ht_api/models/config"
  15. "eta/eta_mini_ht_api/models/eta"
  16. etaDao "eta/eta_mini_ht_api/models/eta"
  17. "eta/eta_mini_ht_api/models/ht"
  18. mediaDao "eta/eta_mini_ht_api/models/image"
  19. reportDao "eta/eta_mini_ht_api/models/report"
  20. userDao "eta/eta_mini_ht_api/models/user"
  21. "math/rand"
  22. "strconv"
  23. "strings"
  24. "time"
  25. )
  26. const (
  27. SourceETA = "ETA"
  28. SourceHT = "HT"
  29. DESC models.Order = "desc"
  30. ASC models.Order = "asc"
  31. ESColumn = "title"
  32. ESRangeColumn = "reportId"
  33. )
  34. var (
  35. sortField = []string{"_score:desc"}
  36. htConfig = config.GetConfig(contants.HT).(*config.HTBizConfig)
  37. )
  38. func elastic() *es.ESClient {
  39. return es.GetInstance()
  40. }
  41. // ESReport Report ES研报mapping
  42. type ESReport struct {
  43. ReportID int `json:"reportId"`
  44. OrgId int `json:"orgId"`
  45. Title string `json:"title"`
  46. Author string `json:"author"`
  47. Source reportDao.ReportSource `json:"source"`
  48. Abstract string `json:"abstract"`
  49. CoverSrc int `json:"coverSrc"`
  50. Status reportDao.ReportStatus `json:"status"`
  51. PublishedTime string `json:"publishedTime"`
  52. }
  53. type ReportDTO struct {
  54. ReportID int `json:"reportId"`
  55. OrgId int `json:"orgId"`
  56. Title string `json:"title"`
  57. Author string `json:"author"`
  58. AuthorInfo []Anthor `json:"authorInfo"`
  59. Source string `json:"source"`
  60. Abstract string `json:"abstract"`
  61. PublishedTime string `json:"publishedTime"`
  62. RiskLevel string `json:"riskLevel"`
  63. PlateName string `json:"-"`
  64. ClassifyId int `json:"-"`
  65. SecondPermission map[int]string `json:"-"`
  66. Permissions map[int]string `json:"-"`
  67. PermissionNames interface{} `json:"permissionNames"`
  68. Highlight []string `json:"highlight"`
  69. Detail json.RawMessage `json:"detail"`
  70. PdfUrl string `json:"pdfUrl"`
  71. CoverSrc int `json:"coverSrc"`
  72. CoverUrl string `json:"coverUrl"`
  73. Login bool `json:"login"`
  74. RiskLevelStatus string `json:"riskLevelStatus"`
  75. IsFree bool `json:"isFree"`
  76. IsSubscribe bool `json:"isSubscribe"`
  77. Price string `json:"price"`
  78. ProductId int `json:"productId"`
  79. IsPackage bool `json:"isPackage"`
  80. }
  81. type Detail struct {
  82. }
  83. type Anthor struct {
  84. Id int `json:"id"`
  85. Name string `json:"name"`
  86. HeadImgUrl string `json:"headImgUrl"`
  87. }
  88. func GetReportById(reportId int) (ReportDTO ReportDTO, err error) {
  89. report, err := reportDao.GetReportById(reportId)
  90. if err != nil {
  91. return
  92. }
  93. ReportDTO = convertReportDTO(report, true)
  94. authorStr := ReportDTO.Author
  95. authorNames := strings.Split(authorStr, ",")
  96. authorNames = stringUtils.RemoveEmptyStrings(authorNames)
  97. var authorList []Anthor
  98. if len(authorNames) > 0 {
  99. for _, name := range authorNames {
  100. var author analystService.FinancialAnalystDTO
  101. author, err = analystService.GetAnalystByName(name)
  102. var item Anthor
  103. if err != nil {
  104. item = Anthor{
  105. Id: 0,
  106. Name: name,
  107. HeadImgUrl: "",
  108. }
  109. } else {
  110. item = Anthor{
  111. Id: author.Id,
  112. Name: author.Name,
  113. HeadImgUrl: author.HeadImgUrl,
  114. }
  115. }
  116. authorList = append(authorList, item)
  117. }
  118. }
  119. ReportDTO.AuthorInfo = authorList
  120. return
  121. }
  122. func GetTotalPageCount() (total int64, latestId int64, err error) {
  123. return reportDao.GetTotalPageCount()
  124. }
  125. func GetTotalPageCountByAnalyst(analyst string, permissionIds []int, riskLevel string) (total int64, latestId int64, ids []int) {
  126. ids, err := reportDao.GetReportsByAnalyst(analyst)
  127. if err != nil {
  128. logger.Error("查询研究研报告列表id失败:%v", err)
  129. return
  130. }
  131. //查询这些包含在列表中的权限的报告ids
  132. htOrgIds, err := GetHTReportIdsByPermissionIds(permissionIds)
  133. if err != nil {
  134. logger.Error("品种筛选ht报告id失败:%v", err)
  135. htOrgIds = []int{}
  136. }
  137. etaOrgIds, err := GetETAReportIdsByPermissionIds(permissionIds)
  138. if err != nil {
  139. logger.Error("品种筛选eta报告id失败:%v", err)
  140. etaOrgIds = []int{}
  141. }
  142. if len(etaOrgIds) == 0 && len(htOrgIds) == 0 {
  143. logger.Info("没有符合权限的研报")
  144. return
  145. }
  146. orgIds := make(map[string][]int, 2)
  147. if len(etaOrgIds) == 0 {
  148. orgIds["ETA"] = []int{}
  149. } else {
  150. orgIds["ETA"] = etaOrgIds
  151. }
  152. if len(htOrgIds) == 0 {
  153. orgIds["HT"] = []int{}
  154. } else {
  155. orgIds["HT"] = htOrgIds
  156. }
  157. permitReportIds, err := reportDao.GetReportIdListByOrgIds(orgIds)
  158. if err != nil {
  159. logger.Error("根据原始报告id获取报告id列表失败:%v", err)
  160. return
  161. }
  162. var filterReportIds []int
  163. for _, id := range ids {
  164. for _, permitReportId := range permitReportIds {
  165. if id == permitReportId {
  166. filterReportIds = append(filterReportIds, id)
  167. }
  168. }
  169. }
  170. if len(filterReportIds) == 0 {
  171. logger.Info("没有符合权限的研究员研报")
  172. return
  173. }
  174. ids = filterReportIds
  175. total = int64(len(filterReportIds))
  176. latestId = int64(findMax(filterReportIds))
  177. return
  178. }
  179. // findMaxWithError 函数用于找到整型数组中的最大值,并返回错误信息
  180. func findMax(nums []int) (max int) {
  181. if len(nums) == 0 {
  182. return 0
  183. }
  184. // 初始化最大值为数组的第一个元素
  185. max = nums[0]
  186. // 遍历数组,找到最大值
  187. for _, num := range nums {
  188. if num > max {
  189. max = num
  190. }
  191. }
  192. return
  193. }
  194. func SearchMaxReportId(key string) (total int64, reportId int64) {
  195. sort := []string{"reportId:desc"}
  196. request := matchAll(sort, key)
  197. //同步es
  198. re, err := elastic().Count(request)
  199. if err != nil {
  200. logger.Error("es搜索异常:%v", err)
  201. }
  202. count := re.Count
  203. total = int64(count)
  204. if total > 0 {
  205. request = match(key, 0, count, sort)
  206. re, err = elastic().Search(request)
  207. if err != nil {
  208. logger.Error("es搜索异常:%v", err)
  209. }
  210. hits := elastic().GetSource(re.Hits)
  211. data := hits[0].Source
  212. report := ReportDTO{}
  213. err = json.Unmarshal(data, &report)
  214. if err != nil {
  215. logger.Error("获取当前最大研报id失败:%v", err)
  216. return
  217. }
  218. reportId = int64(report.ReportID)
  219. }
  220. return
  221. }
  222. func SearchMaxReportIdWithRange(key string, reportIds []int) (total int64) {
  223. sort := []string{"reportId:desc"}
  224. var docIds []string
  225. for _, reportId := range reportIds {
  226. docIds = append(docIds, strconv.Itoa(reportId))
  227. }
  228. request := CountByDocId(key, sort, docIds)
  229. //同步es
  230. re, err := elastic().Count(request)
  231. if err != nil {
  232. logger.Error("es搜索异常:%v", err)
  233. }
  234. count := re.Count
  235. total = int64(count)
  236. if total > 0 {
  237. request = match(key, 0, count, sort)
  238. re, err = elastic().Search(request)
  239. if err != nil {
  240. logger.Error("es搜索异常:%v", err)
  241. }
  242. hits := elastic().GetSource(re.Hits)
  243. data := hits[0].Source
  244. report := ReportDTO{}
  245. err = json.Unmarshal(data, &report)
  246. if err != nil {
  247. logger.Error("获取当前最大研报id失败:%v", err)
  248. return
  249. }
  250. }
  251. return
  252. }
  253. func SearchReportList(key string, ids []int, from int, size int, max int64) (reports []ReportDTO, err error) {
  254. //同步es
  255. var docIds []string
  256. for _, id := range ids {
  257. docIds = append(docIds, strconv.Itoa(id))
  258. }
  259. sorts := append(sortField, "publishedTime:desc")
  260. request := matchRangeByDocId(key, from, size, max, sorts, docIds)
  261. re, err := elastic().Search(request)
  262. if err != nil {
  263. logger.Error("es搜索异常:%v", err)
  264. }
  265. hits := elastic().GetSource(re.Hits)
  266. if len(hits) == 0 {
  267. reports = []ReportDTO{}
  268. return
  269. }
  270. for _, hit := range hits {
  271. var content map[string][]string
  272. err = json.Unmarshal(hit.Highlight, &content)
  273. report := ReportDTO{}
  274. err = json.Unmarshal(hit.Source, &report)
  275. if err != nil {
  276. logger.Error("解析研报数据失败:%v", err)
  277. continue
  278. }
  279. report.Highlight = content[ESColumn]
  280. report.Title = report.Highlight[0]
  281. report.PublishedTime = report.PublishedTime[:10]
  282. reports = append(reports, report)
  283. }
  284. return
  285. }
  286. func GetReportPageByAnalyst(pageInfo page.PageInfo, analyst string, reportIds []int) (list []ReportDTO, err error) {
  287. offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize)
  288. reports, err := reportDao.GetReportPageByAnalyst(pageInfo.LatestId, pageInfo.PageSize, offset, analyst, reportIds)
  289. if err != nil {
  290. logger.Error("分页查询报告列表失败:%v", err)
  291. return
  292. }
  293. list = make([]ReportDTO, 0)
  294. if reports != nil {
  295. for _, report := range reports {
  296. dto := convertReportDTO(report, false)
  297. list = append(list, dto)
  298. }
  299. }
  300. return
  301. }
  302. func GetReportPageByOrgIds(pageInfo page.PageInfo, orgIds map[string][]int, searchAll bool) (list []ReportDTO, err error) {
  303. offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize)
  304. reports, err := reportDao.GetReportPageByOrgIds(pageInfo.LatestId, pageInfo.PageSize, offset, orgIds, searchAll)
  305. if err != nil {
  306. logger.Error("分页查询报告列表失败:%v", err)
  307. return
  308. }
  309. list = make([]ReportDTO, 0)
  310. if reports != nil {
  311. for _, report := range reports {
  312. dto := convertReportDTO(report, false)
  313. list = append(list, dto)
  314. }
  315. }
  316. return
  317. }
  318. func GetNewReportByPublishTime(time time.Time) (reports []ReportDTO) {
  319. list := reportDao.GetNewReportByPublishTime(time)
  320. if list != nil {
  321. for _, report := range list {
  322. dto := convertReportDTO(report, false)
  323. reports = append(reports, dto)
  324. }
  325. }
  326. return
  327. }
  328. func GetReportPage(pageInfo page.PageInfo) (list []ReportDTO, err error) {
  329. offset := page.StartIndex(pageInfo.Current, pageInfo.PageSize)
  330. reports, err := reportDao.GetReportPage(pageInfo.LatestId, pageInfo.PageSize, offset)
  331. if err != nil {
  332. logger.Error("分页查询报告列表失败:%v", err)
  333. return
  334. }
  335. list = make([]ReportDTO, 0)
  336. if reports != nil {
  337. for _, report := range reports {
  338. dto := convertReportDTO(report, false)
  339. list = append(list, dto)
  340. }
  341. }
  342. return
  343. }
  344. func getETAReportFirstPermissions(id int) (permissionDTOs []configService.PermissionDTO) {
  345. classifyId, err := etaDao.GetReportClassifyById(id)
  346. if err != nil || classifyId == 0 {
  347. logger.Error("获取研报分类信息失败:%v", err)
  348. return
  349. }
  350. permissions, err := permissionDao.GetFirstPermissionsByClassifyID(classifyId)
  351. //permissions, err := etaDao.GetFirstPermissionsByClassifyID(classifyId)
  352. if err != nil {
  353. logger.Error("获取研报一级品种信息失败:%v", err)
  354. return
  355. }
  356. for _, permission := range permissions {
  357. permissionDTOs = append(permissionDTOs, convertPermissionDTO(permission))
  358. }
  359. return
  360. }
  361. func getHTReportFirstPermissions(id int) (permissionDTOs []configService.PermissionDTO) {
  362. report, err := reportDao.GetReportByOrgId(id, SourceHT)
  363. if err != nil {
  364. logger.Error("获取报告失败:%v", err)
  365. }
  366. permissionName := report.PlateName
  367. plateName, err := reportDao.GetPlateNameByPermissionName(permissionName)
  368. if err != nil {
  369. return []configService.PermissionDTO{}
  370. }
  371. return []configService.PermissionDTO{
  372. {
  373. PermissionId: 0,
  374. PermissionName: plateName,
  375. ParentId: 0,
  376. },
  377. }
  378. }
  379. func getETAReportSecondPermissions(id int) (permissionDTOs []configService.PermissionDTO) {
  380. classifyId, err := reportDao.GetReportClassifyById(id)
  381. //classifyId, err := etaDao.GetReportClassifyById(id)
  382. if err != nil || classifyId == 0 {
  383. logger.Error("获取研报分类信息失败:%v", err)
  384. return
  385. }
  386. permissions, err := permissionDao.GetSecondPermissionsByClassifyID(classifyId)
  387. //permissions, err := eta.GetSecondPermissionsByClassifyID(classifyId)
  388. if err != nil {
  389. logger.Error("获取研报二级品种信息失败:%v", err)
  390. return
  391. }
  392. for _, permission := range permissions {
  393. permissionDTOs = append(permissionDTOs, convertPermissionDTO(permission))
  394. }
  395. return
  396. }
  397. func getHTReportSecondPermissions(id int) (permissionDTOs []configService.PermissionDTO) {
  398. report, err := reportDao.GetReportByOrgId(id, SourceHT)
  399. if err != nil {
  400. logger.Error("获取报告失败:%v", err)
  401. }
  402. var permission permissionDao.Permission
  403. if report.PlateName != "" {
  404. permission, err = permissionDao.GetPermissionByName(report.PlateName)
  405. if err != nil {
  406. logger.Error("获取品种信息失败:%v", err)
  407. }
  408. }
  409. return []configService.PermissionDTO{
  410. {
  411. PermissionId: permission.PermissionId,
  412. PermissionName: report.PlateName,
  413. ParentId: permission.ParentId,
  414. RiskLevel: permission.RiskLevel,
  415. },
  416. }
  417. }
  418. func (es ESReport) GetId() string {
  419. return strconv.Itoa(es.ReportID)
  420. }
  421. func GetETALatestReportId() (id int, err error) {
  422. return reportDao.GetLatestReportIdBySource(reportDao.SourceETA)
  423. }
  424. func GetHTLatestReportId() (id int, err error) {
  425. return reportDao.GetLatestReportIdBySource(reportDao.SourceHT)
  426. }
  427. func InitETAReportList(list []eta.ETAReport) (err error) {
  428. logger.Info("同步研报数量%d", len(list))
  429. var reports []reportDao.Report
  430. for _, etaRp := range list {
  431. var coverSrc int
  432. //var permissions []etaDao.ChartPermission
  433. //permissions, err = etaDao.GetSecondPermissionsByClassifyID(etaRp.ClassifyID)
  434. var permissions []permissionDao.Permission
  435. permissions, err = permissionDao.GetSecondPermissionsByClassifyID(etaRp.ClassifyID)
  436. if err != nil || len(permissions) == 0 {
  437. logger.Error("获取研报二级品种信息失败:%v", err)
  438. coverSrc = 0
  439. } else {
  440. coverSrc = 0
  441. for _, permission := range permissions {
  442. permissionsId := permission.PermissionId
  443. var ids []int
  444. ids, err = mediaDao.GetIdsByPermissionId(permissionsId)
  445. if err != nil {
  446. logger.Error("获取图片资源失败:%v", err)
  447. continue
  448. }
  449. if ids == nil || len(ids) == 0 {
  450. continue
  451. }
  452. src := rand.NewSource(time.Now().UnixNano())
  453. r := rand.New(src)
  454. // 从切片中随机选择一个元素
  455. randomIndex := r.Intn(len(ids))
  456. coverSrc = ids[randomIndex]
  457. break
  458. }
  459. }
  460. destRp := convertEtaReport(etaRp, reportDao.StatusPublish)
  461. //destRp.Author = authorName
  462. destRp.CoverSrc = coverSrc
  463. reports = append(reports, destRp)
  464. //}
  465. }
  466. err = reportDao.BatchInsertReport(&reports)
  467. if err != nil {
  468. logger.Error("同步ETA研报失败:%v", err)
  469. return
  470. }
  471. return initES(reports)
  472. }
  473. func etaStatus(status int) reportDao.ReportStatus {
  474. if status == etaDao.Passed || status == etaDao.Published {
  475. return reportDao.StatusPublish
  476. } else {
  477. return reportDao.StatusUnPublish
  478. }
  479. }
  480. func SyncETAReportList(list []eta.ETAReport) (err error) {
  481. logger.Info("同步研报数量%d", len(list))
  482. var reports []reportDao.Report
  483. for _, etaRp := range list {
  484. var coverSrc int
  485. //var permissions []etaDao.ChartPermission
  486. //permissions, err = etaDao.GetSecondPermissionsByClassifyID(etaRp.ClassifyID)
  487. var permissions []permissionDao.Permission
  488. permissions, err = permissionDao.GetSecondPermissionsByClassifyID(etaRp.ClassifyID)
  489. if err != nil || len(permissions) == 0 {
  490. logger.Error("获取研报二级品种信息失败:%v", err)
  491. coverSrc = 0
  492. } else {
  493. coverSrc = 0
  494. for _, permission := range permissions {
  495. permissionsId := permission.PermissionId
  496. var ids []int
  497. ids, err = mediaDao.GetIdsByPermissionId(permissionsId)
  498. if err != nil {
  499. logger.Error("获取图片资源失败:%v", err)
  500. continue
  501. }
  502. if ids == nil || len(ids) == 0 {
  503. continue
  504. }
  505. src := rand.NewSource(time.Now().UnixNano())
  506. r := rand.New(src)
  507. // 从切片中随机选择一个元素
  508. randomIndex := r.Intn(len(ids))
  509. coverSrc = ids[randomIndex]
  510. break
  511. }
  512. }
  513. status := etaStatus(etaRp.State)
  514. destRp := convertEtaReport(etaRp, status)
  515. destRp.CoverSrc = coverSrc
  516. reports = append(reports, destRp)
  517. }
  518. esList, err := reportDao.InsertOrUpdateReport(reports, SourceETA)
  519. if esList == nil {
  520. return
  521. }
  522. return syncESAndSendMessage(esList)
  523. }
  524. type UpdateESReport struct {
  525. Title string `json:"title"`
  526. Author string `json:"author"`
  527. Abstract string `json:"abstract"`
  528. PublishedTime string `json:"publishedTime"`
  529. Status string `json:"status"`
  530. }
  531. func syncESAndSendMessage(reports []reportDao.Report) (err error) {
  532. var esReports []es.ESBase
  533. for _, etaRp := range reports {
  534. esRp := convertEsReport(etaRp)
  535. esReports = append(esReports, esRp)
  536. }
  537. //同步es
  538. for _, report := range reports {
  539. var exist bool
  540. exist, err = elastic().Exist(htConfig.GetReportIndex(), report.ID)
  541. if err != nil {
  542. logger.Error("查询es失败,reportId::%d,err:%v", report.ID, err)
  543. }
  544. if exist {
  545. update := UpdateESReport{
  546. Title: report.Title,
  547. Author: report.Author,
  548. PublishedTime: report.PublishedTime,
  549. Abstract: report.Abstract,
  550. Status: string(report.Status),
  551. }
  552. success := elastic().Update(htConfig.GetReportIndex(), report.ID, update)
  553. if !success {
  554. logger.Error("更新es失败,reportId::%d,err:%v", report.ID, err)
  555. }
  556. if report.Status == reportDao.StatusUnPublish {
  557. //隐藏热度搜索
  558. err = userDao.HiddenFlows(report.ID, userDao.ReportSourceType)
  559. if err != nil {
  560. logger.Error("隐藏热度搜索失败,reportId::%d,err:%v", report.ID, err)
  561. }
  562. } else {
  563. err = userDao.ShowFlows(report.ID, userDao.ReportSourceType)
  564. if err != nil {
  565. logger.Error("重置热度搜索失败,reportId::%d,err:%v", report.ID, err)
  566. }
  567. }
  568. } else {
  569. insert := ESReport{
  570. ReportID: report.ID,
  571. OrgId: report.OrgID,
  572. Title: report.Title,
  573. Author: report.Author,
  574. Source: report.Source,
  575. Abstract: report.Abstract,
  576. CoverSrc: report.CoverSrc,
  577. Status: report.Status,
  578. PublishedTime: report.PublishedTime,
  579. }
  580. success := elastic().CreateDocument(htConfig.GetReportIndex(), report.ID, insert)
  581. if !success {
  582. logger.Error("创建es文档失败,reportId::%d,err:%v", report.ID, err)
  583. }
  584. }
  585. }
  586. //err = elastic().BulkInsert(htConfig.GetReportIndex(), esReports)
  587. if err != nil {
  588. logger.Error("同步ETA研报到es失败:%v", err)
  589. return
  590. }
  591. //生产meta信息
  592. logger.Info("生成推送META信息")
  593. for _, report := range reports {
  594. if report.Status == reportDao.StatusUnPublish {
  595. logger.Info("报告取消发布,不需要生成推送消息")
  596. continue
  597. }
  598. var From string
  599. switch report.Source {
  600. case SourceETA:
  601. From = "ETA"
  602. case SourceHT:
  603. From = "HT"
  604. default:
  605. From = "UNKNOWN"
  606. }
  607. authors := strings.Split(report.Author, ",")
  608. authors = stringUtils.RemoveEmptyStrings(authors)
  609. if len(authors) > 0 {
  610. for _, authorName := range authors {
  611. userIds := userService.GetPostUser(authorName, report.PublishedTime)
  612. if len(userIds) > 0 {
  613. logger.Info("推送META信息,用户ID:%v", userIds)
  614. var author analystService.FinancialAnalystDTO
  615. author, err = analystService.GetAnalystByName(authorName)
  616. if err != nil {
  617. logger.Error("获取研报作者失败:%v", err)
  618. continue
  619. }
  620. usersStr := stringUtils.IntToStringSlice(userIds)
  621. Meta := userService.MetaData{
  622. AuthorName: author.Name,
  623. AuthorId: author.Id,
  624. SourceId: report.ID,
  625. PublishedTime: report.PublishedTime,
  626. }
  627. metaStr, _ := json.Marshal(Meta)
  628. toStr := strings.Join(usersStr, ",")
  629. metaContent := userService.MetaInfoDTO{
  630. From: From,
  631. Meta: string(metaStr),
  632. MetaType: "USER_NOTICE",
  633. SourceType: "REPORT",
  634. To: toStr,
  635. }
  636. err = userService.CreateMetaInfo(metaContent)
  637. if err != nil {
  638. logger.Error("创建Meta信息失败:%v", err)
  639. return err
  640. }
  641. }
  642. }
  643. }
  644. }
  645. return
  646. }
  647. func initES(reports []reportDao.Report) (err error) {
  648. var esReports []es.ESBase
  649. for _, etaRp := range reports {
  650. esRp := convertEsReport(etaRp)
  651. esReports = append(esReports, esRp)
  652. }
  653. //同步es
  654. err = elastic().BulkInsert(htConfig.GetReportIndex(), esReports)
  655. if err != nil {
  656. logger.Error("同步ETA研报到es失败:%v", err)
  657. return
  658. }
  659. return
  660. }
  661. func InitHTReportList(list []ht.HTReport) (noRecord bool, err error) {
  662. var reports []reportDao.Report
  663. //获取系统中ht品种名
  664. permissions, err := reportDao.GetGLAuthorNames()
  665. if err != nil {
  666. logger.Error("获取钢联研报作者失败:%v", err)
  667. return
  668. }
  669. for _, htRp := range list {
  670. for _, permission := range permissions {
  671. if htRp.PermissionName == permission.Permission {
  672. if permission.AuthorNames != "" {
  673. htRp.PublishUserName = permission.AuthorNames
  674. }
  675. destRp := convertHTReport(htRp, reportDao.StatusPublish)
  676. var coverSrc int
  677. var permissionId int
  678. permissionId, err = permissionDao.GetPermissionIdByName(htRp.PermissionName)
  679. //permissionId, err = etaDao.GetPermissionIdByName(htRp.PermissionName)
  680. if err != nil {
  681. logger.Error("HT获取eta品种id失败:%v", err)
  682. coverSrc = 0
  683. }
  684. var ids []int
  685. ids, err = mediaDao.GetIdsByPermissionId(permissionId)
  686. if err != nil {
  687. logger.Error("获取图片资源失败:%v", err)
  688. coverSrc = 0
  689. }
  690. if ids == nil || len(ids) == 0 {
  691. coverSrc = 0
  692. } else {
  693. src := rand.NewSource(time.Now().UnixNano())
  694. r := rand.New(src)
  695. // 从切片中随机选择一个元素
  696. randomIndex := r.Intn(len(ids))
  697. coverSrc = ids[randomIndex]
  698. }
  699. destRp.CoverSrc = coverSrc
  700. destRp.PlateName = htRp.PermissionName
  701. reports = append(reports, destRp)
  702. }
  703. //}
  704. }
  705. }
  706. if len(reports) == 0 {
  707. return true, nil
  708. } else {
  709. logger.Info("同步研报数量%d", len(reports))
  710. err = reportDao.BatchInsertReport(&reports)
  711. if err != nil {
  712. logger.Error("同步HT研报失败:%v", err)
  713. return false, err
  714. }
  715. return false, initES(reports)
  716. }
  717. }
  718. func htStatus(status int, isDelete int) reportDao.ReportStatus {
  719. if isDelete == 1 || status != ht.Publish {
  720. return reportDao.StatusUnPublish
  721. }
  722. return reportDao.StatusPublish
  723. }
  724. func SyncHTReportList(list []ht.HTReport) (noRecord bool, err error) {
  725. var reports []reportDao.Report
  726. permissions, err := reportDao.GetGLAuthorNames()
  727. if err != nil {
  728. logger.Error("获取钢联研报作者失败:%v", err)
  729. return
  730. }
  731. for _, htRp := range list {
  732. for _, permission := range permissions {
  733. if htRp.PermissionName == permission.Permission {
  734. if permission.AuthorNames != "" {
  735. htRp.PublishUserName = permission.AuthorNames
  736. }
  737. status := htStatus(htRp.Status, htRp.IsDelete)
  738. destRp := convertHTReport(htRp, status)
  739. var coverSrc int
  740. var permissionId int
  741. permissionId, err = permissionDao.GetPermissionIdByName(htRp.PermissionName)
  742. //permissionId, err = etaDao.GetPermissionIdByName(htRp.PermissionName)
  743. if err != nil {
  744. logger.Error("HT获取eta品种id失败:%v", err)
  745. coverSrc = 0
  746. }
  747. var ids []int
  748. ids, err = mediaDao.GetIdsByPermissionId(permissionId)
  749. if err != nil {
  750. logger.Error("获取图片资源失败:%v", err)
  751. coverSrc = 0
  752. }
  753. if ids == nil || len(ids) == 0 {
  754. coverSrc = 0
  755. } else {
  756. src := rand.NewSource(time.Now().UnixNano())
  757. r := rand.New(src)
  758. // 从切片中随机选择一个元素
  759. randomIndex := r.Intn(len(ids))
  760. coverSrc = ids[randomIndex]
  761. }
  762. destRp.CoverSrc = coverSrc
  763. destRp.PlateName = htRp.PermissionName
  764. reports = append(reports, destRp)
  765. }
  766. //}
  767. }
  768. }
  769. if len(reports) == 0 {
  770. return true, nil
  771. } else {
  772. logger.Info("同步研报数量%d", len(list))
  773. }
  774. esList, err := reportDao.InsertOrUpdateReport(reports, SourceHT)
  775. if esList == nil {
  776. return false, err
  777. }
  778. return false, syncESAndSendMessage(reports)
  779. }
  780. func GetListOrderByConditionWeekly(week bool, column string, limit int, order models.Order) (dtoList []ReportDTO, err error) {
  781. reports, err := reportDao.GetListOrderByCondition(week, column, limit, order)
  782. if err != nil {
  783. logger.Error("获取研报失败:%v", err)
  784. return
  785. }
  786. for _, reportItem := range reports {
  787. dto := convertReportDTO(reportItem, false)
  788. dtoList = append(dtoList, dto)
  789. }
  790. return
  791. }
  792. func GetListByCondition[T any](column string, ids []T) (dtoList []ReportDTO, err error) {
  793. var values []interface{}
  794. for _, id := range ids {
  795. values = append(values, id)
  796. }
  797. reports, err := reportDao.GetListByCondition(column, ids)
  798. if err != nil {
  799. logger.Error("获取研报失败:%v", err)
  800. return
  801. }
  802. for _, reportItem := range reports {
  803. dto := convertReportDTO(reportItem, false)
  804. dtoList = append(dtoList, dto)
  805. }
  806. return
  807. }
  808. func GetReportByIdListByOrgIds(orgIds map[string][]int) (ids []int, err error) {
  809. return reportDao.GetReportIdListByOrgIds(orgIds)
  810. }
  811. func GetTotalPageCountByPermissionIds(permissionIds []int, riskLevel string) (total int64, latestId int64, ids map[string][]int) {
  812. htOrgIds, err := GetHTReportIdsByPermissionIdsWithRiskLevel(permissionIds, riskLevel)
  813. if err != nil {
  814. logger.Error("品种筛选ht报告id失败:%v", err)
  815. htOrgIds = []int{}
  816. }
  817. etaOrgIds, err := GetETAReportIdsByPermissionIdsWithRiskLevel(permissionIds, riskLevel)
  818. if err != nil {
  819. logger.Error("品种筛选eta报告id失败:%v", err)
  820. etaOrgIds = []int{}
  821. }
  822. totalCol := int64(len(etaOrgIds) + len(htOrgIds))
  823. if totalCol == 0 {
  824. latestId = 0
  825. return
  826. }
  827. ids = make(map[string][]int, 2)
  828. if len(etaOrgIds) == 0 {
  829. ids["ETA"] = []int{}
  830. } else {
  831. ids["ETA"] = etaOrgIds
  832. }
  833. if len(htOrgIds) == 0 {
  834. ids["HT"] = []int{}
  835. } else {
  836. ids["HT"] = htOrgIds
  837. }
  838. total, latestId, err = reportDao.GetMaxIdByPermissionIds(ids)
  839. if err != nil {
  840. logger.Error("获取筛选报告的最大记录和记录数失败:%v", err)
  841. return
  842. }
  843. return
  844. }
  845. func convertEtaReport(etaRp eta.ETAReport, status reportDao.ReportStatus) reportDao.Report {
  846. return reportDao.Report{
  847. OrgID: etaRp.ID,
  848. Title: etaRp.Title,
  849. Abstract: etaRp.Abstract,
  850. Author: etaRp.Author,
  851. ClassifyId: etaRp.ClassifyID,
  852. CoverSrc: 0,
  853. PublishedTime: etaRp.PublishTime.Format(time.DateTime),
  854. Source: reportDao.SourceETA,
  855. SendStatus: reportDao.UNSEND,
  856. Status: status,
  857. }
  858. }
  859. func convertHTReport(etaRp ht.HTReport, status reportDao.ReportStatus) reportDao.Report {
  860. return reportDao.Report{
  861. OrgID: etaRp.Id,
  862. Title: etaRp.ReportName,
  863. Author: etaRp.PublishUserName,
  864. PublishedTime: etaRp.PublishedTime,
  865. CoverSrc: 0,
  866. Source: reportDao.SourceHT,
  867. SendStatus: reportDao.UNSEND,
  868. Status: status,
  869. }
  870. }
  871. func convertEsReport(report reportDao.Report) ESReport {
  872. return ESReport{
  873. ReportID: report.ID,
  874. Title: report.Title,
  875. OrgId: report.OrgID,
  876. Author: report.Author,
  877. Source: report.Source,
  878. Abstract: report.Abstract,
  879. Status: report.Status,
  880. CoverSrc: report.CoverSrc,
  881. PublishedTime: report.PublishedTime,
  882. }
  883. }
  884. func convertReportDTO(report reportDao.Report, fullTime bool) (reportDTO ReportDTO) {
  885. reportDTO = ReportDTO{
  886. ReportID: report.ID,
  887. Title: report.Title,
  888. OrgId: report.OrgID,
  889. Author: report.Author,
  890. Source: string(report.Source),
  891. CoverSrc: report.CoverSrc,
  892. Abstract: report.Abstract,
  893. PublishedTime: report.PublishedTime,
  894. PlateName: report.PlateName,
  895. ClassifyId: report.ClassifyId,
  896. }
  897. publishDate, err := time.Parse(time.DateTime, reportDTO.PublishedTime)
  898. if err == nil && !fullTime {
  899. reportDTO.PublishedTime = publishDate.Format(time.DateOnly)
  900. }
  901. return
  902. }
  903. func matchAll(sorts []string, key string) (request *es.ESQueryRequest) {
  904. req := new(es.ESQueryRequest)
  905. return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, 0, 1, sorts, es.MatchAllByCondition).ByCondition("status", "PUBLISH")
  906. }
  907. func match(key string, from int, to int, sorts []string) (request *es.ESQueryRequest) {
  908. req := new(es.ESQueryRequest)
  909. return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, from, to, sorts, es.MatchAllByCondition).ByCondition("status", "PUBLISH")
  910. }
  911. func matchRange(key string, from int, to int, max int64, sorts []string) (request *es.ESQueryRequest) {
  912. req := new(es.ESQueryRequest)
  913. return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, from, to, sorts, es.RangeByCondition).Range(0, max, ESRangeColumn).ByCondition("status", "PUBLISH")
  914. }
  915. func matchRangeByDocId(key string, from int, to int, max int64, sorts []string, docIds []string) (request *es.ESQueryRequest) {
  916. req := new(es.ESQueryRequest)
  917. return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, from, to, sorts, es.RangeByConditionWithDocIds).Range(0, max, ESRangeColumn).ByCondition("status", "PUBLISH").WithDocs(docIds)
  918. }
  919. func CountByDocId(key string, sorts []string, docIds []string) (request *es.ESQueryRequest) {
  920. req := new(es.ESQueryRequest)
  921. return req.CreateESQueryRequest(htConfig.GetReportIndex(), ESColumn, key, 0, 1, sorts, es.CountWithDocIds).WithDocs(docIds)
  922. }