MiFaZhan
248 lines
7.4 KiB
Go
248 lines
7.4 KiB
Go
package lifecycle
|
|
|
|
import (
|
|
"context"
|
|
"sort"
|
|
"strings"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/batchimport"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/entity"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"
|
|
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
)
|
|
|
|
type executorRepository interface {
|
|
ClaimBatch(ctx context.Context, batchNo string) (bool, error)
|
|
LoadTask(ctx context.Context, batchNo string) (*Task, error)
|
|
LoadCards(ctx context.Context, task *Task) (map[string]*entity.TSimCardInfo, error)
|
|
SaveValidationFailure(ctx context.Context, task *Task) error
|
|
SaveOutcome(ctx context.Context, task *Task, row *Row, card *entity.TSimCardInfo, outcome Outcome) error
|
|
SaveDetailFailure(ctx context.Context, task *Task, row *Row) error
|
|
FinalizeBatch(ctx context.Context, task *Task) error
|
|
MarkBatchFailed(ctx context.Context, batchNo, reason string) error
|
|
}
|
|
|
|
type statusChanger interface {
|
|
ChangeSimStatus(ctx context.Context, input *model.SimCardStatusChangeInput) ([]model.SimBaseRes[model.SimCardStatusChangeItem], error)
|
|
}
|
|
|
|
type Executor struct {
|
|
repo executorRepository
|
|
upstream statusChanger
|
|
}
|
|
|
|
func NewExecutor() *Executor {
|
|
return &Executor{repo: NewRepository(), upstream: service.UpstreamChan()}
|
|
}
|
|
|
|
func newExecutor(repo executorRepository, upstream statusChanger) *Executor {
|
|
return &Executor{repo: repo, upstream: upstream}
|
|
}
|
|
|
|
func (e *Executor) Execute(ctx context.Context, batchNo string) error {
|
|
batchNo = strings.TrimSpace(batchNo)
|
|
if batchNo == "" {
|
|
return gerror.New("批次号不能为空")
|
|
}
|
|
claimed, err := e.repo.ClaimBatch(ctx, batchNo)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "抢占停复机批次失败")
|
|
}
|
|
if !claimed {
|
|
return nil
|
|
}
|
|
task, err := e.repo.LoadTask(ctx, batchNo)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
cards, err := e.repo.LoadCards(ctx, task)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "读取SIM卡业务数据失败")
|
|
}
|
|
if !validateTask(task, cards) {
|
|
if err = e.repo.SaveValidationFailure(ctx, task); err != nil {
|
|
return err
|
|
}
|
|
return e.repo.FinalizeBatch(ctx, task)
|
|
}
|
|
|
|
groups := groupRowsByChannel(task.Rows)
|
|
channelKeys := make([]string, 0, len(groups))
|
|
for chanKey := range groups {
|
|
channelKeys = append(channelKeys, chanKey)
|
|
}
|
|
sort.Strings(channelKeys)
|
|
for _, chanKey := range channelKeys {
|
|
rows := groups[chanKey]
|
|
outcomes := e.changeGroup(ctx, task, rows)
|
|
for i, row := range rows {
|
|
outcome := outcomes[i]
|
|
card := cards[row.Iccid]
|
|
if err = e.repo.SaveOutcome(ctx, task, row, card, outcome); err == nil {
|
|
continue
|
|
}
|
|
row.OrderNum = outcome.OrderNum
|
|
row.UpstreamCode = outcome.Code
|
|
row.UpstreamReason = outcome.Message
|
|
row.Reason = "上游已返回，但本地状态落库失败: " + err.Error()
|
|
if saveErr := e.repo.SaveDetailFailure(ctx, task, row); saveErr != nil {
|
|
return gerror.Wrap(saveErr, "记录本地状态落库失败结果失败")
|
|
}
|
|
}
|
|
}
|
|
return e.repo.FinalizeBatch(ctx, task)
|
|
}
|
|
|
|
func (e *Executor) MarkFailed(ctx context.Context, batchNo, reason string) error {
|
|
return e.repo.MarkBatchFailed(ctx, strings.TrimSpace(batchNo), strings.TrimSpace(reason))
|
|
}
|
|
|
|
func validateTask(task *Task, cards map[string]*entity.TSimCardInfo) bool {
|
|
hasFailure := false
|
|
for _, row := range task.Rows {
|
|
if row == nil {
|
|
hasFailure = true
|
|
continue
|
|
}
|
|
if row.Reason != "" {
|
|
hasFailure = true
|
|
continue
|
|
}
|
|
card := cards[row.Iccid]
|
|
if card == nil {
|
|
row.AddErrorReason("ICCID不存在于正式卡记录")
|
|
} else if !row.Authorized {
|
|
row.AddErrorReason("ICCID不在当前操作人的数据权限范围内")
|
|
} else if strings.TrimSpace(card.ChanKey) == "" {
|
|
row.AddErrorReason("ICCID缺少通道标识chanKey")
|
|
} else if row.Msisdn != "" && row.Msisdn != strings.TrimSpace(card.Msisdn) {
|
|
row.AddErrorReason("MSISDN与正式卡记录不一致")
|
|
} else if row.Imsi != "" && row.Imsi != strings.TrimSpace(card.Imsi) {
|
|
row.AddErrorReason("IMSI与正式卡记录不一致")
|
|
} else {
|
|
validateCardStatus(task.Operation, row, card)
|
|
}
|
|
if row.Reason != "" {
|
|
hasFailure = true
|
|
continue
|
|
}
|
|
row.Msisdn = strings.TrimSpace(card.Msisdn)
|
|
row.Imsi = strings.TrimSpace(card.Imsi)
|
|
row.ChanKey = strings.TrimSpace(card.ChanKey)
|
|
row.StatusBefore = card.BossCardStatus
|
|
}
|
|
if hasFailure {
|
|
for _, row := range task.Rows {
|
|
if row != nil && row.Reason == "" {
|
|
row.AddErrorReason(batchimport.NotExecutedReasonPrefix + "本卡校验通过")
|
|
}
|
|
}
|
|
task.FailureReason = "整批业务校验失败，未请求上游"
|
|
return false
|
|
}
|
|
task.ValidationOK = true
|
|
return true
|
|
}
|
|
|
|
func validateCardStatus(operation Operation, row *Row, card *entity.TSimCardInfo) {
|
|
// TODO: 正常系统开发完成后恢复卡状态限制
|
|
// 注释原因：开发环境经常出现卡状态不在预期可操作范围内的情况，先放行
|
|
_ = operation
|
|
_ = row
|
|
_ = card
|
|
// switch operation {
|
|
// case OperationSuspend:
|
|
// if card.BossCardStatus != consts.V3ApiSimBossCardStatusActivated {
|
|
// row.AddErrorReason("仅已激活状态的卡可以自主停机")
|
|
// }
|
|
// case OperationResume:
|
|
// if card.BossCardStatus != consts.V3ApiSimBossCardStatusStopped {
|
|
// row.AddErrorReason("仅停机状态的卡可以自主复机")
|
|
// }
|
|
// }
|
|
}
|
|
|
|
func groupRowsByChannel(rows []*Row) map[string][]*Row {
|
|
groups := make(map[string][]*Row)
|
|
for _, row := range rows {
|
|
groups[row.ChanKey] = append(groups[row.ChanKey], row)
|
|
}
|
|
return groups
|
|
}
|
|
|
|
func (e *Executor) changeGroup(ctx context.Context, task *Task, rows []*Row) []Outcome {
|
|
simKeys := make([]*model.SimKey, 0, len(rows))
|
|
for _, row := range rows {
|
|
simKeys = append(simKeys, &model.SimKey{
|
|
ChanKey: row.ChanKey,
|
|
ICCID: row.Iccid,
|
|
MSISDN: row.Msisdn,
|
|
IMSI: row.Imsi,
|
|
})
|
|
}
|
|
response, err := e.upstream.ChangeSimStatus(ctx, &model.SimCardStatusChangeInput{
|
|
SimKeys: simKeys,
|
|
ReqData: model.SimCardStatusChangeInputItem{OperType: task.Config.OperType},
|
|
})
|
|
if err != nil {
|
|
outcomes := make([]Outcome, len(rows))
|
|
for i := range outcomes {
|
|
outcomes[i] = Outcome{Message: "上游请求失败: " + err.Error()}
|
|
}
|
|
return outcomes
|
|
}
|
|
return mapOutcomes(rows, response)
|
|
}
|
|
|
|
func mapOutcomes(rows []*Row, response []model.SimBaseRes[model.SimCardStatusChangeItem]) []Outcome {
|
|
outcomes := make([]Outcome, len(rows))
|
|
used := make([]bool, len(response))
|
|
for rowIndex, row := range rows {
|
|
matched := -1
|
|
for responseIndex, item := range response {
|
|
if !used[responseIndex] && sameSim(row, item.SimKey) {
|
|
matched = responseIndex
|
|
break
|
|
}
|
|
}
|
|
if matched < 0 && rowIndex < len(response) && !used[rowIndex] && response[rowIndex].SimKey == nil {
|
|
matched = rowIndex
|
|
}
|
|
if matched < 0 {
|
|
outcomes[rowIndex] = Outcome{Message: "上游未返回该卡处理结果"}
|
|
continue
|
|
}
|
|
used[matched] = true
|
|
item := response[matched]
|
|
message := strings.TrimSpace(item.Message)
|
|
if item.Code != 0 && message == "" {
|
|
message = "上游处理失败"
|
|
}
|
|
outcomes[rowIndex] = Outcome{
|
|
Success: item.Code == 0,
|
|
Code: item.Code,
|
|
Message: message,
|
|
OrderNum: strings.TrimSpace(item.ResData.OrderNum),
|
|
}
|
|
}
|
|
return outcomes
|
|
}
|
|
|
|
func sameSim(row *Row, key *model.SimKey) bool {
|
|
if row == nil || key == nil {
|
|
return false
|
|
}
|
|
if value := strings.TrimSpace(key.ICCID); value != "" {
|
|
return value == row.Iccid
|
|
}
|
|
if value := strings.TrimSpace(key.MSISDN); value != "" {
|
|
return value == row.Msisdn
|
|
}
|
|
if value := strings.TrimSpace(key.IMSI); value != "" {
|
|
return value == row.Imsi
|
|
}
|
|
return false
|
|
}
|