acfunlive-remake/danmu.go

// 弹幕录制相关
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

	"github.com/orzogc/acfundanmu"
)

// 不同的视频分辨率对应的弹幕字幕设置
// 字号一律取「短边 × 60/1080」（横屏取高、竖屏取宽），和合并端的自动字号一致
var subConfigs = map[int]acfundanmu.SubConfig{
	0:    {PlayResX: 720, PlayResY: 1280, FontSize: 40}, // 这是手机直播，有一些是 540X960
	540:  {PlayResX: 960, PlayResY: 540, FontSize: 30},
	720:  {PlayResX: 1280, PlayResY: 720, FontSize: 40},
	1080: {PlayResX: 1920, PlayResY: 1080, FontSize: 60},
}

// 获取弹幕字幕设置，根据直播源的码率选择
func (s *streamer) getSubConfig() (cfg acfundanmu.SubConfig, e error) {
	defer func() {
		if err := recover(); err != nil {
			e = fmt.Errorf("getSubConfig() error: %v", err)
		}
	}()

	var ac *acfundanmu.AcFunLive
	var err error
	err = runThrice(func() error {
		ac, err = acfundanmu.NewAcFunLive(acfundanmu.SetLiverUID(int64(s.UID)))
		return err
	})
	checkErr(err)
	sInfo := ac.GetStreamInfo()

	// 选择码率最高的直播源
	index := len(sInfo.StreamList) - 1
	bitrate := sInfo.StreamList[index].Bitrate
	switch {
	case bitrate >= 4000:
		cfg = subConfigs[1080]
	case len(sInfo.StreamList) >= 2 && bitrate >= 2000:
		cfg = subConfigs[720]
	case bitrate == 0:
		cfg = subConfigs[0]
	default:
		cfg = subConfigs[540]
	}

	return cfg, nil
}

// ==================== 弹幕数据双路输出 ====================

// 弹幕持续时间（纳秒），与 acfundanmu 保持一致
const danmuShowDur = int64(10 * time.Second)

// 弹道数量上限，与 acfundanmu 的 queueLen 保持一致
const laneNum = 100

// ass 文件的 Script Info，与 acfundanmu 保持一致
const scriptInfoTmpl = `[Script Info]
; LiveID: %s
; StreamName: %s
Title: %s
ScriptType: v4.00+
Collisions: Normal
PlayResX: %d
PlayResY: %d

`

// ass 文件的 V4+ Styles，与 acfundanmu 保持一致
const stylesTmpl = `[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Danmu,Microsoft YaHei,%d,&H00FFFFFF,&H00FFFFFF,&H00000000,&H00000000,0,0,0,0,100,100,0,0,1,1,0,2,20,20,2,0

`

// ass 文件的 Events
const eventsTmpl = `[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
`

// 原格式弹幕行，与 acfundanmu 的 dialogue 保持一致
const dialogueTmpl = "Dialogue: 0,%s,%s,Danmu,%s(%d),20,20,2,,{\\move(%d,%d,%d,%d)}%s\n"

// 昵称头像版弹幕行，\an7 让锚点固定在文本左上角，前导全角空格是头像占位
const dialogueNickTmpl = "Dialogue: 0,%s,%s,Danmu,%s(%d),20,20,2,,{\\an7\\move(%d,%d,%d,%d)}%s\n"

// jsonl 首行的元数据
type danmuMeta struct {
	PlayResX  int   `json:"playResX"`
	PlayResY  int   `json:"playResY"`
	FontSize  int   `json:"fontSize"`
	StartTime int64 `json:"startTime"` // 录制开始的时间，纳秒 Unix
}

// jsonl 的弹幕记录
type danmuRecord struct {
	Time    float64 `json:"t"` // 相对录制开始的秒数
	UID     int64   `json:"uid"`
	Name    string  `json:"name"`
	Avatar  string  `json:"avatar"` // 头像 URL，弹幕协议自带
	Content string  `json:"content"`
	Lane    int     `json:"lane"`  // 在昵称头像版 ASS 里的弹道行
	Width   int     `json:"width"` // 昵称头像版文本的估算像素宽度
}

// 弹道的占用状态（纳秒）
type laneState struct {
	appear    int64
	emerge    int64
	disappear int64
}

// ass 文件里弹幕出现或消失的时间格式，与 acfundanmu 保持一致
type assTime int64

func (t assTime) String() string {
	if t < 0 {
		return "0:00:00.00"
	}
	tt := time.Unix(0, int64(t)).UTC()
	return fmt.Sprintf("%d:%02d:%02d.%02d", tt.Hour(), tt.Minute(), tt.Second(), tt.Nanosecond()/1e7)
}

// 估算文本的渲染宽度，CJK 及全角字符按全宽、其余按半宽计算
func textWidth(text string, fontSize int) int {
	w := 0
	for _, r := range text {
		if r >= 0x2E80 {
			w += fontSize
		} else {
			w += fontSize / 2
		}
	}
	return w
}

// 在弹道数组里寻找空闲弹道，返回弹道行号，没有空闲返回 -1，公式与 acfundanmu 保持一致
func placeLane(last []laneState, dt laneState, leftTime int64) int {
	for i, t := range last {
		if dt.appear > t.emerge && leftTime > t.disappear {
			last[i] = dt
			return i
		}
	}
	return -1
}

// 弹幕数据双路输出：原格式 ASS + 昵称头像版 ASS + jsonl
type danmuWriter struct {
	cfg   acfundanmu.SubConfig
	ass   *os.File // 原格式
	nick  *os.File // 昵称头像版
	jsonl *os.File
	lastO []laneState // 原格式的弹道状态
	lastN []laneState // 昵称头像版的弹道状态
	err   error       // 写盘出过的错。只记头一个，见 danmuWriter.fail
}

// 记一次写盘错误。只保留第一个：磁盘满了之后每写一条都会失败，
// 全记下来只会把日志刷爆，而第一个就足以说明原因
func (w *danmuWriter) fail(what string, err error) {
	if w.err == nil {
		w.err = fmt.Errorf("%s失败：%v", what, err)
	}
}

// 本场弹幕有没有写坏。写坏的文件不能当正常的用 —— 否则合并出来的成片
// 会缺一大截弹幕，而日志里什么都看不出来
func (w *danmuWriter) failed() bool {
	return w.err != nil
}

// 返回昵称头像版 ASS 和 jsonl 的文件路径
func nickJsonlFile(assFile string) (string, string) {
	base := strings.TrimSuffix(assFile, ".ass")
	return base + ".nick.ass", base + ".jsonl"
}

// 创建弹幕写入器，覆盖写文件头，重启续录时沿用同一个写入器即可
func newDanmuWriter(liveID, streamName string, cfg acfundanmu.SubConfig, assFile string) *danmuWriter {
	nickFile, jsonlFile := nickJsonlFile(assFile)
	w := &danmuWriter{
		cfg:   cfg,
		lastO: make([]laneState, laneNum),
		lastN: make([]laneState, laneNum),
	}
	var err error
	w.ass, err = os.OpenFile(assFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	checkErr(err)
	w.nick, err = os.OpenFile(nickFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	checkErr(err)
	w.jsonl, err = os.OpenFile(jsonlFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	checkErr(err)

	if _, err := fmt.Fprintf(w.ass, scriptInfoTmpl, liveID, streamName, cfg.Title, cfg.PlayResX, cfg.PlayResY); err != nil {
		w.fail("写原格式 ASS 文件头", err)
	}
	if _, err := fmt.Fprintf(w.ass, stylesTmpl, cfg.FontSize); err != nil {
		w.fail("写原格式 ASS 文件头", err)
	}
	if _, err := fmt.Fprint(w.ass, eventsTmpl); err != nil {
		w.fail("写原格式 ASS 文件头", err)
	}
	if _, err := fmt.Fprintf(w.nick, scriptInfoTmpl, liveID, streamName, cfg.Title+" 昵称头像版", cfg.PlayResX, cfg.PlayResY); err != nil {
		w.fail("写昵称头像版 ASS 文件头", err)
	}
	if _, err := fmt.Fprintf(w.nick, stylesTmpl, cfg.FontSize); err != nil {
		w.fail("写昵称头像版 ASS 文件头", err)
	}
	if _, err := fmt.Fprint(w.nick, eventsTmpl); err != nil {
		w.fail("写昵称头像版 ASS 文件头", err)
	}

	meta, err := json.Marshal(danmuMeta{PlayResX: cfg.PlayResX, PlayResY: cfg.PlayResY, FontSize: cfg.FontSize, StartTime: cfg.StartTime})
	checkErr(err)
	if _, err := w.jsonl.Write(meta); err != nil {
		w.fail("写弹幕 jsonl 文件头", err)
	}
	if _, err := w.jsonl.Write([]byte("\n")); err != nil {
		w.fail("写弹幕 jsonl 文件头", err)
	}
	return w
}

// 写入一条弹幕
func (w *danmuWriter) write(c *acfundanmu.Comment) {
	s := w.cfg
	sendTime := c.SendTime * 1e6
	rel := sendTime - s.StartTime

	// 原格式 ASS，碰撞避让和坐标公式与 acfundanmu 完全一致
	length := utf8.RuneCountInString(c.Content) * s.FontSize
	dtO := laneState{
		appear:    rel,
		emerge:    rel + (int64(length)*danmuShowDur)/int64(s.PlayResX+length),
		disappear: rel + danmuShowDur,
	}
	leftO := rel + (int64(s.PlayResX)*danmuShowDur)/int64(s.PlayResX+length)
	if i := placeLane(w.lastO, dtO, leftO); i >= 0 {
		if _, err := fmt.Fprintf(w.ass, dialogueTmpl,
			assTime(dtO.appear), assTime(dtO.disappear),
			strings.ReplaceAll(c.Nickname, ",", " "), c.UserID,
			s.PlayResX+length/2, s.FontSize*(i+1), -length/2, s.FontSize*(i+1), c.Content); err != nil {
			w.fail("写原格式 ASS", err)
		}
	}

	// 昵称头像版 ASS，\an7 锚点在文本左上角，头像由 ffmpeg 叠在文字左侧，不依赖字体渲染宽度
	text := c.Nickname + "(" + strconv.FormatInt(c.UserID, 10) + "):" + c.Content
	width := textWidth(text, s.FontSize)
	avW, avGap := avatarSize(s.FontSize), avatarGap(s.FontSize)
	total := avW + avGap + width // 头像、间隔和文字的总宽度
	dtN := laneState{
		appear:    rel,
		emerge:    rel + (int64(total)*danmuShowDur)/int64(s.PlayResX+total),
		disappear: rel + danmuShowDur,
	}
	leftN := rel + (int64(s.PlayResX)*danmuShowDur)/int64(s.PlayResX+total)
	if i := placeLane(w.lastN, dtN, leftN); i >= 0 {
		if _, err := fmt.Fprintf(w.nick, dialogueNickTmpl,
			assTime(dtN.appear), assTime(dtN.disappear),
			strings.ReplaceAll(c.Nickname, ",", " "), c.UserID,
			s.PlayResX+avW+avGap, s.FontSize*i, -width, s.FontSize*i, text); err != nil {
			w.fail("写昵称头像版 ASS", err)
		}
		rec := danmuRecord{
			Time:    float64(rel) / 1e9,
			UID:     c.UserID,
			Name:    c.Nickname,
			Avatar:  c.Avatar,
			Content: c.Content,
			Lane:    i,
			Width:   width,
		}
		b, err := json.Marshal(rec)
		checkErr(err)
		if _, err := w.jsonl.Write(b); err != nil {
			w.fail("写弹幕 jsonl", err)
		}
		if _, err := w.jsonl.Write([]byte("\n")); err != nil {
			w.fail("写弹幕 jsonl", err)
		}
	}
}

// 消费弹幕队列直到结束，返回说明队列已结束（出错或主播下播）
func (w *danmuWriter) consume(ctx context.Context, ac *acfundanmu.AcFunLive) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			danmu := ac.GetDanmu()
			if danmu == nil {
				return
			}
			for _, d := range danmu {
				if c, ok := d.(*acfundanmu.Comment); ok {
					w.write(c)
				}
			}
		}
	}
}

// 关闭所有文件，并把整个录制过程中出过的写盘错误一并交出来。
// Close 自己的错误也要记：os.File 在部分系统上（NFS、某些网络盘）是到 Close
// 才报出「写不进去」的，只盯每次 Write 的返回值会漏掉这种情况
func (w *danmuWriter) close() error {
	if err := w.ass.Close(); err != nil {
		w.fail("关闭原格式 ASS", err)
	}
	if err := w.nick.Close(); err != nil {
		w.fail("关闭昵称头像版 ASS", err)
	}
	if err := w.jsonl.Close(); err != nil {
		w.fail("关闭弹幕 jsonl", err)
	}
	return w.err
}

// ==================== 录制流程 ====================

// 录制直播弹幕
func (s streamer) getDanmu(ctx context.Context, info liveInfo) {
	defer func() {
		if err := recover(); err != nil {
			lPrintErr("Recovering from panic in getDanmu(), the error is:", err)
			lPrintErrf("录制%s的直播弹幕发生错误，如要重启录制，请运行 startdanmu %d", s.longID(), s.UID)
		}
	}()

	lPrintln("开始录制" + s.longID() + "的直播弹幕")
	lPrintln("本次录制的 ass 文件保存在" + info.assFile)
	lPrintf("如果想提前结束录制%s的直播弹幕，运行 stopdanmu %d", s.longID(), s.UID)

	defer s.moveDanmuFiles(info.assFile)

	ac, err := acfundanmu.NewAcFunLive(acfundanmu.SetLiverUID(int64(s.UID)))
	checkErr(err)
	_ = ac.StartDanmu(ctx, false)
	streamInfo := ac.GetStreamInfo()
	w := newDanmuWriter(streamInfo.LiveID, streamInfo.StreamName, info.cfg, info.assFile)
	// 写盘出错要到关文件时才收得全（有的系统到 Close 才报「写不进去」）。
	// 必须说出来：否则本场照常自动合并，出来的成片缺一大截弹幕，日志里却一切正常
	defer func() {
		if err := w.close(); err != nil {
			lPrintErrf("录制%s的直播弹幕时写盘出错，弹幕文件不完整：%v", s.longID(), err)
			lPrintErrf("本场不做自动合并，弹幕文件保留在 %s，请检查磁盘空间后手动合并", info.assFile)
			updateLiveInfo(info.LiveID, func(i *liveInfo) { i.danmuWriteFailed = true })
		}
	}()
	w.consume(ctx, ac)

	// 断流后先观察一个窗口：网络波动或主播短暂掉线（20 秒内）恢复时沿用同一个写入器接着写，
	// 弹幕时间轴和视频分段保持一致，最终和视频一起合成同一个成片
	fails := 0
	for {
		markResumeWait(info.LiveID)
		resumed := s.waitResume(ctx)
		clearResumeWait(info.LiveID)
		if !resumed {
			break
		}
		if fails >= maxResumeRetry {
			lPrintWarnf("重连%s的直播弹幕连续失败 %d 次，结束本场弹幕录制", s.longID(), fails)
			break
		}
		lPrintWarn("因意外结束录制" + s.longID() + "的直播弹幕，尝试重启录制")
		begin := time.Now()
		ac, err := acfundanmu.NewAcFunLive(acfundanmu.SetLiverUID(int64(s.UID)))
		if err != nil {
			lPrintErrf("重连%s的直播弹幕失败：%v", s.longID(), err)
			fails++
			time.Sleep(time.Second)
			continue
		}
		_ = ac.StartDanmu(ctx, false)
		w.consume(ctx, ac)
		// 连上几秒就断开说明这次重连没成功，计数避免一直高频重连
		if time.Since(begin) < 5*time.Second {
			fails++
		} else {
			fails = 0
		}
		time.Sleep(5 * time.Second)
	}

	lPrintln(s.longID() + "的直播弹幕录制已经结束")
}

// 退出直播弹幕录制相关操作
func (s streamer) quitDanmu(liveID string) {
	finishDanmuPart(liveID)
}

// 初始化弹幕录制
func (s streamer) initDanmu(ctx context.Context, liveID, filename string) {
	dctx, dcancel := context.WithCancel(ctx)
	defer dcancel()
	if info, ok := getLiveInfo(liveID); ok && info.isDanmu {
		lPrintWarnf("已经在录制%s的直播弹幕，如要重启录制，请先运行 stopdanmu %d", s.longID(), s.UID)
		return
	}

	// 获取弹幕字幕设置
	cfg, err := s.getSubConfig()
	if err != nil {
		lPrintErr(err)
		lPrintErrf("无法获取%s的直播源，退出录制直播弹幕，请确定主播正在直播，如要重启录制直播弹幕，请运行 startdanmu %d", s.longID(), s.UID)
		return
	}

	info := liveInfo{LiveID: liveID, uid: s.UID, cfg: cfg}

	assFile := transFilename(filename)
	if assFile == "" {
		return
	}
	info.assFile = assFile + ".ass"
	info.cfg.Title = filepath.Base(assFile)
	info.cfg.StartTime = time.Now().UnixNano()
	info.isDanmu = true
	info.danmuCancel = dcancel
	updateLiveInfo(liveID, func(i *liveInfo) {
		i.uid = info.uid
		i.isDanmu = true
		i.danmuStarted = true
		i.danmuCancel = info.danmuCancel
		i.assFile = info.assFile
		i.cfg = info.cfg
	})
	defer s.quitDanmu(info.LiveID)

	s.getDanmu(dctx, info)
}

// 临时录制指定主播的直播弹幕
func startDanmu(uid int) bool {
	s, ok := getStreamer(uid)
	if !ok {
		name := getName(uid)
		if name == "" {
			lPrintWarnf("不存在 uid 为%d的用户", uid)
			return false
		}
		s = streamer{UID: uid, Name: name}
	}

	liveID := getLiveID(uid)
	if liveID == "" {
		lPrintErr(s.longID() + "不在直播，取消录制直播弹幕")
		return false
	}
	if isDanmu(liveID) {
		lPrintWarnf("已经在录制%s的直播弹幕，如要重启录制，请先运行 stopdanmu %d", s.longID(), s.UID)
		return false
	}

	filename := getTime() + " " + s.Name + " " + s.getTitle()

	// goroutine 是为了快速返回
	safeGo(s.longID()+"的弹幕录制", func() { s.initDanmu(mainCtx, liveID, filename) })
	return true
}

// 停止录制指定主播的直播弹幕
func stopDanmu(uid int) bool {
	infoList, ok := getLiveInfoByUID(uid)
	if !ok {
		lPrintWarnf("没有在录制 uid 为%d的主播的直播弹幕", uid)
		return true
	}

	for _, info := range infoList {
		if info.isDanmu {
			lPrintf("开始停止录制%s的liveID为%s直播弹幕", longID(uid), info.LiveID)
			updateLiveInfo(info.LiveID, func(i *liveInfo) { i.danmuStopped = true })
			info.danmuCancel()
		}
	}

	return true
}