FranklinWH-Toolbox/service/agentmcp/bootstrap.go

package agentmcp

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"

	"github.com/hibiken/asynq"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"

	"fwh-toolbox-server/cache"
	"fwh-toolbox-server/config"
	"fwh-toolbox-server/database"
	"fwh-toolbox-server/logger"
	"fwh-toolbox-server/repositories"
	task_core "fwh-toolbox-server/task/core"
	"fwh-toolbox-server/task/queue/faultRecordQueue"
	"fwh-toolbox-server/task/queue/fwhQueue"
	"fwh-toolbox-server/utils/emspara"
	"fwh-toolbox-server/utils/minio"
)

// DefaultConfigFile fwh-agent-mcp 的独立配置文件（相对工作目录），与主服务的 config.env、旧 MCP 的 config.mcp.env 互不共用。
// 示例见 deploy/config.agent-mcp.env.example。
const DefaultConfigFile = "./config.agent-mcp.env"

// Settings fwh-agent-mcp 自身的运行配置（AGENT_MCP_*）
type Settings struct {
	Token          string            // AGENT_MCP_TOKEN：Bearer 令牌，必填
	SetTokens      map[string]string // AGENT_MCP_TOKEN_<工具集大写>（如 AGENT_MCP_TOKEN_OPS）：该工具集单独的令牌，可选，未配置时用 Token
	Listen         string            // AGENT_MCP_LISTEN：默认 127.0.0.1:8091
	MaxConcurrency int               // AGENT_MCP_MAX_CONCURRENCY：默认 8
	LogPath        string            // AGENT_MCP_LOG_PATH：默认 /var/log/fwh-agent-mcp/agent-mcp.log
}

// LoadSettings 从 configFile 读取 AGENT_MCP_*，环境变量优先。
func LoadSettings(configFile string) (Settings, error) {
	v := viper.New()
	v.SetDefault("AGENT_MCP_LISTEN", "127.0.0.1:8091")
	v.SetDefault("AGENT_MCP_MAX_CONCURRENCY", 8)
	v.SetDefault("AGENT_MCP_LOG_PATH", "/var/log/fwh-agent-mcp/agent-mcp.log")
	v.SetConfigFile(configFile)
	v.SetConfigType("env")
	if err := v.ReadInConfig(); err != nil {
		return Settings{}, fmt.Errorf("读取配置文件 %s 失败: %w", configFile, err)
	}
	v.AutomaticEnv()
	setTokens := map[string]string{}
	for _, set := range AllSets {
		if t := strings.TrimSpace(v.GetString(setTokenKey(set))); t != "" {
			setTokens[set] = t
		}
	}
	return Settings{
		Token:          v.GetString("AGENT_MCP_TOKEN"),
		SetTokens:      setTokens,
		Listen:         v.GetString("AGENT_MCP_LISTEN"),
		MaxConcurrency: v.GetInt("AGENT_MCP_MAX_CONCURRENCY"),
		LogPath:        v.GetString("AGENT_MCP_LOG_PATH"),
	}, nil
}

func setTokenKey(set string) string {
	return "AGENT_MCP_TOKEN_" + strings.ToUpper(set)
}

// TokenFor 返回工具集使用的令牌：AGENT_MCP_TOKEN_<工具集> 优先，否则 AGENT_MCP_TOKEN。
func (s Settings) TokenFor(set string) string {
	if t := s.SetTokens[set]; t != "" {
		return t
	}
	return s.Token
}

// InitLogger 初始化日志：写入独立的日志文件；目录不可写时退回 stderr。
// 同时设置 zap 全局 logger 与 logger 包变量（database 等模块直接引用 logger.Sugar）。
func InitLogger(path string) *zap.Logger {
	outputs := []string{"stderr"}
	if path != "" {
		if err := os.MkdirAll(filepath.Dir(path), 0o755); err == nil {
			outputs = []string{path}
		} else {
			fmt.Fprintf(os.Stderr, "[fwh-agent-mcp] 创建日志目录失败，改为输出到 stderr: %v\n", err)
		}
	}
	cfg := zap.Config{
		Level:    zap.NewAtomicLevelAt(zap.InfoLevel),
		Encoding: "json",
		EncoderConfig: zapcore.EncoderConfig{
			TimeKey:        "ts",
			LevelKey:       "level",
			NameKey:        "logger",
			CallerKey:      "caller",
			MessageKey:     "msg",
			StacktraceKey:  "stacktrace",
			LineEnding:     zapcore.DefaultLineEnding,
			EncodeLevel:    zapcore.LowercaseLevelEncoder,
			EncodeTime:     zapcore.ISO8601TimeEncoder,
			EncodeDuration: zapcore.SecondsDurationEncoder,
			EncodeCaller:   zapcore.ShortCallerEncoder,
		},
		OutputPaths:      outputs,
		ErrorOutputPaths: outputs,
	}
	l, err := cfg.Build(zap.AddStacktrace(zap.ErrorLevel))
	if err != nil {
		fmt.Fprintf(os.Stderr, "[fwh-agent-mcp] logger 初始化失败: %v\n", err)
		l = zap.NewNop()
	}
	zap.ReplaceGlobals(l)
	logger.Log = l
	logger.Sugar = l.Sugar()
	logger.Std = zap.NewStdLog(l)
	return l
}

// InitInfra 从 configFile 加载配置并初始化工具用到的基础设施：数据库、Redis、MinIO、FWH 生产/测试 API、FWH ES、Plane、任务队列入队客户端。
func InitInfra(configFile string) error {
	cfg, err := config.LoadConfig(configFile)
	if err != nil {
		return fmt.Errorf("加载配置失败: %w", err)
	}
	if err := database.Init(&cfg.App); err != nil {
		return fmt.Errorf("数据库初始化失败: %w", err)
	}
	cache.Init(cfg.App.RedisAddr)
	// ems_param_lookup 依赖的 EMS 参数字典（部署目录的 emspara.db -> Redis），与主服务相同
	emspara.Init(emspara.DefaultDBPath)
	if err := repositories.Init(context.Background()); err != nil {
		return fmt.Errorf("Repository 初始化失败: %w", err)
	}
	repositories.FwhHttpInit(&cfg.FwhApi)
	repositories.FwhTestHttpInit(&cfg.FwhTestApi)
	// MQTT 报文统计（iotcore-stat Elasticsearch）：mqtt_traffic_*、mqtt_message_search
	repositories.EsHttpInit(&cfg.FwhEs)
	// 必须用 InitPlane（含 Feedback 专用实例 repositories.Plane），否则 feedback_create 会空指针
	repositories.InitPlane(&cfg.PlaneWebApi)
	// 只入队、不启动 worker：site_info_refresh、故障录波操作投递的任务由主服务执行（含 MQTT 收发）
	taskRedis := asynq.RedisClientOpt{Addr: cfg.App.RedisAddr, DB: task_core.RedisDB}
	fwhQueue.InitEnqueueClient(taskRedis)
	faultRecordQueue.InitEnqueueClient(taskRedis)
	// 故障录波文件（fault-record/ 前缀）的读取；使用 AI 只读账号即可
	minio.InitClient(&cfg.App)
	return nil
}