OMC/be.ems
1
0
Fork 0
Code Issues Releases 3 Activity

perf: 重构接收CDR功能

Browse Source
This commit is contained in:
TsMask
2025-07-15 14:51:16 +08:00
parent d7adfa79ba
commit adb0d5f4e2
22 changed files with 1059 additions and 1563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
src/modules/network_data/repository/cdr_event.go Normal file
View File

@@ -0,0 +1,167 @@
package repository
import (
"fmt"
"strings"
"time"
"be.ems/src/framework/database/db"
"be.ems/src/framework/logger"
"be.ems/src/modules/network_data/model"
)
// 实例化数据层 CDREvent 结构体
var NewCDREvent = &CDREvent{}
// CDREvent CDR会话事件 数据层处理
type CDREvent struct{}
// SelectByPage 分页查询集合
func (r CDREvent) SelectByPage(neType string, query map[string]string) ([]model.CDREvent, int64) {
// 表名
tableName := fmt.Sprintf("cdr_event_%s", strings.ToLower(neType))
tx := db.DB("").Table(tableName).Model(&model.CDREvent{})
// 查询条件拼接
if v, ok := query["rmUID"]; ok && v != "" {
tx = tx.Where("rm_uid = ?", v)
}
if v, ok := query["beginTime"]; ok && v != "" {
if len(v) == 10 {
v = v + "000"
}
tx = tx.Where("timestamp >= ?", v)
}
if v, ok := query["endTime"]; ok && v != "" {
if len(v) == 10 {
v = v + "000"
}
tx = tx.Where("timestamp <= ?", v)
}
// 各个网元特殊查询条件
switch neType {
case "SMSC":
if v, ok := query["callerParty"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.callerParty') = ?", v)
}
if v, ok := query["calledParty"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.calledParty') = ?", v)
}
if v, ok := query["recordType"]; ok && v != "" {
recordTypes := strings.Split(v, ",")
var querytrArr []string
for _, recordType := range recordTypes {
querytrArr = append(querytrArr, fmt.Sprintf("JSON_EXTRACT(cdr_json, '$.recordType') = '%s'", recordType))
}
tx = tx.Where(fmt.Sprintf("( %s )", strings.Join(querytrArr, " OR ")))
}
case "SMF":
if v, ok := query["recordType"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.recordType') = ?", v)
}
if v, ok := query["subscriberID"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.subscriberIdentifier.subscriptionIDData') = ?", v)
}
if v, ok := query["dnn"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.pDUSessionChargingInformation.dNNID') = ?", v)
}
case "SGWC":
if v, ok := query["imsi"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.servedIMSI') = ?", v)
}
if v, ok := query["msisdn"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.servedMSISDN') = ?", v)
}
case "IMS":
if v, ok := query["callerParty"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.callerParty') = ?", v)
}
if v, ok := query["calledParty"]; ok && v != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.calledParty') = ?", v)
}
}
var total int64 = 0
rows := []model.CDREvent{}
// 查询数量 长度为0直接返回
if err := tx.Count(&total).Error; err != nil || total <= 0 {
return rows, total
}
// 分页
pageNum, pageSize := db.PageNumSize(query["pageNum"], query["pageSize"])
tx = tx.Offset(int(pageNum * pageSize)).Limit(int(pageSize))
// 排序
if v, ok := query["sortField"]; ok && v != "" {
sortSql := v
if o, ok := query["sortOrder"]; ok && o != "" {
if o == "desc" {
sortSql += " desc "
} else {
sortSql += " asc "
}
}
tx = tx.Order(sortSql)
}
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query err => %v", err)
}
return rows, total
}
// SelectByIds 通过ID查询
func (r CDREvent) SelectByIds(neType string, ids []int64) []model.CDREvent {
rows := []model.CDREvent{}
if len(ids) <= 0 {
return rows
}
// 表名
tableName := fmt.Sprintf("cdr_event_%s", strings.ToLower(neType))
tx := db.DB("").Table(tableName).Model(&model.CDREvent{})
// 构建查询条件
tx = tx.Where("id in ?", ids)
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows
}
return rows
}
// DeleteByIds 批量删除信息
func (r CDREvent) DeleteByIds(neType string, ids []int64) int64 {
if len(ids) <= 0 {
return 0
}
// 表名
tableName := fmt.Sprintf("cdr_event_%s", strings.ToLower(neType))
tx := db.DB("").Table(tableName).Where("id in ?", ids)
if err := tx.Delete(&model.CDREvent{}).Error; err != nil {
logger.Errorf("delete err => %v", err.Error())
return 0
}
return tx.RowsAffected
}
// Insert 新增信息 返回新增数据ID
func (r CDREvent) Insert(param model.CDREvent) int64 {
if param.NeType == "" {
return 0
}
if param.CreatedAt == 0 {
param.CreatedAt = time.Now().UnixMilli()
}
// 表名
tableName := fmt.Sprintf("cdr_event_%s", strings.ToLower(param.NeType))
// 执行插入
if err := db.DB("").Table(tableName).Create(&param).Error; err != nil {
logger.Errorf("insert err => %v", err.Error())
return 0
}
return param.ID
}

107
src/modules/network_data/repository/cdr_event_ims.go
View File

@@ -1,107 +0,0 @@
package repository
import (
"fmt"
"strings"
"be.ems/src/framework/database/db"
"be.ems/src/framework/logger"
"be.ems/src/modules/network_data/model"
)
// 实例化数据层 CDREventIMS 结构体
var NewCDREventIMS = &CDREventIMS{}
// CDREventIMS CDR会话事件 数据层处理
type CDREventIMS struct{}
// SelectByPage 分页查询集合
func (r CDREventIMS) SelectByPage(query model.CDREventIMSQuery) ([]model.CDREventIMS, int64) {
tx := db.DB("").Model(&model.CDREventIMS{})
// 查询条件拼接
if query.NeType != "" {
tx = tx.Where("ne_type = ?", query.NeType)
}
if query.RmUID != "" {
tx = tx.Where("rm_uid = ?", query.RmUID)
}
if query.BeginTime != 0 {
tx = tx.Where("created_at >= ?", query.BeginTime)
}
if query.EndTime != 0 {
tx = tx.Where("created_at <= ?", query.EndTime)
}
if query.CallerParty != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.callerParty') = ?", query.CallerParty)
}
if query.CalledParty != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.calledParty') = ?", query.CalledParty)
}
if query.RecordType != "" {
recordTypes := strings.Split(query.RecordType, ",")
var querytrArr []string
for _, recordType := range recordTypes {
querytrArr = append(querytrArr, fmt.Sprintf("JSON_EXTRACT(cdr_json, '$.recordType') = '%s'", recordType))
}
tx = tx.Where(fmt.Sprintf("( %s )", strings.Join(querytrArr, " OR ")))
}
// 查询结果
var total int64 = 0
rows := []model.CDREventIMS{}
// 查询数量为0直接返回
if err := tx.Count(&total).Error; err != nil || total <= 0 {
return rows, total
}
// 排序
if query.SortField != "" {
sortField := query.SortField
if query.SortOrder == "desc" {
sortField = sortField + " desc"
}
tx = tx.Order(sortField)
}
// 查询数据分页
pageNum, pageSize := db.PageNumSize(query.PageNum, query.PageSize)
tx = tx.Limit(pageSize).Offset(pageSize * pageNum)
err := tx.Find(&rows).Error
if err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows, total
}
return rows, total
}
// SelectByIds 通过ID查询
func (r *CDREventIMS) SelectByIds(ids []int64) []model.CDREventIMS {
rows := []model.CDREventIMS{}
if len(ids) <= 0 {
return rows
}
tx := db.DB("").Model(&model.CDREventIMS{})
// 构建查询条件
tx = tx.Where("id in ?", ids)
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows
}
return rows
}
// DeleteByIds 批量删除信息
func (r *CDREventIMS) DeleteByIds(ids []int64) int64 {
if len(ids) <= 0 {
return 0
}
tx := db.DB("").Where("id in ?", ids)
if err := tx.Delete(&model.CDREventIMS{}).Error; err != nil {
logger.Errorf("delete err => %v", err.Error())
return 0
}
return tx.RowsAffected
}

95
src/modules/network_data/repository/cdr_event_sgwc.go
View File

@@ -1,95 +0,0 @@
package repository
import (
"be.ems/src/framework/database/db"
"be.ems/src/framework/logger"
"be.ems/src/modules/network_data/model"
)
// 实例化数据层 CDREventSGWC 结构体
var NewCDREventSGWC = &CDREventSGWC{}
// CDREventSGWC CDR会话事件 数据层处理
type CDREventSGWC struct{}
// SelectByPage 分页查询集合
func (r CDREventSGWC) SelectByPage(query model.CDREventSGWCQuery) ([]model.CDREventSGWC, int64) {
tx := db.DB("").Model(&model.CDREventSGWC{})
// 查询条件拼接
if query.NeType != "" {
tx = tx.Where("ne_type = ?", query.NeType)
}
if query.RmUID != "" {
tx = tx.Where("rm_uid = ?", query.RmUID)
}
if query.BeginTime != 0 {
tx = tx.Where("timestamp >= ?", query.BeginTime)
}
if query.EndTime != 0 {
tx = tx.Where("timestamp <= ?", query.EndTime)
}
if query.IMSI != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.servedIMSI') = ?", query.IMSI)
}
if query.MSISDN != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.servedMSISDN') = ?", query.MSISDN)
}
// 查询结果
var total int64 = 0
rows := []model.CDREventSGWC{}
// 查询数量为0直接返回
if err := tx.Count(&total).Error; err != nil || total <= 0 {
return rows, total
}
// 排序
if query.SortField != "" {
sortField := query.SortField
if query.SortOrder == "desc" {
sortField = sortField + " desc"
}
tx = tx.Order(sortField)
}
// 查询数据分页
pageNum, pageSize := db.PageNumSize(query.PageNum, query.PageSize)
tx = tx.Limit(pageSize).Offset(pageSize * pageNum)
err := tx.Find(&rows).Error
if err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows, total
}
return rows, total
}
// SelectByIds 通过ID查询
func (r *CDREventSGWC) SelectByIds(ids []int64) []model.CDREventSGWC {
rows := []model.CDREventSGWC{}
if len(ids) <= 0 {
return rows
}
tx := db.DB("").Model(&model.CDREventSGWC{})
// 构建查询条件
tx = tx.Where("id in ?", ids)
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows
}
return rows
}
// DeleteByIds 批量删除信息
func (r *CDREventSGWC) DeleteByIds(ids []int64) int64 {
if len(ids) <= 0 {
return 0
}
tx := db.DB("").Where("id in ?", ids)
if err := tx.Delete(&model.CDREventSGWC{}).Error; err != nil {
logger.Errorf("delete err => %v", err.Error())
return 0
}
return tx.RowsAffected
}

98
src/modules/network_data/repository/cdr_event_smf.go
View File

@@ -1,98 +0,0 @@
package repository
import (
"be.ems/src/framework/database/db"
"be.ems/src/framework/logger"
"be.ems/src/modules/network_data/model"
)
// 实例化数据层 CDREventSMF 结构体
var NewCDREventSMF = &CDREventSMF{}
// CDREventSMF CDR会话事件 数据层处理
type CDREventSMF struct{}
// SelectByPage 分页查询集合
func (r CDREventSMF) SelectByPage(query model.CDREventSMFQuery) ([]model.CDREventSMF, int64) {
tx := db.DB("").Model(&model.CDREventSMF{})
// 查询条件拼接
if query.NeType != "" {
tx = tx.Where("ne_type = ?", query.NeType)
}
if query.RmUID != "" {
tx = tx.Where("rm_uid = ?", query.RmUID)
}
if query.BeginTime != 0 {
tx = tx.Where("timestamp >= ?", query.BeginTime)
}
if query.EndTime != 0 {
tx = tx.Where("timestamp <= ?", query.EndTime)
}
if query.RecordType != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.recordType') = ?", query.RecordType)
}
if query.SubscriberID != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.subscriberIdentifier.subscriptionIDData') = ?", query.SubscriberID)
}
if query.DNN != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.pDUSessionChargingInformation.dNNID') = ?", query.DNN)
}
// 查询结果
var total int64 = 0
rows := []model.CDREventSMF{}
// 查询数量为0直接返回
if err := tx.Count(&total).Error; err != nil || total <= 0 {
return rows, total
}
// 排序
if query.SortField != "" {
sortField := query.SortField
if query.SortOrder == "desc" {
sortField = sortField + " desc"
}
tx = tx.Order(sortField)
}
// 查询数据分页
pageNum, pageSize := db.PageNumSize(query.PageNum, query.PageSize)
tx = tx.Limit(pageSize).Offset(pageSize * pageNum)
err := tx.Find(&rows).Error
if err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows, total
}
return rows, total
}
// SelectByIds 通过ID查询
func (r *CDREventSMF) SelectByIds(ids []int64) []model.CDREventSMF {
rows := []model.CDREventSMF{}
if len(ids) <= 0 {
return rows
}
tx := db.DB("").Model(&model.CDREventSMF{})
// 构建查询条件
tx = tx.Where("id in ?", ids)
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows
}
return rows
}
// DeleteByIds 批量删除信息
func (r *CDREventSMF) DeleteByIds(ids []int64) int64 {
if len(ids) <= 0 {
return 0
}
tx := db.DB("").Where("id in ?", ids)
if err := tx.Delete(&model.CDREventSMF{}).Error; err != nil {
logger.Errorf("delete err => %v", err.Error())
return 0
}
return tx.RowsAffected
}

107
src/modules/network_data/repository/cdr_event_smsc.go
View File

@@ -1,107 +0,0 @@
package repository
import (
"fmt"
"strings"
"be.ems/src/framework/database/db"
"be.ems/src/framework/logger"
"be.ems/src/modules/network_data/model"
)
// 实例化数据层 CDREventSMSC 结构体
var NewCDREventSMSC = &CDREventSMSC{}
// CDREventSMSC CDR会话事件 数据层处理
type CDREventSMSC struct{}
// SelectByPage 分页查询集合
func (r CDREventSMSC) SelectByPage(query model.CDREventSMSCQuery) ([]model.CDREventSMSC, int64) {
tx := db.DB("").Model(&model.CDREventSMSC{})
// 查询条件拼接
if query.NeType != "" {
tx = tx.Where("ne_type = ?", query.NeType)
}
if query.RmUID != "" {
tx = tx.Where("rm_uid = ?", query.RmUID)
}
if query.BeginTime != 0 {
tx = tx.Where("timestamp >= ?", query.BeginTime)
}
if query.EndTime != 0 {
tx = tx.Where("timestamp <= ?", query.EndTime)
}
if query.CallerParty != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.callerParty') = ?", query.CallerParty)
}
if query.CalledParty != "" {
tx = tx.Where("JSON_EXTRACT(cdr_json, '$.calledParty') = ?", query.CalledParty)
}
if query.RecordType != "" {
recordTypes := strings.Split(query.RecordType, ",")
var querytrArr []string
for _, recordType := range recordTypes {
querytrArr = append(querytrArr, fmt.Sprintf("JSON_EXTRACT(cdr_json, '$.recordType') = '%s'", recordType))
}
tx = tx.Where(fmt.Sprintf("( %s )", strings.Join(querytrArr, " OR ")))
}
// 查询结果
var total int64 = 0
rows := []model.CDREventSMSC{}
// 查询数量为0直接返回
if err := tx.Count(&total).Error; err != nil || total <= 0 {
return rows, total
}
// 排序
if query.SortField != "" {
sortField := query.SortField
if query.SortOrder == "desc" {
sortField = sortField + " desc"
}
tx = tx.Order(sortField)
}
// 查询数据分页
pageNum, pageSize := db.PageNumSize(query.PageNum, query.PageSize)
tx = tx.Limit(pageSize).Offset(pageSize * pageNum)
err := tx.Find(&rows).Error
if err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows, total
}
return rows, total
}
// SelectByIds 通过ID查询
func (r *CDREventSMSC) SelectByIds(ids []int64) []model.CDREventSMSC {
rows := []model.CDREventSMSC{}
if len(ids) <= 0 {
return rows
}
tx := db.DB("").Model(&model.CDREventSMSC{})
// 构建查询条件
tx = tx.Where("id in ?", ids)
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows
}
return rows
}
// DeleteByIds 批量删除信息
func (r *CDREventSMSC) DeleteByIds(ids []int64) int64 {
if len(ids) <= 0 {
return 0
}
tx := db.DB("").Where("id in ?", ids)
if err := tx.Delete(&model.CDREventSMSC{}).Error; err != nil {
logger.Errorf("delete err => %v", err.Error())
return 0
}
return tx.RowsAffected
}