MiFaZhan
750 lines
24 KiB
Go
750 lines
24 KiB
Go
package trafficcutoff
|
|
|
|
import (
|
|
"context"
|
|
"sort"
|
|
"strings"
|
|
|
|
"github.com/gogf/gf/v2/database/gdb"
|
|
"github.com/gogf/gf/v2/encoding/gjson"
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/dao"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/do"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/batchresult"
|
|
trafficcutoff "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/trafficcutoff"
|
|
)
|
|
|
|
const (
|
|
resultFileStatusSuccess = "success"
|
|
resultFileStatusFailed = "failed"
|
|
|
|
validationFailureReason = "批次校验未通过，已跳过流量断网更新"
|
|
)
|
|
|
|
type Repository struct {
|
|
db gdb.DB
|
|
resultFiles batchresult.Generator
|
|
lockEnabled bool
|
|
}
|
|
|
|
func New() *Repository {
|
|
return &Repository{resultFiles: batchresult.New(), lockEnabled: true}
|
|
}
|
|
|
|
func (r *Repository) database() gdb.DB {
|
|
if r.db != nil {
|
|
return r.db
|
|
}
|
|
return g.DB()
|
|
}
|
|
|
|
func (r *Repository) applyLock(model *gdb.Model) *gdb.Model {
|
|
if r.lockEnabled {
|
|
return model.LockUpdate()
|
|
}
|
|
return model
|
|
}
|
|
|
|
func (r *Repository) Load(ctx context.Context, batchNo string) (*trafficcutoff.Task, error) {
|
|
batchNo = strings.TrimSpace(batchNo)
|
|
if batchNo == "" {
|
|
return nil, gerror.New("traffic cutoff batch number is empty")
|
|
}
|
|
|
|
var main entity.BatchMain
|
|
if err := r.database().Model(dao.BatchMain.Table()).Safe().Ctx(ctx).
|
|
Where(dao.BatchMain.Columns().BatchNo, batchNo).
|
|
Where(dao.BatchMain.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Where(dao.BatchMain.Columns().BatchAsyncStatus, consts.BatchAsyncStatusProcessing).
|
|
Scan(&main); err != nil {
|
|
return nil, gerror.Wrap(err, "query traffic cutoff batch main")
|
|
}
|
|
if main.Id == 0 {
|
|
return nil, gerror.New("processing traffic cutoff batch does not exist")
|
|
}
|
|
|
|
var rawDetails []*entity.BatchDetail
|
|
if err := r.database().Model(dao.BatchDetail.Table()).Safe().Ctx(ctx).
|
|
Where(dao.BatchDetail.Columns().BatchNo, batchNo).
|
|
Where(dao.BatchDetail.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Where(dao.BatchDetail.Columns().DetailStatus, consts.BatchDetailStatusRunning).
|
|
OrderAsc(dao.BatchDetail.Columns().RowNo).
|
|
Scan(&rawDetails); err != nil {
|
|
return nil, gerror.Wrap(err, "query running traffic cutoff details")
|
|
}
|
|
if len(rawDetails) == 0 {
|
|
return nil, gerror.New("traffic cutoff batch has no running details")
|
|
}
|
|
trafficCutoffStatus, err := readTrafficCutoffStatus(main.BatchInfo)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
task := &trafficcutoff.Task{
|
|
Batch: trafficcutoff.Batch{
|
|
ID: main.Id,
|
|
BatchNo: main.BatchNo,
|
|
BatchType: main.BatchType,
|
|
TenantID: main.TenantId,
|
|
DeptID: main.DeptId,
|
|
CreatedBy: main.CreatedBy,
|
|
CreatedDept: main.CreatedDept,
|
|
StartTime: main.StartTime,
|
|
BatchInfo: main.BatchInfo,
|
|
},
|
|
TrafficCutoffStatus: trafficCutoffStatus,
|
|
TrafficCutoffThreshold: readTrafficCutoffThreshold(main.BatchInfo),
|
|
ThresholdUnit: readThresholdUnit(main.BatchInfo),
|
|
Details: make([]*trafficcutoff.Detail, 0, len(rawDetails)),
|
|
}
|
|
for _, raw := range rawDetails {
|
|
if raw == nil {
|
|
continue
|
|
}
|
|
task.Details = append(task.Details, buildDetail(raw))
|
|
}
|
|
if len(task.Details) == 0 {
|
|
return nil, gerror.New("traffic cutoff batch has no readable running details")
|
|
}
|
|
if err := r.attachCards(ctx, task); err != nil {
|
|
return nil, err
|
|
}
|
|
return task, nil
|
|
}
|
|
|
|
func buildDetail(raw *entity.BatchDetail) *trafficcutoff.Detail {
|
|
detail := &trafficcutoff.Detail{
|
|
ID: raw.Id,
|
|
RowNo: raw.RowNo,
|
|
ICCID: strings.TrimSpace(raw.ItemId),
|
|
Reason: strings.TrimSpace(raw.ErrorReason),
|
|
}
|
|
if raw.Snapshot != nil {
|
|
if value := firstSnapshot(raw.Snapshot, "ICCID", "iccid"); value != "" {
|
|
detail.ICCID = value
|
|
}
|
|
detail.InputMSISDN = firstSnapshot(raw.Snapshot, "MSISDN", "msisdn", "inputMsisdn", "input_msisdn")
|
|
detail.InputIMSI = firstSnapshot(raw.Snapshot, "IMSI", "imsi", "inputImsi", "input_imsi")
|
|
detail.TrafficCutoffCrossMonthHold = raw.Snapshot.Get("trafficCutoffCrossMonthHold").Int()
|
|
detail.MSISDN = detail.InputMSISDN
|
|
detail.IMSI = detail.InputIMSI
|
|
detail.Authorized = raw.Snapshot.Get("authorized").Bool()
|
|
}
|
|
return detail
|
|
}
|
|
|
|
func firstSnapshot(snapshot *gjson.Json, keys ...string) string {
|
|
if snapshot == nil {
|
|
return ""
|
|
}
|
|
for _, key := range keys {
|
|
if value := strings.TrimSpace(snapshot.Get(key).String()); value != "" {
|
|
return value
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (r *Repository) attachCards(ctx context.Context, task *trafficcutoff.Task) error {
|
|
cards, err := r.loadCards(ctx, task.Batch.TenantID, detailICCIDList(task.Details))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
attachCards(task, cards)
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) loadCards(ctx context.Context, tenantID string, iccids []string) (map[string]*entity.TSimCardInfo, error) {
|
|
result := make(map[string]*entity.TSimCardInfo)
|
|
iccids = uniqueSortedStrings(iccids)
|
|
if len(iccids) == 0 {
|
|
return result, nil
|
|
}
|
|
var cards []*entity.TSimCardInfo
|
|
if err := r.database().Model(dao.TSimCardInfo.Table()).Safe().Ctx(ctx).
|
|
Where(dao.TSimCardInfo.Columns().TenantId, tenantID).
|
|
WhereIn(dao.TSimCardInfo.Columns().Iccid, iccids).
|
|
Scan(&cards); err != nil {
|
|
return nil, gerror.Wrap(err, "query traffic cutoff SIM cards")
|
|
}
|
|
for _, card := range cards {
|
|
if card != nil && strings.TrimSpace(card.Iccid) != "" {
|
|
result[strings.TrimSpace(card.Iccid)] = card
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func attachCards(task *trafficcutoff.Task, cards map[string]*entity.TSimCardInfo) {
|
|
for _, detail := range task.Details {
|
|
if detail == nil {
|
|
continue
|
|
}
|
|
card := cards[strings.TrimSpace(detail.ICCID)]
|
|
if card == nil {
|
|
detail.Card = nil
|
|
continue
|
|
}
|
|
detail.Card = &trafficcutoff.Card{
|
|
ID: card.Id,
|
|
ICCID: strings.TrimSpace(card.Iccid),
|
|
MSISDN: strings.TrimSpace(card.Msisdn),
|
|
IMSI: strings.TrimSpace(card.Imsi),
|
|
BossCardStatus: card.BossCardStatus,
|
|
TrafficCutoffStatus: card.TrafficCutoffStatus,
|
|
TrafficCutoffThreshold: card.TrafficCutoffThreshold,
|
|
}
|
|
detail.MSISDN = detail.Card.MSISDN
|
|
detail.IMSI = detail.Card.IMSI
|
|
detail.BossCardStatusBefore = detail.Card.BossCardStatus
|
|
detail.TrafficCutoffStatusBefore = detail.Card.TrafficCutoffStatus
|
|
detail.TrafficCutoffThresholdBefore = detail.Card.TrafficCutoffThreshold
|
|
}
|
|
}
|
|
|
|
func (r *Repository) SaveValidationFailure(ctx context.Context, task *trafficcutoff.Task) error {
|
|
if task == nil {
|
|
return gerror.New("traffic cutoff task is nil")
|
|
}
|
|
reason := strings.TrimSpace(task.FailureReason)
|
|
if reason == "" {
|
|
reason = validationFailureReason
|
|
}
|
|
now := gtime.Now()
|
|
return r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
if err := updateDetailsFailed(ctx, tx, task, reason); err != nil {
|
|
return err
|
|
}
|
|
return updateMainFailed(ctx, tx, task, len(task.Details), reason, now)
|
|
})
|
|
}
|
|
|
|
func (r *Repository) CommitBusiness(ctx context.Context, task *trafficcutoff.Task) error {
|
|
if task == nil {
|
|
return gerror.New("traffic cutoff task is nil")
|
|
}
|
|
now := gtime.Now()
|
|
iccids := uniqueSortedStrings(detailICCIDList(task.Details))
|
|
return r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
cards, err := r.lockCards(ctx, tx, task.Batch.TenantID, iccids)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "lock traffic cutoff SIM cards")
|
|
}
|
|
attachCards(task, cards)
|
|
if !validateLockedTask(task) {
|
|
reason := strings.TrimSpace(task.FailureReason)
|
|
if reason == "" {
|
|
reason = validationFailureReason
|
|
}
|
|
if err = updateDetailsFailed(ctx, tx, task, reason); err != nil {
|
|
return err
|
|
}
|
|
return updateMainFailed(ctx, tx, task, len(task.Details), reason, now)
|
|
}
|
|
|
|
ids := detailCardIDs(task.Details)
|
|
result, err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).
|
|
Where(dao.TSimCardInfo.Columns().TenantId, task.Batch.TenantID).
|
|
WhereIn(dao.TSimCardInfo.Columns().Id, ids).
|
|
Data(do.TSimCardInfo{
|
|
TrafficCutoffStatus: task.TrafficCutoffStatus,
|
|
TrafficCutoffThreshold: task.TrafficCutoffThreshold,
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff fields")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff card update result")
|
|
}
|
|
if int(affected) != len(ids) {
|
|
return gerror.New("some traffic cutoff cards changed before update")
|
|
}
|
|
if err = markCrossMonthCutoffTriggered(ctx, tx, task); err != nil {
|
|
return err
|
|
}
|
|
for _, detail := range task.Details {
|
|
if detail == nil {
|
|
continue
|
|
}
|
|
detail.TrafficCutoffStatusAfter = task.TrafficCutoffStatus
|
|
detail.TrafficCutoffThresholdAfter = task.TrafficCutoffThreshold
|
|
}
|
|
if err = updateDetailsSuccess(ctx, tx, task); err != nil {
|
|
return err
|
|
}
|
|
return updateMainSuccess(ctx, tx, task, len(task.Details), now)
|
|
})
|
|
}
|
|
|
|
func markCrossMonthCutoffTriggered(ctx context.Context, tx gdb.TX, task *trafficcutoff.Task) error {
|
|
ids := make([]int64, 0, len(task.Details))
|
|
for _, detail := range task.Details {
|
|
if detail != nil && detail.Reason == "" && detail.TrafficCutoffCrossMonthHold == 1 && detail.Card != nil {
|
|
ids = append(ids, detail.Card.ID)
|
|
}
|
|
}
|
|
ids = uniqueSortedInt64(ids)
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
result, err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).
|
|
Where(dao.TSimCardInfo.Columns().TenantId, task.Batch.TenantID).
|
|
WhereIn(dao.TSimCardInfo.Columns().Id, ids).
|
|
Where("traffic_cutoff_cross_month_hold", 1).
|
|
Data(do.TSimCardTrafficCutoffConfig{
|
|
TrafficCutoffTriggeredAt: gtime.Now(),
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff triggered time")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff triggered time update result")
|
|
}
|
|
if int(affected) != len(ids) {
|
|
return gerror.New("some traffic cutoff triggered times changed before update")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) lockCards(ctx context.Context, tx gdb.TX, tenantID string, iccids []string) (map[string]*entity.TSimCardInfo, error) {
|
|
result := make(map[string]*entity.TSimCardInfo)
|
|
if len(iccids) == 0 {
|
|
return result, nil
|
|
}
|
|
var cards []*entity.TSimCardInfo
|
|
if err := r.applyLock(tx.Ctx(ctx).Model(dao.TSimCardInfo.Table())).
|
|
Where(dao.TSimCardInfo.Columns().TenantId, tenantID).
|
|
WhereIn(dao.TSimCardInfo.Columns().Iccid, iccids).
|
|
Scan(&cards); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, card := range cards {
|
|
if card != nil && strings.TrimSpace(card.Iccid) != "" {
|
|
result[strings.TrimSpace(card.Iccid)] = card
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func validateLockedTask(task *trafficcutoff.Task) bool {
|
|
if task == nil || len(task.Details) == 0 {
|
|
if task != nil {
|
|
task.FailureReason = "流量断网批次无有效明细"
|
|
}
|
|
return false
|
|
}
|
|
if task.TrafficCutoffStatus != 1 || task.TrafficCutoffThreshold <= 0 {
|
|
task.FailureReason = validationFailureReason
|
|
for _, detail := range task.Details {
|
|
if detail != nil {
|
|
detail.Reason = validationFailureReason
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
hasFailure := false
|
|
statusSet := make(map[int]struct{})
|
|
for _, detail := range task.Details {
|
|
if detail == nil {
|
|
hasFailure = true
|
|
continue
|
|
}
|
|
detail.Reason = ""
|
|
if detail.Card == nil {
|
|
detail.Reason = "ICCID不存在或不属于当前租户"
|
|
} else if !detail.Authorized {
|
|
detail.Reason = "ICCID不在当前操作员授权范围内"
|
|
} else if detail.InputMSISDN != "" && detail.InputMSISDN != detail.Card.MSISDN {
|
|
detail.Reason = "MSISDN与卡档案不一致"
|
|
} else if detail.InputIMSI != "" && detail.InputIMSI != detail.Card.IMSI {
|
|
detail.Reason = "IMSI与卡档案不一致"
|
|
} else if detail.Card.BossCardStatus == trafficcutoff.BossCardStatusCanceled {
|
|
detail.Reason = "ICCID已销户"
|
|
}
|
|
if detail.Reason != "" {
|
|
hasFailure = true
|
|
continue
|
|
}
|
|
statusSet[detail.Card.BossCardStatus] = struct{}{}
|
|
}
|
|
if len(statusSet) > 1 {
|
|
for _, detail := range task.Details {
|
|
if detail != nil && detail.Reason == "" {
|
|
detail.Reason = "同一批次内卡状态不一致"
|
|
}
|
|
}
|
|
hasFailure = true
|
|
}
|
|
if !hasFailure {
|
|
task.ValidationOK = true
|
|
return true
|
|
}
|
|
for _, detail := range task.Details {
|
|
if detail != nil && detail.Reason == "" {
|
|
detail.Reason = "同一批次存在无效流量断网明细，已跳过更新"
|
|
}
|
|
}
|
|
task.FailureReason = validationFailureReason
|
|
return false
|
|
}
|
|
|
|
func updateDetailsFailed(ctx context.Context, tx gdb.TX, task *trafficcutoff.Task, fallback string) error {
|
|
for _, detail := range task.Details {
|
|
if detail == nil {
|
|
continue
|
|
}
|
|
if strings.TrimSpace(detail.Reason) == "" {
|
|
detail.Reason = fallback
|
|
}
|
|
result, err := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
|
|
Where(dao.BatchDetail.Columns().Id, detail.ID).
|
|
Where(dao.BatchDetail.Columns().BatchNo, task.Batch.BatchNo).
|
|
Where(dao.BatchDetail.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Where(dao.BatchDetail.Columns().DetailStatus, consts.BatchDetailStatusRunning).
|
|
Data(do.BatchDetail{
|
|
DetailStatus: consts.BatchDetailStatusFailed,
|
|
ErrorReason: detail.Reason,
|
|
Snapshot: detail.Snapshot(task),
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff failed detail")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff failed detail result")
|
|
}
|
|
if affected != 1 {
|
|
return gerror.Newf("traffic cutoff detail %d is no longer running", detail.ID)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func updateDetailsSuccess(ctx context.Context, tx gdb.TX, task *trafficcutoff.Task) error {
|
|
for _, detail := range task.Details {
|
|
if detail == nil {
|
|
continue
|
|
}
|
|
result, err := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
|
|
Where(dao.BatchDetail.Columns().Id, detail.ID).
|
|
Where(dao.BatchDetail.Columns().BatchNo, task.Batch.BatchNo).
|
|
Where(dao.BatchDetail.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Where(dao.BatchDetail.Columns().DetailStatus, consts.BatchDetailStatusRunning).
|
|
Data(do.BatchDetail{
|
|
DetailStatus: consts.BatchDetailStatusSuccess,
|
|
ErrorReason: "",
|
|
Snapshot: detail.Snapshot(task),
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff success detail")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff success detail result")
|
|
}
|
|
if affected != 1 {
|
|
return gerror.Newf("traffic cutoff detail %d is no longer running", detail.ID)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func updateMainFailed(ctx context.Context, tx gdb.TX, task *trafficcutoff.Task, total int, reason string, now *gtime.Time) error {
|
|
result, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
|
|
Where(dao.BatchMain.Columns().BatchNo, task.Batch.BatchNo).
|
|
Where(dao.BatchMain.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Where(dao.BatchMain.Columns().BatchAsyncStatus, consts.BatchAsyncStatusProcessing).
|
|
Data(do.BatchMain{
|
|
BatchAsyncStatus: consts.BatchAsyncStatusFailed,
|
|
TotalCount: total,
|
|
ValidCount: total,
|
|
SuccessCount: 0,
|
|
FailCount: total,
|
|
RunResult: consts.BatchRunResultFailed,
|
|
FailReason: reason,
|
|
BatchInfo: buildBatchInfo(task, resultFileStatusFailed, 0, 0, ""),
|
|
EndTime: now,
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff batch main failed")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff failed main result")
|
|
}
|
|
if affected != 1 {
|
|
return gerror.New("traffic cutoff batch is no longer processing")
|
|
}
|
|
task.Batch.BatchInfo = buildBatchInfo(task, resultFileStatusFailed, 0, 0, "")
|
|
return nil
|
|
}
|
|
|
|
func updateMainSuccess(ctx context.Context, tx gdb.TX, task *trafficcutoff.Task, total int, now *gtime.Time) error {
|
|
info := buildBatchInfo(task, "pending", 0, 0, "")
|
|
result, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
|
|
Where(dao.BatchMain.Columns().BatchNo, task.Batch.BatchNo).
|
|
Where(dao.BatchMain.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Where(dao.BatchMain.Columns().BatchAsyncStatus, consts.BatchAsyncStatusProcessing).
|
|
Data(do.BatchMain{
|
|
BatchAsyncStatus: consts.BatchAsyncStatusSuccess,
|
|
TotalCount: total,
|
|
ValidCount: total,
|
|
SuccessCount: total,
|
|
FailCount: 0,
|
|
RunResult: consts.BatchRunResultSuccess,
|
|
FailReason: "",
|
|
BatchInfo: info,
|
|
EndTime: now,
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff batch main success")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff success main result")
|
|
}
|
|
if affected != 1 {
|
|
return gerror.New("traffic cutoff batch is no longer processing")
|
|
}
|
|
task.Batch.BatchInfo = info
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) SaveResultFile(ctx context.Context, task *trafficcutoff.Task, request batchresult.Request) error {
|
|
if task == nil {
|
|
return gerror.New("traffic cutoff task is nil")
|
|
}
|
|
artifact, err := r.artifactGenerator().Create(ctx, request)
|
|
if err != nil {
|
|
return r.updateResultFileStatus(ctx, task, resultFileStatusFailed, 0, 0, err.Error())
|
|
}
|
|
now := gtime.Now()
|
|
var (
|
|
ossID int64
|
|
downloadID int64
|
|
)
|
|
txErr := r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
var insertErr error
|
|
ossID, insertErr = tx.Ctx(ctx).Model(dao.SysOss.Table()).Data(do.SysOss{
|
|
TenantId: task.Batch.TenantID,
|
|
FileName: artifact.FileName,
|
|
OriginalName: artifact.OriginalName,
|
|
FileSuffix: artifact.FileSuffix,
|
|
Path: artifact.Path,
|
|
Url: artifact.URL,
|
|
CreatedDept: task.Batch.CreatedDept,
|
|
CreatedBy: task.Batch.CreatedBy,
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
Service: "local",
|
|
Md5: artifact.MD5,
|
|
FileSize: artifact.Size,
|
|
}).InsertAndGetId()
|
|
if insertErr != nil {
|
|
return gerror.Wrap(insertErr, "insert traffic cutoff result OSS")
|
|
}
|
|
downloadID, insertErr = tx.Ctx(ctx).Model(dao.TResourceBatchDownload.Table()).Data(do.TResourceBatchDownload{
|
|
BatchNo: task.Batch.BatchNo,
|
|
TaskName: batchresult.TaskName(task.Batch.BatchType),
|
|
ResourceType: consts.ResourceTypeExcel,
|
|
ExportScene: task.Batch.BatchType,
|
|
TotalCount: len(task.Details),
|
|
SuccessCount: task.SuccessCount(),
|
|
FailCount: task.FailCount(),
|
|
DownloadStatus: consts.DownloadStatusGenerated,
|
|
OssId: ossID,
|
|
FileName: artifact.FileName,
|
|
FileExt: artifact.FileSuffix,
|
|
FileUrl: artifact.URL,
|
|
FilePath: artifact.Path,
|
|
FileSize: artifact.Size,
|
|
FileMd5: artifact.MD5,
|
|
DownloadCount: 0,
|
|
MaxDownloadCount: 0,
|
|
DeptId: task.Batch.CreatedDept,
|
|
StartTime: task.Batch.StartTime,
|
|
FinishTime: now,
|
|
TenantId: task.Batch.TenantID,
|
|
CreatedDept: task.Batch.CreatedDept,
|
|
CreatedBy: task.Batch.CreatedBy,
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).InsertAndGetId()
|
|
if insertErr != nil {
|
|
return gerror.Wrap(insertErr, "insert traffic cutoff result download")
|
|
}
|
|
return updateBatchInfoOnly(ctx, tx, task, buildBatchInfo(task, resultFileStatusSuccess, ossID, downloadID, ""))
|
|
})
|
|
if txErr != nil {
|
|
batchresult.RemoveLocalFile(artifact)
|
|
return r.updateResultFileStatus(ctx, task, resultFileStatusFailed, 0, 0, txErr.Error())
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) artifactGenerator() batchresult.Generator {
|
|
if r.resultFiles != nil {
|
|
return r.resultFiles
|
|
}
|
|
return batchresult.New()
|
|
}
|
|
|
|
func (r *Repository) updateResultFileStatus(ctx context.Context, task *trafficcutoff.Task, status string, ossID, downloadID int64, reason string) error {
|
|
return r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
return updateBatchInfoOnly(ctx, tx, task, buildBatchInfo(task, status, ossID, downloadID, reason))
|
|
})
|
|
}
|
|
|
|
func updateBatchInfoOnly(ctx context.Context, tx gdb.TX, task *trafficcutoff.Task, info *gjson.Json) error {
|
|
result, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
|
|
Where(dao.BatchMain.Columns().BatchNo, task.Batch.BatchNo).
|
|
Where(dao.BatchMain.Columns().BatchType, consts.BatchTypeTrafficCutoff).
|
|
Data(do.BatchMain{
|
|
BatchInfo: info,
|
|
UpdatedBy: task.Batch.CreatedBy,
|
|
}).Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "update traffic cutoff result file status")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read traffic cutoff result file update result")
|
|
}
|
|
if affected != 1 {
|
|
return gerror.New("traffic cutoff batch disappeared while saving result file")
|
|
}
|
|
task.Batch.BatchInfo = info
|
|
return nil
|
|
}
|
|
|
|
func buildBatchInfo(task *trafficcutoff.Task, resultStatus string, ossID, downloadID int64, reason string) *gjson.Json {
|
|
info := make(map[string]any)
|
|
if task.Batch.BatchInfo != nil {
|
|
for key, value := range task.Batch.BatchInfo.Map() {
|
|
info[key] = value
|
|
}
|
|
}
|
|
info["operation"] = "dataCut"
|
|
info["trafficCutoffStatus"] = task.TrafficCutoffStatus
|
|
info["trafficCutoffThreshold"] = task.TrafficCutoffThreshold
|
|
info["thresholdUnit"] = task.ThresholdUnit
|
|
if resultStatus != "" {
|
|
resultFile := map[string]any{"status": resultStatus}
|
|
if ossID > 0 {
|
|
resultFile["ossId"] = ossID
|
|
}
|
|
if downloadID > 0 {
|
|
resultFile["downloadId"] = downloadID
|
|
}
|
|
if strings.TrimSpace(reason) != "" {
|
|
resultFile["error"] = reason
|
|
}
|
|
info["resultFile"] = resultFile
|
|
}
|
|
return gjson.New(info)
|
|
}
|
|
|
|
func readTrafficCutoffStatus(info *gjson.Json) (int, error) {
|
|
if info == nil || info.IsNil() {
|
|
return 0, gerror.New("traffic cutoff batch missing trafficCutoffStatus")
|
|
}
|
|
if value := info.Get("trafficCutoffStatus"); !value.IsNil() {
|
|
return value.Int(), nil
|
|
}
|
|
return 0, gerror.New("traffic cutoff batch missing trafficCutoffStatus")
|
|
}
|
|
|
|
func readTrafficCutoffThreshold(info *gjson.Json) float64 {
|
|
if info == nil || info.IsNil() {
|
|
return 0
|
|
}
|
|
if value := info.Get("trafficCutoffThreshold"); !value.IsNil() {
|
|
return value.Float64()
|
|
}
|
|
return info.Get("cutoffThreshold").Float64()
|
|
}
|
|
|
|
func readThresholdUnit(info *gjson.Json) string {
|
|
if info != nil {
|
|
if value := strings.TrimSpace(info.Get("thresholdUnit").String()); value != "" {
|
|
return value
|
|
}
|
|
}
|
|
return "MB"
|
|
}
|
|
|
|
func detailICCIDList(details []*trafficcutoff.Detail) []string {
|
|
iccids := make([]string, 0, len(details))
|
|
for _, detail := range details {
|
|
if detail != nil && strings.TrimSpace(detail.ICCID) != "" {
|
|
iccids = append(iccids, strings.TrimSpace(detail.ICCID))
|
|
}
|
|
}
|
|
return iccids
|
|
}
|
|
|
|
func detailCardIDs(details []*trafficcutoff.Detail) []int64 {
|
|
ids := make([]int64, 0, len(details))
|
|
for _, detail := range details {
|
|
if detail != nil && detail.Card != nil {
|
|
ids = append(ids, detail.Card.ID)
|
|
}
|
|
}
|
|
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
|
return ids
|
|
}
|
|
|
|
func uniqueSortedInt64(values []int64) []int64 {
|
|
if len(values) == 0 {
|
|
return nil
|
|
}
|
|
seen := make(map[int64]struct{}, len(values))
|
|
result := make([]int64, 0, len(values))
|
|
for _, value := range values {
|
|
if value <= 0 {
|
|
continue
|
|
}
|
|
if _, ok := seen[value]; ok {
|
|
continue
|
|
}
|
|
seen[value] = struct{}{}
|
|
result = append(result, value)
|
|
}
|
|
sort.Slice(result, func(i, j int) bool { return result[i] < result[j] })
|
|
return result
|
|
}
|
|
|
|
func uniqueSortedStrings(values []string) []string {
|
|
if len(values) == 0 {
|
|
return nil
|
|
}
|
|
seen := make(map[string]struct{}, len(values))
|
|
out := make([]string, 0, len(values))
|
|
for _, value := range values {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
continue
|
|
}
|
|
if _, ok := seen[value]; ok {
|
|
continue
|
|
}
|
|
seen[value] = struct{}{}
|
|
out = append(out, value)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
var _ trafficcutoff.Repository = (*Repository)(nil)
|