package upstreamimpl import ( "context" "encoding/json" "strings" "time" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/consts" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/util/gconv" ) // 移动 OneLink「通信功能开停」接口(2026 接口改版后,替代原 ec/operate/sim-apn-function)。 // // 7.2.1 单卡(CMIOT_API25M19) GET /v5/ec/manage/sim-communication-service // functionType=01(APN上网功能),businessType:0=开 1=停,APN 名称字段为 apn // 7.2.2 批量(CMIOT_API23M09) GET /v5/ec/operate/sim-communication-function/batch // serviceType=11(数据通信服务),operType:0=暂停(停) 1=恢复(开),APN 名称字段为 apnName // // 分流策略(与产品约定): // // 卡数 <10 逐张走 7.2.1 // 10~100 一次走 7.2.2 // >100 按 100 拆块,不足 10 的尾块退回 7.2.1 // // 批量接口 7.2.2 只认 msisdn,缺号码的卡降级走 7.2.1(用 iccid)。 // 两个接口同卡同类型均有限流(3 分钟一次),且 7.2.2 批量中任一张被限流则整批失败(status 13009)。 const ( ydCommunicationServicePath = "ec/manage/sim-communication-service" ydCommunicationFunctionBatchPath = "ec/operate/sim-communication-function/batch" ydCommunicationBatchResultPath = "ec/query/sim-batch-result" // 5.1.2 CMIOT_API23M10 批量办理结果查询 ydCommunicationFunctionTypeApn = "01" // 7.2.1 functionType:APN 上网功能 ydCommunicationServiceTypeData = "11" // 7.2.2 serviceType:数据通信服务 ydCommunicationBatchMin = 10 // 7.2.2 每批最少卡数 ydCommunicationBatchMax = 100 // 7.2.2 每批最多卡数 ) // 批量办理任务状态(5.1.2 sim-batch-result 返回的 jobStatus)。 const ( ydJobStatusDone = "2" // 处理完成(全部成功) ydJobStatusDonePartial = "3" // 处理完成(含失败记录) ydJobStatusFailed = "4" // 处理失败 ) // ydBatchResultQueryFailedMessage 是批量异步任务结果查询失败(轮询超时 / 查询请求失败)时, // 逐卡写入执行日志明细的失败原因:该子批次下每张卡都记这一条,代表"结果未确认"而非"办理失败"。 const ydBatchResultQueryFailedMessage = "查询上游通信结果失败,未取得最终办理结果,请稍后核对" // 批量结果轮询参数;声明为变量以便单测缩短周期。 // // 两个值均维持原值,原因(实测:上游批量任务约 60s 出终态): // - 间隔只决定终态被捕获的延迟(0~interval)。整批 2 万张 = 200 子批 / 并发 5 = 40 波, // 总等待由"每波等上游 job 完成"主导,缩间隔对总时长影响可忽略,却成倍增加查询接口压力。 // - 上限若下调到 60s 会踩在上游耗时边界上,超时即把整个子批 100 张记为失败,而失败结果 // 不回写本地 data_service_status,会造成平台状态与卡真实状态不一致。 // // 真正压缩总等待应把"提交"与"结果查询"分离(全部子批先拿 jobId,再统一并发轮询), // 使总等待从 40 × 上限 降到 ≈ 1 × 上限,而不是靠缩短这里的超时。 var ( ydBatchResultPollInterval = 5 * time.Second ydBatchResultPollTimeout = 180 * time.Second ) // ydApnRequester 抽出移动通道发起请求所需的最小能力,供真实通道(sSimYD)与 mock 通道(sSimYDMock) // 共用同一套开停逻辑,避免两份实现走偏。 type ydApnRequester interface { getUrl(ctx context.Context, serviceUrl string) string getTransId(ctx context.Context) string requestYd(ctx context.Context, operation, url string, params g.Map) (*ydHTTPResponse, error) } // changeSimApnFunctionYD 平台「开通数据(dataStart)/停用数据(dataStop)/达量断网(dataCut)」在移动侧的统一入口。 func changeSimApnFunctionYD(ctx context.Context, c ydApnRequester, input *model.SimCardOperateChangeApnInput) (model.SimCardOperateChangeApnOutput, error) { items := make(model.SimCardOperateChangeApnOutput, 0, len(input.SimKeys)) // 缺 msisdn 的卡不能进批量接口,先分流出来单独走单卡。 withMsisdn := make([]*model.SimKey, 0, len(input.SimKeys)) singleKeys := make([]*model.SimKey, 0) for _, simKey := range input.SimKeys { if simKey == nil { continue } if strings.TrimSpace(simKey.MSISDN) == "" { singleKeys = append(singleKeys, simKey) continue } withMsisdn = append(withMsisdn, simKey) } for start := 0; start < len(withMsisdn); start += ydCommunicationBatchMax { end := start + ydCommunicationBatchMax if end > len(withMsisdn) { end = len(withMsisdn) } chunk := withMsisdn[start:end] if len(chunk) < ydCommunicationBatchMin { singleKeys = append(singleKeys, chunk...) continue } items = append(items, changeApnBatchYD(ctx, c, chunk, input.ReqData)...) } items = append(items, changeApnSingleYD(ctx, c, singleKeys, input.ReqData)...) return items, nil } // changeApnSingleYD 逐张调用 7.2.1 单卡接口办理。 func changeApnSingleYD(ctx context.Context, c ydApnRequester, simKeys []*model.SimKey, req model.SimCardOperateChangeApnInputItem) model.SimCardOperateChangeApnOutput { items := make(model.SimCardOperateChangeApnOutput, 0, len(simKeys)) if len(simKeys) == 0 { return items } businessType, ok := convertYdCommunicationBusinessType(ctx, req.OperType) if !ok { for _, simKey := range simKeys { items = append(items, ydApnErrorItem(simKey, "unsupported operType, expected 0(open) or 1(stop)")) } return items } url := c.getUrl(ctx, ydCommunicationServicePath) // {"status":"0","message":"正确","result":[{"iccid":"898602B2221430000005","orderNum":"xxxxxxx"}]} type ydSingleResp struct { Status string `json:"status"` Message string `json:"message"` Result []struct { OrderNum string `json:"orderNum"` } `json:"result"` } for _, simKey := range simKeys { params := g.Map{ "transid": c.getTransId(ctx), "functionType": ydCommunicationFunctionTypeApn, "businessType": businessType, "apn": req.ApnName, } // 卡号三选一,必须有且只有一项:优先 iccid,其次 msisdn,最后 imsi。 switch { case strings.TrimSpace(simKey.ICCID) != "": params["iccid"] = strings.TrimSpace(simKey.ICCID) case strings.TrimSpace(simKey.MSISDN) != "": params["msisdn"] = strings.TrimSpace(simKey.MSISDN) case strings.TrimSpace(simKey.IMSI) != "": params["imsi"] = strings.TrimSpace(simKey.IMSI) default: g.Log().Warningf(ctx, "changeApnSingleYD missing iccid/msisdn/imsi, simKey %+v", simKey) items = append(items, ydApnErrorItem(simKey, "missing iccid/msisdn/imsi")) continue } resp, err := c.requestYd(ctx, "ChangeSimApnFunction", url, params) if err != nil { g.Log().Errorf(ctx, "changeApnSingleYD request error %+v, url %s, params %+v", err, url, params) items = append(items, ydApnErrorItem(simKey, err.Error())) continue } body := resp.ReadAllString() var ydResp ydSingleResp if err = json.Unmarshal([]byte(body), &ydResp); err != nil { g.Log().Errorf(ctx, "changeApnSingleYD unmarshal error %+v, url %s, params %+v, body %s", err, url, params, body) items = append(items, ydApnErrorItem(simKey, err.Error())) continue } if ydResp.Status != "0" { g.Log().Warningf(ctx, "changeApnSingleYD upstream error %s, url %s, params %+v, body %s", ydResp.Message, url, params, body) items = append(items, ydApnResultItem(simKey, gconv.Int(ydResp.Status), ydResp.Message, "")) continue } orderNum := "" if len(ydResp.Result) > 0 { orderNum = ydResp.Result[0].OrderNum } items = append(items, ydApnResultItem(simKey, 0, ydResp.Message, orderNum)) } return items } // changeApnBatchYD 用一批(10~100 张)调用 7.2.2 批量接口一次办理。 // 该接口异步返回 jobId,status=0 即视为本批全部受理成功,逐卡回填同一个 jobId。 func changeApnBatchYD(ctx context.Context, c ydApnRequester, simKeys []*model.SimKey, req model.SimCardOperateChangeApnInputItem) model.SimCardOperateChangeApnOutput { items := make(model.SimCardOperateChangeApnOutput, 0, len(simKeys)) operType, ok := convertYdCommunicationOperType(ctx, req.OperType) if !ok { for _, simKey := range simKeys { items = append(items, ydApnErrorItem(simKey, "unsupported operType, expected 0(open) or 1(stop)")) } return items } msisdns := make([]string, 0, len(simKeys)) for _, simKey := range simKeys { msisdns = append(msisdns, strings.TrimSpace(simKey.MSISDN)) } url := c.getUrl(ctx, ydCommunicationFunctionBatchPath) params := g.Map{ "transid": c.getTransId(ctx), "msisdns": strings.Join(msisdns, "_"), "serviceType": ydCommunicationServiceTypeData, "operType": operType, "apnName": req.ApnName, } // {"status":"0","message":"正确","result":[{"jobId":"xxxxxxx"}]} type ydBatchResp struct { Status string `json:"status"` Message string `json:"message"` Result []struct { JobId string `json:"jobId"` } `json:"result"` } resp, err := c.requestYd(ctx, "ChangeSimApnFunction", url, params) if err != nil { g.Log().Errorf(ctx, "changeApnBatchYD request error %+v, url %s, params %+v", err, url, params) for _, simKey := range simKeys { items = append(items, ydApnErrorItem(simKey, err.Error())) } return items } body := resp.ReadAllString() var ydResp ydBatchResp if err = json.Unmarshal([]byte(body), &ydResp); err != nil { g.Log().Errorf(ctx, "changeApnBatchYD unmarshal error %+v, url %s, params %+v, body %s", err, url, params, body) for _, simKey := range simKeys { items = append(items, ydApnErrorItem(simKey, err.Error())) } return items } if ydResp.Status != "0" { g.Log().Warningf(ctx, "changeApnBatchYD upstream error %s, url %s, params %+v, body %s", ydResp.Message, url, params, body) for _, simKey := range simKeys { items = append(items, ydApnResultItem(simKey, gconv.Int(ydResp.Status), ydResp.Message, "")) } return items } jobId := "" if len(ydResp.Result) > 0 { jobId = ydResp.Result[0].JobId } if jobId == "" { g.Log().Warningf(ctx, "changeApnBatchYD accepted without jobId, url %s, params %+v, body %s", url, params, body) for _, simKey := range simKeys { items = append(items, ydApnResultItem(simKey, consts.ApiStatusError, "上游受理成功但未返回 jobId", "")) } return items } // 7.2.2 是异步接口:受理(status=0)不等于办理成功。必须轮询结果接口拿到每张卡的最终结果, // 否则执行日志会把"部分失败"的批次整批误记为成功。 result := queryBatchResultYD(ctx, c, jobId) switch { case result.finished && result.failed: for _, simKey := range simKeys { items = append(items, ydApnResultItem(simKey, consts.ApiStatusError, "上游批量任务处理失败", jobId)) } case result.finished: for _, simKey := range simKeys { card, exist := result.cards[strings.TrimSpace(simKey.MSISDN)] if !exist { items = append(items, ydApnResultItem(simKey, consts.ApiStatusError, "上游批量结果未返回该卡", jobId)) continue } items = append(items, ydApnResultItem(simKey, card.code, card.message, jobId)) } default: // 轮询超时或查询失败:该子批下每张卡都按"查询上游通信结果失败"落库,避免误报成功。 // 该文案会逐卡写进执行日志明细;日后若增加"重新查询上游结果"的补偿能力,从这一层补做即可。 g.Log().Warningf(ctx, "changeApnBatchYD result poll not finished, jobId %s, params %+v", jobId, params) for _, simKey := range simKeys { items = append(items, ydApnResultItem(simKey, consts.ApiStatusError, ydBatchResultQueryFailedMessage, jobId)) } } return items } // ydBatchCardResult 批量任务中单张卡的最终办理结果。 type ydBatchCardResult struct { code int message string } // ydBatchQueryResult 批量办理结果查询的结论。 type ydBatchQueryResult struct { finished bool // 是否已取得终态(jobStatus 2/3/4) failed bool // jobStatus=4,整批处理失败 cards map[string]ydBatchCardResult // resultId(msisdn)-> 单卡结果 } // ydBatchResultItem 5.1.2 sim-batch-result 的逐卡结果项。 type ydBatchResultItem struct { Status string `json:"status"` Message string `json:"message"` ResultType string `json:"resultType"` ResultId string `json:"resultId"` } // ydBatchResultResp 5.1.2 sim-batch-result 的响应体。 type ydBatchResultResp struct { Status string `json:"status"` Message string `json:"message"` Result []struct { JobStatus string `json:"jobStatus"` ResultList []ydBatchResultItem `json:"resultList"` } `json:"result"` } // queryBatchResultYD 轮询 7.2.2 批量任务结果,直到取得终态或超时。 // 超时 / 请求失败均返回 finished=false,由调用方按"未确认"处理。 func queryBatchResultYD(ctx context.Context, c ydApnRequester, jobId string) ydBatchQueryResult { url := c.getUrl(ctx, ydCommunicationBatchResultPath) deadline := time.Now().Add(ydBatchResultPollTimeout) for { params := g.Map{ "transid": c.getTransId(ctx), "jobId": jobId, } resp, err := c.requestYd(ctx, "QuerySimBatchResult", url, params) if err != nil { g.Log().Warningf(ctx, "queryBatchResultYD request error %+v, jobId %s", err, jobId) } else { body := resp.ReadAllString() var parsed ydBatchResultResp if jsonErr := json.Unmarshal([]byte(body), &parsed); jsonErr != nil { g.Log().Warningf(ctx, "queryBatchResultYD unmarshal error %+v, jobId %s, body %s", jsonErr, jobId, body) } else if parsed.Status != "0" { g.Log().Warningf(ctx, "queryBatchResultYD upstream error %s, jobId %s, body %s", parsed.Message, jobId, body) } else if len(parsed.Result) > 0 { switch parsed.Result[0].JobStatus { case ydJobStatusDone, ydJobStatusDonePartial: return ydBatchQueryResult{finished: true, cards: toYdBatchCardMap(parsed.Result[0].ResultList)} case ydJobStatusFailed: return ydBatchQueryResult{finished: true, failed: true} } } } if time.Now().After(deadline) { return ydBatchQueryResult{} } select { case <-ctx.Done(): return ydBatchQueryResult{} case <-time.After(ydBatchResultPollInterval): } } } // toYdBatchCardMap 把逐卡结果按 resultId(msisdn)建索引。 func toYdBatchCardMap(list []ydBatchResultItem) map[string]ydBatchCardResult { cards := make(map[string]ydBatchCardResult, len(list)) for _, item := range list { id := strings.TrimSpace(item.ResultId) if id == "" { continue } cards[id] = ydBatchCardResult{code: gconv.Int(item.Status), message: item.Message} } return cards } func ydApnErrorItem(simKey *model.SimKey, message string) *model.SimBaseRes[model.SimCardOperateChangeApnItem] { return &model.SimBaseRes[model.SimCardOperateChangeApnItem]{ SimKey: simKey, Code: consts.ApiStatusError, Message: message, } } func ydApnResultItem(simKey *model.SimKey, code int, message, orderNum string) *model.SimBaseRes[model.SimCardOperateChangeApnItem] { return &model.SimBaseRes[model.SimCardOperateChangeApnItem]{ SimKey: simKey, Code: code, Message: message, ResData: model.SimCardOperateChangeApnItem{OrderNum: orderNum}, } }