FranklinWH-Toolbox/service/task/queue/simCardQueue/simCardStatistics.go

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
}