FranklinWH-Toolbox/service/agentmcp/tools/mqttstats.go

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]) + "…（已截断）"
}