Merge branch 'bzq/eiaSteo_custom' of eta_server/eta_task into master

.gitignore

@@ -15,3 +15,4 @@ eta_task
 /README.md
 /eta_task.exe~
 /*.mod
+/test/

models/data_manage/base_from_eia_steo.go

@@ -1,8 +1,11 @@
 package data_manage
 
 import (
-	"github.com/beego/beego/v2/client/orm"
+	"eta/eta_task/utils"
+
 	"time"
+
+	"github.com/beego/beego/v2/client/orm"
 )
 
 // BaseFromEiaSteoIndex EiaSteo指标
@@ -56,6 +59,13 @@ func GetBaseFromEiaSteoIndexAll(dateStr string) (list []*BaseFromEiaSteoIndex, e
 	return
 }
 
+func GetBaseFromEiaSteoIndexCodeListByDate(dateStr string) (list []string, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT index_code FROM base_from_eia_steo_index WHERE modify_time>=?`
+	_, err = o.Raw(sql, dateStr).QueryRows(&list)
+	return
+}
+
 // Update 更新作者基础信息
 func (item *BaseFromEiaSteoIndex) Update(cols []string) (err error) {
 	o := orm.NewOrm()
@@ -68,6 +78,8 @@ type BaseFromEiaSteoClassify struct {
 	BaseFromEiaSteoClassifyId int       `orm:"column(base_from_eia_steo_classify_id);pk"`
 	ClassifyName              string    `description:"分类名称（中文名称）"`
 	ClassifyNameOriginal      string    `description:"分类名称（原始名称）"`
+	ParentId                  int       `description:"父级id"`
+	Level                     int       `description:"层级"`
 	ModifyTime                time.Time `description:"最新更新时间"`
 	CreateTime                time.Time `description:"创建时间"`
 }
@@ -94,6 +106,12 @@ type BaseFromEiaSteoData struct {
 	CreateTime             time.Time `description:"创建时间"`
 }
 
+func (b *BaseFromEiaSteoData) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	_, err = o.Update(b, cols...)
+	return
+}
+
 func AddBaseFromEiaSteoData(item *BaseFromEiaSteoData) (lastId int64, err error) {
 	o := orm.NewOrm()
 	lastId, err = o.Insert(item)
@@ -108,6 +126,28 @@ func GetBaseFromEiaSteoClassifyAll() (list []*BaseFromEiaSteoClassify, err error
 	return
 }
 
+func AddBaseEiaSteoClassify(item *BaseFromEiaSteoClassify) (lastId int64, err error) {
+	o := orm.NewOrm()
+	lastId, err = o.Insert(item)
+	return
+}
+
+func (item *BaseFromEiaSteoClassify) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	_, err = o.Update(item, cols...)
+	return
+}
+
+func BatchDeleteBaseFromEiaSteoClassify(id []int) (err error) {
+	if len(id) == 0 {
+		return
+	}
+	o := orm.NewOrm()
+	sql := `DELETE FROM base_from_eia_steo_classify WHERE base_from_eia_steo_classify_id IN (` + utils.GetOrmInReplace(len(id)) + `)`
+	_, err = o.Raw(sql, id).Exec()
+	return
+}
+
 type EiaSteoIndexResp struct {
 	Ret     int
 	Msg     string
@@ -124,6 +164,14 @@ type EiaSteoIndexDataResp struct {
 	Data    []*BaseFromEiaSteoData
 }
 
+type EiaSteoClassifyDataResp struct {
+	Ret     int
+	Msg     string
+	ErrMsg  string
+	ErrCode string
+	Data    []*BaseFromEiaSteoClassify
+}
+
 func GetBaseFromEiaSteoIndexDataAll(dateStr string) (list []*BaseFromEiaSteoData, err error) {
 	o := orm.NewOrm()
 	sql := `SELECT * FROM base_from_eia_steo_data WHERE create_time>=?`
@@ -131,9 +179,30 @@ func GetBaseFromEiaSteoIndexDataAll(dateStr string) (list []*BaseFromEiaSteoData
 	return
 }
 
+func GetBaseFromEiaSteoIndexDataByIndexCode(indexCode, dateStr string) (list []*BaseFromEiaSteoData, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT * FROM base_from_eia_steo_data WHERE index_code=? AND modify_time>=?`
+	_, err = o.Raw(sql, indexCode, dateStr).QueryRows(&list)
+	return
+}
+
+func GetBaseFromEiaSteoIndexDataByDate(dateStr string) (list []*BaseFromEiaSteoData, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT * FROM base_from_eia_steo_data WHERE modify_time>=?`
+	_, err = o.Raw(sql, dateStr).QueryRows(&list)
+	return
+}
+
 func GetBaseFromEiaSteoIndexMaxDate() (max_date time.Time, err error) {
 	o := orm.NewOrm()
 	sql := ` SELECT max(a.data_time)as max_date FROM base_from_eia_steo_data as a `
 	err = o.Raw(sql).QueryRow(&max_date)
 	return
 }
+
+func GetBaseFromEiaSteoIndexMaxCreateDate() (max_date time.Time, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT max(a.create_time)as max_date FROM base_from_eia_steo_index as a `
+	err = o.Raw(sql).QueryRow(&max_date)
+	return
+}

models/db.go

@@ -6,9 +6,10 @@ import (
 	"eta/eta_task/models/data_manage/future_good"
 	"eta/eta_task/models/report"
 	"eta/eta_task/utils"
-	_ "github.com/go-sql-driver/mysql"
 	"time"
 
+	_ "github.com/go-sql-driver/mysql"
+
 	"github.com/beego/beego/v2/client/orm"
 )
 
@@ -114,6 +115,7 @@ func initEdbDataTable() {
 		new(data_manage.BaseFromTradeIneIndex),
 		new(data_manage.BaseFromChangesVisitorsCovid),
 		new(data_manage.BaseFromEiaSteoIndex),
+		new(data_manage.BaseFromEiaSteoClassify),
 		new(data_manage.BaseFromEiaSteoData),
 		new(data_manage.ComTradeIndex),           // 联合国指标表
 		new(data_manage.ComTradeCodeMapping),     // 联合国指标关系表

services/eia_steo.go

@@ -11,7 +11,7 @@ import (
 // SyncEiaSteoIndex EIA STEO报告 指标
 func SyncEiaSteoIndex() (err error) {
 	var startDate string
-	maxDate, err := data_manage.GetBaseFromEiaSteoIndexMaxDate()
+	maxDate, err := data_manage.GetBaseFromEiaSteoIndexMaxCreateDate()
 	if err != nil || maxDate.IsZero() {
 		startDate = time.Now().AddDate(0, 0, -1).Format(utils.FormatDate) + " 00:00:00"
 	} else {
@@ -24,6 +24,11 @@ func SyncEiaSteoIndex() (err error) {
 	data["StartDate"] = startDate
 	//data["EndDate"] = endDate
 	result, err := HttpPost("SyncEiaSteoIndex", method, data)
+	if err != nil {
+		fmt.Println("HttpPost err:", err)
+		utils.FileLog.Info("HttpPost err:", err)
+		return
+	}
 	utils.FileLog.Info(result)
 	fmt.Println(result)
 
@@ -47,6 +52,9 @@ func SyncEiaSteoIndex() (err error) {
 
 	for _, zv := range respObj.Data {
 		//if _, ok := existIndexMap[zv.IndexCode]; !ok {
+		if zv.BaseFromEiaSteoIndexId <= 0 {
+			continue
+		}
 		newID, err := data_manage.InsertOrUpdateBaseFromEiaSteoIndex(zv)
 		if err != nil {
 			fmt.Println("InsertOrUpdateBaseFromEiaSteoIndex error:", err)
@@ -78,6 +86,11 @@ func SyncEiaSteoIndexData() (err error) {
 
 		var result string
 		result, err = HttpPost("SyncEiaSteoIndexData", method, data)
+		if err != nil {
+			fmt.Println("HttpPost err:", err)
+			utils.FileLog.Info("HttpPost err:", err)
+			return
+		}
 		utils.FileLog.Info(result)
 		fmt.Println(result)
 
@@ -110,3 +123,155 @@ func SyncEiaSteoIndexData() (err error) {
 	}
 	return err
 }
+
+// EIA STEO报告-分类
+func SyncEiaSteoClassify() (err error) {
+	data := make(map[string]interface{})
+	data["Source"] = utils.DATA_SOURCE_EIA_STEO
+	method := `classify/list`
+	result, err := HttpPost("SyncEiaSteoClassify", method, data)
+	if err != nil {
+		fmt.Println("SyncEiaSteoClassify HttpPost Err:", err.Error())
+		return err
+	}
+	utils.FileLog.Info("SyncEiaSteoClassify:" + result)
+	fmt.Println("SyncEiaSteoClassify result:", result)
+
+	respObj := new(data_manage.EiaSteoClassifyDataResp)
+	err = json.Unmarshal([]byte(result), &respObj)
+	if err != nil {
+		return err
+	}
+	//获取所有指标信息  某一天的
+	allClassify, err := data_manage.GetBaseFromEiaSteoClassifyAll()
+	if err != nil {
+		return
+	}
+
+	existClassifyMap := make(map[int]*data_manage.BaseFromEiaSteoClassify)
+	for _, v := range allClassify {
+		existClassifyMap[v.BaseFromEiaSteoClassifyId] = v
+	}
+
+	for _, item := range respObj.Data {
+		if oldClassify, ok := existClassifyMap[item.BaseFromEiaSteoClassifyId]; !ok {
+			newID, err := data_manage.AddBaseEiaSteoClassify(item)
+			if err != nil {
+				fmt.Println("InsertOrUpdateBaseEiaSteoClassify error:", err)
+			}
+			fmt.Println("InsertOrUpdateBaseEiaSteoClassify new indexID:", newID)
+		} else {
+			var upDateCols []string
+			if oldClassify.ClassifyName != item.ClassifyName {
+				oldClassify.ClassifyName = item.ClassifyName
+				upDateCols = append(upDateCols, "ClassifyName")
+			}
+			if oldClassify.ClassifyNameOriginal != item.ClassifyNameOriginal {
+				oldClassify.ClassifyNameOriginal = item.ClassifyNameOriginal
+				upDateCols = append(upDateCols, "ClassifyNameOriginal")
+			}
+			if oldClassify.ParentId != item.ParentId {
+				oldClassify.ParentId = item.ParentId
+				upDateCols = append(upDateCols, "ParentId")
+			}
+			if oldClassify.Level != item.Level {
+				oldClassify.Level = item.Level
+				upDateCols = append(upDateCols, "Level")
+			}
+			if len(upDateCols) > 0 {
+				oldClassify.ModifyTime = time.Now()
+				upDateCols = append(upDateCols, "ModifyTime")
+				err := oldClassify.Update(upDateCols)
+				if err != nil {
+					fmt.Println("UpdateBaseFromEiaSteoClassify error:", err)
+				}
+			}
+			delete(existClassifyMap, oldClassify.BaseFromEiaSteoClassifyId)
+		}
+	}
+	var deleteId []int
+	for k, _ := range existClassifyMap {
+		deleteId = append(deleteId, k)
+	}
+	err = data_manage.BatchDeleteBaseFromEiaSteoClassify(deleteId)
+	if err != nil {
+		fmt.Println("BatchDeleteBaseFromEiaSteoClassify error:", err)
+	}
+	return err
+}
+
+func SyncEiaSteoIndexDataV2() (err error) {
+	startDate := time.Now().AddDate(0, 0, -1).Format(utils.FormatDate) + " 00:00:00"
+
+	method := `index/data/list`
+
+	//获取所有指标信息  某一天的
+	allIndexCode, err := data_manage.GetBaseFromEiaSteoIndexCodeListByDate(startDate)
+	if err != nil {
+		fmt.Println("get GetBaseFromEiaSteoIndexAll err:" + err.Error())
+		utils.FileLog.Info("get GetBaseFromEiaSteoIndexCodeList err:", err)
+		return
+	}
+
+	for _, indexCode := range allIndexCode {
+		data := make(map[string]interface{})
+		data["Source"] = utils.DATA_SOURCE_EIA_STEO
+		data["StartDate"] = startDate
+		data["IndexCode"] = indexCode
+
+		var result string
+		result, err = HttpPost("SyncEiaSteoIndexData", method, data)
+		if err != nil {
+			fmt.Println("HttpPost err:", err)
+			utils.FileLog.Info("HttpPost err:", err)
+			return
+		}
+		utils.FileLog.Info(result)
+		fmt.Println(result)
+
+		respObj := new(data_manage.EiaSteoIndexDataResp)
+		err = json.Unmarshal([]byte(result), &respObj)
+		if err != nil {
+			fmt.Println("json.Unmarshal err:" + err.Error())
+			return err
+		}
+
+		existDataMap := make(map[int]*data_manage.BaseFromEiaSteoData)
+		allData, err1 := data_manage.GetBaseFromEiaSteoIndexDataByIndexCode(indexCode, startDate)
+		if err1 != nil {
+			utils.FileLog.Error("get GetBaseFromEiaSteoIndexDataByDate err:", err1.Error())
+			return
+		}
+		for _, dv := range allData {
+			existDataMap[dv.BaseFromEiaSteoDataId] = dv
+		}
+
+		for _, dv := range respObj.Data {
+			if v, ok := existDataMap[dv.BaseFromEiaSteoDataId]; !ok {
+				newID, err := data_manage.AddBaseFromEiaSteoData(dv)
+				if err != nil {
+					utils.FileLog.Error("add error:", err)
+				}
+				fmt.Println("insert new indexID:", newID)
+			} else {
+				upDateCols := make([]string, 0)
+				if !v.DataTime.Equal(dv.DataTime) {
+					v.DataTime = dv.DataTime
+					upDateCols = append(upDateCols, "data_time")
+				}
+				if v.Value != dv.Value {
+					v.Value = dv.Value
+					v.ModifyTime = dv.ModifyTime
+					upDateCols = append(upDateCols, []string{"value", "modify_time"}...)
+				}
+				if len(upDateCols) > 0 {
+					err = v.Update(upDateCols)
+					if err != nil {
+						utils.FileLog.Error("update error:", err)
+					}
+				}
+			}
+		}
+	}
+	return err
+}

services/sync_hz_data.go

@@ -60,6 +60,11 @@ func SyncHzDataIndex(cont context.Context) (err error) {
 	}
 
 	// EiaSteo
+	err = SyncEiaSteoClassify()
+	if err != nil {
+		fmt.Println("SyncEiaSteoClassify Err:" + err.Error())
+		return
+	}
 	err = SyncEiaSteoIndex()
 	if err != nil {
 		fmt.Println("SyncEiaSteoIndex Err:" + err.Error())
@@ -140,7 +145,7 @@ func SyncHzDataIndex(cont context.Context) (err error) {
 func SyncHzDataIndexData() {
 	var err error
 	//EiaSteo
-	err = SyncEiaSteoIndexData()
+	err = SyncEiaSteoIndexDataV2()
 	if err != nil {
 		fmt.Println("SyncEiaSteoIndexData Err:" + err.Error())
 		return