base_from_kpler.go 28 KB

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