diff --git a/backend/internal/modules/chat/handler_qrcode.go b/backend/internal/modules/chat/handler_qrcode.go index 262ed86..a21a8ca 100644 --- a/backend/internal/modules/chat/handler_qrcode.go +++ b/backend/internal/modules/chat/handler_qrcode.go @@ -128,6 +128,7 @@ func (h *Handler) RecognizeQrCodeGroupNameHandler(c *gin.Context) { return } if errors.Is(err, ErrQrCodeOCRUnavailable) { + response.RecordError(c, err) c.JSON(http.StatusBadGateway, gin.H{"error": "PaddleOCR 服务暂时不可用"}) return } @@ -164,6 +165,7 @@ func (h *Handler) SubmitOCRJobHandler(c *gin.Context) { return } if errors.Is(err, ErrQrCodeOCRUnavailable) { + response.RecordError(c, err) c.JSON(http.StatusBadGateway, gin.H{"error": "PaddleOCR 服务暂时不可用"}) return } diff --git a/backend/internal/modules/chat/qrcode_ocr.go b/backend/internal/modules/chat/qrcode_ocr.go index 0d5e728..6e4c225 100644 --- a/backend/internal/modules/chat/qrcode_ocr.go +++ b/backend/internal/modules/chat/qrcode_ocr.go @@ -6,21 +6,46 @@ import ( "encoding/json" "errors" "fmt" + "image" + "image/color" + "image/draw" + _ "image/gif" + "image/jpeg" + _ "image/png" "io" + "math" "mime/multipart" "net/http" "net/textproto" "regexp" "sort" + "strconv" "strings" + "sync" "time" "unicode/utf8" + "hfb_sys/backend/internal/logging" + "github.com/google/uuid" + "go.uber.org/zap" + xdraw "golang.org/x/image/draw" + _ "golang.org/x/image/webp" ) const ( maxQrCodeOCRFileSize = 10 * 1024 * 1024 + maxQrCodeOCRPixels = 40_000_000 + maxQrCodeOCRSide = 2560 + qrCodeOCRJPEGQuality = 92 + + ocrRequestConcurrency = 2 + ocrRequestMinInterval = 250 * time.Millisecond + ocrRequestMaxAttempts = 3 + ocrRetryBaseDelay = 500 * time.Millisecond + ocrResponseBodyLimit = 2 * 1024 * 1024 + ocrResultBodyLimit = 10 * 1024 * 1024 + ocrLogBodyLimit = 2048 ocrJobRedisKeyPrefix = "ocr:job:" ocrJobRedisTTL = 10 * time.Minute @@ -38,9 +63,87 @@ var ( ErrQrCodeOCRJobNotFound = errors.New("OCR 任务不存在或已过期") ErrQrCodeOCRRedisDisabled = errors.New("OCR 异步模式需要 Redis") - ocrHTTPClient = &http.Client{Timeout: 30 * time.Second} + ocrHTTPClient = &http.Client{Timeout: 30 * time.Second} + sharedOCRRequestGate = newOCRRequestGate(ocrRequestConcurrency, ocrRequestMinInterval) ) +type ocrRequestGate struct { + slots chan struct{} + mu sync.Mutex + nextAllowed time.Time + minInterval time.Duration +} + +func newOCRRequestGate(concurrency int, minInterval time.Duration) *ocrRequestGate { + if concurrency < 1 { + concurrency = 1 + } + return &ocrRequestGate{ + slots: make(chan struct{}, concurrency), + minInterval: max(0, minInterval), + } +} + +func (g *ocrRequestGate) acquire(ctx context.Context, sleep func(context.Context, time.Duration) error) (func(), error) { + select { + case g.slots <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + } + + release := func() { <-g.slots } + g.mu.Lock() + now := time.Now() + wait := max(time.Duration(0), g.nextAllowed.Sub(now)) + startAt := now.Add(wait) + g.nextAllowed = startAt.Add(g.minInterval) + g.mu.Unlock() + + if wait > 0 { + if err := sleep(ctx, wait); err != nil { + release() + return nil, err + } + } + return release, nil +} + +type ocrUpstreamError struct { + Operation string + StatusCode int + Code string + Message string + TraceID string + RetryAfter time.Duration + BodySnippet string + Cause error + Retriable bool +} + +func (e *ocrUpstreamError) Error() string { + parts := []string{ErrQrCodeOCRUnavailable.Error(), "operation=" + e.Operation} + if e.StatusCode != 0 { + parts = append(parts, "status="+strconv.Itoa(e.StatusCode)) + } + if e.Code != "" { + parts = append(parts, "code="+e.Code) + } + if e.Message != "" { + parts = append(parts, "message="+e.Message) + } + if e.TraceID != "" { + parts = append(parts, "trace_id="+e.TraceID) + } + if e.Cause != nil { + parts = append(parts, "cause="+e.Cause.Error()) + } + return strings.Join(parts, " ") +} + +func (e *ocrUpstreamError) Unwrap() error { + return ErrQrCodeOCRUnavailable +} + type QrCodeOCRResult struct { GroupName string `json:"group_name"` Candidates []string `json:"candidates"` @@ -79,13 +182,46 @@ func readAndValidateOCRFile(reader io.Reader, contentType string) ([]byte, strin if err != nil || len(data) == 0 || len(data) > maxQrCodeOCRFileSize { return nil, "", ErrQrCodeOCRInvalidFile } - if contentType == "" { - contentType = http.DetectContentType(data) + contentType = strings.ToLower(strings.TrimSpace(strings.Split(contentType, ";")[0])) + if contentType == "" || !strings.HasPrefix(contentType, "image/") { + contentType = strings.ToLower(strings.TrimSpace(strings.Split(http.DetectContentType(data), ";")[0])) } if !strings.HasPrefix(contentType, "image/") { return nil, "", ErrQrCodeOCRInvalidFile } - return data, contentType, nil + config, _, err := image.DecodeConfig(bytes.NewReader(data)) + if err != nil || config.Width <= 0 || config.Height <= 0 || int64(config.Width)*int64(config.Height) > maxQrCodeOCRPixels { + return nil, "", ErrQrCodeOCRInvalidFile + } + source, _, err := image.Decode(bytes.NewReader(data)) + if err != nil { + return nil, "", ErrQrCodeOCRInvalidFile + } + normalized := resizeOCRImage(source, maxQrCodeOCRSide) + var buffer bytes.Buffer + if err := jpeg.Encode(&buffer, normalized, &jpeg.Options{Quality: qrCodeOCRJPEGQuality}); err != nil { + return nil, "", ErrQrCodeOCRInvalidFile + } + if buffer.Len() == 0 || buffer.Len() > maxQrCodeOCRFileSize { + return nil, "", ErrQrCodeOCRInvalidFile + } + return buffer.Bytes(), "image/jpeg", nil +} + +func resizeOCRImage(source image.Image, maxSide int) image.Image { + bounds := source.Bounds() + width := bounds.Dx() + height := bounds.Dy() + if width <= 0 || height <= 0 { + return source + } + scale := math.Min(1, float64(maxSide)/float64(max(width, height))) + targetWidth := max(1, int(math.Round(float64(width)*scale))) + targetHeight := max(1, int(math.Round(float64(height)*scale))) + target := image.NewRGBA(image.Rect(0, 0, targetWidth, targetHeight)) + draw.Draw(target, target.Bounds(), &image.Uniform{C: color.White}, image.Point{}, draw.Src) + xdraw.ApproxBiLinear.Scale(target, target.Bounds(), source, bounds, draw.Over, nil) + return target } func ocrJobKey(jobID string) string { @@ -132,15 +268,15 @@ func (r *Repository) RecognizeQrCodeGroupName(ctx context.Context, filename, con return nil, err } - paddleJobID, err := submitPaddleOCRJob(ctx, config, filename, contentType, data) + paddleJobID, err := r.submitPaddleOCRJob(ctx, config, filename, contentType, data) if err != nil { return nil, err } - jsonURL, err := waitPaddleOCRJob(ctx, config, paddleJobID) + jsonURL, err := r.waitPaddleOCRJob(ctx, config, paddleJobID) if err != nil { return nil, err } - rawText, err := fetchPaddleOCRText(ctx, jsonURL) + rawText, err := r.fetchPaddleOCRText(ctx, jsonURL) if err != nil { return nil, err } @@ -173,7 +309,7 @@ func (r *Repository) SubmitOCRJob(ctx context.Context, filename, contentType str return "", err } - paddleJobID, err := submitPaddleOCRJob(ctx, config, filename, contentType, data) + paddleJobID, err := r.submitPaddleOCRJob(ctx, config, filename, contentType, data) if err != nil { return "", err } @@ -206,47 +342,48 @@ func (r *Repository) PollOCRJobResult(ctx context.Context, jobID string) (*OCRJo } statusURL := strings.TrimRight(config.JobURL, "/") + "/" + record.PaddleJobID - req, err := http.NewRequestWithContext(ctx, http.MethodGet, statusURL, nil) + respBody, err := r.doPaddleRequest(ctx, "status", func(ctx context.Context) (*http.Request, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, statusURL, nil) + if err == nil { + req.Header.Set("Authorization", "bearer "+config.Token) + } + return req, err + }) if err != nil { - return nil, fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - req.Header.Set("Authorization", "bearer "+config.Token) - - resp, err := ocrHTTPClient.Do(req) - if err != nil { - // 网络错误不更新 Redis，保留当前状态让前端重试 - return record, nil - } - respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024)) - resp.Body.Close() - - if resp.StatusCode != http.StatusOK { + // 保留当前状态，让下一次前端轮询继续尝试。 return record, nil } var statusPayload paddleOCRJobStatusResponse if err := json.Unmarshal(respBody, &statusPayload); err != nil { + upstreamErr := newOCRUpstreamError("status_decode", http.StatusOK, nil, respBody, err) + r.logOCRUpstreamFailure(ctx, upstreamErr, 1, true, 0) return record, nil } switch statusPayload.Data.State { case "pending", "running": record.Status = ocrJobStatusRunning - _ = r.saveOCRJobRecord(ctx, jobID, record) + r.saveOCRJobRecordWithLog(ctx, jobID, record) return record, nil case "done": if statusPayload.Data.ResultURL.JSONURL == "" { record.Status = ocrJobStatusFailed record.Error = "未返回识别结果地址" - _ = r.saveOCRJobRecord(ctx, jobID, record) + r.ocrLogger(ctx).Error("PaddleOCR 任务缺少结果地址", + zap.String("operation", "status"), + zap.String("paddle_job_id", record.PaddleJobID), + zap.String("upstream_state", statusPayload.Data.State), + ) + r.saveOCRJobRecordWithLog(ctx, jobID, record) return record, nil } - rawText, err := fetchPaddleOCRText(ctx, statusPayload.Data.ResultURL.JSONURL) + rawText, err := r.fetchPaddleOCRText(ctx, statusPayload.Data.ResultURL.JSONURL) if err != nil { record.Status = ocrJobStatusFailed record.Error = err.Error() - _ = r.saveOCRJobRecord(ctx, jobID, record) + r.saveOCRJobRecordWithLog(ctx, jobID, record) return record, nil } candidates := parseQrCodeGroupNameCandidates(rawText) @@ -260,7 +397,7 @@ func (r *Repository) PollOCRJobResult(ctx context.Context, jobID string) (*OCRJo Candidates: candidates, RawText: rawText, } - _ = r.saveOCRJobRecord(ctx, jobID, record) + r.saveOCRJobRecordWithLog(ctx, jobID, record) return record, nil case "failed": @@ -270,14 +407,36 @@ func (r *Repository) PollOCRJobResult(ctx context.Context, jobID string) (*OCRJo } else { record.Error = "PaddleOCR 识别失败" } - _ = r.saveOCRJobRecord(ctx, jobID, record) + r.ocrLogger(ctx).Error("PaddleOCR 任务执行失败", + zap.String("operation", "status"), + zap.String("paddle_job_id", record.PaddleJobID), + zap.String("upstream_state", statusPayload.Data.State), + zap.String("upstream_message", truncateOCRLogValue(record.Error, 512)), + ) + r.saveOCRJobRecordWithLog(ctx, jobID, record) return record, nil + + default: + upstreamErr := newOCRUpstreamError("status_payload", http.StatusOK, nil, respBody, errors.New("unexpected OCR job state")) + r.logOCRUpstreamFailure(ctx, upstreamErr, 1, true, 0) } return record, nil } -func submitPaddleOCRJob(ctx context.Context, config *QrCodeOCRConfig, filename, contentType string, data []byte) (string, error) { +func (r *Repository) saveOCRJobRecordWithLog(ctx context.Context, jobID string, record *OCRJobRecord) { + if err := r.saveOCRJobRecord(ctx, jobID, record); err != nil { + r.ocrLogger(ctx).Error("OCR 任务状态保存失败", + zap.String("ocr_job_id", jobID), + zap.String("paddle_job_id", record.PaddleJobID), + zap.String("ocr_status", record.Status), + zap.Error(err), + ) + } +} + +func (r *Repository) submitPaddleOCRJob(ctx context.Context, config *QrCodeOCRConfig, filename, contentType string, data []byte) (string, error) { + filename = normalizeOCRFilename(filename) body := &bytes.Buffer{} writer := multipart.NewWriter(body) _ = writer.WriteField("model", config.Model) @@ -301,51 +460,69 @@ func submitPaddleOCRJob(ctx context.Context, config *QrCodeOCRConfig, filename, return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, config.JobURL, body) + requestBody := append([]byte(nil), body.Bytes()...) + formContentType := writer.FormDataContentType() + respBody, err := r.doPaddleRequest(ctx, "submit", func(ctx context.Context) (*http.Request, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, config.JobURL, bytes.NewReader(requestBody)) + if err == nil { + req.Header.Set("Authorization", "bearer "+config.Token) + req.Header.Set("Content-Type", formContentType) + } + return req, err + }) if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - req.Header.Set("Authorization", "bearer "+config.Token) - req.Header.Set("Content-Type", writer.FormDataContentType()) - - resp, err := ocrHTTPClient.Do(req) - if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - defer resp.Body.Close() - respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024)) - if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("%w: status=%d body=%s", ErrQrCodeOCRUnavailable, resp.StatusCode, string(respBody)) + return "", err } var payload paddleOCRJobResponse if err := json.Unmarshal(respBody, &payload); err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) + upstreamErr := newOCRUpstreamError("submit", http.StatusOK, nil, respBody, err) + r.logOCRUpstreamFailure(ctx, upstreamErr, 1, true, 0) + return "", upstreamErr } if payload.Data.JobID == "" { - return "", fmt.Errorf("%w: 未返回 jobId", ErrQrCodeOCRUnavailable) + upstreamErr := newOCRUpstreamError("submit", http.StatusOK, nil, respBody, errors.New("missing jobId")) + r.logOCRUpstreamFailure(ctx, upstreamErr, 1, true, 0) + return "", upstreamErr } + r.ocrLogger(ctx).Info("PaddleOCR 任务提交成功", + zap.String("operation", "submit"), + zap.String("paddle_job_id", payload.Data.JobID), + zap.String("model", config.Model), + zap.String("file_content_type", contentType), + zap.Int("file_size", len(data)), + ) return payload.Data.JobID, nil } -func waitPaddleOCRJob(ctx context.Context, config *QrCodeOCRConfig, jobID string) (string, error) { +func normalizeOCRFilename(filename string) string { + filename = strings.TrimSpace(filename) + if filename == "" { + return "qrcode.jpg" + } + if dot := strings.LastIndex(filename, "."); dot > 0 { + filename = filename[:dot] + } + return filename + ".jpg" +} + +func (r *Repository) waitPaddleOCRJob(ctx context.Context, config *QrCodeOCRConfig, jobID string) (string, error) { for i := 0; i < 12; i++ { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(config.JobURL, "/")+"/"+jobID, nil) + statusURL := strings.TrimRight(config.JobURL, "/") + "/" + jobID + respBody, err := r.doPaddleRequest(ctx, "status", func(ctx context.Context) (*http.Request, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, statusURL, nil) + if err == nil { + req.Header.Set("Authorization", "bearer "+config.Token) + } + return req, err + }) if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - req.Header.Set("Authorization", "bearer "+config.Token) - resp, err := ocrHTTPClient.Do(req) - if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024)) - _ = resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("%w: status=%d body=%s", ErrQrCodeOCRUnavailable, resp.StatusCode, string(respBody)) + return "", err } var payload paddleOCRJobStatusResponse if err := json.Unmarshal(respBody, &payload); err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) + upstreamErr := newOCRUpstreamError("status", http.StatusOK, nil, respBody, err) + r.logOCRUpstreamFailure(ctx, upstreamErr, 1, true, 0) + return "", upstreamErr } switch payload.Data.State { case "done": @@ -359,36 +536,232 @@ func waitPaddleOCRJob(ctx context.Context, config *QrCodeOCRConfig, jobID string } return "", ErrQrCodeOCRUnavailable } - select { - case <-ctx.Done(): - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, ctx.Err()) - case <-time.After(2 * time.Second): + if err := r.ocrSleep(ctx, 2*time.Second); err != nil { + return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) } } return "", fmt.Errorf("%w: 识别超时", ErrQrCodeOCRUnavailable) } -func fetchPaddleOCRText(ctx context.Context, jsonURL string) (string, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, jsonURL, nil) +func (r *Repository) fetchPaddleOCRText(ctx context.Context, jsonURL string) (string, error) { + body, err := r.doPaddleRequest(ctx, "result", func(ctx context.Context) (*http.Request, error) { + return http.NewRequestWithContext(ctx, http.MethodGet, jsonURL, nil) + }) if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - resp, err := ocrHTTPClient.Do(req) - if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024)) - return "", fmt.Errorf("%w: status=%d body=%s", ErrQrCodeOCRUnavailable, resp.StatusCode, string(respBody)) - } - body, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024*1024)) - if err != nil { - return "", fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) + return "", err } return collectPaddleOCRJSONLText(string(body)), nil } +type ocrRequestFactory func(context.Context) (*http.Request, error) + +func (r *Repository) doPaddleRequest(ctx context.Context, operation string, factory ocrRequestFactory) ([]byte, error) { + var lastErr *ocrUpstreamError + for attempt := 1; attempt <= ocrRequestMaxAttempts; attempt++ { + release, err := r.ocrGate.acquire(ctx, r.ocrSleep) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) + } + req, err := factory(ctx) + if err != nil { + release() + return nil, fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) + } + startedAt := time.Now() + resp, requestErr := r.ocrHTTPClient.Do(req) + duration := time.Since(startedAt) + if requestErr != nil { + release() + lastErr = newOCRUpstreamError(operation, 0, nil, nil, requestErr) + lastErr.Retriable = isRetriableOCRNetworkError(ctx, requestErr) + r.logOCRUpstreamFailure(ctx, lastErr, attempt, attempt == ocrRequestMaxAttempts || !lastErr.Retriable, duration) + } else { + bodyLimit := int64(ocrResponseBodyLimit) + if operation == "result" { + bodyLimit = ocrResultBodyLimit + } + body, readErr := io.ReadAll(io.LimitReader(resp.Body, bodyLimit+1)) + _ = resp.Body.Close() + release() + if int64(len(body)) > bodyLimit { + body = body[:bodyLimit] + readErr = errors.New("upstream response exceeded limit") + } + if readErr == nil && resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices { + return body, nil + } + lastErr = newOCRUpstreamError(operation, resp.StatusCode, resp.Header, body, readErr) + lastErr.Retriable = isRetriableOCRStatus(resp.StatusCode) || readErr != nil + r.logOCRUpstreamFailure(ctx, lastErr, attempt, attempt == ocrRequestMaxAttempts || !lastErr.Retriable, duration) + } + + if !lastErr.Retriable || attempt == ocrRequestMaxAttempts { + return nil, lastErr + } + delay := ocrRetryDelay(attempt, lastErr.RetryAfter) + if err := r.ocrSleep(ctx, delay); err != nil { + return nil, fmt.Errorf("%w: %v", ErrQrCodeOCRUnavailable, err) + } + } + return nil, lastErr +} + +func newOCRUpstreamError(operation string, statusCode int, headers http.Header, body []byte, cause error) *ocrUpstreamError { + upstreamErr := &ocrUpstreamError{ + Operation: operation, + StatusCode: statusCode, + BodySnippet: truncateOCRLogValue(string(body), ocrLogBodyLimit), + Cause: cause, + } + if headers != nil { + upstreamErr.RetryAfter = parseOCRRetryAfter(headers.Get("Retry-After"), time.Now()) + upstreamErr.TraceID = firstNonEmpty( + headers.Get("X-Request-ID"), + headers.Get("X-Trace-ID"), + headers.Get("Trace-ID"), + ) + } + var payload map[string]any + if json.Unmarshal(body, &payload) == nil { + upstreamErr.Code = firstJSONText(payload, "code", "errorCode", "error_code") + upstreamErr.Message = firstJSONText(payload, "message", "msg", "error", "errorMsg", "error_message") + if upstreamErr.TraceID == "" { + upstreamErr.TraceID = firstJSONText(payload, "traceId", "trace_id", "requestId", "request_id") + } + } + return upstreamErr +} + +func (r *Repository) logOCRUpstreamFailure(ctx context.Context, upstreamErr *ocrUpstreamError, attempt int, final bool, duration time.Duration) { + if upstreamErr == nil { + return + } + fields := []zap.Field{ + zap.String("operation", upstreamErr.Operation), + zap.Int("attempt", attempt), + zap.Bool("final", final), + zap.Bool("retriable", upstreamErr.Retriable), + zap.Int("upstream_status", upstreamErr.StatusCode), + zap.String("upstream_code", upstreamErr.Code), + zap.String("upstream_message", upstreamErr.Message), + zap.String("upstream_trace_id", upstreamErr.TraceID), + zap.String("upstream_body", upstreamErr.BodySnippet), + zap.Int64("retry_after_ms", upstreamErr.RetryAfter.Milliseconds()), + zap.Float64("duration_ms", float64(duration.Microseconds())/1000), + } + if upstreamErr.Cause != nil { + fields = append(fields, zap.Error(upstreamErr.Cause)) + } + if final { + r.ocrLogger(ctx).Error("PaddleOCR 请求失败", fields...) + } else { + r.ocrLogger(ctx).Warn("PaddleOCR 请求失败，准备重试", fields...) + } +} + +func (r *Repository) ocrLogger(ctx context.Context) *zap.Logger { + logger := r.logger + if logger == nil { + logger = zap.NewNop() + } + fields := []zap.Field{zap.String("module", "chat_ocr")} + if requestID := logging.RequestIDFromContext(ctx); requestID != "" { + fields = append(fields, zap.String("request_id", requestID)) + } + if adminID := logging.AdminIDFromContext(ctx); adminID != 0 { + fields = append(fields, zap.Uint64("admin_id", adminID)) + } + return logger.With(fields...) +} + +func isRetriableOCRStatus(status int) bool { + switch status { + case http.StatusRequestTimeout, http.StatusTooEarly, http.StatusTooManyRequests, + http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable, + http.StatusGatewayTimeout: + return true + default: + return false + } +} + +func isRetriableOCRNetworkError(ctx context.Context, err error) bool { + return err != nil && ctx.Err() == nil +} + +func ocrRetryDelay(attempt int, retryAfter time.Duration) time.Duration { + if retryAfter > 0 { + return min(retryAfter, 10*time.Second) + } + delay := ocrRetryBaseDelay * time.Duration(1<<(attempt-1)) + // Deterministic jitter avoids synchronized retries without relying on global randomness. + jitter := time.Duration((attempt*137)%250) * time.Millisecond + return delay + jitter +} + +func parseOCRRetryAfter(value string, now time.Time) time.Duration { + value = strings.TrimSpace(value) + if value == "" { + return 0 + } + if seconds, err := strconv.Atoi(value); err == nil { + return max(0, time.Duration(seconds)*time.Second) + } + if when, err := http.ParseTime(value); err == nil { + return max(0, when.Sub(now)) + } + return 0 +} + +func sleepWithContext(ctx context.Context, duration time.Duration) error { + if duration <= 0 { + return nil + } + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func firstJSONText(payload map[string]any, keys ...string) string { + for _, key := range keys { + value, ok := payload[key] + if !ok || value == nil { + continue + } + switch typed := value.(type) { + case string: + if text := strings.TrimSpace(typed); text != "" { + return truncateOCRLogValue(text, 512) + } + case float64: + return strconv.FormatFloat(typed, 'f', -1, 64) + } + } + return "" +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if value = strings.TrimSpace(value); value != "" { + return truncateOCRLogValue(value, 256) + } + } + return "" +} + +func truncateOCRLogValue(value string, limit int) string { + value = strings.Join(strings.Fields(value), " ") + if len(value) <= limit { + return value + } + return value[:limit] + "..." +} + func collectPaddleOCRJSONLText(raw string) string { var texts []string for _, line := range strings.Split(raw, "\n") { diff --git a/backend/internal/modules/chat/qrcode_ocr_test.go b/backend/internal/modules/chat/qrcode_ocr_test.go new file mode 100644 index 0000000..75f02f7 --- /dev/null +++ b/backend/internal/modules/chat/qrcode_ocr_test.go @@ -0,0 +1,224 @@ +package chat + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "hash/crc32" + "image" + "image/color" + "image/draw" + "image/png" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" +) + +func newOCRTestRepository(client *http.Client, logger *zap.Logger, sleep func(context.Context, time.Duration) error) *Repository { + return NewRepository(nil, nil, nil, + withOCRHTTPClient(client), + withOCRGate(newOCRRequestGate(ocrRequestConcurrency, 0)), + WithLogger(logger), + withOCRSleep(sleep), + ) +} + +func TestDoPaddleRequestRetries429AndHonorsRetryAfter(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if requests.Add(1) == 1 { + w.Header().Set("Retry-After", "2") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(`{"code":"rate_limit","message":"busy"}`)) + return + } + _, _ = w.Write([]byte("ok")) + })) + t.Cleanup(server.Close) + + var delays []time.Duration + repo := newOCRTestRepository(server.Client(), zap.NewNop(), func(_ context.Context, delay time.Duration) error { + delays = append(delays, delay) + return nil + }) + body, err := repo.doPaddleRequest(t.Context(), "submit", func(ctx context.Context) (*http.Request, error) { + return http.NewRequestWithContext(ctx, http.MethodPost, server.URL, nil) + }) + if err != nil { + t.Fatalf("Paddle 请求失败: %v", err) + } + if string(body) != "ok" || requests.Load() != 2 { + t.Fatalf("body = %q, requests = %d", body, requests.Load()) + } + if len(delays) != 1 || delays[0] != 2*time.Second { + t.Fatalf("retry delays = %v, want [2s]", delays) + } +} + +func TestDoPaddleRequestDoesNotRetryDeterministic4xx(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requests.Add(1) + w.WriteHeader(http.StatusUnsupportedMediaType) + _, _ = w.Write([]byte(`{"code":"invalid_format","message":"unsupported image"}`)) + })) + t.Cleanup(server.Close) + + repo := newOCRTestRepository(server.Client(), zap.NewNop(), func(context.Context, time.Duration) error { return nil }) + _, err := repo.doPaddleRequest(t.Context(), "submit", func(ctx context.Context) (*http.Request, error) { + return http.NewRequestWithContext(ctx, http.MethodPost, server.URL, nil) + }) + var upstreamErr *ocrUpstreamError + if !errors.As(err, &upstreamErr) { + t.Fatalf("error = %v, want ocrUpstreamError", err) + } + if requests.Load() != 1 || upstreamErr.Retriable { + t.Fatalf("requests = %d, retriable = %v", requests.Load(), upstreamErr.Retriable) + } + if upstreamErr.StatusCode != http.StatusUnsupportedMediaType || upstreamErr.Code != "invalid_format" { + t.Fatalf("upstream error = %+v", upstreamErr) + } +} + +func TestDoPaddleRequestLogsStructured502Failure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("X-Trace-ID", "trace-502") + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte(`{"code":"capacity","message":"upstream busy"}`)) + })) + t.Cleanup(server.Close) + core, observed := observer.New(zap.DebugLevel) + repo := newOCRTestRepository(server.Client(), zap.New(core), func(context.Context, time.Duration) error { return nil }) + + _, err := repo.doPaddleRequest(t.Context(), "submit", func(ctx context.Context) (*http.Request, error) { + req, requestErr := http.NewRequestWithContext(ctx, http.MethodPost, server.URL, nil) + if requestErr == nil { + req.Header.Set("Authorization", "bearer secret-token") + } + return req, requestErr + }) + var upstreamErr *ocrUpstreamError + if !errors.As(err, &upstreamErr) { + t.Fatalf("error = %v, want ocrUpstreamError", err) + } + if upstreamErr.StatusCode != http.StatusBadGateway || upstreamErr.TraceID != "trace-502" || upstreamErr.Code != "capacity" { + t.Fatalf("upstream error = %+v", upstreamErr) + } + entries := observed.AllUntimed() + if len(entries) != ocrRequestMaxAttempts { + t.Fatalf("log entries = %d, want %d", len(entries), ocrRequestMaxAttempts) + } + fields := entries[len(entries)-1].ContextMap() + if fields["final"] != true || fields["upstream_status"] != int64(http.StatusBadGateway) || fields["upstream_trace_id"] != "trace-502" { + t.Fatalf("final log fields = %#v", fields) + } + for _, entry := range entries { + if strings.Contains(entry.Message, "secret-token") || strings.Contains(fmt.Sprint(entry.Context), "secret-token") { + t.Fatal("OCR 日志泄露了 Authorization Token") + } + } +} + +func TestReadAndValidateOCRFileNormalizesJPEGAndSize(t *testing.T) { + source := image.NewNRGBA(image.Rect(0, 0, 3000, 1200)) + draw.Draw(source, source.Bounds(), &image.Uniform{C: color.NRGBA{R: 25, G: 80, B: 160, A: 255}}, image.Point{}, draw.Src) + var input bytes.Buffer + if err := png.Encode(&input, source); err != nil { + t.Fatalf("编码测试图片失败: %v", err) + } + + data, contentType, err := readAndValidateOCRFile(&input, "image/png") + if err != nil { + t.Fatalf("归一化图片失败: %v", err) + } + if contentType != "image/jpeg" || len(data) < 2 || data[0] != 0xff || data[1] != 0xd8 { + t.Fatalf("content_type = %q, signature = %x", contentType, data[:min(2, len(data))]) + } + config, format, err := image.DecodeConfig(bytes.NewReader(data)) + if err != nil { + t.Fatalf("读取归一化图片失败: %v", err) + } + if format != "jpeg" || config.Width != 2560 || config.Height != 1024 { + t.Fatalf("format = %q, size = %dx%d", format, config.Width, config.Height) + } +} + +func TestReadAndValidateOCRFileRejectsExcessiveDecodedPixels(t *testing.T) { + data := pngHeader(8001, 5000) + config, _, decodeErr := image.DecodeConfig(bytes.NewReader(data)) + if decodeErr != nil || config.Width != 8001 || config.Height != 5000 { + t.Fatalf("测试 PNG 头无效: config=%+v, error=%v", config, decodeErr) + } + _, _, err := readAndValidateOCRFile(bytes.NewReader(data), "image/png") + if !errors.Is(err, ErrQrCodeOCRInvalidFile) { + t.Fatalf("error = %v, want ErrQrCodeOCRInvalidFile", err) + } +} + +func TestOCRRequestGateLimitsConcurrency(t *testing.T) { + gate := newOCRRequestGate(2, 0) + sleep := func(context.Context, time.Duration) error { return nil } + releaseFirst, err := gate.acquire(t.Context(), sleep) + if err != nil { + t.Fatal(err) + } + releaseSecond, err := gate.acquire(t.Context(), sleep) + if err != nil { + t.Fatal(err) + } + + third := make(chan func(), 1) + go func() { + release, acquireErr := gate.acquire(t.Context(), sleep) + if acquireErr == nil { + third <- release + } + }() + select { + case release := <-third: + release() + t.Fatal("第三个请求在并发槽释放前进入") + case <-time.After(30 * time.Millisecond): + } + releaseFirst() + select { + case release := <-third: + release() + case <-time.After(time.Second): + t.Fatal("并发槽释放后第三个请求仍未进入") + } + releaseSecond() +} + +func TestNormalizeOCRFilenameUsesJPEGExtension(t *testing.T) { + if got := normalizeOCRFilename("group.qrcode.webp"); got != "group.qrcode.jpg" { + t.Fatalf("filename = %q", got) + } + if got := normalizeOCRFilename(""); got != "qrcode.jpg" { + t.Fatalf("empty filename = %q", got) + } +} + +func pngHeader(width, height uint32) []byte { + var result bytes.Buffer + result.Write([]byte{137, 80, 78, 71, 13, 10, 26, 10}) + data := make([]byte, 13) + binary.BigEndian.PutUint32(data[0:4], width) + binary.BigEndian.PutUint32(data[4:8], height) + data[8] = 8 + data[9] = 2 + binary.Write(&result, binary.BigEndian, uint32(len(data))) + result.WriteString("IHDR") + result.Write(data) + checksum := crc32.ChecksumIEEE(append([]byte("IHDR"), data...)) + binary.Write(&result, binary.BigEndian, checksum) + return result.Bytes() +} diff --git a/backend/internal/modules/chat/repository.go b/backend/internal/modules/chat/repository.go index 79bff37..d9891df 100644 --- a/backend/internal/modules/chat/repository.go +++ b/backend/internal/modules/chat/repository.go @@ -1,27 +1,81 @@ package chat import ( + "context" "errors" + "net/http" + "time" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" "gorm.io/gorm" "gorm.io/gorm/clause" "hfb_sys/backend/internal/model" "hfb_sys/backend/internal/modules/chathub" - "time" ) type Repository struct { - db *gorm.DB - hub *chathub.Hub - redis *redis.Client + db *gorm.DB + hub *chathub.Hub + redis *redis.Client + logger *zap.Logger + ocrHTTPClient *http.Client + ocrGate *ocrRequestGate + ocrSleep func(context.Context, time.Duration) error } const ( defaultSupportRoleCode = "cs" ) -func NewRepository(db *gorm.DB, hub *chathub.Hub, redis *redis.Client) *Repository { - return &Repository{db: db, hub: hub, redis: redis} +type RepositoryOption func(*Repository) + +func WithLogger(logger *zap.Logger) RepositoryOption { + return func(repo *Repository) { + if logger != nil { + repo.logger = logger + } + } +} + +func withOCRHTTPClient(client *http.Client) RepositoryOption { + return func(repo *Repository) { + if client != nil { + repo.ocrHTTPClient = client + } + } +} + +func withOCRGate(gate *ocrRequestGate) RepositoryOption { + return func(repo *Repository) { + if gate != nil { + repo.ocrGate = gate + } + } +} + +func withOCRSleep(sleep func(context.Context, time.Duration) error) RepositoryOption { + return func(repo *Repository) { + if sleep != nil { + repo.ocrSleep = sleep + } + } +} + +func NewRepository(db *gorm.DB, hub *chathub.Hub, redis *redis.Client, options ...RepositoryOption) *Repository { + repo := &Repository{ + db: db, + hub: hub, + redis: redis, + logger: zap.NewNop(), + ocrHTTPClient: ocrHTTPClient, + ocrGate: sharedOCRRequestGate, + ocrSleep: sleepWithContext, + } + for _, option := range options { + option(repo) + } + return repo } func EnsureOrderConversation(tx *gorm.DB, order model.RentalOrder) (*model.ChatConversation, error) { var existing model.ChatConversation diff --git a/backend/internal/router/router.go b/backend/internal/router/router.go index 94166b7..3edea5a 100644 --- a/backend/internal/router/router.go +++ b/backend/internal/router/router.go @@ -177,7 +177,7 @@ func New(cfg config.Config, deps Dependencies, logger *zap.Logger) *gin.Engine { } var chatRepo *chat.Repository if deps.DB != nil { - chatRepo = chat.NewRepository(deps.DB, chatHub, deps.Redis) + chatRepo = chat.NewRepository(deps.DB, chatHub, deps.Redis, chat.WithLogger(logger)) } // 创建 chat 适配器用于 listing diff --git a/frontend/src/features/admin/views/AdminQrCodePoolView.vue b/frontend/src/features/admin/views/AdminQrCodePoolView.vue index 249340a..d53a026 100644 --- a/frontend/src/features/admin/views/AdminQrCodePoolView.vue +++ b/frontend/src/features/admin/views/AdminQrCodePoolView.vue @@ -32,6 +32,7 @@ import { formatDateTime } from '@/shared/utils/time' import AdminTablePagination from '../components/AdminTablePagination.vue' import AuthImage from '@/shared/components/business/AuthImage.vue' import { readError } from '@/shared/utils/error' +import { optimizeImageForUpload } from '@/shared/utils/imageUpload' import { useAdminSessionStore } from '@/stores/adminSession' const adminSession = useAdminSessionStore() @@ -243,34 +244,33 @@ function parseGroupNameFromOcrText(text: string) { ) } -const ocrConcurrencyLimit = 4 +const ocrConcurrencyLimit = 2 let ocrRunningCount = 0 -const ocrWaitQueue: (() => void)[] = [] +class OCRQueueResetError extends Error {} +const ocrWaitQueue: { resolve: () => void; reject: (error: Error) => void }[] = [] function ocrAcquireSlot(): Promise { if (ocrRunningCount < ocrConcurrencyLimit) { ocrRunningCount++ return Promise.resolve() } - return new Promise(resolve => { - ocrWaitQueue.push(() => { - ocrRunningCount++ - resolve() - }) + return new Promise((resolve, reject) => { + ocrWaitQueue.push({ resolve, reject }) }) } function ocrReleaseSlot() { ocrRunningCount = Math.max(0, ocrRunningCount - 1) const next = ocrWaitQueue.shift() - if (next) next() + if (next) { + ocrRunningCount++ + next.resolve() + } } function ocrResetSlots() { - ocrRunningCount = 0 - while (ocrWaitQueue.length > 0) { - const next = ocrWaitQueue.shift() - next?.() + for (const waiter of ocrWaitQueue.splice(0)) { + waiter.reject(new OCRQueueResetError('OCR queue reset')) } } @@ -322,9 +322,11 @@ async function handleFileUpload(options: { file: File }) { } uploadedImages.value.push(item) + const normalizedFile = await optimizeImageForUpload(options.file, 'qrcode') const [uploaded, groupName] = await Promise.all([ - uploadAdminFile(options.file, 'qrcode'), - recognizeGroupNameWithLimit(options.file).catch(error => { + uploadAdminFile(normalizedFile, 'qrcode', { optimize: false }), + recognizeGroupNameWithLimit(normalizedFile).catch(error => { + if (error instanceof OCRQueueResetError) return '' console.warn('二维码群名 OCR 失败', error) ElMessage.warning(readError(error, '二维码群名 OCR 失败')) return '' diff --git a/frontend/src/shared/api/files.ts b/frontend/src/shared/api/files.ts index 45404dc..71d2f0e 100644 --- a/frontend/src/shared/api/files.ts +++ b/frontend/src/shared/api/files.ts @@ -23,8 +23,12 @@ export async function uploadFile(file: File, scene: string) { return data.data } -export async function uploadAdminFile(file: File, scene: string) { - const uploadTarget = await optimizeImageForUpload(file, scene) +export async function uploadAdminFile( + file: File, + scene: string, + options: { optimize?: boolean } = {} +) { + const uploadTarget = options.optimize === false ? file : await optimizeImageForUpload(file, scene) const form = new FormData() form.append('file', uploadTarget) form.append('scene', scene)