Merge branch 'hotfix/pcsg_bloomberg_0508'

controllers/base_from_bloomberg.go

@@ -271,3 +271,49 @@ func (this *BloombergController) PCSGRefreshMonthly() {
 	br.Success = true
 	br.Msg = "操作成功"
 }
+
+// PCSGImportHistoryData
+// @Title 中石油新加坡-导入历史数据
+// @Description  中石油新加坡-导入历史数据
+// @Success 200 {object} models.PCSGImportHistoryDataReq
+// @router /pcsg/import_history_data [post]
+func (this *BloombergController) PCSGImportHistoryData() {
+	br := new(models.BaseResponse).Init()
+	defer func() {
+		if br.ErrMsg == "" {
+			br.IsSendEmail = false
+		}
+		this.Data["json"] = br
+		this.ServeJSON()
+	}()
+	var req models.PCSGImportHistoryDataReq
+	if e := json.Unmarshal(this.Ctx.Input.RequestBody, &req); e != nil {
+		br.Msg = "参数解析异常！"
+		br.ErrMsg = "参数解析失败，Err:" + e.Error()
+		return
+	}
+
+	var indexes []models.BaseFromBloombergApiIndexAndData
+	var index models.BaseFromBloombergApiIndexAndData
+	var indexData []models.BaseFromBloombergApiIndexData
+	index.IndexCode = req.IndexCode
+	for k, v := range req.DataMap {
+		indexData = append(indexData, models.BaseFromBloombergApiIndexData{
+			DataTime: k,
+			Value:    v,
+		})
+	}
+	index.Data = indexData
+	indexes = append(indexes, index)
+
+	// 写入数据
+	if e := services.PCSGWrite2BaseBloomberg(indexes); e != nil {
+		br.Msg = "刷新失败"
+		br.ErrMsg = "PCSG-写入Bloomberg数据源失败, Err: " + e.Error()
+		return
+	}
+
+	br.Ret = 200
+	br.Success = true
+	br.Msg = "操作成功"
+}

logic/base_edb_info.go

@@ -29,6 +29,8 @@ func RefreshBaseEdbInfo(edbInfo *models.EdbInfo, startDate string) (isHandling b
 		err = models.RefreshEdbDataFromMysteelChemical(edbInfo.EdbInfoId, edbInfo.EdbCode, startDate)
 	case utils.DATA_SOURCE_YS:
 		err = models.RefreshEdbDataFromSmm(edbInfo.EdbInfoId, edbInfo.EdbCode, startDate)
+	case utils.DATA_SOURCE_BLOOMBERG:
+		err = models.RefreshEdbDataFromBloomberg(edbInfo.EdbInfoId, edbInfo.EdbCode, startDate)
 	default:
 		return
 	}

models/base_from_bloomberg.go

@@ -43,6 +43,12 @@ func (m *BaseFromBloombergIndex) Create() (err error) {
 	return
 }
 
+func (m *BaseFromBloombergIndex) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	_, err = o.Update(m, cols...)
+	return
+}
+
 func GetBaseFromBloombergIndexByCode(indexCode string) (item *BaseFromBloombergIndex, err error) {
 	o := orm.NewOrm()
 	sql := `SELECT * FROM base_from_bloomberg_index WHERE index_code = ? LIMIT 1`
@@ -80,7 +86,7 @@ func GetBaseFromBloombergDataByCondition(condition string, pars []interface{}) (
 func MultiInsertOrUpdateBaseFromBloombergData(inserts, updates []*BaseFromBloombergData) (err error) {
 	o := orm.NewOrm()
 	if len(inserts) > 0 {
-		_, e := o.InsertMulti(len(inserts), inserts)
+		_, e := o.InsertMulti(1000, inserts)
 		if e != nil {
 			err = fmt.Errorf("insert multi err: %s", e.Error())
 			return
@@ -345,3 +351,9 @@ type BaseFromBloombergApiIndexData struct {
 	DataTime time.Time `description:"数据日期"`
 	Value    float64   `description:"数据值"`
 }
+
+// PCSGImportHistoryDataReq 导入历史数据
+type PCSGImportHistoryDataReq struct {
+	IndexCode string                `description:"指标编码"`
+	DataMap   map[time.Time]float64 `description:"数据日期/值"`
+}

routers/commentsRouter.go

@@ -88,6 +88,15 @@ func init() {
             Filters: nil,
             Params: nil})
 
+    beego.GlobalControllerRouter["eta/eta_index_lib/controllers:BloombergController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:BloombergController"],
+        beego.ControllerComments{
+            Method: "PCSGImportHistoryData",
+            Router: `/pcsg/import_history_data`,
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
     beego.GlobalControllerRouter["eta/eta_index_lib/controllers:BloombergController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:BloombergController"],
         beego.ControllerComments{
             Method: "PCSGRefreshDaily",

services/base_from_pcsg.go

@@ -213,6 +213,10 @@ func PCSGWrite2BaseBloomberg(indexes []models.BaseFromBloombergApiIndexAndData)
 		if v.IndexCode == "" {
 			continue
 		}
+		// 无数据的情况不处理
+		if len(v.Data) == 0 {
+			continue
+		}
 
 		// 指标是否存在
 		index, e := models.GetBaseFromBloombergIndexByCode(v.IndexCode)
@@ -236,6 +240,18 @@ func PCSGWrite2BaseBloomberg(indexes []models.BaseFromBloombergApiIndexAndData)
 				return
 			}
 			index = newIndex
+		} else {
+			// 无指标名称的情况下更新指标基础信息
+			if index.IndexName == "" {
+				index.IndexName = v.IndexName
+				index.Unit = v.Unit
+				index.Frequency = v.Frequency
+				index.ModifyTime = time.Now().Local()
+				if e = index.Update([]string{"IndexName", "Unit", "Frequency", "ModifyTime"}); e != nil {
+					err = fmt.Errorf("更新Bloomberg原始指标失败, err: %s", e.Error())
+					return
+				}
+			}
 		}
 
 		// 更新指标数据
@@ -249,6 +265,7 @@ func PCSGWrite2BaseBloomberg(indexes []models.BaseFromBloombergApiIndexAndData)
 			return
 		}
 		dateExist := make(map[string]*models.BaseFromBloombergData)
+		newValExist := make(map[string]bool)
 		if len(indexData) > 0 {
 			for _, d := range indexData {
 				strDate := d.DataTime.Format(utils.FormatDate)
@@ -270,6 +287,11 @@ func PCSGWrite2BaseBloomberg(indexes []models.BaseFromBloombergApiIndexAndData)
 				originData.ModifyTime = time.Now().Local()
 				updateData = append(updateData, originData)
 			} else {
+				// 新增的数据去重
+				if newValExist[strDate] {
+					continue
+				}
+				newValExist[strDate] = true
 				newData := new(models.BaseFromBloombergData)
 				newData.BaseFromBloombergIndexId = index.BaseFromBloombergIndexId
 				newData.IndexCode = index.IndexCode