report_service.go 35 KB

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