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

package broadcast

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"pbx-api-gin/internal/app/ami/action"
	"pbx-api-gin/internal/app/ami/model"
	"pbx-api-gin/internal/app/stc/active"
	msgdata "pbx-api-gin/internal/app/stc/data"
	"pbx-api-gin/internal/app/stc/priority"
	alstatus "pbx-api-gin/internal/app/stc/sendstatus"
	"pbx-api-gin/pkg/lfshook"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/sirupsen/logrus"
	"gopkg.in/ini.v1"
)

func HandleStcCmd(ctx context.Context, conn net.Conn) {

	for {
		select {
		case <-ctx.Done():
			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 {
						conn.Close()
					}
					return
				}
				buf.Write(tmp[:n])
			}

			for {
				packet, err := msgdata.ExtractPacket(&buf)
				if err != nil {
					break
				}
				go processPacket(packet)
			}
		}
	}
}

// 处理单个数据包（原 switch 逻辑迁移过来）
func processPacket(packet []byte) {

	//check if Master role
	if !active.Master {
		lfshook.NewLogger().Logger.Infof("=========Not Master Role Ignore data=============")
		return
	}

	if len(packet) < 6 {
		fmt.Println("Invalid packet length")
		return
	}

	//for recv data log debug
	//if packet[5] != 0x03 && packet[5] != 0x0c && packet[5] != 0x01 {
	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[8]})
		return
	}

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

		//PAD-OCC异常情况处理
		if priority.OCCAnswer == 1 {
			//定时监测ICP queue 转到OCC queue， 避免异常情形下PAD进入ICP queue之后无法被接听
			resCaller, err := action.QueueStatus("0300", "") // check ICP queue, get entries
			if err != nil {
				lfshook.NewLogger().Infof("==ICP=QueueStatus==%+v", err)
				return
			}
			if resCaller.Calls != "0" {
				sort.Slice(resCaller.Entrys, func(i, j int) bool {
					return resCaller.Entrys[i].Position < resCaller.Entrys[j].Position
				})

				for _, caller := range resCaller.Entrys {
					priority.ICPAnswer = 0
					lfshook.NewLogger().Infof("====Redirect to 0301 entry:%s=Pos:%s==", caller.CallerIDNum, caller.Position)
					//order by pos
					action.RedirectInQueue(caller.CallerIDNum, "0301", "queues-occ", caller.CallerIDNum) // redirect All ICP-PAD redirect to OCC queue
					time.Sleep(time.Microsecond * 200)                                                   //200 ms delay
				}
			}

			//定时监测OCC  queue， 避免异常情况下PAD 在OCC queue里面不能自动转到OCC
			res, err := action.QueueStatus("0301", "") // check OCC queue ,if empty PAD end
			if err != nil {
				lfshook.NewLogger().Infof("==OCC=QueueStatus==%+v", err)
				return
			}
			if res.Calls != "0" { //OCC queue is not empty
				//	HangupAllLocalChan()
				lfshook.NewLogger().Infof("====Start OCC-PAD===next==%+v", res)
				if active.ActivedCab == "1" && action.ExtenStatus("1411") == "Idle" { //check active and OCC status
					time.Sleep(time.Second)
					PADChan := ""
					for _, chanEntry := range res.Entrys {
						lfshook.NewLogger().Infof("====PAD answered by OCC1 pos:%s===chan:%s=", chanEntry.Position, chanEntry.Channel)
						if chanEntry.Position == "1" {
							PADChan = chanEntry.Channel
							break
						}
					}

					if PADChan != "" {
						alstatus.AlarmStatus(strings.Split(strings.Split(res.Entrys[0].Channel, "/")[1], "-")[0], "connect")
						go action.RedirectInQueue(PADChan, "1411", "pad-page-occ-icp", "1") //PAD Page(OCC+ICPs)

						go action.Dial("0401", "0512", "pad-rule-pacus-occ", "ano1", "ano1", "1") // PACUs dial OCC1
					} else {
						lfshook.NewLogger().Infof("===OCC-QueueStatus==PADCchan NULL")
					}
					break
				} else if active.ActivedCab == "8" && action.ExtenStatus("1481") == "Idle" {
					time.Sleep(time.Second)
					PADChan := ""
					for _, chanEntry := range res.Entrys {
						lfshook.NewLogger().Infof("====PAD answered by OCC1 pos:%s===chan:%s=", chanEntry.Position, chanEntry.Channel)
						if chanEntry.Position == "1" {
							PADChan = chanEntry.Channel
							break
						}
					}

					if PADChan != "" {
						alstatus.AlarmStatus(strings.Split(strings.Split(res.Entrys[0].Channel, "/")[1], "-")[0], "connect")
						go action.RedirectInQueue(PADChan, "1481", "pad-page-occ-icp", "8") //PAD Page(OCC+ICPs)

						go action.Dial("0401", "0512", "pad-rule-pacus-occ", "ano8", "ano8", "8") // PACUs dial OCC1
					} else {
						lfshook.NewLogger().Infof("===OCC-QueueStatus==PADCchan NULL")
					}
					break
				}
			}
		}

	case 0x02: // STN
		if priority.CheckPriority("STN") {
			action.HangupRunningTask("STN") //STN interrupt other
			StationAnn(packet)
		} else {
			alstatus.PaStatus("", "STN", "refuse")
		}

	case 0x05: // SPC
		if priority.CheckPriority("SPC") {
			action.HangupRunningTask("SPC") //SPC interrupt other
			SpecialAnn(packet)
		} else {
			alstatus.PaStatus("", "SPC", "refuse")
		}

	case 0x06: // EMG
		if priority.CheckPriority("EMG") {
			action.HangupRunningTask("EMG") //EMG interrupt other
			EmgMsg(packet)
		} else {
			alstatus.PaStatus("", "EMG", "refuse")
		}
	case 0x07: // STOP
		AnnStop([4]byte{packet[8], packet[9], packet[10], packet[11]})

	case 0x08: // DCS
		if priority.CheckPriority("DCS") {
			action.HangupRunningTask("DCS") //DCS interrupt other
			DcsAnn(packet)
		} else {
			alstatus.PaStatus("", "DCS", "refuse")
		}

	case 0x09: // SELF CHECK
		if priority.CheckPriority("CHK") {
			action.HangupRunningTask("CHK") //CHK interrupt other
			SelfCheck(packet)
		} else {
			alstatus.PaStatus("", "CHK", "refuse")
		}

	case 0x0a: // Tone-test
		if priority.CheckPriority("VOL") {
			action.HangupRunningTask("VOL") //VOL interrupt other
			ToneTest(packet)
		} else {
			alstatus.PaStatus("", "VOL", "refuse")
		}

	case 0x0e: //TMS answer PAD
		if priority.CheckPriority("PAD-TMS") {
			action.HangupRunningTask("PAD-TMS") //PAD-TMS interrupt other

			AlarmHandleTMS(packet)
			active.QueueTimer.Stop()
		} else {
			alstatus.PaStatus("", "PAD-TMS", "refuse")
		}

	case 0x0b: // reset all PAD
		AlarmHoldResetAll(packet[8]) // reset all pad

	case 0x0c: // recored config
		RecordStorageConf(packet[8:]) // RCD setting

	case 0x0d: // ICP answer PAD
		if priority.CheckPriority("PAD-ICP") {

			action.HangupRunningTask("PAD-ICP") //PAD-ICP interrupt other
			active.QueueTimer.Stop()
			AlarmHandleICP(packet) //
		} else {
			alstatus.PaStatus("", "PAD-ICP", "refuse")
		}

	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)

	//set spc voice tag
	priority.SpecialVoice = specialVoice

	action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "STN")
	return nil
}

// 激活信号
func Active(data [1]byte) {

	//var info model.Sysinfo
	//active.Actived = true
	Signal := int(data[0])
	//lfshook.NewLogger().Logger.Infof("Active data : %x", Signal)

	switch Signal {
	case 0:

		//lfshook.NewLogger().Logger.Infof("=================Inactive==================")
		active.ActivedCab = ""
		action.InActiveHangup()

	case 1:
		active.ActivedCab = "1"
		//lfshook.NewLogger().Logger.Infof("=================active===MC1===============")
		if active.CabNum == "8" { // local cab is MC1
			action.InActiveHangup()
		}

	case 8:

		active.ActivedCab = "8"
		//lfshook.NewLogger().Logger.Infof("=================active===MC8===============")
		if active.CabNum == "1" { //Local cab is MC8
			action.InActiveHangup()
		}
	}
}

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

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

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

	if int(cycleCount) == 255 {
		action.PlaybackPacu(strconv.Quote(filename), 9999999, int(delay), "SPC")
	} else {
		action.PlaybackPacu(strconv.Quote(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)

	if int(cycleCount) == 255 {
		action.PlaybackPacu(strconv.Quote(filename), 9999999, int(delay), "EMG")
		priority.ResumeEmgPara = priority.BroadcastResumeParas{FileName: filename, Count: 99999999, Delay: int(delay), BroadcastType: "EMG"}
	} else {
		action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "EMG")
	}
}

// 停止指定类型广播
func AnnStop(data [4]byte) {

	//PaType := ""
	lfshook.NewLogger().Logger.Infof("=AnnStop Type  %x", data[0])
	switch data[0] {
	case 0x03:
		if priority.RunningType == "DCS" {
			action.HangupRunningTask("") //STOP DCS
		}
	case 0x04:
		if priority.RunningType == "EMG" {
			action.HangupRunningTask("AnnStop") //STOP EMG
		}
	case 0x07:
		if priority.RunningType == "SPC" {
			action.HangupRunningTask("AnnStop") //STOP SPC
		}
	case 0x08:
		if priority.RunningType == "STN" {
			action.HangupRunningTask("AnnStop") //STOP STN
		}
	case 0x09:
		if priority.RunningType == "CHK" {
			action.HangupRunningTask("AnnStop") //STOP CHK
		}
	case 0x0a:
		if priority.RunningType == "VOL" {
			action.HangupRunningTask("AnnStop") //STOP VOL
		}
	default:
		action.HangupRunningTask("AnnStop")
	}
}

// DCS 语音
func DcsAnn(data []byte) {
	delay := data[8]
	cycleCount := data[9]
	datalen := int(data[10])

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

	if int(cycleCount) == 255 {
		action.PlaybackPacu(strconv.Quote(filename), 9999999, int(delay), "DCS")
	} else {
		action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "DCS")
	}
}

// tone-test广播
func ToneTest(data []byte) {

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

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

	switch check {
	case 0x01: //start
		action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "VOL")

	case 0x02: //stop

		//alstatus.PaStatus("", "VOL", "end")
		action.HangupAllExcept("")
	}
}

// 自检广播
func SelfCheck(data []byte) {

	check := data[8]
	delay := data[9]
	//cycleCount := data[10]
	cycleCount := 0x32
	datalen := int(data[11])

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

	switch check {
	case 0x01: //start

		action.PlaybackPacu(strconv.Quote(filename), int(cycleCount), int(delay), "CHK")
	case 0x02: //stop

		//Pa status report
		priority.CleanPriorityTag()
		//alstatus.PaStatus("", "CHK", "end")
		action.HangupAllExcept("")
	}
}

// 全局变量：记录正在抑制的 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 AlarmHandleICP(data []byte) {
	handler := data[8]
	carr := data[12]
	pos := data[13]
	exten := fmt.Sprintf("24%c%c", carr, pos)

	key := suppressKey(exten, handler)

	//Drop other handler in 2 sec
	//PACUs---call---->ICP1
	//PAD---->Chanspy(WEq)-->ICP1;PAD--->Call---->ICP2
	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
		}

		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")
		priority.ICPAnswer = 1
		if priority.PADStart == 0 {
			alstatus.PaStatus(exten, "PAD", "start")
			priority.PADStart = 1
		}
		lfshook.NewLogger().Logger.Infof("================ICP Answer PAD================:%s ", exten)
		if active.ActivedCab == "1" {
			action.Dial("0402", "0511", "pad-rule-pacus", "ani1", exten, "1") // PACUs dial ICP1
			//goto ami event ConfbridgeJoin, ICP answer PAD
		} else if active.ActivedCab == "8" {
			action.Dial("0402", "0511", "pad-rule-pacus", "ani8", exten, "8") // PACUs dial ICP8
			//goto ami event ConfbridgeJoin, ICP answer PAD
		}

	case 0x02: //hold  重新放回队列里面
		NotifyPaiu(exten, "hold")
		err := action.RedirectInQueue(exten, "0300", "default", "1")
		if err != nil {
			lfshook.NewLogger().Info(err)
		}
		action.HangupAllLocalChan()

	case 0x03: //hangup
		//NotifyPaiu(exten, "hangup")
		action.Hangup(exten) //Pad
		action.HangupAllLocalChan()
		action.HangupICP()
		priority.CleanPriorityTag()
	}
}

// TMS操作乘客报警(根据激活信息判断转到1车还是8车================)
func AlarmHandleTMS(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)
		//PACU---call---->ICP1
		//PAD---->Chanspy(WEq)-->ICP1;PAD--->Call---->ICP2
		if priority.PADStart == 0 {
			alstatus.PaStatus(exten, "PAD", "start")
			priority.PADStart = 1
		}
		priority.ICPAnswer = 1
		lfshook.NewLogger().Logger.Infof("================TMS Answer PAD:%s===PACU:%s==========", exten, PacuNum)
		if action.ExtenStatus(PacuNum) == "Idle" {
			if active.ActivedCab == "1" {
				action.Dial("0403", PacuNum, "default", PacuNum, exten, "1") // PACU dial ICP1
				//goto ami event BridgeEnter, ICP8 whisper ICP1
			} else if active.ActivedCab == "8" {
				action.Dial("0403", PacuNum, "default", PacuNum, exten, "8") // PACU dial ICP8
				//goto ami event BridgeEnter, ICP1 whisper ICP8
			}
		} else {
			action.RedirectInQueue(exten, "0405", "default", exten) // PAD dial ICPs
		}

	case 0x02: //hold  重新放回队列里面
		NotifyPaiu(exten, "hold")
		err := action.RedirectInQueue(exten, "0300", "default", "1")
		if err != nil {
			lfshook.NewLogger().Info(err)
		}
		action.HangupAllLocalChan()

	case 0x03: //hangup
		//NotifyPaiu(exten, "hangup")
		action.Hangup(exten) //Pad
		action.HangupAllLocalChan()
		action.HangupICP()
		priority.CleanPriorityTag()
	}
}

// 挂断所有报警器
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 AlarmHoldResetAll(handler byte) {

	//all hold

	//hangup all actived PAD
	action.HangupAllPAD()

	//hangup running task
	if priority.RunningType == "PAD-ICP" || priority.RunningType == "PAD-OCC" || priority.RunningType == "PAD-TMS" {
		action.HangupRunningTask("AlarmHoldResetAll") //Reset PAD ALL
		priority.CleanPriorityTag()
	}
}

func RecordStorageConf(data []byte) {

	var info model.RcdConf

	info.PadRcdEnable = strconv.Itoa(int(data[0]))
	info.PadRcdStorageDays = strconv.Itoa(int(data[1]))
	info.PaRcdStorageDays = strconv.Itoa(int(data[2]))
	info.CpaRcdStorageDays = strconv.Itoa(int(data[3]))

	info.PadRcdDelDays = strconv.Itoa(int(data[4]))
	info.PaRcdDelDays = strconv.Itoa(int(data[5]))
	info.CpaRcdDelDays = strconv.Itoa(int(data[6]))
	//info.OpaRcdStorageDays = int(data[7])
	//info.OpaRcdDelDays = int(data[8])

	//lfshook.NewLogger().Infof("=============Set record Conf : %+v", info)

	filePath := "/etc/asterisk/recording.conf"
	_, err := os.Stat(filePath)
	if err != nil {
		logrus.Error(err)
		return
	}
	iniFile, err := ini.Load(filePath)
	if err != nil {
		logrus.Error(err)
		return
	}

	iniFile.Section("general").Key("PADRCD").SetValue(info.PadRcdEnable)
	iniFile.Section("general").Key("PADRCDDAYS").SetValue(info.PadRcdStorageDays)
	iniFile.Section("general").Key("PARCDDAYS").SetValue(info.PaRcdStorageDays)
	iniFile.Section("general").Key("CPARCDDAYS").SetValue(info.CpaRcdStorageDays)
	iniFile.Section("general").Key("PADRCDDELDAYS").SetValue(info.PadRcdDelDays)
	iniFile.Section("general").Key("PARCDDELDAYS").SetValue(info.PaRcdDelDays)
	iniFile.Section("general").Key("CPARCDDELDAYS").SetValue(info.CpaRcdDelDays)
	iniFile.SaveTo(filePath)

}