// 内容监控的落库逻辑:基线建立、变更判定、防抖、人工处理。 // // 判定原则(避免误报的核心): // 1. 拨测本身失败(连不上/超时/非预期状态码)→ 完全不参与内容比对 // 2. 拿到响应但规范化判定为"采集异常"(空/过短/非 JSON)→ 也不参与比对 // 3. 只有"拿到了有效数据且与基线不同"才算内容变更 package siteprobe import ( "encoding/json" "fmt" "time" "ezops/model" "gorm.io/gorm" "gorm.io/gorm/clause" ) // ApplyContentCheck 对一次成功拨测的响应体执行内容比对。 // // 返回 changed=true 表示本次达到确认阈值、已写入变更记录(应触发告警)。 // Logan.Li func ApplyContentCheck(db *gorm.DB, task *model.SiteTask, body string, probeOK bool) (bool, error) { if task == nil || !ContentCheckEnabled(task.ContentCheckMode) { return false, nil } // 拨测本身失败时不做内容判定:否则网络抖动会被误报成"内容变更"。 // Logan.Li if !probeOK { return false, nil } snap := BuildSnapshot( task.ContentCheckMode, body, ParseRules(task.ContentPaths), ParseRules(task.ContentIgnoreRules), task.ContentMinLength, task.ContentSortLines, ) if snap.Invalid != "" { // 采集异常:既不算变更,也不覆盖基线,等下次恢复。 return false, fmt.Errorf("内容采集异常: %s", snap.Invalid) } // 无基线 → 自动建立 v1(按用户选择采用"自动接受",否则新任务永远跑不动)。 // 必须用行锁串行化:RunTaskNow 与后台 worker 可能同时拨测, // 不加锁会各自看到 current_baseline_id=0 并各建一条基线(实测出现 v1/v2 重复)。 // Logan.Li if task.CurrentBaselineID == 0 { return false, db.Transaction(func(tx *gorm.DB) error { var locked model.SiteTask if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&locked, task.ID).Error; err != nil { return err } if locked.CurrentBaselineID != 0 { // 已被其它协程建立,本次直接返回。 // Logan.Li task.CurrentBaselineID = locked.CurrentBaselineID return nil } baseline, err := CreateBaseline(tx, task.ID, snap, "auto", "", "首次采集自动建立基线") if err != nil { return err } return tx.Model(&model.SiteTask{}).Where("id = ?", task.ID). Updates(map[string]interface{}{ "current_baseline_id": baseline.ID, "pending_change_count": 0, }).Error }) } var baseline model.SiteBaseline if err := db.First(&baseline, task.CurrentBaselineID).Error; err != nil { // 基线丢了(被清理/迁移),重新建立而不是刷变更。 // Logan.Li nb, cerr := CreateBaseline(db, task.ID, snap, "auto", "", "基线缺失,自动重建") if cerr != nil { return false, cerr } db.Model(&model.SiteTask{}).Where("id = ?", task.ID). Update("current_baseline_id", nb.ID) return false, nil } // 内容与基线一致 → 重置连续计数,说明之前的"变更"是抖动。 if snap.Hash == baseline.ContentHash { if task.PendingChangeCount > 0 { db.Model(&model.SiteTask{}).Where("id = ?", task.ID). Update("pending_change_count", 0) } return false, nil } // 内容不同 → 累计连续次数,达到阈值才落变更记录并告警。 // Logan.Li confirmTimes := task.ContentConfirmTimes if confirmTimes <= 0 { confirmTimes = 2 } pending := task.PendingChangeCount + 1 if pending < confirmTimes { db.Model(&model.SiteTask{}).Where("id = ?", task.ID). Update("pending_change_count", pending) return false, nil } // 同一 hash 的 pending 变更已存在时不重复创建,避免连续两轮各写一条。 // Logan.Li var existing model.SiteContentChange if err := db.Where("task_id = ? AND status = ? AND content_hash = ?", task.ID, model.ContentChangePending, snap.Hash). Order("id DESC").Take(&existing).Error; err == nil { db.Model(&model.SiteTask{}).Where("id = ?", task.ID). Update("pending_change_count", 0) return false, nil } diffType := "text_line" var payload any if task.ContentCheckMode == ContentModeJSON { diffType = "json_field" payload = DiffJSONFields(baseline.ContentText, snap.Canonical) } else { payload = DiffText(baseline.ContentText, snap.Canonical) } diffJSON, _ := json.Marshal(payload) change := model.SiteContentChange{ TaskID: task.ID, FromBaselineID: baseline.ID, FromVersion: baseline.Version, ContentHash: snap.Hash, ContentText: snap.Canonical, DiffType: diffType, DiffPayload: string(diffJSON), Status: model.ContentChangePending, DetectedAt: time.Now(), } if err := db.Create(&change).Error; err != nil { return false, err } db.Model(&model.SiteTask{}).Where("id = ?", task.ID). Update("pending_change_count", 0) return true, nil } // CreateBaseline 追加一个新版本基线。 // Logan.Li func CreateBaseline(db *gorm.DB, taskID uint, snap ContentSnapshot, source, acceptedBy, note string) (*model.SiteBaseline, error) { var maxVersion int db.Model(&model.SiteBaseline{}).Where("task_id = ?", taskID). Select("COALESCE(MAX(version), 0)").Scan(&maxVersion) baseline := model.SiteBaseline{ TaskID: taskID, Version: maxVersion + 1, ContentHash: snap.Hash, ContentText: snap.Canonical, Source: source, AcceptedBy: acceptedBy, Note: note, CreatedAt: time.Now(), } if err := db.Create(&baseline).Error; err != nil { return nil, err } return &baseline, nil } // AcceptChange 把某次变更接受为新基线。 // // 只有人工确认才走这里:真实变更(上游确实改了)与误报(我方配置问题)需人工判断, // 自动接受会让告警失去意义。 // Logan.Li func AcceptChange(db *gorm.DB, changeID uint, acceptedBy, note string) error { var change model.SiteContentChange if err := db.First(&change, changeID).Error; err != nil { return err } if change.Status != model.ContentChangePending { return fmt.Errorf("该变更已处理(当前状态 %s)", change.Status) } var task model.SiteTask if err := db.First(&task, change.TaskID).Error; err != nil { return err } snap := ContentSnapshot{ Mode: task.ContentCheckMode, Canonical: change.ContentText, Hash: change.ContentHash, } baseline, err := CreateBaseline(db, change.TaskID, snap, "manual", acceptedBy, note) if err != nil { return err } now := time.Now() return db.Transaction(func(tx *gorm.DB) error { if err := tx.Model(&model.SiteContentChange{}).Where("id = ?", change.ID). Updates(map[string]interface{}{ "status": model.ContentChangeAccepted, "handled_by": acceptedBy, "handled_at": now, }).Error; err != nil { return err } // 同一基线的其它 pending 变更一并作废,避免重复要求确认。 if err := tx.Model(&model.SiteContentChange{}). Where("task_id = ? AND status = ? AND from_baseline_id = ?", change.TaskID, model.ContentChangePending, change.FromBaselineID). Updates(map[string]interface{}{ "status": model.ContentChangeAccepted, "handled_by": acceptedBy, "handled_at": now, }).Error; err != nil { return err } return tx.Model(&model.SiteTask{}).Where("id = ?", change.TaskID). Updates(map[string]interface{}{ "current_baseline_id": baseline.ID, "pending_change_count": 0, }).Error }) } // IgnoreChange 把变更标记为误报忽略(不更新基线)。 // // 忽略后如果内容仍未恢复,下一轮会再次累积到阈值并重新产生变更记录, // 这是期望行为——避免"忽略一次就永久静默"。 // Logan.Li func IgnoreChange(db *gorm.DB, changeID uint, handledBy string) error { var change model.SiteContentChange if err := db.First(&change, changeID).Error; err != nil { return err } if change.Status != model.ContentChangePending { return fmt.Errorf("该变更已处理(当前状态 %s)", change.Status) } now := time.Now() return db.Model(&model.SiteContentChange{}).Where("id = ?", change.ID). Updates(map[string]interface{}{ "status": model.ContentChangeIgnored, "handled_by": handledBy, "handled_at": now, }).Error } // ContentCheckEnabled 判断任务是否启用了内容监控。 // Logan.Li func ContentCheckEnabled(mode string) bool { return mode == ContentModeText || mode == ContentModeJSON }