newRobotProcess/service/metric_process_config.go

package service

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"text/template"
	"time"

	jsoniter "github.com/json-iterator/go"
	"github.com/tidwall/gjson"
	"newaterobot-process/entity"
	"newaterobot-process/model"
	"newaterobot-process/utils"
)

// MetricProcessConfigService 指标处理配置服务
type MetricProcessConfigService struct {
	configModel *model.MetricProcessConfigModel
	httpClient  *http.Client
}

// NewMetricProcessConfigService 创建指标处理配置服务实例
func NewMetricProcessConfigService() *MetricProcessConfigService {
	return &MetricProcessConfigService{
		configModel: &model.MetricProcessConfigModel{},
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
		},
	}
}

// ProcessMetric 处理指标
func (s *MetricProcessConfigService) ProcessMetric(inputData entity.MetricProcessRequest) (entity.MetricProcessResponse, error) {
	ret := entity.MetricProcessResponse{}
	// 根据指标类型获取配置
	configs, err := s.configModel.GetByMetricType(inputData.MetricType, inputData.Intent)
	if err != nil {
		return ret, err
	}

	configsMap := make(map[string]entity.MetricProcessConfig)
	for _, config := range configs {
		key := fmt.Sprintf("%d_%d", config.Intent, config.MetricType)
		configsMap[key] = config
	}

	key := fmt.Sprintf("%d_%d", inputData.Intent, inputData.MetricType)
	var config entity.MetricProcessConfig
	var ok bool

	// 未配置该处理分支直接返回，在dify中处理
	if config, ok = configsMap[key]; !ok {
		ret.Flag = entity.NoExecute
		return ret, nil
	}
	if config.MetricType == 24 {
		inputData.STime = time.Now().Format("2006-01-02") + " 00:00:00"
		inputData.ETime = time.Now().Format("2006-01-02") + " 23:59:59"
	} else if config.MetricType == 29 {
		inputData.STime = time.Now().Format("2006-01")
	} else if config.MetricType == 30 {
		inputData.STime = time.Now().Format("2006-01-02") + " 00:00:00"
	} else if config.MetricType == 33 {
		inputData.STime = time.Now().Format("2006-01-02")
		inputData.ETime = time.Now().AddDate(0, 0, 180).Format("2006-01-02")
	} else if config.MetricType == 59 {
		inputData.DateTime = strings.Split(inputData.DateTime, " ")[0]
	} else if config.MetricType == 42 {
		inputData.STime = strconv.Itoa(time.Now().Year())
		inputData.ETime = strconv.Itoa(int(time.Now().Month()))
	}
	// 直接回复分支
	templateVarMap := s.objToMap(inputData)
	if config.Type == 1 {
		contextTemplate := config.ContextTemplate
		if inputData.IsEnglish == 1 {
			contextTemplate = config.EngContextTemplate
		}
		replyContext, err := s.parseTemplate(contextTemplate, templateVarMap)
		if err != nil {
			return ret, err
		}
		ret.ReplyContext = replyContext
		return ret, nil
	}

	// 发送http 请求
	var responseData string
	if strings.Contains(config.URLTemplate, ";") {
		urlList := strings.Split(config.URLTemplate, ";")
		pamList := strings.Split(config.QueryTemplate, ";")
		for idx, url := range urlList {
			config.URLTemplate = url
			config.QueryTemplate = pamList[idx]
			if strings.HasPrefix(pamList[idx], "{") {
				config.HTTPMethod = 1
			} else {
				config.HTTPMethod = 0
			}
			response, err := s.sendHTTPRequest(&config, templateVarMap, inputData.JwtToken)
			if err != nil {
				return ret, err
			}
			responseData = responseData + response + "#@#@#@"
		}
		responseData = strings.TrimRight(responseData, "#@#@#@")
	} else {
		responseData, err = s.sendHTTPRequest(&config, templateVarMap, inputData.JwtToken)
		if err != nil {
			return ret, err
		}
	}

	// 抽取字段
	responseExtract := make(map[string]string)
	err = json.Unmarshal([]byte(config.ResponseExtract), &responseExtract)
	if err != nil {
		return ret, err
	}

	// 额外逻辑处理
	switch config.MetricType {
	case 2:
		responseData, _ = TransformType2(responseData)
	case 4:
		responseData, _ = TransformType4(responseData)
	case 5:
		responseData, _ = TransformType5(responseData)
	case 6:
		responseData, _ = TransformType6(responseData)
	case 8:
		responseData, _ = TransformType8(responseData)
	case 15:
		responseData, _ = TransformType15(responseData)
	case 17:
		responseData, _ = TransformType17(responseData)
	case 19:
		responseData, _ = TransformType19(responseData)
	case 21:
		responseData, _ = TransformType21(responseData)
	case 22:
		responseData, _ = TransformType22(responseData)
	case 23:
		responseData, _ = TransformType23(responseData)
	case 24:
		responseData, _ = TransformType24(responseData)
	case 25, 26, 27:
		responseData, _ = TransformType25(responseData, inputData.IsEnglish)
	case 28:
		responseData, _ = TransformType28(responseData, inputData.IsEnglish)
	case 29:
		responseData, _ = TransformType29(responseData)
	case 30:
		responseData, _ = TransformType30(responseData)
	case 35:
		responseData, _ = TransformType35(responseData)
	case 36:
		responseData, _ = TransformType36(responseData)
	case 38:
		responseData, _ = TransformType38(responseData)
	case 39:
		responseData, _ = TransformType39(responseData)
	case 42:
		responseData, _ = TransformType42(responseData)
	case 51:
		responseData, _ = TransformType51(responseData)
	case 59:
		responseData, _ = TransformType59(responseData, inputData.IsEnglish, inputData.DateTime)
	}

	extractMap := s.extractFields(responseExtract, responseData)
	for k, v := range extractMap {
		templateVarMap[k] = v
	}

	// 解析返回模板
	replyContext, err := s.parseTemplate(config.ContextTemplate, templateVarMap)
	if err != nil {
		return ret, err
	}
	ret.Flag = entity.HttpReply
	ret.ReplyContext = replyContext

	info := make(map[string]interface{})
	info["pid"] = inputData.ID
	info["metric_type"] = config.MetricType
	info["datetime"] = inputData.DateTime
	info["metric"] = replyContext
	info["is_english"] = inputData.IsEnglish
	info["type"] = config.Type
	k := fmt.Sprintf("deepseek2:%s:%s", inputData.ConversationId, inputData.DialogueCount)
	msg, _ := jsoniter.MarshalToString(info)
	GetRedisService().SetObject(k, msg, 3600*24*7)
	return ret, nil
}

func (s *MetricProcessConfigService) ProcessMate(cid string, mid string) (map[string]interface{}, error) {
	k := fmt.Sprintf("deepseek2:%s:%s", cid, mid)
	row, err := GetRedisService().GetString(k)
	if err != nil {
		return nil, err
	}
	info := make(map[string]interface{})
	err = json.Unmarshal([]byte(row), &info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

func (s *MetricProcessConfigService) objToMap(req entity.MetricProcessRequest) map[string]interface{} {
	ret := make(map[string]interface{})
	ret["name"] = req.Name
	ret["think_str"] = entity.ThinkStr
	ret["date_src"] = req.DateSrc
	ret["s_time"] = req.STime
	ret["e_time"] = req.ETime
	ret["local_url"] = req.LocalUrl
	ret["id"] = req.ID
	ret["datetime"] = req.DateTime
	return ret
}

// sendHTTPRequest 发送HTTP请求
func (s *MetricProcessConfigService) sendHTTPRequest(config *entity.MetricProcessConfig, inputData map[string]interface{}, jwtToken string) (string, error) {
	// 解析URL模板
	reqUrl, err := s.parseTemplate(config.URLTemplate, inputData)
	if err != nil {
		return "", fmt.Errorf("解析URL模板失败: %v", err)
	}

	var req *http.Request
	switch config.HTTPMethod {
	case 0: // GET
		// 解析查询参数模板
		query, err := s.parseTemplate(config.QueryTemplate, inputData)
		if err != nil {
			return "", fmt.Errorf("解析查询参数模板失败: %v", err)
		}

		// 对查询参数进行urlencode处理
		if query != "" {
			pairs := strings.Split(query, "&")
			encodedQuery := ""
			for i, pair := range pairs {
				if i > 0 {
					encodedQuery += "&"
				}
				// 分割键值对
				kv := strings.Split(pair, "=")
				if len(kv) == 2 {
					// 对键和值都进行urlencode
					encodedKey := url.QueryEscape(kv[0])
					encodedValue := url.QueryEscape(kv[1])
					encodedQuery += encodedKey + "=" + encodedValue
				}

			}
			reqUrl += "?" + encodedQuery
		}

		req, err = http.NewRequest("GET", reqUrl, nil)
		if err != nil {
			return "", fmt.Errorf("创建GET请求失败: %v", err)
		}

	case 1: // POST

		// 解析请求体模板
		bodyStr, err := s.parseTemplate(config.QueryTemplate, inputData)
		if err != nil {
			return "", fmt.Errorf("解析请求体模板失败: %v", err)
		}

		body := strings.NewReader(bodyStr)
		req, err = http.NewRequest("POST", reqUrl, body)
		if err != nil {
			return "", fmt.Errorf("创建POST请求失败: %v", err)
		}

		// 设置默认内容类型
		req.Header.Set("Content-Type", "application/json")
	default:
		return "", fmt.Errorf("不支持的HTTP方法: %d", config.HTTPMethod)
	}
	req.Header.Set("JWT-TOKEN", jwtToken)
	// 发送请求
	resp, err := s.httpClient.Do(req)
	if err != nil {
		return "", fmt.Errorf("发送请求失败: %v", err)
	}
	defer resp.Body.Close()

	// 读取响应
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("读取响应失败: %v", err)
	}

	responseString := string(body)

	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("请求失败，状态码: %d，响应: %s", resp.StatusCode, responseString)
	}

	return responseString, nil
}

// extractFields 提取指定字段
func (s *MetricProcessConfigService) extractFields(rules map[string]string, responseBody string) map[string]interface{} {
	ret := make(map[string]interface{})
	for field, rule := range rules {
		result := gjson.Get(responseBody, rule)
		ret[field] = result.Value()
	}
	return ret
}

// parseTemplate 解析模板
func (s *MetricProcessConfigService) parseTemplate(tpl string, data map[string]interface{}) (string, error) {
	if tpl == "" {
		return "", nil
	}
	funcMap := template.FuncMap{
		"SplitAndFetch": utils.SplitAndFetch,
	}
	// 创建模板
	t, err := template.New("").Funcs(funcMap).Parse(tpl)
	if err != nil {
		return "", fmt.Errorf("解析模板失败: %v", err)
	}

	// 执行模板
	var buf bytes.Buffer
	if err := t.Execute(&buf, data); err != nil {
		return "", fmt.Errorf("执行模板失败: %v", err)
	}

	return buf.String(), nil
}