eta_crawler/models/com_trade_data.go

package models

import (
	"github.com/beego/beego/v2/client/orm"
	"github.com/shopspring/decimal"
	"time"
)

// ComTradeIndex 联合国商品贸易数据指标表
type ComTradeIndex struct {
	ComTradeId  int       `orm:"column(com_trade_id);pk"`
	IndexCode   string    `description:"联合国商品贸易编码"`
	IndexName   string    `description:"联合国商品名称"`
	IndexNameCn string    `description:"联合国商品名称（中文）"`
	CreateTime  time.Time `description:"创建时间"`
}

// GetComTradeIndexList 获取指标数据列表
func GetComTradeIndexList(condition string, pars []interface{}) (list []*ComTradeIndex, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := `SELECT * FROM com_trade_index WHERE 1=1 `
	if condition != "" {
		sql += condition
	}
	sql += `ORDER BY com_trade_id DESC `
	_, err = o.Raw(sql, pars).QueryRows(&list)
	return
}

// GetAllComTradeIndexList 获取指标数据列表
func GetAllComTradeIndexList() (list []*ComTradeIndex, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := `SELECT * FROM com_trade_index WHERE 1=1  ORDER BY com_trade_id DESC `
	_, err = o.Raw(sql).QueryRows(&list)
	return
}

// ComTradeCodeMapping 联合国商品贸易数据指标编码与分类编码的关系表
type ComTradeCodeMapping struct {
	Code       string    `orm:"column(code);pk"`
	IndexCode  string    `description:"联合国商品贸易编码"`
	Type       int       `description:"1:价格；2：重量"`
	Name       string    `description:"指标名称"`
	CreateTime time.Time `description:"创建时间"`
}

// GetAllComTradeCodeMappingList 获取联合国商品贸易数据指标编码与分类编码的关系列表
func GetAllComTradeCodeMappingList() (list []*ComTradeCodeMapping, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := `SELECT * FROM com_trade_code_mapping WHERE 1=1  ORDER BY create_time ASC `
	_, err = o.Raw(sql).QueryRows(&list)
	return
}

func AddComTradeCodeMapping(item *ComTradeCodeMapping) (err error) {
	o := orm.NewOrmUsingDB("data")
	_, err = o.Insert(item)
	if err != nil {
		return
	}
	return
}

func UpdateComTradeCodeMapping(item *ComTradeCodeMapping,cols []string) (err error) {
	o := orm.NewOrmUsingDB("data")
	_, err = o.Update(item, cols...)
	if err != nil {
		return
	}
	return
}

func AddMultiComTradeCodeMapping(items []*ComTradeCodeMapping) (err error) {
	o := orm.NewOrmUsingDB("data")
	_, err = o.InsertMulti(len(items), items)
	if err != nil {
		return
	}
	return
}

// ComTradeData 联合国商品贸易数据表
type ComTradeData struct {
	ComTradeDataId     int       `orm:"column(com_trade_data_id);pk"`
	ComTradeId         int       `description:"指标id"`
	IndexCode          string    `description:"联合国商品贸易编码"`
	Flow               string    `description:"贸易流向：X（Export：出口）；M（Import：进口）"`
	ReporterCode       int       `description:"出口国id"`
	ReporterName       string    `description:"出口国名称"`
	PartnerCode        int       `description:"进口国id"`
	PartnerName        string    `description:"进口国名称"`
	Partner2Code       int       `description:"第二进口国id"`
	Partner2Name       string    `description:"第二进口国名称"`
	DateType           int       `description:"日期类型，1：年度，2：月度"`
	DataTime           string    `description:"数据日期"`
	IndexTradeCode     string    `description:"离岸价编码"`
	TradeValue         float64   `description:"离岸价（美元）"`
	IndexNetWeightCode string    `description:"净重编码"`
	NetWeightValue     float64   `description:"净重/公斤"`
	ModifyTime         time.Time `description:"最新更新时间"`
	CreateTime         time.Time `description:"创建时间"`
}

// Update 更新基础信息
func (item *ComTradeData) Update(cols []string) (err error) {
	o := orm.NewOrmUsingDB("data")
	_, err = o.Update(item, cols...)
	return
}

//HandleComTradeData 数据处理
func HandleComTradeData(dataList []*ComTradeData, indexCode, flowCode, dataTime, partnerCode string) (err error) {
	o := orm.NewOrmUsingDB("data")
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			to.Rollback()
		} else {
			to.Commit()
		}
	}()
	var existList []*ComTradeData
	sql := `SELECT * FROM com_trade_data WHERE index_code = ? and flow = ? and partner_code = ? and data_time = ? order by data_time asc`
	_, err = to.Raw(sql, indexCode, flowCode, partnerCode, dataTime).QueryRows(&existList)
	if err != nil {
		return err
	}

	existMap := make(map[string]*ComTradeData)
	for _, v := range existList {
		existMap[v.IndexTradeCode] = v
	}

	// 待添加的数据列表
	addDataList := make([]*ComTradeData, 0)

	for _, v := range dataList {
		existData, ok := existMap[v.IndexTradeCode]
		if !ok {
			v.IndexTradeCode = v.IndexTradeCode
			v.IndexNetWeightCode = v.IndexNetWeightCode
			addDataList = append(addDataList, v)
		} else {
			existTradeValue := decimal.NewFromFloat(existData.TradeValue)
			existNetWeightValue := decimal.NewFromFloat(existData.NetWeightValue)
			nowTradeValue := decimal.NewFromFloat(v.TradeValue)
			nowNetWeightValue := decimal.NewFromFloat(v.NetWeightValue)

			updateColStr := make([]string, 0)
			//fmt.Println(date, "=========", nowVal.Equal(valDecimal))
			if !existTradeValue.Equal(nowTradeValue) {
				existData.TradeValue = v.TradeValue
			}
			if !existNetWeightValue.Equal(nowNetWeightValue) {
				existData.NetWeightValue = v.NetWeightValue
			}
			if len(updateColStr) > 0 {
				_, err = to.Update(existData, updateColStr...)
				if err != nil {
					return
				}
			}
		}
		delete(existMap, v.DataTime)
	}

	lenAddDataList := len(addDataList)
	if lenAddDataList > 0 {
		_, err = to.InsertMulti(lenAddDataList, addDataList)
	}

	//existMapNum := len(existMap)
	//if existMapNum > 0 {
	//	idList := make([]int, 0)
	//	for _, v := range existMap {
	//		idList = append(idList, v.BaseFromEiaSteoDataId)
	//	}
	//
	//	sql = `DELETE FROM base_from_eia_steo_data WHERE base_from_eia_steo_data_id in(` + utils.GetOrmInReplace(existMapNum) + `) `
	//	_, err = to.Raw(sql, idList).Exec()
	//	if err != nil {
	//		return err
	//	}
	//}

	// 更新指标信息
	sql = `UPDATE com_trade_index SET update_time = now() WHERE index_code = ? `
	_, err = to.Raw(sql, indexCode).Exec()
	if err != nil {
		return err
	}
	//{
	//	isUpdateIndexDate := false
	//	if endDate.After(indexInfo.EndDate) {
	//		isUpdateIndexDate = true
	//		indexInfo.EndDate = endDate
	//	}
	//	if startDate.Before(indexInfo.StartDate) {
	//		isUpdateIndexDate = true
	//		indexInfo.StartDate = startDate
	//	}
	//	if isUpdateIndexDate {
	//		_, err = to.Update(indexInfo, "StartDate", "EndDate")
	//		if err != nil {
	//			return
	//		}
	//	}
	//}
	return
}

// GetAllComTradeDataList 获取联合国商品贸易数据指标编码与分类编码的关系列表
func GetAllComTradeDataList() (list []*ComTradeData, err error) {
	o := orm.NewOrmUsingDB("data")
	//sql := `SELECT * FROM com_trade_data WHERE 1=1  ORDER BY create_time ASC `
	sql := `SELECT  * FROM com_trade_data WHERE 1=1 GROUP BY index_trade_code`
	_, err = o.Raw(sql).QueryRows(&list)
	return
}

// UpdateComTradeData 更新
func UpdateComTradeData(indexTradeCode, indexNetWeightCode, oldIndexTradeCode string) (err error) {
	o := orm.NewOrmUsingDB("data")
	sql := `UPDATE com_trade_data SET index_trade_code = ?,index_net_weight_code=?  WHERE index_trade_code = ? ORDER BY create_time ASC `
	_, err = o.Raw(sql, indexTradeCode, indexNetWeightCode, oldIndexTradeCode).Exec()
	return
}