package tools import ( "context" "fmt" "time" "github.com/mark3labs/mcp-go/mcp" "fwh-toolbox-server/agentmcp/kit" "fwh-toolbox-server/domain" "fwh-toolbox-server/repositories" "fwh-toolbox-server/services" ) // 站点信息表同步(SiteInfoManageRecord)的状态判断。 // 同步由主服务的任务队列执行,耗时可能超过 1.5 小时;记录只有 progress、createTime(请求时间)、 // updateTime(完成时间,仅 complete 时更新)与 total,没有心跳,只能按请求时间判断是否疑似中断。 const ( // syncStaleAfter pending / running 超过该时长视为疑似中断(例如主服务重启),需要 force 才重新触发 syncStaleAfter = 6 * time.Hour // syncFreshWithin 最近一次成功同步在该时长内时认为数据较新,不重新触发(force 除外) syncFreshWithin = 24 * time.Hour ) // 同步状态 const ( syncStateNever = "never" // 从未同步 syncStateSyncing = "syncing" // 同步中(pending / running) syncStateStale = "stale" // pending / running 但超过 syncStaleAfter,疑似中断 syncStateIdle = "idle" // 没有进行中的同步(最近一次已完成或失败) ) var beijing = time.FixedZone("CST", 8*3600) func fmtBeijing(t time.Time) string { return t.In(beijing).Format(time.DateTime) } func fmtHours(d time.Duration) string { if d < time.Hour { return fmt.Sprintf("%d 分钟", int(d.Minutes())) } return fmt.Sprintf("%.1f 小时", d.Hours()) } type siteInfoSync struct { State string Latest *domain.SiteInfoUpdateRecord LastSuccess *domain.SiteInfoUpdateRecord now time.Time } func loadSiteInfoSync() (*siteInfoSync, error) { latest, err := repositories.GetSiteInfoUpdateNewRecord() if err != nil { return nil, err } lastComplete, err := repositories.GetSiteInfoUpdateLastComplete() if err != nil { return nil, err } return evalSiteInfoSync(latest, lastComplete, time.Now()), nil } // evalSiteInfoSync 根据最近一次记录与最近一次成功记录判断同步状态 func evalSiteInfoSync(latest, lastComplete *domain.SiteInfoUpdateRecord, now time.Time) *siteInfoSync { s := &siteInfoSync{Latest: latest, LastSuccess: lastComplete, now: now} switch { case latest == nil: s.State = syncStateNever case latest.Progress == domain.ProgressPending || latest.Progress == domain.ProgressRunning: s.State = syncStateSyncing if s.now.Sub(latest.CreateTime) > syncStaleAfter { s.State = syncStateStale } default: s.State = syncStateIdle } return s } // decideRefresh 决定 site_info_refresh 是否触发同步;trigger=false 时 action 说明原因 func (s *siteInfoSync) decideRefresh(force bool) (action string, trigger bool) { switch { case s.State == syncStateSyncing: return "tracking", false case s.State == syncStateStale && !force: return "skipped_stale", false case s.State == syncStateIdle && s.Latest.Progress == domain.ProgressComplete && s.fresh() && !force: return "skipped_fresh", false } return "triggered", true } // fresh 最近一次成功同步在 syncFreshWithin 内 func (s *siteInfoSync) fresh() bool { return s.LastSuccess != nil && s.now.Sub(s.LastSuccess.UpdateTime) < syncFreshWithin } func (s *siteInfoSync) view() map[string]any { out := map[string]any{"state": s.State} if r := s.Latest; r != nil { latest := map[string]any{ "id": r.Id.Hex(), "progress": r.Progress, "requestedAt": fmtBeijing(r.CreateTime), } if r.Progress == domain.ProgressComplete { latest["completedAt"] = fmtBeijing(r.UpdateTime) latest["duration"] = fmtHours(r.UpdateTime.Sub(r.CreateTime)) latest["total"] = r.Total } else { latest["elapsed"] = fmtHours(s.now.Sub(r.CreateTime)) } out["latest"] = latest } if r := s.LastSuccess; r != nil { out["lastSuccess"] = map[string]any{ "completedAt": fmtBeijing(r.UpdateTime), "age": fmtHours(s.now.Sub(r.UpdateTime)), "total": r.Total, } } out["advice"] = s.advice() return out } func (s *siteInfoSync) advice() string { switch s.State { case syncStateNever: return "从未同步过,可以调用 site_info_refresh 触发同步。" case syncStateSyncing: return fmt.Sprintf("同步进行中(已 %s),无需重新触发。同步耗时可能超过 1.5 小时,稍后再用 site_info_sync_status 查看。", fmtHours(s.now.Sub(s.Latest.CreateTime))) case syncStateStale: return fmt.Sprintf("最近一次同步已请求 %s 仍未完成,超过 %s,疑似中断(如主服务重启)。确认后可用 site_info_refresh 并设置 force=true 重新同步。", fmtHours(s.now.Sub(s.Latest.CreateTime)), fmtHours(syncStaleAfter)) } if s.Latest.Progress != domain.ProgressComplete { return "最近一次同步失败,可以调用 site_info_refresh 重新同步。" } if s.fresh() { return fmt.Sprintf("最近一次同步在 %s 前完成,数据较新,一般无需重新同步;确需最新数据时用 site_info_refresh 并设置 force=true。", fmtHours(s.now.Sub(s.LastSuccess.UpdateTime))) } return fmt.Sprintf("最近一次同步在 %s 前完成,超过 %s,可以调用 site_info_refresh 重新同步。", fmtHours(s.now.Sub(s.LastSuccess.UpdateTime)), fmtHours(syncFreshWithin)) } // ---- site_info_sync_status ---- var SiteInfoSyncStatus = kit.Def[kit.NoArgs]{ Name: "site_info_sync_status", Title: "查询站点信息表同步状态", Description: "查询 ToolBox 站点信息表(site_search 的数据来源)从 FWH 平台同步的状态,只读。" + "返回 state:never(从未同步)、syncing(同步中)、stale(超过 6 小时未完成,疑似中断)、idle(没有进行中的同步);" + "latest(最近一次同步:progress 为 pending/running/complete/error,requestedAt 请求时间,完成时有 completedAt、duration、total 站点数,未完成时有 elapsed);" + "lastSuccess(最近一次成功同步的 completedAt、age 距今时长、total);advice(是否需要重新同步的建议)。时间为北京时间。" + "同步耗时可能超过 1.5 小时:触发同步前先调用本工具,同步中时用本工具跟踪进度,不要重复触发。", ReadOnly: true, Handler: func(ctx context.Context, _ kit.NoArgs) (*mcp.CallToolResult, error) { s, err := loadSiteInfoSync() if err != nil { return kit.Upstream("ToolBox 站点信息同步记录", err), nil } return kit.Object(s.view()), nil }, } // ---- site_info_refresh ---- var refreshMessages = map[string]string{ "tracking": "同步进行中,未重复触发。", "skipped_stale": "上次同步疑似中断,未触发;确认后设置 force=true 重新同步。", "skipped_fresh": "最近 24 小时内已成功同步,未触发;确需最新数据时设置 force=true。", "triggered": "已触发同步,任务在主服务后台执行,耗时可能超过 1.5 小时;用 site_info_sync_status 跟踪进度。", } type siteInfoRefreshArgs struct { Force bool `json:"force,omitempty" jsonschema:"为 true 时:最近 24 小时内已成功同步、或上次同步疑似中断(stale)时仍重新触发。同步进行中(syncing)时任何情况下都不会重复触发"` } var SiteInfoRefresh = kit.Def[siteInfoRefreshArgs]{ Name: "site_info_refresh", Title: "触发站点信息表同步", Description: "触发 ToolBox 站点信息表从 FWH 平台重新同步,立即返回,同步在主服务后台执行,耗时可能超过 1.5 小时。" + "触发前先检查同步状态:同步进行中时不重复触发,返回当前进度(action=tracking);最近 24 小时内已成功同步时不触发(action=skipped_fresh);" + "上次同步超过 6 小时未完成、疑似中断时不触发(action=skipped_stale);以上两种不触发的情况可用 force=true 强制触发。" + "已触发时 action=triggered。返回中的 status 与 site_info_sync_status 相同,之后用 site_info_sync_status 跟踪进度。" + "建议先调用 site_info_sync_status 判断是否需要同步,再调用本工具。这是写操作,执行前会请求确认。", ReadOnly: false, Handler: func(ctx context.Context, a siteInfoRefreshArgs) (*mcp.CallToolResult, error) { s, err := loadSiteInfoSync() if err != nil { return kit.Upstream("ToolBox 站点信息同步记录", err), nil } action, trigger := s.decideRefresh(a.Force) if trigger { if err := services.UpdateSiteInfoDate(); err != nil { return kit.Upstream("ToolBox 站点信息同步", err), nil } if s, err = loadSiteInfoSync(); err != nil { return kit.Upstream("ToolBox 站点信息同步记录", err), nil } } return kit.Object(map[string]any{"action": action, "message": refreshMessages[action], "status": s.view()}), nil }, }