pig/pig-farm-controller
1
0
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects Releases Wiki Activity

CreateDeviceThresholdAlarm

Browse Source 
UpdateDeviceThresholdAlarm
CreateAreaThresholdAlarm
UpdateAreaThresholdAlarm
This commit is contained in:
huang
2025-11-10 17:35:22 +08:00
parent d4e8aba1fd
commit f2b0c2987f
10 changed files with 375 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
internal/app/dto/alarm_dto.go
View File

@@ -81,3 +81,35 @@ type ListHistoricalAlarmResponse struct {
List []HistoricalAlarmDTO `json:"list"`
Pagination PaginationDTO `json:"pagination"`
}
// CreateDeviceThresholdAlarmDTO 创建设备阈值告警的请求DTO
type CreateDeviceThresholdAlarmDTO struct {
DeviceID uint `json:"device_id" binding:"required"` // 设备ID
SensorType models.SensorType `json:"sensor_type" binding:"required"` // 传感器类型
Thresholds float64 `json:"thresholds" binding:"required"` // 阈值
Operator models.Operator `json:"operator" binding:"required"` // 操作符 (使用string类型，与前端交互更通用)
Level models.SeverityLevel `json:"level,omitempty"` // 告警等级，可选，如果未提供则使用默认值
}
// UpdateDeviceThresholdAlarmDTO 更新设备阈值告警的请求DTO
type UpdateDeviceThresholdAlarmDTO struct {
Thresholds float64 `json:"thresholds" binding:"required"` // 新的阈值
Operator models.Operator `json:"operator" binding:"required"` // 新的操作符
Level models.SeverityLevel `json:"level,omitempty"` // 新的告警等级，可选
}
// CreateAreaThresholdAlarmDTO 创建区域阈值告警的请求DTO
type CreateAreaThresholdAlarmDTO struct {
AreaControllerID uint `json:"area_controller_id" binding:"required"` // 区域主控ID
SensorType models.SensorType `json:"sensor_type" binding:"required"` // 传感器类型
Thresholds float64 `json:"thresholds" binding:"required"` // 阈值
Operator models.Operator `json:"operator" binding:"required"` // 操作符
Level models.SeverityLevel `json:"level,omitempty"` // 告警等级，可选
}
// UpdateAreaThresholdAlarmDTO 更新区域阈值告警的请求DTO
type UpdateAreaThresholdAlarmDTO struct {
Thresholds float64 `json:"thresholds" binding:"required"` // 新的阈值
Operator models.Operator `json:"operator" binding:"required"` // 新的操作符
Level models.SeverityLevel `json:"level,omitempty"` // 新的告警等级，可选
}

3
internal/app/service/plan_service.go
View File

@@ -7,6 +7,7 @@ import (
"git.huangwc.com/pig/pig-farm-controller/internal/app/dto"
"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"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
)
@@ -147,7 +148,7 @@ func (s *planService) UpdatePlan(ctx context.Context, id uint, req *dto.UpdatePl
planToUpdate.ID = id // 确保ID被设置
// 调用领域服务更新计划
updatedPlan, err := s.domainPlanService.UpdatePlan(serviceCtx, planToUpdate)
updatedPlan, err := s.domainPlanService.UpdatePlan(serviceCtx, planToUpdate, models.PlanTypeCustom)
if err != nil {
logger.Errorf("%s: 领域服务更新计划失败: %v, ID: %d", actionType, err, id)
return nil, err // 直接返回领域层错误

283
internal/app/service/threshold_alarm_service.go
View File

@@ -2,11 +2,15 @@ package service
import (
"context"
"fmt"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/dto"
domainAlarm "git.huangwc.com/pig/pig-farm-controller/internal/domain/alarm"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/plan"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/task"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
)
@@ -21,21 +25,42 @@ type ThresholdAlarmService interface {
ListActiveAlarms(ctx context.Context, req *dto.ListActiveAlarmRequest) (*dto.ListActiveAlarmResponse, error)
// ListHistoricalAlarms 批量查询历史告警。
ListHistoricalAlarms(ctx context.Context, req *dto.ListHistoricalAlarmRequest) (*dto.ListHistoricalAlarmResponse, error)
// CreateDeviceThresholdAlarm 创建一个设备阈值告警。
CreateDeviceThresholdAlarm(ctx context.Context, req *dto.CreateDeviceThresholdAlarmDTO) error
// UpdateDeviceThresholdAlarm 更新一个设备阈值告警。
UpdateDeviceThresholdAlarm(ctx context.Context, taskID int, req *dto.UpdateDeviceThresholdAlarmDTO) error
// CreateAreaThresholdAlarm 创建一个区域阈值告警。
CreateAreaThresholdAlarm(ctx context.Context, req *dto.CreateAreaThresholdAlarmDTO) error
// UpdateAreaThresholdAlarm 更新一个区域阈值告警。
UpdateAreaThresholdAlarm(ctx context.Context, taskID int, req *dto.UpdateAreaThresholdAlarmDTO) error
}
// thresholdAlarmService 是 ThresholdAlarmService 接口的具体实现。
type thresholdAlarmService struct {
ctx context.Context
alarmService domainAlarm.AlarmService
alarmRepo repository.AlarmRepository
planService plan.Service
alarmRepo repository.AlarmRepository
planRepo repository.PlanRepository
deviceRepo repository.DeviceRepository
}
// NewThresholdAlarmService 创建一个新的 ThresholdAlarmService 实例。
func NewThresholdAlarmService(ctx context.Context, alarmService domainAlarm.AlarmService, alarmRepo repository.AlarmRepository) ThresholdAlarmService {
func NewThresholdAlarmService(ctx context.Context,
alarmService domainAlarm.AlarmService,
planService plan.Service,
alarmRepo repository.AlarmRepository,
planRepo repository.PlanRepository,
deviceRepo repository.DeviceRepository,
) ThresholdAlarmService {
return &thresholdAlarmService{
ctx: ctx,
alarmService: alarmService,
planService: planService,
alarmRepo: alarmRepo,
planRepo: planRepo,
deviceRepo: deviceRepo,
}
}
@@ -95,3 +120,257 @@ func (s *thresholdAlarmService) ListHistoricalAlarms(ctx context.Context, req *d
return dto.NewListHistoricalAlarmResponse(alarms, total, req.Page, req.PageSize), nil
}
// CreateDeviceThresholdAlarm 实现了创建一个设备阈值告警的逻辑。
func (s *thresholdAlarmService) CreateDeviceThresholdAlarm(ctx context.Context, req *dto.CreateDeviceThresholdAlarmDTO) error {
serviceCtx, logger := logs.Trace(ctx, s.ctx, "CreateThresholdAlarm")
device, err := s.deviceRepo.FindByID(serviceCtx, req.DeviceID)
if err != nil {
return fmt.Errorf("获取设备 %v 失败: %v", req.DeviceID, err)
}
plan, err := s.planRepo.GetSystemPlanByName(serviceCtx, models.PlanNamePeriodicSystemHealthCheck)
if err != nil {
return fmt.Errorf("获取系统计划 %v 失败: %v", models.PlanNamePeriodicSystemHealthCheck, err)
}
if plan == nil {
logger.Panicf("系统计划 %v 不存在", models.PlanNamePeriodicSystemHealthCheck)
}
// 系统计划肯定是子任务
for i, t := range plan.Tasks {
switch t.Type {
case models.TaskTypeDeviceThresholdCheck: // 检查任务是否存在
var params task.DeviceThresholdCheckParams
err = t.ParseParameters(&params)
if err != nil {
return fmt.Errorf("任务 %v: 解析设备阈值检查任务参数失败: %v", t.ID, err)
}
if params.DeviceID == req.DeviceID && params.SensorType == req.SensorType {
return fmt.Errorf("设备 %v: 该设备已存在阈值检查任务", req.DeviceID)
}
case models.TaskTypeAreaCollectorThresholdCheck: // 向区域阈值检查任务过滤列表中添加该设备
params := task.AreaThresholdCheckParams{
ExcludeDeviceIDs: []uint{},
}
err = t.ParseParameters(&params)
if err != nil {
return fmt.Errorf("任务 %v: 解析区域阈值检查任务参数失败: %v", t.ID, err)
}
if params.AreaControllerID == device.AreaControllerID {
has := false
for _, d := range params.ExcludeDeviceIDs {
if d == req.DeviceID {
has = true
break
}
}
if !has {
params.ExcludeDeviceIDs = append(params.ExcludeDeviceIDs, req.DeviceID)
err = plan.Tasks[i].SaveParameters(params)
if err != nil {
return fmt.Errorf("任务 %v: 保存任务参数失败: %v", t.ID, err)
}
}
}
default:
continue
}
}
t := models.Task{
PlanID: plan.ID,
Name: fmt.Sprintf("设备 %v 的阈值检测任务", req.DeviceID),
Description: fmt.Sprintf("检测该设备 %v 是否 %v %v", req.SensorType, req.Operator, req.Thresholds),
ExecutionOrder: len(plan.Tasks) + 1,
Type: models.TaskTypeDeviceThresholdCheck,
}
err = t.SaveParameters(task.DeviceThresholdCheckParams{
DeviceID: req.DeviceID,
SensorType: req.SensorType,
Thresholds: req.Thresholds,
Level: req.Level,
Operator: req.Operator,
})
if err != nil {
return fmt.Errorf("保存任务参数失败: %v", err)
}
plan.Tasks = append(plan.Tasks, t)
plan.ReorderSteps()
_, err = s.planService.UpdatePlan(serviceCtx, plan, models.PlanTypeSystem)
if err != nil {
return fmt.Errorf("更新计划失败: %v", err)
}
return nil
}
// UpdateDeviceThresholdAlarm 实现了更新一个设备阈值告警的逻辑。
func (s *thresholdAlarmService) UpdateDeviceThresholdAlarm(ctx context.Context, taskID int, req *dto.UpdateDeviceThresholdAlarmDTO) error {
serviceCtx, logger := logs.Trace(ctx, s.ctx, "UpdateDeviceThresholdAlarm")
// 1. 获取系统健康检查计划
plan, err := s.planRepo.GetSystemPlanByName(serviceCtx, models.PlanNamePeriodicSystemHealthCheck)
if err != nil {
return fmt.Errorf("获取系统健康检查计划失败: %w", err)
}
if plan == nil {
// 这个系统计划必须存在
logger.Panicf("系统计划 %v 不存在", models.PlanNamePeriodicSystemHealthCheck)
}
// 2. 遍历任务列表，查找并更新目标任务
taskFound := false
for i, t := range plan.Tasks {
if t.ID == taskID && t.Type == models.TaskTypeDeviceThresholdCheck {
taskFound = true
var params task.DeviceThresholdCheckParams
if err = t.ParseParameters(&params); err != nil {
return fmt.Errorf("任务 %d: 解析现有参数失败: %w", taskID, err)
}
params.Thresholds = req.Thresholds
params.Operator = req.Operator
params.Level = req.Level
// 刷新任务说明
plan.Tasks[i].Description = fmt.Sprintf("检测该设备 %v 是否 %v %v", params.SensorType, params.Operator, params.Thresholds)
err = plan.Tasks[i].SaveParameters(params)
if err != nil {
return fmt.Errorf("任务 %d: 保存参数失败: %w", taskID, err)
}
break
}
}
if !taskFound {
return fmt.Errorf("任务 %d: 不存在", taskID)
}
// 全量更新计划
_, err = s.planService.UpdatePlan(serviceCtx, plan, models.PlanTypeSystem)
return err
}
// CreateAreaThresholdAlarm 实现了创建一个区域阈值告警的逻辑。
func (s *thresholdAlarmService) CreateAreaThresholdAlarm(ctx context.Context, req *dto.CreateAreaThresholdAlarmDTO) error {
serviceCtx, logger := logs.Trace(ctx, s.ctx, "CreateAreaThresholdAlarm")
// 1. 获取系统健康检查计划
plan, err := s.planRepo.GetSystemPlanByName(serviceCtx, models.PlanNamePeriodicSystemHealthCheck)
if err != nil {
return fmt.Errorf("获取系统健康检查计划失败: %w", err)
}
if plan == nil {
logger.Panicf("系统计划 %v 不存在", models.PlanNamePeriodicSystemHealthCheck)
}
// 2. 获取目标区域下的所有设备，并建立快速查找表
devicesInArea, err := s.deviceRepo.ListByAreaControllerID(serviceCtx, req.AreaControllerID)
if err != nil {
return fmt.Errorf("获取区域 %d 下的设备列表失败: %w", req.AreaControllerID, err)
}
devicesInAreaMap := make(map[uint]struct{}, len(devicesInArea))
for _, device := range devicesInArea {
devicesInAreaMap[device.ID] = struct{}{}
}
// 3. 遍历计划，检查存在性并收集需要排除的设备ID
var excludeDeviceIDs []uint
for _, t := range plan.Tasks {
switch t.Type {
case models.TaskTypeAreaCollectorThresholdCheck:
var params task.AreaThresholdCheckParams
if err := t.ParseParameters(&params); err != nil {
return fmt.Errorf("任务 %d: 解析区域阈值检查任务参数失败: %w", t.ID, err)
}
if params.AreaControllerID == req.AreaControllerID && params.SensorType == req.SensorType {
return fmt.Errorf("区域 %d: 该区域已存在针对 %s 的阈值检查任务", req.AreaControllerID, req.SensorType)
}
case models.TaskTypeDeviceThresholdCheck:
var params task.DeviceThresholdCheckParams
if err := t.ParseParameters(&params); err != nil {
return fmt.Errorf("任务 %d: 解析设备阈值检查任务参数失败: %w", t.ID, err)
}
// 检查该设备是否属于目标区域
if _, ok := devicesInAreaMap[params.DeviceID]; ok {
excludeDeviceIDs = append(excludeDeviceIDs, params.DeviceID)
}
}
}
// 4. 创建新任务
newTask := models.Task{
PlanID: plan.ID,
Name: fmt.Sprintf("区域 %d 的 %s 阈值检测任务", req.AreaControllerID, req.SensorType),
Description: fmt.Sprintf("检测区域 %d 的 %s 是否 %v %v", req.AreaControllerID, req.SensorType, req.Operator, req.Thresholds),
ExecutionOrder: len(plan.Tasks) + 1,
Type: models.TaskTypeAreaCollectorThresholdCheck,
}
err = newTask.SaveParameters(task.AreaThresholdCheckParams{
AreaControllerID: req.AreaControllerID,
SensorType: req.SensorType,
Thresholds: req.Thresholds,
Operator: req.Operator,
Level: req.Level,
ExcludeDeviceIDs: excludeDeviceIDs,
})
if err != nil {
return fmt.Errorf("保存新区域任务的参数失败: %w", err)
}
// 5. 更新计划
plan.Tasks = append(plan.Tasks, newTask)
plan.ReorderSteps()
_, err = s.planService.UpdatePlan(serviceCtx, plan, models.PlanTypeSystem)
return err
}
// UpdateAreaThresholdAlarm 实现了更新一个区域阈值告警的逻辑。
func (s *thresholdAlarmService) UpdateAreaThresholdAlarm(ctx context.Context, taskID int, req *dto.UpdateAreaThresholdAlarmDTO) error {
serviceCtx, logger := logs.Trace(ctx, s.ctx, "UpdateAreaThresholdAlarm")
// 1. 获取系统健康检查计划
plan, err := s.planRepo.GetSystemPlanByName(serviceCtx, models.PlanNamePeriodicSystemHealthCheck)
if err != nil {
return fmt.Errorf("获取系统健康检查计划失败: %w", err)
}
if plan == nil {
logger.Panicf("系统计划 %v 不存在", models.PlanNamePeriodicSystemHealthCheck)
}
// 2. 遍历任务列表，查找并更新目标任务
taskFound := false
for i, t := range plan.Tasks {
if t.ID == taskID && t.Type == models.TaskTypeAreaCollectorThresholdCheck {
taskFound = true
var params task.AreaThresholdCheckParams
if err = t.ParseParameters(&params); err != nil {
return fmt.Errorf("任务 %d: 解析现有参数失败: %w", taskID, err)
}
params.Thresholds = req.Thresholds
params.Operator = req.Operator
params.Level = req.Level
// 刷新任务说明
plan.Tasks[i].Description = fmt.Sprintf("检测区域 %d 的 %s 是否 %v %v", params.AreaControllerID, params.SensorType, params.Operator, params.Thresholds)
err = plan.Tasks[i].SaveParameters(params)
if err != nil {
return fmt.Errorf("任务 %d: 保存参数失败: %w", taskID, err)
}
break
}
}
if !taskFound {
return fmt.Errorf("任务 %d: 不存在或类型不匹配", taskID)
}
// 全量更新计划
_, err = s.planService.UpdatePlan(serviceCtx, plan, models.PlanTypeSystem)
return err
}

13
internal/core/data_initializer.go
View File

@@ -10,13 +10,6 @@ import (
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
)
const (
// PlanNamePeriodicSystemHealthCheck 是周期性系统健康检查计划的名称
PlanNamePeriodicSystemHealthCheck = "周期性系统健康检查"
// PlanNameAlarmNotification 是告警通知发送计划的名称
PlanNameAlarmNotification = "告警通知发送"
)
// initializeState 在应用启动时准备其初始数据状态。
// 它遵循一个严格的顺序：清理 -> 更新 -> 刷新，以确保数据的一致性和正确性。
func (app *Application) initializeState(ctx context.Context) error {
@@ -121,7 +114,7 @@ func (app *Application) initializePeriodicSystemHealthCheckPlan(ctx context.Cont
newTasks = append(newTasks, fullCollectionTask, delayTask)
// 如果计划已存在，则获取其现有任务并追加到新任务列表后（排除预设任务）
if foundExistingPlan, ok := existingPlanMap[PlanNamePeriodicSystemHealthCheck]; ok {
if foundExistingPlan, ok := existingPlanMap[models.PlanNamePeriodicSystemHealthCheck]; ok {
for _, existingTask := range foundExistingPlan.Tasks {
// 排除已预设的全量采集和延时任务
if existingTask.Type != models.TaskTypeFullCollection && existingTask.Type != models.TaskTypeWaiting {
@@ -136,7 +129,7 @@ func (app *Application) initializePeriodicSystemHealthCheckPlan(ctx context.Cont
}
predefinedPlan := &models.Plan{
Name: PlanNamePeriodicSystemHealthCheck,
Name: models.PlanNamePeriodicSystemHealthCheck,
Description: fmt.Sprintf("这是一个系统预定义的计划, 每 %d 分钟自动触发一次全量数据采集, 并进行阈值校验告警。", app.Config.Collection.Interval),
PlanType: models.PlanTypeSystem,
ExecutionType: models.PlanExecutionTypeAutomatic,
@@ -186,7 +179,7 @@ func (app *Application) initializeAlarmNotificationPlan(ctx context.Context, exi
appCtx, logger := logs.Trace(ctx, app.Ctx, "initializeAlarmNotificationPlan")
predefinedPlan := &models.Plan{
Name: PlanNameAlarmNotification,
Name: models.PlanNameAlarmNotification,
Description: "这是一个系统预定义的计划, 每分钟自动触发一次告警通知发送。",
PlanType: models.PlanTypeSystem,
ExecutionType: models.PlanExecutionTypeAutomatic,

13
internal/domain/plan/plan_service.go
View File

@@ -44,8 +44,8 @@ type Service interface {
GetPlanByID(ctx context.Context, id uint) (*models.Plan, error)
// ListPlans 获取计划列表，支持过滤和分页
ListPlans(ctx context.Context, opts repository.ListPlansOptions, page, pageSize int) ([]models.Plan, int64, error)
// UpdatePlan 更新计划
UpdatePlan(ctx context.Context, plan *models.Plan) (*models.Plan, error)
// UpdatePlan 更新计划, wantPlanType 表示期望被修改的计划是什么类型
UpdatePlan(ctx context.Context, plan *models.Plan, wantPlanType models.PlanType) (*models.Plan, error)
// DeletePlan 删除计划（软删除）
DeletePlan(ctx context.Context, id uint) error
// StartPlan 启动计划
@@ -207,8 +207,8 @@ func (s *planServiceImpl) ListPlans(ctx context.Context, opts repository.ListPla
return plans, total, nil
}
// UpdatePlan 更新计划
func (s *planServiceImpl) UpdatePlan(ctx context.Context, planToUpdate *models.Plan) (*models.Plan, error) {
// UpdatePlan 更新计划, wantPlanType 表示期望被修改的计划是什么类型
func (s *planServiceImpl) UpdatePlan(ctx context.Context, planToUpdate *models.Plan, wantPlanType models.PlanType) (*models.Plan, error) {
planCtx, logger := logs.Trace(ctx, s.ctx, "UpdatePlan")
const actionType = "领域层：更新计划"
@@ -222,9 +222,8 @@ func (s *planServiceImpl) UpdatePlan(ctx context.Context, planToUpdate *models.P
return nil, err
}
// 系统计划不允许修改
if existingPlan.PlanType == models.PlanTypeSystem {
logger.Warnf("%s: 尝试修改系统计划, ID: %d", actionType, planToUpdate.ID)
if existingPlan.PlanType != wantPlanType {
logger.Warnf("%s: 禁止修改 %v 类型计划, ID: %d", actionType, wantPlanType, planToUpdate.ID)
return nil, ErrPlanCannotBeModified
}

15
internal/domain/task/area_threshold_check_task.go
View File

@@ -14,11 +14,12 @@ import (
// AreaThresholdCheckParams 定义了区域阈值检查任务的参数
type AreaThresholdCheckParams struct {
AreaControllerID uint `json:"area_controller_id"` // 区域主控ID
SensorType models.SensorType `json:"sensor_type"` // 传感器类型
Thresholds float64 `json:"thresholds"` // 阈值
Operator models.Operator `json:"operator"` // 操作符
ExcludeDeviceIDs []uint `json:"exclude_device_ids"` // 排除的传感器ID
AreaControllerID uint `json:"area_controller_id"` // 区域主控ID
SensorType models.SensorType `json:"sensor_type"` // 传感器类型
Thresholds float64 `json:"thresholds"` // 阈值
Operator models.Operator `json:"operator"` // 操作符
Level models.SeverityLevel `json:"level"` // 告警级别
ExcludeDeviceIDs []uint `json:"exclude_device_ids"` // 排除的传感器ID
}
// AreaThresholdCheckTask 是一个任务，用于检查区域阈值并触发告警, 区域主控下的所有没有独立校验任务的设备都会受到此任务的检查
@@ -78,6 +79,7 @@ func (a *AreaThresholdCheckTask) Execute(ctx context.Context) error {
DeviceID: device.ID,
SensorType: a.params.SensorType,
Thresholds: a.params.Thresholds,
Level: a.params.Level,
Operator: a.params.Operator,
})
if err != nil {
@@ -150,6 +152,9 @@ func (a *AreaThresholdCheckTask) parseParameters(ctx context.Context) error {
if params.AreaControllerID == 0 {
err = fmt.Errorf("任务 %v: 未配置区域主控ID", a.taskLog.TaskID)
}
if params.Level == "" {
params.Level = models.WarnLevel
}
if params.ExcludeDeviceIDs == nil {
params.ExcludeDeviceIDs = []uint{}
}

14
internal/domain/task/device_threshold_check_task.go
View File

@@ -14,10 +14,11 @@ import (
)
type DeviceThresholdCheckParams struct {
DeviceID uint `json:"device_id"` // 设备ID
SensorType models.SensorType `json:"sensor_type"` // 传感器类型
Thresholds float64 `json:"thresholds"` // 阈值
Operator models.Operator `json:"operator"` // 操作符
DeviceID uint `json:"device_id"` // 设备ID
SensorType models.SensorType `json:"sensor_type"` // 传感器类型
Thresholds float64 `json:"thresholds"` // 阈值
Operator models.Operator `json:"operator"` // 操作符
Level models.SeverityLevel `json:"level"` // 告警等级
}
// DeviceThresholdCheckTask 是一个任务，用于检查设备传感器数据是否满足阈值条件。
@@ -99,7 +100,7 @@ func (d *DeviceThresholdCheckTask) Execute(ctx context.Context) error {
AlarmCode: alarmCode,
AlarmSummary: summary,
AlarmDetails: details,
Level: models.WarnLevel, // 默认告警等级，可后续根据需求调整
Level: d.params.Level,
TriggerTime: time.Now(),
}
@@ -172,6 +173,9 @@ func (d *DeviceThresholdCheckTask) parseParameters(ctx context.Context) error {
if params.DeviceID == 0 {
err = fmt.Errorf("任务 %v: 未配置设备ID", d.taskLog.TaskID)
}
if params.Level == "" {
params.Level = models.WarnLevel
}
d.params = params

20
internal/infra/models/plan.go
View File

@@ -11,11 +11,13 @@ import (
"gorm.io/gorm"
)
type PlanName string
const (
// PlanNamePeriodicSystemHealthCheck 是周期性系统健康检查计划的名称
PlanNamePeriodicSystemHealthCheck = "周期性系统健康检查"
PlanNamePeriodicSystemHealthCheck PlanName = "周期性系统健康检查"
// PlanNameAlarmNotification 是告警通知发送计划的名称
PlanNameAlarmNotification = "告警通知发送"
PlanNameAlarmNotification PlanName = "告警通知发送"
)
// PlanExecutionType 定义了计划的执行类型
@@ -38,11 +40,13 @@ const (
type TaskType string
const (
TaskPlanAnalysis TaskType = "计划分析" // 解析Plan的Task列表并添加到待执行队列的特殊任务
TaskTypeWaiting TaskType = "等待" // 等待任务
TaskTypeReleaseFeedWeight TaskType = "下料" // 下料口释放指定重量任务
TaskTypeFullCollection TaskType = "全量采集" // 新增的全量采集任务
TaskTypeAlarmNotification TaskType = "告警通知" // 告警通知任务
TaskPlanAnalysis TaskType = "计划分析" // 解析Plan的Task列表并添加到待执行队列的特殊任务
TaskTypeWaiting TaskType = "等待" // 等待任务
TaskTypeReleaseFeedWeight TaskType = "下料" // 下料口释放指定重量任务
TaskTypeFullCollection TaskType = "全量采集" // 新增的全量采集任务
TaskTypeAlarmNotification TaskType = "告警通知" // 告警通知任务
TaskTypeDeviceThresholdCheck TaskType = "设备阈值检查" // 设备阈值检查任务
TaskTypeAreaCollectorThresholdCheck TaskType = "区域阈值检查" // 区域阈值检查任务
)
// -- Task Parameters --
@@ -72,7 +76,7 @@ const (
type Plan struct {
gorm.Model
Name string `gorm:"not null" json:"name"`
Name PlanName `gorm:"not null" json:"name"`
Description string `json:"description"`
PlanType PlanType `gorm:"not null;index" json:"plan_type"` // 任务类型, 包括系统任务和用户自定义任务
ExecutionType PlanExecutionType `gorm:"not null;index" json:"execution_type"`

19
internal/infra/repository/plan_repository.go
View File

@@ -49,6 +49,8 @@ type PlanRepository interface {
GetPlanByID(ctx context.Context, id uint) (*models.Plan, error)
// GetPlansByIDs 根据ID列表获取计划，不包含子计划和任务详情
GetPlansByIDs(ctx context.Context, ids []uint) ([]models.Plan, error)
// GetSystemPlanByName 根据计划名称获取系统计划，包含子计划和任务详情
GetSystemPlanByName(ctx context.Context, planName models.PlanName) (*models.Plan, error)
// CreatePlan 创建一个新的计划
CreatePlan(ctx context.Context, plan *models.Plan) error
// CreatePlanTx 在指定事务中创建一个新的计划
@@ -164,6 +166,23 @@ func (r *gormPlanRepository) GetPlansByIDs(ctx context.Context, ids []uint) ([]m
return plans, nil
}
// GetSystemPlanByName 根据计划名称获取系统计划，包含子计划和任务详情, 系统任务不该有重名情况, 所以可以这么查询
func (r *gormPlanRepository) GetSystemPlanByName(ctx context.Context, planName models.PlanName) (*models.Plan, error) {
repoCtx := logs.AddFuncName(ctx, r.ctx, "GetSystemPlanByName")
var plan models.Plan
// 首先只查询计划的基本信息，获取其ID
err := r.db.WithContext(repoCtx).Select("id").Where("name = ? AND plan_type = ?", planName, models.PlanTypeSystem).First(&plan).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil // 未找到系统计划不是错误
}
if err != nil {
return nil, fmt.Errorf("查询系统计划 '%s' 失败: %w", planName, err)
}
// 如果找到了计划ID，则复用 GetPlanByID 来获取完整的计划详情
return r.GetPlanByID(repoCtx, plan.ID)
}
// GetPlanByID 根据ID获取计划，包含子计划和任务详情
func (r *gormPlanRepository) GetPlanByID(ctx context.Context, id uint) (*models.Plan, error) {
repoCtx := logs.AddFuncName(ctx, r.ctx, "GetPlanByID")

1
project_structure.txt
View File

@@ -103,6 +103,7 @@ internal/domain/plan/plan_execution_manager.go
internal/domain/plan/plan_service.go
internal/domain/plan/task.go
internal/domain/task/alarm_notification_task.go
internal/domain/task/area_threshold_check_task.go
internal/domain/task/delay_task.go
internal/domain/task/device_threshold_check_task.go
internal/domain/task/full_collection_task.go