acfunlive-remake/cycle.go

// 循环相关
package main

import (
	"context"
	"time"
)

// 处理管道信号
func (s *streamer) handleMsg(msg controlMsg) {
	switch msg.c {
	case startCycle:
		lPrintln("重启监听" + s.longID() + "的直播状态")
		mainCh <- msg
	case stopCycle:
		lPrintln("退出" + s.longID() + "的循环")
		sInfoMap.Lock()
		if _, ok := sInfoMap.info[s.UID]; ok {
			delete(sInfoMap.info, s.UID)
		} else {
			lPrintErr("sInfoMap没有%s的key", s.longID())
		}
		sInfoMap.Unlock()
	case quit:
	default:
		lPrintErrf("未知的 controlMsg：%+v", msg)
	}
}

// 掉线后又上线的判定窗口。AcFun 的接口和直播源偶尔会抖动一下，几秒到十几秒内
// 「下播又上播」基本都是网络波动而不是真的换场次，窗口内恢复就当成同一场继续录，
// 这样最后是一个成片，而不是一场直播被切成两份文件
var liveResumeWindow = 20 * time.Second

// 判断主播是否还在直播（走 wap 页面）。抽成变量是为了让测试能替换掉这次网络请求
var liveOnPage = func(s *streamer) bool { return s.isLiveOnByPage() }

// 断流后等待直播恢复：窗口内恢复返回 true，超时或用户主动停止返回 false
func (s *streamer) waitResume(ctx context.Context) bool {
	deadline := time.Now().Add(liveResumeWindow)
	for {
		if ctx.Err() != nil {
			return false
		}
		// 页面说还在直播就继续；liveID 变了也接受，短暂下播又上播大多只是网络波动
		if liveOnPage(s) {
			return true
		}
		if !time.Now().Before(deadline) {
			return false
		}
		time.Sleep(time.Second)
	}
}

// 标记这条记录正在等直播恢复，这段时间里不要给同一个主播另起一场
func markResumeWait(liveID string) {
	until := time.Now().Add(liveResumeWindow).UnixNano()
	updateLiveInfo(liveID, func(i *liveInfo) { i.resumeUntil = until })
}

// 取消等待标记（恢复录制、或者本场真的结束了）
func clearResumeWait(liveID string) {
	updateLiveInfo(liveID, func(i *liveInfo) { i.resumeUntil = 0 })
}

// 主播是否还有没真正结束的录制会话：本场还有部分在录，或者断流后录制循环还在重连窗口里等。
// cycle() 用它判断「下播又上播」是不是同一场的延续，避免同一场被重复录成两份文件
func sessionAliveForUID(uid int) bool {
	now := time.Now().UnixNano()
	lInfoMap.RLock()
	defer lInfoMap.RUnlock()
	for _, info := range lInfoMap.info {
		if info.uid != uid {
			continue
		}
		if info.resumeUntil > now {
			return true
		}
		// 本场开始录过的部分还有没结束的，说明本场还在进行；
		// 录制循环退出时一定会标记 Done，所以这里不会把已经死掉的录制当成还活着
		if (info.recordStarted && !info.recordDone) || (info.danmuStarted && !info.danmuDone) {
			return true
		}
	}
	return false
}

// 循环获取指定主播的直播状态，自动录制直播弹幕
func (s streamer) cycle(liveID string) {
	defer func() {
		if err := recover(); err != nil {
			lPrintErr("Recovering from panic in cycle(), the error is:", err)
			lPrintErr(s.longID() + "的循环处理发生错误，尝试重启循环")

			restart := controlMsg{s: s, c: startCycle}
			mainCh <- restart
		}
	}()

	ch := make(chan controlMsg, 20)
	var modify bool
	sInfoMap.Lock()
	if m, ok := sInfoMap.info[s.UID]; ok {
		m.ch = ch
		modify = m.modify
		m.modify = false
	} else {
		sInfoMap.info[s.UID] = &streamerInfo{ch: ch}
	}
	sInfoMap.Unlock()

	lPrintln("开始监听" + s.longID() + "的直播状态")

	var isLive bool
	var lastTitle string
	var offlineAt time.Time // 首次读不到直播状态的时刻，用来判断是网络波动还是真下播
	var resumeWait bool     // 已经提示过「本场还在继续」，避免每秒重复打日志
	for {
		select {
		case msg := <-ch:
			msg.liveID = liveID
			s.handleMsg(msg)
			return
		default:
			if s.isLiveOn() {
				isLive = true
				newLiveID := s.getLiveID()
				if newLiveID == "" {
					lPrintErrf("无法获取%s的liveID", s.longID())
					time.Sleep(time.Second)
					continue
				}

				if !offlineAt.IsZero() {
					lPrintf("%s 的直播状态恢复（掉线约 %.0f 秒），判定为网络波动，继续本场录制",
						s.longID(), time.Since(offlineAt).Seconds())
					offlineAt = time.Time{}
				}

				if liveID != newLiveID || modify {
					// 只有真正换了场次才算新开播：改设置导致的重新开始不重复推送开播通知
					newLive := liveID != newLiveID
					// 同一主播的上一场还没真正结束（断流后录制循环还在重连窗口里等），
					// 说明这次「上播」只是网络波动：先不新开一场，避免重复录制和重复通知。
					// 这里刻意不改 liveID，等本场真的结束后下一轮再按新 liveID 开一场，
					// 否则本场要是 20 秒后才结束，这一场新直播就永远开始不了了
					if newLive && !modify && sessionAliveForUID(s.UID) {
						if !resumeWait {
							resumeWait = true
							lPrintln(s.longID() + " 本场录制还在继续（判定为网络波动），不重新开始录制")
						}
						time.Sleep(time.Second)
						continue
					}
					resumeWait = false
					liveID = newLiveID
					modify = false
					title := s.getTitle()
					liveURL := s.getURL()
					lastTitle = title
					lPrintln(s.longID() + "正在直播：" + title)
					lPrintln(s.Name + "的直播观看地址：" + liveURL)

					// 刷新主播数据，使用最新的开关设置
					if cur, ok := getStreamer(s.UID); ok {
						s = cur
					}
					if newLive && notifiedLiveOnce(liveID) {
						sendNotify(notifyEventLiveOn, s.UID, s.Name, title, liveURL, "",
							s.Name+" 开播了\n"+title+"\n"+liveURL)
					}
					filename := getTime() + " " + s.Name + " " + title
					if s.Record {
						ss := s
						safeGo(ss.longID()+"的视频录制", func() { ss.initRecord(mainCtx, liveID, filename) })
						// 没有 ffmpeg 时录不了视频，避免发出假的「开始录制」通知
						if newLive && getFFmpeg() != "" {
							sendNotify(notifyEventRecordStart, s.UID, s.Name, title, liveURL, "",
								s.Name+" 开始录制直播视频\n"+title+"\n"+liveURL)
						}
					}
					if s.Danmu {
						ss := s
						safeGo(ss.longID()+"的弹幕录制", func() { ss.initDanmu(mainCtx, liveID, filename) })
					}
				}
			} else {
				// 应付 AcFun API 可能出现的 bug：主播没下播但 API 显示下播
				if isLive && !s.isLiveOnByPage() {
					if offlineAt.IsZero() {
						// 先不下结论：网络波动也会这样，等窗口过后再确认，免得发一条假的下播通知
						offlineAt = time.Now()
						lPrintWarn(s.longID() + " 读不到直播状态，等待 " + itoa(int(liveResumeWindow.Seconds())) +
							" 秒确认（网络波动会继续本场录制）")
					} else if time.Since(offlineAt) >= liveResumeWindow {
						isLive = false
						offlineAt = time.Time{}
						resumeWait = false
						lPrintln(s.longID() + "已经下播")
						sendNotify(notifyEventLiveOff, s.UID, s.Name, lastTitle, s.getURL(), "",
							s.Name+" 已经下播\n最后一场："+lastTitle)
					}
				}
			}
		}

		time.Sleep(time.Second)
	}
}

// 录制全部结束、且没有再产生合并任务后，liveInfo 的保留时长
const liveInfoKeepDuration = 30 * time.Minute

// 循环检测删除 lInfoMap.info 里没有录制弹幕的 key
func cycleDelKey(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			sweepLiveInfo()

			// 每分钟循环一次
			time.Sleep(time.Minute)
		}
	}
}

// 清理已经彻底结束的直播记录，返回删除的条数
func sweepLiveInfo() int {
	now := time.Now()
	nowNs := now.UnixNano()
	removed := 0
	lInfoMap.Lock()
	defer lInfoMap.Unlock()
	for liveID, info := range lInfoMap.info {
		// 正在录制或正在合并的记录不能删
		if info.isDanmu || info.isRecording || info.mergeStarted {
			continue
		}
		// 断流后录制循环还在重连窗口里等着，这条记录后面还要接着用
		if info.resumeUntil > nowNs {
			continue
		}
		// 刚结束录制的记录要留住：弹幕和录播的结束时间可能相差较久（录播分段之间还有间隔），
		// 删掉 key 会让后结束的那部分在 finishPart 里找不到记录，自动合并被静默跳过
		if info.danmuDone || info.recordDone {
			if info.doneAt == 0 || now.Sub(time.Unix(0, info.doneAt)) < liveInfoKeepDuration {
				continue
			}
		}
		delete(lInfoMap.info, liveID)
		removed++
	}
	return removed
}

// 循环获取 AcFun 直播间数据
func cycleFetch(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			if ok := fetchAllRooms(); ok {
				if len(liveRooms.newRooms) == 0 {
					lPrintWarn("没有人在直播")
				}

				liveRooms.Lock()
				for uid, room := range liveRooms.rooms {
					delete(liveRooms.rooms, uid)
					liveRoomPool.Put(room)
				}
				liveRooms.rooms = liveRooms.newRooms
				liveRooms.Unlock()
			}

			// 每 10 秒循环一次
			time.Sleep(10 * time.Second)
		}
	}
}