Files
dygou_go/internal/client/client.go
T
2026-06-24 19:11:02 +08:00

586 lines
18 KiB
Go

package client
import (
"encoding/json"
"errors"
"fmt"
"io"
"mime/multipart"
"net/http"
"reflect"
"strconv"
"strings"
"time"
"douyu_go/internal/constants"
"douyu_go/internal/logger"
"douyu_go/internal/trace"
"douyu_go/internal/types"
"douyu_go/internal/utils"
)
// DouyuClient 斗鱼接口客户端。
type DouyuClient struct {
HTTPClient *http.Client
TimeAPIURL string
Trace *trace.Recorder
}
// New 创建斗鱼客户端。
func New(timeAPIURL string, tr *trace.Recorder) *DouyuClient {
return &DouyuClient{
TimeAPIURL: timeAPIURL,
Trace: tr,
HTTPClient: &http.Client{
Timeout: 8 * time.Second,
Transport: &http.Transport{
MaxConnsPerHost: 120,
MaxIdleConns: 120,
MaxIdleConnsPerHost: 80,
IdleConnTimeout: 30 * time.Second,
},
},
}
}
// ServerTime 从商品详情接口读取 data.preExchange.currentTime。
func (c *DouyuClient) ServerTime() (time.Time, error) {
sample, err := c.ServerTimeSample()
return sample.EstimatedServerNow(), err
}
// ServerTimeSample 从斗鱼秒级 currentTime 估算毫秒级服务端时间。
func (c *DouyuClient) ServerTimeSample() (types.TimeSample, error) {
var lastErr error
for i := 0; i < 3; i++ {
sample, err := c.fetchServerTimeOnce()
if err == nil {
sample.Samples = 1
sample.Source = "single"
return sample, nil
}
lastErr = err
time.Sleep(100 * time.Millisecond)
}
now := time.Now()
return types.TimeSample{
ServerTime: now.UTC(),
LocalTime: now,
Offset: 0,
Source: "localFallback",
Fallback: true,
}, fmt.Errorf("服务端时间获取失败,已降级本地时间: %w", lastErr)
}
// CalibratedServerTimeSample 通过等待 currentTime 跳秒估算本地与斗鱼秒边界的偏移。
func (c *DouyuClient) CalibratedServerTimeSample(maxWait time.Duration) (types.TimeSample, error) {
first, err := c.ServerTimeSample()
if err != nil || maxWait <= 0 {
return first, err
}
deadline := time.Now().Add(maxWait)
prev := first
for time.Now().Before(deadline) {
time.Sleep(40 * time.Millisecond)
current, err := c.fetchServerTimeOnce()
if err != nil {
continue
}
if current.RawCurrentTime <= prev.RawCurrentTime {
prev = current
continue
}
sampleGap := current.LocalTime.Sub(prev.LocalTime)
localBoundary := prev.LocalTime.Add(sampleGap / 2)
serverBoundary := time.Unix(current.RawCurrentTime, 0).UTC()
offset := serverBoundary.Sub(localBoundary.UTC())
current.Offset = offset
current.ServerTime = current.EstimatedServerNow()
current.LocalTime = time.Now()
current.RTT += first.RTT
current.ErrorBound = sampleGap / 2
current.Samples = 2
current.Source = "boundary"
return current, nil
}
first.ServerTime = first.EstimatedServerNow()
return first, nil
}
// BestServerTimeCalibration 多次校准,选择误差范围最小、RTT更低的一次。
func (c *DouyuClient) BestServerTimeCalibration(rounds int, maxWait time.Duration) (types.TimeCalibration, error) {
if rounds <= 0 {
rounds = 1
}
var best types.TimeSample
var lastErr error
for i := 0; i < rounds; i++ {
sample, err := c.CalibratedServerTimeSample(maxWait)
if err != nil {
lastErr = err
}
if sample.LocalTime.IsZero() {
continue
}
sample.Samples = max(sample.Samples, 1)
if best.LocalTime.IsZero() || betterTimeSample(sample, best) {
best = sample
}
}
if best.LocalTime.IsZero() {
now := time.Now()
return types.TimeCalibration{
CreatedAt: now,
Source: "localFallback",
Fallback: true,
}, fmt.Errorf("斗鱼时间校准失败: %w", lastErr)
}
return types.TimeCalibration{
Offset: best.Offset,
ErrorBound: best.ErrorBound,
RTT: best.RTT,
Samples: rounds,
CreatedAt: time.Now(),
Source: best.Source,
Fallback: best.Fallback,
}, lastErr
}
func betterTimeSample(candidate, current types.TimeSample) bool {
if candidate.ErrorBound != 0 && current.ErrorBound == 0 {
return true
}
if candidate.ErrorBound == 0 && current.ErrorBound != 0 {
return false
}
if candidate.ErrorBound != current.ErrorBound {
return candidate.ErrorBound < current.ErrorBound
}
return candidate.RTT < current.RTT
}
func (c *DouyuClient) fetchServerTimeOnce() (types.TimeSample, error) {
url := utils.AppendCacheBuster(c.TimeAPIURL)
var result struct {
types.APIResponse
Data struct {
PreExchange struct {
CurrentTime string `json:"currentTime"`
} `json:"preExchange"`
} `json:"data"`
}
start := time.Now()
if err := c.getJSON(url, "", "服务时间", &result); err != nil {
return types.TimeSample{}, err
}
end := time.Now()
ts, err := strconv.ParseInt(result.Data.PreExchange.CurrentTime, 10, 64)
if err != nil || ts <= 0 {
return types.TimeSample{}, fmt.Errorf("currentTime 无效: %q", result.Data.PreExchange.CurrentTime)
}
rtt := end.Sub(start)
estimatedLocal := start.Add(rtt / 2)
serverTime := time.Unix(ts, 0).Add(time.Duration(estimatedLocal.Nanosecond())).UTC()
return types.TimeSample{
ServerTime: serverTime,
LocalTime: estimatedLocal,
RTT: rtt,
Offset: serverTime.Sub(estimatedLocal.UTC()),
RawCurrentTime: ts,
}, nil
}
// Products 拉取商城商品列表。
func (c *DouyuClient) Products() ([]types.Product, error) {
var result struct {
types.APIResponse
Data struct {
List []types.Product `json:"list"`
} `json:"data"`
}
if err := c.getJSON(constants.StoreHomeURL, "", "商品列表", &result); err != nil {
return nil, err
}
if result.Error != 0 {
return nil, fmt.Errorf("商品接口失败: %s", result.Msg)
}
return result.Data.List, nil
}
// Score 查询账号积分余额。
func (c *DouyuClient) Score(cookie string) (int, error) {
req, err := http.NewRequest("POST", constants.ScoreURL, strings.NewReader(`{}`))
if err != nil {
return 0, err
}
c.setJSONHeaders(req, cookie)
req.Header.Set("Referer", constants.BaseURL+"/")
var result struct {
types.APIResponse
Data struct {
Balance int `json:"balance"`
} `json:"data"`
}
if _, err := c.doJSON(req, &result, "积分余额"); err != nil {
return 0, err
}
if result.Error != 0 {
return 0, fmt.Errorf("积分接口失败: %s", result.Msg)
}
return result.Data.Balance, nil
}
// GenerateCSRF 调用 generateCsrf,并从 Set-Cookie 中取 cvl_csrf_token。
func (c *DouyuClient) GenerateCSRF(cookie string) (string, error) {
req, err := http.NewRequest("POST", constants.CSRFURL, nil)
if err != nil {
return "", err
}
c.setBrowserHeaders(req, cookie)
start := time.Now()
reqBody := c.Trace.RequestBody(req)
resp, err := c.HTTPClient.Do(req)
if err != nil {
c.Trace.Record("生成CSRF", req, 0, time.Since(start), reqBody, nil, err.Error(), 0, "")
logNetworkError("生成CSRF", req, time.Since(start), err)
return "", err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
c.Trace.RecordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, nil, err.Error(), 0, "")
logNetworkError("生成CSRF", req, time.Since(start), err)
return "", err
}
if resp.StatusCode != http.StatusOK {
c.Trace.RecordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "")
logNetworkHTTPError("生成CSRF", req, resp.StatusCode, time.Since(start), body)
return "", fmt.Errorf("CSRF HTTP %d: %s", resp.StatusCode, utils.TrimForLog(string(body), 120))
}
var api types.APIResponse
if err := json.Unmarshal(body, &api); err != nil {
c.Trace.RecordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, body, err.Error(), 0, "")
logNetworkParseError("生成CSRF", req, time.Since(start), body, err)
return "", fmt.Errorf("CSRF 响应解析失败: %w", err)
}
traceErr := ""
if api.Error != 0 {
traceErr = api.Msg
c.Trace.RecordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, body, traceErr, api.Error, api.Msg)
logNetworkAPIResult("生成CSRF", req, time.Since(start), api.Error, api.Msg)
return "", fmt.Errorf("CSRF 接口失败: %s", api.Msg)
}
for _, ck := range resp.Cookies() {
if ck.Name == "cvl_csrf_token" && ck.Value != "" {
c.Trace.RecordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", api.Error, api.Msg)
logNetworkSuccess("生成CSRF", req, time.Since(start), 0, "")
return ck.Value, nil
}
}
c.Trace.RecordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, body, "响应没有 cvl_csrf_token", api.Error, api.Msg)
logger.LogEvent("网络", "warn", "接口返回异常: 生成CSRF %s %s 耗时=%s msg=响应没有 cvl_csrf_token",
req.Method, requestPath(req), utils.RoundDuration(time.Since(start)))
return "", errors.New("CSRF 响应没有 cvl_csrf_token")
}
func csrfFromResponse(resp *http.Response) string {
if resp == nil {
return ""
}
for _, ck := range resp.Cookies() {
if ck.Name == "cvl_csrf_token" && ck.Value != "" {
return ck.Value
}
}
return ""
}
// BatchExchangeLimit 请求兑换限制接口。
func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID string, csrf *string) (int, error) {
currentCSRF := ""
if csrf != nil {
currentCSRF = *csrf
}
var result struct {
types.APIResponse
Data struct {
Limit string `json:"limit"`
} `json:"data"`
}
if err := c.postMultipart(constants.BatchLimitURL, cookie, csrf, "批量限制", map[string]string{
"manualID": constants.ManualID,
"commodityID": commodityID,
"rid": constants.RoomID,
"csrfToken": currentCSRF,
}, &result); err != nil {
return 0, err
}
if result.Error != 0 {
return 0, fmt.Errorf("批量限制接口失败: %s", result.Msg)
}
limit, err := strconv.Atoi(result.Data.Limit)
if err != nil {
return 0, fmt.Errorf("批量限制值无效: %q", result.Data.Limit)
}
return limit, nil
}
// PreExchange 按抓包流程执行预兑换检查。
func (c *DouyuClient) PreExchange(cookie, commodityID string, csrf *string) error {
currentCSRF := ""
if csrf != nil {
currentCSRF = *csrf
}
var result types.APIResponse
if err := c.postMultipart(constants.PreExchangeURL, cookie, csrf, "预兑换", map[string]string{
"manualID": constants.ManualID,
"rid": constants.RoomID,
"commodityID": commodityID,
"num": "1",
"csrfToken": currentCSRF,
}, &result); err != nil {
return err
}
if result.Error != 0 {
return fmt.Errorf("预兑换失败: %s", result.Msg)
}
return nil
}
// Exchange 正式兑换一个商品。
func (c *DouyuClient) Exchange(cookie, commodityID string, csrf *string) (string, error) {
currentCSRF := ""
if csrf != nil {
currentCSRF = *csrf
}
var result struct {
types.APIResponse
Data struct {
ExchangeID int `json:"exchangeId"`
ExchangeNum string `json:"exchangeNum"`
CommodityTyp int `json:"commodityType"`
} `json:"data"`
}
if err := c.postMultipart(constants.ExchangeURL, cookie, csrf, "正式兑换", map[string]string{
"manualID": constants.ManualID,
"commodityID": commodityID,
"num": "1",
"rid": constants.RoomID,
"csrfToken": currentCSRF,
}, &result); err != nil {
return "", err
}
if result.Error != 0 {
if strings.TrimSpace(result.Msg) == "" {
return "", errors.New("兑换失败,接口未返回原因")
}
return "", fmt.Errorf("兑换失败: %s", result.Msg)
}
return fmt.Sprintf("exchangeId=%d,num=%s", result.Data.ExchangeID, result.Data.ExchangeNum), nil
}
// RoleName 查询绑定角色名。
func (c *DouyuClient) RoleName(cookie string) (string, error) {
var result struct {
types.APIResponse
Data struct {
GameRole struct {
RoleName string `json:"roleName"`
} `json:"gameRole"`
} `json:"data"`
}
if err := c.getJSON(constants.BindInfoURL, cookie, "角色绑定", &result); err != nil {
return "", err
}
if result.Error != 0 {
return "", fmt.Errorf("角色接口失败: %s", result.Msg)
}
return result.Data.GameRole.RoleName, nil
}
func (c *DouyuClient) getJSON(url, cookie, label string, out any) error {
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return err
}
c.setBrowserHeaders(req, cookie)
_, err = c.doJSON(req, out, label)
return err
}
func (c *DouyuClient) postMultipart(url, cookie string, csrf *string, label string, fields map[string]string, out any) error {
var body strings.Builder
writer := multipart.NewWriter(&body)
for key, value := range fields {
if err := writer.WriteField(key, value); err != nil {
return err
}
}
if err := writer.Close(); err != nil {
return err
}
req, err := http.NewRequest("POST", url, strings.NewReader(body.String()))
if err != nil {
return err
}
currentCSRF := ""
if csrf != nil {
currentCSRF = *csrf
}
c.setBrowserHeaders(req, utils.MergeCookie(cookie, "cvl_csrf_token", currentCSRF))
req.Header.Set("Content-Type", writer.FormDataContentType())
resp, err := c.doJSON(req, out, label)
if resp != nil {
if nextCSRF := csrfFromResponse(resp); nextCSRF != "" && csrf != nil {
if nextCSRF != *csrf {
logger.LogEvent("网络", "info", "响应刷新CSRF已同步: %s %s %s", label, req.Method, requestPath(req))
}
*csrf = nextCSRF
}
}
return err
}
func (c *DouyuClient) doJSON(req *http.Request, out any, label string) (*http.Response, error) {
start := time.Now()
reqBody := c.Trace.RequestBody(req)
resp, err := c.HTTPClient.Do(req)
if err != nil {
c.Trace.Record(label, req, 0, time.Since(start), reqBody, nil, err.Error(), 0, "")
logNetworkError(label, req, time.Since(start), err)
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
c.Trace.RecordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, nil, err.Error(), 0, "")
logNetworkError(label, req, time.Since(start), err)
return resp, err
}
if resp.StatusCode != http.StatusOK {
c.Trace.RecordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "")
logNetworkHTTPError(label, req, resp.StatusCode, time.Since(start), body)
return resp, fmt.Errorf("HTTP %d: %s", resp.StatusCode, utils.TrimForLog(string(body), 160))
}
if err := json.Unmarshal(body, out); err != nil {
c.Trace.RecordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, err.Error(), 0, "")
logNetworkParseError(label, req, time.Since(start), body, err)
return resp, fmt.Errorf("JSON 解析失败: %w, body=%s", err, utils.TrimForLog(string(body), 160))
}
if apiError, apiMsg, ok := extractAPIStatus(out); ok && apiError != 0 {
c.Trace.RecordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, apiMsg, apiError, apiMsg)
logNetworkAPIResult(label, req, time.Since(start), apiError, apiMsg)
} else if shouldLogNetworkSuccess(label) {
c.Trace.RecordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "")
logNetworkSuccess(label, req, time.Since(start), 0, "")
} else {
c.Trace.RecordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "")
}
return resp, nil
}
func shouldLogNetworkSuccess(label string) bool {
return label != "服务时间"
}
func logNetworkSuccess(label string, req *http.Request, cost time.Duration, apiError int, apiMsg string) {
if !shouldLogNetworkSuccess(label) {
return
}
extra := ""
if apiMsg != "" || apiError != 0 {
extra = fmt.Sprintf(" apiError=%d msg=%s", apiError, utils.TrimForLog(apiMsg, 80))
}
logger.LogEvent("网络", "info", "请求完成: %s %s %s 耗时=%s%s", label, req.Method, requestPath(req), utils.RoundDuration(cost), extra)
}
func logNetworkAPIResult(label string, req *http.Request, cost time.Duration, apiError int, apiMsg string) {
logger.LogEvent("网络", "warn", "接口返回异常: %s %s %s 耗时=%s apiError=%d msg=%s",
label, req.Method, requestPath(req), utils.RoundDuration(cost), apiError, utils.TrimForLog(apiMsg, 120))
}
func logNetworkHTTPError(label string, req *http.Request, status int, cost time.Duration, body []byte) {
logger.LogEvent("网络", "error", "HTTP异常: %s %s %s 状态=%d 耗时=%s 响应=%s",
label, req.Method, requestPath(req), status, utils.RoundDuration(cost), utils.TrimForLog(string(body), 160))
}
func logNetworkParseError(label string, req *http.Request, cost time.Duration, body []byte, err error) {
logger.LogEvent("网络", "error", "响应解析失败: %s %s %s 耗时=%s 错误=%v 响应=%s",
label, req.Method, requestPath(req), utils.RoundDuration(cost), err, utils.TrimForLog(string(body), 160))
}
func logNetworkError(label string, req *http.Request, cost time.Duration, err error) {
logger.LogEvent("网络", "error", "请求失败: %s %s %s 耗时=%s 错误=%v",
label, req.Method, requestPath(req), utils.RoundDuration(cost), err)
}
func requestPath(req *http.Request) string {
if req == nil || req.URL == nil {
return "-"
}
return req.URL.Path
}
func extractAPIStatus(out any) (int, string, bool) {
if out == nil {
return 0, "", false
}
v := reflect.ValueOf(out)
if v.Kind() == reflect.Pointer {
if v.IsNil() {
return 0, "", false
}
v = v.Elem()
}
if !v.IsValid() {
return 0, "", false
}
if v.Type() == reflect.TypeOf(types.APIResponse{}) {
api := v.Interface().(types.APIResponse)
return api.Error, api.Msg, true
}
if v.Kind() != reflect.Struct {
return 0, "", false
}
if field := v.FieldByName("APIResponse"); field.IsValid() && field.CanInterface() {
if api, ok := field.Interface().(types.APIResponse); ok {
return api.Error, api.Msg, true
}
}
errorField := v.FieldByName("Error")
msgField := v.FieldByName("Msg")
if errorField.IsValid() && msgField.IsValid() && errorField.CanInt() && msgField.Kind() == reflect.String {
return int(errorField.Int()), msgField.String(), true
}
return 0, "", false
}
func (c *DouyuClient) setBrowserHeaders(req *http.Request, cookie string) {
req.Header.Set("Accept", "application/json, text/plain, */*")
req.Header.Set("Accept-Language", "zh-CN,zh;q=0.9,en-US;q=0.6,en;q=0.5")
req.Header.Set("User-Agent", constants.UserAgent)
req.Header.Set("Origin", constants.BaseURL)
req.Header.Set("Referer", constants.ShopReferer)
req.Header.Set("Sec-Fetch-Dest", "empty")
req.Header.Set("Sec-Fetch-Mode", "cors")
req.Header.Set("Sec-Fetch-Site", "same-origin")
if strings.TrimSpace(cookie) != "" {
req.Header.Set("Cookie", cookie)
}
}
func (c *DouyuClient) setJSONHeaders(req *http.Request, cookie string) {
c.setBrowserHeaders(req, cookie)
req.Header.Set("Content-Type", "application/json")
}