package simcardquery import ( "context" "fmt" "sort" "strings" "sync" "time" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/dao" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/exportcenter" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/tdengine" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/xgorm/handler" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model" "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gtime" ) // usageHistoryMonths 近一年月流水的月份个数(不含当月)。 // 例:当前 2026-10,导出 2026-09 ~ 2025-10 共 12 个月。 const usageHistoryMonths = 12 // usageHistoryTDEngineChunk 单次向 TDengine 查询的 ICCID 条数。 // 父表按 iccid 标签过滤会剪枝到对应子表,500 条一批可把单次查询控制在秒级。 const usageHistoryTDEngineChunk = 500 // dept 模式没有输入号码,靠公司范围圈卡,必须设上限,否则顶级账号一次拖全库。 // deptExportMaxRows 公司模式(含下级)单次导出上限。 // // 与号码模式的 20000 分开取值:号码模式受「用户手上有多少号码」约束, // 公司模式是整批圈卡,按 2 万设限只能靠不断切分下级公司,反而更容易出错。 // // 取值依据(实测,见 export_cost_test.go):5 万行写 Excel 约 3.2s、文件 4.5MB, // 远低于导出中心 10 分钟超时 / 100MB 上限;真正的约束是当前实现会把整批行 // 先构造成内存切片再分页写出,5 万行内存占用约 50MB 量级,10 万行开始偏高。 // 若要支持 10 万以上,需要把公司模式改成「按页直接查库」的流式导出。 const deptExportMaxRows = 50000 var usageHistoryFixedFields = []string{"iccid", "msisdn", "imsi", "outStockTime", "expireAt"} // 月用量列的取值口径: // // 查询失败(服务异常)→ 留空,原因写进「失败原因」列 // 查到了但该月无记录 → "-" // 该月有记录且用量为 0 → 0 // // 失败格留空而不是写文字:这样整列是「数字 + 空 + 横杠」,Excel 求和/透视不会被文本干扰; // 同时「失败原因」列会写明该卡流量查询失败,信息不会丢。 const usageNoDataText = "-" // usageQueryFailedNote 流量查询失败时写入「失败原因」列的说明。 const usageQueryFailedNote = "该卡近12个月流量数据本次查询失败,月份列留空,可稍后重新导出" // cardUsageResult 单张卡的历史用量查询结果。 type cardUsageResult struct { // Months 键为 "YYYY-MM";缺失表示查询成功但该月无记录。 Months map[string]float64 // Failed 表示该卡所属批次的流量查询失败,数据不可信。 Failed bool } // ExportUsageHistory 月历史用量查询导出。 func (s *Service) ExportUsageHistory(ctx context.Context, in *model.SimCardUsageHistoryExportParam, input ExcelInput) (*exportcenter.SubmitResult, error) { months := LastCompleteMonths(time.Now(), usageHistoryMonths) columns := resolveUsageHistoryColumns(in, months) // 公司模式的卡列表必须在后台任务中解析,避免提交请求同步读取最多 2 万张卡。 // 号码模式仍在提交阶段校验输入格式,实际查库统一延迟到导出 runner。 var lines []inputLine var err error if in.Mode != model.SimCardQueryModeDept { lines, err = parseInputs(ctx, &in.SimCardQueryExportParam, input) if err != nil { return nil, err } } stats := &resolveStats{} once := &sync.Once{} var cached []*model.SimCardUsageHistoryRow var cacheErr error resolve := func(ctx context.Context) ([]*model.SimCardUsageHistoryRow, error) { once.Do(func() { switch in.Mode { case model.SimCardQueryModeDept: cached, cacheErr = s.resolveDeptRows(ctx, in, months, stats) case model.SimCardQueryModeRange: cached, cacheErr = s.resolveUsageRangeRows(ctx, &in.SimCardQueryExportParam, months, stats) default: cached, cacheErr = s.resolveUsageHistoryRows(ctx, lines, months, stats) } }) return cached, cacheErr } probeTotal := len(lines) switch in.Mode { case model.SimCardQueryModeRange: // 号段模式:号段包含多少个号码就是预计条数,与是否查到无关。 if size, sizeErr := rangeSize(&in.SimCardQueryExportParam); sizeErr == nil { probeTotal = int(size) } else { probeTotal = -1 } case model.SimCardQueryModeDept: // 公司卡数只在后台 runner 中查询;提交探针不扫描业务表,条数未知。 probeTotal = -1 } return exportcenter.Submit(ctx, &exportcenter.Request{ BizKey: "tSimCardUsageHistory", BizName: "月历史用量查询", PermCode: "cpm:iot:tSimCardUsageHistory:view", SheetName: "月历史用量", QueryParam: usageHistorySnapshot(in, months, probeTotal), Columns: columns, RowFields: append(append([]string{}, usageHistoryFixedFields...), append(months, queryResultField, failReasonField)...), Tags: func() ([]string, error) { return usageHistoryTags(months), nil }, ToRow: func(v interface{}) []interface{} { row, ok := v.(*model.SimCardUsageHistoryRow) if !ok || row == nil { return nil } values := make([]interface{}, 0, len(usageHistoryFixedFields)+len(months)+2) values = append(values, row.Iccid, row.Msisdn, row.Imsi, formatTime(row.OutStockTime), formatTime(row.ExpireAt)) values = append(values, formatMonthValues(row, months)...) return append(values, row.QueryResult, row.FailReason) }, Fetch: func(ctx context.Context, page, pageSize int) ([]interface{}, int, error) { if pageSize <= 1 && page <= 1 { // probeTotal 为 -1 表示提交阶段无法确定条数(按公司模式),放行由后台实际取数。 return nil, probeTotal, nil } rows, err := resolve(ctx) if err != nil { return nil, 0, err } total := len(rows) start := (page - 1) * pageSize if start >= total { return nil, total, nil } end := start + pageSize if end > total { end = total } out := make([]interface{}, 0, end-start) for _, row := range rows[start:end] { out = append(out, row) } return out, total, nil }, Result: stats.resultSummary, }) } // resolveUsageHistoryRows 号码模式:解析输入 → 查卡 → 查近 12 个月用量。 func (s *Service) resolveUsageHistoryRows(ctx context.Context, lines []inputLine, months []string, stats *resolveStats) ([]*model.SimCardUsageHistoryRow, error) { values := collectValues(lines) cards, err := queryCardsByValues(ctx, values) if err != nil { return nil, err } byIccid := indexCards(cards, func(card *cardBrief) string { return card.Iccid }) byMsisdn := indexCards(cards, func(card *cardBrief) string { return card.Msisdn }) byImsi := indexCards(cards, func(card *cardBrief) string { return card.Imsi }) iccids := make([]string, 0, len(cards)) for _, card := range cards { if card.Iccid != "" { iccids = append(iccids, card.Iccid) } } usage, err := queryMonthlyUsage(ctx, iccids, months) if err != nil { return nil, err } rows := make([]*model.SimCardUsageHistoryRow, 0, len(lines)) // 同一张卡重复出现只导出一次(同号码重复填写,或 ICCID 与其 MSISDN 各写一行)。 exported := make(map[int64]struct{}, len(lines)) for _, line := range lines { row := &model.SimCardUsageHistoryRow{MonthValues: map[string]float64{}} card, reason := matchCard(line, byIccid, byMsisdn, byImsi) if card == nil { row.Iccid = firstValue(line.Values) row.QueryResult = queryResultMiss row.FailReason = inputFailure(line, reason) stats.add(false) rows = append(rows, row) continue } if _, ok := exported[card.Id]; ok { stats.addDup() continue } exported[card.Id] = struct{}{} rows = append(rows, newUsageRow(card, usage)) stats.add(true) } return rows, nil } // resolveUsageRangeRows 号段模式:按起止号段取卡,再补近 12 个月用量。 func (s *Service) resolveUsageRangeRows(ctx context.Context, in *model.SimCardQueryExportParam, months []string, stats *resolveStats) ([]*model.SimCardUsageHistoryRow, error) { cards, err := queryCardsByRange(ctx, in) if err != nil { return nil, err } iccids := make([]string, 0, len(cards)) for _, card := range cards { if card.Iccid != "" { iccids = append(iccids, card.Iccid) } } usage, err := queryMonthlyUsage(ctx, iccids, months) if err != nil { return nil, err } // 按号段逐个号码输出:命中出行数据,未命中出「未查询到数据」的失败行, // 保证导出条数与号段大小一致(号段里可能一部分可见、一部分不可见)。 dimension := strings.TrimSpace(in.Dimension) index := rangeCardIndex(cards, dimension) size, sizeErr := rangeSize(in) if sizeErr != nil { size = int64(len(cards)) } numbers := rangeNumbers(in, size) rows := make([]*model.SimCardUsageHistoryRow, 0, len(numbers)) for _, number := range numbers { if card, ok := index[number]; ok { rows = append(rows, newUsageRow(card, usage)) stats.add(true) continue } row := &model.SimCardUsageHistoryRow{ MonthValues: map[string]float64{}, QueryResult: queryResultMiss, FailReason: rangeMissReason, } switch dimension { case "msisdn": row.Msisdn = number case "imsi": row.Imsi = number default: row.Iccid = number } stats.add(false) rows = append(rows, row) } return rows, nil } // resolveDeptRows 在后台按提交人的数据权限重新解析公司(含下级)卡列表。 func (s *Service) resolveDeptRows(ctx context.Context, in *model.SimCardUsageHistoryExportParam, months []string, stats *resolveStats) ([]*model.SimCardUsageHistoryRow, error) { iccids, err := s.cardsByDept(ctx, in) if err != nil { return nil, err } cards, err := queryCardsByIccids(ctx, iccids) if err != nil { return nil, err } usage, err := queryMonthlyUsage(ctx, iccids, months) if err != nil { return nil, err } rows := make([]*model.SimCardUsageHistoryRow, 0, len(cards)) for _, card := range cards { stats.add(true) rows = append(rows, newUsageRow(card, usage)) } return rows, nil } // cardsByDept 取公司(含全部下级公司)名下的 ICCID 列表。 // // 选中公司即包含下级,与 DeptOptions 的卡数统计口径一致: // 顶级公司自身通常不挂卡,只查本级会得到 0 条。 func (s *Service) cardsByDept(ctx context.Context, in *model.SimCardUsageHistoryExportParam) ([]string, error) { if in.DeptId <= 0 { return nil, gerror.New("请选择所属公司") } columns := dao.TSimCardInfo.Columns() base := func() *gdb.Model { mod := handler.Model(dao.TSimCardInfo.Ctx(ctx), &handler.Option{ FilterTenant: true, FilterAuth: true, }).WhereIn(columns.DeptId, deptSubtreeIDs(ctx, in.DeptId)) if start := strings.TrimSpace(in.ExpireStartTime); start != "" { mod = mod.WhereGTE(columns.FlowPkgEndAt, start) } if end := strings.TrimSpace(in.ExpireEndTime); end != "" { mod = mod.WhereLTE(columns.FlowPkgEndAt, end) } return mod } count, err := base().Fields(columns.Iccid).Count() if err != nil { return nil, gerror.Wrap(err, "统计公司名下卡数量失败") } if count <= 0 { return nil, gerror.New("该公司(含下级)下没有符合条件可导出的卡") } if count > deptExportMaxRows { return nil, gerror.Newf( "该公司(含下级)共 %d 张卡,超过单次导出上限 %d 张,请改选下级公司或改用号码导入", count, deptExportMaxRows, ) } values, err := base().Fields(columns.Iccid).OrderAsc(columns.Id).Limit(deptExportMaxRows).Array() if err != nil { return nil, gerror.Wrap(err, "查询公司名下卡失败") } out := make([]string, 0, len(values)) for _, value := range values { if iccid := strings.TrimSpace(value.String()); iccid != "" { out = append(out, iccid) } } return out, nil } // queryCardsByIccids 按 ICCID 批量取导出所需字段。 func queryCardsByIccids(ctx context.Context, iccids []string) ([]*cardBrief, error) { if len(iccids) == 0 { return nil, nil } columns := dao.TSimCardInfo.Columns() out := make([]*cardBrief, 0, len(iccids)) for start := 0; start < len(iccids); start += chunkSize { end := start + chunkSize if end > len(iccids) { end = len(iccids) } list := make([]*cardBrief, 0, end-start) err := handler.Model(dao.TSimCardInfo.Ctx(ctx), &handler.Option{ FilterTenant: true, FilterAuth: true, }). Fields( columns.Id, columns.Iccid, columns.Msisdn, columns.Imsi, columns.OutStockTime, columns.FlowPkgEndAt, ). WhereIn(columns.Iccid, iccids[start:end]). Scan(&list) if err != nil { return nil, gerror.Wrap(err, "查询卡信息失败,请稍后重试") } out = append(out, list...) } return out, nil } // queryMonthlyUsage 从 TDengine 取多卡的多月用量。 // // 每张卡带上自己的查询状态,三种情况必须在导出里可区分: // - Failed=true:该批查询异常,月份格写「查询失败」 // - Months 里缺该月:查询成功但没有该月记录,月份格写 "-" // - Months 里有该月且为 0:查询成功、零用量,月份格写 0 // // 部分批次失败不拖垮整单,只有全部批次失败(服务整体不可用)才返回错误。 func queryMonthlyUsage(ctx context.Context, iccids []string, months []string) (map[string]*cardUsageResult, error) { result := make(map[string]*cardUsageResult, len(iccids)) if len(iccids) == 0 || len(months) == 0 { return result, nil } if !tdengine.Enabled(ctx) { g.Log().Warning(ctx, "[simcardquery] 流量库未启用") return nil, gerror.New("流量数据服务暂时不可用,本次导出未能完成。请稍后重试,若持续出现可联系技术支持。") } cfg := tdengine.GetConfig(ctx) if err := validateTDEngineNames(cfg.DbName, cfg.MonthlyStable); err != nil { g.Log().Errorf(ctx, "[simcardquery] 流量库配置异常: %v", err) return nil, gerror.New("流量数据服务暂时不可用,本次导出未能完成。请稍后重试,若持续出现可联系技术支持。") } start, end, err := monthRange(months) if err != nil { return nil, err } table := fmt.Sprintf("%s.%s", cfg.DbName, cfg.MonthlyStable) // 用秒级时间字面量:月表写入点是当月最后一天 23:59:59,秒级精度已能覆盖。 startText := start.Format("2006-01-02 15:04:05") endText := end.Format("2006-01-02 15:04:05") batches, failedBatches := 0, 0 for chunkStart := 0; chunkStart < len(iccids); chunkStart += usageHistoryTDEngineChunk { chunkEnd := chunkStart + usageHistoryTDEngineChunk if chunkEnd > len(iccids) { chunkEnd = len(iccids) } chunk := iccids[chunkStart:chunkEnd] // TDengine 驱动不会给字符串参数补引号,用 ? 绑定会让时间条件变成 // `ts >= 2025-10-01 00:00:00` 这种非法表达式(报 syntax error)。 // 因此这里把 ICCID 与时间都拼成带引号的字面量, // ICCID 必须经过字符白名单校验,杜绝拼接注入。 quoted := make([]string, 0, len(chunk)) safe := make([]string, 0, len(chunk)) for _, iccid := range chunk { if !isSafeIccid(iccid) { continue } safe = append(safe, iccid) quoted = append(quoted, "'"+iccid+"'") } if len(quoted) == 0 { continue } batches++ // 先按「查询成功、暂无记录」建条目;一旦本批出错再整体标记为查询失败。 for _, iccid := range safe { if result[iccid] == nil { result[iccid] = &cardUsageResult{Months: make(map[string]float64, len(months))} } } markFailedBatch := func() { for _, iccid := range safe { if entry := result[iccid]; entry != nil { entry.Failed = true entry.Months = map[string]float64{} } } } sql := fmt.Sprintf( "SELECT iccid, ts, usage_mb FROM %s WHERE iccid IN (%s) AND ts >= '%s' AND ts <= '%s'", table, strings.Join(quoted, ","), startText, endText, ) rows, queryErr := tdengine.Query(ctx, sql) if queryErr != nil { // 单批失败不拖垮整个导出:本批卡片标记为「查询失败」,其余批次照常。 failedBatches++ markFailedBatch() g.Log().Errorf(ctx, "[simcardquery] 查询月流量失败 cards=%d range=%s~%s err=%v", len(chunk), startText, endText, queryErr) continue } latest := make(map[string]map[string]time.Time, len(chunk)) scanFailed := false for rows.Next() { var iccid string var ts time.Time var usage float64 if scanErr := rows.Scan(&iccid, &ts, &usage); scanErr != nil { scanFailed = true g.Log().Errorf(ctx, "[simcardquery] 解析月流量结果失败 err=%v", scanErr) break } key := ts.Format("2006-01") if latest[iccid] == nil { latest[iccid] = make(map[string]time.Time, len(months)) } previous, exists := latest[iccid][key] if exists && !ts.After(previous) { continue } entry := result[iccid] if entry == nil { entry = &cardUsageResult{Months: make(map[string]float64, len(months))} result[iccid] = entry } latest[iccid][key] = ts entry.Months[key] = usage } if !scanFailed { if rowsErr := rows.Err(); rowsErr != nil { scanFailed = true g.Log().Errorf(ctx, "[simcardquery] 遍历月流量结果失败 err=%v", rowsErr) } } rows.Close() if scanFailed { failedBatches++ markFailedBatch() } } // 全部批次都失败才判定服务不可用。 if batches > 0 && failedBatches == batches { g.Log().Errorf(ctx, "[simcardquery] 流量服务全部批次查询失败 batches=%d", batches) return nil, gerror.New("流量数据服务暂时不可用,本次导出未能完成。请稍后重试,若持续出现可联系技术支持。") } if failedBatches > 0 { g.Log().Warningf(ctx, "[simcardquery] 流量服务部分批次失败 batches=%d failed=%d", batches, failedBatches) } return result, nil } // formatMonthValues 把某行的月度用量格式化成导出的单元格值,三种状态严格区分: // // 查询失败(服务异常)→ 留空(原因见「失败原因」列) // 查询正常、该月无记录 → "-" // 查询正常、有记录 → 数值(零用量即 0) func formatMonthValues(row *model.SimCardUsageHistoryRow, months []string) []interface{} { values := make([]interface{}, 0, len(months)) for _, month := range months { if row.UsageQueryFailed || row.QueryResult == queryResultMiss { // 整卡流量查询失败,或这张卡本身没查到:月份列留空, // 原因由「失败原因」列说明,保证月份列始终是「数字 + 半角横杠 + 空」。 values = append(values, "") continue } usage, exists := row.MonthValues[month] if !exists { values = append(values, usageNoDataText) continue } values = append(values, roundMb(usage)) } return values } // newUsageRow 组装月历史导出行,并统一处理「流量查询失败」的说明。 func newUsageRow(card *cardBrief, usage map[string]*cardUsageResult) *model.SimCardUsageHistoryRow { row := &model.SimCardUsageHistoryRow{ Iccid: card.Iccid, Msisdn: card.Msisdn, Imsi: card.Imsi, OutStockTime: card.OutStockTime, ExpireAt: card.ExpireAt, MonthValues: usageMonths(usage, card.Iccid), UsageQueryFailed: usageFailed(usage, card.Iccid), QueryResult: queryResultHit, } if row.UsageQueryFailed { row.FailReason = usageQueryFailedNote } return row } // usageMonths 取某卡的月度用量;该卡查询失败或没有条目时返回空 map。 func usageMonths(usage map[string]*cardUsageResult, iccid string) map[string]float64 { if entry := usage[iccid]; entry != nil { return entry.Months } return map[string]float64{} } // usageFailed 该卡的流量数据是否查询失败。 func usageFailed(usage map[string]*cardUsageResult, iccid string) bool { if entry := usage[iccid]; entry != nil { return entry.Failed } return false } // LastCompleteMonths 返回最近 N 个「已结束的月份」,按时间倒序(最近的在前)。 // 当月不计入:当月数据仍在累积,导出口径会和下个月对不上。 func LastCompleteMonths(now time.Time, count int) []string { months := make([]string, 0, count) cursor := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location()).AddDate(0, -1, 0) for i := 0; i < count; i++ { months = append(months, cursor.Format("2006-01")) cursor = cursor.AddDate(0, -1, 0) } return months } // monthRange 把月份清单换算成查询时间范围(含首尾)。 func monthRange(months []string) (time.Time, time.Time, error) { if len(months) == 0 { return time.Time{}, time.Time{}, gerror.New("月份清单为空") } parsed := make([]time.Time, 0, len(months)) for _, month := range months { value, err := time.ParseInLocation("2006-01", month, time.Local) if err != nil { return time.Time{}, time.Time{}, gerror.Wrapf(err, "月份格式错误:%s", month) } parsed = append(parsed, value) } sort.Slice(parsed, func(i, j int) bool { return parsed[i].Before(parsed[j]) }) start := parsed[0] end := parsed[len(parsed)-1].AddDate(0, 1, 0).Add(-time.Second) return start, end, nil } // isSafeIccid ICCID 拼进 SQL 字面量前的白名单校验。 // ICCID 正常只含数字,个别批次可能带字母;除此之外一律拒绝,避免拼接注入。 func isSafeIccid(value string) bool { if value == "" || len(value) > 32 { return false } for _, r := range value { switch { case r >= '0' && r <= '9': case r >= 'a' && r <= 'z': case r >= 'A' && r <= 'Z': default: return false } } return true } func validateTDEngineNames(database, stable string) error { if database == "" || stable == "" { return gerror.New("流量库配置不完整") } for _, value := range []string{database, stable} { for _, r := range value { if !(r == '_' || (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9')) { return gerror.Newf("流量库标识非法:%s", value) } } } return nil } func formatTime(value *gtime.Time) string { if value == nil || value.IsZero() { return "" } return value.Format("Y-m-d H:i:s") }