crrc/internal/app/stc/broadcast/stc-broadcast.go

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())
	}
	*/
}