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

package broadcast

import (
	"bytes"
	"fmt"
	"io"
	"net"
	"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"
	"sync"
	"time"
)

var Pacus = []string{"2111", "2121", "2131", "2141", "2151", "2161", "2171", "2181"}

func HandleStcCmd(conn net.Conn) {
	var buf bytes.Buffer // 用于累积未处理完的数据流
	tmp := make([]byte, 1024)

	for {
		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])
	}
}

/*
func HandleStcCmd(conn net.Conn) {

	for {
		buffer := make([]byte, 1024)
		n, err := conn.Read(buffer)
		if err != nil {
			fmt.Println("Error reading from server:", err)
			return
		}
		if buffer[5] != 0x01 {
			lfshook.NewLogger().Logger.Infof("Get data from STC ===============:%x", buffer[:n])
		}

		switch buffer[5] {
		case 0x01: //heartbeat

		case 0x02: //STN
			StationAnn(buffer)

		case 0x03: //ACTIVE
			Active([1]byte{buffer[8]})

		case 0x05: //SPC
			SpecialAnn(buffer)

		case 0x06: //EMG
			EmgMsg(buffer)

		case 0x07: //STOP
			AnnStop([4]byte{buffer[8], buffer[9], buffer[10], buffer[11]})

		case 0x08: //DCS
			DcsAnn(buffer)

		case 0x09: //SELF CHECK
			SelfCheck(buffer)

		case 0x0a: //
			AlarmHandle(buffer)

		case 0x0b: //
			AlarmResetAll()

		case 0x0c: //
			RecordStorageConf(buffer[8:])
		}
	}
}
*/
// 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)

	//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)

		err := action.RedirectInQueue(exten, "0402", "ani-rule", "1") // 1车ICP接听
		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接听
		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  重新放回队列里面
		err := action.RedirectInQueue(exten, "0300", "default", "1")
		if err != nil {
			lfshook.NewLogger().Info(err)
		}

	case 0x03: //hangup
		action.Hangup(exten)
	}
}

// 挂断所有报警器
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())
	}
	*/
}