fadd
This commit is contained in:
432
internal/infrastructure/external/rongxing/rongxing_service.go
vendored
Normal file
432
internal/infrastructure/external/rongxing/rongxing_service.go
vendored
Normal file
@@ -0,0 +1,432 @@
|
||||
package rongxing
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"hyapi-server/internal/shared/external_logger"
|
||||
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultRequestTimeout = 10 * time.Second
|
||||
apiKeyLogin = "auth_login"
|
||||
pathAuthLogin = "/auth/login"
|
||||
headerDmsToken = "dms-token"
|
||||
)
|
||||
|
||||
// serviceConfig 戎行服务运行时配置
|
||||
type serviceConfig struct {
|
||||
BaseURL string
|
||||
Account string
|
||||
Password string
|
||||
AppID string
|
||||
PrivateKey string
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
// RongxingService 戎行数据源服务
|
||||
type RongxingService struct {
|
||||
config serviceConfig
|
||||
logger *external_logger.ExternalServiceLogger
|
||||
|
||||
tokenMu sync.RWMutex
|
||||
cachedToken string
|
||||
}
|
||||
|
||||
// apiResponse 戎行统一响应。code 可能是数字或字符串；扣费以 consumeFlag 为准。
|
||||
type apiResponse struct {
|
||||
Flag bool `json:"flag"`
|
||||
Code json.RawMessage `json:"code"`
|
||||
Msg string `json:"msg"`
|
||||
Message string `json:"message"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
ConsumeFlag int `json:"consumeFlag"`
|
||||
}
|
||||
|
||||
func (r apiResponse) text() string {
|
||||
if r.Msg != "" {
|
||||
return r.Msg
|
||||
}
|
||||
return r.Message
|
||||
}
|
||||
|
||||
func (r apiResponse) code() string {
|
||||
return parseCode(r.Code)
|
||||
}
|
||||
|
||||
// NewRongxingService 创建戎行服务实例
|
||||
func NewRongxingService(cfg serviceConfig, logger *external_logger.ExternalServiceLogger) *RongxingService {
|
||||
if cfg.Timeout <= 0 {
|
||||
cfg.Timeout = defaultRequestTimeout
|
||||
}
|
||||
cfg.BaseURL = strings.TrimRight(strings.TrimSpace(cfg.BaseURL), "/")
|
||||
return &RongxingService{config: cfg, logger: logger}
|
||||
}
|
||||
|
||||
// GetConfig 获取运行时配置
|
||||
func (s *RongxingService) GetConfig() serviceConfig {
|
||||
return s.config
|
||||
}
|
||||
|
||||
// CallAPI 通用业务接口调用。
|
||||
// apiPath 为相对路径（如 /third/loan/info360），reqData 为已组装好的请求体。
|
||||
// Token 获取与 Header 注入由服务内部处理；401/403 时自动刷新 Token 并重试一次。
|
||||
func (s *RongxingService) CallAPI(ctx context.Context, apiPath string, reqData map[string]interface{}) ([]byte, error) {
|
||||
apiKey := strings.Trim(apiPath, "/")
|
||||
|
||||
var transactionID string
|
||||
if id, ok := ctx.Value("transaction_id").(string); ok {
|
||||
transactionID = id
|
||||
}
|
||||
|
||||
if err := s.validateConfig(); err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logError(transactionID, apiKey, "", err, nil)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(apiPath, "/") {
|
||||
apiPath = "/" + apiPath
|
||||
}
|
||||
requestURL := s.config.BaseURL + apiPath
|
||||
|
||||
bodyBytes, err := json.Marshal(reqData)
|
||||
if err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logError(transactionID, apiKey, "", err, reqData)
|
||||
return nil, err
|
||||
}
|
||||
bodyStr := string(bodyBytes)
|
||||
|
||||
for attempt := 0; attempt < 2; attempt++ {
|
||||
startTime := time.Now()
|
||||
|
||||
token, err := s.getToken(ctx, transactionID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
headers := map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
headerDmsToken: token,
|
||||
}
|
||||
curlCmd := generateCurlCommand(http.MethodPost, requestURL, headers, bodyStr)
|
||||
s.logCurl(transactionID, apiKey, requestURL, curlCmd)
|
||||
|
||||
if s.logger != nil {
|
||||
s.logger.LogRequest("", transactionID, apiKey, requestURL)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL, bytes.NewBuffer(bodyBytes))
|
||||
if err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logErrorWithCurl(transactionID, apiKey, err, reqData, curlCmd, "")
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set(headerDmsToken, token)
|
||||
|
||||
respBody, statusCode, err := s.doHTTP(req)
|
||||
if err != nil {
|
||||
err = errors.Join(ErrDatasource, err)
|
||||
s.logErrorWithCurl(transactionID, apiKey, err, reqData, curlCmd, "")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
duration := time.Since(startTime)
|
||||
respStr := string(respBody)
|
||||
|
||||
if statusCode == http.StatusUnauthorized || statusCode == http.StatusForbidden {
|
||||
s.clearToken()
|
||||
if attempt == 0 {
|
||||
continue
|
||||
}
|
||||
err = errors.Join(ErrDatasource, fmt.Errorf("HTTP状态码 %d", statusCode))
|
||||
s.logErrorWithCurl(transactionID, apiKey, err, reqData, curlCmd, respStr)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if statusCode != http.StatusOK {
|
||||
err = errors.Join(ErrDatasource, fmt.Errorf("HTTP状态码 %d", statusCode))
|
||||
s.logErrorWithCurl(transactionID, apiKey, err, reqData, curlCmd, respStr)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var resp apiResponse
|
||||
if err := json.Unmarshal(respBody, &resp); err != nil {
|
||||
err = errors.Join(ErrSystem, fmt.Errorf("响应解析失败: %w, body=%s", err, respStr))
|
||||
s.logErrorWithCurl(transactionID, apiKey, err, reqData, curlCmd, respStr)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
code := resp.code()
|
||||
payload := extractBusinessPayload(resp.Data)
|
||||
s.logResponse(transactionID, apiKey, statusCode, duration, respStr)
|
||||
|
||||
// 扣费只看 consumeFlag：1 扣费（按成功返回），0 不扣费
|
||||
if IsBillable(resp.ConsumeFlag) {
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
sentinel := MapNonBillableToErr(code)
|
||||
err = errors.Join(sentinel, NewRongxingError(code, resp.text()))
|
||||
if !errors.Is(sentinel, ErrNotFound) {
|
||||
s.logErrorWithCurl(transactionID, apiKey, err, reqData, curlCmd, respStr)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return nil, errors.Join(ErrDatasource, errors.New("请求失败"))
|
||||
}
|
||||
|
||||
func (s *RongxingService) getToken(ctx context.Context, transactionID string) (string, error) {
|
||||
s.tokenMu.RLock()
|
||||
token := s.cachedToken
|
||||
s.tokenMu.RUnlock()
|
||||
if token != "" {
|
||||
return token, nil
|
||||
}
|
||||
|
||||
s.tokenMu.Lock()
|
||||
defer s.tokenMu.Unlock()
|
||||
if s.cachedToken != "" {
|
||||
return s.cachedToken, nil
|
||||
}
|
||||
|
||||
token, err := s.login(ctx, transactionID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
s.cachedToken = token
|
||||
return token, nil
|
||||
}
|
||||
|
||||
func (s *RongxingService) clearToken() {
|
||||
s.tokenMu.Lock()
|
||||
s.cachedToken = ""
|
||||
s.tokenMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *RongxingService) login(ctx context.Context, transactionID string) (string, error) {
|
||||
privateKey, err := ParsePrivateKey(s.config.PrivateKey)
|
||||
if err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logError(transactionID, apiKeyLogin, "", err, nil)
|
||||
return "", err
|
||||
}
|
||||
|
||||
passwordB64 := EncodePasswordBase64(s.config.Password)
|
||||
timestamp := time.Now().UnixMilli()
|
||||
signParams := map[string]interface{}{
|
||||
"account": s.config.Account,
|
||||
"password": passwordB64,
|
||||
"appId": s.config.AppID,
|
||||
"timestamp": timestamp,
|
||||
}
|
||||
content := BuildSignContent(signParams)
|
||||
sign, err := SignSHA256WithRSA(content, privateKey)
|
||||
if err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logError(transactionID, apiKeyLogin, "", err, nil)
|
||||
return "", err
|
||||
}
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"account": s.config.Account,
|
||||
"password": passwordB64,
|
||||
"appId": s.config.AppID,
|
||||
"timestamp": timestamp,
|
||||
"sign": sign,
|
||||
}
|
||||
requestURL := s.config.BaseURL + pathAuthLogin
|
||||
|
||||
bodyBytes, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logError(transactionID, apiKeyLogin, "", err, map[string]interface{}{
|
||||
"account": s.config.Account,
|
||||
"appId": s.config.AppID,
|
||||
})
|
||||
return "", err
|
||||
}
|
||||
bodyStr := string(bodyBytes)
|
||||
|
||||
headers := map[string]string{"Content-Type": "application/json"}
|
||||
curlCmd := generateCurlCommand(http.MethodPost, requestURL, headers, bodyStr)
|
||||
s.logCurl(transactionID, apiKeyLogin, requestURL, curlCmd)
|
||||
|
||||
if s.logger != nil {
|
||||
s.logger.LogRequest("", transactionID, apiKeyLogin, requestURL)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL, bytes.NewBuffer(bodyBytes))
|
||||
if err != nil {
|
||||
err = errors.Join(ErrSystem, err)
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, "")
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
startTime := time.Now()
|
||||
respBody, statusCode, err := s.doHTTP(req)
|
||||
if err != nil {
|
||||
err = errors.Join(ErrDatasource, err)
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, "")
|
||||
return "", err
|
||||
}
|
||||
duration := time.Since(startTime)
|
||||
respStr := string(respBody)
|
||||
|
||||
if statusCode != http.StatusOK {
|
||||
err = errors.Join(ErrDatasource, fmt.Errorf("登录 HTTP状态码 %d, body=%s", statusCode, respStr))
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, respStr)
|
||||
return "", err
|
||||
}
|
||||
|
||||
var loginResp apiResponse
|
||||
if err := json.Unmarshal(respBody, &loginResp); err != nil {
|
||||
err = errors.Join(ErrSystem, fmt.Errorf("登录响应解析失败: %w, body=%s", err, respStr))
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, respStr)
|
||||
return "", err
|
||||
}
|
||||
|
||||
code := loginResp.code()
|
||||
if code != CodeSuccess {
|
||||
rxErr := NewRongxingError(code, loginResp.text())
|
||||
err = errors.Join(ErrDatasource, rxErr)
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, respStr)
|
||||
return "", err
|
||||
}
|
||||
|
||||
var token string
|
||||
if err := json.Unmarshal(loginResp.Data, &token); err != nil {
|
||||
err = errors.Join(ErrSystem, fmt.Errorf("登录响应 Token 解析失败: %w, body=%s", err, respStr))
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, respStr)
|
||||
return "", err
|
||||
}
|
||||
token = strings.TrimSpace(token)
|
||||
if token == "" {
|
||||
err = errors.Join(ErrSystem, fmt.Errorf("登录响应 Token 为空, body=%s", respStr))
|
||||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, respStr)
|
||||
return "", err
|
||||
}
|
||||
|
||||
s.logResponse(transactionID, apiKeyLogin, statusCode, duration, respStr)
|
||||
return token, nil
|
||||
}
|
||||
|
||||
func (s *RongxingService) doHTTP(req *http.Request) ([]byte, int, error) {
|
||||
client := &http.Client{Timeout: s.config.Timeout}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, resp.StatusCode, err
|
||||
}
|
||||
return body, resp.StatusCode, nil
|
||||
}
|
||||
|
||||
func (s *RongxingService) validateConfig() error {
|
||||
if s.config.BaseURL == "" {
|
||||
return errors.New("戎行 url 未配置")
|
||||
}
|
||||
if strings.TrimSpace(s.config.Account) == "" {
|
||||
return errors.New("戎行 account 未配置")
|
||||
}
|
||||
if strings.TrimSpace(s.config.Password) == "" {
|
||||
return errors.New("戎行 password 未配置")
|
||||
}
|
||||
if strings.TrimSpace(s.config.AppID) == "" {
|
||||
return errors.New("戎行 app_id 未配置")
|
||||
}
|
||||
if strings.TrimSpace(s.config.PrivateKey) == "" {
|
||||
return errors.New("戎行 private_key 未配置")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func parseCode(raw json.RawMessage) string {
|
||||
if len(raw) == 0 {
|
||||
return ""
|
||||
}
|
||||
var s string
|
||||
if err := json.Unmarshal(raw, &s); err == nil {
|
||||
return strings.TrimSpace(s)
|
||||
}
|
||||
var n json.Number
|
||||
if err := json.Unmarshal(raw, &n); err == nil {
|
||||
return n.String()
|
||||
}
|
||||
return strings.Trim(string(raw), `"`)
|
||||
}
|
||||
|
||||
// extractBusinessPayload 提取对外返回的业务 data。
|
||||
// 若外层 data 内还嵌套 data（如 Info360），则取内层标签对象。
|
||||
func extractBusinessPayload(data json.RawMessage) []byte {
|
||||
if len(data) == 0 || string(data) == "null" {
|
||||
return []byte("{}")
|
||||
}
|
||||
|
||||
var wrap struct {
|
||||
Data json.RawMessage `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(data, &wrap); err == nil &&
|
||||
len(wrap.Data) > 0 && string(wrap.Data) != "null" {
|
||||
return wrap.Data
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
func (s *RongxingService) logCurl(transactionID, apiKey, url, curlCmd string) {
|
||||
if s.logger == nil {
|
||||
return
|
||||
}
|
||||
s.logger.LogInfo("rongxing curl",
|
||||
zap.String("transaction_id", transactionID),
|
||||
zap.String("api_code", apiKey),
|
||||
zap.String("url", url),
|
||||
zap.String("curl", curlCmd),
|
||||
)
|
||||
}
|
||||
|
||||
func (s *RongxingService) logResponse(transactionID, apiKey string, statusCode int, duration time.Duration, responseBody string) {
|
||||
if s.logger == nil {
|
||||
return
|
||||
}
|
||||
s.logger.LogResponseWithBody("", transactionID, apiKey, statusCode, duration, responseBody)
|
||||
}
|
||||
|
||||
func (s *RongxingService) logError(transactionID, apiKey, requestID string, err error, payload interface{}) {
|
||||
if s.logger == nil {
|
||||
return
|
||||
}
|
||||
s.logger.LogError(requestID, transactionID, apiKey, err, payload)
|
||||
}
|
||||
|
||||
func (s *RongxingService) logErrorWithCurl(transactionID, apiKey string, err error, payload interface{}, curlCmd, respBody string) {
|
||||
if s.logger == nil {
|
||||
return
|
||||
}
|
||||
s.logger.LogErrorWithFields("rongxing API错误",
|
||||
zap.String("transaction_id", transactionID),
|
||||
zap.String("api_code", apiKey),
|
||||
zap.Error(err),
|
||||
zap.Any("params", payload),
|
||||
zap.String("curl", curlCmd),
|
||||
zap.String("response_body", respBody),
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user