25c081fe60
Some checks failed
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Failing after 8h37m36s
- M2 批量换绑表头改为“平台登录账号”，只认手机号，细分格式、不存在、多匹配、非平台、已禁用失败原因 - M3 后台订单列表与详情返回 C 端在线购买的交易流水号、支付渠道、商户名称，详情新增全部支付单 payments - M16 预警规则不再返回 real_data_mb，预警列表/详情新增 threshold_mb，导出末列追加“阈值流量(MB)” - M23 设备导入 IMEI 必填，空 IMEI 行记为失败 - OpenSpec Change fix-august-iteration-follow-up-gaps：13/18，导入任务、导出任务的 Worker 端到端验证与前端/运营通知待部署后完成
355 lines
15 KiB
Go
355 lines
15 KiB
Go
package exporter
|
|
|
|
import (
|
|
"context"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/domain/packagetrafficalert"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
|
)
|
|
|
|
// PackageTrafficAlertDataSource 套餐真流量达量预警导出数据源。
|
|
//
|
|
// 粒度为一条预警记录。套餐、用量、总量、阈值、到期时间与资产标识类列一律读预警行冻结的触发快照；
|
|
// 店铺、业务员与用户组按导出执行时当前归属补充，用户组按既有实时推导，不写入店铺表。
|
|
// 本场景只对超级管理员与平台账号开放：受控入口已做角色门禁，这里再校验一次，
|
|
// 阻止通过通用导出入口以代理身份创建本场景任务后读到预警数据。
|
|
type PackageTrafficAlertDataSource struct {
|
|
db *gorm.DB
|
|
}
|
|
|
|
// NewPackageTrafficAlertDataSource 创建套餐真流量达量预警导出数据源。
|
|
func NewPackageTrafficAlertDataSource(db *gorm.DB) *PackageTrafficAlertDataSource {
|
|
return &PackageTrafficAlertDataSource{db: db}
|
|
}
|
|
|
|
// Scene 返回导出场景编码。
|
|
func (s *PackageTrafficAlertDataSource) Scene() string {
|
|
return constants.ExportTaskScenePackageTrafficAlert
|
|
}
|
|
|
|
// Count 统计导出预警行数。
|
|
func (s *PackageTrafficAlertDataSource) Count(ctx context.Context, params ExportParams) (int, error) {
|
|
if err := ensurePackageTrafficAlertExportAllowed(params); err != nil {
|
|
return 0, err
|
|
}
|
|
var total int64
|
|
query, err := s.applyFilters(s.baseQuery(ctx, params), params)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if err := query.Count(&total).Error; err != nil {
|
|
return 0, err
|
|
}
|
|
return int(total), nil
|
|
}
|
|
|
|
// Headers 返回套餐真流量达量预警导出表头。
|
|
// 表头在 dispatch 阶段冻结，历史任务重导出沿用同一列序；不含任何运营商通道列。
|
|
func (s *PackageTrafficAlertDataSource) Headers(context.Context, ExportParams) ([]string, error) {
|
|
return []string{
|
|
"资产类型", "资产标识", "对应标识符", "卡标识", "设备类型", "设备型号",
|
|
"套餐名称", "真流量已用量(MB)", "真流量额度(MB)", "比例(%)", "阈值快照(%)",
|
|
"到期时间", "剩余天数", "触发时间", "店铺", "业务员", "用户组", "通知投递结果",
|
|
packageTrafficAlertThresholdMBHeader,
|
|
}, nil
|
|
}
|
|
|
|
// packageTrafficAlertThresholdMBHeader 阈值对应流量列，固定追加在最后一列；历史任务冻结表头没有该列。
|
|
const packageTrafficAlertThresholdMBHeader = "阈值流量(MB)"
|
|
|
|
// alertHeadersIncludeThresholdMB 判断冻结表头是否为带“阈值流量(MB)”末列的新表头。
|
|
func alertHeadersIncludeThresholdMB(headers []string) bool {
|
|
return len(headers) == 0 || headers[len(headers)-1] == packageTrafficAlertThresholdMBHeader
|
|
}
|
|
|
|
// Fetch 按 offset/limit 查询预警导出数据。
|
|
func (s *PackageTrafficAlertDataSource) Fetch(ctx context.Context, params ExportParams, offset, limit int) ([][]string, error) {
|
|
if limit <= 0 {
|
|
return [][]string{}, nil
|
|
}
|
|
if err := ensurePackageTrafficAlertExportAllowed(params); err != nil {
|
|
return nil, err
|
|
}
|
|
var items []packageTrafficAlertExportRow
|
|
filtered, err := s.applyFilters(s.baseQuery(ctx, params), params)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query := filtered.
|
|
Select(`
|
|
a.asset_type,
|
|
a.asset_identifier_snapshot,
|
|
a.counterpart_identifier_snapshot,
|
|
a.card_identifier_snapshot,
|
|
a.device_type_snapshot,
|
|
a.device_model_snapshot,
|
|
a.package_name_snapshot,
|
|
a.used_mb_snapshot,
|
|
a.limit_mb_snapshot,
|
|
a.usage_percent_snapshot,
|
|
a.threshold_percent_snapshot,
|
|
a.expires_at_snapshot,
|
|
a.triggered_at,
|
|
a.shop_id_snapshot,
|
|
a.shop_name_snapshot,
|
|
a.business_owner_account_id_snapshot,
|
|
a.business_owner_name_snapshot,
|
|
a.notification_event_id,
|
|
sh.id AS current_shop_id,
|
|
COALESCE(sh.shop_name, '') AS current_shop_name,
|
|
owner.id AS current_owner_id,
|
|
COALESCE(owner.username, '') AS current_owner_name,
|
|
oe.status AS outbox_status,
|
|
n.id AS notification_id
|
|
`).
|
|
Order("a.triggered_at DESC").Order("a.id DESC").
|
|
Limit(limit).Offset(offset)
|
|
if err := query.Scan(&items).Error; err != nil {
|
|
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询套餐真流量达量预警导出数据失败")
|
|
}
|
|
groupNames, err := s.loadBusinessUserGroupNames(ctx, items)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
now := time.Now().UTC()
|
|
includeThresholdMB := alertHeadersIncludeThresholdMB(params.ResolvedHeaders)
|
|
rows := make([][]string, 0, len(items))
|
|
for _, item := range items {
|
|
row := []string{
|
|
assetTypeName(item.AssetType),
|
|
item.AssetIdentifier,
|
|
item.CounterpartIdentifier,
|
|
item.CardIdentifier,
|
|
item.DeviceType,
|
|
item.DeviceModel,
|
|
item.PackageName,
|
|
strconv.FormatInt(item.UsedMB, 10),
|
|
strconv.FormatInt(item.LimitMB, 10),
|
|
formatPercentValue(item.UsagePercent),
|
|
formatPercentValue(item.ThresholdPercent),
|
|
formatOptionalTime(item.ExpiresAt),
|
|
formatRemainingDays(item.ExpiresAt, now),
|
|
item.TriggeredAt.Format(exportTimeLayout),
|
|
item.CurrentShopName,
|
|
item.CurrentOwnerName,
|
|
currentOwnerGroupName(groupNames, item.CurrentOwnerID),
|
|
constants.GetPackageTrafficAlertNotifyStatusName(resolveAlertNotifyStatus(item)),
|
|
}
|
|
if includeThresholdMB {
|
|
row = append(row, strconv.FormatFloat(packagetrafficalert.ThresholdMB(item.LimitMB, item.ThresholdPercent), 'f', 2, 64))
|
|
}
|
|
rows = append(rows, row)
|
|
}
|
|
return rows, nil
|
|
}
|
|
|
|
// baseQuery 构造预警导出基础查询。
|
|
// 归属展示列按执行时当前归属补充：资产 → 当前店铺 → 店铺当前业务员；用户组随后按业务员账号实时推导。
|
|
func (s *PackageTrafficAlertDataSource) baseQuery(ctx context.Context, params ExportParams) *gorm.DB {
|
|
query := s.db.WithContext(ctx).Table("tb_package_traffic_alert AS a").
|
|
Joins("LEFT JOIN tb_iot_card AS c ON a.asset_type = ? AND c.id = a.asset_id AND c.deleted_at IS NULL",
|
|
constants.AssetTypeIotCard).
|
|
Joins("LEFT JOIN tb_device AS d ON a.asset_type = ? AND d.id = a.asset_id AND d.deleted_at IS NULL",
|
|
constants.AssetTypeDevice).
|
|
Joins("LEFT JOIN tb_shop AS sh ON sh.id = COALESCE(c.shop_id, d.shop_id) AND sh.deleted_at IS NULL").
|
|
Joins("LEFT JOIN tb_account AS owner ON owner.id = sh.business_owner_account_id AND owner.deleted_at IS NULL").
|
|
Joins("LEFT JOIN tb_outbox_event AS oe ON oe.event_id = a.notification_event_id").
|
|
Joins("LEFT JOIN tb_notification AS n ON n.event_id = a.notification_event_id")
|
|
// 数据范围使用导出侧范围过滤（空范围拒绝），不得使用请求上下文版过滤（空范围语义相反）。
|
|
return applyExportShopScope(query, params, "a.shop_id_snapshot")
|
|
}
|
|
|
|
// applyFilters 应用导出筛选快照。
|
|
// 筛选口径与列表一致，都作用在触发快照列上；时间范围按触发时间闭区间解析，
|
|
// 冻结值一律按统一严格解析器解析，非法值返回错误由调用方落任务失败。
|
|
func (s *PackageTrafficAlertDataSource) applyFilters(query *gorm.DB, params ExportParams) (*gorm.DB, error) {
|
|
if packageID, ok := filterUint(params.Filters, "package_id"); ok {
|
|
query = query.Where("a.package_id = ?", packageID)
|
|
}
|
|
if shopID, ok := filterUint(params.Filters, "shop_id"); ok {
|
|
query = query.Where("a.shop_id_snapshot = ?", shopID)
|
|
}
|
|
if ownerID, ok := filterUint(params.Filters, "business_owner_account_id"); ok {
|
|
query = query.Where("a.business_owner_account_id_snapshot = ?", ownerID)
|
|
}
|
|
if assetType, ok := filterString(params.Filters, "asset_type"); ok {
|
|
query = query.Where("a.asset_type = ?", assetType)
|
|
}
|
|
if identifier, ok := filterString(params.Filters, "asset_identifier"); ok {
|
|
pattern := "%" + identifier + "%"
|
|
query = query.Where("(a.asset_identifier_snapshot ILIKE ? OR a.card_identifier_snapshot ILIKE ? "+
|
|
"OR a.counterpart_identifier_snapshot ILIKE ?)", pattern, pattern, pattern)
|
|
}
|
|
if threshold, ok := alertFilterFloat(params.Filters, "threshold_percent"); ok {
|
|
query = query.Where("a.threshold_percent_snapshot = ?",
|
|
packagetrafficalert.NormalizeThresholdPercent(threshold))
|
|
}
|
|
startTime, endTime, err := strictTimeRange(params.Filters)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if startTime != nil {
|
|
query = query.Where("a.triggered_at >= ?", *startTime)
|
|
}
|
|
if endTime != nil {
|
|
query = query.Where("a.triggered_at <= ?", *endTime)
|
|
}
|
|
if status, ok := filterInt(params.Filters, "notification_status"); ok {
|
|
query = applyAlertNotificationStatusFilter(query, status)
|
|
}
|
|
return query, nil
|
|
}
|
|
|
|
// applyAlertNotificationStatusFilter 按通知投递结果筛选，口径与读侧列表一致。
|
|
func applyAlertNotificationStatusFilter(query *gorm.DB, status int) *gorm.DB {
|
|
const hasEvent = "a.notification_event_id <> ''"
|
|
const hasNotification = "n.id IS NOT NULL"
|
|
switch status {
|
|
case constants.PackageTrafficAlertNotifyNoBusinessOwner:
|
|
return query.Where("a.notification_event_id = ''")
|
|
case constants.PackageTrafficAlertNotifyNotified:
|
|
return query.Where(hasEvent).Where(hasNotification)
|
|
case constants.PackageTrafficAlertNotifyPending:
|
|
return query.Where(hasEvent).Where("NOT ("+hasNotification+")").
|
|
Where("oe.status IN ?", []int{constants.OutboxStatusPending, constants.OutboxStatusDelivering})
|
|
case constants.PackageTrafficAlertNotifyFailed:
|
|
return query.Where(hasEvent).Where("NOT ("+hasNotification+")").
|
|
Where("oe.status = ?", constants.OutboxStatusFailed)
|
|
case constants.PackageTrafficAlertNotifyRecipientGone:
|
|
return query.Where(hasEvent).Where("NOT ("+hasNotification+")").
|
|
Where("oe.status = ?", constants.OutboxStatusDelivered)
|
|
default:
|
|
return query
|
|
}
|
|
}
|
|
|
|
// loadBusinessUserGroupNames 按执行时当前业务员账号批量推导业务用户组名称。
|
|
// 用户组不落在店铺库表上，按既有实时推导读取，多个组按排序拼接。
|
|
func (s *PackageTrafficAlertDataSource) loadBusinessUserGroupNames(ctx context.Context,
|
|
items []packageTrafficAlertExportRow) (map[uint]string, error) {
|
|
ownerIDs := make([]uint, 0, len(items))
|
|
seen := make(map[uint]struct{}, len(items))
|
|
for _, item := range items {
|
|
if item.CurrentOwnerID == nil || *item.CurrentOwnerID == 0 {
|
|
continue
|
|
}
|
|
if _, exists := seen[*item.CurrentOwnerID]; exists {
|
|
continue
|
|
}
|
|
seen[*item.CurrentOwnerID] = struct{}{}
|
|
ownerIDs = append(ownerIDs, *item.CurrentOwnerID)
|
|
}
|
|
return businessUserGroupNames(ctx, s.db, ownerIDs)
|
|
}
|
|
|
|
// packageTrafficAlertExportRow 是预警导出的一行原始投影。
|
|
type packageTrafficAlertExportRow struct {
|
|
AssetType string `gorm:"column:asset_type"`
|
|
AssetIdentifier string `gorm:"column:asset_identifier_snapshot"`
|
|
CounterpartIdentifier string `gorm:"column:counterpart_identifier_snapshot"`
|
|
CardIdentifier string `gorm:"column:card_identifier_snapshot"`
|
|
DeviceType string `gorm:"column:device_type_snapshot"`
|
|
DeviceModel string `gorm:"column:device_model_snapshot"`
|
|
PackageName string `gorm:"column:package_name_snapshot"`
|
|
UsedMB int64 `gorm:"column:used_mb_snapshot"`
|
|
LimitMB int64 `gorm:"column:limit_mb_snapshot"`
|
|
UsagePercent float64 `gorm:"column:usage_percent_snapshot"`
|
|
ThresholdPercent float64 `gorm:"column:threshold_percent_snapshot"`
|
|
ExpiresAt *time.Time `gorm:"column:expires_at_snapshot"`
|
|
TriggeredAt time.Time `gorm:"column:triggered_at"`
|
|
ShopIDSnapshot uint `gorm:"column:shop_id_snapshot"`
|
|
ShopNameSnapshot string `gorm:"column:shop_name_snapshot"`
|
|
BusinessOwnerID *uint `gorm:"column:business_owner_account_id_snapshot"`
|
|
BusinessOwnerName string `gorm:"column:business_owner_name_snapshot"`
|
|
NotificationEventID string `gorm:"column:notification_event_id"`
|
|
CurrentShopID *uint `gorm:"column:current_shop_id"`
|
|
CurrentShopName string `gorm:"column:current_shop_name"`
|
|
CurrentOwnerID *uint `gorm:"column:current_owner_id"`
|
|
CurrentOwnerName string `gorm:"column:current_owner_name"`
|
|
OutboxStatus *int `gorm:"column:outbox_status"`
|
|
NotificationID *uint `gorm:"column:notification_id"`
|
|
}
|
|
|
|
// resolveAlertNotifyStatus 推导导出行的通知投递结果，与列表、详情同口径。
|
|
func resolveAlertNotifyStatus(item packageTrafficAlertExportRow) int {
|
|
return constants.ResolvePackageTrafficAlertNotifyStatus(
|
|
item.NotificationEventID != "", item.OutboxStatus, item.NotificationID != nil)
|
|
}
|
|
|
|
// ensurePackageTrafficAlertExportAllowed 只允许超级管理员与平台账号使用本场景。
|
|
// 通用导出入口不做场景级角色校验，因此这一层门禁是防止代理越权读取预警数据的必要防线。
|
|
func ensurePackageTrafficAlertExportAllowed(params ExportParams) error {
|
|
if params.UserType == constants.UserTypeSuperAdmin || params.UserType == constants.UserTypePlatform {
|
|
return nil
|
|
}
|
|
return errors.New(errors.CodeForbidden, constants.PlatformManagementForbiddenMessage)
|
|
}
|
|
|
|
// alertFilterFloat 解析导出筛选中的小数百分比。
|
|
// 阈值筛选只在预警导出使用，为避免改动既有共享筛选助手文件，这里就地解析。
|
|
func alertFilterFloat(filters map[string]any, key string) (float64, bool) {
|
|
value, ok := filters[key]
|
|
if !ok || value == nil {
|
|
return 0, false
|
|
}
|
|
switch typed := value.(type) {
|
|
case float64:
|
|
return typed, true
|
|
case float32:
|
|
return float64(typed), true
|
|
case int:
|
|
return float64(typed), true
|
|
case int64:
|
|
return float64(typed), true
|
|
case string:
|
|
parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
|
|
if err != nil {
|
|
return 0, false
|
|
}
|
|
return parsed, true
|
|
default:
|
|
return 0, false
|
|
}
|
|
}
|
|
|
|
// currentOwnerGroupName 返回执行时当前业务员的用户组名称，无有效业务员时为空。
|
|
func currentOwnerGroupName(groupNames map[uint]string, ownerID *uint) string {
|
|
if ownerID == nil {
|
|
return ""
|
|
}
|
|
return groupNames[*ownerID]
|
|
}
|
|
|
|
// assetTypeName 返回资产类型的中文名称。
|
|
func assetTypeName(assetType string) string {
|
|
if assetType == constants.AssetTypeDevice {
|
|
return "设备"
|
|
}
|
|
return "物联网卡"
|
|
}
|
|
|
|
// formatPercentValue 输出保留两位小数的百分比。
|
|
func formatPercentValue(value float64) string {
|
|
return strconv.FormatFloat(value, 'f', 2, 64)
|
|
}
|
|
|
|
// formatRemainingDays 按上海自然日推算剩余天数；无到期时间时输出空字符串。
|
|
func formatRemainingDays(expiresAt *time.Time, now time.Time) string {
|
|
if expiresAt == nil {
|
|
return ""
|
|
}
|
|
location := time.FixedZone("Asia/Shanghai", 8*60*60)
|
|
localExpires := expiresAt.In(location)
|
|
localNow := now.In(location)
|
|
expiresDate := time.Date(localExpires.Year(), localExpires.Month(), localExpires.Day(), 0, 0, 0, 0, location)
|
|
nowDate := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, location)
|
|
days := int(expiresDate.Sub(nowDate).Hours() / 24)
|
|
return strconv.Itoa(days)
|
|
}
|