task.go 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573
  1. package services
  2. import (
  3. "eta/eta_api/models"
  4. "eta/eta_api/services/binlog"
  5. "eta/eta_api/services/data"
  6. edbmonitor "eta/eta_api/services/edb_monitor"
  7. "eta/eta_api/services/llm"
  8. "eta/eta_api/utils"
  9. "fmt"
  10. "strings"
  11. "time"
  12. )
  13. func Task() {
  14. fmt.Println("task start")
  15. {
  16. // 修复客户试用数据
  17. //FixCompanyUpdateData()
  18. //FixCompanyTryDay()
  19. //FixCompanyTryDay()
  20. //StaticCompanyTryDay()
  21. //return
  22. }
  23. //FixPermissionStatus()
  24. //GetCompanyInfo()
  25. //ReportCount()
  26. //windSourceUrl:=`http://47.100.166.55:7002/edbInfo/wind/?EdbCode=M0001427&StartDate=2020-11-01&EndDate=2021-03-01`
  27. //data.AddAllArticle()
  28. //data.GetSmmIndex()
  29. //data.GetSmmIndexData()
  30. go AutoInsertLogToDB()
  31. //手工数据表格导入后的指标库刷新
  32. go ImportManualDataRefresh()
  33. // 加入钢联化工指标数据之后的刷新
  34. go MysteelChemicalDataAdd()
  35. //修复用户关注标识
  36. //GetWxUsersSubscribe()
  37. go AutoInsertAdminOperateRecordToDB()
  38. // 指标刷新
  39. go data.HandleEdbRefreshQueue()
  40. // 进行指标替换操作
  41. go DealReplaceEdbCache()
  42. // TODO:监听这里需要找下达梦的方案,主流程先跑起来再处理
  43. // 监听binlog
  44. if utils.MYSQL_DATA_BINLOG_URL != "" && utils.DbDriverName == `mysql` {
  45. go binlog.ListenMysql()
  46. go edbmonitor.HandleEdbMonitorEdbInfo()
  47. // 监听数据源binlog写入es
  48. go binlog.HandleDataSourceChange2Es()
  49. }
  50. go StartSessionManager()
  51. go llm.SaveAllChatRecordsToDB()
  52. // TODO:数据修复
  53. //FixNewEs()
  54. fmt.Println("task end")
  55. }
  56. //// 每日发布晨报
  57. //func AutoPublishDayReport() {
  58. // defer func() {
  59. // if err := recover(); err != nil {
  60. // fmt.Println("[AutoPublishDayReport]", err)
  61. // }
  62. // }()
  63. //
  64. // // 每日8:42发布晨报
  65. // ticker := time.Tick(50 * time.Second)
  66. // for range ticker {
  67. // nowTime := time.Now()
  68. // clock := nowTime.Format("1504")
  69. // if clock == "0842" {
  70. // if err := PublishTodayDayReport(); err != nil {
  71. // go alarm_msg.SendAlarmMsg(fmt.Sprint("每日晨报自动发送 AutoPublishDayReport ERR:", err), 3)
  72. // //utils.SendEmail(utils.APPNAME+" "+utils.RunMode+" 失败提醒", fmt.Sprint("AutoPublishDayReport ERR:", err), utils.EmailSendToUsers)
  73. // }
  74. // }
  75. // }
  76. //}
  77. // ImportManualDataRefresh 导入手工数据后的刷新
  78. func ImportManualDataRefresh() {
  79. defer func() {
  80. if err := recover(); err != nil {
  81. fmt.Println("[ImportManualDataRefresh]", err)
  82. }
  83. }()
  84. for {
  85. utils.Rc.Brpop(utils.CACHE_IMPORT_MANUAL_DATA, func(b []byte) {
  86. edbCode := string(b)
  87. edbCode = strings.TrimPrefix(edbCode, `"`)
  88. edbCode = strings.TrimSuffix(edbCode, `"`)
  89. data.RefreshManualData(edbCode)
  90. })
  91. }
  92. }
  93. // MysteelChemicalDataAdd 加入钢联化工指标数据之后的刷新
  94. func MysteelChemicalDataAdd() {
  95. defer func() {
  96. if err := recover(); err != nil {
  97. fmt.Println("[ImportManualDataRefresh]", err)
  98. }
  99. }()
  100. for {
  101. utils.Rc.Brpop(utils.CACHE_MYSTEEL_CHEMICAL_ADD_DATA, func(b []byte) {
  102. edbCode := string(b)
  103. edbCode = strings.TrimPrefix(edbCode, `"`)
  104. edbCode = strings.TrimSuffix(edbCode, `"`)
  105. data.RefreshMysteelChemicalData(edbCode)
  106. })
  107. }
  108. }
  109. //func init() {
  110. // fmt.Println("start task init")
  111. // UpdateEnglishEmailLogErrMsg()
  112. // fmt.Println("end task init")
  113. //}
  114. //
  115. //// UpdateEnglishEmailLogErrMsg 更新英文邮件日志的ErrMsg(研报后台4.2上线后执行, 仅一次)
  116. //func UpdateEnglishEmailLogErrMsg() {
  117. // var cond string
  118. // var pars []interface{}
  119. // list, e := models.GetEnglishReportEmailLogList(cond, pars)
  120. // if e != nil {
  121. // fmt.Println("获取日志列表失败")
  122. // return
  123. // }
  124. // for _, v := range list {
  125. // if v.SendStatus != 0 || v.Source != 1 || v.Result == "" || v.ErrMsg != "" {
  126. // continue
  127. // }
  128. // // 取出错误信息
  129. // fmt.Printf("正在更新%d\n", v.Id)
  130. // r := new(AliyunEmailResult)
  131. // if e = json.Unmarshal([]byte(v.Result), &r); e != nil {
  132. // fmt.Println("JSON解析报错了1" + e.Error())
  133. // continue
  134. // }
  135. // rd := new(AliyunEmailResultData)
  136. // res := strings.Replace(r.Data, `\`, ``, -1)
  137. // if e = json.Unmarshal([]byte(res), &rd); e != nil {
  138. // fmt.Println("JSON解析报错了2" + e.Error())
  139. // continue
  140. // }
  141. // v.ErrMsg = rd.Message
  142. // if e = v.Update([]string{"ErrMsg"}); e != nil {
  143. // fmt.Println("更新失败了" + e.Error())
  144. // continue
  145. // }
  146. // }
  147. // fmt.Println("更新成功")
  148. //}
  149. //func FixCompanyUpdateData1() {
  150. // list, err := company.GetCompanyProductUpdateLogList()
  151. // if err != nil {
  152. // fmt.Println("获取客户变更数据失败:", err)
  153. // return
  154. // }
  155. //
  156. // nowTime := time.Now()
  157. // for _, v := range list {
  158. // //item, tmpErr := company.GetCompanyProductLogItem(v.CompanyId, v.ProductId, v.CreateTime)
  159. // //if tmpErr != nil {
  160. // // fmt.Println(v.Id, "找数据,", tmpErr)
  161. // // continue
  162. // //}
  163. //
  164. // //permissionList := make([]*company.CompanyReportPermission, 0)
  165. // //switch v.Source {
  166. // //case "add", "receive", "thaw", "delay", "apply_receive":
  167. // //
  168. // //}
  169. // permissionList, err := company.GetCompanyReportPermission(v.CompanyId, v.ProductId)
  170. // if err != nil {
  171. // fmt.Println("err:", err)
  172. // continue
  173. // }
  174. //
  175. // startDate := v.CreateTime
  176. // endDate := v.CreateTime.AddDate(0, 2, 0)
  177. // isStop := 1
  178. // realEndDate := endDate
  179. // if realEndDate.After(nowTime) {
  180. // realEndDate = nowTime
  181. // isStop = 0
  182. // }
  183. // for _, permission := range permissionList {
  184. // permission.StartDate = startDate.Format(utils.FormatDate)
  185. // permission.EndDate = v.CreateTime.Format(utils.FormatDate)
  186. // }
  187. // companyProductTryOutUpdateLogItem := &company.CompanyProductTryOutUpdateLog{
  188. // Id: 0,
  189. // CompanyId: v.CompanyId,
  190. // ProductId: v.ProductId,
  191. // SellerId: v.SellerId,
  192. // SellerName: v.SellerName,
  193. // Source: v.Source,
  194. // StartDate: startDate,
  195. // EndDate: endDate,
  196. // RealEndDate: realEndDate,
  197. // IsStop: isStop,
  198. // CreateTime: v.CreateTime,
  199. // }
  200. // err = company.AddCompanyProductTryOutUpdateLog(companyProductTryOutUpdateLogItem, permissionList)
  201. // }
  202. // //fmt.Println("结束")
  203. //}
  204. //func FixCompanyUpdateData2() {
  205. // list, err := company.GetTryOutCompanyOperationRecordList()
  206. // if err != nil {
  207. // fmt.Println("获取客户变更数据失败:", err)
  208. // return
  209. // }
  210. //
  211. // nowTime := time.Now()
  212. // for _, v := range list {
  213. // //item, tmpErr := company.GetCompanyProductLogItem(v.CompanyId, v.ProductId, v.CreateTime)
  214. // //if tmpErr != nil {
  215. // // fmt.Println(v.Id, "找数据,", tmpErr)
  216. // // continue
  217. // //}
  218. //
  219. // //permissionList := make([]*company.CompanyReportPermission, 0)
  220. // //switch v.Source {
  221. // //case "add", "receive", "thaw", "delay", "apply_receive":
  222. // //
  223. // //}
  224. // permissionList, err := company.GetCompanyReportPermission(v.CompanyId, v.ProductId)
  225. // if err != nil {
  226. // fmt.Println("err:", err)
  227. // continue
  228. // }
  229. //
  230. // startDate := v.CreateTime
  231. // endDate := v.CreateTime.AddDate(0, 2, 0)
  232. // isStop := 1
  233. // realEndDate := endDate
  234. // if realEndDate.After(nowTime) {
  235. // realEndDate = nowTime
  236. // isStop = 0
  237. // }
  238. // for _, permission := range permissionList {
  239. // permission.StartDate = startDate.Format(utils.FormatDate)
  240. // permission.EndDate = v.CreateTime.Format(utils.FormatDate)
  241. // }
  242. // sellerName := ``
  243. // {
  244. // sysUser, _ := system.GetSysAdminById(v.SellerId)
  245. // if sysUser != nil {
  246. // sellerName = sysUser.RealName
  247. // }
  248. // }
  249. // companyProductTryOutUpdateLogItem := &company.CompanyProductTryOutUpdateLog{
  250. // Id: 0,
  251. // CompanyId: v.CompanyId,
  252. // ProductId: v.ProductId,
  253. // SellerId: v.SellerId,
  254. // SellerName: sellerName,
  255. // Source: "formal_to_try_out",
  256. // StartDate: startDate,
  257. // EndDate: endDate,
  258. // RealEndDate: realEndDate,
  259. // IsStop: isStop,
  260. // CreateTime: v.CreateTime,
  261. // }
  262. // err = company.AddCompanyProductTryOutUpdateLog(companyProductTryOutUpdateLogItem, permissionList)
  263. // }
  264. // fmt.Println("结束")
  265. //}
  266. // FixCompanyTryDay 修复试用天数
  267. //func FixCompanyTryDay() {
  268. // list, err := company.GetCompanyProductTryOutUpdateGroup()
  269. // if err != nil {
  270. // fmt.Println("获取客户变更数据失败:", err)
  271. // return
  272. // }
  273. //
  274. // lenList := len(list)
  275. // for k, v := range list {
  276. // fmt.Println("剩余", lenList-k-1, "条数据修复")
  277. // companyProduct, tmpErr := company.GetCompanyProductByCompanyIdAndProductId(v.CompanyId, v.ProductId)
  278. // if tmpErr != nil {
  279. // fmt.Println("company_id:", v.CompanyId, ";product_id:", v.ProductId, ";找不到对应的客户,Err:", tmpErr)
  280. // continue
  281. // }
  282. //
  283. // logList, err := company.GetCompanyProductTryOutUpdateList(v.CompanyId, v.ProductId)
  284. // if err != nil {
  285. // fmt.Println("查找客户日志失败,err:", err)
  286. // continue
  287. // }
  288. //
  289. // //lenLog := len(logList)
  290. // var day int //实际试用天数
  291. // var endDate time.Time
  292. // for _, log := range logList {
  293. // startDate := log.StartDate
  294. // if endDate.IsZero() {
  295. // endDate = log.RealEndDate
  296. // day = utils.GetTimeSubDay(startDate, log.RealEndDate) + 1
  297. // } else {
  298. // if log.RealEndDate.After(endDate) {
  299. // if endDate.After(startDate) {
  300. // startDate = endDate
  301. // }
  302. // if startDate.Equal(log.EndDate) {
  303. // day += utils.GetTimeSubDay(startDate, log.RealEndDate)
  304. // } else {
  305. // day += utils.GetTimeSubDay(startDate, log.RealEndDate) + 1
  306. // }
  307. // endDate = log.RealEndDate
  308. // }
  309. // }
  310. // }
  311. // companyProduct.TryOutDayTotal = day
  312. // companyProduct.Update([]string{"TryOutDayTotal"})
  313. // }
  314. // fmt.Println("结束")
  315. //}
  316. // StaticCompanyTryDay 定时任务每天更新试用天数
  317. //func StaticCompanyTryDay() {
  318. // list, err := company.GetCompanyProductTryOutUpdateNoStopGroup()
  319. // if err != nil {
  320. // fmt.Println("获取客户变更数据失败:", err)
  321. // return
  322. // }
  323. //
  324. // lenList := len(list)
  325. // for k, v := range list {
  326. // isAdd := false //是否要增加一天,默认不加
  327. // fmt.Println("剩余", lenList-k-1, "条数据修复")
  328. // permissionList, tmpErr := company.GetCompanyReportPermissionByStatus(v.CompanyId, v.ProductId, "试用")
  329. // if tmpErr == nil {
  330. // fmt.Println("company_id:", v.CompanyId, ";product_id:", v.ProductId, ";寻找对应的试用品种失败,Err:", tmpErr)
  331. // continue
  332. // }
  333. // currPermissionIdList := make([]int, 0) //当前试用的品种
  334. // for _, permission := range permissionList {
  335. // currPermissionIdList = append(currPermissionIdList, permission.ChartPermissionId)
  336. // }
  337. //
  338. // //获取所有未停止的记录列表
  339. // logList, err := company.GetCompanyProductTryOutUpdateNoStopListByEndDate(v.CompanyId, v.ProductId)
  340. // if err != nil {
  341. // fmt.Println("查找客户日志失败,err:", err)
  342. // continue
  343. // }
  344. //
  345. // logPermissionIdList := make([]int, 0) //当前日志中的试用的品种
  346. // for _, log := range logList {
  347. // //获取所有未停止的品种试用记录列表
  348. // logPermissionList, tmpErr := company.GetCompanyProductTryOutPermissionUpdateNoStopListByEndDate(log.Id)
  349. // if tmpErr != nil {
  350. // fmt.Println("查找客户品种变更日志失败,err:", tmpErr)
  351. // continue
  352. // }
  353. // lenLogPermissionList := len(logPermissionList) //当前日志存在试用的品种数量
  354. // stopPermission := 0 //当前日志需要停止的品种数量
  355. //
  356. // currTime := time.Now() //当前时间
  357. // for _, logPermission := range logPermissionList {
  358. // if utils.InArrayByInt(logPermissionIdList, logPermission.ChartPermissionId) {
  359. // // 如果已经被其他记录使用了,那么就将当前记录给标记停止
  360. // logPermission.IsStop = 1
  361. //
  362. // // 如果当前时间晚于结束日期,那么实际结束日期就是今天之前一天,否则就是当天
  363. // if currTime.After(logPermission.EndDate) {
  364. // logPermission.RealEndDate = currTime.AddDate(0, 0, -1)
  365. // } else {
  366. // logPermission.RealEndDate = time.Now()
  367. // }
  368. // logPermission.Update([]string{"IsStop", "RealEndDate"})
  369. // stopPermission++
  370. // continue
  371. // } else if !utils.InArrayByInt(currPermissionIdList, logPermission.ChartPermissionId) {
  372. // // 如果该品种不在当前客户的品种里面,那么也要将当前记录给标记停止
  373. // logPermission.IsStop = 1
  374. // // 如果当前时间晚于结束日期,那么实际结束日期就是今天之前一天,否则就是当天
  375. // if currTime.After(logPermission.EndDate) {
  376. // logPermission.RealEndDate = currTime.AddDate(0, 0, -1)
  377. // } else {
  378. // logPermission.RealEndDate = time.Now()
  379. // }
  380. // logPermission.Update([]string{"IsStop", "RealEndDate"})
  381. // stopPermission++
  382. // continue
  383. // }
  384. //
  385. // // 剩下的说明还处于试用状态,需要添加1天试用期,且需要把该品种加入到当前日志中的试用的品种列表
  386. // isAdd = true
  387. // logPermissionIdList = append(logPermissionIdList, logPermission.ChartPermissionId)
  388. // }
  389. //
  390. // //如果当前日志存在试用的品种数量 == 当前日志需要停止的品种数量
  391. // // 那么当前日志也是处于停用状态
  392. // if lenLogPermissionList == stopPermission {
  393. // log.IsStop = 1
  394. // // 如果当前时间晚于结束日期,那么实际结束日期就是今天之前一天,否则就是当天
  395. // if currTime.After(log.EndDate) {
  396. // log.RealEndDate = currTime.AddDate(0, 0, -1)
  397. // } else {
  398. // log.RealEndDate = time.Now()
  399. // }
  400. // log.Update([]string{"IsStop", "RealEndDate"})
  401. // }
  402. //
  403. // }
  404. //
  405. // // 如果需要添加,那么将该客户品种添加1天
  406. // if isAdd {
  407. //
  408. // }
  409. // // 更新客户产品的试用天数
  410. // tmpErr = company.AddCompanyProductTryOutDayTotal(v.CompanyId, v.ProductId)
  411. // if tmpErr != nil {
  412. // fmt.Println("company_id:", v.CompanyId, ";product_id:", v.ProductId, ";更新客户产品的试用天数,Err:", tmpErr)
  413. // continue
  414. // }
  415. // }
  416. // fmt.Println("结束")
  417. //}
  418. //func Task2() {
  419. // fmt.Println("task start")
  420. //
  421. // //_, _ = maycur.GetPublicOfferingSaleLeader()
  422. // //_ = maycur.TestSyncCompanyProfile()
  423. // //_ = maycur.SyncCompanyProfile()
  424. // //_ = maycur.ImportExcelEmployeeId()
  425. // //_ = maycur.ApiTest()
  426. // //_ = maycur.ApiTest2()
  427. // fmt.Println("task end")
  428. //}
  429. // FixEnCompanyPermission 英文权限上线时修复英文客户拥有所有权限(一次性)
  430. func FixEnCompanyPermission() {
  431. var err error
  432. defer func() {
  433. if err != nil {
  434. fmt.Println("FixEnCompanyPermission Err: ", err.Error())
  435. }
  436. }()
  437. // 获取正式客户
  438. companies := make([]*models.EnglishCompany, 0)
  439. {
  440. cond := ` AND status = ?`
  441. pars := make([]interface{}, 0)
  442. pars = append(pars, 1)
  443. list, e := models.GetEnglishCompanyList(cond, pars, "")
  444. if e != nil {
  445. err = fmt.Errorf("GetEnglishCompanyList err: %s", e.Error())
  446. return
  447. }
  448. companies = list
  449. }
  450. // 获取所有权限
  451. permissions := make([]*models.EnPermission, 0)
  452. {
  453. cond := ` AND parent_id > ?`
  454. pars := make([]interface{}, 0)
  455. pars = append(pars, 0)
  456. ob := new(models.EnPermission)
  457. list, e := ob.GetItemsByCondition(cond, pars, []string{}, "")
  458. if e != nil {
  459. err = fmt.Errorf("GetPermissionItemsByCondition err: %s", e.Error())
  460. return
  461. }
  462. permissions = list
  463. }
  464. for _, c := range companies {
  465. ps := make([]*models.EnCompanyPermission, 0)
  466. for _, p := range permissions {
  467. ps = append(ps, &models.EnCompanyPermission{
  468. EnCompanyId: c.CompanyId,
  469. EnPermissionId: p.EnPermissionId,
  470. CreateTime: time.Now().Local(),
  471. })
  472. }
  473. if e := models.ClearAndCreateEnCompanyPermissions(c.CompanyId, ps); e != nil {
  474. err = fmt.Errorf("ClearAndCreateEnCompanyPermissions err: %s", e.Error())
  475. return
  476. }
  477. }
  478. fmt.Println("修复完成")
  479. }
  480. // ModifyEsEnglishReport 批量修改es里的英文研报信息和线上路演信息
  481. func ModifyEsEnglishReport() {
  482. fmt.Println("开始")
  483. err := ModifyAllEsEnglishReportVideo()
  484. if err != nil {
  485. err = fmt.Errorf("重置es中的英文研报信息失败:ModifyAllEnglishReportInEs err: %s", err.Error())
  486. return
  487. }
  488. fmt.Println("结束")
  489. }
  490. //func InsertBloombergIndex() {
  491. // fmt.Println("开始写入")
  492. //
  493. // start := 100000
  494. // now := time.Now()
  495. // for i := 1; i <= 100; i++ {
  496. // fmt.Printf("写入第%d个\n", i)
  497. //
  498. // start += 1
  499. // index := new(data_manage.BaseFromBloombergIndex)
  500. // index.IndexCode = fmt.Sprintf("BLID%d", start)
  501. // index.IndexName = fmt.Sprintf("模拟Bloomberg-%s", index.IndexCode)
  502. // index.Unit = "无"
  503. // index.Source = utils.DATA_SOURCE_BLOOMBERG
  504. // index.Frequency = "日度"
  505. // index.StartDate = now.AddDate(0, 0, -i)
  506. // index.EndDate = now
  507. // index.CreateTime = time.Now().Local()
  508. // index.ModifyTime = time.Now().Local()
  509. // if e := index.Create(); e != nil {
  510. // fmt.Printf("新增指标失败, IndexCode: %s, err: %s", index.IndexCode, e.Error())
  511. // return
  512. // }
  513. //
  514. // insertData := make([]*data_manage.BaseFromBloombergData, 0)
  515. // for ii := 0; ii <= 50; ii++ {
  516. // indexData := new(data_manage.BaseFromBloombergData)
  517. // indexData.BaseFromBloombergIndexId = index.BaseFromBloombergIndexId
  518. // indexData.IndexCode = index.IndexCode
  519. // indexData.DataTime = now.AddDate(0, 0, -ii)
  520. // va := GenerateRandomFloat64InRange()
  521. // va += float64(ii)
  522. // indexData.Value = va
  523. // indexData.CreateTime = time.Now().Local()
  524. // indexData.ModifyTime = time.Now().Local()
  525. // indexData.DataTimestamp = int(indexData.DataTime.UnixNano() / 1e6)
  526. // insertData = append(insertData, indexData)
  527. // }
  528. // ob := new(data_manage.BaseFromBloombergData)
  529. // if e := ob.CreateMulti(insertData); e != nil {
  530. // fmt.Printf("新增指标数据失败, IndexCode: %s, err: %s", index.IndexCode, e.Error())
  531. // return
  532. // }
  533. // }
  534. //
  535. // fmt.Println("结束写入")
  536. //}
  537. //
  538. //func GenerateRandomFloat64InRange() float64 {
  539. // var rnd = rand.New(rand.NewSource(time.Now().UnixNano())) // 设置随机数种子
  540. //
  541. // return rnd.Float64()*11000 - 1000
  542. //}