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

fix: file export to reset data

Browse Source
This commit is contained in:
simonzhangsz
2025-04-10 10:37:40 +08:00
parent 6329d638a4
commit a1eaaaebd7
1 changed files with 69 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
src/modules/crontask/processor/exportUEData/exportUEData.go
View File

@@ -10,11 +10,11 @@ import (
// "path/filepath"
"time"
ueService "be.ems/features/ue/service"
"be.ems/lib/dborm"
"be.ems/lib/log"
"be.ems/src/framework/cron"
ueService "be.ems/features/ue/service"
neService "be.ems/src/modules/network_data/service"
neService "be.ems/src/modules/network_data/service"
)
var NewProcessor = &BarProcessor{
@@ -31,20 +31,20 @@ type BarProcessor struct {
}
type BarParams struct {
TableName string `json:"tableName"`
Columns string `json:"columns"` // exported column name of time string
Extras string `json:"extras"` // extras condition for where
TableName string `json:"tableName"`
Columns string `json:"columns"` // exported column name of time string
Extras string `json:"extras"` // extras condition for where
ServiceName string `json:"serviceName"` // data service name
OrderBy string `json:"orderBy"`
OrderType string `json:"orderType"` // order type: asc/desc
FileType string `json:"fileType"` // file type: txt/csv
FilePath string `json:"filePath"` // file path
OrderBy string `json:"orderBy"`
OrderType string `json:"orderType"` // order type: asc/desc
FileType string `json:"fileType"` // file type: txt/csv
FilePath string `json:"filePath"` // file path
}
const (
FILE_TYPE_TXT = "txt"
FILE_TYPE_CSV = "csv"
NE_TYPE_UDM = "UDM"
NE_TYPE_UDM = "UDM"
)
func (s *BarProcessor) Execute(data any) (any, error) {
@@ -75,7 +75,7 @@ func (s *BarProcessor) Execute(data any) (any, error) {
// 返回结果，用于记录执行结果
return map[string]any{
"msg": "sucess",
"msg": "sucess",
"results": results,
}, nil
}
@@ -97,25 +97,30 @@ func (s *BarProcessor) exportUEData(param BarParams) (map[string]any, error) {
Find(&neIDs).Error
if err != nil {
log.Error("failed to get ne_ids:", err)
return nil, err
return nil, err
}
var fps []string
var affectedArr []int64
for _, neID := range neIDs {
log.Trace("ne_id:", neID)
// 1. 加载最新数据, 如果数据服务存在，则重新加载数据
dataService, err := GetService(param.ServiceName)
if err != nil {
log.Warn("failed to get data service:", err)
} else if dataService != nil {
// 重新加载数据
data := dataService.ResetData(neID)
if data == 0 {
log.Error("failed to load data:", err)
return nil, err
}
log.Trace("load data:", data)
result := ResetDataWithResult(param.ServiceName, neID)
log.Trace("reset data count=", result.Count)
// 后续可以等待异步操作返回结果
if err := <-result.Err; err != nil {
log.Error("failed to reset data: ", err)
}
// dataService, err := GetService(param.ServiceName)
// if err != nil {
// log.Warn("failed to get data service:", err)
// } else if dataService != nil {
// // 重新加载数据
// data := dataService.ResetData(neID)
// if data == 0 {
// log.Error("failed to load data:", err)
// return nil, err
// }
// log.Trace("load data:", data)
// }
// 2. 构造查询语句
var query string
@@ -138,11 +143,11 @@ func (s *BarProcessor) exportUEData(param BarParams) (map[string]any, error) {
// 3. 构造文件路径
var filePath string
if param.FileType == FILE_TYPE_TXT {
filePath = fmt.Sprintf("%s/%s_export_%s_%s.txt",
filePath = fmt.Sprintf("%s/%s_export_%s_%s.txt",
param.FilePath, param.TableName, time.Now().Format("20060102150405"), neID)
} else {
// 默认导出 csv 文件
filePath = fmt.Sprintf("%s/%s_export_%s_%s.csv",
filePath = fmt.Sprintf("%s/%s_export_%s_%s.csv",
param.FilePath, param.TableName, time.Now().Format("20060102150405"), neID)
}
@@ -159,12 +164,12 @@ func (s *BarProcessor) exportUEData(param BarParams) (map[string]any, error) {
// 5. 返回结果
result := map[string]any{
"msg": "sucess",
"table": param.TableName,
"ne_id": neIDs,
"table": param.TableName,
"ne_id": neIDs,
"affected": affectedArr,
}
return result, nil
}
}
func (s *BarProcessor) exportData(query, filePath string, fileType string) (int64, error) {
rows, err := dborm.XCoreDB().Query(query)
@@ -226,24 +231,51 @@ func (s *BarProcessor) exportData(query, filePath string, fileType string) (int6
return affected, nil
}
type ResetDataResult struct {
Count int64 // 同步返回的数据长度
Err <-chan error // 异步返回 ClearAndInsert 的执行结果
}
// ResetDataWithResult 重置鉴权用户数据，清空数据库重新同步Redis数据
// 立即返回数据量，同时通过 channel 返回 ClearAndInsert 的执行结果
func ResetDataWithResult(serivceName, neID string) ResetDataResult {
dataService, err := GetService(serivceName)
if err != nil {
log.Warn("failed to get data service:", err)
}
authArr := dataService.dataByRedis("*", neID)
resultCh := make(chan error, 1)
go func() {
resultCh <- dataService.ClearAndInsert(neID, authArr)
}()
return ResetDataResult{
Count: int64(len(authArr)),
Err: resultCh,
}
}
// ResettableService 接口定义
type ResettableService interface {
ResetData(neID string) int64
dataByRedis(key, neID string) []map[string]string // 从Redis获取数据
ClearAndInsert(neID string, authArr []map[string]string) error // 清空数据库并插入数据
//ResetData(neID string) int64
}
// 服务注册表
var serviceRegistry = make(map[string]ResettableService)
func RegisterService(name string, service ResettableService) {
serviceRegistry[name] = service
serviceRegistry[name] = service
}
// 获取服务
func GetService(name string) (ResettableService, error) {
service, exists := serviceRegistry[name]
if !exists {
return nil, fmt.Errorf("service %s not found", name)
}
return service, nil
service, exists := serviceRegistry[name]
if !exists {
return nil, fmt.Errorf("service %s not found", name)
}
return service, nil
}
// 初始化注册表
@@ -253,5 +285,5 @@ func init() {
RegisterService("UDMVoIPAuth", ueService.NewVoIPAuthService)
RegisterService("UDMIMSUser", ueService.NewIMSUserService)
// 这里注册更多服务
// 这里注册更多服务
}