增加异常情况邮件提醒，增加WebSocket服务
This commit is contained in:
parent
34ab8962e0
commit
2fb1d934db
537
main.go
537
main.go
@ -47,7 +47,7 @@ func init() {
|
||||
Password: redisPassword,
|
||||
DB: redisDB,
|
||||
})
|
||||
//go startWebSocket() // 启动 WebSocket 服务，使用协程方式运行
|
||||
go startWebSocket() // 启动 WebSocket 服务，使用协程方式运行
|
||||
applogger.Info(fmt.Sprintf("程序启动，加载%s环境，尝试执行...", env))
|
||||
go downloadDecompression() // 启动立即执行一次数据下载、处理
|
||||
}
|
||||
@ -150,10 +150,13 @@ func downloadDecompression() {
|
||||
if exists, _ := redisClient.Exists("iniDataStatus").Result(); exists == 1 {
|
||||
fmt.Print("上一批次执行中，跳过本次\n")
|
||||
} else {
|
||||
subject := "丝芙兰短信处理程序异常"
|
||||
// 写入执行中标记
|
||||
err := redisClient.Set("iniDataStatus", 1, 0).Err()
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("写入任务执行中标记失败：%v", err))
|
||||
body := fmt.Sprintf("写入任务执行中标记失败：%v", err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
// Connect to SFTP server
|
||||
sshConfig := &ssh.ClientConfig{
|
||||
@ -165,22 +168,28 @@ func downloadDecompression() {
|
||||
}
|
||||
sshClient, err := ssh.Dial("tcp", sftpAddress, sshConfig)
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("写入任务执行中标记失败：%v", err))
|
||||
body := fmt.Sprintf("sshClient连接失败：%v", err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
sftpClient, err := sftp.NewClient(sshClient)
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("写入任务执行中标记失败：%v", err))
|
||||
body := fmt.Sprintf("sftp连接失败：%v", err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
defer sftpClient.Close()
|
||||
files, err := sftpClient.ReadDir(sftpDir)
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("sftp目录不存在: 错误信息：%v", err))
|
||||
body := fmt.Sprintf("sftp目录%s不存在, 错误信息：%v", sftpDir, err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
it := 1
|
||||
fmt.Printf("共%d个文件\n", len(files))
|
||||
|
||||
sort.Sort(FileSorter(files))
|
||||
|
||||
processingStatus := -1
|
||||
for _, file := range files {
|
||||
fmt.Printf("第%d个文件处理中\n", it)
|
||||
it++
|
||||
@ -190,41 +199,45 @@ func downloadDecompression() {
|
||||
fmt.Println("跳过已处理过的文件：" + file.Name())
|
||||
continue
|
||||
}
|
||||
|
||||
if filepath.Ext(file.Name()) == ".zip" {
|
||||
//fmt.Println("下载开始（数据包）：" + file.Name())
|
||||
// Download file
|
||||
srcFile, err := sftpClient.Open(path.Join(sftpDir, file.Name()))
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("打开sftp文件失败，文件名:%s，错误信息： %v", file.Name(), err))
|
||||
body := fmt.Sprintf("打开sftp文件失败，文件名:%s，错误信息： %v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
defer srcFile.Close()
|
||||
dstFile, err := os.Create(path.Join(executableDir, zipPath, file.Name()))
|
||||
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("创建本地文件失败，文件名: %s，错误信息：%v", file.Name(), err))
|
||||
body := fmt.Sprintf("创建本地文件失败，文件名: %s，错误信息：%v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
defer dstFile.Close()
|
||||
if _, err := io.Copy(dstFile, srcFile); err != nil {
|
||||
applogger.Error(fmt.Sprintf("下载文件失败，文件名: %s，错误信息：%v", file.Name(), err))
|
||||
body := fmt.Sprintf("下载文件失败，文件名: %s，错误信息：%v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
//fmt.Printf("%s（数据包）下载完成\n", file.Name())
|
||||
applogger.Info(fmt.Sprintf("%s（数据包）下载完成", file.Name()))
|
||||
// Unzip file
|
||||
zipReader, err := zip.OpenReader(path.Join(executableDir, zipPath, file.Name()))
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("解压文件失败: 文件名：%s，错误信息：%v", file.Name(), err))
|
||||
body := fmt.Sprintf("解压文件失败: 文件名：%s，错误信息：%v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
defer zipReader.Close()
|
||||
//fmt.Println("压缩报文件数量：", len(zipReader.File))
|
||||
for _, zipFile := range zipReader.File {
|
||||
zipFileReader, err := zipFile.Open()
|
||||
if strings.Contains(zipFile.Name, "__MACOSX/._") {
|
||||
//fmt.Print("系统文件.DS_Store,跳过处理", zipFile.Name)
|
||||
continue
|
||||
} else if filepath.Ext(zipFile.Name) != ".txt" {
|
||||
applogger.Error(fmt.Sprintf("文件类型不正确,跳过处理: %v", zipFile.Name))
|
||||
@ -239,60 +252,71 @@ func downloadDecompression() {
|
||||
// Create the file
|
||||
unzipFile, err := os.Create(path.Join(executableDir, txtPath, zipFile.Name))
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("创建压缩后的文件失败，文件名:%s，错误信息： %v", zipFile.Name, err))
|
||||
body := fmt.Sprintf("创建压缩后的文件失败，文件名:%s，错误信息： %v", zipFile.Name, err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
defer unzipFile.Close()
|
||||
// Write the unzip data to the file
|
||||
_, err = io.Copy(unzipFile, zipFileReader)
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("文件解压失败，文件名：%s，错误信息： %v", zipFileReader, err))
|
||||
body := fmt.Sprintf("文件解压失败，文件名：%s，错误信息： %v", zipFileReader, err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
applogger.Info(fmt.Sprintf("%s（数据包）解压完成", zipFile.Name))
|
||||
//fmt.Printf("%s（数据包）解压完成\n", zipFile.Name)
|
||||
batchDataInsert(zipFile.Name)
|
||||
|
||||
processingStatus = batchDataInsert(zipFile.Name)
|
||||
}
|
||||
} else if filepath.Ext(file.Name()) == ".txt" {
|
||||
srcFile, err := sftpClient.Open(path.Join(sftpDir, file.Name()))
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("打开sftp文件失败，文件名:%s，错误信息： %v", file.Name(), err))
|
||||
body := fmt.Sprintf("打开sftp文件失败，文件名:%s，错误信息： %v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
defer srcFile.Close()
|
||||
dstFile, err := os.Create(path.Join(executableDir, txtPath, file.Name()))
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("创建本地文件失败，文件名: %s，错误信息：%v", file.Name(), err))
|
||||
body := fmt.Sprintf("创建本地文件失败，文件名: %s，错误信息：%v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
defer dstFile.Close()
|
||||
if _, err := io.Copy(dstFile, srcFile); err != nil {
|
||||
applogger.Error(fmt.Sprintf("下载文件失败，文件名: %s，错误信息：%v", file.Name(), err))
|
||||
body := fmt.Sprintf("下载文件失败，文件名: %s，错误信息：%v", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
continue
|
||||
}
|
||||
//fmt.Printf("%s（批次文件）下载完成 \n", file.Name())
|
||||
applogger.Info(fmt.Sprintf("%s（批次文件）下载完成", file.Name()))
|
||||
batchInsert(file.Name())
|
||||
processingStatus = batchInsert(file.Name())
|
||||
}
|
||||
err = redisClient.Set(fileKey, 1, 0).Err() //写入下载、解压、入库完成标记
|
||||
if err != nil {
|
||||
applogger.Info(fmt.Sprintf("写入文件处理完成标记失败文件名：%s，错误信息：%v\n", file.Name(), err))
|
||||
if processingStatus == 0 {
|
||||
err = redisClient.Set(fileKey, 1, 0).Err() //写入文件处理完成
|
||||
if err != nil {
|
||||
body := fmt.Sprintf("写入文件处理完成标记失败文件名：%s，错误信息：%v\n", file.Name(), err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
redisClient.Del("iniDataStatus") //删除任务执行中标记
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// 批次入库
|
||||
func batchInsert(fileName string) {
|
||||
func batchInsert(fileName string) int {
|
||||
//fmt.Print("批次处理开始")
|
||||
start := time.Now()
|
||||
db, _ := connectToDB()
|
||||
file, err := os.Open(path.Join(executableDir, txtPath, fileName))
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("文件打开失败文件名：%s，错误信息%v", fileName, err))
|
||||
return -1
|
||||
} else {
|
||||
defer file.Close()
|
||||
reader := csv.NewReader(bufio.NewReader(file))
|
||||
@ -301,6 +325,7 @@ func batchInsert(fileName string) {
|
||||
for {
|
||||
record, err := reader.Read()
|
||||
if err != nil {
|
||||
return -1
|
||||
break
|
||||
}
|
||||
|
||||
@ -327,7 +352,7 @@ func batchInsert(fileName string) {
|
||||
sid, err := CreateBatch(batchParams)
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
return
|
||||
return -1
|
||||
}
|
||||
fmt.Println(sid)
|
||||
|
||||
@ -346,11 +371,14 @@ func batchInsert(fileName string) {
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
subject := "丝芙兰批次文件处理完成"
|
||||
body := "批次文件：" + fileName + ";\n批次数：" + strconv.Itoa(batchRows) + ";\n处理完成，请前往管理平台查看处理。"
|
||||
body := fmt.Sprintf("批次数：%d;\n批次文件：%s;\n处理完成，请前往管理平台查看处理。", batchRows, fileName)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
|
||||
applogger.Info(fmt.Sprintf("%s（批次文件）入库完成", fileName))
|
||||
applogger.Info(fmt.Sprintf("%s（批次文件）执行时间%s 插入批次次数：%d", fileName, elapsed, batchRows))
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
@ -397,41 +425,38 @@ func smsApi(method string, sendSMSDataJson string) (int, error) {
|
||||
return -1, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// 解析响应数据
|
||||
var retobj map[string]interface{}
|
||||
err = json.NewDecoder(resp.Body).Decode(&retobj)
|
||||
if err != nil {
|
||||
return -1, err
|
||||
}
|
||||
fmt.Print(retobj)
|
||||
|
||||
jsonStr, err := json.Marshal(retobj)
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
return -1, err
|
||||
}
|
||||
fmt.Printf("API 返回：%s\n", string(jsonStr))
|
||||
|
||||
code := int(retobj["code"].(float64))
|
||||
|
||||
//fmt.Print("code：", code)
|
||||
|
||||
if code == 0 {
|
||||
//sid := int(retobj["sid"].(float64))
|
||||
fmt.Printf("提交批次成功code：%d\n", code)
|
||||
return code, nil
|
||||
} else {
|
||||
return -1, fmt.Errorf("create batch failed, error code: %d", code)
|
||||
applogger.Error(string(jsonStr))
|
||||
return -1, fmt.Errorf(string(jsonStr))
|
||||
}
|
||||
}
|
||||
|
||||
func batchDataInsert(fileName string) {
|
||||
func batchDataInsert(fileName string) int {
|
||||
start := time.Now()
|
||||
// Open file
|
||||
fileKey := fmt.Sprintf("downloaded:%s", strings.Replace(fileName, filepath.Ext(fileName), ".zip", 1))
|
||||
file, err := os.Open(path.Join(executableDir, txtPath, fileName))
|
||||
if err != nil {
|
||||
redisClient.Del(fileKey) //删除文件处理完成的标志位
|
||||
applogger.Error(fmt.Sprintf("文件打开失败,文件名：%s，错误信息%v", fileName, err))
|
||||
return -1
|
||||
} else {
|
||||
defer file.Close()
|
||||
db, _ := connectToDB()
|
||||
@ -461,249 +486,258 @@ func batchDataInsert(fileName string) {
|
||||
}()
|
||||
}
|
||||
|
||||
// Create and initialize hashset
|
||||
hs := make(map[string]bool)
|
||||
// Prepare batch data
|
||||
dataBatch := make([]BatcheData, 0, batchSize)
|
||||
dataBatchDuplicate := make([]BatchDataDuplicateLog, 0, batchSize)
|
||||
// Parse file line by line and insert data in batches
|
||||
scanner := bufio.NewScanner(file)
|
||||
scanner.Split(bufio.ScanLines)
|
||||
scanner.Scan() // skip first line
|
||||
bi := 0
|
||||
duplicateCount := make(map[string]int)
|
||||
insertsCount := make(map[string]int)
|
||||
//sendSMSData := make(map[string][]Sms)
|
||||
sendMobiles := make(map[string]map[string]interface{})
|
||||
var count int
|
||||
ccids := make(map[string]bool)
|
||||
hs := make(map[string]bool) //排重
|
||||
dataBatch := make([]BatcheData, 0, batchSize) //短信数据
|
||||
dataBatchDuplicate := make([]BatchDataDuplicateLog, 0, batchSize) //重复短信数据
|
||||
bi := 0 //重复总数
|
||||
count := 0 // 总数
|
||||
//数据文件中批次
|
||||
batches := []Batch{} //查询批次
|
||||
|
||||
// 通过下划线将文件名拆分为多个字段
|
||||
fields := strings.Split(fileName, "_")
|
||||
|
||||
// 最后一个字段是日期和时间，我们只需要日期部分
|
||||
datetime := fields[len(fields)-1]
|
||||
fileNameDate := datetime[:8]
|
||||
var batches []Batch
|
||||
// 模糊查询文件名包含“20230103”字符串的批次记录
|
||||
db.Table("batches b1").
|
||||
Select("b1.*").
|
||||
Where("b1.data_file_name LIKE ?", "%"+fileNameDate+"%").
|
||||
Joins("INNER JOIN (SELECT communication_channel_id, MAX(created_at) AS created_at FROM batches GROUP BY communication_channel_id) b2 ON b1.communication_channel_id = b2.communication_channel_id AND b1.created_at = b2.created_at").
|
||||
Order("b1.created_at desc").
|
||||
Find(&batches)
|
||||
|
||||
// 模糊查询包含“20230103”字符串的记录
|
||||
db.Where("data_file_name LIKE ?", "%"+fileNameDate+"%").Find(&batches)
|
||||
batchCount := len(batches)
|
||||
sendMobiles := make(map[string]SmsData, batchCount) //短信数据，用于提交发送平台
|
||||
duplicateCount := make(map[string]int, batchCount) //按批次重复数
|
||||
insertsCount := make(map[string]int, batchCount) //按批次插入数
|
||||
ccids := make(map[string]bool, batchCount)
|
||||
|
||||
// 定义一个名为result的map类型的变量，并以CommunicationChannelID作为键
|
||||
result := make(map[string][]Batch)
|
||||
for _, batch := range batches {
|
||||
cckdKey := strconv.FormatUint(uint64(batch.CommunicationChannelID), 10)
|
||||
result[cckdKey] = append(result[cckdKey], batch)
|
||||
print(batch.CommunicationChannelID, "\n")
|
||||
}
|
||||
|
||||
//ci := 1
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
row, err := csv.NewReader(strings.NewReader(line)).Read()
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("文件按行读取失败,文件名：%s，错误信息%v", fileName, err))
|
||||
continue
|
||||
}
|
||||
reservedFields := map[string]string{ //合并个性化字段
|
||||
"ReservedField1": row[5],
|
||||
"ReservedField2": row[6],
|
||||
"ReservedField3": row[7],
|
||||
"ReservedField4": row[8],
|
||||
"ReservedField5": row[9],
|
||||
"DataFileName": fileName,
|
||||
}
|
||||
reservedFieldsJson, err := json.Marshal(reservedFields) //个性化字段转json
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("个性化字段合并失败,文件名：%s，错误信息%v", fileName, err))
|
||||
continue
|
||||
}
|
||||
if _, ok := duplicateCount[row[2]]; !ok {
|
||||
duplicateCount[row[2]] = 0
|
||||
if batchCount > 0 { //如果查询到数据需要的批次信息，开始处理数据
|
||||
// 定义一个名为result的map类型的变量，并以CommunicationChannelID作为键
|
||||
result := make(map[string][]Batch)
|
||||
for _, batch := range batches {
|
||||
cckdKey := strconv.FormatUint(uint64(batch.CommunicationChannelID), 10)
|
||||
result[cckdKey] = append(result[cckdKey], batch)
|
||||
print(batch.CommunicationChannelID, "\n")
|
||||
}
|
||||
|
||||
// Check if record exists in hashset
|
||||
key := fmt.Sprintf("%s-%s-%s", row[2], row[3], row[5])
|
||||
if _, exists := hs[key]; exists { //如果批次数据重复
|
||||
bi++
|
||||
// Increment duplicate count
|
||||
duplicateCount[row[2]]++
|
||||
dataBatchDuplicate = append(dataBatchDuplicate, BatchDataDuplicateLog{
|
||||
scanner := bufio.NewScanner(file)
|
||||
scanner.Split(bufio.ScanLines)
|
||||
scanner.Scan() // skip first line
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
row, err := csv.NewReader(strings.NewReader(line)).Read()
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("文件按行读取失败,文件名：%s，错误信息%v", fileName, err))
|
||||
continue
|
||||
}
|
||||
reservedFields := map[string]string{ //合并个性化字段
|
||||
"ReservedField1": row[5],
|
||||
"ReservedField2": row[6],
|
||||
"ReservedField3": row[7],
|
||||
"ReservedField4": row[8],
|
||||
"ReservedField5": row[9],
|
||||
"DataFileName": fileName,
|
||||
}
|
||||
reservedFieldsJson, err := json.Marshal(reservedFields) //个性化字段转json
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("个性化字段合并失败,文件名：%s，错误信息%v", fileName, err))
|
||||
continue
|
||||
}
|
||||
if _, ok := duplicateCount[row[2]]; !ok {
|
||||
duplicateCount[row[2]] = 0
|
||||
}
|
||||
|
||||
// Check if record exists in hashset
|
||||
key := fmt.Sprintf("%s-%s-%s", row[2], row[3], row[5])
|
||||
if _, exists := hs[key]; exists { //如果批次数据重复
|
||||
bi++
|
||||
// Increment duplicate count
|
||||
duplicateCount[row[2]]++
|
||||
dataBatchDuplicate = append(dataBatchDuplicate, BatchDataDuplicateLog{
|
||||
CommunicationChannelID: row[2],
|
||||
Mobile: row[3],
|
||||
FullName: row[4],
|
||||
ReservedField: string(reservedFieldsJson),
|
||||
})
|
||||
|
||||
if len(dataBatchDuplicate) >= batchSize {
|
||||
err = db.CreateInBatches(dataBatchDuplicate, insertSize).Error
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("插入重复数据失败,文件名：%s，错误信息%v", fileName, err))
|
||||
} else {
|
||||
dataBatchDuplicate = make([]BatchDataDuplicateLog, 0, batchSize)
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
// Add record to hashset
|
||||
hs[key] = true
|
||||
ccids[row[2]] = true
|
||||
dataBatch = append(dataBatch, BatcheData{
|
||||
CommunicationChannelID: row[2],
|
||||
Mobile: row[3],
|
||||
FullName: row[4],
|
||||
ReservedField: string(reservedFieldsJson),
|
||||
})
|
||||
|
||||
if len(dataBatchDuplicate) >= batchSize {
|
||||
err = db.CreateInBatches(dataBatchDuplicate, insertSize).Error
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("插入重复数据失败,文件名：%s，错误信息%v", fileName, err))
|
||||
} else {
|
||||
dataBatchDuplicate = make([]BatchDataDuplicateLog, 0, batchSize)
|
||||
}
|
||||
|
||||
if len(dataBatch) >= batchSize {
|
||||
dataBatchChan <- dataBatch
|
||||
dataBatch = make([]BatcheData, 0, batchSize)
|
||||
}
|
||||
continue
|
||||
}
|
||||
// Add record to hashset
|
||||
hs[key] = true
|
||||
//tccid, _ := strconv.ParseUint(row[2], 10, 32)
|
||||
ccids[row[2]] = true
|
||||
dataBatch = append(dataBatch, BatcheData{
|
||||
CommunicationChannelID: row[2],
|
||||
Mobile: row[3],
|
||||
FullName: row[4],
|
||||
ReservedField: string(reservedFieldsJson),
|
||||
})
|
||||
|
||||
if len(dataBatch) >= batchSize {
|
||||
if batches, ok := result[row[2]]; ok && len(batches) > 0 && batches[0].Content != "" {
|
||||
content := batches[0].Content
|
||||
Sid := batches[0].Sid
|
||||
pattern := regexp.MustCompile(`\{.*\}`)
|
||||
matched := pattern.MatchString(content)
|
||||
if matched { //个性化短信
|
||||
smsData, ok := sendMobiles[row[2]]
|
||||
if !ok {
|
||||
smsData = SmsData{
|
||||
Sid: Sid,
|
||||
IsPersonalizedMsg: true,
|
||||
SmsList: make([]SmsList, 0),
|
||||
}
|
||||
}
|
||||
placeholderMap := map[string]string{
|
||||
"{RESERVED_FIELD_1}": row[5],
|
||||
"{RESERVED_FIELD_2}": row[6],
|
||||
"{RESERVED_FIELD_3}": row[7],
|
||||
"{RESERVED_FIELD_4}": row[8],
|
||||
"{RESERVED_FIELD_5}": row[9],
|
||||
}
|
||||
for k, v := range placeholderMap {
|
||||
content = strings.ReplaceAll(content, k, v)
|
||||
}
|
||||
smsData.SmsList = append(smsData.SmsList, SmsList{M: row[3], C: content, F: 8})
|
||||
if len(smsData.SmsList) >= batchSize/2 {
|
||||
sd := Sms{Sid: Sid, Data: smsData.SmsList, Token: token}
|
||||
sendSMSDataJson, _ := json.Marshal(sd)
|
||||
//
|
||||
resp, err := smsApi("appendbatchdata", string(sendSMSDataJson))
|
||||
if resp != 0 {
|
||||
fmt.Printf("smsApi returned error: %d\n", err)
|
||||
//发送提醒邮件
|
||||
subject := "丝芙兰数据包数据提交异常"
|
||||
body := fmt.Sprintf("数据包：%s;\nCommunicationChannelID：%s;\n错误信息：%s\n 请前往管理平台查看处理。", fileName, row[2], err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
smsData.SmsList = []SmsList{} // reset mobiles slice
|
||||
}
|
||||
sendMobiles[row[2]] = smsData
|
||||
} else {
|
||||
// 处理非个性化短信
|
||||
smsData, ok := sendMobiles[row[2]]
|
||||
if !ok {
|
||||
smsData = SmsData{
|
||||
Sid: Sid,
|
||||
IsPersonalizedMsg: false,
|
||||
Content: content,
|
||||
Mobiles: make([]string, 0),
|
||||
}
|
||||
}
|
||||
smsData.Mobiles = append(smsData.Mobiles, row[3])
|
||||
if len(smsData.Mobiles) >= batchSize {
|
||||
mobiles := smsData.Mobiles
|
||||
mobileStr := strings.Join(mobiles, ",")
|
||||
var sl []SmsList
|
||||
sl = append(sl, SmsList{M: mobileStr, C: content, F: 8})
|
||||
sd := Sms{Sid: Sid, Data: sl, Token: token}
|
||||
sendSMSDataJson, _ := json.Marshal(sd)
|
||||
resp, err := smsApi("appendbatchdata", string(sendSMSDataJson))
|
||||
if resp != 0 {
|
||||
fmt.Printf("smsApi returned error: %d\n", err)
|
||||
//发送提醒邮件
|
||||
subject := "丝芙兰数据包数据提交异常"
|
||||
body := fmt.Sprintf("数据包：%s;\nCommunicationChannelID：%s;\n错误信息：%s\n 请前往管理平台查看处理。", fileName, row[2], err)
|
||||
applogger.Error(body)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
}
|
||||
smsData.Mobiles = []string{} // reset mobiles slice
|
||||
}
|
||||
sendMobiles[row[2]] = smsData
|
||||
}
|
||||
}
|
||||
if _, ok := insertsCount[row[2]]; !ok {
|
||||
insertsCount[row[2]] = 0
|
||||
}
|
||||
insertsCount[row[2]]++
|
||||
count++
|
||||
}
|
||||
|
||||
if len(dataBatch) > 0 {
|
||||
dataBatchChan <- dataBatch
|
||||
dataBatch = make([]BatcheData, 0, batchSize)
|
||||
//fmt.Print("提交通次数" + strconv.Itoa(ci))
|
||||
//ci++
|
||||
}
|
||||
close(dataBatchChan)
|
||||
|
||||
if batches, ok := result[row[2]]; ok && len(batches) > 0 && batches[0].Content != "" {
|
||||
content := batches[0].Content
|
||||
Sid := batches[0].Sid
|
||||
pattern := regexp.MustCompile(`\{.*\}`)
|
||||
|
||||
matched := pattern.MatchString(content)
|
||||
if matched { //个性化短信
|
||||
if _, ok := sendMobiles[row[2]]; !ok {
|
||||
sendMobiles[row[2]] = make(map[string]interface{})
|
||||
sendMobiles[row[2]]["sid"] = Sid
|
||||
sendMobiles[row[2]]["isPersonalizedMsg"] = true
|
||||
sendMobiles[row[2]]["mobiles"] = make([]SmsList, 0)
|
||||
}
|
||||
placeholderMap := map[string]string{
|
||||
"{RESERVED_FIELD_1}": row[5],
|
||||
"{RESERVED_FIELD_2}": row[6],
|
||||
"{RESERVED_FIELD_3}": row[7],
|
||||
"{RESERVED_FIELD_4}": row[8],
|
||||
"{RESERVED_FIELD_5}": row[9],
|
||||
}
|
||||
for k, v := range placeholderMap {
|
||||
content = strings.ReplaceAll(content, k, v)
|
||||
}
|
||||
|
||||
sendMobiles[row[2]]["mobiles"] = append(sendMobiles[row[2]]["mobiles"].([]SmsList), SmsList{M: row[3], C: content, F: 8})
|
||||
|
||||
if len(sendMobiles[row[2]]["mobiles"].([]SmsList)) >= batchSize/2 {
|
||||
sd := Sms{Sid: Sid, Data: sendMobiles[row[2]]["mobiles"].([]SmsList), Token: token}
|
||||
for ccid := range ccids {
|
||||
smsData, ok := sendMobiles[ccid]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if smsData.IsPersonalizedMsg { //个性化
|
||||
smsList := smsData.SmsList
|
||||
if len(smsList) > 0 {
|
||||
sd := Sms{Sid: smsData.Sid, Data: smsList, Token: token}
|
||||
sendSMSDataJson, _ := json.Marshal(sd)
|
||||
smsApi("appendbatchdata", string(sendSMSDataJson))
|
||||
//fmt.Print(string(sendSMSDataJson))
|
||||
sendMobiles[row[2]]["mobiles"] = []SmsList{} // reset mobiles slice
|
||||
smsData.SmsList = []SmsList{} // reset mobiles slice
|
||||
}
|
||||
|
||||
} else {
|
||||
// 处理非个性化短信
|
||||
if _, ok := sendMobiles[row[2]]; !ok {
|
||||
sendMobiles[row[2]] = make(map[string]interface{})
|
||||
sendMobiles[row[2]]["sid"] = Sid
|
||||
sendMobiles[row[2]]["isPersonalizedMsg"] = false
|
||||
sendMobiles[row[2]]["content"] = content
|
||||
sendMobiles[row[2]]["mobiles"] = []string{}
|
||||
}
|
||||
|
||||
sendMobiles[row[2]]["mobiles"] = append(sendMobiles[row[2]]["mobiles"].([]string), row[3])
|
||||
|
||||
if len(sendMobiles[row[2]]["mobiles"].([]string)) >= batchSize {
|
||||
mobiles := sendMobiles[row[2]]["mobiles"].([]string)
|
||||
} else { //非个性化
|
||||
mobiles, ok := smsData.Mobiles, len(smsData.Mobiles) > 0 && true
|
||||
if ok && len(mobiles) > 0 {
|
||||
mobileStr := strings.Join(mobiles, ",")
|
||||
var sl []SmsList
|
||||
sl = append(sl, SmsList{M: mobileStr, C: content, F: 8})
|
||||
sd := Sms{Sid: Sid, Data: sl, Token: token}
|
||||
sl = append(sl, SmsList{M: mobileStr, C: smsData.Content, F: 8})
|
||||
sd := Sms{Sid: smsData.Sid, Data: sl, Token: token}
|
||||
sendSMSDataJson, _ := json.Marshal(sd)
|
||||
smsApi("appendbatchdata", string(sendSMSDataJson))
|
||||
sendMobiles[row[2]]["mobiles"] = []string{} // reset mobiles slice
|
||||
|
||||
smsData.Mobiles = []string{} // reset mobiles slice
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if _, ok := insertsCount[row[2]]; !ok {
|
||||
insertsCount[row[2]] = 0
|
||||
}
|
||||
insertsCount[row[2]]++
|
||||
count++
|
||||
}
|
||||
|
||||
if len(dataBatch) > 0 {
|
||||
dataBatchChan <- dataBatch
|
||||
}
|
||||
|
||||
close(dataBatchChan)
|
||||
|
||||
//fmt.Println("结束批次v%\n", ccids)
|
||||
//fmt.Print("ccids 长度", len(ccids), "\n")
|
||||
for ccid := range ccids { // 处理各个批次剩余数据，同时处理批次结束api
|
||||
|
||||
fmt.Println("循环处理批次结束\n", ccid)
|
||||
|
||||
//batches := result[ccid]
|
||||
if sendMobiles[ccid]["isPersonalizedMsg"].(bool) { //个性化
|
||||
smsList := sendMobiles[ccid]["mobiles"].([]SmsList)
|
||||
if len(smsList) > 0 {
|
||||
sd := Sms{Sid: sendMobiles[ccid]["sid"].(int), Data: sendMobiles[ccid]["mobiles"].([]SmsList), Token: token}
|
||||
sendSMSDataJson, _ := json.Marshal(sd)
|
||||
smsApi("appendbatchdata", string(sendSMSDataJson))
|
||||
//fmt.Print(string(sendSMSDataJson))
|
||||
sendMobiles[ccid]["mobiles"] = []SmsList{} // reset mobiles slice
|
||||
sf := SmsFinish{Sid: smsData.Sid, Token: token}
|
||||
sfJson, _ := json.Marshal(sf)
|
||||
smsApi("finishbatch", string(sfJson))
|
||||
bpi := []BatchProcessingInformation{}
|
||||
bpi = append(bpi, BatchProcessingInformation{
|
||||
CommunicationChannelID: ccid,
|
||||
RepeatTargetsMember: duplicateCount[ccid],
|
||||
InsertsTargetsMember: insertsCount[ccid],
|
||||
DataFileName: fileName,
|
||||
})
|
||||
err = db.CreateInBatches(bpi, insertSize).Error
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("插入批次处理信息失败,文件名：%s，错误信息%v", fileName, err))
|
||||
}
|
||||
smsData = SmsData{}
|
||||
sendMobiles[ccid] = smsData
|
||||
}
|
||||
|
||||
} else { //非个性化
|
||||
if mobiles, ok := sendMobiles[ccid]["mobiles"].([]string); ok && len(mobiles) > 0 {
|
||||
mobileStr := strings.Join(sendMobiles[ccid]["mobiles"].([]string), ",")
|
||||
var sl []SmsList
|
||||
sl = append(sl, SmsList{M: mobileStr, C: sendMobiles[ccid]["content"].(string), F: 8})
|
||||
wg.Wait() //所有入库全部完成
|
||||
|
||||
sd := Sms{Sid: sendMobiles[ccid]["sid"].(int), Data: sl, Token: token}
|
||||
sendSMSDataJson, _ := json.Marshal(sd)
|
||||
smsApi("appendbatchdata", string(sendSMSDataJson))
|
||||
//插入批此重复数据
|
||||
if len(dataBatchDuplicate) > 0 {
|
||||
err = db.CreateInBatches(dataBatchDuplicate, insertSize).Error
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("插入重复数据失败,文件名：%s，错误信息%v", fileName, err))
|
||||
} else {
|
||||
dataBatchDuplicate = nil
|
||||
}
|
||||
}
|
||||
|
||||
//关闭批次
|
||||
sf := SmsFinish{Sid: sendMobiles[ccid]["sid"].(int), Token: token}
|
||||
sfJson, _ := json.Marshal(sf)
|
||||
smsApi("finishbatch", string(sfJson))
|
||||
elapsed := time.Since(start)
|
||||
|
||||
//插入批次处理数据
|
||||
bpi := []BatchProcessingInformation{}
|
||||
bpi = append(bpi, BatchProcessingInformation{
|
||||
CommunicationChannelID: ccid,
|
||||
RepeatTargetsMember: duplicateCount[ccid],
|
||||
InsertsTargetsMember: insertsCount[ccid],
|
||||
DataFileName: fileName,
|
||||
})
|
||||
err = db.CreateInBatches(bpi, insertSize).Error
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("插入批次处理信息失败,文件名：%s，错误信息%v", fileName, err))
|
||||
}
|
||||
//发送提醒邮件
|
||||
subject := "丝芙兰数据包处理完成"
|
||||
body := fmt.Sprintf("总数：%d;\n数据包：%s;\n过滤重复数：%d;\n过滤后总数：%d;\n处理完成，请前往管理平台查看处理。", count+bi, fileName, bi, count)
|
||||
SendEmail(subject, body) //发送邮件
|
||||
applogger.Info(fmt.Sprintf("%s（数据包） 入库完成", fileName))
|
||||
applogger.Info(fmt.Sprintf("%s（数据包）执行时间:%s 插入数据:%d条 过滤数数：%d条", fileName, elapsed, count, bi))
|
||||
return 0
|
||||
} else { //如果没有查询到批次信息，跳过处理
|
||||
return -1
|
||||
}
|
||||
wg.Wait() //所有入库全部完成
|
||||
|
||||
//插入批此重复数据
|
||||
if len(dataBatchDuplicate) > 0 {
|
||||
err = db.CreateInBatches(dataBatchDuplicate, insertSize).Error
|
||||
if err != nil {
|
||||
applogger.Error(fmt.Sprintf("插入重复数据失败,文件名：%s，错误信息%v", fileName, err))
|
||||
} else {
|
||||
dataBatchDuplicate = nil
|
||||
}
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
|
||||
//发送提醒邮件
|
||||
subject := "丝芙兰数据包处理完成"
|
||||
body := "数据包：" + fileName + ";\n总数：" + strconv.Itoa(count+bi) + ";\n过滤重复数：" + strconv.Itoa(bi) + ";\n过滤后总数：" + strconv.Itoa(count) + ";\n处理完成，请前往管理平台查看处理。"
|
||||
SendEmail(subject, body) //发送邮件
|
||||
applogger.Info(fmt.Sprintf("%s（数据包） 入库完成", fileName))
|
||||
applogger.Info(fmt.Sprintf("%s（数据包）执行时间:%s 插入数据:%d条 过滤数数：%d条", fileName, elapsed, count, bi))
|
||||
}
|
||||
}
|
||||
|
||||
@ -998,3 +1032,10 @@ type SmsFinish struct {
|
||||
Sid int `json:"sid"`
|
||||
Token string `json:"token"`
|
||||
}
|
||||
type SmsData struct {
|
||||
Sid int
|
||||
IsPersonalizedMsg bool
|
||||
Content string
|
||||
Mobiles []string
|
||||
SmsList []SmsList
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user