实现DeviceThresholdCheckTask(除Execute外)

This commit is contained in:
2025-11-08 21:12:51 +08:00
parent 26df398392
commit a35a9a1038
2 changed files with 88 additions and 2 deletions

View File

@@ -67,7 +67,7 @@
1. **任务类型**: 提供两种可供用户配置的阈值告警任务类型,分别对应 **区域主控****普通设备** 告警。
2. **参数结构**:
* **通用参数**: 任务参数将包含 `Value` (阈值) 和 `Operator` (操作符,如 `>``<`) 字段。
* **通用参数**: 任务参数将包含 `Thresholds` (阈值) 和 `Operator` (操作符,如 `>``<`) 字段。
* **普通设备任务**: 配置包含 `DeviceID`
* **区域主控任务**: 配置包含 `AreaControllerID`, `SensorType`, 以及一个 `ExcludeDeviceIDs` (需要排除的设备ID列表)。
@@ -132,4 +132,4 @@
2. 重构部分枚举, 让models包不依赖其他项目中的包
3. 创建仓库层对象(不包含方法)
4. 实现告警发送任务
5. 实现告警通知发送计划
5. 实现告警通知发送计划/全量采集计划改名

View File

@@ -0,0 +1,86 @@
package task
import (
"context"
"fmt"
"sync"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/plan"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
)
type Operator string
const (
OperatorLessThan Operator = "<"
OperatorLessThanOrEqualTo Operator = "<="
OperatorGreaterThan Operator = ">"
OperatorGreaterThanOrEqualTo Operator = ">="
OperatorEqualTo Operator = "="
OperatorNotEqualTo Operator = "!="
)
type DeviceThresholdCheckParams struct {
DeviceID uint `json:"device_id"` // 设备ID
Thresholds float64 `json:"thresholds"` // 阈值
Operator Operator `json:"operator"` // 操作符
}
type DeviceThresholdCheckTask struct {
ctx context.Context
onceParse sync.Once
taskLog *models.TaskExecutionLog
params DeviceThresholdCheckParams
}
func NewDeviceThresholdCheckTask(ctx context.Context, taskLog *models.TaskExecutionLog) plan.Task {
return &DeviceThresholdCheckTask{
ctx: ctx,
taskLog: taskLog,
}
}
func (d *DeviceThresholdCheckTask) Execute(ctx context.Context) error {
//TODO implement me
panic("implement me")
}
// parseParameters 解析任务参数
func (d *DeviceThresholdCheckTask) parseParameters(ctx context.Context) error {
logger := logs.TraceLogger(ctx, d.ctx, "parseParameters")
var err error
d.onceParse.Do(func() {
if d.taskLog.Task.Parameters == nil {
logger.Errorf("任务 %v: 缺少参数", d.taskLog.TaskID)
err = fmt.Errorf("任务 %v: 参数不全", d.taskLog.TaskID)
return
}
var params DeviceThresholdCheckParams
err = d.taskLog.Task.ParseParameters(&params)
if err != nil {
logger.Errorf("任务 %v: 解析参数失败: %v", d.taskLog.TaskID, err)
err = fmt.Errorf("任务 %v: 解析参数失败: %v", d.taskLog.TaskID, err)
return
}
d.params = params
})
return err
}
func (d *DeviceThresholdCheckTask) OnFailure(ctx context.Context, executeErr error) {
logger := logs.TraceLogger(ctx, d.ctx, "OnFailure")
logger.Errorf("设备阈值检测任务执行失败, 任务ID: %v: 执行失败: %v", d.taskLog.TaskID, executeErr)
}
func (d *DeviceThresholdCheckTask) ResolveDeviceIDs(ctx context.Context) ([]uint, error) {
taskCtx := logs.AddFuncName(ctx, d.ctx, "ResolveDeviceIDs")
if err := d.parseParameters(taskCtx); err != nil {
return nil, err
}
return []uint{d.params.DeviceID}, nil
}