From 86374fdd4d12af48b1dac46900f261c60af06b53 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sat, 20 Jun 2026 05:54:25 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=8A=A2=E8=B4=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- douyu_client.go | 98 ++++++++++++++++++++++++++++++++++--------------- handlers.go | 4 ++ seckill.go | 29 +++++++++++++-- trace.go | 76 ++++++++++++++++++++++++++++++++++++-- types.go | 3 ++ 5 files changed, 172 insertions(+), 38 deletions(-) diff --git a/douyu_client.go b/douyu_client.go index 8ff060c..6690783 100644 --- a/douyu_client.go +++ b/douyu_client.go @@ -216,7 +216,7 @@ func (c *DouyuClient) Score(cookie string) (int, error) { Balance int `json:"balance"` } `json:"data"` } - if err := c.doJSON(req, &result, "积分余额"); err != nil { + if _, err := c.doJSON(req, &result, "积分余额"); err != nil { return 0, err } if result.Error != 0 { @@ -245,44 +245,60 @@ func (c *DouyuClient) GenerateCSRF(cookie string) (string, error) { body, err := io.ReadAll(resp.Body) if err != nil { - c.trace.record("生成CSRF", req, resp.StatusCode, time.Since(start), reqBody, nil, err.Error(), 0, "") + 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.record("生成CSRF", req, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") + 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, trimForLog(string(body), 120)) } var api APIResponse if err := json.Unmarshal(body, &api); err != nil { - c.trace.record("生成CSRF", req, resp.StatusCode, time.Since(start), reqBody, body, err.Error(), 0, "") + 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.record("生成CSRF", req, resp.StatusCode, time.Since(start), reqBody, body, traceErr, api.Error, 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.record("生成CSRF", req, resp.StatusCode, time.Since(start), reqBody, body, "", api.Error, api.Msg) + 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.record("生成CSRF", req, resp.StatusCode, time.Since(start), reqBody, body, "响应没有 cvl_csrf_token", api.Error, api.Msg) + c.trace.recordResponse("生成CSRF", req, resp, resp.StatusCode, time.Since(start), reqBody, body, "响应没有 cvl_csrf_token", api.Error, api.Msg) logEvent("网络", "warn", "接口返回异常: 生成CSRF %s %s 耗时=%s msg=响应没有 cvl_csrf_token", req.Method, requestPath(req), 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, csrf string) (int, error) { +func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID string, csrf *string) (int, error) { + currentCSRF := "" + if csrf != nil { + currentCSRF = *csrf + } var result struct { APIResponse Data struct { @@ -293,7 +309,7 @@ func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID, csrf string) (int, "manualID": manualID, "commodityID": commodityID, "rid": roomID, - "csrfToken": csrf, + "csrfToken": currentCSRF, }, &result); err != nil { return 0, err } @@ -308,14 +324,18 @@ func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID, csrf string) (int, } // PreExchange 按抓包流程执行预兑换检查。 -func (c *DouyuClient) PreExchange(cookie, commodityID, csrf string) error { +func (c *DouyuClient) PreExchange(cookie, commodityID string, csrf *string) error { + currentCSRF := "" + if csrf != nil { + currentCSRF = *csrf + } var result APIResponse if err := c.postMultipart(preExchangeURL, cookie, csrf, "预兑换", map[string]string{ "manualID": manualID, "rid": roomID, "commodityID": commodityID, "num": "1", - "csrfToken": csrf, + "csrfToken": currentCSRF, }, &result); err != nil { return err } @@ -326,7 +346,11 @@ func (c *DouyuClient) PreExchange(cookie, commodityID, csrf string) error { } // Exchange 正式兑换一个商品，num 固定为 1，和浏览器抓包一致。 -func (c *DouyuClient) Exchange(cookie, commodityID, csrf string) (string, error) { +func (c *DouyuClient) Exchange(cookie, commodityID string, csrf *string) (string, error) { + currentCSRF := "" + if csrf != nil { + currentCSRF = *csrf + } var result struct { APIResponse Data struct { @@ -340,7 +364,7 @@ func (c *DouyuClient) Exchange(cookie, commodityID, csrf string) (string, error) "commodityID": commodityID, "num": "1", "rid": roomID, - "csrfToken": csrf, + "csrfToken": currentCSRF, }, &result); err != nil { return "", err } @@ -379,11 +403,12 @@ func (c *DouyuClient) getJSON(url, cookie, label string, out any) error { return err } c.setBrowserHeaders(req, cookie) - return c.doJSON(req, out, label) + _, err = c.doJSON(req, out, label) + return err } -// postMultipart 发送 multipart/form-data 请求并解析 JSON。 -func (c *DouyuClient) postMultipart(url, cookie, csrf, label string, fields map[string]string, out any) error { +// postMultipart 发送 multipart/form-data 请求并解析 JSON，并同步响应里的 CSRF Cookie。 +func (c *DouyuClient) postMultipart(url, cookie string, csrf *string, label string, fields map[string]string, out any) error { var body bytes.Buffer writer := multipart.NewWriter(&body) for key, value := range fields { @@ -399,49 +424,62 @@ func (c *DouyuClient) postMultipart(url, cookie, csrf, label string, fields map[ if err != nil { return err } - c.setBrowserHeaders(req, mergeCookie(cookie, "cvl_csrf_token", csrf)) + currentCSRF := "" + if csrf != nil { + currentCSRF = *csrf + } + c.setBrowserHeaders(req, mergeCookie(cookie, "cvl_csrf_token", currentCSRF)) req.Header.Set("Content-Type", writer.FormDataContentType()) - return c.doJSON(req, out, label) + resp, err := c.doJSON(req, out, label) + if resp != nil { + if nextCSRF := csrfFromResponse(resp); nextCSRF != "" && csrf != nil { + if nextCSRF != *csrf { + logEvent("网络", "info", "响应刷新CSRF已同步: %s %s %s", label, req.Method, requestPath(req)) + } + *csrf = nextCSRF + } + } + return err } // doJSON 执行请求、检查 HTTP 状态码并解析 JSON。 -func (c *DouyuClient) doJSON(req *http.Request, out any, label string) error { +func (c *DouyuClient) doJSON(req *http.Request, out any, label string) (*http.Response, error) { start := time.Now() reqBody := traceRequestBody(c.trace, 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 err + return nil, err } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { - c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, nil, err.Error(), 0, "") + c.trace.recordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, nil, err.Error(), 0, "") logNetworkError(label, req, time.Since(start), err) - return err + return resp, err } if resp.StatusCode != http.StatusOK { - c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") + c.trace.recordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") logNetworkHTTPError(label, req, resp.StatusCode, time.Since(start), body) - return fmt.Errorf("HTTP %d: %s", resp.StatusCode, trimForLog(string(body), 160)) + return resp, fmt.Errorf("HTTP %d: %s", resp.StatusCode, trimForLog(string(body), 160)) } if err := json.Unmarshal(body, out); err != nil { - c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, body, err.Error(), 0, "") + 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 fmt.Errorf("JSON 解析失败: %w, body=%s", err, trimForLog(string(body), 160)) + return resp, fmt.Errorf("JSON 解析失败: %w, body=%s", err, trimForLog(string(body), 160)) } if apiError, apiMsg, ok := extractAPIStatus(out); ok && apiError != 0 { - c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, body, apiMsg, apiError, apiMsg) + 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.record(label, req, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") + c.trace.recordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") logNetworkSuccess(label, req, time.Since(start), 0, "") } else { - c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") + c.trace.recordResponse(label, req, resp, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") } - return nil + return resp, nil } func shouldLogNetworkSuccess(label string) bool { diff --git a/handlers.go b/handlers.go index 9ffda28..13b681b 100644 --- a/handlers.go +++ b/handlers.go @@ -91,6 +91,10 @@ func (s *appState) handleSeckill(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, errors.New("请选择账号和商品")) return } + if !s.tryStartRun() { + writeError(w, http.StatusConflict, errors.New("抢购任务正在运行，请勿重复提交")) + return + } go s.runNow(cloneStringMap(req.AccountProducts)) writeJSON(w, map[string]any{"success": true, "msg": "抢购已启动"}) } diff --git a/seckill.go b/seckill.go index cbfff32..f351a1a 100644 --- a/seckill.go +++ b/seckill.go @@ -18,6 +18,7 @@ const ( // runNow 立即执行一批抢购任务。 func (s *appState) runNow(accountProducts map[string]string) { + defer s.finishRun() s.clearResults() attempts := s.buildAttempts(accountProducts) if len(attempts) == 0 { @@ -50,6 +51,22 @@ func (s *appState) runNow(accountProducts map[string]string) { logEvent("抢购", "info", "抢购任务结束") } +func (s *appState) tryStartRun() bool { + s.runMu.Lock() + defer s.runMu.Unlock() + if s.running { + return false + } + s.running = true + return true +} + +func (s *appState) finishRun() { + s.runMu.Lock() + defer s.runMu.Unlock() + s.running = false +} + // buildAttempts 根据前端传入的账号到商品映射，生成具体兑换尝试。 func (s *appState) buildAttempts(accountProducts map[string]string) []ExchangeAttempt { accounts := s.accountSnapshot() @@ -186,7 +203,7 @@ func (s *appState) stopSchedulesLocked() { } // runAttempt 按抓包链路执行一次兑换：CSRF -> 限制 -> 预兑换 -> 正式兑换。 -// 同一次抢购尝试只生成一个 CSRF，后续接口共用，减少临门请求耗时。 +// 同一次抢购尝试只生成一次 CSRF；如果后续响应刷新 token，则沿链路继续复用最新值。 func (s *appState) runAttempt(attempt ExchangeAttempt) ExchangeResult { result := ExchangeResult{ Account: attempt.Account.Name, @@ -212,7 +229,7 @@ func (s *appState) runAttempt(attempt ExchangeAttempt) ExchangeResult { logEvent("抢购", "info", "CSRF已生成，本次兑换链路复用同一个 token: 账号=%s 商品=%s", attempt.Account.Name, attempt.Product.CommodityName) } - limit, err := s.client.BatchExchangeLimit(attempt.Account.Cookie, attempt.Product.CommodityID, csrf) + limit, err := s.client.BatchExchangeLimit(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf) if err != nil { result.Message = "失败|批量限制检查失败: " + err.Error() return result @@ -223,12 +240,12 @@ func (s *appState) runAttempt(attempt ExchangeAttempt) ExchangeResult { } logEvent("抢购", "info", "批量限制通过: 账号=%s 商品=%s limit=%d", attempt.Account.Name, attempt.Product.CommodityName, limit) - if err := s.client.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, csrf); err != nil { + if err := s.client.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf); err != nil { result.Message = "失败|预兑换失败: " + err.Error() return result } - exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, csrf) + exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf) if err != nil { result.Message = "失败|" + err.Error() return result @@ -352,6 +369,10 @@ func (s *appState) scheduleTimers(accountProducts map[string]string, attempts [] estimatedServer.In(cst).Format("15:04:05.000"), calibration.Offset.Milliseconds(), calibration.ErrorBound.Milliseconds()) + if !s.tryStartRun() { + logEvent("抢购", "warn", "已有抢购任务正在运行，本次定时触发已跳过") + return + } s.runNow(accountProducts) }) if resync { diff --git a/trace.go b/trace.go index 4cf04cc..db31afd 100644 --- a/trace.go +++ b/trace.go @@ -42,9 +42,9 @@ func initTrace(logDir string, cfg TraceConfig) (*traceRecorder, func()) { filePath: path, } if cfg.MaxBodyBytes > 0 { - logEvent("Trace", "warn", "trace 已启用: %s，单段正文最多保留 %d 字节，不做脱敏", path, cfg.MaxBodyBytes) + logEvent("Trace", "warn", "trace 已启用: %s，单段正文最多保留 %d 字节；请求头不脱敏，响应 Set-Cookie 隐藏值", path, cfg.MaxBodyBytes) } else { - logEvent("Trace", "warn", "trace 已启用: %s，全量保留请求和响应，不做脱敏", path) + logEvent("Trace", "warn", "trace 已启用: %s，全量保留请求和响应；请求头不脱敏，响应 Set-Cookie 隐藏值", path) } return recorder, func() { _ = recorder.Close() @@ -60,6 +60,10 @@ func (t *traceRecorder) Close() error { } func (t *traceRecorder) record(label string, req *http.Request, status int, cost time.Duration, reqBody, respBody []byte, errText string, apiError int, apiMsg string) { + t.recordResponse(label, req, nil, status, cost, reqBody, respBody, errText, apiError, apiMsg) +} + +func (t *traceRecorder) recordResponse(label string, req *http.Request, resp *http.Response, status int, cost time.Duration, reqBody, respBody []byte, errText string, apiError int, apiMsg string) { if t == nil || !t.enabled || req == nil || req.URL == nil { return } @@ -76,6 +80,7 @@ func (t *traceRecorder) record(label string, req *http.Request, status int, cost Status: status, Cost: cost, ReqHeader: sanitizeHeader(req.Header), + RespHeader: sanitizeResponseHeader(resp), ReqBody: sanitizeBodyForTrace(reqBody, req.Header.Get("Content-Type"), t.maxBodyBytes), RespBody: sanitizeResponseBody(respBody, t.maxBodyBytes), APIError: apiError, @@ -114,6 +119,66 @@ func sanitizeHeader(header http.Header) map[string][]string { return out } +func sanitizeResponseHeader(resp *http.Response) map[string][]string { + if resp == nil { + return nil + } + out := sanitizeHeader(resp.Header) + if len(out) == 0 { + return out + } + if _, ok := out["Set-Cookie"]; ok { + out["Set-Cookie"] = summarizeSetCookies(resp) + } + return out +} + +func summarizeSetCookies(resp *http.Response) []string { + if resp == nil { + return nil + } + cookies := resp.Cookies() + if len(cookies) == 0 { + values := resp.Header.Values("Set-Cookie") + summary := make([]string, 0, len(values)) + for range values { + summary = append(summary, "") + } + return summary + } + summary := make([]string, 0, len(cookies)) + for _, ck := range cookies { + if ck == nil || ck.Name == "" { + continue + } + parts := []string{ck.Name + "="} + if ck.Path != "" { + parts = append(parts, "Path="+ck.Path) + } + if ck.Domain != "" { + parts = append(parts, "Domain="+ck.Domain) + } + if ck.MaxAge != 0 { + parts = append(parts, fmt.Sprintf("Max-Age=%d", ck.MaxAge)) + } + if !ck.Expires.IsZero() { + parts = append(parts, "Expires="+ck.Expires.In(cst).Format(time.RFC3339)) + } + if ck.HttpOnly { + parts = append(parts, "HttpOnly") + } + if ck.Secure { + parts = append(parts, "Secure") + } + summary = append(summary, strings.Join(parts, "; ")) + } + if len(summary) == 0 { + return []string{""} + } + sort.Strings(summary) + return summary +} + func sanitizeBodyForTrace(body []byte, contentType string, max int) string { if len(body) == 0 { return "" @@ -192,6 +257,7 @@ type traceEntry struct { Status int Cost time.Duration ReqHeader map[string][]string + RespHeader map[string][]string ReqBody string RespBody string APIError int @@ -220,6 +286,8 @@ func formatTraceEntry(e traceEntry) string { if e.ErrorText != "" { fmt.Fprintf(&b, "错误: %s\n", e.ErrorText) } + b.WriteString("\n[响应头]\n") + b.WriteString(formatTraceHeaders(e.RespHeader)) b.WriteString("\n[请求头]\n") b.WriteString(formatTraceHeaders(e.ReqHeader)) b.WriteString("\n[请求体]\n") @@ -227,9 +295,9 @@ func formatTraceEntry(e traceEntry) string { b.WriteString("\n[响应体]\n") b.WriteString(formatTraceBody(e.RespBody)) if e.MaxBodyBytes > 0 { - fmt.Fprintf(&b, "\n提示: 单段正文最多保留 %d 字节；trace 不做脱敏。\n", e.MaxBodyBytes) + fmt.Fprintf(&b, "\n提示: 单段正文最多保留 %d 字节；请求头不脱敏，响应 Set-Cookie 隐藏值。\n", e.MaxBodyBytes) } else { - b.WriteString("\n提示: trace 全量保留请求和响应，不做脱敏。\n") + b.WriteString("\n提示: trace 全量保留请求和响应；请求头不脱敏，响应 Set-Cookie 隐藏值。\n") } b.WriteByte('\n') return b.String() diff --git a/types.go b/types.go index 6bbb54b..fd22c3c 100644 --- a/types.go +++ b/types.go @@ -100,6 +100,9 @@ type appState struct { resultsMu sync.RWMutex results map[string]string + runMu sync.Mutex + running bool + scheduleMu sync.Mutex schedule *time.Timer warmupSchedule *time.Timer