base_from_kpler.go 28 KB

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