Files
iotlcm/internal/service/renewchange/prepare.go
T
MiFaZhan 2bec257289 chore: 初始开源发布
iotlcm-worker-svr:异步执行中枢(RabbitMQ 消费者 + 全量同步编排)。

- Go module:iotlcm-worker-svr → git.5iot.com/iotlcm-public/iotlcm-worker-svr(214 文件)
- 凭据与内部标识脱敏:MySQL/Redis 凭据、私网与生产 IP、内网域名、
  库名 xiujie_admin → iotlcm、哨兵名占位
- 159 处真实手机号 → 合法 11 位编造号(保留号段与后 4 位,维持互异,
  不破坏「校验 MSISDN 格式」等测试断言)
- 移除内部资产:CI 触发器、Jenkins 触发测试文档、预演门禁与镜像
  digest 记录
- Dockerfile 改用公共基础镜像(golang:1.25-alpine → alpine:3.23),
  移除写死的 git 凭据
- README 重写(架构/全量同步模型/配置/已知失败),补 Apache-2.0
  LICENSE,LF 行尾 + .gitattributes

已知:3 个测试包在源仓同一提交上同样失败(client/upstream、
service/allocate、service/inbound),属既有缺陷,本仓未引入亦未修复。
2026-10-22 16:37:56 +08:00

1300 lines
41 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package renewchange
import (
"context"
"sort"
"strconv"
"strings"
"time"
"github.com/gogf/gf/v2/container/gvar"
"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/model/entity"
rc "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/renewchange"
)
const prepareProcessingRetryAfter = 2 * time.Minute
// PrepareExecutor 处理 batch_type=4 的预处理批次。
// 它根据 batch_info.targetBatchType 把工作路由到 renew/change 的准备逻辑上。
//
// 成功:保持 batch_type=4, batch_async_status=Success, run_result=Success;
//
// 写 batch_info 报价/客户/套餐快照;不改业务表。
//
// 失败:转为目标业务类型后写 Failed;明细失败 + 原因;不改业务表。
// 旧/重复消息:被处理过的批次(async != Processing)会直接安全忽略。
type PrepareExecutor struct {
repo *Repository
nowFunc NowFunc
}
// NewPrepareExecutor 返回默认的预处理执行器。
func NewPrepareExecutor(repo *Repository) *PrepareExecutor {
if repo == nil {
repo = NewRepository()
}
return &PrepareExecutor{repo: repo, nowFunc: timeNow}
}
// SetNowFunc 注入时钟,便于测试。
func (p *PrepareExecutor) SetNowFunc(now NowFunc) { p.nowFunc = now }
// CanExecuteProcessing allows a redelivered message to reclaim a stale prepare
// batch after a worker crash or shutdown between claim and finalization.
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 main.StartTime == nil || main.StartTime.IsZero() {
return true, nil
}
return p.nowFunc().Sub(main.StartTime.Time) >= prepareProcessingRetryAfter, nil
}
// Execute 实现 batch.Executor。
func (p *PrepareExecutor) Execute(ctx context.Context, batchNo string) error {
batchNo = strings.TrimSpace(batchNo)
if batchNo == "" {
return gerror.New("prepare batch number is empty")
}
main, err := p.repo.LoadProcessingBatch(ctx, batchNo, consts.BatchTypePrepare)
if err != nil {
return err
}
targetType := readTargetBatchType(main.BatchInfo)
if targetType != consts.BatchTypeRenew && targetType != consts.BatchTypeChange {
return p.finalizeBatchFailure(ctx, main, "预处理批次缺少合法 targetBatchType(必须是 101 或 102)", nil)
}
details, err := p.repo.LoadRunningDetails(ctx, batchNo, consts.BatchTypePrepare)
if err != nil {
return gerror.Wrap(err, "load prepare details failed")
}
if len(details) == 0 {
return p.finalizeBatchFailure(ctx, main, "预处理批次无可处理的明细", nil)
}
options := loadPrepareOptions(main.BatchInfo)
result, err := p.validate(ctx, main, targetType, options, details)
if err != nil {
return gerror.Wrap(err, "prepare validate failed")
}
if result.hasFailure {
return p.finalizeBatchFailure(ctx, main, result.failureReason, mergeRowErrors(result.perRow, result.rowErrors))
}
return p.finalizeBatchSuccess(ctx, main, targetType, result, details)
}
type prepareOptions struct {
customerAgentID int64
targetAgentID int64
priceScope int
// change 专用
newPackageCode string
durationValue int
durationSet bool
}
type prepareOutcome struct {
hasFailure bool
failureReason string
perRow map[int]string
priceSnap *RenewPriceSnapshot
changeSnap *ChangePrepareSnapshot
changePrice *ChangePriceSnapshot
customer *customerInfo
packageInfo *packageInfo
rowErrors map[int]string
rows map[int]*RenewRow
}
type customerInfo struct {
AgentID int64 `json:"agentId"`
Name string `json:"name"`
DeptID int64 `json:"deptId"`
TenantID string `json:"tenantId"`
}
type packageInfo struct {
Code string `json:"code"`
Name string `json:"name"`
Type int `json:"type"`
FlowMb float64 `json:"flowMb"`
Duration int `json:"durationMonths"`
UnitPrice float64 `json:"unitPrice"`
}
const renewSamePeriodFailureReason = "同一批续费卡当前套餐开始时间和到期时间必须一致"
const renewExpireWindowMonths = 6
const renewExpireFailureReason = "距离到期超过6个月,不能续费"
func (p *PrepareExecutor) validate(ctx context.Context, main *rc.BatchMainRow, targetType int, options prepareOptions, details []*rc.BatchDetailRow) (*prepareOutcome, error) {
now := p.nowFunc()
rows := make([]*RenewRow, 0, len(details))
iccids := make([]string, 0, len(details))
rowErrors := map[int]string{}
for _, d := range details {
if d == nil {
continue
}
iccid := strings.TrimSpace(d.ItemId)
if iccid == "" && d.Snapshot != nil {
iccid = prepareJSONValue(d.Snapshot, "iccid", "ICCID")
}
row := &RenewRow{
RowNo: d.RowNo,
ICCID: iccid,
Reason: strings.TrimSpace(d.ErrorReason),
}
if d.Snapshot != nil {
row.MSISDN = prepareJSONValue(d.Snapshot, "msisdn", "MSISDN")
row.IMSI = prepareJSONValue(d.Snapshot, "imsi", "IMSI")
row.NewPkgCode = strings.TrimSpace(d.Snapshot.Get("newPackageCode").String())
row.PackageCode = strings.TrimSpace(d.Snapshot.Get("packageCode").String())
row.OldPkgCode = strings.TrimSpace(d.Snapshot.Get("oldPackageCode").String())
row.OldPkgName = strings.TrimSpace(d.Snapshot.Get("oldPackageName").String())
row.OutboundBatchNo = strings.TrimSpace(d.Snapshot.Get("outboundBatchNo").String())
row.SourceBatchNo = strings.TrimSpace(d.Snapshot.Get("sourceBatchNo").String())
}
rows = append(rows, row)
}
if err := p.resolvePrepareRowIdentities(ctx, main.TenantId, rows, rowErrors); err != nil {
return nil, err
}
iccids = collectPrepareIccids(rows, rowErrors)
cards, err := p.repo.Tables().LoadCardsByIccids(ctx, iccids)
if err != nil {
return nil, gerror.Wrap(err, "load sim cards failed")
}
var bindRows []*rc.SimCardBindPkg
if targetType == consts.BatchTypeRenew {
bindRows, err = p.repo.Tables().LoadRenewBindRowsByIccids(ctx, iccids)
} else {
bindRows, err = p.repo.Tables().LoadActiveBindRowsByIccids(ctx, iccids)
}
if err != nil {
return nil, gerror.Wrap(err, "load current subscriptions failed")
}
out := &prepareOutcome{
perRow: map[int]string{},
rowErrors: rowErrors,
rows: make(map[int]*RenewRow, len(rows)),
}
for _, row := range rows {
out.rows[row.RowNo] = row
}
sourceBatchNo := executionSourceBatchNo(main, main.BatchInfo)
// 选择基准套餐/价格:先按相同 packageCode 收敛;多个则统一失败。
packageCodes := map[string]struct{}{}
customerIDs := map[int64]struct{}{}
for _, row := range rows {
if row.Reason != "" {
out.rowErrors[row.RowNo] = row.Reason
}
}
for _, row := range rows {
if strings.TrimSpace(out.rowErrors[row.RowNo]) != "" {
continue
}
if row.ICCID == "" {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], "ICCID不能为空")
continue
}
card, ok := cards[row.ICCID]
if !ok || card == nil {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], rcLocalize("errCardNotFound"))
continue
}
if card.TenantId != "" && main.TenantId != "" && card.TenantId != main.TenantId {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], "ICCID不属于当前租户")
continue
}
if card.DeptId <= 0 {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], rcLocalize("errCardNotFound"))
continue
}
if strings.TrimSpace(card.CurrentPackageCode) == "" {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], rcLocalize("errCardNoCurrentPackage"))
continue
}
bind, futureBinds, bindErr := selectCurrentSubscription(bindRows, row.ICCID)
if bindErr != nil {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], bindErr.Error())
continue
}
if bindErr = validateCurrentSubscription(card, bind, subscriptionValidation{
ICCID: row.ICCID,
MSISDN: row.MSISDN,
IMSI: row.IMSI,
TenantID: main.TenantId,
CustomerDeptID: card.DeptId,
PackageCode: card.CurrentPackageCode,
SourceBatchNo: sourceBatchNo,
}); bindErr != nil {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], bindErr.Error())
continue
}
periodBind := bind
if targetType == consts.BatchTypeRenew {
periodBind = renewBaseSubscriptionForCard(card, bind, futureBinds)
if periodBind == nil || periodBind.ExpectedExpireTime == nil {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], "没有预计到期时间,不能续费")
continue
}
if renewExpireAfterAllowedWindow(periodBind.ExpectedExpireTime, now) {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], renewExpireFailureReason)
continue
}
}
row.MSISDN = strings.TrimSpace(card.Msisdn)
row.IMSI = strings.TrimSpace(card.Imsi)
row.CustomerID = card.DeptId
row.PackageCode = strings.TrimSpace(card.CurrentPackageCode)
row.OldPkgCode = strings.TrimSpace(periodBind.PackageCode)
row.OldBindSubNo = strings.TrimSpace(periodBind.SubscriptionNo)
row.OldBindOrdNo = strings.TrimSpace(periodBind.OrderCode)
row.OldEffTime = periodBind.EffectiveTime
row.OldExpireTime = periodBind.ExpectedExpireTime
row.SourceBatchNo = sourceBatchNo
customerIDs[card.DeptId] = struct{}{}
packageCodes[row.PackageCode] = struct{}{}
}
if setPrepareFailureFromRows(out, rows) {
return out, nil
}
// 续费/变更专属校验
if targetType == consts.BatchTypeRenew {
if len(packageCodes) > 1 {
out.rowErrors[0] = rcLocalize("errMultiplePackages")
out.hasFailure = true
out.failureReason = rcLocalize("errMultiplePackages")
return out, nil
}
if err := validateRenewSameCurrentPeriod(rows, out.rowErrors); err != nil {
out.rowErrors[0] = err.Error()
out.hasFailure = true
out.failureReason = err.Error()
return out, nil
}
} else {
if options.newPackageCode == "" {
out.rowErrors[0] = rcLocalize("errNewPackageNotInChange")
out.hasFailure = true
out.failureReason = "变更批次缺少新套餐编码"
return out, nil
}
}
// 同批客户单一性:多个客户按全批失败处理。
if len(customerIDs) > 1 {
out.rowErrors[0] = rcLocalize("errMultipleCustomers")
out.hasFailure = true
out.failureReason = rcLocalize("errMultipleCustomers")
return out, nil
}
// 选择客户
var customerID int64
for id := range customerIDs {
customerID = id
}
// 加载套餐、客户、价格
var (
curPkg *rc.FlowPackageInfo
newPkg *rc.FlowPackageInfo
)
for code := range packageCodes {
pkg, err := p.repo.Tables().LoadPackageByCode(ctx, main.TenantId, code)
if err != nil || pkg == nil {
out.rowErrors[0] = rcLocalize("errPkgNotFound")
out.hasFailure = true
out.failureReason = "当前套餐不存在或未启用"
return out, nil
}
if pkg.Enabled != 1 {
out.rowErrors[0] = rcLocalize("errPkgDisabled")
out.hasFailure = true
out.failureReason = "当前套餐未启用"
return out, nil
}
curPkg = pkg
for _, row := range rows {
if row.PackageCode == pkg.PackageCode {
row.OldPkgName = pkg.PackageName
}
}
}
if targetType == consts.BatchTypeChange {
newPkg, err = p.repo.Tables().LoadPackageByCode(ctx, main.TenantId, options.newPackageCode)
if err != nil || newPkg == nil {
out.rowErrors[0] = rcLocalize("errTargetPkgNotFound")
out.hasFailure = true
out.failureReason = "目标套餐不存在或未启用"
return out, nil
}
if newPkg.Enabled != 1 {
out.rowErrors[0] = rcLocalize("errPkgDisabled")
out.hasFailure = true
out.failureReason = "目标套餐未启用"
return out, nil
}
if curPkg != nil && curPkg.PackageType != newPkg.PackageType {
out.rowErrors[0] = rcLocalize("errPackageTypeMismatch")
out.hasFailure = true
out.failureReason = rcLocalize("errPackageTypeMismatch")
return out, nil
}
if curPkg != nil && curPkg.PackageCode == newPkg.PackageCode {
out.rowErrors[0] = rcLocalize("errNewPackageSameAsOld")
out.hasFailure = true
out.failureReason = rcLocalize("errNewPackageSameAsOld")
return out, nil
}
if curPkg.ChangeablePackageCodes == nil || curPkg.ChangeablePackageCodes.IsNil() {
out.rowErrors[0] = rcLocalize("errNoChangeablePackages")
out.hasFailure = true
out.failureReason = rcLocalize("errNoChangeablePackages")
return out, nil
}
allowed := false
for _, c := range curPkg.ChangeablePackageCodes.Interfaces() {
if strings.TrimSpace(stringify(c)) == newPkg.PackageCode {
allowed = true
break
}
}
if !allowed {
out.rowErrors[0] = rcLocalize("errNewPackageNotInChange")
out.hasFailure = true
out.failureReason = rcLocalize("errNewPackageNotInChange")
return out, nil
}
for _, row := range rows {
row.NewPkgCode = newPkg.PackageCode
}
}
customerName, err := p.repo.Tables().LoadDeptName(ctx, main.TenantId, customerID)
if err != nil || customerName == "" {
out.rowErrors[0] = rcLocalize("errDeptNotFound")
out.hasFailure = true
out.failureReason = rcLocalize("errDeptNotFound")
return out, nil
}
customer := &customerInfo{
AgentID: options.customerAgentID,
Name: customerName,
DeptID: customerID,
TenantID: main.TenantId,
}
out.customer = customer
// 单卡校验
for _, row := range rows {
if row.ICCID == "" {
continue
}
card := cards[row.ICCID]
if card == nil {
continue
}
// 续费时长校验
if targetType == consts.BatchTypeRenew {
if curPkg.RenewableDurationConfig == nil || curPkg.RenewableDurationConfig.IsNil() {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], rcLocalize("errRenewDurationEmpty"))
continue
}
opts, err := renewableYearOptions(curPkg.RenewableDurationConfig)
if err != nil || len(opts) == 0 {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], rcLocalize("errRenewDurationEmpty"))
continue
}
if options.durationSet && !rc.ContainsInt(opts, options.durationValue) {
out.rowErrors[row.RowNo] = appendIfEmpty(out.rowErrors[row.RowNo], rcLocalize("errRenewableDurationMiss"))
continue
}
}
}
// 解析价格
priceSnap, changePrice, err := p.resolvePrice(ctx, main, options, targetType, curPkg, newPkg, customer)
if err != nil {
out.rowErrors[0] = err.Error()
out.hasFailure = true
out.failureReason = err.Error()
return out, nil
}
out.priceSnap = priceSnap
if targetType == consts.BatchTypeRenew {
fillRenewBeginSnapshot(priceSnap, rows, out.rowErrors)
}
if changePrice != nil {
out.changePrice = changePrice
}
if targetType == consts.BatchTypeChange && curPkg != nil {
out.changeSnap = &ChangePrepareSnapshot{
SourceBatchNo: executionSourceBatchNo(main, main.BatchInfo),
CustomerAgentId: customer.DeptID,
CustomerName: customer.Name,
OldPackageCode: curPkg.PackageCode,
OldPackageName: curPkg.PackageName,
OldPackageType: curPkg.PackageType,
OldPackageFlowMb: curPkg.FlowMb,
Quantity: len(rows),
}
}
if curPkg != nil {
out.packageInfo = &packageInfo{
Code: curPkg.PackageCode,
Name: curPkg.PackageName,
Type: curPkg.PackageType,
FlowMb: curPkg.FlowMb,
Duration: options.durationValue,
UnitPrice: priceSnap.RenewUnitPrice,
}
}
// 把行级错误汇总到 hasFailure
for _, msg := range out.rowErrors {
if strings.TrimSpace(msg) != "" {
out.hasFailure = true
break
}
}
if out.hasFailure {
if out.failureReason == "" {
if len(rows) == 1 {
out.failureReason = strings.TrimSpace(out.rowErrors[rows[0].RowNo])
}
if out.failureReason != "" {
return out, nil
}
out.failureReason = "同批存在不满足条件的卡,本次未执行"
}
}
return out, nil
}
func setPrepareFailureFromRows(out *prepareOutcome, rows []*RenewRow) bool {
if out == nil {
return false
}
for _, msg := range out.rowErrors {
if strings.TrimSpace(msg) != "" {
out.hasFailure = true
break
}
}
if !out.hasFailure {
return false
}
if len(rows) == 1 {
out.failureReason = strings.TrimSpace(out.rowErrors[rows[0].RowNo])
}
if out.failureReason == "" {
out.failureReason = "同批存在不满足条件的卡,本次未执行"
}
return true
}
func (p *PrepareExecutor) resolvePrepareRowIdentities(ctx context.Context, tenantID string, rows []*RenewRow, rowErrors map[int]string) error {
keys := collectPrepareIdentityKeys(rows, rowErrors)
cards, err := p.repo.Tables().LoadCardsByIdentityValues(ctx, tenantID, keys)
if err != nil {
return gerror.Wrap(err, "load prepare identity cards failed")
}
resolvePrepareIdentityRowsFromCards(rows, rowErrors, cards)
return nil
}
func collectPrepareIdentityKeys(rows []*RenewRow, rowErrors map[int]string) []string {
keys := make([]string, 0, len(rows)*3)
for _, row := range rows {
if row == nil || row.Reason != "" || strings.TrimSpace(rowErrors[row.RowNo]) != "" {
continue
}
keys = append(keys, row.ICCID, row.MSISDN, row.IMSI)
}
return uniquePrepareStrings(keys)
}
func collectPrepareIccids(rows []*RenewRow, rowErrors map[int]string) []string {
seen := make(map[string]int, len(rows))
iccids := make([]string, 0, len(rows))
for _, row := range rows {
if row == nil || row.Reason != "" || strings.TrimSpace(rowErrors[row.RowNo]) != "" {
continue
}
row.ICCID = strings.TrimSpace(row.ICCID)
if row.ICCID == "" {
continue
}
if firstRowNo, exists := seen[row.ICCID]; exists {
reason := rcLocalize("errDuplicateICCID")
rowErrors[row.RowNo] = appendIfEmpty(rowErrors[row.RowNo], reason)
rowErrors[firstRowNo] = appendIfEmpty(rowErrors[firstRowNo], reason)
continue
}
seen[row.ICCID] = row.RowNo
iccids = append(iccids, row.ICCID)
}
return iccids
}
func resolvePrepareIdentityRowsFromCards(rows []*RenewRow, rowErrors map[int]string, cards []*rc.SimCardInfo) {
resolver := newPrepareIdentityResolver(cards)
for _, row := range rows {
if row == nil || row.Reason != "" || strings.TrimSpace(rowErrors[row.RowNo]) != "" {
continue
}
identities := uniquePrepareStrings([]string{row.ICCID, row.MSISDN, row.IMSI})
if len(identities) == 0 {
rowErrors[row.RowNo] = "ICCID/MSISDN/IMSI不能为空"
continue
}
card, err := resolver.resolve(identities)
if err != nil {
rowErrors[row.RowNo] = err.Error()
continue
}
if card == nil {
continue
}
if !prepareCardIdentityMatchesAll(card, identities) {
rowErrors[row.RowNo] = "ICCID/MSISDN/IMSI与卡信息不一致"
continue
}
row.ICCID = strings.TrimSpace(card.Iccid)
row.MSISDN = strings.TrimSpace(card.Msisdn)
row.IMSI = strings.TrimSpace(card.Imsi)
}
}
type prepareIdentityResolver struct {
byValue map[string]map[int64]*rc.SimCardInfo
}
func newPrepareIdentityResolver(cards []*rc.SimCardInfo) *prepareIdentityResolver {
resolver := &prepareIdentityResolver{byValue: map[string]map[int64]*rc.SimCardInfo{}}
for _, card := range cards {
resolver.addCard(card)
}
return resolver
}
func (r *prepareIdentityResolver) addCard(card *rc.SimCardInfo) {
if r == nil || card == nil || card.Id <= 0 {
return
}
r.addValue(card.Iccid, card)
r.addValue(card.Msisdn, card)
r.addValue(card.Imsi, card)
}
func (r *prepareIdentityResolver) addValue(value string, card *rc.SimCardInfo) {
value = strings.TrimSpace(value)
if r == nil || value == "" || card == nil || card.Id <= 0 {
return
}
if r.byValue == nil {
r.byValue = map[string]map[int64]*rc.SimCardInfo{}
}
matches := r.byValue[value]
if matches == nil {
matches = map[int64]*rc.SimCardInfo{}
r.byValue[value] = matches
}
matches[card.Id] = card
}
func (r *prepareIdentityResolver) resolve(values []string) (*rc.SimCardInfo, error) {
if r == nil || len(values) == 0 {
return nil, nil
}
matches := map[int64]*rc.SimCardInfo{}
for _, value := range uniquePrepareStrings(values) {
for id, card := range r.byValue[strings.TrimSpace(value)] {
matches[id] = card
}
}
switch len(matches) {
case 0:
return nil, nil
case 1:
for _, card := range matches {
return card, nil
}
}
return nil, gerror.New("卡号对应多条卡记录,请输入 ICCID")
}
func prepareCardIdentityMatchesAll(card *rc.SimCardInfo, values []string) bool {
if card == nil {
return false
}
for _, value := range values {
if !prepareCardIdentityMatches(card, value) {
return false
}
}
return true
}
func prepareCardIdentityMatches(card *rc.SimCardInfo, value string) bool {
value = strings.TrimSpace(value)
if value == "" || card == nil {
return false
}
return value == strings.TrimSpace(card.Iccid) ||
value == strings.TrimSpace(card.Msisdn) ||
value == strings.TrimSpace(card.Imsi)
}
func prepareJSONValue(json *gjson.Json, keys ...string) string {
if json == nil {
return ""
}
for _, key := range keys {
if value := strings.TrimSpace(json.Get(key).String()); value != "" {
return value
}
}
return ""
}
func uniquePrepareStrings(list []string) []string {
if len(list) == 0 {
return nil
}
seen := make(map[string]struct{}, len(list))
out := make([]string, 0, len(list))
for _, value := range list {
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
}
func (p *PrepareExecutor) resolvePrice(ctx context.Context, main *rc.BatchMainRow, _ prepareOptions, targetType int, curPkg, newPkg *rc.FlowPackageInfo, customer *customerInfo) (*RenewPriceSnapshot, *ChangePriceSnapshot, error) {
tenantID := main.TenantId
pkg := newPkg
if pkg == nil {
pkg = curPkg
}
if pkg == nil {
return nil, nil, gerror.New("缺失套餐信息")
}
unitPrice := 0.0
priceSource := PriceSourceSystem
var (
agentDefaultID int64
targetID int64
)
priceOwnerDeptID := int64(0)
if targetType == consts.BatchTypeRenew {
if customer == nil || customer.DeptID <= 0 {
return nil, nil, gerror.New("renew customer department is missing")
}
dept, err := p.repo.Tables().LoadDept(ctx, tenantID, customer.DeptID)
if err != nil {
return nil, nil, gerror.Wrap(err, "load renew customer department failed")
}
priceOwnerDeptID = renewPriceOwnerDeptID(dept)
salePrice, err := resolveRenewEffectiveSalePrice(ctx, p.repo.Tables(), tenantID, priceOwnerDeptID, customer.DeptID, pkg)
if err != nil {
return nil, nil, gerror.Wrap(err, "calculate renew effective price failed")
}
unitPrice = salePrice.AgentShipmentPrice
priceSource = salePrice.Rule.PriceSource
agentDefaultID = salePrice.Rule.AgentDefaultPriceId
targetID = salePrice.Rule.TargetPriceId
} else {
priceOwnerDeptID = main.DeptId
if priceOwnerDeptID <= 0 {
priceOwnerDeptID = main.CreatedDept
}
salePrice, err := resolveRenewEffectiveSalePrice(ctx, p.repo.Tables(), tenantID, priceOwnerDeptID, customer.DeptID, pkg)
if err != nil {
return nil, nil, gerror.Wrap(err, "calculate change effective price failed")
}
unitPrice = salePrice.AgentShipmentPrice
priceSource = salePrice.Rule.PriceSource
agentDefaultID = salePrice.Rule.AgentDefaultPriceId
targetID = salePrice.Rule.TargetPriceId
}
unitPrice = rc.RoundAmount(unitPrice)
if targetType == consts.BatchTypeRenew {
opts, _ := renewableYearOptions(curPkg.RenewableDurationConfig)
durationUnit := renewDurationUnitForPackage(curPkg)
renewableCodes, renewablePackages, err := p.renewablePackageOptions(ctx, tenantID, customer.DeptID, priceOwnerDeptID, curPkg)
if err != nil {
return nil, nil, err
}
return &RenewPriceSnapshot{
CustomerAgentId: customer.DeptID,
PriceOwnerDeptId: priceOwnerDeptID,
PackageCode: curPkg.PackageCode,
PackageName: curPkg.PackageName,
BillingMethod: curPkg.BillingMethod,
RenewDurationUnit: durationUnit,
RenewDurationUnitText: consts.RenewDurationUnitText(durationUnit),
PriceSource: priceSource,
AgentDefaultPriceId: agentDefaultID,
TargetPriceId: targetID,
RenewUnitPrice: unitPrice,
RenewableDurationOptions: opts,
RenewablePackageCodes: renewableCodes,
RenewablePackages: renewablePackages,
}, nil, nil
}
oldOpts, _ := renewableYearOptions(curPkg.RenewableDurationConfig)
newOpts, _ := changeDurationOptions(pkg.RenewableDurationConfig)
oldDurationUnit := renewDurationUnitForPackage(curPkg)
renewSnap := &RenewPriceSnapshot{
CustomerAgentId: customer.DeptID,
PackageCode: curPkg.PackageCode,
PackageName: curPkg.PackageName,
BillingMethod: curPkg.BillingMethod,
RenewDurationUnit: oldDurationUnit,
RenewDurationUnitText: consts.RenewDurationUnitText(oldDurationUnit),
PriceSource: priceSource,
AgentDefaultPriceId: agentDefaultID,
TargetPriceId: targetID,
RenewUnitPrice: rc.RoundAmount(curPkg.CostPrice),
RenewableDurationOptions: oldOpts,
}
changeSnap := &ChangePriceSnapshot{
CustomerAgentId: customer.DeptID,
NewPackageCode: newPkg.PackageCode,
NewPackageName: newPkg.PackageName,
PriceSource: priceSource,
AgentDefaultPriceId: agentDefaultID,
TargetPriceId: targetID,
ChangeUnitPrice: unitPrice,
RenewableDurationOptions: newOpts,
}
return renewSnap, changeSnap, nil
}
func fillRenewBeginSnapshot(snapshot *RenewPriceSnapshot, rows []*RenewRow, rowErrors map[int]string) {
if snapshot == nil {
return
}
dates := make([]string, 0, len(rows))
for _, row := range rows {
if row == nil || row.OldExpireTime == nil {
continue
}
if strings.TrimSpace(rowErrors[row.RowNo]) != "" {
continue
}
renewBeginAt := renewBeginDateFromExpire(row.OldExpireTime)
if renewBeginAt == nil {
continue
}
date := strings.TrimSpace(renewBeginAt.Format("Y-m-d"))
if date != "" {
dates = append(dates, date)
}
}
sameDate, minDate, maxDate, same := summarizeRenewBeginDates(dates)
snapshot.RenewBeginDate = sameDate
snapshot.RenewBeginDateMin = minDate
snapshot.RenewBeginDateMax = maxDate
snapshot.SameRenewBeginDate = same
}
func validateRenewSameCurrentPeriod(rows []*RenewRow, rowErrors map[int]string) error {
var beginTime, expireTime string
hasBase := false
for _, row := range rows {
if row == nil {
continue
}
if strings.TrimSpace(rowErrors[row.RowNo]) != "" {
continue
}
rowBegin := formatTime(row.OldEffTime)
rowExpire := formatTime(row.OldExpireTime)
if rowBegin == "" || rowExpire == "" {
return gerror.New(renewSamePeriodFailureReason)
}
if !hasBase {
beginTime = rowBegin
expireTime = rowExpire
hasBase = true
continue
}
if rowBegin != beginTime || rowExpire != expireTime {
return gerror.New(renewSamePeriodFailureReason)
}
}
return nil
}
func summarizeRenewBeginDates(dates []string) (sameDate string, minDate string, maxDate string, same bool) {
if len(dates) == 0 {
return "", "", "", false
}
sort.Strings(dates)
minDate = dates[0]
maxDate = dates[len(dates)-1]
same = minDate == maxDate
if same {
sameDate = minDate
}
return sameDate, minDate, maxDate, same
}
func renewPriceOwnerDeptID(dept *rc.SysDeptRow) int64 {
if dept == nil {
return 0
}
return dept.ParentId
}
func renewExpireAfterAllowedWindow(expireAt *gtime.Time, now time.Time) bool {
if expireAt == nil {
return false
}
return expireAt.Time.After(now.AddDate(0, renewExpireWindowMonths, 0))
}
func priceByPackage(pkg *rc.FlowPackageInfo, override *rc.FlowPackagePrice) float64 {
if pkg == nil {
return 0
}
ratio := pkg.SysDefaultRatio
profit := pkg.SysDefaultProfit
if override != nil {
ratio = override.AgentCostRatio
profit = override.AgentCostProfit
}
return rc.RoundAmount(pkg.CostPrice*normalizePriceRatio(ratio) + profit)
}
func (p *PrepareExecutor) renewablePackageOptions(ctx context.Context, tenantID string, customerDeptID, priceOwnerDeptID int64, currentPkg *rc.FlowPackageInfo) ([]string, []RenewPackageOption, error) {
if currentPkg == nil {
return nil, nil, gerror.New("当前套餐不存在或未启用")
}
codes := renewPackageCodesFromJSON(currentPkg.RenewablePackageCodes)
if len(codes) == 0 {
return nil, nil, gerror.New("当前套餐未配置可续费套餐")
}
options := make([]RenewPackageOption, 0, len(codes))
validCodes := make([]string, 0, len(codes))
for _, code := range codes {
pkg, err := p.repo.Tables().LoadPackageByCode(ctx, tenantID, code)
if err != nil || pkg == nil || pkg.Enabled != 1 {
continue
}
unitPrice, err := p.renewPackageUnitPrice(ctx, tenantID, customerDeptID, priceOwnerDeptID, pkg)
if err != nil {
return nil, nil, err
}
durations, err := renewableYearOptions(pkg.RenewableDurationConfig)
if err != nil {
continue
}
validCodes = append(validCodes, pkg.PackageCode)
options = append(options, RenewPackageOption{
PackageCode: pkg.PackageCode,
PackageName: pkg.PackageName,
RenewUnitPrice: unitPrice,
RenewableDurationOptions: durations,
})
}
if len(options) == 0 {
return nil, nil, gerror.New("当前套餐未配置可用的可续费套餐")
}
return validCodes, options, nil
}
func (p *PrepareExecutor) renewPackageUnitPrice(ctx context.Context, tenantID string, customerDeptID, priceOwnerDeptID int64, pkg *rc.FlowPackageInfo) (float64, error) {
if pkg == nil {
return 0, gerror.New("续费套餐不存在")
}
salePrice, err := resolveRenewEffectiveSalePrice(ctx, p.repo.Tables(), tenantID, priceOwnerDeptID, customerDeptID, pkg)
if err != nil {
return 0, gerror.Wrap(err, "calculate renew package effective price failed")
}
return salePrice.AgentShipmentPrice, nil
}
func renewPackageCodesFromJSON(values *gjson.Json) []string {
if values == nil || values.IsNil() {
return nil
}
seen := make(map[string]struct{})
codes := make([]string, 0, len(values.Interfaces()))
for _, value := range values.Interfaces() {
code := strings.TrimSpace(gconvString(value))
if code == "" {
continue
}
if _, ok := seen[code]; ok {
continue
}
seen[code] = struct{}{}
codes = append(codes, code)
}
return codes
}
func normalizePriceRatio(ratio float64) float64 {
if ratio > 10 {
return ratio / 100
}
return ratio
}
func (p *PrepareExecutor) finalizeBatchFailure(ctx context.Context, main *rc.BatchMainRow, reason string, perRow map[int]string) error {
g.Log().Warningf(ctx, "prepare batch failed, batchNo:%s, reason:%s", main.BatchNo, reason)
now := p.nowFunc()
err := p.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
finalBatchType := main.BatchType
// 明细级失败
if perRow != nil {
for rowNo, msg := range perRow {
if rowNo <= 0 || msg == "" {
continue
}
if err := p.repo.MarkDetailFailed(ctx, tx, main.BatchNo, main.BatchType, rowNo, msg, main.CreatedBy); err != nil {
return err
}
}
}
// 兜底:所有仍是 running 的明细批量置为失败
if err := p.repo.MarkAllDetailsFailed(ctx, tx, main.BatchNo, main.BatchType, reason, main.CreatedBy); err != nil {
return err
}
if targetType := readTargetBatchType(main.BatchInfo); targetType == consts.BatchTypeRenew || targetType == consts.BatchTypeChange {
convertResult, err := p.repo.ConvertPrepareToRenewChange(ctx, tx, main.BatchNo, targetType)
if err != nil {
return err
}
g.Log().Warningf(ctx, "prepare failure converted to target batch, batchNo:%s, targetType:%d, mainAffected:%d, detailAffected:%d, reason:%s",
main.BatchNo, targetType, convertResult.MainAffected, convertResult.DetailAffected, reason)
finalBatchType = targetType
}
return p.repo.FinalizeBatchMain(ctx, tx, main.BatchNo, finalBatchType, &rc.BatchMainRow{
BatchAsyncStatus: failedStatus(),
RunResult: consts.BatchRunResultFailed,
FailReason: reason,
SuccessCount: 0,
FailCount: main.TotalCount,
ValidCount: 0,
TotalCount: main.TotalCount,
EndTime: gtime.NewFromTime(now),
UpdatedBy: main.CreatedBy,
})
})
if err != nil {
return gerror.Wrap(err, "finalize prepare failure failed")
}
return nil
}
func (p *PrepareExecutor) finalizeBatchSuccess(ctx context.Context, main *rc.BatchMainRow, targetType int, out *prepareOutcome, details []*rc.BatchDetailRow) error {
now := p.nowFunc()
info := buildPrepareInfo(main.BatchInfo, targetType, out, len(details))
tenantID := main.TenantId
err := p.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
for _, d := range details {
if d == nil {
continue
}
snap := buildPrepareDetailSnapshot(d, out, targetType)
itemID := ""
if row := out.rows[d.RowNo]; row != nil {
itemID = row.ICCID
}
if err := p.repo.MarkPrepareDetailSuccess(ctx, tx, main.BatchNo, d.RowNo, itemID, snap, main.CreatedBy); err != nil {
return err
}
}
// 批量将所有 running 置为 success(兜底覆盖)
if err := p.repo.MarkAllDetailsSuccess(ctx, tx, main.BatchNo, main.BatchType, main.CreatedBy, nil); err != nil {
return err
}
if err := p.repo.UpdateBatchMainInfo(ctx, tx, main.BatchNo, main.BatchType, &rc.BatchMainRow{
BatchType: main.BatchType,
BatchAsyncStatus: consts.BatchAsyncStatusSuccess,
RunResult: consts.BatchRunResultSuccess,
TotalCount: main.TotalCount,
ValidCount: len(details),
SuccessCount: len(details),
FailCount: 0,
BatchInfo: gjson.New(info),
EndTime: gtime.NewFromTime(now),
UpdatedBy: main.CreatedBy,
}); err != nil {
return err
}
return nil
})
if err != nil {
return gerror.Wrap(err, "finalize prepare success failed")
}
g.Log().Infof(ctx, "prepare batch success, batchNo:%s, targetType:%d, tenant:%s", main.BatchNo, targetType, tenantID)
return nil
}
func buildPrepareInfo(prev *gjson.Json, targetType int, out *prepareOutcome, quantity int) map[string]any {
info := map[string]any{}
if prev != nil {
for k, v := range prev.Map() {
info[k] = v
}
}
info["targetBatchType"] = targetType
info["quantity"] = quantity
if out.customer != nil {
info["customerAgentId"] = out.customer.DeptID
info["customerName"] = out.customer.Name
}
if out.priceSnap != nil {
info["priceSnapshot"] = out.priceSnap
if targetType == consts.BatchTypeRenew {
info["packageCode"] = out.priceSnap.PackageCode
info["packageName"] = out.priceSnap.PackageName
info["billingMethod"] = out.priceSnap.BillingMethod
info["renewDurationUnit"] = out.priceSnap.RenewDurationUnit
info["renewDurationUnitText"] = out.priceSnap.RenewDurationUnitText
info["priceSource"] = out.priceSnap.PriceSource
info["priceOwnerDeptId"] = out.priceSnap.PriceOwnerDeptId
info["agentDefaultPriceId"] = out.priceSnap.AgentDefaultPriceId
info["targetPriceId"] = out.priceSnap.TargetPriceId
info["renewUnitPrice"] = out.priceSnap.RenewUnitPrice
info["renewableDurationOptions"] = out.priceSnap.RenewableDurationOptions
info["renewablePackageCodes"] = out.priceSnap.RenewablePackageCodes
info["renewablePackages"] = out.priceSnap.RenewablePackages
info["renewBeginDate"] = out.priceSnap.RenewBeginDate
info["renewBeginDateMin"] = out.priceSnap.RenewBeginDateMin
info["renewBeginDateMax"] = out.priceSnap.RenewBeginDateMax
info["sameRenewBeginDate"] = out.priceSnap.SameRenewBeginDate
}
}
if out.changePrice != nil {
info["changePriceSnapshot"] = out.changePrice
if targetType == consts.BatchTypeChange {
info["newPackageCode"] = out.changePrice.NewPackageCode
info["newPackageName"] = out.changePrice.NewPackageName
info["priceSource"] = out.changePrice.PriceSource
info["agentDefaultPriceId"] = out.changePrice.AgentDefaultPriceId
info["targetPriceId"] = out.changePrice.TargetPriceId
info["changeUnitPrice"] = out.changePrice.ChangeUnitPrice
info["renewableDurationOptions"] = out.changePrice.RenewableDurationOptions
}
}
if out.changeSnap != nil {
info["changePrepareSnapshot"] = out.changeSnap
if targetType == consts.BatchTypeChange {
info["sourceBatchNo"] = out.changeSnap.SourceBatchNo
info["oldPackageCode"] = out.changeSnap.OldPackageCode
info["oldPackageName"] = out.changeSnap.OldPackageName
info["oldPackageType"] = out.changeSnap.OldPackageType
info["oldPackageFlowMb"] = out.changeSnap.OldPackageFlowMb
if quantity <= 0 {
info["quantity"] = out.changeSnap.Quantity
}
}
}
if out.customer != nil {
info["customer"] = out.customer
}
if out.packageInfo != nil {
info["packageInfo"] = out.packageInfo
}
info["prepareTime"] = timeNow().UTC().Format("2006-01-02 15:04:05")
info["status"] = "success"
return info
}
func buildPrepareDetailSnapshot(d *rc.BatchDetailRow, out *prepareOutcome, targetType int) *gjson.Json {
base := map[string]any{}
if d.Snapshot != nil {
for k, v := range d.Snapshot.Map() {
base[k] = v
}
}
base["rowNo"] = d.RowNo
base["success"] = true
if row := out.rows[d.RowNo]; row != nil {
base["iccid"] = row.ICCID
base["msisdn"] = row.MSISDN
base["imsi"] = row.IMSI
base["oldPackageCode"] = row.OldPkgCode
base["oldPackageName"] = row.OldPkgName
base["oldSubscriptionNo"] = row.OldBindSubNo
base["oldOrderSn"] = row.OldBindOrdNo
base["sourceBatchNo"] = row.SourceBatchNo
if out.changePrice != nil {
base["newPackageCode"] = out.changePrice.NewPackageCode
base["newPackageName"] = out.changePrice.NewPackageName
}
}
if out.priceSnap != nil {
base["priceSnapshot"] = out.priceSnap
}
if out.changePrice != nil {
base["changePriceSnapshot"] = out.changePrice
}
if out.changeSnap != nil {
base["changePrepareSnapshot"] = out.changeSnap
}
if out.customer != nil {
base["customer"] = out.customer
}
if out.packageInfo != nil {
base["packageInfo"] = out.packageInfo
}
base["targetBatchType"] = targetType
return gjson.New(base)
}
func loadPrepareOptions(info *gjson.Json) prepareOptions {
opts := prepareOptions{priceScope: PriceScopeAgentDefault}
if info == nil {
return opts
}
opts.customerAgentID = info.Get("customerAgentId").Int64()
opts.targetAgentID = info.Get("targetAgentId").Int64()
opts.priceScope = int(info.Get("priceScope").Int64())
if opts.priceScope == 0 {
opts.priceScope = PriceScopeAgentDefault
}
opts.newPackageCode = strings.TrimSpace(info.Get("newPackageCode").String())
if v := firstPrepareOptionValue(info, "durationValue", "durationYears"); v != nil {
opts.durationValue = int(v.Int64())
opts.durationSet = opts.durationValue > 0
} else if v := firstPrepareOptionValue(info, "durationMonths"); v != nil {
opts.durationValue = normalizeLegacyRenewDurationValue(int(v.Int64()))
opts.durationSet = opts.durationValue > 0
}
return opts
}
func firstPrepareOptionValue(info *gjson.Json, paths ...string) *gvar.Var {
if info == nil {
return nil
}
for _, path := range paths {
value := info.Get(path)
if value != nil && !value.IsNil() {
return value
}
}
return nil
}
func mergeRowErrors(perRow, rowErrors map[int]string) map[int]string {
if len(perRow) == 0 && len(rowErrors) == 0 {
return nil
}
out := make(map[int]string, len(perRow)+len(rowErrors))
for rowNo, msg := range perRow {
if strings.TrimSpace(msg) != "" {
out[rowNo] = msg
}
}
for rowNo, msg := range rowErrors {
if strings.TrimSpace(msg) == "" {
continue
}
if _, ok := out[rowNo]; !ok {
out[rowNo] = msg
}
}
return out
}
func readTargetBatchType(info *gjson.Json) int {
if info == nil {
return 0
}
if v := info.Get("targetBatchType"); v != nil {
return int(v.Int64())
}
if v := info.Get("target_batch_type"); v != nil {
return int(v.Int64())
}
return 0
}
func failedStatus() int { return consts.BatchAsyncStatusFailed }
func rcLocalize(key string) string {
switch key {
case "errCardNotFound":
return "ICCID不存在正式卡记录"
case "errMsisdnMismatch":
return "MSISDN与SIM卡记录不一致"
case "errImsiMismatch":
return "IMSI与SIM卡记录不一致"
case "errCardNoCurrentPackage":
return "ICCID没有当前套餐"
case "errMultiplePackages":
return "当前批次下存在多个套餐,不能统一处理"
case "errMultipleCustomers":
return "当前批次下存在多个客户,不能统一处理"
case "errPkgNotFound":
return "套餐不存在"
case "errPkgDisabled":
return "套餐未启用或不存在"
case "errDeptNotFound":
return "客户不存在"
case "errNoChangeablePackages":
return "当前套餐未配置可变更套餐范围"
case "errNewPackageNotInChange":
return "新套餐不在当前套餐可变更范围内"
case "errNewPackageSameAsOld":
return "新套餐不能与当前套餐相同"
case "errPackageTypeMismatch":
return "目标套餐类型与当前套餐类型不一致"
case "errTargetPkgNotFound":
return "目标套餐不存在"
case "errRenewDurationEmpty":
return "续费/变更套餐未配置可续费时长"
case "errRenewableDurationMiss":
return "续费时长不在当前套餐可续费范围内"
}
return key
}
func appendIfEmpty(existing, addition string) string {
addition = strings.TrimSpace(addition)
if addition == "" {
return existing
}
if existing == "" {
return addition
}
return existing + ";" + addition
}
func stringify(value any) string {
switch v := value.(type) {
case string:
return v
case int:
return strconv.Itoa(v)
case int32:
return strconv.FormatInt(int64(v), 10)
case int64:
return strconv.FormatInt(v, 10)
case float32:
return strconv.FormatFloat(float64(v), 'f', -1, 32)
case float64:
return strconv.FormatFloat(v, 'f', -1, 64)
}
return ""
}