yml/dygou_go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化日志

Browse Source
This commit is contained in:
yml2213
2026-06-19 13:02:28 +08:00
parent 59b9dd12ef
commit 878daa4ab8
6 changed files with 419 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.go
+30
View File
@@ -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}
douyu_client.go
+21 -1
View File
@@ -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
}
log_store.go
+62 -5
View File
@@ -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")
main.go
+7 -2
View File
@@ -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)
}
}
trace.go
+275
View File
@@ -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("...<truncated %d bytes>", 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 "<empty>\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 "<empty>\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
}
types.go
+24
View File
@@ -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 {