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"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/hibiken/asynq"
	"go.uber.org/zap"
)

const taskTypeSimCardDailyReport = "SimCard:DailyReport"

const (
	simCardHeavyUsageKB      = 100 * 1024 // 进入日报清单的日用量阈值（严格大于）
	simCardHeavyUsageBoldKB  = 500 * 1024 // 清单中加粗显示的日用量阈值（严格大于）
	simCardHeavyUsageMaxRows = 30         // 清单最多列出的行数，避免超过钉钉消息长度上限
)

// simCardHeavyUsage 是日用量超过阈值的一张卡；SN 为空表示没有关联站点。
type simCardHeavyUsage struct {
	ICCID       string
	IncrementKB float64
	SN          string
	Version     string
}

type simCardDailyReport struct {
	Stats            domain.SimCardDailyStatistics
	HeavyCount       int                 // 超过阈值的卡总数
	Heavy            []simCardHeavyUsage // 按日用量降序，最多 simCardHeavyUsageMaxRows 条
	SiteLookupFailed bool
}

func calculateSimCardStatistics(date time.Time) (domain.SimCardDailyStatistics, []simCardHeavyUsage, error) {
	stats := domain.SimCardDailyStatistics{Date: date.Format(time.DateOnly)}
	var heavy []simCardHeavyUsage
	err := mongo.StreamSimCardDailyUsage(date, func(simId string, total float64, previous *float64) error {
		if increment, ok := accumulateSimCardStatistics(&stats, total, previous); ok && increment > simCardHeavyUsageKB {
			heavy = append(heavy, simCardHeavyUsage{ICCID: simId, IncrementKB: increment})
		}
		return nil
	})
	return stats, heavy, err
}

// buildSimCardDailyReport 排序截断清单并补充站点信息；站点查询失败不影响统计部分。
func buildSimCardDailyReport(stats domain.SimCardDailyStatistics, heavy []simCardHeavyUsage,
	lookup func(giccids []string) ([]domain.SiteInfoDomain, error)) simCardDailyReport {
	sort.SliceStable(heavy, func(i, j int) bool { return heavy[i].IncrementKB > heavy[j].IncrementKB })
	report := simCardDailyReport{Stats: stats, HeavyCount: len(heavy)}
	if len(heavy) > simCardHeavyUsageMaxRows {
		heavy = heavy[:simCardHeavyUsageMaxRows]
	}
	report.Heavy = heavy
	if len(heavy) == 0 {
		return report
	}
	ids := make([]string, 0, len(heavy))
	for _, h := range heavy {
		ids = append(ids, h.ICCID)
	}
	sites, err := lookup(ids)
	if err != nil {
		zap.L().Error("SIM card daily report: site lookup failed", zap.String("date", stats.Date), zap.Error(err))
		report.SiteLookupFailed = true
		return report
	}
	for i := range report.Heavy {
		var candidates []domain.SiteInfoDomain
		for _, site := range sites {
			if report.Heavy[i].ICCID != "" && strings.Contains(strings.ToUpper(site.ICCID), strings.ToUpper(report.Heavy[i].ICCID)) {
				candidates = append(candidates, site)
			}
		}
		if site := pickSimCardSite(candidates); site != nil {
			report.Heavy[i].SN = site.Sn
			report.Heavy[i].Version = site.Version
		}
	}
	return report
}

// pickSimCardSite 在一张卡关联的多个站点中选一个：优先在线站点（多个在线取最后上线最晚的），
// 都不在线时取最后离线时间最接近现在的。时间格式为 YYYY-MM-DD HH:mm:SS，可直接按字符串比较。
func pickSimCardSite(sites []domain.SiteInfoDomain) *domain.SiteInfoDomain {
	var best *domain.SiteInfoDomain
	for i := range sites {
		site := &sites[i]
		if best == nil || simCardSiteBetter(site, best) {
			best = site
		}
	}
	return best
}

func simCardSiteBetter(a, b *domain.SiteInfoDomain) bool {
	aOnline, bOnline := a.Status == 1, b.Status == 1
	if aOnline != bOnline {
		return aOnline
	}
	if aOnline {
		return derefString(a.OnLineTime) > derefString(b.OnLineTime)
	}
	return derefString(a.OffLineTime) > derefString(b.OffLineTime)
}

func derefString(s *string) string {
	if s == nil {
		return ""
	}
	return *s
}

// accumulateSimCardStatistics 返回该卡的日增量；没有前一日数据时 ok 为 false。
func accumulateSimCardStatistics(stats *domain.SimCardDailyStatistics, total float64, previous *float64) (increment float64, ok bool) {
	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 0, false
	}
	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++
	}
	return increment, true
}

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, heavy, err := calculateSimCardStatistics(date)
	if err != nil {
		return err
	}
	if err := mongo.SaveSimCardDailyStatistics(stats); err != nil {
		return err
	}
	report := buildSimCardDailyReport(stats, heavy, mongo.GetSiteInfoListFromGiccids)
	if err := sendSimCardStatistics(ctx, processor.webhook, processor.secret, report); err != nil {
		return err
	}
	return mongo.MarkSimCardReportSent(dateStr)
}

func formatSimCardDailyReport(report simCardDailyReport) string {
	var b strings.Builder
	b.WriteString(formatSimCardStatistics(report.Stats))
	fmt.Fprintf(&b, "\n\n日用量超过 %d MB 的站点（%d 张卡，SN · 日用量 · 固件版本）：\n\n",
		simCardHeavyUsageKB/1024, report.HeavyCount)
	if report.HeavyCount == 0 {
		b.WriteString("- 无")
		return b.String()
	}
	if report.SiteLookupFailed {
		b.WriteString("> 站点信息查询失败，以下仅列出 ICCID\n\n")
	}
	for i, h := range report.Heavy {
		name, version := h.SN, h.Version
		if name == "" {
			name = "ICCID " + h.ICCID
			if !report.SiteLookupFailed {
				name += "（未绑定站点）"
			}
		}
		if version == "" {
			version = "-"
		}
		line := fmt.Sprintf("%s · %.2f MB · %s", name, h.IncrementKB/1024, version)
		if h.IncrementKB > simCardHeavyUsageBoldKB {
			line = "**" + line + "**"
		}
		if i > 0 {
			b.WriteString("\n")
		}
		b.WriteString("- " + line)
	}
	if rest := report.HeavyCount - len(report.Heavy); rest > 0 {
		fmt.Fprintf(&b, "\n- 其余 %d 张省略", rest)
	}
	return b.String()
}

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, report simCardDailyReport) error {
	return postSimCardStatistics(ctx, &http.Client{Timeout: 10 * time.Second}, webhook, secret, report)
}

func postSimCardStatistics(ctx context.Context, client *http.Client, webhook, secret string, report simCardDailyReport) 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 卡流量日报 " + report.Stats.Date,
		"text":  formatSimCardDailyReport(report),
	}}
	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
}