hongze_task/models/data_manage/edb_data_lz.go

package data_manage

import (
	"fmt"
	"hongze/hongze_task/utils"
	"github.com/beego/beego/v2/client/orm"
	"strconv"
	"strings"
	"time"
)

type lzSurveyData struct {
	DataTime   string `orm:"column(data_time)" description:"日期"`
	InputValue string `orm:"column(input_value)" description:"值"`
}

type LongzhongSurveyData struct {
	SurveyDataId         int `orm:"column(survey_data_id);pk"`
	SurveyProductId      int
	ProjectQuotaId       int64
	BreedId              string
	BreedName            string
	QuotaId              string
	QuotaName            string
	UnitId               string
	UnitName             string
	SampleType           int64
	SampleId             string
	SampleName           string
	DeviceId             string
	Device               string
	ProductCraftId       string
	ProductCraft         string
	ProductLine          string
	InputMode            int64
	Frequency            int64
	InputValue           string
	TaskShouldFinishTime int64
	CustomId             string
	CustomType           int64
	Custom               string
	QuotaSampleId        int64
	TaskActualFinishTime int64
	AreaName             string
	ProvinceName         string
	ResearchStartData    int64
	ResearchStopData     int64
	DataTime             string
}

func GetLzSurveyDataByTradeCode(condition string, pars []interface{}) (item []*lzSurveyData, err error) {
	sql := ` SELECT  a.* FROM longzhong_survey_data AS a
				INNER JOIN longzhong_survey_product AS b ON a.survey_product_id=b.survey_product_id
				WHERE 1=1 `
	o := orm.NewOrmUsingDB("data")
	if condition != "" {
		sql += condition
	}
	sql += ` ORDER BY a.data_time DESC `
	_, err = o.Raw(sql, pars).QueryRows(&item)
	return
}

func GetLzSurveyDataExistByTradeCode(surveyProductId int) (items []*LongzhongSurveyData, err error) {
	sql := ` SELECT  * FROM longzhong_survey_data WHERE 1=1 AND survey_product_id=? `
	o := orm.NewOrmUsingDB("data")
	_, err = o.Raw(sql, surveyProductId).QueryRows(&items)
	return
}

func GetEdbDataLzByCodeAndDate(edbCode string, startDate string) (count int, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := ` SELECT COUNT(1) AS count FROM edb_data_lz WHERE edb_code=? AND data_time=? `
	err = o.Raw(sql, edbCode, startDate).QueryRow(&count)
	return
}

func ModifyEdbDataLz(edbInfoId int64, dataTime, value string) (err error) {
	o := orm.NewOrmUsingDB("data")
	sql := ` UPDATE edb_data_lz SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
	_, err = o.Raw(sql, value, edbInfoId, dataTime).Exec()
	return
}

//刷新隆众指标数据
func RefreshEdbDataByLz(edbInfoId int, edbCode, startDate, endDate string) (err error) {
	o := orm.NewOrmUsingDB("data")
	tx,err:=o.Begin()
	defer func() {
		if err != nil {
			tx.Rollback()
		} else {
			tx.Commit()
		}
	}()

	if err != nil {
		return
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		condition += " AND b.lz_code=? "
		pars = append(pars, edbCode)
	}

	if startDate != "" {
		condition += " AND a.data_time>=? "
		pars = append(pars, startDate)
	}

	if endDate != "" {
		condition += " AND a.data_time<=? "
		pars = append(pars, endDate)
	}

	lzDataList, err := GetLzSurveyDataByTradeCode(condition, pars)

	//获取已存在指标所有数据
	existDataList := make([]*EdbDataBase, 0)
	dataTableName := GetEdbDataTableName(utils.DATA_SOURCE_LZ)
	sql := `SELECT * FROM %s WHERE edb_info_id=? `
	sql = fmt.Sprintf(sql, dataTableName)
	_, err = o.Raw(sql, edbInfoId).QueryRows(&existDataList)
	if err != nil {
		return err
	}
	existDataMap := make(map[string]string)
	for _, v := range existDataList {
		existDataMap[v.DataTime] = v.Value
	}

	addSql := ` INSERT INTO edb_data_lz(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values `
	var isAdd bool
	addExistMap := make(map[string]string)
	for _, v := range lzDataList {
		item := v
		if _, ok := addExistMap[v.DataTime]; !ok {
			if existVal, existOk := existDataMap[v.DataTime]; !existOk {
				eDate := item.DataTime
				sValue := item.InputValue
				if sValue != "" {
					dataTime, err := time.Parse(utils.FormatDate, eDate)
					if err != nil {
						return err
					}
					timestamp := dataTime.UnixNano() / 1e6
					timeStr := fmt.Sprintf("%d", timestamp)
					addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, sValue)
					isAdd = true
				}
			} else {
				if existVal != item.InputValue {
					sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
					sql = fmt.Sprintf(sql, dataTableName)
					_, err = tx.Raw(sql, item.InputValue, edbInfoId, v.DataTime).Exec()
					if err != nil {
						return err
					}
				}
			}
		}
		addExistMap[v.DataTime] = v.InputValue
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = tx.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}