不透過 YouTube Data API 側錄直播聊天室留言

• 技術分享
• YouTube
• 網頁爬蟲
• 直播聊天室

前言

官方 API 的做法

整體流程

1. 解析影片 ID

const ID_PATTERN = /^[A-Za-z0-9_-]{6,20}$/

export const parseVideoId = (input: string): string => {
  const value = input.trim()
  if (ID_PATTERN.test(value)) return value

  const url = new URL(value.includes('://') ? value : `https://${value}`)
  const host = url.hostname.toLowerCase().replace(/^www\./, '')

  // https://youtu.be/<id>
  if (host === 'youtu.be') return url.pathname.split('/').filter(Boolean)[0]

  // https://www.youtube.com/watch?v=<id>
  const queryId = url.searchParams.get('v')
  if (queryId) return queryId

  // https://www.youtube.com/live/<id>、/embed/<id>、/shorts/<id>
  const [kind, id] = url.pathname.split('/').filter(Boolean)
  if (['live', 'embed', 'shorts'].includes(kind) && id) return id

  throw new Error('請提供 YouTube 直播網址或影片 ID。')
}

2. 從聊天室網頁取得初始參數

https://www.youtube.com/live_chat?is_popout=1&v=<影片ID>

const USER_AGENT =
  'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 ' +
  '(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36'

// HTML 裡是 JSON 字串，可能含有 \u0026 這類跳脫字元，要還原
const decodeJsonString = (value: string): string => JSON.parse(`"${value}"`)

export const fetchBootstrap = async (videoId: string) => {
  const url = new URL('https://www.youtube.com/live_chat')
  url.searchParams.set('is_popout', '1')
  url.searchParams.set('v', videoId)

  const response = await fetch(url, { headers: { 'User-Agent': USER_AGENT } })
  if (!response.ok) throw new Error(`無法載入聊天室頁面（HTTP ${response.status}）`)
  const html = await response.text()

  const apiKey = html.match(/"INNERTUBE_API_KEY":"((?:\\.|[^"\\])*)"/)?.[1]
  const clientVersion =
    html.match(/"INNERTUBE_CONTEXT_CLIENT_VERSION":"((?:\\.|[^"\\])*)"/)?.[1] ??
    html.match(/"clientVersion":"((?:\\.|[^"\\])*)"/)?.[1]
  const continuation =
    html.match(/"continuation":"((?:\\.|[^"\\])*)","clickTrackingParams":/)?.[1]

  if (!apiKey || !clientVersion || !continuation) {
    throw new Error('頁面沒有可用的聊天室資訊；請確認直播中且聊天室已啟用。')
  }
  return {
    apiKey: decodeJsonString(apiKey),
    clientVersion: decodeJsonString(clientVersion),
    continuation: decodeJsonString(continuation),
  }
}

3. 輪詢聊天室端點

export const fetchLiveChatBatch = async (
  bootstrap: { apiKey: string, clientVersion: string },
  continuation: string,
) => {
  const url = new URL('https://www.youtube.com/youtubei/v1/live_chat/get_live_chat')
  url.searchParams.set('key', bootstrap.apiKey)
  url.searchParams.set('prettyPrint', 'false')

  const response = await fetch(url, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'User-Agent': USER_AGENT },
    body: JSON.stringify({
      context: { client: { clientName: 'WEB', clientVersion: bootstrap.clientVersion } },
      continuation,
    }),
  })
  if (!response.ok) throw new Error(`聊天室輪詢失敗（HTTP ${response.status}）`)
  return response.json()
}

4. 解析留言

continuationContents
  └ liveChatContinuation
      ├ actions[]         這一批的事件
      └ continuations[]   下一次要用的 token 與等待時間

const textFromRuns = (message: any): string => {
  if (!message) return ''
  if (typeof message.simpleText === 'string') return message.simpleText

  return (message.runs ?? [])
    .map((run: any) => {
      if (typeof run.text === 'string') return run.text
      // 表情符號：優先用 :shortcut: 形式，其次搜尋詞，最後才用 emojiId
      const emoji = run.emoji
      if (!emoji) return ''
      return emoji.shortcuts?.[0]
        ?? (emoji.searchTerms?.[0] ? `:${emoji.searchTerms[0]}:` : undefined)
        ?? emoji.emojiId
        ?? '[emoji]'
    })
    .join('')
    .trim()
}

const toIso = (usec: string) => new Date(Number(BigInt(usec) / 1000n)).toISOString()

5. 下一個 token 與輪詢間隔

const nextContinuation = (liveChatContinuation: any) => {
  for (const entry of liveChatContinuation.continuations ?? []) {
    const data = entry.invalidationContinuationData
      ?? entry.timedContinuationData
      ?? entry.liveChatReplayContinuationData
    if (!data) continue
    const timeout = Number(data.timeoutMs)
    return {
      continuation: data.continuation ?? null,
      // 限制在 0.25–30 秒之間，避免異常值讓程式狂打或卡住
      pollAfterMs: Number.isFinite(timeout) ? Math.min(30_000, Math.max(250, timeout)) : 1_000,
    }
  }
  return { continuation: null, pollAfterMs: 1_000 }
}

6. 去重與寫檔

const seen = new Set<string>()

for (const message of batch.messages) {
  if (seen.has(message.id)) continue
  seen.add(message.id)
  await file.appendFile(`${JSON.stringify(message)}\n`, 'utf8')
}

7. 主迴圈與錯誤處理

const run = async (input: string, signal: AbortSignal) => {
  const videoId = parseVideoId(input)
  let bootstrap = await fetchBootstrap(videoId)
  let continuation = bootstrap.continuation
  let failures = 0

  while (!signal.aborted) {
    try {
      const payload = await fetchLiveChatBatch(bootstrap, continuation)
      const live = payload?.continuationContents?.liveChatContinuation
      if (!live) throw new Error('回應沒有 liveChatContinuation，直播可能已結束')

      await saveMessages(live.actions ?? [])        // 第 4、6 步
      const next = nextContinuation(live)            // 第 5 步
      if (!next.continuation) break                  // 直播結束
      continuation = next.continuation
      failures = 0
      await sleep(next.pollAfterMs, signal)
    } catch (error) {
      if (signal.aborted) break
      failures += 1
      // 連續失敗 3 次，token 可能已經失效，重新從網頁取得
      if (failures >= 3) {
        bootstrap = await fetchBootstrap(videoId)
        continuation = bootstrap.continuation
        failures = 0
      }
      // 指數退避，最多等 30 秒
      await sleep(Math.min(30_000, 1_000 * 2 ** failures), signal)
    }
  }
}

隱私上的處理

import { createHmac, randomUUID } from 'node:crypto'

const sessionKey = randomUUID() // 每次側錄重新產生，不寫入檔案
const pseudonym = (channelId: string) =>
  `viewer_${createHmac('sha256', sessionKey).update(channelId).digest('hex').slice(0, 12)}`

已知限制

結語