500 lines
15 KiB
Go
500 lines
15 KiB
Go
package rongxing
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
"net/url"
|
||
"os"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"hyapi-server/internal/shared/external_logger"
|
||
|
||
"go.uber.org/zap"
|
||
"golang.org/x/net/proxy"
|
||
)
|
||
|
||
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
|
||
Proxy string
|
||
}
|
||
|
||
// RongxingService 戎行数据源服务
|
||
type RongxingService struct {
|
||
config serviceConfig
|
||
logger *external_logger.ExternalServiceLogger
|
||
client *http.Client
|
||
|
||
// loginMu 避免并发登录打爆对方；不缓存 token，每次业务调用重新登录。
|
||
loginMu sync.Mutex
|
||
}
|
||
|
||
// 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, error) {
|
||
if cfg.Timeout <= 0 {
|
||
cfg.Timeout = defaultRequestTimeout
|
||
}
|
||
cfg.BaseURL = strings.TrimRight(strings.TrimSpace(cfg.BaseURL), "/")
|
||
client, err := newHTTPClient(cfg.Timeout, cfg.Proxy)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &RongxingService{config: cfg, logger: logger, client: client}, nil
|
||
}
|
||
|
||
// newHTTPClient 构建 HTTP 客户端；proxyURL 支持 socks5/socks5h/http/https，空则直连。
|
||
func newHTTPClient(timeout time.Duration, proxyURL string) (*http.Client, error) {
|
||
client := &http.Client{Timeout: timeout}
|
||
proxyURL = strings.TrimSpace(proxyURL)
|
||
if proxyURL == "" {
|
||
return client, nil
|
||
}
|
||
|
||
u, err := url.Parse(proxyURL)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("解析 proxy 失败: %w", err)
|
||
}
|
||
|
||
switch strings.ToLower(u.Scheme) {
|
||
case "socks5", "socks5h":
|
||
dialer, err := proxy.FromURL(u, proxy.Direct)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("创建 SOCKS 代理失败: %w", err)
|
||
}
|
||
transport := &http.Transport{}
|
||
if cd, ok := dialer.(proxy.ContextDialer); ok {
|
||
transport.DialContext = cd.DialContext
|
||
} else {
|
||
transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||
return dialer.Dial(network, addr)
|
||
}
|
||
}
|
||
client.Transport = transport
|
||
case "http", "https":
|
||
client.Transport = &http.Transport{Proxy: http.ProxyURL(u)}
|
||
default:
|
||
return nil, fmt.Errorf("不支持的 proxy scheme: %s（支持 socks5/socks5h/http/https）", u.Scheme)
|
||
}
|
||
return client, nil
|
||
}
|
||
|
||
// GetConfig 获取运行时配置
|
||
func (s *RongxingService) GetConfig() serviceConfig {
|
||
return s.config
|
||
}
|
||
|
||
// CallAPI 通用业务接口调用。
|
||
// apiPath 为相对路径（如 /third/loan/info360），reqData 为已组装好的请求体。
|
||
// 每次调用重新登录取 Token（不本地缓存）；若业务码/HTTP 仍为 401/403，再登录重试一次。
|
||
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++ {
|
||
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.config.Proxy)
|
||
|
||
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
|
||
}
|
||
|
||
respStr := string(respBody)
|
||
|
||
if statusCode == http.StatusUnauthorized || statusCode == http.StatusForbidden {
|
||
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()
|
||
// 对方常以 HTTP 200 + body.code=401 表示 token 无效（非 HTTP 401）
|
||
if isTokenInvalidCode(code) && attempt == 0 {
|
||
continue
|
||
}
|
||
|
||
payload := extractBusinessPayload(resp.Data)
|
||
|
||
// 扣费只看 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("请求失败"))
|
||
}
|
||
|
||
// getToken 每次重新登录，不缓存 token。
|
||
func (s *RongxingService) getToken(ctx context.Context, transactionID string) (string, error) {
|
||
s.loginMu.Lock()
|
||
defer s.loginMu.Unlock()
|
||
return s.login(ctx, transactionID)
|
||
}
|
||
|
||
func isTokenInvalidCode(code string) bool {
|
||
switch strings.TrimSpace(code) {
|
||
case "401", "403":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
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.config.Proxy)
|
||
|
||
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")
|
||
|
||
respBody, statusCode, err := s.doHTTP(req)
|
||
if err != nil {
|
||
err = errors.Join(ErrDatasource, err)
|
||
s.logErrorWithCurl(transactionID, apiKeyLogin, err, nil, curlCmd, "")
|
||
return "", err
|
||
}
|
||
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
|
||
}
|
||
|
||
return token, nil
|
||
}
|
||
|
||
func (s *RongxingService) doHTTP(req *http.Request) ([]byte, int, error) {
|
||
resp, err := s.client.Do(req)
|
||
if err != nil {
|
||
return nil, 0, s.wrapTransportError(req.URL.String(), err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
body, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
return nil, resp.StatusCode, err
|
||
}
|
||
return body, resp.StatusCode, nil
|
||
}
|
||
|
||
// wrapTransportError 把超时/拒连/代理失败等包装成可读诊断，便于区分「联不通」原因。
|
||
func (s *RongxingService) wrapTransportError(targetURL string, err error) error {
|
||
diag := classifyTransportError(err)
|
||
proxyMode := strings.TrimSpace(s.config.Proxy)
|
||
if proxyMode == "" {
|
||
proxyMode = "(直连，未配置 proxy)"
|
||
}
|
||
return fmt.Errorf(
|
||
"戎行 HTTP 失败 target=%s proxy=%s diagnosis=%s cause=%w",
|
||
targetURL, proxyMode, diag, err,
|
||
)
|
||
}
|
||
|
||
func classifyTransportError(err error) string {
|
||
if err == nil {
|
||
return "unknown"
|
||
}
|
||
msg := err.Error()
|
||
|
||
switch {
|
||
case os.IsTimeout(err) || errors.Is(err, context.DeadlineExceeded) ||
|
||
strings.Contains(msg, "Client.Timeout") || strings.Contains(msg, "deadline exceeded"):
|
||
return "请求超时(未在 timeout 内收到响应头；可能：目标 192.168.3.43:7007 不可达、VPN/SOCKS 未转发、或服务无响应)"
|
||
case strings.Contains(msg, "connection refused"):
|
||
return "连接被拒绝(端口未监听或代理/目标拒绝)"
|
||
case strings.Contains(msg, "no such host") || strings.Contains(msg, "lookup"):
|
||
return "DNS/主机名解析失败(检查 rongxing-vpn 服务名或目标域名)"
|
||
case strings.Contains(msg, "network is unreachable") || strings.Contains(msg, "no route to host"):
|
||
return "网络不可达(无路由；直连内网 IP 时常见于未走 VPN/代理)"
|
||
case strings.Contains(msg, "i/o timeout") || strings.Contains(msg, "TLS handshake timeout"):
|
||
return "传输层超时(链路通但握手/读写超时)"
|
||
case strings.Contains(msg, "proxy") || strings.Contains(msg, "socks"):
|
||
return "代理链路异常(检查 socks5://rongxing-vpn:1080 与 VPN 容器)"
|
||
}
|
||
|
||
var netErr net.Error
|
||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||
return "网络超时"
|
||
}
|
||
// 完整原始错误在同一条日志的 error/cause 字段中，不在别的文件
|
||
return "其他网络错误(见本条日志 error 全文)"
|
||
}
|
||
|
||
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) 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
|
||
}
|
||
proxyMode := strings.TrimSpace(s.config.Proxy)
|
||
if proxyMode == "" {
|
||
proxyMode = "(直连，未配置 proxy)"
|
||
}
|
||
s.logger.LogErrorWithFields("rongxing API错误",
|
||
zap.String("transaction_id", transactionID),
|
||
zap.String("api_code", apiKey),
|
||
zap.String("base_url", s.config.BaseURL),
|
||
zap.String("proxy", proxyMode),
|
||
zap.String("diagnosis", extractDiagnosis(err)),
|
||
zap.Error(err),
|
||
zap.Any("params", payload),
|
||
zap.String("curl", curlCmd),
|
||
zap.String("response_body", respBody),
|
||
)
|
||
}
|
||
|
||
func extractDiagnosis(err error) string {
|
||
if err == nil {
|
||
return ""
|
||
}
|
||
const marker = "diagnosis="
|
||
msg := err.Error()
|
||
if i := strings.Index(msg, marker); i >= 0 {
|
||
rest := msg[i+len(marker):]
|
||
if j := strings.Index(rest, " cause="); j >= 0 {
|
||
return rest[:j]
|
||
}
|
||
return rest
|
||
}
|
||
return classifyTransportError(err)
|
||
}
|