base_from_kpler.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896
  1. package services
  2. import (
  3. "encoding/json"
  4. "eta/eta_index_lib/logic"
  5. "eta/eta_index_lib/models"
  6. "eta/eta_index_lib/services/alarm_msg"
  7. "eta/eta_index_lib/utils"
  8. "fmt"
  9. "io/ioutil"
  10. "net/http"
  11. "strconv"
  12. "strings"
  13. "time"
  14. )
  15. // GetKplerDataByApi 获取开普勒数据
  16. func GetKplerDataByApi(params models.KplerSearchEdbReq, terminalCode string) (indexes []*models.KplerIndexItem, apiQueryUrl string, terminalInfo *models.EdbTerminal, err error) {
  17. terminal, e := GetApiTerminal(utils.DATA_SOURCE_KPLER, terminalCode)
  18. if e != nil {
  19. err = fmt.Errorf("获取开普勒终端配置失败, %v", e)
  20. return
  21. }
  22. if terminal.ServerUrl == "" {
  23. err = fmt.Errorf("开普勒终端地址未配置")
  24. return
  25. }
  26. // 走API
  27. if terminal.IsApi == 1 {
  28. indexes, apiQueryUrl, err = getKplerDataByApi(params, terminal.ServerUrl)
  29. if err != nil {
  30. err = fmt.Errorf("获取开普勒指标数据失败, %v", err)
  31. return
  32. }
  33. terminalInfo = terminal
  34. return
  35. }
  36. return
  37. }
  38. // getEdbDataFromThsHfHttp API-获取高频指标数据
  39. func getKplerDataByApi(params models.KplerSearchEdbReq, serverUrl string) (list []*models.KplerIndexItem, apiQueryUrl string, err error) {
  40. defer func() {
  41. if err != nil {
  42. tips := fmt.Sprintf("开普勒指标API-getKplerDataByApi err: %v", err)
  43. utils.FileLog.Info(tips)
  44. }
  45. }()
  46. // todo 判断指标是否已存在
  47. productNamesStr := params.ProductNames
  48. fromZoneNamesStr := params.FromZoneNames
  49. toZoneNamesStr := params.ToZoneNames
  50. libreq := new(models.KplerFlowDataLibReq)
  51. libreq.Products = productNamesStr
  52. libreq.FromZones = fromZoneNamesStr
  53. libreq.ToZones = toZoneNamesStr
  54. libreq.Split = params.Split
  55. libreq.FlowDirection = params.FlowDirection
  56. libreq.Granularity = params.Granularity
  57. libreq.Unit = params.Unit
  58. libreq.WithIntraRegion = "true"
  59. libreq.WithForecast = "true"
  60. libreq.OnlyRealized = "false"
  61. libreq.StartDate = "2025-01-01"
  62. libreq.EndDate = time.Now().Format(utils.FormatDate)
  63. // 请求接口
  64. apiResp, err := getKplerFlowDataLib(serverUrl, libreq)
  65. if err != nil {
  66. err = fmt.Errorf("开普勒指标API-getKplerDataByApi err: %v", err)
  67. return
  68. }
  69. if apiResp.Ret != 200 {
  70. err = fmt.Errorf("开普勒指标API-状态码: %d, 提示信息: %s", apiResp.Ret, apiResp.ErrMsg)
  71. return
  72. }
  73. indexes := make([]*models.KplerIndexItem, 0)
  74. if len(apiResp.Data) == 0 {
  75. utils.FileLog.Info("开普勒指标API-无数据")
  76. return
  77. }
  78. indexCodeList := make([]string, 0)
  79. existIndexCodeMap := make(map[string]bool)
  80. flowDirection := params.FlowDirection
  81. prefixIndexName := fmt.Sprintf("kpler%s%s%s%s", fromZoneNamesStr, flowDirection, toZoneNamesStr, productNamesStr)
  82. lastIndexName := params.Granularity
  83. prefixIndexCode := fmt.Sprintf("k%s%s%s%s", fromZoneNamesStr, flowDirection, toZoneNamesStr, productNamesStr)
  84. // 获取首字母
  85. lastIndexCode := strings.ToUpper(params.Granularity[:1])
  86. // Tables中的每一个对应一个证券代码
  87. for _, v := range apiResp.Data {
  88. index := new(models.KplerIndexItem)
  89. indexName := ""
  90. indexCode := ""
  91. indexName = fmt.Sprintf("%s%s%s", prefixIndexName, v.SplitItem, lastIndexName)
  92. indexCode = fmt.Sprintf("%s%s%s", prefixIndexCode,v.SplitItem, lastIndexCode)
  93. indexCodeList = append(indexCodeList, indexCode)
  94. index.IndexCode = indexCode
  95. index.IndexName = indexName
  96. index.Frequency = getKplerFrequency(params.Granularity)
  97. index.Unit = params.Unit
  98. index.IndexData = v.IndexData
  99. indexes = append(indexes, index)
  100. }
  101. indexObj := new(models.BaseFromKplerIndex)
  102. // 查询指标是否存在
  103. existList, err := indexObj.GetByIndexCodes(indexCodeList)
  104. if err != nil {
  105. err = fmt.Errorf("查询指标是否存在失败, %v", err)
  106. return
  107. }
  108. for _, v := range existList {
  109. existIndexCodeMap[v.IndexCode] = true
  110. }
  111. // 过滤已经存在的指标
  112. list = make([]*models.KplerIndexItem, 0)
  113. for _, v := range indexes {
  114. if _, ok := existIndexCodeMap[v.IndexCode]; !ok {
  115. list = append(list, v)
  116. }
  117. }
  118. return
  119. }
  120. func AddKplerIndexByApi(indexList []*models.KplerIndexItem, req *models.KplerSearchEdbReq, ApiQueryUrl string, classifyId int, terminalCode string) (err error) {
  121. errMsgList := make([]string, 0)
  122. defer func() {
  123. if len(errMsgList) > 0 {
  124. msg := fmt.Sprint("数据源-Kpler数据处理失败,err:", strings.Join(errMsgList, "\n"))
  125. utils.FileLog.Info(msg)
  126. go alarm_msg.SendAlarmMsg(msg, 3)
  127. }
  128. }()
  129. maxSort := 0
  130. for _, indexInfo := range indexList {
  131. indexObj := new(models.BaseFromKplerIndex)
  132. indexId := 0
  133. //判断指标是否存在
  134. var isAdd int
  135. item, er := indexObj.GetByIndexCode(indexInfo.IndexCode)
  136. if er != nil {
  137. if er.Error() == utils.ErrNoRow() {
  138. isAdd = 1
  139. err = nil
  140. } else {
  141. errMsgList = append(errMsgList, fmt.Sprintf("查询数据源指标库失败 GetByIndexCode Err:%s", er))
  142. continue
  143. }
  144. }
  145. if item != nil && item.BaseFromKplerIndexId > 0 {
  146. fmt.Println("item:", item)
  147. isAdd = 2
  148. } else {
  149. isAdd = 1
  150. }
  151. // 批量新增指标
  152. if isAdd == 1 {
  153. indexObj.IndexCode = indexInfo.IndexCode
  154. indexObj.IndexName = indexInfo.IndexName
  155. indexObj.Frequency = indexInfo.Frequency
  156. indexObj.ClassifyId = int(classifyId)
  157. indexObj.FromZoneName = req.FromZoneNames
  158. indexObj.ToZoneName = req.ToZoneNames
  159. indexObj.FlowDirection = req.FlowDirection
  160. indexObj.Granularity = req.Granularity
  161. indexObj.Split = req.Split
  162. indexObj.Unit = req.Unit
  163. indexObj.Sort = maxSort
  164. indexObj.ApiQueryUrl = ApiQueryUrl
  165. indexObj.ModifyTime = time.Now()
  166. indexObj.CreateTime = time.Now()
  167. indexObj.TerminalCode = terminalCode
  168. err = indexObj.Add()
  169. if err != nil {
  170. err = fmt.Errorf("数据源新增Kpler指标失败 Err:%s", err)
  171. return
  172. }
  173. indexId = indexObj.BaseFromKplerIndexId
  174. maxSort++
  175. } else if isAdd == 2 {
  176. indexId = item.BaseFromKplerIndexId
  177. if item.TerminalCode == `` && terminalCode != `` {
  178. item.TerminalCode = terminalCode
  179. err = item.Update([]string{"TerminalCode"})
  180. if err != nil {
  181. err = fmt.Errorf("数据源更新Kpler指标失败 Err:%s", err)
  182. return
  183. }
  184. }
  185. indexObj.BaseFromKplerIndexId = item.BaseFromKplerIndexId
  186. indexObj.IndexName = indexInfo.IndexName
  187. indexObj.Frequency = indexInfo.Frequency
  188. indexObj.ClassifyId = int(classifyId)
  189. indexObj.ProductNames = req.ProductNames
  190. indexObj.FromZoneName = req.FromZoneNames
  191. indexObj.ToZoneName = req.ToZoneNames
  192. indexObj.FlowDirection = req.FlowDirection
  193. indexObj.Granularity = req.Granularity
  194. indexObj.Split = req.Split
  195. indexObj.Unit = req.Unit
  196. indexObj.ApiQueryUrl = ApiQueryUrl
  197. indexObj.ModifyTime = time.Now()
  198. //修改数据
  199. updateColsArr := make([]string, 0)
  200. updateColsArr = append(updateColsArr, "index_name")
  201. updateColsArr = append(updateColsArr, "classify_id")
  202. updateColsArr = append(updateColsArr, "product_names")
  203. updateColsArr = append(updateColsArr, "from_zone_name")
  204. updateColsArr = append(updateColsArr, "to_zone_name")
  205. updateColsArr = append(updateColsArr, "flow_direction")
  206. updateColsArr = append(updateColsArr, "granularity")
  207. updateColsArr = append(updateColsArr, "split")
  208. updateColsArr = append(updateColsArr, "frequency")
  209. updateColsArr = append(updateColsArr, "modify_time")
  210. e := indexObj.Update(updateColsArr)
  211. if e != nil {
  212. fmt.Println("Index Update Err:" + e.Error())
  213. return
  214. }
  215. }
  216. dataObj := new(models.BaseFromKplerData)
  217. exitDataMap := make(map[string]*models.BaseFromKplerData)
  218. //获取已存在的所有数据
  219. var exitDataList []*models.BaseFromKplerData
  220. exitDataList, err = dataObj.GetByIndexCode(indexInfo.IndexCode)
  221. if err != nil {
  222. err = fmt.Errorf("数据源查询Kpler指标数据失败 Err:%s", err)
  223. return
  224. }
  225. fmt.Println("exitDataListLen:", len(exitDataList))
  226. for _, v := range exitDataList {
  227. dateStr := v.DataTime
  228. exitDataMap[dateStr] = v
  229. }
  230. addDataList := make([]*models.BaseFromKplerData, 0)
  231. // 遍历excel数据,然后跟现有的数据做校验,不存在则入库
  232. for _, value := range indexInfo.IndexData {
  233. date := value.DataTime
  234. value := value.Value
  235. if findData, ok := exitDataMap[date]; !ok {
  236. _, err = time.ParseInLocation(utils.FormatDate, date, time.Local)
  237. if err != nil {
  238. err = fmt.Errorf("%s 转换日期格式失败 Err:%s", date, err)
  239. return
  240. }
  241. var saveDataTime time.Time
  242. if strings.Contains(date, "00:00:00") {
  243. saveDataTime, err = time.Parse(utils.FormatDateTime, date)
  244. } else {
  245. saveDataTime, err = time.Parse(utils.FormatDate, date)
  246. }
  247. if err != nil {
  248. err = fmt.Errorf("%s 转换日期格式失败 Err:%s", date, err)
  249. continue
  250. }
  251. timestamp := saveDataTime.UnixNano() / 1e6
  252. dataItem := new(models.BaseFromKplerData)
  253. dataItem.BaseFromKplerIndexId = int(indexId)
  254. dataItem.IndexCode = indexInfo.IndexCode
  255. dataItem.DataTime = date
  256. dataItem.Value = value
  257. dataItem.CreateTime = time.Now()
  258. dataItem.ModifyTime = time.Now()
  259. dataItem.DataTimestamp = timestamp
  260. addDataList = append(addDataList, dataItem)
  261. if len(addDataList) > 500 {
  262. err = dataObj.AddMulti(addDataList)
  263. if err != nil {
  264. err = fmt.Errorf("批量新增指标失败 Err:%s", err)
  265. return
  266. }
  267. addDataList = make([]*models.BaseFromKplerData, 0)
  268. }
  269. } else {
  270. if findData != nil && findData.Value != value && !strings.Contains(value, "#N/A") { //修改数据
  271. // 过滤0.50和0.5的比较
  272. oldV, _ := strconv.ParseFloat(findData.Value, 64)
  273. newV, _ := strconv.ParseFloat(value, 64)
  274. if oldV == newV {
  275. continue
  276. }
  277. dataObj.BaseFromKplerIndexId = findData.BaseFromKplerIndexId
  278. dataObj.Value = value
  279. dataObj.ModifyTime = time.Now()
  280. updateDataColsArr := make([]string, 0)
  281. updateDataColsArr = append(updateDataColsArr, "value")
  282. updateDataColsArr = append(updateDataColsArr, "modify_time")
  283. dataObj.Update(updateDataColsArr)
  284. }
  285. }
  286. }
  287. if len(addDataList) > 0 {
  288. err = dataObj.AddMulti(addDataList)
  289. if err != nil {
  290. err = fmt.Errorf("批量新增指标失败 Err:%s", err)
  291. return
  292. }
  293. }
  294. var dateItem *models.EdbInfoMaxAndMinInfo
  295. dateItem, err = dataObj.GetMaxAndMinDateByIndexCode(indexInfo.IndexCode)
  296. if err != nil {
  297. err = fmt.Errorf("查询指标最新日期失败 Err:%s", err)
  298. return
  299. }
  300. go func() {
  301. indexObj.ModifyIndexMaxAndMinDate(indexInfo.IndexCode, dateItem)
  302. }()
  303. // 同步刷新ETA指标库的指标
  304. {
  305. // 获取指标详情
  306. baseObj := new(models.BaseFromKpler)
  307. var edbInfo *models.EdbInfo
  308. edbInfo, err = models.GetEdbInfoByEdbCode(baseObj.GetSource(), indexInfo.IndexCode)
  309. if err != nil {
  310. if !utils.IsErrNoRow(err) {
  311. errMsgList = append(errMsgList, fmt.Sprint("刷新ETA指标异常,指标编码:", indexInfo.IndexCode, err.Error()))
  312. return
  313. } else {
  314. err = nil
  315. }
  316. }
  317. // 已经加入到指标库的话,那么就去更新ETA指标库吧
  318. if edbInfo != nil {
  319. go logic.RefreshBaseEdbInfo(edbInfo, ``)
  320. }
  321. }
  322. }
  323. return
  324. }
  325. // HandleKplerIndex 处理Kpler的excel数据
  326. func HandleKplerIndex(req *models.HandleKplerExcelDataReq) (err error) {
  327. errMsgList := make([]string, 0)
  328. defer func() {
  329. if len(errMsgList) > 0 {
  330. msg := fmt.Sprint("数据源-Kpler数据处理失败,err:", strings.Join(errMsgList, "\n"))
  331. utils.FileLog.Info(msg)
  332. go alarm_msg.SendAlarmMsg(msg, 3)
  333. }
  334. }()
  335. // 查询所有的一级分类
  336. classifyObj := new(models.BaseFromKplerClassify)
  337. classifyList, err := classifyObj.GetParentClassify()
  338. if err != nil {
  339. err = fmt.Errorf("查询一级目录信息失败 Err:%s", err)
  340. return
  341. }
  342. classifyMap := make(map[string]int, 0)
  343. for _, v := range classifyList {
  344. classifyMap[v.ClassifyName] = int(v.ClassifyId)
  345. }
  346. for _, v := range req.List {
  347. err = handleKplerIndex(v, req.TerminalCode, classifyMap)
  348. if err != nil {
  349. errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常,指标编码:%s, Err: %s", v.IndexCode, err))
  350. return
  351. }
  352. }
  353. return
  354. }
  355. // getKplerFlowDataLib 获取开普勒流数据
  356. func getKplerFlowDataLib(libUrl string, dataMap *models.KplerFlowDataLibReq) (resp *models.KplerFlowDataLibResp, err error) {
  357. postUrl := fmt.Sprintf("%s%s", libUrl, "/v1/kpler/getFlowData")
  358. postData, err := json.Marshal(dataMap)
  359. if err != nil {
  360. return
  361. }
  362. result, err := HttpPost(postUrl, string(postData), "application/json")
  363. if err != nil {
  364. return
  365. }
  366. utils.FileLog.Info("postRefreshEdbData:" + postUrl + ";" + string(postData) + ";result:" + string(result))
  367. err = json.Unmarshal(result, &resp)
  368. if err != nil {
  369. return
  370. }
  371. return resp, nil
  372. }
  373. func getKplerProductLib(libUrl string) (resp *models.KplerProductLibResp, err error) {
  374. postUrl := fmt.Sprintf("%s%s", libUrl, "/v1/kpler/getProductData")
  375. result, err := HttpPost(postUrl, "", "application/json")
  376. if err != nil {
  377. return
  378. }
  379. utils.FileLog.Info("postRefreshEdbData:" + postUrl + ";result:" + string(result))
  380. err = json.Unmarshal(result, &resp)
  381. if err != nil {
  382. return
  383. }
  384. return resp, nil
  385. }
  386. func getKplerZoneLib(libUrl string) (resp *models.KplerZoneLibResp, err error) {
  387. postUrl := fmt.Sprintf("%s%s", libUrl, "/v1/kpler/getZoneData")
  388. result, err := HttpPost(postUrl, "", "application/json")
  389. if err != nil {
  390. return
  391. }
  392. utils.FileLog.Info("postRefreshEdbData:" + postUrl + ";result:" + string(result))
  393. err = json.Unmarshal(result, &resp)
  394. if err != nil {
  395. return
  396. }
  397. return resp, nil
  398. }
  399. func handleKplerIndex(req *models.HandleKplerExcelData, terminalCode string, classifyMap map[string]int) (err error) {
  400. indexName := req.IndexName
  401. indexCode := req.IndexCode
  402. excelDataMap := req.ExcelDataMap
  403. errMsgList := make([]string, 0)
  404. defer func() {
  405. if len(errMsgList) > 0 {
  406. msg := fmt.Sprint("数据源-Kpler数据处理失败,err:", strings.Join(errMsgList, "\n"))
  407. utils.FileLog.Info(msg)
  408. go alarm_msg.SendAlarmMsg(msg, 3)
  409. }
  410. }()
  411. indexObj := new(models.BaseFromKplerIndex)
  412. dataObj := new(models.BaseFromKplerData)
  413. classifyObj := new(models.BaseFromKplerClassify)
  414. var indexId int64
  415. addDataList := make([]*models.BaseFromKplerData, 0)
  416. exitDataMap := make(map[string]*models.BaseFromKplerData)
  417. // 修改指标信息
  418. if indexName == "" {
  419. utils.FileLog.Info("未刷新到指标数据:indexName:" + indexName)
  420. return
  421. }
  422. if req.ClassifyName == "" {
  423. // 获取产品的名称,把产品的第一个名称作为分类ID
  424. productNames := strings.Split(req.ProductNames, ",")
  425. if len(productNames) > 0 {
  426. req.ClassifyName = productNames[0]
  427. }
  428. }
  429. fromZoneNamesStr := req.FromZoneName
  430. flowDirection := req.FlowDirection
  431. toZoneNamesStr := req.ToZoneName
  432. productNamesStr := req.ProductNames
  433. // 拼接指标编码
  434. prefixIndexCode := fmt.Sprintf("k%s%s%s%s", fromZoneNamesStr, flowDirection, toZoneNamesStr, productNamesStr)
  435. // 获取首字母
  436. lastIndexCode := strings.ToUpper(req.Granularity[:1])
  437. indexCode = fmt.Sprintf("%s%s%s", prefixIndexCode,req.SplitName, lastIndexCode)
  438. // 判断目录是否存在
  439. var classifyId int64
  440. now := time.Now()
  441. if req.ClassifyName != "" {
  442. classifyParentId := 0
  443. level := 1
  444. classifyParentId, _ = classifyMap[req.ParentClassifyName]
  445. if classifyParentId > 0 {
  446. level = 2
  447. }
  448. classifyObj, err = classifyObj.GetByClassifyName(req.ClassifyName)
  449. if err != nil {
  450. if err.Error() == utils.ErrNoRow() {
  451. //新增分类
  452. classifyObj = &models.BaseFromKplerClassify{
  453. ClassifyName: req.ClassifyName,
  454. ClassifyNameEn: req.ClassifyName,
  455. ParentId: classifyParentId,
  456. SysUserId: 0,
  457. SysUserRealName: "",
  458. Level: level,
  459. Sort: req.ClassifySort,
  460. ModifyTime: now,
  461. CreateTime: now,
  462. }
  463. classifyId, err = classifyObj.Add()
  464. if err != nil {
  465. err = fmt.Errorf("新增分类失败 Err:%s", err)
  466. return
  467. }
  468. classifyObj.ClassifyId = int(classifyId)
  469. } else {
  470. return
  471. }
  472. } else {
  473. classifyId = int64(classifyObj.ClassifyId)
  474. classifyObj.ModifyTime = now
  475. classifyObj.ParentId = classifyParentId
  476. e := classifyObj.Update([]string{"ParentId", "ModifyTime"})
  477. if e != nil {
  478. fmt.Println("classifyObj Update Err:" + e.Error())
  479. return
  480. }
  481. }
  482. }
  483. //判断指标是否存在
  484. var isAdd int
  485. item, err := indexObj.GetByIndexCode(indexCode)
  486. if err != nil {
  487. if err.Error() == utils.ErrNoRow() {
  488. isAdd = 1
  489. err = nil
  490. } else {
  491. isAdd = -1
  492. err = fmt.Errorf("查询数据源指标库失败 GetByIndexCode Err:%s", err)
  493. return
  494. }
  495. }
  496. if item != nil && item.BaseFromKplerIndexId > 0 {
  497. fmt.Println("item:", item)
  498. isAdd = 2
  499. } else {
  500. isAdd = 1
  501. }
  502. if isAdd == 1 {
  503. indexObj.IndexCode = indexCode
  504. indexObj.IndexName = indexName
  505. indexObj.Frequency = getKplerFrequency(req.Granularity)
  506. indexObj.ClassifyId = int(classifyId)
  507. indexObj.ProductNames = req.ProductNames
  508. indexObj.FromZoneId = req.FromZoneId
  509. indexObj.FromZoneName = req.FromZoneName
  510. indexObj.ToZoneId = req.ToZoneId
  511. indexObj.ToZoneName = req.ToZoneName
  512. indexObj.FlowDirection = req.FlowDirection
  513. indexObj.Granularity = req.Granularity
  514. indexObj.Split = req.Split
  515. indexObj.Unit = req.Unit
  516. indexObj.Sort = req.Sort
  517. indexObj.ExcelQueryUrl = req.ExcelQueryUrl
  518. indexObj.ModifyTime = time.Now()
  519. indexObj.CreateTime = time.Now()
  520. indexObj.TerminalCode = terminalCode
  521. err = indexObj.Add()
  522. if err != nil {
  523. err = fmt.Errorf("数据源新增Kpler指标失败 Err:%s", err)
  524. return
  525. }
  526. indexId = int64(indexObj.BaseFromKplerIndexId)
  527. } else if isAdd == 2 {
  528. indexId = int64(item.BaseFromKplerIndexId)
  529. if item.TerminalCode == `` && terminalCode != `` {
  530. item.TerminalCode = terminalCode
  531. err = item.Update([]string{"TerminalCode"})
  532. if err != nil {
  533. err = fmt.Errorf("数据源更新Kpler指标失败 Err:%s", err)
  534. return
  535. }
  536. }
  537. indexObj.BaseFromKplerIndexId = item.BaseFromKplerIndexId
  538. indexObj.IndexName = indexName
  539. indexObj.Frequency = getKplerFrequency(req.Granularity)
  540. indexObj.ClassifyId = int(classifyId)
  541. indexObj.ProductNames = req.ProductNames
  542. indexObj.FromZoneId = req.FromZoneId
  543. indexObj.FromZoneName = req.FromZoneName
  544. indexObj.ToZoneId = req.ToZoneId
  545. indexObj.ToZoneName = req.ToZoneName
  546. indexObj.FlowDirection = req.FlowDirection
  547. indexObj.Granularity = req.Granularity
  548. indexObj.Split = req.Split
  549. indexObj.ExcelQueryUrl = req.ExcelQueryUrl
  550. indexObj.Unit = req.Unit
  551. indexObj.Sort = req.Sort
  552. indexObj.ModifyTime = time.Now()
  553. //修改数据
  554. updateColsArr := make([]string, 0)
  555. updateColsArr = append(updateColsArr, "index_name")
  556. updateColsArr = append(updateColsArr, "classify_id")
  557. updateColsArr = append(updateColsArr, "product_names")
  558. updateColsArr = append(updateColsArr, "from_zone_id")
  559. updateColsArr = append(updateColsArr, "from_zone_name")
  560. updateColsArr = append(updateColsArr, "to_zone_id")
  561. updateColsArr = append(updateColsArr, "to_zone_name")
  562. updateColsArr = append(updateColsArr, "flow_direction")
  563. updateColsArr = append(updateColsArr, "granularity")
  564. updateColsArr = append(updateColsArr, "split")
  565. updateColsArr = append(updateColsArr, "frequency")
  566. updateColsArr = append(updateColsArr, "sort")
  567. updateColsArr = append(updateColsArr, "modify_time")
  568. e := indexObj.Update(updateColsArr)
  569. if e != nil {
  570. fmt.Println("Index Update Err:" + e.Error())
  571. return
  572. }
  573. }
  574. //获取已存在的所有数据
  575. var exitDataList []*models.BaseFromKplerData
  576. exitDataList, err = dataObj.GetByIndexCode(indexCode)
  577. if err != nil {
  578. err = fmt.Errorf("数据源查询Kpler指标数据失败 Err:%s", err)
  579. return
  580. }
  581. fmt.Println("exitDataListLen:", len(exitDataList))
  582. for _, v := range exitDataList {
  583. dateStr := v.DataTime
  584. exitDataMap[dateStr] = v
  585. }
  586. // 遍历excel数据,然后跟现有的数据做校验,不存在则入库
  587. for date, value := range excelDataMap {
  588. if findData, ok := exitDataMap[date]; !ok {
  589. _, err = time.ParseInLocation(utils.FormatDate, date, time.Local)
  590. if err != nil {
  591. err = fmt.Errorf("%s 转换日期格式失败 Err:%s", date, err)
  592. return
  593. }
  594. var saveDataTime time.Time
  595. if strings.Contains(date, "00:00:00") {
  596. saveDataTime, err = time.Parse(utils.FormatDateTime, date)
  597. } else {
  598. saveDataTime, err = time.Parse(utils.FormatDate, date)
  599. }
  600. if err != nil {
  601. err = fmt.Errorf("%s 转换日期格式失败 Err:%s", date, err)
  602. continue
  603. }
  604. timestamp := saveDataTime.UnixNano() / 1e6
  605. dataItem := new(models.BaseFromKplerData)
  606. dataItem.BaseFromKplerIndexId = int(indexId)
  607. dataItem.IndexCode = indexCode
  608. dataItem.DataTime = date
  609. dataItem.Value = value
  610. dataItem.CreateTime = time.Now()
  611. dataItem.ModifyTime = time.Now()
  612. dataItem.DataTimestamp = timestamp
  613. addDataList = append(addDataList, dataItem)
  614. if len(addDataList) > 500 {
  615. err = dataObj.AddMulti(addDataList)
  616. if err != nil {
  617. err = fmt.Errorf("批量新增指标失败 Err:%s", err)
  618. return
  619. }
  620. addDataList = make([]*models.BaseFromKplerData, 0)
  621. }
  622. } else {
  623. if findData != nil && findData.Value != value && !strings.Contains(value, "#N/A") { //修改数据
  624. // 过滤0.50和0.5的比较
  625. oldV, _ := strconv.ParseFloat(findData.Value, 64)
  626. newV, _ := strconv.ParseFloat(value, 64)
  627. if oldV == newV {
  628. continue
  629. }
  630. dataObj.BaseFromKplerIndexId = findData.BaseFromKplerIndexId
  631. dataObj.Value = value
  632. dataObj.ModifyTime = time.Now()
  633. updateDataColsArr := make([]string, 0)
  634. updateDataColsArr = append(updateDataColsArr, "value")
  635. updateDataColsArr = append(updateDataColsArr, "modify_time")
  636. dataObj.Update(updateDataColsArr)
  637. }
  638. }
  639. }
  640. if len(addDataList) > 0 {
  641. err = dataObj.AddMulti(addDataList)
  642. if err != nil {
  643. err = fmt.Errorf("批量新增指标失败 Err:%s", err)
  644. return
  645. }
  646. }
  647. var dateItem *models.EdbInfoMaxAndMinInfo
  648. dateItem, err = dataObj.GetMaxAndMinDateByIndexCode(indexCode)
  649. if err != nil {
  650. err = fmt.Errorf("查询指标最新日期失败 Err:%s", err)
  651. return
  652. }
  653. go func() {
  654. indexObj.ModifyIndexMaxAndMinDate(indexCode, dateItem)
  655. }()
  656. // 同步刷新ETA指标库的指标
  657. {
  658. // 获取指标详情
  659. baseObj := new(models.BaseFromKpler)
  660. var edbInfo *models.EdbInfo
  661. edbInfo, err = models.GetEdbInfoByEdbCode(baseObj.GetSource(), indexCode)
  662. if err != nil {
  663. if !utils.IsErrNoRow(err) {
  664. errMsgList = append(errMsgList, fmt.Sprint("刷新ETA指标异常,指标编码:", indexCode, err.Error()))
  665. return
  666. } else {
  667. err = nil
  668. }
  669. }
  670. // 已经加入到指标库的话,那么就去更新ETA指标库吧
  671. if edbInfo != nil {
  672. go logic.RefreshBaseEdbInfo(edbInfo, ``)
  673. }
  674. }
  675. return
  676. }
  677. func HttpPost(url, postData string, params ...string) ([]byte, error) {
  678. body := ioutil.NopCloser(strings.NewReader(postData))
  679. client := &http.Client{}
  680. req, err := http.NewRequest("POST", url, body)
  681. if err != nil {
  682. return nil, err
  683. }
  684. contentType := "application/x-www-form-urlencoded;charset=utf-8"
  685. if len(params) > 0 && params[0] != "" {
  686. contentType = params[0]
  687. }
  688. req.Header.Set("Content-Type", contentType)
  689. req.Header.Set("authorization", utils.MD5(utils.APP_EDB_DATA_ANALYSIS+utils.EDB_DATA_ANALYSIS_Md5_KEY))
  690. resp, err := client.Do(req)
  691. if err != nil {
  692. return nil, err
  693. }
  694. defer resp.Body.Close()
  695. b, err := ioutil.ReadAll(resp.Body)
  696. utils.FileLog.Debug("HttpPost:" + string(b))
  697. return b, err
  698. }
  699. func getKplerFrequency(granularity string) (frequency string) {
  700. switch granularity {
  701. case "daily", "day":
  702. return "日度"
  703. case "weekly", "weeks":
  704. return "周度"
  705. case "monthly", "month":
  706. return "月度"
  707. case "yearly", "year":
  708. return "年度"
  709. }
  710. return ""
  711. }
  712. func InitKplerProduct ()(err error) {
  713. libUrl := "http://127.0.0.1:8915"
  714. libResp, err := getKplerProductLib(libUrl)
  715. if err != nil {
  716. return
  717. }
  718. if libResp.Ret != 200 {
  719. fmt.Println("获取开普勒产品库失败", libResp.ErrMsg)
  720. utils.FileLog.Info("获取开普勒产品库失败", libResp.ErrMsg)
  721. return
  722. }
  723. classifyList, err := models.GetAllBaseFromKplerClassify()
  724. if err != nil {
  725. return
  726. }
  727. classifyMap := make(map[int]*models.BaseFromKplerClassify)
  728. for _, v := range classifyList {
  729. classifyMap[v.ProductId] = v
  730. }
  731. // 批量插入分类表中
  732. // family > group(commodity) > product() > grade
  733. sort := 0
  734. for _, v := range libResp.Data {
  735. id, _ := strconv.Atoi(v.Id)
  736. if classifyObj, ok := classifyMap[id]; !ok {
  737. classifyObj = new(models.BaseFromKplerClassify)
  738. classifyObj.ClassifyName = v.Name
  739. classifyObj.ClassifyNameEn = v.Name
  740. classifyObj.ModifyTime = time.Now()
  741. classifyObj.CreateTime = time.Now()
  742. classifyObj.Sort = sort
  743. classifyObj.Level = 1
  744. classifyObj.ParentId = 0
  745. classifyObj.SysUserId = 0
  746. classifyObj.SysUserRealName = ""
  747. classifyObj.ProductId = id
  748. classifyObj.ProductName = v.Name
  749. classifyObj.AncestorFamilyId, _ = strconv.Atoi(v.FamilyId)
  750. classifyObj.AncestorFamilyName = v.Family
  751. classifyObj.AncestorGroupId, _ = strconv.Atoi(v.GroupId)
  752. classifyObj.AncestorGroupName = v.Group
  753. classifyObj.AncestorProductId, _ = strconv.Atoi(v.ProductId)
  754. classifyObj.AncestorProductName = v.Product
  755. classifyObj.AncestorGradeId, _ = strconv.Atoi(v.GradeId)
  756. classifyObj.AncestorGradeName = v.Grade
  757. classifyObj.ClassifyType = v.Type
  758. _, err = classifyObj.Add()
  759. if err != nil {
  760. fmt.Println("新增开普勒产品库失败", err)
  761. utils.FileLog.Info("新增开普勒产品库失败", err)
  762. }
  763. classifyMap[id] = classifyObj
  764. }
  765. }
  766. for _, v := range classifyMap {
  767. if v.ParentId == 0 {
  768. if v.AncestorGradeId > 0 && v.AncestorGradeName != v.ClassifyName {
  769. v.ParentId = classifyMap[v.AncestorGradeId].ClassifyId
  770. } else if v.AncestorProductId > 0 && v.AncestorProductName != v.ClassifyName {
  771. v.ParentId = classifyMap[v.AncestorProductId].ClassifyId
  772. } else if v.AncestorGroupId > 0 && v.AncestorGroupName != v.ClassifyName {
  773. v.ParentId = classifyMap[v.AncestorGroupId].ClassifyId
  774. }
  775. err = v.Update([]string{"ParentId"})
  776. if err != nil {
  777. fmt.Println("更新开普勒产品库失败", err)
  778. utils.FileLog.Info("更新开普勒产品库失败", err)
  779. }
  780. }
  781. }
  782. fmt.Println("classifyList:", classifyList)
  783. return
  784. }
  785. func InitKplerZone() (err error) {
  786. libUrl := "http://127.0.0.1:8915"
  787. libResp, err := getKplerZoneLib(libUrl)
  788. if err != nil {
  789. return
  790. }
  791. if libResp.Ret != 200 {
  792. fmt.Println("获取开普勒区域库失败", libResp.ErrMsg)
  793. utils.FileLog.Info("获取开普勒区域库失败", libResp.ErrMsg)
  794. return
  795. }
  796. // 批量插入区域表中
  797. zoneMap := make(map[int]*models.BaseFromKplerZone)
  798. for _, v := range libResp.Data {
  799. zoneObj := new(models.BaseFromKplerZone)
  800. descendantId, _ := strconv.Atoi(v.DescendantId)
  801. ancestorId, _ := strconv.Atoi(v.AncestorId)
  802. if _, ok := zoneMap[descendantId]; !ok {
  803. zoneObj.ZoneName = v.DescendantName
  804. //zoneObj.ZoneType = v.DescendantType
  805. zoneObj.AncestorId = ancestorId
  806. zoneObj.AncestorType = v.AncestorType
  807. zoneObj.AncestorName = v.AncestorName
  808. zoneObj.DescendantId = descendantId
  809. zoneObj.DescendantName = v.DescendantName
  810. zoneObj.ModifyTime = time.Now()
  811. zoneObj.CreateTime = time.Now()
  812. _, err = zoneObj.Add()
  813. if err != nil {
  814. fmt.Println("新增开普勒区域库失败", err)
  815. utils.FileLog.Info("新增开普勒区域库失败", err)
  816. }
  817. zoneMap[descendantId] = zoneObj
  818. if _, ok := zoneMap[ancestorId]; !ok {
  819. zoneObj := new(models.BaseFromKplerZone)
  820. zoneObj.ZoneName = v.AncestorName
  821. zoneObj.ZoneType = v.AncestorType
  822. zoneObj.DescendantId = ancestorId
  823. zoneObj.DescendantName = v.AncestorName
  824. zoneObj.ModifyTime = time.Now()
  825. zoneObj.CreateTime = time.Now()
  826. _, err = zoneObj.Add()
  827. if err != nil {
  828. fmt.Println("新增开普勒区域库失败", err)
  829. utils.FileLog.Info("新增开普勒区域库失败", err)
  830. }
  831. zoneMap[ancestorId] = zoneObj
  832. }
  833. }
  834. }
  835. return
  836. }