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 }