MiFaZhan
832 lines
27 KiB
Go
832 lines
27 KiB
Go
package addpackage
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"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/batch"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/library/rabbitmq"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
|
|
rc "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/renewchange"
|
|
renewchangeservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/renewchange"
|
|
)
|
|
|
|
const addPackagePrepareProcessingRetryAfter = 2 * time.Minute
|
|
|
|
type PrepareExecutor struct {
|
|
repo *renewchangeservice.Repository
|
|
inner *executor
|
|
targetBatchType int
|
|
nowFunc NowFunc
|
|
}
|
|
|
|
func NewCardPrepareExecutor(repo *renewchangeservice.Repository) *PrepareExecutor {
|
|
if repo == nil {
|
|
repo = renewchangeservice.NewRepository()
|
|
}
|
|
return &PrepareExecutor{
|
|
repo: repo,
|
|
inner: &executor{repo: repo, batchType: consts.BatchTypePrepare, nowFunc: timeNow},
|
|
targetBatchType: consts.BatchTypeCardAddPackage,
|
|
nowFunc: timeNow,
|
|
}
|
|
}
|
|
|
|
func NewPoolPrepareExecutor(repo *renewchangeservice.Repository) *PrepareExecutor {
|
|
if repo == nil {
|
|
repo = renewchangeservice.NewRepository()
|
|
}
|
|
return &PrepareExecutor{
|
|
repo: repo,
|
|
inner: &executor{repo: repo, batchType: consts.BatchTypePrepare, poolMode: true, nowFunc: timeNow},
|
|
targetBatchType: consts.BatchTypePoolAddPackage,
|
|
nowFunc: timeNow,
|
|
}
|
|
}
|
|
|
|
func (p *PrepareExecutor) CanExecuteProcessing(_ context.Context, main *entity.BatchMain) (bool, error) {
|
|
if main == nil || main.BatchType != consts.BatchTypePrepare || main.BatchAsyncStatus != consts.BatchAsyncStatusProcessing {
|
|
return false, nil
|
|
}
|
|
if targetBatchTypeFromInfo(main.BatchInfo) != p.targetBatchType {
|
|
return false, nil
|
|
}
|
|
if main.StartTime == nil || main.StartTime.IsZero() {
|
|
return true, nil
|
|
}
|
|
return p.now().Sub(main.StartTime.Time) >= addPackagePrepareProcessingRetryAfter, nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) Execute(ctx context.Context, batchNo string) error {
|
|
batchNo = strings.TrimSpace(batchNo)
|
|
if batchNo == "" {
|
|
return gerror.New("add package prepare batch number is empty")
|
|
}
|
|
main, err := p.repo.LoadProcessingBatch(ctx, batchNo, consts.BatchTypePrepare)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if targetBatchTypeFromInfo(main.BatchInfo) != p.targetBatchType {
|
|
return p.finalizePrepareFailure(ctx, main, "加油包预处理批次类型不匹配", nil)
|
|
}
|
|
details, err := p.repo.LoadRunningDetails(ctx, batchNo, consts.BatchTypePrepare)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "load add package prepare details failed")
|
|
}
|
|
task := &batchTask{
|
|
main: main,
|
|
total: len(details),
|
|
plans: make([]*detailPlan, 0, len(details)),
|
|
paymentIntent: readPaymentIntent(main.BatchInfo),
|
|
paymentSnapshot: make(map[int64]*PaymentSnapshot),
|
|
prepareOnly: true,
|
|
}
|
|
if len(details) == 0 {
|
|
return p.finalizePrepareFailure(ctx, main, "加油包预处理批次没有可处理明细", task)
|
|
}
|
|
p.inner.nowFunc = p.now
|
|
if err := p.inner.buildTask(ctx, task, details); err != nil {
|
|
return p.finalizePrepareFailure(ctx, main, err.Error(), task)
|
|
}
|
|
if task.fail > 0 || task.valid != len(details) {
|
|
reason := strings.TrimSpace(task.reason)
|
|
if reason == "" {
|
|
reason = "加油包预处理存在不满足条件的明细"
|
|
}
|
|
return p.finalizePrepareFailure(ctx, main, reason, task)
|
|
}
|
|
return p.finalizePrepareSuccess(ctx, task)
|
|
}
|
|
|
|
func (p *PrepareExecutor) finalizePrepareFailure(ctx context.Context, main *rc.BatchMainRow, reason string, task *batchTask) error {
|
|
if main == nil {
|
|
return gerror.New("add package prepare batch is empty")
|
|
}
|
|
if isAutoConfirmPrepare(main) {
|
|
return p.finalizeAutoPrepareFailure(ctx, main, reason, task)
|
|
}
|
|
reason = strings.TrimSpace(reason)
|
|
if reason == "" {
|
|
reason = "加油包预处理失败"
|
|
}
|
|
now := p.now()
|
|
failCount := main.TotalCount
|
|
err := p.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
if task != nil {
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.detail == nil {
|
|
continue
|
|
}
|
|
rowReason := strings.TrimSpace(plan.reason)
|
|
if rowReason == "" {
|
|
rowReason = reason
|
|
}
|
|
if err := p.repo.Tables().UpdateBatchDetailSnapshot(ctx, tx, main.BatchNo, main.BatchType, plan.detail.RowNo, consts.BatchDetailStatusFailed, buildDetailSnapshot(plan, ""), rowReason, main.CreatedBy); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
if err := p.repo.MarkAllDetailsFailed(ctx, tx, main.BatchNo, main.BatchType, reason, main.CreatedBy); err != nil {
|
|
return err
|
|
}
|
|
return p.repo.FinalizeBatchMain(ctx, tx, main.BatchNo, main.BatchType, &rc.BatchMainRow{
|
|
BatchAsyncStatus: consts.BatchAsyncStatusFailed,
|
|
RunResult: consts.BatchRunResultFailed,
|
|
FailReason: reason,
|
|
TotalCount: main.TotalCount,
|
|
ValidCount: 0,
|
|
SuccessCount: 0,
|
|
FailCount: failCount,
|
|
EndTime: gtime.NewFromTime(now),
|
|
UpdatedBy: main.CreatedBy,
|
|
})
|
|
})
|
|
return gerror.Wrap(err, "finalize add package prepare failure failed")
|
|
}
|
|
|
|
func (p *PrepareExecutor) finalizePrepareSuccess(ctx context.Context, task *batchTask) error {
|
|
if task == nil || task.main == nil {
|
|
return gerror.New("add package prepare task is empty")
|
|
}
|
|
now := p.now()
|
|
info, err := p.buildPrepareInfo(ctx, task)
|
|
if err != nil {
|
|
if isAutoConfirmPrepare(task.main) {
|
|
return p.finalizeAutoPrepareFailure(ctx, task.main, err.Error(), task)
|
|
}
|
|
return err
|
|
}
|
|
if isAutoConfirmPrepare(task.main) {
|
|
return p.finalizeAutoPrepareSuccess(ctx, task, info, now)
|
|
}
|
|
err = p.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.detail == nil {
|
|
continue
|
|
}
|
|
if err := p.repo.Tables().UpdateBatchDetailSnapshot(ctx, tx, task.main.BatchNo, task.main.BatchType, plan.detail.RowNo, consts.BatchDetailStatusSuccess, buildDetailSnapshot(plan, ""), "", task.main.CreatedBy); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := p.repo.MarkAllDetailsSuccess(ctx, tx, task.main.BatchNo, task.main.BatchType, task.main.CreatedBy, nil); err != nil {
|
|
return err
|
|
}
|
|
return p.repo.UpdateBatchMainInfo(ctx, tx, task.main.BatchNo, task.main.BatchType, &rc.BatchMainRow{
|
|
BatchType: task.main.BatchType,
|
|
BatchAsyncStatus: consts.BatchAsyncStatusSuccess,
|
|
RunResult: consts.BatchRunResultSuccess,
|
|
TotalCount: task.total,
|
|
ValidCount: task.valid,
|
|
SuccessCount: task.valid,
|
|
FailCount: 0,
|
|
FailReason: "",
|
|
BatchInfo: gjson.New(info),
|
|
EndTime: gtime.NewFromTime(now),
|
|
UpdatedBy: task.main.CreatedBy,
|
|
})
|
|
})
|
|
return gerror.Wrap(err, "finalize add package prepare success failed")
|
|
}
|
|
|
|
func (p *PrepareExecutor) finalizeAutoPrepareSuccess(ctx context.Context, task *batchTask, info map[string]any, now time.Time) error {
|
|
if task == nil || task.main == nil {
|
|
return gerror.New("add package auto prepare task is empty")
|
|
}
|
|
targetBatchType := p.targetBatchType
|
|
info["autoConfirmed"] = true
|
|
info["prepareBatchType"] = consts.BatchTypePrepare
|
|
info["status"] = "pending_execute"
|
|
err := p.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.detail == nil {
|
|
continue
|
|
}
|
|
if err := p.repo.Tables().UpdateBatchDetailSnapshot(ctx, tx, task.main.BatchNo, task.main.BatchType, plan.detail.RowNo, consts.BatchDetailStatusSuccess, buildDetailSnapshot(plan, ""), "", task.main.CreatedBy); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := convertAddPackagePrepareDetails(ctx, tx, task.main.BatchNo, targetBatchType, consts.BatchDetailStatusPending, "", task.main.CreatedBy, now); err != nil {
|
|
return err
|
|
}
|
|
return convertAddPackagePrepareMain(ctx, tx, task.main.BatchNo, targetBatchType, &rc.BatchMainRow{
|
|
BatchAsyncStatus: batch.AsyncStatusPending,
|
|
RunResult: 0,
|
|
TotalCount: task.total,
|
|
ValidCount: 0,
|
|
SuccessCount: 0,
|
|
FailCount: 0,
|
|
FailReason: "",
|
|
BatchInfo: gjson.New(info),
|
|
UpdatedBy: task.main.CreatedBy,
|
|
})
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "auto confirm add package prepare failed")
|
|
}
|
|
p.publishExecution(ctx, task.main.BatchNo, targetBatchType)
|
|
g.Log().Infof(ctx, "add package prepare auto confirmed, batchNo:%s, targetType:%d", task.main.BatchNo, targetBatchType)
|
|
return nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) finalizeAutoPrepareFailure(ctx context.Context, main *rc.BatchMainRow, reason string, task *batchTask) error {
|
|
reason = strings.TrimSpace(reason)
|
|
if reason == "" {
|
|
reason = "加油包预处理失败"
|
|
}
|
|
now := p.now()
|
|
targetBatchType := p.targetBatchType
|
|
info := mergeBatchInfo(main.BatchInfo, map[string]any{
|
|
"targetBatchType": targetBatchType,
|
|
"prepareBatchType": consts.BatchTypePrepare,
|
|
"autoConfirm": true,
|
|
"autoConfirmed": false,
|
|
"status": "failed",
|
|
})
|
|
err := p.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
|
if task != nil {
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.detail == nil {
|
|
continue
|
|
}
|
|
rowReason := strings.TrimSpace(plan.reason)
|
|
if rowReason == "" {
|
|
rowReason = reason
|
|
}
|
|
if err := p.repo.Tables().UpdateBatchDetailSnapshot(ctx, tx, main.BatchNo, main.BatchType, plan.detail.RowNo, consts.BatchDetailStatusFailed, buildDetailSnapshot(plan, ""), rowReason, main.CreatedBy); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
if err := p.repo.MarkAllDetailsFailed(ctx, tx, main.BatchNo, main.BatchType, reason, main.CreatedBy); err != nil {
|
|
return err
|
|
}
|
|
detailReason := reason
|
|
if task != nil && len(task.plans) > 0 {
|
|
detailReason = ""
|
|
}
|
|
if err := convertAddPackagePrepareDetails(ctx, tx, main.BatchNo, targetBatchType, consts.BatchDetailStatusFailed, detailReason, main.CreatedBy, now); err != nil {
|
|
return err
|
|
}
|
|
return convertAddPackagePrepareMain(ctx, tx, main.BatchNo, targetBatchType, &rc.BatchMainRow{
|
|
BatchAsyncStatus: consts.BatchAsyncStatusFailed,
|
|
RunResult: consts.BatchRunResultFailed,
|
|
FailReason: reason,
|
|
TotalCount: main.TotalCount,
|
|
ValidCount: 0,
|
|
SuccessCount: 0,
|
|
FailCount: main.TotalCount,
|
|
BatchInfo: gjson.New(info),
|
|
EndTime: gtime.NewFromTime(now),
|
|
UpdatedBy: main.CreatedBy,
|
|
})
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "finalize auto add package prepare failure failed")
|
|
}
|
|
rows, _ := p.repo.ResolveResultSnapshot(ctx, main.BatchNo, targetBatchType)
|
|
_ = p.repo.WriteResultFileSafe(ctx, main.BatchNo, targetBatchType, main.TenantId, main.CreatedDept, main.CreatedBy, rows)
|
|
g.Log().Warningf(ctx, "add package prepare auto failed, batchNo:%s, targetType:%d, reason:%s", main.BatchNo, targetBatchType, reason)
|
|
return nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) publishExecution(ctx context.Context, batchNo string, batchType int) {
|
|
if err := rabbitmq.Publish(ctx, consts.QueueBatchExecute, batch.Message{BatchNo: batchNo, BatchType: batchType}); err != nil {
|
|
g.Log().Errorf(ctx, "publish add package execution after prepare failed, batchNo:%s, batchType:%d, err:%+v", batchNo, batchType, err)
|
|
}
|
|
}
|
|
|
|
func (p *PrepareExecutor) buildPrepareInfo(ctx context.Context, task *batchTask) (map[string]any, error) {
|
|
info := mergeBatchInfo(task.main.BatchInfo, map[string]any{
|
|
"targetBatchType": p.targetBatchType,
|
|
"batchMode": map[bool]string{true: "pool", false: "card"}[p.inner.poolMode],
|
|
})
|
|
quantity := 0
|
|
totalAmount := 0.0
|
|
var sample *detailPlan
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.reason != "" {
|
|
continue
|
|
}
|
|
if sample == nil {
|
|
sample = plan
|
|
}
|
|
quantity++
|
|
if plan.price == nil {
|
|
continue
|
|
}
|
|
totalAmount = roundAmount(totalAmount + plan.amount)
|
|
}
|
|
if task.prepareOnly && sample != nil && strings.TrimSpace(sample.packageCode) == "" {
|
|
options, err := p.prepareAvailablePackages(ctx, task)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(options) == 0 {
|
|
return nil, gerror.New("当前没有共同可购的加油包套餐")
|
|
}
|
|
info["quantity"] = quantity
|
|
info["basePackageKey"] = sample.basePackageKey
|
|
info["availablePackages"] = options
|
|
info["packageOptions"] = options
|
|
if _, err := p.fillPreparePaymentView(ctx, task, info); err != nil {
|
|
return nil, err
|
|
}
|
|
info["prepareTime"] = p.now().Format("2006-01-02 15:04:05")
|
|
info["status"] = "success"
|
|
return info, nil
|
|
}
|
|
if sample == nil || sample.price == nil {
|
|
return nil, gerror.New("加油包预处理没有可确认的明细")
|
|
}
|
|
info["quantity"] = quantity
|
|
info["basePackageKey"] = sample.basePackageKey
|
|
info["packageCode"] = sample.packageCode
|
|
info["packageName"] = sample.price.PackageName
|
|
info["flowMb"] = sample.price.FlowMb
|
|
info["unitPrice"] = sample.price.UnitPrice
|
|
info["totalAmount"] = totalAmount
|
|
info["effectiveMode"] = sample.effectiveMode
|
|
info["effectiveModeText"] = effectiveModeText(sample.effectiveMode)
|
|
info["effectiveTime"] = gtime.NewFromTime(sample.effectiveTime).Format("Y-m-d H:i:s")
|
|
info["expectedExpireTime"] = gtime.NewFromTime(sample.expireTime).Format("Y-m-d H:i:s")
|
|
info["priceSnapshot"] = sample.price
|
|
accounts, err := p.preparePaymentAccounts(ctx, task)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
info["paymentAccounts"] = accounts
|
|
isTop, err := p.fillPreparePaymentView(ctx, task, info)
|
|
if isAutoConfirmPrepare(task.main) {
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
intent, err := normalizePaymentIntent(task.paymentIntent, isTop)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
task.paymentIntent = intent
|
|
info["paymentMode"] = intent.PaymentMode
|
|
info["offlineContractCompleted"] = intent.OfflineContractCompleted
|
|
if intent.PaymentMode == "balance" {
|
|
if err := validatePrepareBalanceEnough(accounts); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
info["prepareTime"] = p.now().Format("2006-01-02 15:04:05")
|
|
info["status"] = "success"
|
|
return info, nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) fillPreparePaymentView(ctx context.Context, task *batchTask, info map[string]any) (bool, error) {
|
|
isTop, err := p.inner.isTopDept(ctx, task.main.TenantId, task.main.CreatedDept)
|
|
if err == nil {
|
|
info["isTopDept"] = isTop
|
|
info["defaultPaymentMode"] = "balance"
|
|
}
|
|
balance, balanceErr := p.availableBalance(ctx, task.main.TenantId, task.main.CreatedDept)
|
|
if balanceErr != nil {
|
|
return isTop, balanceErr
|
|
}
|
|
info["availableBalance"] = balance
|
|
return isTop, err
|
|
}
|
|
|
|
func (p *PrepareExecutor) prepareAvailablePackages(ctx context.Context, task *batchTask) ([]map[string]any, error) {
|
|
sample := firstValidAddPackagePlan(task)
|
|
if sample == nil {
|
|
return nil, gerror.New("加油包预处理没有可确认的明细")
|
|
}
|
|
codes, err := p.availablePackageCodes(ctx, task.main.TenantId, sample)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
options := make([]map[string]any, 0, len(codes))
|
|
for _, code := range codes {
|
|
option, err := p.buildAvailablePackageOption(ctx, task, code)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if option != nil {
|
|
options = append(options, option)
|
|
}
|
|
}
|
|
return options, nil
|
|
}
|
|
|
|
func firstValidAddPackagePlan(task *batchTask) *detailPlan {
|
|
if task == nil {
|
|
return nil
|
|
}
|
|
for _, plan := range task.plans {
|
|
if plan != nil && plan.reason == "" {
|
|
return plan
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) availablePackageCodes(ctx context.Context, tenantID string, sample *detailPlan) ([]string, error) {
|
|
if p.inner.poolMode {
|
|
return p.availablePoolPackageCodes(ctx, tenantID, sample)
|
|
}
|
|
basePkg, err := p.repo.Tables().LoadPackageByCode(ctx, tenantID, sample.basePackageKey)
|
|
if err != nil || basePkg == nil {
|
|
return nil, gerror.New("当前卡基础套餐不存在")
|
|
}
|
|
codes := packageCodesFromJSON(basePkg.TopupPackageCodes)
|
|
if len(codes) == 0 {
|
|
return nil, gerror.New("当前卡基础套餐未配置可购加油包")
|
|
}
|
|
return codes, nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) availablePoolPackageCodes(ctx context.Context, tenantID string, sample *detailPlan) ([]string, error) {
|
|
if sample.pool == nil || sample.pool.PoolPackageCode <= 0 {
|
|
return nil, gerror.New("流量池未配置基础套餐")
|
|
}
|
|
var packages []*rc.FlowPackageInfo
|
|
if err := p.repo.DB().Model(rc.TableSimFlowPackageInfo).Safe().Ctx(ctx).
|
|
Where("tenant_id", tenantID).
|
|
Where("pool_package_code", sample.pool.PoolPackageCode).
|
|
Where("package_type", consts.PackageTypePoolAddPackage).
|
|
Where("enabled", 1).
|
|
OrderAsc("sort_no").
|
|
Scan(&packages); err != nil {
|
|
return nil, gerror.Wrap(err, "查询流量池加油包套餐失败")
|
|
}
|
|
codes := make([]string, 0, len(packages))
|
|
for _, pkg := range packages {
|
|
if pkg == nil {
|
|
continue
|
|
}
|
|
if code := strings.TrimSpace(pkg.PackageCode); code != "" {
|
|
codes = append(codes, code)
|
|
}
|
|
}
|
|
if len(codes) == 0 {
|
|
return nil, gerror.New("所选流量池没有共同可购的加油包套餐")
|
|
}
|
|
return codes, nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) buildAvailablePackageOption(ctx context.Context, task *batchTask, packageCode string) (map[string]any, error) {
|
|
packageCode = strings.TrimSpace(packageCode)
|
|
if packageCode == "" {
|
|
return nil, nil
|
|
}
|
|
var (
|
|
samplePrice *PriceSnapshot
|
|
totalAmount float64
|
|
unitPrice float64
|
|
priceAvailable = true
|
|
priceReason string
|
|
accounts = make(map[int64]*prepareAccount)
|
|
deptIDs = make([]int64, 0)
|
|
)
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.reason != "" {
|
|
continue
|
|
}
|
|
pkg, price, err := p.inner.resolvePrice(ctx, task.main.TenantId, plan.customerDeptID, packageCode)
|
|
if err != nil {
|
|
priceAvailable = false
|
|
if priceReason == "" {
|
|
priceReason = err.Error()
|
|
}
|
|
continue
|
|
}
|
|
if p.inner.poolMode {
|
|
if pkg.PackageType != consts.PackageTypePoolAddPackage || plan.pool == nil || pkg.PoolPackageCode != plan.pool.PoolPackageCode {
|
|
priceAvailable = false
|
|
priceReason = "加油包不属于当前流量池套餐"
|
|
continue
|
|
}
|
|
} else if pkg.PackageType != consts.PackageTypeCardAddPackage {
|
|
priceAvailable = false
|
|
priceReason = "目标套餐类型不是单卡加油包"
|
|
continue
|
|
}
|
|
if samplePrice == nil {
|
|
samplePrice = price
|
|
unitPrice = price.UnitPrice
|
|
} else if !sameAddPackageAmount(unitPrice, price.UnitPrice) {
|
|
priceAvailable = false
|
|
priceReason = addPackageSamePriceFailureReason
|
|
}
|
|
totalAmount = roundAmount(totalAmount + price.UnitPrice)
|
|
account := accounts[plan.customerDeptID]
|
|
if account == nil {
|
|
account = &prepareAccount{payerDeptID: plan.customerDeptID}
|
|
accounts[plan.customerDeptID] = account
|
|
deptIDs = append(deptIDs, plan.customerDeptID)
|
|
}
|
|
account.count++
|
|
account.required = roundAmount(account.required + price.UnitPrice)
|
|
}
|
|
if samplePrice == nil {
|
|
return nil, nil
|
|
}
|
|
paymentAccounts, err := p.prepareAccountItems(ctx, task.main.TenantId, accounts, deptIDs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
option := map[string]any{
|
|
"packageCode": samplePrice.PackageCode,
|
|
"packageName": samplePrice.PackageName,
|
|
"flowMb": samplePrice.FlowMb,
|
|
"unitPrice": unitPrice,
|
|
"totalAmount": totalAmount,
|
|
"priceAvailable": priceAvailable,
|
|
"paymentAccounts": paymentAccounts,
|
|
}
|
|
if priceReason != "" {
|
|
option["priceReason"] = priceReason
|
|
}
|
|
return option, nil
|
|
}
|
|
|
|
type prepareAccount struct {
|
|
payerDeptID int64
|
|
count int
|
|
required float64
|
|
}
|
|
|
|
func (p *PrepareExecutor) prepareAccountItems(ctx context.Context, tenantID string, accounts map[int64]*prepareAccount, deptIDs []int64) ([]map[string]any, error) {
|
|
sort.Slice(deptIDs, func(i, j int) bool { return deptIDs[i] < deptIDs[j] })
|
|
out := make([]map[string]any, 0, len(deptIDs))
|
|
for _, deptID := range deptIDs {
|
|
item := accounts[deptID]
|
|
if item == nil {
|
|
continue
|
|
}
|
|
balance, err := p.availableBalance(ctx, tenantID, deptID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, map[string]any{
|
|
"payerDeptId": item.payerDeptID,
|
|
"targetCount": item.count,
|
|
"requiredAmount": item.required,
|
|
"availableBalance": balance,
|
|
"balanceEnough": balance+0.0001 >= item.required,
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func validatePrepareBalanceEnough(accounts []map[string]any) error {
|
|
for _, account := range accounts {
|
|
if account == nil {
|
|
continue
|
|
}
|
|
if enough, _ := account["balanceEnough"].(bool); enough {
|
|
continue
|
|
}
|
|
return gerror.Newf(
|
|
"账户余额不足，付款部门%v需支付%.2f元，当前余额%.2f元",
|
|
account["payerDeptId"],
|
|
floatFromAny(account["requiredAmount"]),
|
|
floatFromAny(account["availableBalance"]),
|
|
)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func floatFromAny(value any) float64 {
|
|
switch v := value.(type) {
|
|
case float64:
|
|
return v
|
|
case float32:
|
|
return float64(v)
|
|
case int:
|
|
return float64(v)
|
|
case int64:
|
|
return float64(v)
|
|
case int32:
|
|
return float64(v)
|
|
case uint:
|
|
return float64(v)
|
|
case uint64:
|
|
return float64(v)
|
|
case uint32:
|
|
return float64(v)
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func (p *PrepareExecutor) preparePaymentAccounts(ctx context.Context, task *batchTask) ([]map[string]any, error) {
|
|
type account struct {
|
|
payerDeptID int64
|
|
count int
|
|
required float64
|
|
}
|
|
byDept := make(map[int64]*account)
|
|
deptIDs := make([]int64, 0)
|
|
for _, plan := range task.plans {
|
|
if plan == nil || plan.reason != "" {
|
|
continue
|
|
}
|
|
payerDeptID := plan.customerDeptID
|
|
if payerDeptID <= 0 {
|
|
return nil, gerror.New("加油包付款部门不能为空")
|
|
}
|
|
item := byDept[payerDeptID]
|
|
if item == nil {
|
|
item = &account{payerDeptID: payerDeptID}
|
|
byDept[payerDeptID] = item
|
|
deptIDs = append(deptIDs, payerDeptID)
|
|
}
|
|
item.count++
|
|
item.required = roundAmount(item.required + plan.amount)
|
|
}
|
|
sort.Slice(deptIDs, func(i, j int) bool { return deptIDs[i] < deptIDs[j] })
|
|
out := make([]map[string]any, 0, len(deptIDs))
|
|
for _, deptID := range deptIDs {
|
|
item := byDept[deptID]
|
|
balance, err := p.availableBalance(ctx, task.main.TenantId, deptID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, map[string]any{
|
|
"payerDeptId": item.payerDeptID,
|
|
"targetCount": item.count,
|
|
"requiredAmount": item.required,
|
|
"availableBalance": balance,
|
|
"balanceEnough": balance+0.0001 >= item.required,
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) availableBalance(ctx context.Context, tenantID string, deptID int64) (float64, error) {
|
|
if deptID <= 0 {
|
|
return 0, nil
|
|
}
|
|
var row agentBalanceRow
|
|
if err := p.repo.DB().Model("t_agent_balance").Ctx(ctx).
|
|
Where("tenant_id", tenantID).
|
|
Where("dept_id", deptID).
|
|
Scan(&row); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return 0, nil
|
|
}
|
|
return 0, gerror.Wrap(err, "query add package balance failed")
|
|
}
|
|
if row.Id == 0 {
|
|
return 0, nil
|
|
}
|
|
return roundAmount(row.Balance), nil
|
|
}
|
|
|
|
func (p *PrepareExecutor) now() time.Time {
|
|
if p == nil || p.nowFunc == nil {
|
|
return time.Now()
|
|
}
|
|
return p.nowFunc()
|
|
}
|
|
|
|
type PrepareMux struct {
|
|
repo *renewchangeservice.Repository
|
|
renew batch.Executor
|
|
card *PrepareExecutor
|
|
pool *PrepareExecutor
|
|
}
|
|
|
|
func NewPrepareMux(repo *renewchangeservice.Repository, renew batch.Executor) *PrepareMux {
|
|
if repo == nil {
|
|
repo = renewchangeservice.NewRepository()
|
|
}
|
|
return &PrepareMux{
|
|
repo: repo,
|
|
renew: renew,
|
|
card: NewCardPrepareExecutor(repo),
|
|
pool: NewPoolPrepareExecutor(repo),
|
|
}
|
|
}
|
|
|
|
func (m *PrepareMux) Execute(ctx context.Context, batchNo string) error {
|
|
main, err := m.repo.LoadProcessingBatch(ctx, strings.TrimSpace(batchNo), consts.BatchTypePrepare)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch targetBatchTypeFromInfo(main.BatchInfo) {
|
|
case consts.BatchTypeCardAddPackage:
|
|
return m.card.Execute(ctx, batchNo)
|
|
case consts.BatchTypePoolAddPackage:
|
|
return m.pool.Execute(ctx, batchNo)
|
|
default:
|
|
if m.renew != nil {
|
|
return m.renew.Execute(ctx, batchNo)
|
|
}
|
|
return gerror.Newf("unsupported prepare target batch type: %d", targetBatchTypeFromInfo(main.BatchInfo))
|
|
}
|
|
}
|
|
|
|
func (m *PrepareMux) CanExecuteProcessing(ctx context.Context, main *entity.BatchMain) (bool, error) {
|
|
if main == nil || main.BatchType != consts.BatchTypePrepare || main.BatchAsyncStatus != consts.BatchAsyncStatusProcessing {
|
|
return false, nil
|
|
}
|
|
switch targetBatchTypeFromInfo(main.BatchInfo) {
|
|
case consts.BatchTypeCardAddPackage:
|
|
return m.card.CanExecuteProcessing(ctx, main)
|
|
case consts.BatchTypePoolAddPackage:
|
|
return m.pool.CanExecuteProcessing(ctx, main)
|
|
default:
|
|
if processing, ok := m.renew.(batch.ProcessingExecutor); ok {
|
|
return processing.CanExecuteProcessing(ctx, main)
|
|
}
|
|
return false, nil
|
|
}
|
|
}
|
|
|
|
func targetBatchTypeFromInfo(info *gjson.Json) int {
|
|
if info == nil || info.IsNil() {
|
|
return 0
|
|
}
|
|
if v := info.Get("targetBatchType"); v != nil && !v.IsNil() {
|
|
return v.Int()
|
|
}
|
|
if v := info.Get("target_batch_type"); v != nil && !v.IsNil() {
|
|
return v.Int()
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func isAutoConfirmPrepare(main *rc.BatchMainRow) bool {
|
|
if main == nil || main.SourceType != consts.BatchSourceTypeFile || main.BatchInfo == nil || main.BatchInfo.IsNil() {
|
|
return false
|
|
}
|
|
return main.BatchInfo.Get("autoConfirm").Bool()
|
|
}
|
|
|
|
func convertAddPackagePrepareDetails(ctx context.Context, tx gdb.TX, batchNo string, targetBatchType int, detailStatus int, reason string, userID int64, now time.Time) error {
|
|
data := map[string]any{
|
|
"batch_type": targetBatchType,
|
|
"detail_status": detailStatus,
|
|
"updated_by": userID,
|
|
"updated_at": gtime.NewFromTime(now),
|
|
}
|
|
if strings.TrimSpace(reason) != "" {
|
|
data["error_reason"] = reason
|
|
}
|
|
_, err := tx.Ctx(ctx).Model(rc.TableBatchDetail).
|
|
Where("batch_no", batchNo).
|
|
Where("batch_type", consts.BatchTypePrepare).
|
|
Data(data).
|
|
Update()
|
|
return gerror.Wrap(err, "convert add package prepare details failed")
|
|
}
|
|
|
|
func convertAddPackagePrepareMain(ctx context.Context, tx gdb.TX, batchNo string, targetBatchType int, data *rc.BatchMainRow) error {
|
|
if data == nil {
|
|
return gerror.New("add package batch conversion data is empty")
|
|
}
|
|
update := map[string]any{
|
|
"batch_type": targetBatchType,
|
|
"batch_async_status": data.BatchAsyncStatus,
|
|
"run_result": data.RunResult,
|
|
"total_count": data.TotalCount,
|
|
"valid_count": data.ValidCount,
|
|
"success_count": data.SuccessCount,
|
|
"fail_count": data.FailCount,
|
|
"fail_reason": data.FailReason,
|
|
"batch_info": data.BatchInfo,
|
|
"start_time": nil,
|
|
"end_time": data.EndTime,
|
|
"updated_by": data.UpdatedBy,
|
|
"updated_at": gtime.Now(),
|
|
}
|
|
if data.EndTime == nil {
|
|
update["end_time"] = nil
|
|
}
|
|
result, err := tx.Ctx(ctx).Model(rc.TableBatchMain).
|
|
Where("batch_no", batchNo).
|
|
Where("batch_type", consts.BatchTypePrepare).
|
|
Data(update).
|
|
Update()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "convert add package prepare main failed")
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return gerror.Wrap(err, "read add package prepare main conversion result failed")
|
|
}
|
|
if affected == 0 {
|
|
return gerror.New("加油包预处理批次已变化，请重试")
|
|
}
|
|
return nil
|
|
}
|