sparrow/services/scene-service/internal/service/manager/weather.go

package manager

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gogf/gf/container/gmap"
	"github.com/gogf/gf/encoding/gjson"
	"sparrow/pkg/rpcs"
	"sparrow/pkg/server"
	"sparrow/pkg/utils"
	"time"
	weather "weather-api-sdk"
)

// WeatherSceneConfig 天气监控任务配置
type WeatherSceneConfig struct {
	SceneId      string              `json:"scene_id"`      // 任务id
	DecisionExpr string              `json:"decision_expr"` // 条件表达式 and or
	Conditions   []*WeatherCondition `json:"conditions"`    // 条件
	Interval     int                 `json:"interval"`      // 检查间隔(分钟)
	Actions      []*Action           `json:"actions"`       // 执行动作
	ticker       *time.Ticker        `json:"-"`             // 定时器
	stopChan     chan struct{}       `json:"-"`             // 停止信号通道
}

type WeatherCondition struct {
	ConditionId string `json:"condition_id"` // 条件id
	Location    string `json:"location"`     // 地点
	FieldType   int    `json:"field_type"`   // 字段类型 1字符串 2数值
	Operator    int    `json:"operator"`     // 比较类型  数值比较 1 >  2 >= 3 = 4 <= 5 < 6 !=   字符串比较  1 相等  2  不相等
	FieldName   string `json:"field_name"`   // 字段名
	Field       string `json:"field"`        // 字段
	TargetValue string `json:"target_value"` // 目标值
}

type CheckResult struct {
	Result      bool
	ConditionId []string
}

type WeatherSceneService struct {
	tasks *gmap.HashMap
}

func NewWeatherSceneService() *WeatherSceneService {
	return &WeatherSceneService{
		tasks: gmap.NewHashMap(true),
	}
}

func (w *WeatherSceneService) Add(config string) error {
	var c WeatherSceneConfig
	err := json.Unmarshal([]byte(config), &c)
	if err != nil {
	}
	if len(c.Conditions) == 0 {
		return errors.New("天气监控任务配置错误：判断条件不能为空")
	}
	// 初始化Ticker和停止通道
	c.ticker = time.NewTicker(600 * time.Second)
	c.stopChan = make(chan struct{})

	// 启动监控协程
	go w.monitorTask(c)
	w.tasks.Set(c.SceneId, c)
	return nil
}

// monitorTask 监控任务：使用select监听Ticker和停止信号
func (w *WeatherSceneService) monitorTask(task WeatherSceneConfig) {
	for {
		select {
		case <-task.ticker.C: // 定时触发
			result, err := w.checkWeatherCondition(task)
			if err != nil {
				server.Log.Errorf("compare weather condition error :%s", err.Error())
			}
			if result.Result {
				taskExecutor := NewTaskExecutor(task.Actions)
				if err = taskExecutor.Do(task.SceneId); err != nil {
					server.Log.Errorf("weather do taskid :%s error:%s", task.SceneId, err.Error())
				}
				err = taskExecutor.saveHis(task.SceneId, result.ConditionId)
			}
		case <-task.stopChan: // 收到停止信号
			task.ticker.Stop()
			return
		}
	}
}

func (w *WeatherSceneService) Update(config string) error {
	var c WeatherSceneConfig
	err := json.Unmarshal([]byte(config), &c)
	if err != nil {
		server.Log.Errorf("config to timerConfig error :%s", err.Error())
	}
	_ = w.Stop(c.SceneId)

	// 初始化Ticker和停止通道
	c.ticker = time.NewTicker(600 * time.Second)
	c.stopChan = make(chan struct{})

	// 启动监控协程
	go w.monitorTask(c)
	w.tasks.Set(c.SceneId, c)
	server.Log.Debugf("UpdateWeatherScene :%s", config)
	return nil
}

func (w *WeatherSceneService) Remove(id string) error {
	scene := w.tasks.Get(id)
	if scene == nil {
		return nil
	}
	c := scene.(WeatherSceneConfig)
	c.stopChan <- struct{}{}
	w.tasks.Remove(c.SceneId)
	server.Log.Debugf("RemoveTimeScene :%s", c.SceneId)
	return nil
}

// Start 停止任务
func (w *WeatherSceneService) Start(id string) error {
	if !w.tasks.Contains(id) {
		return errors.New("任务不存在")
	}
	task := w.tasks.Get(id)
	c := task.(WeatherSceneConfig)
	go w.monitorTask(c)
	return nil
}

// Stop 停止任务
func (w *WeatherSceneService) Stop(id string) error {
	if !w.tasks.Contains(id) {
		return errors.New("任务不存在")
	}
	task := w.tasks.Get(id)
	c := task.(WeatherSceneConfig)
	c.stopChan <- struct{}{}
	return nil
}

// checkWeatherCondition 检查天气
func (w *WeatherSceneService) checkWeatherCondition(config WeatherSceneConfig) (CheckResult, error) {
	results := make([]bool, len(config.Conditions))
	var checkResult CheckResult
	var err error
	for _, condition := range config.Conditions {
		// 获取天气数据
		weatherInfo, err := getWeatherInfo(condition.Location)
		fmt.Printf("任务 %s: 获取天气数据成功: %v\n", config.SceneId, weatherInfo)
		if err != nil {
			fmt.Printf("任务 %s: 获取天气数据失败: %v\n", config.SceneId, err)
			return checkResult, err
		}
		result := utils.CheckValue(condition.TargetValue, weatherInfo[condition.Field], condition.FieldType, condition.Operator)
		if result {
			checkResult.ConditionId = append(checkResult.ConditionId, condition.ConditionId)
		}
		results = append(results, result)

	}
	if len(results) == 0 {
		return checkResult, nil
	}
	switch config.DecisionExpr {
	case "and":
		for _, v := range results {
			if !v {
				checkResult.Result = false
			}
		}
		checkResult.Result = true

	case "or":
		for _, v := range results {
			if v {
				checkResult.Result = true
			}
		}
		checkResult.Result = false
	default:
		err = errors.New("无效的判断逻辑")
	}
	return checkResult, err
}

func getWeatherInfo(location string) (map[string]interface{}, error) {

	// redis获取天气数据
	args := rpcs.ArgsGetWeather{Location: location}
	reply := rpcs.ReplayWeatherInfo{}
	err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.GetWeatherInfo", args, &reply)
	if err != nil {
		server.Log.Errorf("天气状态获取失败：%v", err)
		return nil, err
	}

	if reply.Info != "" {
		info := gjson.New(reply.Info)
		// 判断数据是否过期
		if !info.GetTime("last_update").Before(time.Now().Add(-time.Minute * 15)) {
			return info.Map(), nil
		}
	}

	weatherInfo, err := weather.GetWeatherInfo(location, "SgqF9ZGBUj7R0dLAb")
	if err != nil {
		return nil, err
	}
	j := gjson.New(weatherInfo)
	err = redisSaveWeatherInfo(location, j.MustToJsonString())
	return j.Map(), nil

}

func redisSaveWeatherInfo(location string, info string) error {
	args := rpcs.ArgsSetWeather{
		Location: location,
		Info:     info,
	}
	reply := rpcs.ReplySetStatus{}
	err := server.RPCCallByName(nil, rpcs.DeviceManagerName, "DeviceManager.SetWeatherInfo", args, &reply)
	if err != nil {
		server.Log.Errorf("天气状态获取失败：%v", err)
		return err
	}
	return nil
}