package broadcast import ( "bytes" "context" "fmt" "io" "net" "net/http" "pbx-api-gin/internal/app/ami/action" "pbx-api-gin/internal/app/ami/model" "pbx-api-gin/internal/app/mysql" msgdata "pbx-api-gin/internal/app/stc/data" "pbx-api-gin/pkg/lfshook" "strings" "sync" "time" ) var Pacus = []string{"2111", "2121", "2131", "2141", "2151", "2161", "2171", "2181"} func HandleStcCmd(ctx context.Context, conn net.Conn) { for { select { case <-ctx.Done(): lfshook.NewLogger().Logger.Infof("HandleStcCmd===ctx==ret======") return default: var buf bytes.Buffer // 用于累积未处理完的数据流 tmp := make([]byte, 1024) if conn != nil { n, err := conn.Read(tmp) if err != nil { if err != io.EOF { fmt.Println("Error reading from server:", err) conn.Close() } return } // 将新读取的数据追加到缓冲区 buf.Write(tmp[:n]) } // 尝试从缓冲区中提取完整数据包 for { packet, err := msgdata.ExtractPacket(&buf) if err != nil { break // 没有完整包或出错,等待更多数据 } // 成功提取一个包,进行处理 go processPacket(packet) // 使用 goroutine 避免阻塞接收 } } } } // 处理单个数据包(原 switch 逻辑迁移过来) func processPacket(packet []byte) { if len(packet) < 6 { fmt.Println("Invalid packet length") return } lfshook.NewLogger().Logger.Infof("Get data from STC ===============:%x", packet) switch packet[5] { case 0x01: // heartbeat // handle heartbeat case 0x02: // STN StationAnn(packet) case 0x03: // ACTIVE Active([1]byte{packet[8]}) case 0x05: // SPC SpecialAnn(packet) case 0x06: // EMG EmgMsg(packet) case 0x07: // STOP AnnStop([4]byte{packet[8], packet[9], packet[10], packet[11]}) case 0x08: // DCS DcsAnn(packet) case 0x09: // SELF CHECK SelfCheck(packet) case 0x0a: AlarmHandle(packet) case 0x0b: AlarmResetAll() case 0x0c: RecordStorageConf(packet[8:]) default: fmt.Printf("Unknown command: %x\n", packet[5]) } } // STN , 自动报站广播 func StationAnn(data []byte) (err error) { specialVoice := int(data[8]) delay := data[9] cycleCount := data[10] datalen := int(data[11]) filename := msgdata.SubstrByRune(string(data[12:]), 0, datalen-4) filename = strings.ReplaceAll(filename, ".wav", "") filename = strings.ReplaceAll(filename, ".mp3", "") lfshook.NewLogger().Logger.Infof("=============Get filename : %v", filename) //update special voice _, er := mysql.DBOrmInstance.In("exten", Pacus).Update(&model.Extension{Special: specialVoice, PaType: "STN"}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) return er } action.PlaybackPacu(filename, int(cycleCount), int(delay), "STN") return nil } // 激活信号 func Active(data [1]byte) { Num := int(data[0]) switch Num { // 设置全局的激活信号,并通过协议(待定)通知终端注册到对应的激活主机上 case 0: case 1: case 8: case 9: } } // SPC ,特殊服务消息广播 func SpecialAnn(data []byte) { delay := data[8] cycleCount := data[9] datalen := int(data[10]) filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-3) //update pa type _, er := mysql.DBOrmInstance.In("exten", Pacus).Update(&model.Extension{PaType: "SPC"}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } if int(cycleCount) == 255 { action.PlaybackPacu(filename, 9999999, int(delay), "SPC") } else { action.PlaybackPacu(filename, int(cycleCount), int(delay), "SPC") } } // EMG ,紧急服务消息广播 func EmgMsg(data []byte) { delay := data[8] cycleCount := data[9] datalen := int(data[10]) filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-3) //update pa type _, er := mysql.DBOrmInstance.In("exten", Pacus).Update(&model.Extension{PaType: "EMG"}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } if int(cycleCount) == 255 { action.PlaybackPacu(filename, 9999999, int(delay), "EMG") } else { action.PlaybackPacu(filename, int(cycleCount), int(delay), "EMG") } } // 停止指定类型广播 func AnnStop(data [4]byte) { switch data[0] { case 0x03: case 0x04: case 0x07: case 0x08: case 0x09: } for _, ext := range Pacus { action.Hangup(ext) } //update pa type _, er := mysql.DBOrmInstance.Cols("patype").In("exten", Pacus).Update(&model.Extension{PaType: ""}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } } // DCS 语音 func DcsAnn(data []byte) { delay := data[8] cycleCount := data[9] datalen := int(data[10]) filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-3) //update pa type _, er := mysql.DBOrmInstance.In("exten", Pacus).Update(&model.Extension{PaType: "DCS"}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } if int(cycleCount) == 255 { action.PlaybackPacu(filename, 9999999, int(delay), "DCS") } else { action.PlaybackPacu(filename, int(cycleCount), int(delay), "DCS") } } // 自检广播 func SelfCheck(data []byte) { check := data[8] delay := data[9] cycleCount := data[10] datalen := int(data[11]) filename := msgdata.SubstrByRune(string(data[12:]), 0, datalen-4) //update pa type _, er := mysql.DBOrmInstance.In("exten", Pacus).Update(&model.Extension{PaType: "CHK"}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } switch check { case 0x01: action.PlaybackPacu(filename, int(cycleCount), int(delay), "CHK") case 0x02: //asterisk.Hangup(Exten) } } // 全局变量:记录正在抑制的 exten var ( suppressedExts = sync.Map{} // map[string]struct{},值存在即表示被抑制 suppressionMu sync.Mutex // 保护初始化和清理操作(可选) ) // suppressKey 生成用于抑制的 key(可以根据需求扩展) func suppressKey(exten string, handler byte) string { return fmt.Sprintf("%s_h%x", exten, handler) } // ICP操作乘客报警(根据激活信息判断转到1车还是8车================) func AlarmHandle(data []byte) { handler := data[8] //extlen := data[9] carr := data[12] pos := data[13] exten := fmt.Sprintf("24%c%c", carr, pos) lfshook.NewLogger().Logger.Infof("================ICP Answer PAD================:%s ", exten) key := suppressKey(exten, handler) // 只对 handler == 0x01 做 2 秒去重 if handler == 0x01 { if _, loaded := suppressedExts.LoadOrStore(key, struct{}{}); loaded { lfshook.NewLogger().Logger.Infof("Suppressed duplicate ICP Alarm (handler=0x01) for exten: %s within 2 seconds", exten) return // 已存在,说明在2秒窗口期内,直接丢弃 } // 设置2秒后删除该 key,允许下次通过 time.AfterFunc(4*time.Second, func() { suppressedExts.Delete(key) lfshook.NewLogger().Logger.Debugf("Suppression released for key: %s", key) }) } // 设置2秒后删除该 key,允许下次通过 time.AfterFunc(1*time.Second, func() { suppressedExts.Delete(key) lfshook.NewLogger().Logger.Debugf("Suppression released for key: %s", key) }) switch handler { case 0x01: //answer(ICP+Alarm+PACU) //NotifyPaiu(exten, "answer") err := action.RedirectInQueue(exten, "0402", "ani-rule", "1") // 1车ICP接听PAIU if err != nil { lfshook.NewLogger().Logger.Infof("================ICP Answer PAD====ERR============ : %+v", err.Error()) } //invite PACU join in //action.Hangup("PACU") //action.ChanSpy("PACU", exten, false, true) case 0x04: //answer(ICP+Alarm+PACU) err := action.RedirectInQueue(exten, "0401", "ano-rule", "1") // 1车OCC接听PAIU if err != nil { //lfshook.NewLogger().Info(err) lfshook.NewLogger().Logger.Infof("================ICP Answer PAD====ERR============ : %+v", err.Error()) } //invite PACU join in //action.Hangup("PACU") //action.ChanSpy("PACU", exten, false, true) lfshook.NewLogger().Logger.Infof("================ICP Answer PAD================:%s ", exten) case 0x02: //hold 重新放回队列里面 NotifyPaiu(exten, "hold") err := action.RedirectInQueue(exten, "0300", "default", "1") if err != nil { lfshook.NewLogger().Info(err) } case 0x03: //hangup //NotifyPaiu(exten, "hangup") action.Hangup(exten) } } // 挂断所有报警器 func NotifyPaiu(Exten, Action string) { url := "" switch Action { case "answer": url = fmt.Sprintf("http://10.0.24.%s/api/sipphone?action=answer", Exten[2:]) case "hold": url = fmt.Sprintf("http://10.0.24.%s/api/sipphone?action=hold", Exten[2:]) case "hangup": url = fmt.Sprintf("http://10.0.24.%s/api/sipphone?action=hangup", Exten[2:]) } lfshook.NewLogger().Logger.Infof("======Notify PAIU Alarm====:%+v ", url) resp, err := http.Get(url) if err != nil { lfshook.NewLogger().Logger.Infof("======Notify PAIU Alarm====:%+v ", err) return } defer resp.Body.Close() /* body, err := io.ReadAll(resp.Body) if err != nil { // 读取数据错误 lfshook.NewLogger().Warn("ioutil ReadAll failed :", err.Error()) return } fmt.Printf("状态码: %d\n", resp.StatusCode) fmt.Printf("响应内容: %s\n", body) */ } // 挂断所有报警器 func AlarmResetAll() { var AlarmExts []model.Extension er := mysql.DBOrmInstance.Where("exttype = ?", "PAIU").Find(&AlarmExts) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } for _, ext := range AlarmExts { action.Hangup(ext.Extension) } } func RecordStorageConf(data []byte) { /*padRcd := data[0] padRcdStorage := data[1] paRcdStorage := data[2] cpaRcdStorage := data[3] padRcdDel := data[4] PaRcdDel := data[5] cpaRcdDel := data[6] //update pa type _, er := mysql.DBOrmInstance.In("exten", Pacus).Update(&model.Extension{PaType: "CHK"}) if er != nil { lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error()) } */ }