HandleMessage 处理从 NSQ 读取的消息
(message *nsq.Message)
| 86 | |
| 87 | // HandleMessage 处理从 NSQ 读取的消息 |
| 88 | func (h *NSQHandler) HandleMessage(message *nsq.Message) error { |
| 89 | log.Printf("Received message: %s", string(message.Body)) |
| 90 | |
| 91 | // 解析消息为结构体 |
| 92 | var data NSQMessage |
| 93 | err := json.Unmarshal(message.Body, &data) |
| 94 | if err != nil { |
| 95 | log.Printf("Failed to unmarshal message: %v", err) |
| 96 | return nil |
| 97 | } |
| 98 | |
| 99 | // 将时间字符串转换为 time.Time |
| 100 | timestamp, err := time.Parse(time.RFC3339, data.TimeISO8601) |
| 101 | if err != nil { |
| 102 | log.Printf("Failed to parse timestamp: %v", err) |
| 103 | return nil |
| 104 | } |
| 105 | |
| 106 | day := time.Date(timestamp.Year(), timestamp.Month(), timestamp.Day(), 0, 0, 0, 0, timestamp.Location()) |
| 107 | hour := time.Date(timestamp.Year(), timestamp.Month(), timestamp.Day(), timestamp.Hour(), 0, 0, 0, timestamp.Location()) |
| 108 | minute := time.Date(timestamp.Year(), timestamp.Month(), timestamp.Day(), timestamp.Hour(), timestamp.Minute(), 0, 0, timestamp.Location()) |
| 109 | return h.transaction.Transaction(context.Background(), func(ctx context.Context) error { |
| 110 | finalStatus := &AIProviderStatus{} |
| 111 | for _, s := range data.AI.ProviderStats { |
| 112 | status := ToKeyStatus(s.Status).Int() |
| 113 | key := genAIKey(s.Key, s.Provider) |
| 114 | err = h.aiKeyService.Save(ctx, key, &ai_key.Edit{ |
| 115 | Status: &status, |
| 116 | }) |
| 117 | if err != nil { |
| 118 | log.Printf("Failed to save AI key: %v", err) |
| 119 | return nil |
| 120 | } |
| 121 | if s.Provider != data.AI.Provider { |
| 122 | |
| 123 | pStatus := ai_dto.ProviderAbnormal.Int() |
| 124 | err = h.aiService.Save(ctx, s.Provider, &ai.SetProvider{ |
| 125 | Status: &pStatus, |
| 126 | }) |
| 127 | } else { |
| 128 | pStatus := ai_dto.ProviderEnabled.Int() |
| 129 | err = h.aiService.Save(ctx, s.Provider, &ai.SetProvider{ |
| 130 | Status: &pStatus, |
| 131 | }) |
| 132 | } |
| 133 | finalStatus = &s |
| 134 | } |
| 135 | if finalStatus != nil { |
| 136 | //keys := strings.Split(finalStatus.Key, "@") |
| 137 | key := genAIKey(finalStatus.Key, finalStatus.Provider) |
| 138 | err = h.aiKeyService.IncrUseToken(ctx, key, convertInt(data.AI.TotalToken)) |
| 139 | if err != nil { |
| 140 | log.Printf("Failed to increment AI key token: %v", err) |
| 141 | return nil |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | // 调用 AI API 接口 |
nothing calls this directly
no test coverage detected