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

package broadcast

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"pbx-api-gin/api/panel/asterisk"
	"pbx-api-gin/internal/app/ami/action"
	"pbx-api-gin/internal/app/ami/model"
	"pbx-api-gin/internal/app/mysql"
	"pbx-api-gin/internal/app/stc/active"
	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(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
				}
				if active.Actived { // check actived or not
					// 将新读取的数据追加到缓冲区
					buf.Write(tmp[:n])
				} else {
					break
				}
			}
			// 尝试从缓冲区中提取完整数据包
			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)
	//check if the cmd type is avtive
	if packet[5] == 0x03 { // ACTIVE
		Active([1]byte{packet[7]})
	}

	switch packet[5] {
	case 0x01: // heartbeat
		// handle heartbeat
	case 0x02: // STN
		lfshook.NewLogger().Infof("=====STN=========")
		StationAnn(packet)
	case 0x03: // ACTIVE

		Active([1]byte{packet[8]})
	case 0x05: // SPC
		lfshook.NewLogger().Infof("=====SPC=========")
		SpecialAnn(packet)
	case 0x06: // EMG
		lfshook.NewLogger().Infof("=====EMG=========")
		EmgMsg(packet)
	case 0x07: // STOP
		AnnStop([4]byte{packet[8], packet[9], packet[10], packet[11]})
	case 0x08: // DCS
		lfshook.NewLogger().Infof("=====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)
	lfshook.NewLogger().Logger.Infof("=============Get filename  : %v", filename)

	//update stn 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) {

	var info model.Sysinfo

	Num := int(data[0])
	lfshook.NewLogger().Logger.Infof("Active data : %x", Num)

	switch Num { // 设置全局的激活信号，并通过协议（待定）通知终端注册到对应的激活主机上
	case 0:
		lfshook.NewLogger().Logger.Infof("=================Inactive==================")
		info.Name = "cab_active"
		info.Value = "0"

		_, er := mysql.DBOrmInstance.Where("name = ?", "cab_active").Update(&info)
		if er != nil {
			lfshook.NewLogger().Logger.Infof("update sysinfo err : %+v", er.Error())
			return
		}
	case 1:
		lfshook.NewLogger().Logger.Infof("=================active===MC1===============")
		if active.CabNum == "1" { // local cab is MC1
			info.Name = "cab_active"
			info.Value = "1"
			active.Actived = true
		} else {
			info.Name = "cab_active"
			info.Value = "0"
			active.Actived = false
		}
		_, er := mysql.DBOrmInstance.Where("name = ?", "cab_active").Update(&info)
		if er != nil {
			lfshook.NewLogger().Logger.Infof("update sysinfo err : %+v", er.Error())
			return
		}
	case 8:
		lfshook.NewLogger().Logger.Infof("=================active===MC8===============")
		if active.CabNum == "8" { //Local cab is MC8
			info.Name = "cab_active"
			info.Value = "1"
			active.Actived = true
		} else {
			info.Name = "cab_active"
			info.Value = "0"
			active.Actived = false
		}

		_, er := mysql.DBOrmInstance.Where("name = ?", "cab_active").Update(&info)
		if er != nil {
			lfshook.NewLogger().Logger.Infof("update sysinfo err : %+v", er.Error())
			return
		}
	}
}

// SPC ,特殊服务消息广播
func SpecialAnn(data []byte) {

	delay := data[8]
	cycleCount := data[9]
	datalen := int(data[10])

	filename := msgdata.SubstrByRune(string(data[11:]), 0, datalen-4)

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

	lfshook.NewLogger().Infof("======count:%x", cycleCount)
	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-4)

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

	lfshook.NewLogger().Infof("======count:%x", cycleCount)
	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) {
	RunningType := ""

	switch data[0] {
	case 0x03:
		RunningType = "DCS"
	case 0x04:
		RunningType = "EMG"
	case 0x07:
		RunningType = "SPC"
	case 0x08:
		RunningType = "STN"
	case 0x09:
		RunningType = "CHK"
	}
	for _, ext := range Pacus {
		action.Hangup(ext)
	}

	//update pa type
	_, er := mysql.DBOrmInstance.Cols("patype").Where("pa_type = ?", RunningType).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-4)

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

	//lfshook.NewLogger().Infof("======count:%x", cycleCount)
	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:
		for _, ext := range Pacus {
			asterisk.Hangup(ext)
		}
	}
}

// 全局变量：记录正在抑制的 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)
	PacuNum := fmt.Sprintf("21%c%c", carr, pos)

	key := suppressKey(exten, handler)

	//Drop other handler in 2 sec
	// 只对 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 4 seconds", exten)
			return // 已存在，说明在2秒窗口期内，直接丢弃
		}

		// 设置4秒后删除该 key，允许下次通过
		time.AfterFunc(4*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")

		//get pacu status
		var infoExt model.Extension
		_, er := mysql.DBOrmInstance.Where("exten = ?", PacuNum).Get(&infoExt)
		if er != nil {
			lfshook.NewLogger().Logger.Infof("update special voice to exten err : %+v", er.Error())
		}

		lfshook.NewLogger().Logger.Infof("================ICP Answer PAD================:%s ", exten)
		if infoExt.Status == "Idle" {
			if active.CabNum == "1" && active.Actived {
				action.Dial("0402", PacuNum, "default", PacuNum, exten, "1") // PACU dial ICP
			} else if active.CabNum == "8" && active.Actived {
				action.Dial("0402", PacuNum, "default", PacuNum, exten, "8") // PACU dial ICP
			}
		} else {
			if active.CabNum == "1" && active.Actived {
				action.RedirectInQueue(exten, "0402", "default", "1") // PACU dial ICP
			} else if active.CabNum == "8" && active.Actived {
				action.RedirectInQueue(exten, "0402", "default", "8") // PACU dial ICP
			}
		}

	case 0x02: //hold  重新放回队列里面
		NotifyPaiu(exten, "hold")
		err := action.RedirectInQueue(exten, "0300", "default", "1")
		if err != nil {
			lfshook.NewLogger().Info(err)
		}
		if active.CabNum == "1" && active.Actived {
			action.Hangup("2311") //1 车接听
		} else if active.CabNum == "8" && active.Actived {
			action.Hangup("2381") //8 车接听
		}
	case 0x03: //hangup
		//NotifyPaiu(exten, "hangup")
		action.Hangup(exten)
		if active.CabNum == "1" && active.Actived {
			action.Hangup("2311") //1 车接听
		} else if active.CabNum == "8" && active.Actived {
			action.Hangup("2381") //8 车接听
		}
	}
}

// 挂断所有报警器
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("dev_type = ?", "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)
	}

	action.Hangup("2311")
}

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