diff --git a/config.go b/config.go index 088fa0c..28148de 100644 --- a/config.go +++ b/config.go @@ -5,6 +5,7 @@ import ( "errors" "os" "path/filepath" + "strconv" ) // defaultAppPaths 收集运行时路径,支持环境变量覆盖,方便不同平台和部署方式调整。 @@ -25,6 +26,35 @@ func envOrDefault(key, fallback string) string { return fallback } +// loadTraceConfig 读取 trace 开关;默认关闭,避免保存敏感调试数据。 +func loadTraceConfig() TraceConfig { + return TraceConfig{ + Enabled: isTruthyEnv("DOUYU_TRACE"), + MaxBodyBytes: envInt("DOUYU_TRACE_MAX_BODY", 0), + } +} + +func isTruthyEnv(key string) bool { + switch os.Getenv(key) { + case "1", "true", "TRUE", "yes", "YES", "on", "ON": + return true + default: + return false + } +} + +func envInt(key string, fallback int) int { + raw := os.Getenv(key) + if raw == "" { + return fallback + } + value, err := strconv.Atoi(raw) + if err != nil || value <= 0 { + return fallback + } + return value +} + // loadConfig 读取配置文件;缺失时安静使用内置默认配置。 func loadConfig(path string) Config { cfg := Config{TimeAPIURL: defaultTimeAPI} diff --git a/douyu_client.go b/douyu_client.go index 7f419c9..5a11c67 100644 --- a/douyu_client.go +++ b/douyu_client.go @@ -15,9 +15,10 @@ import ( ) // NewDouyuClient 创建一个专门请求斗鱼接口的客户端。 -func NewDouyuClient(timeAPIURL string) *DouyuClient { +func NewDouyuClient(timeAPIURL string, trace *traceRecorder) *DouyuClient { return &DouyuClient{ timeAPIURL: timeAPIURL, + trace: trace, httpClient: &http.Client{ Timeout: 8 * time.Second, Transport: &http.Transport{ @@ -115,8 +116,10 @@ func (c *DouyuClient) GenerateCSRF(cookie string) (string, error) { c.setBrowserHeaders(req, cookie) start := time.Now() + reqBody := traceRequestBody(c.trace, 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 } @@ -124,29 +127,37 @@ 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, "") 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, "") 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, "") 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) 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) 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) 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") @@ -278,8 +289,10 @@ func (c *DouyuClient) postMultipart(url, cookie, csrf, label string, fields map[ // doJSON 执行请求、检查 HTTP 状态码并解析 JSON。 func (c *DouyuClient) doJSON(req *http.Request, out any, label string) 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 } @@ -287,21 +300,28 @@ func (c *DouyuClient) doJSON(req *http.Request, out any, label string) error { body, err := io.ReadAll(resp.Body) if err != nil { + c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, nil, err.Error(), 0, "") logNetworkError(label, req, time.Since(start), err) return err } if resp.StatusCode != http.StatusOK { + c.trace.record(label, req, 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)) } if err := json.Unmarshal(body, out); err != nil { + c.trace.record(label, req, 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)) } if apiError, apiMsg, ok := extractAPIStatus(out); ok && apiError != 0 { + c.trace.record(label, req, 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, "") logNetworkSuccess(label, req, time.Since(start), 0, "") + } else { + c.trace.record(label, req, resp.StatusCode, time.Since(start), reqBody, body, "", 0, "") } return nil } diff --git a/log_store.go b/log_store.go index f6dabd9..776d282 100644 --- a/log_store.go +++ b/log_store.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "sync/atomic" "time" ) @@ -27,27 +28,83 @@ func initLogging(store *memoryLogStore, logDir string) func() { logDir = defaultLogDir } if err := os.MkdirAll(logDir, 0755); err != nil { - log.SetOutput(io.MultiWriter(os.Stdout, store)) + async := newAsyncLogWriter(io.MultiWriter(os.Stdout, store), 2048) + log.SetOutput(async) log.Printf("日志目录创建失败,仅写入终端和内存: %v", err) - return func() {} + return func() { _ = async.Close() } } filePath := filepath.Join(logDir, "douyu_"+time.Now().In(cst).Format("20060102_150405")+".log") file, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) if err != nil { - log.SetOutput(io.MultiWriter(os.Stdout, store)) + async := newAsyncLogWriter(io.MultiWriter(os.Stdout, store), 2048) + log.SetOutput(async) log.Printf("日志文件创建失败,仅写入终端和内存: %v", err) - return func() {} + return func() { _ = async.Close() } } store.setFilePath(filePath) - log.SetOutput(io.MultiWriter(os.Stdout, file, store)) + async := newAsyncLogWriter(io.MultiWriter(os.Stdout, file, store), 2048) + log.SetOutput(async) log.Printf("日志文件已启用: %s", filePath) return func() { + _ = async.Close() _ = file.Close() } } +func newAsyncLogWriter(dst io.Writer, buffer int) *asyncLogWriter { + if buffer <= 0 { + buffer = 1024 + } + w := &asyncLogWriter{ + dst: dst, + ch: make(chan []byte, buffer), + } + w.wg.Add(1) + go w.run() + return w +} + +// Write 异步写日志;队列满时丢弃,避免抢购请求被磁盘 I/O 阻塞。 +func (w *asyncLogWriter) Write(p []byte) (int, error) { + if w == nil { + return len(p), nil + } + if atomic.LoadUint32(&w.closed) == 1 { + return len(p), nil + } + buf := append([]byte(nil), p...) + select { + case w.ch <- buf: + default: + atomic.AddUint64(&w.dropped, 1) + } + return len(p), nil +} + +func (w *asyncLogWriter) run() { + defer w.wg.Done() + for p := range w.ch { + _, _ = w.dst.Write(p) + } + if dropped := atomic.LoadUint64(&w.dropped); dropped > 0 { + _, _ = fmt.Fprintf(w.dst, "%s [日志][warn] 日志队列已丢弃 %d 条,已优先保护抢购链路\n", + time.Now().In(cst).Format("2006/01/02 15:04:05.000000"), dropped) + } +} + +func (w *asyncLogWriter) Close() error { + if w == nil { + return nil + } + if atomic.CompareAndSwapUint32(&w.closed, 0, 1) { + close(w.ch) + w.wg.Wait() + } + return nil +} + // Write 实现 io.Writer,把标准 log 输出转存为内存日志。 func (s *memoryLogStore) Write(p []byte) (int, error) { text := strings.TrimRight(string(p), "\r\n") diff --git a/main.go b/main.go index 3f7f2f1..54edb0b 100644 --- a/main.go +++ b/main.go @@ -13,9 +13,12 @@ func main() { state.logs = newMemoryLogStore(800) closeLogFile := initLogging(state.logs, state.paths.LogDir) defer closeLogFile() + trace, closeTrace := initTrace(state.paths.LogDir, loadTraceConfig()) + state.trace = trace + defer closeTrace() state.config = loadConfig(state.paths.ConfigFile) - state.client = NewDouyuClient(state.config.TimeAPIURL) + state.client = NewDouyuClient(state.config.TimeAPIURL, state.trace) state.loadAccounts() state.loadFont() if err := os.MkdirAll(state.paths.SuccessDir, 0755); err != nil { @@ -32,5 +35,7 @@ func main() { state.registerRoutes(mux) logEvent("系统", "success", "服务启动: http://%s | 账号: %d", serverAddr, state.accountCount()) - log.Fatal(http.ListenAndServe(serverAddr, mux)) + if err := http.ListenAndServe(serverAddr, mux); err != nil { + logEvent("系统", "error", "服务退出: %v", err) + } } diff --git a/trace.go b/trace.go new file mode 100644 index 0000000..4cf04cc --- /dev/null +++ b/trace.go @@ -0,0 +1,275 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "mime" + "mime/multipart" + "net/http" + "os" + "path/filepath" + "sort" + "strings" + "sync/atomic" + "time" +) + +func initTrace(logDir string, cfg TraceConfig) (*traceRecorder, func()) { + if !cfg.Enabled { + return &traceRecorder{enabled: false}, func() {} + } + if logDir == "" { + logDir = defaultLogDir + } + if err := os.MkdirAll(logDir, 0755); err != nil { + logEvent("Trace", "error", "trace 目录创建失败,已关闭 trace: %v", err) + return &traceRecorder{enabled: false}, func() {} + } + + path := filepath.Join(logDir, "douyu_trace_"+time.Now().In(cst).Format("20060102_150405")+".log") + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) + if err != nil { + logEvent("Trace", "error", "trace 文件创建失败,已关闭 trace: %v", err) + return &traceRecorder{enabled: false}, func() {} + } + + recorder := &traceRecorder{ + enabled: true, + maxBodyBytes: cfg.MaxBodyBytes, + writer: newAsyncLogWriter(file, 2048), + filePath: path, + } + if cfg.MaxBodyBytes > 0 { + logEvent("Trace", "warn", "trace 已启用: %s,单段正文最多保留 %d 字节,不做脱敏", path, cfg.MaxBodyBytes) + } else { + logEvent("Trace", "warn", "trace 已启用: %s,全量保留请求和响应,不做脱敏", path) + } + return recorder, func() { + _ = recorder.Close() + _ = file.Close() + } +} + +func (t *traceRecorder) Close() error { + if t == nil || t.writer == nil { + return nil + } + return t.writer.Close() +} + +func (t *traceRecorder) record(label string, req *http.Request, 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 + } + if label == "服务时间" && status == http.StatusOK && errText == "" && apiError == 0 { + return + } + entry := traceEntry{ + Seq: atomic.AddUint64(&t.seq, 1), + Time: time.Now().In(cst).Format("2006-01-02 15:04:05.000"), + Label: label, + Method: req.Method, + URL: req.URL.Redacted(), + Path: req.URL.Path, + Status: status, + Cost: cost, + ReqHeader: sanitizeHeader(req.Header), + ReqBody: sanitizeBodyForTrace(reqBody, req.Header.Get("Content-Type"), t.maxBodyBytes), + RespBody: sanitizeResponseBody(respBody, t.maxBodyBytes), + APIError: apiError, + APIMessage: apiMsg, + ErrorText: errText, + MaxBodyBytes: t.maxBodyBytes, + } + _, _ = t.writer.Write([]byte(formatTraceEntry(entry))) +} + +func traceRequestBody(t *traceRecorder, req *http.Request) []byte { + if t == nil || !t.enabled || req == nil || req.GetBody == nil { + return nil + } + body, err := req.GetBody() + if err != nil { + return nil + } + defer body.Close() + if t.maxBodyBytes > 0 { + data, _ := io.ReadAll(io.LimitReader(body, int64(t.maxBodyBytes)+1)) + return data + } + data, _ := io.ReadAll(body) + return data +} + +func sanitizeHeader(header http.Header) map[string][]string { + out := make(map[string][]string, len(header)) + for key, values := range header { + canonical := http.CanonicalHeaderKey(key) + copied := make([]string, len(values)) + copy(copied, values) + out[canonical] = copied + } + return out +} + +func sanitizeBodyForTrace(body []byte, contentType string, max int) string { + if len(body) == 0 { + return "" + } + mediaType, params, _ := mime.ParseMediaType(contentType) + if strings.HasPrefix(mediaType, "multipart/") { + boundary := params["boundary"] + if boundary != "" { + return sanitizeMultipartBody(body, boundary, max) + } + } + return sanitizeTextBody(body, max) +} + +func sanitizeResponseBody(body []byte, max int) string { + return sanitizeTextBody(body, max) +} + +func sanitizeTextBody(body []byte, max int) string { + raw := string(body) + if pretty, ok := prettyJSON(raw); ok { + return truncateTraceText(pretty, max) + } + return truncateTraceText(raw, max) +} + +func sanitizeMultipartBody(body []byte, boundary string, max int) string { + reader := multipartReader(bytes.NewReader(body), boundary) + fields := make(map[string][]string) + for { + part, err := reader.NextPart() + if err == io.EOF { + break + } + if err != nil { + return truncateTraceText(string(body), max) + } + name := part.FormName() + if name == "" { + continue + } + var valueBytes []byte + if max > 0 { + valueBytes, _ = io.ReadAll(io.LimitReader(part, int64(max)+1)) + } else { + valueBytes, _ = io.ReadAll(part) + } + value := truncateTraceText(string(valueBytes), max) + fields[name] = append(fields[name], value) + } + encoded, err := json.Marshal(fields) + if err != nil { + return "" + } + return string(encoded) +} + +func multipartReader(r io.Reader, boundary string) *multipart.Reader { + return multipart.NewReader(r, boundary) +} + +func truncateTraceText(s string, max int) string { + if max <= 0 || len(s) <= max { + return s + } + return s[:max] + fmt.Sprintf("...", len(s)-max) +} + +type traceEntry struct { + Seq uint64 + Time string + Label string + Method string + URL string + Path string + Status int + Cost time.Duration + ReqHeader map[string][]string + ReqBody string + RespBody string + APIError int + APIMessage string + ErrorText string + MaxBodyBytes int +} + +func formatTraceEntry(e traceEntry) string { + var b strings.Builder + statusText := "成功" + if e.Status < 200 || e.Status >= 300 || e.APIError != 0 || e.ErrorText != "" { + statusText = "异常" + } + + b.WriteString(strings.Repeat("=", 96)) + b.WriteByte('\n') + fmt.Fprintf(&b, "#%06d %s [%s] %s %s\n", e.Seq, e.Time, statusText, e.Method, e.URL) + fmt.Fprintf(&b, "接口: %s\n", emptyDefault(e.Label, "-")) + fmt.Fprintf(&b, "路径: %s\n", emptyDefault(e.Path, "-")) + fmt.Fprintf(&b, "结果: HTTP %d | 耗时 %s | apiError=%d", e.Status, roundDuration(e.Cost), e.APIError) + if e.APIMessage != "" { + fmt.Fprintf(&b, " | msg=%s", e.APIMessage) + } + b.WriteByte('\n') + if e.ErrorText != "" { + fmt.Fprintf(&b, "错误: %s\n", e.ErrorText) + } + b.WriteString("\n[请求头]\n") + b.WriteString(formatTraceHeaders(e.ReqHeader)) + b.WriteString("\n[请求体]\n") + b.WriteString(formatTraceBody(e.ReqBody)) + b.WriteString("\n[响应体]\n") + b.WriteString(formatTraceBody(e.RespBody)) + if e.MaxBodyBytes > 0 { + fmt.Fprintf(&b, "\n提示: 单段正文最多保留 %d 字节;trace 不做脱敏。\n", e.MaxBodyBytes) + } else { + b.WriteString("\n提示: trace 全量保留请求和响应,不做脱敏。\n") + } + b.WriteByte('\n') + return b.String() +} + +func formatTraceHeaders(headers map[string][]string) string { + if len(headers) == 0 { + return "\n" + } + keys := make([]string, 0, len(headers)) + for key := range headers { + keys = append(keys, key) + } + sort.Strings(keys) + var b strings.Builder + for _, key := range keys { + fmt.Fprintf(&b, "%s: %s\n", key, strings.Join(headers[key], ", ")) + } + return b.String() +} + +func formatTraceBody(body string) string { + body = strings.TrimSpace(body) + if body == "" { + return "\n" + } + if pretty, ok := prettyJSON(body); ok { + return pretty + "\n" + } + return body + "\n" +} + +func prettyJSON(s string) (string, bool) { + var value any + if err := json.Unmarshal([]byte(s), &value); err != nil { + return "", false + } + out, err := json.MarshalIndent(value, "", " ") + if err != nil { + return "", false + } + return string(out), true +} diff --git a/types.go b/types.go index 7d71dfb..a44c80b 100644 --- a/types.go +++ b/types.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "io" "net/http" "sync" "time" @@ -21,6 +22,11 @@ type appPaths struct { FontFiles []string } +type TraceConfig struct { + Enabled bool + MaxBodyBytes int +} + type Account struct { Name string `json:"name"` Password string `json:"password"` @@ -82,6 +88,7 @@ type appState struct { paths appPaths client *DouyuClient logs *memoryLogStore + trace *traceRecorder accountsMu sync.RWMutex accounts []Account @@ -103,6 +110,23 @@ type appState struct { type DouyuClient struct { httpClient *http.Client timeAPIURL string + trace *traceRecorder +} + +type traceRecorder struct { + enabled bool + maxBodyBytes int + writer *asyncLogWriter + filePath string + seq uint64 +} + +type asyncLogWriter struct { + dst io.Writer + ch chan []byte + wg sync.WaitGroup + closed uint32 + dropped uint64 } type ExchangeAttempt struct {