// 115 开放平台 HTTP 客户端(集成全局三级令牌桶限流、熔断保护与防 405 重定向策略)。 package cloud115 import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "strings" "sync" "time" ) // OpenClient 是 115 开放平台客户端。 type OpenClient struct { AppID string HTTP *http.Client AccessToken string RefreshTokenStr string executor *QueueExecutor // OnTokenRefreshed 在 access_token 刷新成功后回调(参数为新令牌对), // 供上层持久化新令牌使用;nil 安全,且在 tokenMu 释放后调用以避免死锁。 OnTokenRefreshed func(accessToken, refreshToken string) // tokenMu 保护 AccessToken / RefreshTokenStr 的并发读写:业务请求中途 // access_token 失效时自动刷新重试,多 goroutine(同步列表 + 下载队列) // 并发下只允许一次刷新进行。 tokenMu sync.RWMutex } // default115HTTPClient 创建带有防 405 重定向保护的 http.Client。 func default115HTTPClient() *http.Client { return &http.Client{ Timeout: 60 * time.Second, // 防止 Go 标准库在遇到 301/302/307 重定向时将 POST 降级为 GET 导致 115 报 405 Method Not Allowed CheckRedirect: func(req *http.Request, via []*http.Request) error { if len(via) >= 5 { return errors.New("stopped after 5 redirects") } // 如果原请求是 POST,重定向时不允许静默转为 GET(直接终止自动重定向,由业务层处理响应) if len(via) > 0 && via[len(via)-1].Method == http.MethodPost { return http.ErrUseLastResponse } return nil }, } } // NewOpenClient 构造客户端。 func NewOpenClient(appID, accessToken, refreshToken string) *OpenClient { return &OpenClient{ AppID: appID, HTTP: default115HTTPClient(), AccessToken: accessToken, RefreshTokenStr: refreshToken, executor: GetGlobalExecutor(), } } // SetAuthToken 更新认证令牌(并发安全)。 func (c *OpenClient) SetAuthToken(accessToken, refreshToken string) { c.tokenMu.Lock() c.setAuthTokenLocked(accessToken, refreshToken) c.tokenMu.Unlock() } // setAuthTokenLocked 无锁更新令牌,调用方必须已持有 tokenMu 写锁 // (tryRefreshTokenLocked 等已持锁流程内部使用,避免重入死锁)。 func (c *OpenClient) setAuthTokenLocked(accessToken, refreshToken string) { c.AccessToken = accessToken c.RefreshTokenStr = refreshToken } // currentAccessToken 返回当前 access_token(并发安全)。 func (c *OpenClient) currentAccessToken() string { c.tokenMu.RLock() defer c.tokenMu.RUnlock() return c.AccessToken } // currentRefreshToken 返回当前 refresh_token(并发安全)。 func (c *OpenClient) currentRefreshToken() string { c.tokenMu.RLock() defer c.tokenMu.RUnlock() return c.RefreshTokenStr } // CurrentAccessToken 返回当前 access_token 快照(并发安全), // 供上层在无锁环境下安全读取(如 Ping 时探测令牌是否存在)。 func (c *OpenClient) CurrentAccessToken() string { return c.currentAccessToken() } // RespState 兼容 115 不同端点返回的 state 类型(proapi 返回布尔、passport 返回数字)。 type RespState bool func (s *RespState) UnmarshalJSON(data []byte) error { switch string(data) { case "true", "1": *s = true return nil case "false", "0", "null", "": *s = false return nil } var n float64 if err := json.Unmarshal(data, &n); err != nil { return fmt.Errorf("115: 无法解析 state 字段 %s", string(data)) } *s = n != 0 return nil } // RespBase 是 115 开放平台统一响应外壳。 type RespBase struct { State RespState `json:"state"` Code int `json:"code"` Errno int `json:"errno"` Message string `json:"message"` Error string `json:"error"` Count int64 `json:"count"` Data json.RawMessage `json:"data"` Raw json.RawMessage `json:"-"` // 原始响应体(外层附加字段用) } // doJSON 执行 HTTP 请求并解析为统一响应;带 AccessToken(access=true 时)。 func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map[string]string, access bool, retries int, uas ...string) (*RespBase, error) { executor := c.executor if executor == nil { executor = GetGlobalExecutor() } ua := "" if len(uas) > 0 { ua = uas[0] } var lastErr error for attempt := 0; attempt <= retries; attempt++ { // 1. 获取全局三级令牌桶(QPS/QPM/QPH)令牌,若在熔断状态则阻塞等待冷却 if err := executor.Acquire(ctx); err != nil { return nil, err } req, err := c.buildRequestWithUA(ctx, method, rawURL, form, access, ua) if err != nil { return nil, err } resp, err := c.HTTP.Do(req) if err != nil { lastErr = err if attempt < retries { time.Sleep(time.Duration(attempt+1) * 1 * time.Second) continue } return nil, err } body, readErr := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) _ = resp.Body.Close() if readErr != nil { lastErr = readErr continue } // 检查 HTTP 状态码:特别处理 405/406/429 等限流和异常阻断 if resp.StatusCode < 200 || resp.StatusCode >= 300 { if resp.StatusCode == http.StatusMethodNotAllowed || resp.StatusCode == http.StatusNotAcceptable || resp.StatusCode == http.StatusTooManyRequests { // 触发全局熔断冷却,防止短时间连续重试导致 IP 被完全拉黑 executor.MarkThrottled() lastErr = fmt.Errorf("115 接口触发频控/安全拦截(HTTP %d):%s", resp.StatusCode, strings.TrimSpace(string(body))) if attempt < retries { // 阻塞等待 115 限流冷却解封后自动重试当前请求 if waitErr := executor.throttleManager.WaitThrottleRecovery(ctx); waitErr != nil { return nil, waitErr } continue } return nil, lastErr } lastErr = fmt.Errorf("115 接口返回 HTTP %d:%s", resp.StatusCode, strings.TrimSpace(string(body))) if attempt < retries { time.Sleep(time.Duration(attempt+1) * 1 * time.Second) continue } return nil, lastErr } var base RespBase if err := json.Unmarshal(body, &base); err != nil { return nil, fmt.Errorf("115 接口响应解析失败:%w", err) } base.Raw = body if base.State { return &base, nil } // 业务返回限流错误码(406 / 770004):标记全局熔断,等待冷却后重试 if IsThrottleCode(base.Code) { executor.MarkThrottled() lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115 访问频率达到上限,已进入冷却") if attempt < retries { // 阻塞等待限流冷却解封后自动重试当前请求 if waitErr := executor.throttleManager.WaitThrottleRecovery(ctx); waitErr != nil { return nil, waitErr } continue } return &base, lastErr } // Token 失效(access_token 过期,如长时间同步中途过期):自动用 // refresh_token 刷新后重试一次。刷新失败或重试后仍失败才返回, // 避免长时间同步因 token 过期而整体失败。 if isTokenCode(base.Code) { if access && c.tryRefreshTokenLocked(ctx) { continue } if access { // 刷新失败(或已刷新仍失败)时返回明确错误 lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115: access_token 校验失败且刷新未成功") } else { // 未携带令牌的请求(登录/刷新流程)命中 token 类错误码: // 必须返回显式 error,避免调用方把 (resp, nil) 当作成功处理 msg := base.Message if msg == "" { msg = base.Error } lastErr = fmt.Errorf("115: 认证失败(code=%d): %s", base.Code, msg) } return &base, lastErr } lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115 接口调用失败") if attempt < retries { time.Sleep(time.Duration(attempt+1) * 1 * time.Second) continue } return &base, lastErr } return nil, lastErr } func (c *OpenClient) buildRequest(ctx context.Context, method, rawURL string, form map[string]string, access bool) (*http.Request, error) { return c.buildRequestWithUA(ctx, method, rawURL, form, access, "") } func (c *OpenClient) buildRequestWithUA(ctx context.Context, method, rawURL string, form map[string]string, access bool, ua string) (*http.Request, error) { method = strings.ToUpper(method) var body io.Reader if method == http.MethodPost && len(form) > 0 { values := url.Values{} for k, v := range form { values.Set(k, v) } body = bytes.NewBufferString(values.Encode()) } else if method == http.MethodGet && len(form) > 0 { u, err := url.Parse(rawURL) if err != nil { return nil, err } query := u.Query() for k, v := range form { query.Set(k, v) } u.RawQuery = query.Encode() rawURL = u.String() } req, err := http.NewRequestWithContext(ctx, method, rawURL, body) if err != nil { return nil, err } targetUA := DefaultUA if strings.TrimSpace(ua) != "" { targetUA = strings.TrimSpace(ua) } req.Header.Set("User-Agent", targetUA) if method == http.MethodPost && len(form) > 0 { req.Header.Set("Content-Type", "application/x-www-form-urlencoded") } if access { // RLock 读取令牌,避免与刷新流程的写入产生数据竞争 if accessToken := c.currentAccessToken(); accessToken != "" { req.Header.Set("Authorization", "Bearer "+accessToken) } } return req, nil } // doAuthJSON 带 AccessToken 的业务请求。 func (c *OpenClient) doAuthJSON(ctx context.Context, method, rawURL string, form map[string]string, retries int) (*RespBase, error) { return c.doJSON(ctx, method, rawURL, form, true, retries) } // doAuthJSONWithUA 带自定义 User-Agent 的业务请求(换取直链等防盗链接口使用)。 func (c *OpenClient) doAuthJSONWithUA(ctx context.Context, method, rawURL string, form map[string]string, retries int, ua string) (*RespBase, error) { return c.doJSON(ctx, method, rawURL, form, true, retries, ua) } // tryRefreshTokenLocked 并发安全地刷新 access_token;成功返回 true(调用方 // 应使用内存中的新 token 重试原请求)。 // // 拿到写锁后在锁内读取 oldAccess,与持锁期间的当前值对比:若已被其他 // goroutine 刷新过则直接复用新 token,避免并发请求连环轮转消耗 115 的 // 一次性 refresh_token。全程持写锁读写 token 字段,无 TOCTOU 窗口。 // // 对"refresh_token 本身已失效/被吊销"(IsRefreshTokenDead,如 40140114/116/119/120) // 这类不可恢复的错误直接放弃并清空内存 token(提示需重新授权)。 // 对其它失败(网络瞬时抖动、刷新接口可重试错误码等)做指数退避重试几次再放弃, // 避免同步长任务中途 token 到期时恰好撞上一个短暂的刷新失败就整体失败。 func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context) bool { c.tokenMu.Lock() // 在已持有写锁内读取当前 token 作为"刷新前快照",消除双重加锁窗口: // 若在拿锁期间已有其他 goroutine 完成刷新,refreshTokenWhileLocked // 内的 c.AccessToken != oldAccess 判断会立即命中并返回复用。 oldAccess := c.AccessToken newToken, ok := c.refreshTokenWhileLocked(ctx, oldAccess) c.tokenMu.Unlock() // 回调必须在 tokenMu 释放后调用,避免上层在回调内访问客户端时死锁 if ok && newToken != nil && c.OnTokenRefreshed != nil { c.OnTokenRefreshed(newToken.AccessToken, newToken.RefreshToken) } return ok } // refreshTokenWhileLocked 在已持有 tokenMu 写锁的前提下执行刷新。 // 返回 (新令牌, 是否成功);命中"他人已刷新"捷径时新令牌为 nil。 func (c *OpenClient) refreshTokenWhileLocked(ctx context.Context, oldAccess string) (*TokenData, bool) { if c.AccessToken != oldAccess { // 其他 goroutine 刚刷新过:直接复用内存中的新 token 重试原请求 return nil, true } refreshToken := c.RefreshTokenStr for attempt := 0; attempt < refreshAttempts; attempt++ { token, err := c.doRefreshToken(refreshToken) if err == nil { c.setAuthTokenLocked(token.AccessToken, token.RefreshToken) return token, true } if IsRefreshTokenDead(err) { c.setAuthTokenLocked("", "") return nil, false } // 可恢复失败:退避后重试。ctx 取消时立即放弃。 if attempt < refreshAttempts-1 { select { case <-ctx.Done(): return nil, false case <-time.After(refreshBackoff(attempt)): } } } return nil, false } // refreshAttempts 是刷新 access_token 失败时的最大尝试次数(含首次)。 const refreshAttempts = 3 // refreshBackoff 返回第 attempt 次(从 0 计)刷新失败后的退避时长(指数退避)。 func refreshBackoff(attempt int) time.Duration { return time.Duration(200*(1<