|
|
@@ -0,0 +1,186 @@
|
|
|
+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]) + "…（已截断）"
|
|
|
+}
|
Jerry Liang