package tools // MQTT 报文统计(FWH iotcore-stat Elasticsearch,与 ToolBox“MQTT 报文管理 / MQTT 数据看板”同源)。 // ES 按 UTC 日建索引:日期、时间参数一律按 UTC 解释。 import ( "context" "encoding/json" "errors" "fmt" "strconv" "strings" "time" "github.com/mark3labs/mcp-go/mcp" "fwh-toolbox-server/agentmcp/kit" "fwh-toolbox-server/repositories" "fwh-toolbox-server/services" ) const mqttEsSource = "FWH MQTT 报文统计(ES)" // esDate 把 yyyy-MM-dd 转为 ES 索引日期 yyyy.MM.dd func esDate(name, v string) (string, error) { t, err := kit.ParseDate(name, v, false) if err != nil { return "", err } return t.Format("2006.01.02"), nil } // ---- mqtt_traffic_trend ---- type mqttTrendArgs struct { Month string `json:"month" jsonschema:"月份,yyyy-MM(UTC)"` } var MqttTrafficTrend = kit.Def[mqttTrendArgs]{ Name: "mqtt_traffic_trend", Title: "查询 MQTT 报文月度趋势", Description: "查询全部站点某月的 MQTT 报文总量与每日总量(UTC 日,整月补齐,缺失的天为 0)。" + "用于观察报文量的整体趋势、定位异常的日期;某天哪些站点报文最多用 mqtt_traffic_sites。返回 month、total、days(date、total)。", ReadOnly: true, Handler: func(ctx context.Context, a mqttTrendArgs) (*mcp.CallToolResult, error) { t, err := time.Parse("2006-01", strings.TrimSpace(a.Month)) if err != nil { return kit.Failf("参数 month 格式错误,需要 yyyy-MM:%s", a.Month), nil } total, err := services.GetMqttMonthlyTotal(t.Year(), int(t.Month())) if err != nil { return kit.Upstream(mqttEsSource, err), nil } days, err := services.GetMqttMonthlyDailyTrend(t.Year(), int(t.Month())) if err != nil { return kit.Upstream(mqttEsSource, err), nil } return kit.Object(map[string]any{"month": t.Format("2006-01"), "total": total.Total, "days": days}), nil }, } // ---- mqtt_traffic_sites ---- type mqttSitesArgs struct { Date string `json:"date" jsonschema:"日期,yyyy-MM-dd(UTC)"` Page int `json:"page,omitempty" jsonschema:"页码,默认 1"` Size int `json:"size,omitempty" jsonschema:"每页条数,1~100,默认 20"` } var MqttTrafficSites = kit.Def[mqttSitesArgs]{ Name: "mqtt_traffic_sites", Title: "查询站点 MQTT 报文量排行", Description: "查询某一天(UTC)各站点的 MQTT 报文量,按报文量从多到少分页返回。用于找出报文量异常偏高的站点。" + "返回 siteCount(当天有报文的站点数)与 items(sn、total)。某站点各命令类型的分布用 mqtt_traffic_site_cmd。", ReadOnly: true, Handler: func(ctx context.Context, a mqttSitesArgs) (*mcp.CallToolResult, error) { date, err := esDate("date", a.Date) if err != nil { return kit.Fail(err), nil } page, size := kit.Page(a.Page), kit.Limit(a.Size, 20, 100) res, err := services.GetMqttSiteDailyTotals(date, page, size) if err != nil { return kit.Upstream(mqttEsSource, err), nil } items := make([]map[string]any, 0, len(res.Items)) for _, it := range res.Items { items = append(items, map[string]any{"sn": it.EquipNo, "total": it.Total}) } return kit.List(items, int(res.SiteCount), size, fmt.Sprintf("第 %d 页,按报文量降序,可用 page 翻页", page)), nil }, } // ---- mqtt_traffic_site_cmd ---- type mqttSiteCmdArgs struct { SN string `json:"sn" jsonschema:"站点 SN(aGate 序列号)"` Date string `json:"date" jsonschema:"日期,yyyy-MM-dd(UTC)"` } var MqttTrafficSiteCmd = kit.Def[mqttSiteCmdArgs]{ Name: "mqtt_traffic_site_cmd", Title: "查询站点 MQTT 命令类型分布", Description: "查询单个站点某一天(UTC)各 MQTT 命令类型(cmdType)的报文量,按报文量降序。用于分析某站点报文量异常的来源。" + "返回 cmdTypeCount 与 items(cmdType、total)。具体报文用 mqtt_message_search 或 mqtt_log_search。", ReadOnly: true, Handler: func(ctx context.Context, a mqttSiteCmdArgs) (*mcp.CallToolResult, error) { sn, err := kit.Required("sn", a.SN) if err != nil { return kit.Fail(err), nil } date, err := esDate("date", a.Date) if err != nil { return kit.Fail(err), nil } res, err := services.GetMqttSiteCmdTypeTotals(sn, date, 1, 200) if err != nil { return kit.Upstream(mqttEsSource, err), nil } return kit.List(res.Items, int(res.CmdTypeCount), 200, ""), nil }, } // ---- mqtt_message_search ---- type mqttMessageSearchArgs struct { SN string `json:"sn,omitempty" jsonschema:"站点 SN;不填查全部站点(此时建议指定 cmdType 并缩短时间范围)"` CmdType *int `json:"cmdType,omitempty" jsonschema:"MQTT 命令类型,如 0、101、105、107、201、211"` From string `json:"from" jsonschema:"开始时间,yyyy-MM-dd HH:mm:ss(UTC)"` To string `json:"to" jsonschema:"结束时间,同上;与开始时间相差不超过 7 天"` Limit int `json:"limit,omitempty" jsonschema:"最多返回条数,1~200,默认 50"` } var MqttMessageSearch = kit.Def[mqttMessageSearchArgs]{ Name: "mqtt_message_search", Title: "检索 MQTT 报文(ES)", Description: "在 MQTT 报文统计库(ES)中按站点、命令类型、时间(UTC)检索逐条报文,按时间倒序。可以不指定站点,跨站点查找某类报文。" + "返回 total(命中数,可能为下界)与 items:time(UTC)、sn、cmdType,有内容时附 event(超过 1500 字截断)。" + "统计库主要保存报文的时间、站点与命令类型,多数报文没有 event 内容,适合统计与定位时间点;需要报文内容时用 mqtt_log_search(FWH 平台日志,时间为站点当地时间)。", ReadOnly: true, Handler: func(ctx context.Context, a mqttMessageSearchArgs) (*mcp.CallToolResult, error) { from, to, err := kit.ParseRange(a.From, a.To, 7*24*time.Hour) if err != nil { return kit.Fail(err), nil } // ParseRange 按本地时区解析;ES 的时间是 UTC,按字面值换为 UTC from = time.Date(from.Year(), from.Month(), from.Day(), from.Hour(), from.Minute(), from.Second(), 0, time.UTC) to = time.Date(to.Year(), to.Month(), to.Day(), to.Hour(), to.Minute(), to.Second(), 0, time.UTC) limit := kit.Limit(a.Limit, 50, 200) var sn *string if v := strings.TrimSpace(a.SN); v != "" { sn = &v } var cmd *int64 if a.CmdType != nil { v := int64(*a.CmdType) cmd = &v } res, err := services.QueryMqttMessages(sn, cmd, &from, &to, 1, limit) if errors.Is(err, repositories.ErrEsDeepPaging) { return kit.Failf("命中的报文太多,请缩短时间范围或指定站点、命令类型。"), nil } if err != nil { return kit.Upstream(mqttEsSource, err), nil } items := make([]map[string]any, 0, len(res.Items)) for _, m := range res.Items { item := map[string]any{"time": m.Timestamp, "sn": m.EquipNo, "cmdType": strconv.FormatInt(m.CmdType, 10)} if len(m.Event) > 0 { if b, err := json.Marshal(m.Event); err == nil { item["event"] = truncate(string(b), 1500) } } items = append(items, item) } return kit.List(items, int(res.Total), limit, "超过 limit 时请缩短时间范围或增加条件"), nil }, } func truncate(s string, n int) string { r := []rune(s) if len(r) <= n { return s } return string(r[:n]) + "…(已截断)" }