package bus_service import ( "bytes" "errors" "fmt" "github.com/go-admin-team/go-admin-core/logger" "github.com/go-admin-team/go-admin-core/sdk/service" "github.com/xuri/excelize/v2" "go-admin/app/admin/models/bus_models" "go-admin/common/redisx" "golang.org/x/net/context" "gorm.io/gorm" "math/rand" "strconv" "strings" "time" "unicode/utf8" ) const ( maxPhonesPerShard = 10000 cacheExpire = time.Hour TimeFormat = "2006-01-02T15:04:05+08:00" ExportFile = "/www/server/images/export/" MiGuExportUrl = "https://telecom.deovo.com/load/export/" ) type SmsService struct { service.Service } // ReadExcelFile 读取 Excel 文件并提取第一列的手机号 func (s *SmsService) ReadExcelFile(file []byte) ([]string, error) { // 使用 excelize.OpenReader 直接从文件流读取内容 f, err := excelize.OpenReader(bytes.NewReader(file)) if err != nil { return nil, fmt.Errorf("无法打开 Excel 文件: %v", err) } var phones []string // 获取所有工作表列表 sheetList := f.GetSheetList() if len(sheetList) == 0 { return nil, fmt.Errorf("excel 文件中没有工作表") } // 选择第一个工作表 sheet := sheetList[0] rows, err := f.GetRows(sheet) if err != nil { return nil, fmt.Errorf("读取 Excel 行失败: %v", err) } // 假设第一列是手机号 for _, row := range rows { if len(row) > 0 { phone := row[0] phones = append(phones, phone) } } return phones, nil } func (s *SmsService) CachePhonesByShard(importSerial string, phones []string) error { ctx := context.Background() total := 0 for i := 0; i < len(phones); i += maxPhonesPerShard { end := i + maxPhonesPerShard if end > len(phones) { end = len(phones) } shard := phones[i:end] total++ key := fmt.Sprintf("sms:import:%s:%d", importSerial, total) if err := redisx.Client.RPush(ctx, key, shard).Err(); err != nil { return err } redisx.Client.Expire(ctx, key, cacheExpire) } // 保存总分片数量 totalKey := fmt.Sprintf("sms:import:%s:total", importSerial) err := redisx.Client.Set(ctx, totalKey, total, cacheExpire).Err() return err } func (s *SmsService) AppendPhonesToRedis(serial string, phones []string) error { ctx := context.Background() // 拿当前最大分片号 totalKey := fmt.Sprintf("sms:import:%s:total", serial) totalStr, err := redisx.Client.Get(ctx, totalKey).Result() if err != nil { return fmt.Errorf("无法读取缓存分片: %v", err) } total, _ := strconv.Atoi(totalStr) if total == 0 { return fmt.Errorf("缓存分片不存在") } // 直接往最后一片中添加(你也可以按量分片再扩展) lastKey := fmt.Sprintf("sms:import:%s:%d", serial, total) values := make([]interface{}, len(phones)) for i, p := range phones { values[i] = p } err = redisx.Client.RPush(ctx, lastKey, values...).Err() return err } func (s *SmsService) SubmitSmsTaskFromRedis(ctx context.Context, serial, content, sendTime, coopNum, coopName string) error { // 读取分片数量 totalKey := fmt.Sprintf("sms:import:%s:total", serial) totalStr, err := redisx.Client.Get(ctx, totalKey).Result() if err != nil { return fmt.Errorf("获取缓存分片数失败: %v", err) } total, _ := strconv.Atoi(totalStr) if total == 0 { return fmt.Errorf("分片数为 0,无法创建任务") } // 统计总手机号数量 var totalPhones int phoneCounts := make([]int, total) for i := 1; i <= total; i++ { redisKey := fmt.Sprintf("sms:import:%s:%d", serial, i) count, err := redisx.Client.LLen(ctx, redisKey).Result() if err != nil { return fmt.Errorf("读取缓存分片 %d 失败: %v", i, err) } totalPhones += int(count) phoneCounts[i-1] = int(count) } if totalPhones == 0 { return fmt.Errorf("手机号为空,不能创建任务") } var planTime time.Time // 判断是否设置了定时发送 if sendTime != "" { loc, _ := time.LoadLocation("Asia/Shanghai") planTime, err = time.ParseInLocation(TimeFormat, sendTime, loc) if err != nil { return fmt.Errorf("解析时间出错:%s", err.Error()) } } // 计算短信条数(按70字分割) // 短信条数计算:70字以内算1条,超过则每67字拆1条(长短信按67字分段) contentCost := (len([]rune(content)) + 66) / 67 totalSmsCount := totalPhones * contentCost // 使用数据库事务 return s.Orm.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // 生成批次ID batchID, _ := GenerateBatchID(s.Orm) // 插入任务 task := &bus_models.SmsTask{ CooperativeNumber: coopNum, // 可根据需要填充 CooperativeName: coopName, BatchID: batchID, ImportID: serial, SmsContent: content, SmsContentCost: contentCost, TotalPhoneCount: totalPhones, TotalSmsCount: totalSmsCount, Status: 0, InterceptFailCount: 0, ChannelFailCount: 0, ScheduleTime: &planTime, } if err := tx.Create(task).Error; err != nil { return fmt.Errorf("创建短信任务失败: %v", err) } // 插入批次记录 for i := 1; i <= total; i++ { batch := &bus_models.SmsTaskBatch{ CooperativeNumber: coopNum, // 可根据需要填充 CooperativeName: coopName, TaskID: task.ID, BatchID: batchID, ImportID: serial, Num: i, PhoneCount: phoneCounts[i-1], SmsCount: phoneCounts[i-1] * contentCost, SmsContent: content, Status: 0, ScheduleTime: &planTime, } if err := tx.Create(batch).Error; err != nil { return fmt.Errorf("创建批次失败: %v", err) } } // ✅ 提示:不在这里生成 SmsSendRecord,后续通过消费者从 Redis 中读取批次生成 // 否则数据量大时会影响事务和响应速度 return nil }) } // GenerateBatchID 生成唯一的批次ID:日期(YYYYMMDD)+ 8位随机数(共16位) func GenerateBatchID(db *gorm.DB) (string, error) { // 获取当前日期(年月日) dateStr := time.Now().Format("20060102") // 例如:20250411 // 生成一个8位随机数(范围:00000000 ~ 99999999) rand.Seed(time.Now().UnixNano()) randomNum := rand.Int63n(100000000) // 8位最大是1亿,即10^8 // 格式化为8位字符串(左侧补0) randomNumStr := fmt.Sprintf("%08d", randomNum) // 拼接成 batch_id batchID := fmt.Sprintf("%s%s", dateStr, randomNumStr) // 检查 sms_task 表中是否已存在该 batch_id var count int64 err := db.Table("sms_task").Where("batch_id = ?", batchID).Count(&count).Error if err != nil { return "", err } // 如果已存在,则递归重试 if count > 0 { return GenerateBatchID(db) } return batchID, nil } func (s *SmsService) ExportPhoneListToExcel(importSerial string, directPhones []string) (string, error) { ctx := context.Background() var allPhones []string // 从 Redis 获取导入的号码 if importSerial != "" { totalKey := fmt.Sprintf("sms:import:%s:total", importSerial) totalStr, err := redisx.Client.Get(ctx, totalKey).Result() if err == nil { total, _ := strconv.Atoi(totalStr) for i := 1; i <= total; i++ { key := fmt.Sprintf("sms:import:%s:%d", importSerial, i) shardPhones, err := redisx.Client.LRange(ctx, key, 0, -1).Result() if err == nil { allPhones = append(allPhones, shardPhones...) } } } } // 追加直接输入的号码 if len(directPhones) > 0 { allPhones = append(allPhones, directPhones...) } if len(allPhones) == 0 { return "", fmt.Errorf("没有可导出的号码") } // 创建 Excel 文件 file := excelize.NewFile() sheet := "Sheet1" for i, phone := range allPhones { cell := fmt.Sprintf("A%d", i+1) file.SetCellValue(sheet, cell, phone) } // 设置样式 style, _ := file.NewStyle(&excelize.Style{ Alignment: &excelize.Alignment{ Horizontal: "center", Vertical: "center", }, }) _ = file.SetCellStyle(sheet, "A1", fmt.Sprintf("A%d", len(allPhones)), style) file.SetColWidth(sheet, "A", "A", 20) // 保存文件 fileName := time.Now().Format("20060102150405") + "_号码导出.xlsx" url := MiGuExportUrl + fileName if err := file.SaveAs(ExportFile + fileName); err != nil { logger.Errorf("导出Excel失败: %v", err) return "", err } return url, nil } // CheckSensitiveWords 检查短信内容是否包含敏感词 func (s *SmsService) CheckSensitiveWords(content string) ([]string, error) { var sensitiveWords []bus_models.SensitiveWord err := s.Orm.Table("sensitive_words").Where("is_enabled = ?", 1). Find(&sensitiveWords).Error if err != nil { return nil, err } var hits []string for _, sw := range sensitiveWords { if strings.Contains(content, sw.Word) { hits = append(hits, sw.Word) } } return hits, nil } // QuerySmsTaskList 查询短信下行记录 func (s *SmsService) QuerySmsTaskList(req bus_models.SmsTaskQueryRequest, db *gorm.DB) (*bus_models.SmsTaskQueryResponse, error) { var tasks []bus_models.SmsTask var total int64 query := db.Model(&bus_models.SmsTask{}).Where("status = 2") if req.BatchID != "" { query = query.Where("batch_id = ?", req.BatchID) } if req.MinTotalSms > 0 { query = query.Where("total_sms_count >= ?", req.MinTotalSms) } if req.MinPhoneCount > 0 { query = query.Where("total_phone_count >= ?", req.MinPhoneCount) } if req.Status != nil { query = query.Where("status = ?", *req.Status) } if req.StartTime != "" && req.EndTime != "" { query = query.Where("created_at BETWEEN ? AND ?", req.StartTime, req.EndTime) } // 统计总数 if err := query.Count(&total).Error; err != nil { return nil, err } // 分页处理 page := req.Page if page < 1 { page = 1 } pageSize := req.PageSize if pageSize <= 0 { pageSize = 10 } offset := (page - 1) * pageSize // 查询分页数据 if err := query. Order("created_at DESC"). Limit(pageSize). Offset(offset). Find(&tasks).Error; err != nil { return nil, err } totalPage := int((total + int64(pageSize) - 1) / int64(pageSize)) if totalPage < 1 { totalPage = 1 } return &bus_models.SmsTaskQueryResponse{ List: tasks, Total: total, Page: page, PageSize: pageSize, TotalPage: totalPage, }, nil } func (s *SmsService) QuerySmsSendRecords(req bus_models.SmsSendRecordQueryReq) (bus_models.SmsSendRecordQueryResp, error) { var resp bus_models.SmsSendRecordQueryResp db := s.Orm.Model(&bus_models.SmsSendRecord{}) if req.BatchID != "" { db = db.Where("batch_id = ?", req.BatchID) } if req.Phone != "" { db = db.Where("phone = ?", req.Phone) } if req.SmsCode != "" { db = db.Where("sms_code = ?", req.SmsCode) } if req.StartTime != "" { db = db.Where("receive_time >= ?", req.StartTime) } if req.EndTime != "" { db = db.Where("receive_time <= ?", req.EndTime) } if req.MinSegments > 0 { // 计算短信计费条数的粗略估算:每67字1条,注意实际可以存字段 db = db.Where("(length(sms_content) / 67) + 1 >= ?", req.MinSegments) } err := db.Count(&resp.Total).Error if err != nil { return resp, err } // 分页处理 page := req.Page if page < 1 { page = 1 } pageSize := req.PageSize if pageSize <= 0 { pageSize = 10 } offset := (page - 1) * pageSize // 查询分页数据 if err = db. Order("created_at DESC"). Limit(pageSize). Offset(offset). Find(&resp.List).Error; err != nil { return resp, err } totalPage := int((resp.Total + int64(pageSize) - 1) / int64(pageSize)) if totalPage < 1 { totalPage = 1 } resp.Page = page resp.PageSize = pageSize resp.TotalPage = totalPage return resp, nil } // GetPhonesFromCache 获取导入手机号前 limit 条 func (s *SmsService) GetPhonesFromCache(importSerial string, limit int) ([]string, error) { ctx := context.Background() totalKey := fmt.Sprintf("sms:import:%s:total", importSerial) totalStr, err := redisx.Client.Get(ctx, totalKey).Result() if err != nil { return nil, err } totalShards, err := strconv.Atoi(totalStr) if err != nil { return nil, err } result := make([]string, 0, limit) for i := 1; i <= totalShards; i++ { key := fmt.Sprintf("sms:import:%s:%d", importSerial, i) phones, err := redisx.Client.LRange(ctx, key, 0, -1).Result() if err != nil { return nil, err } result = append(result, phones...) if len(result) >= limit { return result[:limit], nil } } return result, nil } // GetSentPhonesByBatchID 从数据库查询已发送的手机号,最多返回 limit 条 func (s *SmsService) GetSentPhonesByBatchID(db *gorm.DB, batchID string, limit int) ([]string, error) { var records []bus_models.SmsSendRecord err := db. Where("batch_id = ?", batchID). Order("id DESC"). Limit(limit). Find(&records).Error if err != nil { return nil, err } var phones []string for _, record := range records { phones = append(phones, record.Phone) } return phones, nil } // QueryScheduledSmsTaskList 查询定时短信任务 func (s *SmsService) QueryScheduledSmsTaskList(req bus_models.SmsTaskScheduledQueryRequest, db *gorm.DB) (*bus_models.SmsTaskQueryResponse, error) { var tasks []bus_models.SmsTask var total int64 query := db.Model(&bus_models.SmsTask{}). Where("schedule_time IS NOT NULL"). Where("status = ?", 0) if req.StartTime != "" && req.EndTime != "" { query = query.Where("schedule_time BETWEEN ? AND ?", req.StartTime, req.EndTime) } // 统计总数 if err := query.Count(&total).Error; err != nil { return nil, err } // 分页处理 page := req.Page if page < 1 { page = 1 } pageSize := req.PageSize if pageSize <= 0 { pageSize = 10 } offset := (page - 1) * pageSize if err := query. Order("schedule_time ASC"). Limit(pageSize). Offset(offset). Find(&tasks).Error; err != nil { return nil, err } totalPage := int((total + int64(pageSize) - 1) / int64(pageSize)) if totalPage < 1 { totalPage = 1 } return &bus_models.SmsTaskQueryResponse{ List: tasks, Total: total, Page: page, PageSize: pageSize, TotalPage: totalPage, }, nil } func (s *SmsService) QuerySmsUplinkList(req bus_models.SmsUplinkQueryRequest, db *gorm.DB) (bus_models.SmsUplinkQueryResponse, error) { var resp bus_models.SmsUplinkQueryResponse var total int64 query := db.Table("sms_uplink_log AS uplink"). Select(` uplink.id AS uplink_id, uplink.phone_number, sr.id AS send_id, sr.sms_content, uplink.reply_content, uplink.created_at AS receive_time, sr.created_at AS send_time `). Joins("LEFT JOIN sms_send_record sr ON uplink.batch_id = sr.batch_id AND uplink.phone_number = sr.phone") // 条件拼接 if req.PhoneNumber != "" { query = query.Where("uplink.phone_number LIKE ?", "%"+req.PhoneNumber+"%") } if req.SendID > 0 { query = query.Where("sr.id = ?", req.SendID) } if req.UplinkID > 0 { query = query.Where("uplink.id = ?", req.UplinkID) } if req.ReplyContent != "" { query = query.Where("uplink.reply_content LIKE ?", "%"+req.ReplyContent+"%") } if req.StartTime != "" && req.EndTime != "" { query = query.Where("uplink.created_at BETWEEN ? AND ?", req.StartTime, req.EndTime) } // 统计总数 if err := query.Count(&total).Error; err != nil { return resp, err } // 分页处理 page := req.Page if page <= 0 { page = 1 } pageSize := req.PageSize if pageSize <= 0 { pageSize = 10 } offset := (page - 1) * pageSize var results []bus_models.SmsUplinkRecordResponse // 查询数据 err := query.Order("uplink.created_at DESC").Limit(pageSize).Offset(offset).Scan(&results).Error if err != nil { return resp, err } resp.List = results resp.Total = total resp.Page = page resp.PageSize = pageSize resp.TotalPage = (total + int64(page) - 1) / int64(pageSize) if resp.TotalPage < 1 { resp.TotalPage = 1 } return resp, nil } // BatchUpdateSmsContent 修改短信内容 func (s *SmsService) BatchUpdateSmsContent(req bus_models.BatchUpdateSmsContentRequest, db *gorm.DB) error { // 计算短信内容消耗(每70字符1条,超出部分算1条) runeCount := utf8.RuneCountInString(req.SmsContent) smsCost := runeCount / 70 if runeCount%70 != 0 { smsCost++ } // 批量更新 SmsTask if err := db.Model(&bus_models.SmsTask{}). Where("id IN ?", req.TaskIDs). Updates(map[string]interface{}{ "sms_content": req.SmsContent, "sms_content_cost": smsCost, }).Error; err != nil { return err } // 同步更新 SmsTaskBatch 中相应内容 if err := db.Model(&bus_models.SmsTaskBatch{}). Where("task_id IN ?", req.TaskIDs). Updates(map[string]interface{}{ "sms_content": req.SmsContent, }).Error; err != nil { return err } return nil } func (s *SmsService) BatchCancelSmsTasks(req bus_models.BatchUpdateRequest, db *gorm.DB) error { if len(req.TaskIDs) == 0 { return errors.New("任务ID列表不能为空") } // 取消任务状态为 4(取消) if err := db.Model(&bus_models.SmsTask{}). Where("id IN ?", req.TaskIDs). Updates(map[string]interface{}{ "status": 4, }).Error; err != nil { return err } // 同步取消子任务 if err := db.Model(&bus_models.SmsTaskBatch{}). Where("task_id IN ?", req.TaskIDs). Updates(map[string]interface{}{ "status": 4, }).Error; err != nil { return err } return nil } // BatchResetScheduleTime 批量重置定时时间 func (s *SmsService) BatchResetScheduleTime(req bus_models.BatchResetScheduleTimeRequest, db *gorm.DB) error { if len(req.TaskIDs) == 0 || req.ScheduleTime == "" { return errors.New("任务ID和定时时间不能为空") } var planTime time.Time var err error // 判断是否设置了定时发送 if req.ScheduleTime != "" { loc, _ := time.LoadLocation("Asia/Shanghai") planTime, err = time.ParseInLocation(TimeFormat, req.ScheduleTime, loc) if err != nil { return fmt.Errorf("解析时间出错:%s", err.Error()) } } // 更新 SmsTask if err := db.Model(&bus_models.SmsTask{}). Where("id IN ?", req.TaskIDs). Updates(map[string]interface{}{ "schedule_time": planTime, }).Error; err != nil { return err } // 同步更新 SmsTaskBatch if err := db.Model(&bus_models.SmsTaskBatch{}). Where("task_id IN ?", req.TaskIDs). Updates(map[string]interface{}{ "schedule_time": planTime, }).Error; err != nil { return err } return nil } // CreateSignatureRealname 创建签名实名制记录 func (s *SmsService) CreateSignatureRealname(data *bus_models.SmsSignatureRealname, db *gorm.DB) error { return db.Create(data).Error } // UpdateSignatureRealname 编辑签名实名制记录 func (s *SmsService) UpdateSignatureRealname(data *bus_models.SmsSignatureRealname, db *gorm.DB) error { if data.ID == 0 { return errors.New("ID不能为空") } return db.Model(&bus_models.SmsSignatureRealname{}). Where("id = ?", data.ID). Updates(data).Error } // BatchDeleteSignatureRealname 删除签名实名制记录 func (s *SmsService) BatchDeleteSignatureRealname(ids []uint, db *gorm.DB) error { if len(ids) == 0 { return errors.New("删除ID列表不能为空") } return db.Where("id IN ?", ids).Delete(&bus_models.SmsSignatureRealname{}).Error } func (s *SmsService) ListSignatureRealname(req bus_models.SignatureRealnameQuery, db *gorm.DB) (bus_models.SignatureRealnameQueryResp, error) { var resp bus_models.SignatureRealnameQueryResp var list []bus_models.SmsSignatureRealname var total int64 query := db.Model(&bus_models.SmsSignatureRealname{}) if req.Signature != "" { query = query.Where("signature LIKE ?", "%"+req.Signature+"%") } if req.CooperativeName != "" { query = query.Where("cooperative_name LIKE ?", "%"+req.CooperativeName+"%") } if req.CompanyName != "" { query = query.Where("company_name LIKE ?", "%"+req.CompanyName+"%") } if req.CompanyCreditCode != "" { query = query.Where("company_credit_code LIKE ?", "%"+req.CompanyCreditCode+"%") } if req.ResponsibleName != "" { query = query.Where("responsible_name LIKE ?", "%"+req.ResponsibleName+"%") } if req.UsageCategory != 0 { query = query.Where("usage_category = ?", req.UsageCategory) } if req.IsActive != 0 { query = query.Where("is_active = ?", req.IsActive) } // 获取总条数 err := query.Count(&total).Error if err != nil { return resp, err } // 分页处理 page := req.Page if page <= 0 { page = 1 } pageSize := req.PageSize if pageSize <= 0 { pageSize = 10 } offset := (page - 1) * pageSize // 分页查询 err = query.Order("id desc"). Offset(offset). Limit(pageSize). Find(&list).Error if err != nil { return resp, err } // 构建分页响应 totalPage := (total + int64(pageSize) - 1) / int64(pageSize) if totalPage < 1 { totalPage = 1 } resp = bus_models.SignatureRealnameQueryResp{ List: list, Total: total, Page: page, PageSize: pageSize, TotalPage: totalPage, } return resp, nil } func (s *SmsService) ListContacts(req bus_models.ContactQuery, db *gorm.DB) (bus_models.ContactQueryResp, error) { var resp bus_models.ContactQueryResp var list []bus_models.SmsContact var total int64 query := db.Model(&bus_models.SmsContact{}) if req.Name != "" { query = query.Where("name LIKE ?", "%"+req.Name+"%") } if req.PhoneNumber != "" { query = query.Where("phone_number LIKE ?", "%"+req.PhoneNumber+"%") } err := query.Count(&total).Error if err != nil { return resp, err } page := req.Page if page <= 0 { page = 1 } pageSize := req.PageSize if pageSize <= 0 { pageSize = 10 } offset := (page - 1) * pageSize err = query.Order("id desc"). Offset(offset). Limit(pageSize). Find(&list).Error if err != nil { return resp, err } totalPage := (total + int64(pageSize) - 1) / int64(pageSize) if totalPage < 1 { totalPage = 1 } resp = bus_models.ContactQueryResp{ List: list, Total: total, Page: page, PageSize: pageSize, TotalPage: totalPage, } return resp, nil } // AddContact 添加联系人 func (s *SmsService) AddContact(contact bus_models.SmsContact, db *gorm.DB) error { return db.Create(&contact).Error } // EditContact 编辑联系人 func (s *SmsService) EditContact(id string, contact bus_models.SmsContact, db *gorm.DB) error { var existingContact bus_models.SmsContact if err := db.Where("id = ?", id).First(&existingContact).Error; err != nil { return err } // 更新联系人字段 existingContact.Name = contact.Name existingContact.PhoneNumber = contact.PhoneNumber existingContact.Gender = contact.Gender existingContact.Birthday = contact.Birthday existingContact.Company = contact.Company existingContact.Address = contact.Address existingContact.Remark = contact.Remark existingContact.CooperativeName = contact.CooperativeName existingContact.CooperativeNumber = contact.CooperativeNumber return db.Save(&existingContact).Error } // BulkDeleteContacts 批量删除联系人 func (s *SmsService) BulkDeleteContacts(req bus_models.ContactDeleteRequest, db *gorm.DB) error { return db.Where("id IN (?)", req.ContactIDs).Delete(&bus_models.SmsContact{}).Error }