task.go 17 KB

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