package simCardQueue import ( "bytes" "context" "crypto/hmac" "crypto/sha256" "encoding/base64" "encoding/json" "fmt" "fwh-toolbox-server/domain" "fwh-toolbox-server/repositories/mongo" "net/http" "net/url" "strconv" "time" "github.com/hibiken/asynq" ) const taskTypeSimCardDailyReport = "SimCard:DailyReport" func calculateSimCardStatistics(date time.Time) (domain.SimCardDailyStatistics, error) { stats := domain.SimCardDailyStatistics{Date: date.Format(time.DateOnly)} err := mongo.StreamSimCardDailyUsage(date, func(total float64, previous *float64) error { accumulateSimCardStatistics(&stats, total, previous) return nil }) return stats, err } func accumulateSimCardStatistics(stats *domain.SimCardDailyStatistics, total float64, previous *float64) { stats.CardCount++ if total > stats.MaxTotalKB { stats.MaxTotalKB = total } if total > 200*1024 { stats.TotalOver200MB++ } if total > 1024*1024 { stats.TotalOver1GB++ } if total > 2*1024*1024 { stats.TotalOver2GB++ } if previous == nil { return } increment := total - *previous // Monthly counters can reset at a cycle boundary. The new counter then // represents the usage accumulated since the reset. if increment < 0 { increment = total } stats.IncrementCardCount++ if stats.MaxIncrementKB == nil || increment > *stats.MaxIncrementKB { stats.MaxIncrementKB = &increment } if increment > 20*1024 { stats.IncrementOver20MB++ } if increment > 50*1024 { stats.IncrementOver50MB++ } if increment > 200*1024 { stats.IncrementOver200MB++ } } func (processor *SimCardProcessor) enqueueSimCardReportIfComplete(date string) error { if processor.webhook == "" { return nil } complete, err := mongo.IsSimCardRunComplete(date) if err != nil || !complete { return err } _, err = TaskClient.Enqueue(asynq.NewTask(taskTypeSimCardDailyReport, []byte(date), asynq.Queue("default"), asynq.MaxRetry(10), asynq.Timeout(3*time.Minute), asynq.TaskID("sim-card-daily-report-"+date), asynq.Retention(7*24*time.Hour))) if err == asynq.ErrTaskIDConflict { return nil } return err } func (processor *SimCardProcessor) handleSimCardDailyReport(ctx context.Context, task *asynq.Task) error { if processor.webhook == "" { return nil } dateStr := string(task.Payload()) date, err := time.Parse(time.DateOnly, dateStr) if err != nil { return err } complete, err := mongo.IsSimCardRunComplete(dateStr) if err != nil || !complete { return err } stats, err := calculateSimCardStatistics(date) if err != nil { return err } if err := mongo.SaveSimCardDailyStatistics(stats); err != nil { return err } if err := sendSimCardStatistics(ctx, processor.webhook, processor.secret, stats); err != nil { return err } return mongo.MarkSimCardReportSent(dateStr) } func formatSimCardStatistics(stats domain.SimCardDailyStatistics) string { maxIncrement := "无可比数据" if stats.MaxIncrementKB != nil { maxIncrement = fmt.Sprintf("%.2f MB", *stats.MaxIncrementKB/1024) } return fmt.Sprintf("### SIM 卡流量日报(%s UTC)\n\n"+ "总流量(%d 张卡,月累计):\n\n"+ "- 超过 200 MB:%d 张\n- 超过 1 GB:%d 张\n- 超过 2 GB:%d 张\n- 最大总量:%.2f MB\n\n"+ "增量流量(%d 张卡有前一日数据):\n\n"+ "- 超过 20 MB:%d 张\n- 超过 50 MB:%d 张\n- 超过 200 MB:%d 张\n- 最大增量:%s", stats.Date, stats.CardCount, stats.TotalOver200MB, stats.TotalOver1GB, stats.TotalOver2GB, stats.MaxTotalKB/1024, stats.IncrementCardCount, stats.IncrementOver20MB, stats.IncrementOver50MB, stats.IncrementOver200MB, maxIncrement) } func sendSimCardStatistics(ctx context.Context, webhook, secret string, stats domain.SimCardDailyStatistics) error { return postSimCardStatistics(ctx, &http.Client{Timeout: 10 * time.Second}, webhook, secret, stats) } func postSimCardStatistics(ctx context.Context, client *http.Client, webhook, secret string, stats domain.SimCardDailyStatistics) error { if webhook == "" { return fmt.Errorf("SIM_CARD_DINGTALK_WEBHOOK is not configured") } u, err := url.Parse(webhook) if err != nil || u.Scheme != "https" || u.Host == "" { return fmt.Errorf("SIM_CARD_DINGTALK_WEBHOOK must be an HTTPS URL") } if secret != "" { timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10) mac := hmac.New(sha256.New, []byte(secret)) _, _ = mac.Write([]byte(timestamp + "\n" + secret)) query := u.Query() query.Set("timestamp", timestamp) query.Set("sign", base64.StdEncoding.EncodeToString(mac.Sum(nil))) u.RawQuery = query.Encode() } message := map[string]any{"msgtype": "markdown", "markdown": map[string]string{ "title": "SIM 卡流量日报 " + stats.Date, "text": formatSimCardStatistics(stats), }} body, err := json.Marshal(message) if err != nil { return err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := client.Do(req) if err != nil { return fmt.Errorf("send SIM card statistics to DingTalk: %w", err) } defer resp.Body.Close() var result struct { ErrCode *int `json:"errcode"` ErrMsg string `json:"errmsg"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return err } if resp.StatusCode != http.StatusOK || result.ErrCode == nil || *result.ErrCode != 0 { return fmt.Errorf("DingTalk rejected SIM card statistics: HTTP %d, code %v, %s", resp.StatusCode, result.ErrCode, result.ErrMsg) } return nil }