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), ) }