property-mqtt/model/gate_record.go

// Copyright 2019 github.com. All rights reserved.
// Use of this source code is governed by github.com.

package model

import (
	"encoding/json"
	"property-mqtt/errors"
	"strings"
	"time"

	"git.getensh.com/common/gopkgs/logger"
	"go.uber.org/zap"

	"git.getensh.com/common/gopkgs/influxdb"
	iclient "github.com/influxdata/influxdb/client/v2"
)
const (
	DatabaseNotFound = "database not found"
	GateDbname = "property_device_gate"
)


func WriteGateData(deviceId string, tags map[string]string, fields map[string]interface{}, t time.Time) error {
	// 主动获取数据转换

	bp, err := iclient.NewBatchPoints(iclient.BatchPointsConfig{
		Database:  GateDbname,
		Precision: "ns",
	})

	if err != nil {
		logger.Error("func",
			zap.String("call", "NewBatchPoints"),
			zap.String("args", GateDbname),
			zap.String("error", err.Error()))
		return errors.DataBaseError
	}

	pt, err := iclient.NewPoint(
		deviceId,
		tags,
		fields,
		t,
	)
	if err != nil {
		logger.Error("func",
			zap.String("call", "NewPoint"),
			zap.String("args", deviceId),
			zap.String("error", err.Error()))
		return errors.DataBaseError
	}

	bp.AddPoint(pt)

	err = influxdb.InfluxCli.Write(bp)
	if err != nil {
		logger.Error("func",
			zap.String("call", "InfluxCli.Write"),
			zap.String("args", deviceId),
			zap.String("error", err.Error()))

		if strings.Contains(err.Error(), DatabaseNotFound) {
			Create(influxdb.InfluxCli)
			err = influxdb.InfluxCli.Write(bp)
			if err != nil {
				return errors.DataBaseError
			}
			return nil
		}
		return errors.DataBaseError
	}

	return nil
}

func Query(sql string,  dbname string, result interface{},) ([]map[string]interface{},  error) {
	qt := iclient.Query{
		Database: dbname,
		Command:  sql,
	}
	r, err := influxdb.InfluxCli.Query(qt)
	if err != nil {
		logger.Error("func",
			zap.String("call", "Query "+ dbname),
			zap.String("args", sql),
			zap.String("error", err.Error()))
		return nil, err
	}
	if r == nil {
		return nil, nil
	}
	if len(r.Results) == 0 {
		return nil, nil
	}
	if len(r.Results[0].Series) == 0 {
		return nil, nil
	}
	colNames := r.Results[0].Series[0].Columns
	var marray = make([]map[string]interface{}, len(r.Results[0].Series[0].Values))
	for i, row := range r.Results[0].Series[0].Values {
		item := map[string]interface{}{}
		for j, v := range row {
			if colNames[j] == "time" {
				t, _ := time.Parse(time.RFC3339, v.(string))
				v = t.Format("2006-01-02 15:04:05")
			}
			item[colNames[j]] = v
		}
		marray[i] = item
	}

	if result != nil {
		bytes, _ := json.Marshal(marray)
		err = json.Unmarshal(bytes, result)
		if err != nil {
			return nil, err
		}
	}

	return marray, nil
}

func Create(conn iclient.Client) error {
	qt := iclient.Query{
		Command: "create database " + GateDbname,
	}
	_, err := conn.Query(qt)
	if err != nil {
		return err
	}
	return nil
}

func WriteGateDatas(deviceId string, tags map[string]string, fields []map[string]interface{}, t []time.Time) error {
	// 主动获取数据转换

	bp, err := iclient.NewBatchPoints(iclient.BatchPointsConfig{
		Database:  GateDbname,
		Precision: "ns",
	})

	if err != nil {
		logger.Error("func",
			zap.String("call", "NewBatchPoints"),
			zap.String("args", GateDbname),
			zap.String("error", err.Error()))
		return errors.DataBaseError
	}
	pts := make([]*iclient.Point, len(fields))
	for i, _ := range fields {
		pt, err := iclient.NewPoint(
			deviceId,
			tags,
			fields[i],
			t[i],
		)
		if err != nil {
			logger.Error("func",
				zap.String("call", "NewPoint"),
				zap.String("args", deviceId),
				zap.String("error", err.Error()))
			return errors.DataBaseError
		}
		pts[i] = pt
	}

	bp.AddPoints(pts)

	err = influxdb.InfluxCli.Write(bp)
	if err != nil {
		logger.Error("func",
			zap.String("call", "InfluxCli.Write"),
			zap.String("args", deviceId),
			zap.String("error", err.Error()))

		if strings.Contains(err.Error(), DatabaseNotFound) {
			Create(influxdb.InfluxCli)
			err = influxdb.InfluxCli.Write(bp)
			if err != nil {
				return errors.DataBaseError
			}
			return nil
		}
		return errors.DataBaseError
	}

	return nil
}