GtDataStore/app/model/dcitemhistorydatamodel.go

package model

import (
	"GtDataStore/app/cmd/organization/pb"
	"context"
	"database/sql"
	"fmt"
	"github.com/zeromicro/go-zero/core/stores/sqlc"
	"github.com/zeromicro/go-zero/core/stores/sqlx"
	"strings"
	"time"
)

var _ DcItemHistoryDataModel = (*customDcItemHistoryDataModel)(nil)

type (
	// DcItemHistoryDataModel is an interface to be customized, add more methods here,
	// and implement the added methods in customDcItemHistoryDataModel.
	DcItemHistoryDataModel interface {
		dcItemHistoryDataModel
		MultiInsert(ctx context.Context, projectId int64, datas []*pb.ItemHistoryData) (int64, error)
		QueryHistoryDataByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) ([]*ItemHistoryData, error)
		QueryHistoryDataMaxMinByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) (*MaxMinData, error)
		QueryHistoryDataFirstByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) (*ItemHistoryData, error)
		QueryHistoryDataLastByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) (*ItemHistoryData, error)
	}

	customDcItemHistoryDataModel struct {
		*defaultDcItemHistoryDataModel
	}

	MaxMinData struct {
		MaxVal sql.NullFloat64 `db:"max_val"` // 最大值
		MinVal sql.NullFloat64 `db:"min_val"` // 最小值
	}

	ItemHistoryData struct {
		ItemName string    `db:"item_name"` // 点位名
		Val      float64   `db:"val"`       // 值
		HTime    time.Time `db:"h_time"`    // 采集数据时间
	}
)

// NewDcItemHistoryDataModel returns a model for the database table.
func NewDcItemHistoryDataModel(conn sqlx.SqlConn) DcItemHistoryDataModel {
	return &customDcItemHistoryDataModel{
		defaultDcItemHistoryDataModel: newDcItemHistoryDataModel(conn),
	}
}

func (m *defaultDcItemHistoryDataModel) MultiInsert(ctx context.Context, projectId int64, datas []*pb.ItemHistoryData) (int64, error) {
	query := fmt.Sprintf("insert ignore into %s (`item_name`,`val`,`h_time`) values (?, ?, ?)", m.getTableName(projectId))

	if bulk, err := sqlx.NewBulkInserter(m.conn, query); err == nil {
		for _, data := range datas {
			if err = bulk.Insert(data.ItemName, data.Val, data.HTime); err != nil {
				return 0, err
			}
		}

		bulk.Flush()
		return int64(len(datas)), nil
	} else {
		return 0, err
	}
}

func (m *defaultDcItemHistoryDataModel) QueryHistoryDataByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) ([]*ItemHistoryData, error) {
	resp := make([]*ItemHistoryData, 0)
	var err error
	if strings.Index(in.ItemName, ",") > 0 {
		ItemNames := strings.Split(in.ItemName, ",")
		var ItemNameStrs []string
		for _, val := range ItemNames {
			ItemNameStrs = append(ItemNameStrs, "'"+val+"'")
		}
		inClause := strings.Join(ItemNameStrs, ",")
		query := fmt.Sprintf("SELECT hd.* FROM %s AS hd "+
			" LEFT JOIN dc_item_config as ic ON hd.item_name = ic.item_name "+
			" WHERE hd.item_name in (%s) AND h_time BETWEEN ? AND ? AND (ic.id IS NULL OR (hd.val > ic.min_val AND hd.val < ic.max_val))", m.getTableName(in.ProjectId), inClause)
		err = m.conn.QueryRowsCtx(ctx, &resp, query, in.Stime, in.Etime)
	} else {
		query := fmt.Sprintf("SELECT hd.* FROM %s AS hd "+
			" LEFT JOIN dc_item_config as ic ON hd.item_name = ic.item_name "+
			" WHERE hd.item_name in (?) AND h_time BETWEEN ? AND ? AND (ic.id IS NULL OR (hd.val > ic.min_val AND hd.val < ic.max_val))", m.getTableName(in.ProjectId))
		err = m.conn.QueryRowsCtx(ctx, &resp, query, in.ItemName, in.Stime, in.Etime)
	}

	switch err {
	case nil:
		return resp, nil
	case sqlc.ErrNotFound:
		return nil, ErrNotFound
	default:
		return nil, err
	}
}

func (m *defaultDcItemHistoryDataModel) QueryHistoryDataMaxMinByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) (*MaxMinData, error) {
	resp := &MaxMinData{}
	var err error
	query := fmt.Sprintf("SELECT max(hd.val) as max_val, min(hd.val) as min_val FROM %s AS hd "+
		" LEFT JOIN dc_item_config as ic ON hd.item_name = ic.item_name "+
		" WHERE hd.item_name = ? AND h_time BETWEEN ? AND ? AND (ic.id IS NULL OR (hd.val > ic.min_val AND hd.val < ic.max_val))", m.getTableName(in.ProjectId))
	err = m.conn.QueryRowCtx(ctx, resp, query, in.ItemName, in.Stime, in.Etime)

	switch err {
	case nil:
		if !resp.MaxVal.Valid {
			resp.MaxVal.Float64 = 0
		}
		if !resp.MinVal.Valid {
			resp.MinVal.Float64 = 0
		}
		return resp, nil
	case sqlc.ErrNotFound:
		return nil, ErrNotFound
	default:
		return nil, err
	}
}

func (m *defaultDcItemHistoryDataModel) QueryHistoryDataFirstByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) (*ItemHistoryData, error) {
	resp := &ItemHistoryData{}
	var err error
	query := fmt.Sprintf("SELECT hd.* FROM %s AS hd "+
		" LEFT JOIN dc_item_config as ic ON hd.item_name = ic.item_name "+
		" WHERE hd.item_name = ? AND h_time BETWEEN ? AND ? AND (ic.id IS NULL OR (hd.val > ic.min_val AND hd.val < ic.max_val))"+
		" Order by h_time asc limit 1", m.getTableName(in.ProjectId))
	err = m.conn.QueryRowCtx(ctx, resp, query, in.ItemName, in.Stime, in.Etime)

	switch err {
	case nil:
		return resp, nil
	case sqlc.ErrNotFound:
		return nil, ErrNotFound
	default:
		return nil, err
	}
}

func (m *defaultDcItemHistoryDataModel) QueryHistoryDataLastByTime(ctx context.Context, in *pb.ItemHistoryDataByTimeReq) (*ItemHistoryData, error) {
	resp := &ItemHistoryData{}
	var err error
	if in.Stime != "" && in.Etime != "" {
		query := fmt.Sprintf("SELECT hd.* FROM %s AS hd "+
			" LEFT JOIN dc_item_config as ic ON hd.item_name = ic.item_name "+
			" WHERE hd.item_name = ? AND h_time BETWEEN ? AND ? AND (ic.id IS NULL OR (hd.val > ic.min_val AND hd.val < ic.max_val))"+
			" Order by h_time desc limit 1", m.getTableName(in.ProjectId))
		err = m.conn.QueryRowCtx(ctx, resp, query, in.ItemName, in.Stime, in.Etime)
	} else {
		query := fmt.Sprintf("SELECT hd.* FROM %s AS hd "+
			" LEFT JOIN dc_item_config as ic ON hd.item_name = ic.item_name "+
			" WHERE hd.item_name = ? AND (ic.id IS NULL OR (hd.val > ic.min_val AND hd.val < ic.max_val))"+
			" Order by h_time desc limit 1", m.getTableName(in.ProjectId))
		err = m.conn.QueryRowCtx(ctx, resp, query, in.ItemName)
	}

	switch err {
	case nil:
		return resp, nil
	case sqlc.ErrNotFound:
		return nil, ErrNotFound
	default:
		return nil, err
	}
}

func (m *defaultDcItemHistoryDataModel) getTableName(projectId int64) string {
	return fmt.Sprintf("dc_item_history_data_%d", projectId)
}

/*
func (m *defaultDcItemHistoryDataModel) QueryHistoryData(ctx context.Context, in *pb.ItemHistoryDataListReq) ([]DcItemHistoryData, error) {
	resp := make([]DcItemHistoryData, 0)
	var err error
	var query string
	if in.Aggregator == "realtime" {
		query = fmt.Sprintf("SELECT project_id,item_name,val,DATE_FORMAT(c_time, '%%Y-%%m-%%d %%H:00:00') as c_time FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY DATE_FORMAT(c_time, '%%Y-%%m-%%d %%H:00:00') ORDER BY TIME_TO_SEC(c_time) ASC) AS row_num FROM %s WHERE project_id = ? AND item_name = ? AND c_time >= ? AND c_time < ? AND val <> 0) subquery WHERE row_num = 1;", m.table)
	} else if in.Aggregator == "new" {
		groupSql := ""
		timeSql := ""
		if in.Interval == "minute" {
			query = fmt.Sprintf(`
					SELECT val, DATE_ADD(DATE_FORMAT(min_c_time, '%%Y-%%m-%%d %%H:%%i'), INTERVAL (FLOOR(TIME_TO_SEC(min_c_time)/600)*%d) MINUTE) as c_time
					FROM (
						SELECT MAX(val) AS val,
							MIN(c_time) AS min_c_time
						FROM %s
						WHERE project_id = ? AND item_name = ?
							AND c_time >= ? AND c_time < ?
						GROUP BY DATE_FORMAT(c_time, '%%Y-%%m-%%d %%H:%%i'), FLOOR(TIME_TO_SEC(c_time) / 600)
					) AS subquery
					ORDER BY min_c_time ASC;`, in.Size, m.table)
		} else if in.Interval == "h" {
			timeSql = fmt.Sprintf("DATE_ADD(concat(DATE_FORMAT(ts,'%%Y-%%m-%%d '),LPAD(FLOOR(DATE_FORMAT(ts, '%%H')/%d)*%d,2,0),':00:00'), INTERVAL %d hour)", in.Size, in.Size, in.Size)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m-%%d' ),FLOOR(DATE_FORMAT(ts, '%%H' )/%d)", in.Size)

		} else {
			// day
			timeSql = fmt.Sprintf("DATE_ADD(concat(DATE_FORMAT(ts,'%%Y-%%m-'),LPAD(FLOOR(DATE_FORMAT(ts, '%%d')/%d)*%d,2,0),' 00:00:00'), INTERVAL %d day)", in.Size, in.Size, in.Size)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m' ),FLOOR(DATE_FORMAT(ts, '%%d' )/%d)", in.Size)
		}
		tableName := "ws_scada." + TableName(itemId)
		table := fmt.Sprintf(`(select *
								from %s
								where item_name = "%s" and project_id = %s
									and ts >= "%s" and ts < "%s"
								order by ts desc
								limit 99999999) A `, tableName, itemId, projectId, sTime, eTime)
		rows, err = DB.New().Table(table).
			Select(fmt.Sprintf("cast(item_value as DECIMAL(11,4)) as val, %s as htime", timeSql)).
			Group(groupSql).
			Order("htime asc").
			Rows()
	}
	err = m.conn.QueryRowCtx(ctx, &resp, query, in.ProjectId, in.ItemName, in.Stime, in.Etime)
	switch err {
	case nil:
		return resp, nil
	case sqlc.ErrNotFound:
		return nil, ErrNotFound
	default:
		return nil, err
	}

	var rows *sql.Rows

	if aggregator == "realtime" {
		rows, err = DB.New().Table("ws_scada."+TableName(itemId)).
			Select("item_value as val,CASE   WHEN MINUTE(ts) >= 30  THEN DATE_FORMAT(DATE_ADD(ts, INTERVAL 1 HOUR), '%Y-%m-%d %H:00:00') ELSE DATE_FORMAT(ts, '%Y-%m-%d %H:00:00') END AS htime").
			Where("item_name = ? ", itemId).
			Where("ts >= ? and ts < ?", sTime, eTime).
			Where("project_id = ?", projectId).
			Group("htime").
			Rows()
	} else if aggregator == "new" {
		groupSql := ""
		timeSql := ""
		if interval == "minute" {
			timeSql = fmt.Sprintf("DATE_ADD(concat(DATE_FORMAT(ts,'%%Y-%%m-%%d %%H:'),LPAD(FLOOR(DATE_FORMAT(ts, '%%i')/%d)*%d,2,0),':00'), INTERVAL %d minute)", sizeInt, sizeInt, sizeInt)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m-%%d %%H' ),FLOOR(DATE_FORMAT(ts, '%%i' )/%d)", sizeInt)
		} else if interval == "h" {
			timeSql = fmt.Sprintf("DATE_ADD(concat(DATE_FORMAT(ts,'%%Y-%%m-%%d '),LPAD(FLOOR(DATE_FORMAT(ts, '%%H')/%d)*%d,2,0),':00:00'), INTERVAL %d hour)", sizeInt, sizeInt, sizeInt)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m-%%d' ),FLOOR(DATE_FORMAT(ts, '%%H' )/%d)", sizeInt)
		} else {
			// day
			timeSql = fmt.Sprintf("DATE_ADD(concat(DATE_FORMAT(ts,'%%Y-%%m-'),LPAD(FLOOR(DATE_FORMAT(ts, '%%d')/%d)*%d,2,0),' 00:00:00'), INTERVAL %d day)", sizeInt, sizeInt, sizeInt)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m' ),FLOOR(DATE_FORMAT(ts, '%%d' )/%d)", sizeInt)
		}
		tableName := "ws_scada." + TableName(itemId)
		table := fmt.Sprintf(`(select *
								from %s
								where item_name = "%s" and project_id = %s
									and ts >= "%s" and ts < "%s"
								order by ts desc
								limit 99999999) A `, tableName, itemId, projectId, sTime, eTime)
		rows, err = DB.New().Table(table).
			Select(fmt.Sprintf("cast(item_value as DECIMAL(11,4)) as val, %s as htime", timeSql)).
			Group(groupSql).
			Order("htime asc").
			Rows()
	} else {
		groupSql := ""
		timeSql := ""
		if interval == "minute" {
			timeSql = fmt.Sprintf("concat(DATE_FORMAT(ts,'%%Y-%%m-%%d %%H:'),LPAD(FLOOR(DATE_FORMAT(ts, '%%i')/%d)*%d,2,0),':00')", sizeInt, sizeInt)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m-%%d %%H' ),FLOOR(DATE_FORMAT(ts, '%%i' )/%d)", sizeInt)
		} else if interval == "h" {
			timeSql = fmt.Sprintf("concat(DATE_FORMAT(ts, '%%Y-%%m-%%d ' ),LPAD(FLOOR(DATE_FORMAT(ts, '%%H' )/%d)*%d,2,0),':00:00')", sizeInt, sizeInt)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m-%%d' ),FLOOR(DATE_FORMAT(ts, '%%H' )/%d)", sizeInt)
		} else {
			//day
			timeSql = fmt.Sprintf("concat(DATE_FORMAT(ts, '%%Y-%%m-' ),LPAD(FLOOR(DATE_FORMAT(ts, '%%d' )/%d)*%d,2,0),' 00:00:00')", sizeInt, sizeInt)
			groupSql = fmt.Sprintf("DATE_FORMAT(ts, '%%Y-%%m' ),FLOOR(DATE_FORMAT(ts, '%%d' )/%d)", sizeInt)
		}

		rows, err = DB.New().Table("ws_scada."+TableName(itemId)).
			Select(fmt.Sprintf("%s(cast(item_value as DECIMAL(11,4))) as val, %s as htime", aggregator, timeSql)).
			Where("item_name = ? ", itemId).
			Where("ts >= ? and ts < ?", sTime, eTime).
			Where("project_id = ?", projectId).
			Group(groupSql).
			Rows()
	}
	defer rows.Close()
	if err != nil {
		log.Error(err)
		return nil, err
	}

	// 查询点位信息 导出表格需要
	configMap, err := GetDrdcByItemNameAndDeviceId(itemId, fmt.Sprintf("%d", deviceId))
	if err != nil {
		log.Error(err)
		return nil, err
	}

	for rows.Next() {
		di := postgres.JinkeHistoryData{}
		var val sql.NullString
		err := rows.Scan(&val, &di.Htime)
		di.Val = NullStringConverse(val)
		if err != nil { // 获得的都是字符串
			log.Error("Some amazing wrong happens in the process of queryAll.", err)
			return nil, err
		}
		di.Name = configMap.ItemAlias
		//判断小数点
		if configMap.IsBool == false {
			di.Val = processPrecise(di.Val, configMap.ItemPrecise)
		}
		resp = append(resp, di)
	}
	return resp, nil
}
*/