base_from_kpler.go 31 KB

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