diff --git a/.env.example b/.env.example index 6d47a9f..88587c8 100644 --- a/.env.example +++ b/.env.example @@ -75,6 +75,15 @@ YTDLP_JS_RUNTIME=node # Необязательно: каталог с локальной медиатекой (смонтируйте том). # LOCAL_MEDIA_DIR=/media/music +# Показывать клип в голосовом канале как демонстрацию экрана (только YouTube). +# Требует прав Video у бота и включённого видео в конфигурации инстанса. +# YouTube отдаёт одним файлом только 360p, а стримить можно лишь такой формат — +# отсюда размер по умолчанию. Кодирование видео заметно грузит CPU сервера. +VIDEO_ENABLED=false +VIDEO_WIDTH=640 +VIDEO_HEIGHT=360 +VIDEO_FPS=24 + DEFAULT_VOLUME=60 MAX_QUEUE_SIZE=500 diff --git a/README.md b/README.md index 464fa47..227d615 100644 --- a/README.md +++ b/README.md @@ -225,6 +225,36 @@ docker compose build --build-arg YTDLP_VERSION=$(date +%Y.%m.%d) && docker compo Актуальный тег — на [странице релизов yt-dlp](https://github.com/yt-dlp/yt-dlp/releases). +## Видео: клип как демонстрация экрана + +`VIDEO_ENABLED=true` — и вместе со звуком бот публикует картинку клипа отдельной дорожкой +(screen share), которую видно в голосовом канале. + +```dotenv +VIDEO_ENABLED=true +VIDEO_WIDTH=640 +VIDEO_HEIGHT=360 +VIDEO_FPS=24 +``` + +Что нужно на стороне Stoat: у роли бота — право **Video** в канале, а на инстансе включённое +видео (`VIDEO_ENABLED` при генерации конфига, он же `video_resolution` в `Revolt.toml`). Токен +на вход в звонок выдаёт разрешение публиковать `screen_share` только при обоих условиях; иначе +бот сообщит в чат, что видео недоступно, и продолжит играть звук. + +Как это устроено и почему так: + +- **Только YouTube и только 360p.** Стримить можно лишь прогрессивный формат — один файл со + звуком и картинкой. Раздельные дорожки (720p и выше) yt-dlp обязан сначала скачать целиком + и лишь потом склеить, то есть воспроизведение началось бы после полной загрузки. Отдавать + ffmpeg прямые ссылки на CDN тоже нельзя: YouTube их для сторонних клиентов подвешивает. +- **Один процесс ffmpeg, два выхода:** PCM для голосовой дорожки и сырые I420-кадры для видео. + Темп задаёт `-re`, иначе кадры улетали бы вперёд звука и съедали память — распакованный кадр + 720p весит 1.4 МБ. +- **Цена.** Кодирование видео ложится на CPU сервера и держится всё время трека, в отличие от + почти бесплатного звука. Ставьте `VIDEO_FPS` пониже, если нагрузка мешает. +- Живые трансляции и не-YouTube источники играют звуком, как раньше. + ## Прокси, когда YouTube недоступен `YTDLP_PROXY` пропускает через прокси **все** обращения yt-dlp — поиск, метаданные и сам diff --git a/package.json b/package.json index 6bfb8fb..1018957 100644 --- a/package.json +++ b/package.json @@ -20,7 +20,9 @@ "@fastify/cookie": "^11.0.2", "@fastify/static": "^8.1.1", "@fastify/websocket": "^11.0.2", + "@livekit/rtc-node": "^0.13.34", "fastify": "^5.2.1", + "ffmpeg-static": "^5.3.0", "jose": "^6.0.10", "pino": "^9.6.0", "pino-pretty": "^13.0.0", diff --git a/src/config.ts b/src/config.ts index 80802b9..4e81b5f 100644 --- a/src/config.ts +++ b/src/config.ts @@ -62,6 +62,17 @@ const schema = z.object({ YTDLP_JS_RUNTIME: z.string().default("node"), LOCAL_MEDIA_DIR: z.string().optional(), + /** Publish the clip as a screen share alongside the audio (YouTube only). */ + VIDEO_ENABLED: z + .enum(["true", "false"]) + .default("false") + .transform((value) => value === "true"), + // YouTube only serves 360p as a single progressive file, and only such files + // can be streamed, so that is the realistic default. + VIDEO_WIDTH: z.coerce.number().int().positive().default(640), + VIDEO_HEIGHT: z.coerce.number().int().positive().default(360), + VIDEO_FPS: z.coerce.number().int().min(1).max(60).default(24), + DEFAULT_VOLUME: z.coerce.number().min(0).max(200).default(60), MAX_QUEUE_SIZE: z.coerce.number().int().positive().default(500), /** Upper bound on how many tracks one pasted playlist may add. */ diff --git a/src/core/player.ts b/src/core/player.ts index 5166038..1e5b17f 100644 --- a/src/core/player.ts +++ b/src/core/player.ts @@ -9,6 +9,7 @@ import { type PlayerStatus, type Track, } from "../types.js"; +import { VideoPublisher } from "./video.js"; import { MediaPlayer, parseFfmpegDuration, @@ -73,6 +74,7 @@ export class GuildPlayer extends EventEmitter { /** Tracked from connection events: revoice's own `connected` getter is broken. */ private voiceReady = false; private media: MediaPlayerLike | null = null; + private videoPublisher: VideoPublisher | null = null; private currentInput: PlaybackInput | null = null; private seekOffset = 0; /** Set while we tear playback down ourselves, so the resulting `finish` is ignored. */ @@ -217,6 +219,7 @@ export class GuildPlayer extends EventEmitter { async leaveVoice(): Promise { this.cancelLeaveTimer(); + this.stopScreenShare(); this.stopTicker(); this.teardownPlayback(); this.current = null; @@ -293,6 +296,10 @@ export class GuildPlayer extends EventEmitter { media.setVolume(this.volume / 100); this.startTicker(); + if (input.video && this.connection) { + await this.startScreenShare(input.video); + } + // A downloader that dies mid-stream just looks like a very short track, so // say why instead of silently moving on. void input.failure?.then((reason) => { @@ -331,11 +338,36 @@ export class GuildPlayer extends EventEmitter { this.log.debug({ err }, "media.stop() threw"); } } + this.stopScreenShare(); this.currentInput?.cleanup(); this.currentInput = null; this.seekOffset = 0; } + private async startScreenShare(video: NonNullable): Promise { + const room = this.connection?.room; + if (!room) return; + const publisher = new VideoPublisher({ width: video.width, height: video.height }); + try { + await publisher.start(room, video.stream, (reason) => { + this.notify(`📺 Видео отключено: ${reason}.`); + }); + this.videoPublisher = publisher; + } catch (err) { + this.log.warn({ err }, "could not publish screen share"); + this.notify( + "📺 Не удалось показать клип: инстанс не разрешает видео (нужны право Video у бота и включённое видео в конфигурации Stoat). Звук играет как обычно.", + ); + await publisher.stop().catch(() => {}); + } + } + + private stopScreenShare(): void { + const publisher = this.videoPublisher; + this.videoPublisher = null; + if (publisher) void publisher.stop().catch(() => {}); + } + private async handleFinish(): Promise { if (this.expectingStop) { this.expectingStop = false; diff --git a/src/core/revoice.ts b/src/core/revoice.ts index ca2eaf0..12012e3 100644 --- a/src/core/revoice.ts +++ b/src/core/revoice.ts @@ -1,5 +1,6 @@ import { createRequire } from "node:module"; import type { Readable } from "node:stream"; +import type { Room } from "@livekit/rtc-node"; import { UserFacingError } from "../types.js"; // revoice.js is CommonJS and its bundled typings lag behind the LiveKit rewrite, @@ -29,6 +30,8 @@ export const VOICE_STATE_OFFLINE = "off"; export interface VoiceConnectionLike { channelId: string; + /** The underlying LiveKit room, used to publish the screen share. */ + room: Room; play(media: MediaPlayerLike): Promise; leave(): Promise; destroy(): Promise; diff --git a/src/core/video.ts b/src/core/video.ts new file mode 100644 index 0000000..713e142 --- /dev/null +++ b/src/core/video.ts @@ -0,0 +1,136 @@ +import { createRequire } from "node:module"; +import type { Readable } from "node:stream"; +import type { LocalVideoTrack, Room, VideoSource } from "@livekit/rtc-node"; +import { logger } from "../logger.js"; + +// revoice.js pulls in the CommonJS build of @livekit/rtc-node, and the native +// FFI client is per-module-instance: importing the ESM build here would give us +// a second instance whose tracks the room (created by revoice) cannot publish. +// So we require the very same module object it uses. +const require = createRequire(import.meta.url); +const rtc = require("@livekit/rtc-node") as typeof import("@livekit/rtc-node"); + +const log = logger.child({ mod: "video" }); +/** Give up on the video track if the source never produced a picture. */ +const FIRST_FRAME_TIMEOUT_MS = 15_000; + +export interface VideoPublisherOptions { + width: number; + height: number; +} + +/** + * Publishes a raw I420 stream into a LiveKit room as a screen share. ffmpeg + * feeds frames at playback speed, so this class only has to slice the byte + * stream into frames and hand them over. + */ +export class VideoPublisher { + private readonly frameSize: number; + private source: VideoSource | null = null; + private track: LocalVideoTrack | null = null; + private trackSid: string | null = null; + private room: Room | null = null; + private stream: Readable | null = null; + private pending: Buffer[] = []; + private pendingBytes = 0; + private frames = 0; + private firstFrameTimer: NodeJS.Timeout | null = null; + + constructor(private readonly options: VideoPublisherOptions) { + // I420: one luma plane plus two half-resolution chroma planes. + this.frameSize = Math.floor((options.width * options.height * 3) / 2); + } + + get isPublishing(): boolean { + return this.trackSid !== null; + } + + /** Publishes the track and starts pumping frames. Resolves once published. */ + async start(room: Room, stream: Readable, onGiveUp?: (reason: string) => void): Promise { + const { width, height } = this.options; + const source = new rtc.VideoSource(width, height); + const track = rtc.LocalVideoTrack.createVideoTrack("screen", source); + + const options = new rtc.TrackPublishOptions(); + options.source = rtc.TrackSource.SOURCE_SCREENSHARE; + + const participant = room.localParticipant; + if (!participant) throw new Error("room has no local participant yet"); + const publication = await participant.publishTrack(track, options); + + this.room = room; + this.source = source; + this.track = track; + this.trackSid = publication.sid ?? null; + this.stream = stream; + + stream.on("data", (chunk: Buffer) => this.consume(chunk)); + stream.once("error", (err) => log.debug({ err }, "video stream error")); + + this.firstFrameTimer = setTimeout(() => { + if (this.frames > 0) return; + log.warn("no video frames arrived, dropping the screen share"); + onGiveUp?.("источник не отдал видео"); + void this.stop(); + }, FIRST_FRAME_TIMEOUT_MS); + this.firstFrameTimer.unref?.(); + + log.info({ width, height, sid: publication.sid }, "screen share published"); + } + + private consume(chunk: Buffer): void { + if (!this.source) return; + this.pending.push(chunk); + this.pendingBytes += chunk.length; + if (this.pendingBytes < this.frameSize) return; + + let buffer = this.pending.length === 1 ? (this.pending[0] as Buffer) : Buffer.concat(this.pending, this.pendingBytes); + while (buffer.length >= this.frameSize) { + const frame = buffer.subarray(0, this.frameSize); + try { + this.source.captureFrame( + new rtc.VideoFrame(frame, this.options.width, this.options.height, rtc.VideoBufferType.I420), + ); + this.frames += 1; + } catch (err) { + log.debug({ err }, "captureFrame failed"); + } + buffer = buffer.subarray(this.frameSize); + } + // Keep only the tail of a partial frame for the next chunk. + this.pending = buffer.length > 0 ? [Buffer.from(buffer)] : []; + this.pendingBytes = buffer.length; + } + + async stop(): Promise { + if (this.firstFrameTimer) { + clearTimeout(this.firstFrameTimer); + this.firstFrameTimer = null; + } + this.stream?.removeAllListeners("data"); + this.stream = null; + this.pending = []; + this.pendingBytes = 0; + + const { room, trackSid } = this; + this.trackSid = null; + if (room?.localParticipant && trackSid) { + try { + await room.localParticipant.unpublishTrack(trackSid, true); + } catch (err) { + log.debug({ err }, "unpublishTrack failed"); + } + } + + try { + this.track?.close?.(); + await this.source?.close?.(); + } catch (err) { + log.debug({ err }, "closing video track failed"); + } + this.track = null; + this.source = null; + this.room = null; + this.frames = 0; + } +} diff --git a/src/sources/index.ts b/src/sources/index.ts index 0d2fb2a..680030e 100644 --- a/src/sources/index.ts +++ b/src/sources/index.ts @@ -3,6 +3,7 @@ import { config } from "../config.js"; import { UserFacingError, type Requester, type SearchResult, type Track } from "../types.js"; import * as direct from "./direct.js"; import * as local from "./local.js"; +import { openVideoPipeline } from "./video-pipeline.js"; import * as ytdlp from "./ytdlp.js"; export { checkAvailable as checkYtDlp, checkCookies, checkProxy, describeProxy } from "./ytdlp.js"; @@ -165,6 +166,13 @@ export interface PlaybackInput { cleanup(): void; /** Resolves with a reason if the downloader died on its own, for reporting. */ failure?: Promise; + /** Raw I420 frames to publish as a screen share, when video is on. */ + video?: { stream: Readable; width: number; height: number }; +} + +/** Only YouTube reliably carries a picture worth showing next to the audio. */ +function canShowVideo(track: Track): boolean { + return config.VIDEO_ENABLED && track.source === "youtube" && !track.isLive; } const HTTP_RESILIENCE = [ @@ -201,16 +209,9 @@ export async function openPlayback(track: Track, seekSeconds = 0): Promise 0 && !config.YTDLP_PROXY) { - // Seeking over a pipe would mean decoding everything up to the offset, so we - // resolve the CDN URL instead and let ffmpeg do an HTTP range request. - const streamUrl = await ytdlp.resolveStreamUrl(track.url); - return { input: streamUrl, inputOptions: [...HTTP_RESILIENCE, ...seekOptions], cleanup: () => {} }; - } - - // With a proxy configured we always pipe through yt-dlp, which honours it; - // handing a CDN URL to ffmpeg would leak the request past the proxy (and would - // simply fail for SOCKS). Seeking then costs a decode up to the offset. + // Everything goes through yt-dlp: YouTube stalls direct CDN URLs fetched by + // anything else, and ffmpeg would bypass a SOCKS proxy anyway. Seeking + // therefore costs a decode up to the offset instead of an HTTP range request. const proc = ytdlp.openAudioStream(track.url); return { input: proc.stream, diff --git a/src/sources/video-pipeline.ts b/src/sources/video-pipeline.ts new file mode 100644 index 0000000..7aaadef --- /dev/null +++ b/src/sources/video-pipeline.ts @@ -0,0 +1,121 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { createRequire } from "node:module"; +import type { Readable } from "node:stream"; +import { config } from "../config.js"; +import { logger } from "../logger.js"; +import { downloadArgs, progressiveFormat } from "./ytdlp.js"; + +const require = createRequire(import.meta.url); +/** ffmpeg-static ships the binary revoice.js already relies on. */ +const FFMPEG_PATH = require("ffmpeg-static") as string; + +const log = logger.child({ mod: "video-pipeline" }); + +export interface VideoPipeline { + /** Raw PCM for the audio track. */ + audio: Readable; + /** ffmpeg input options describing that raw PCM. */ + audioInputOptions: string[]; + /** Raw I420 frames for the screen share. */ + video: Readable; + width: number; + height: number; + kill(): void; +} + +/** + * One download, one decode, two outputs: PCM for the voice track and I420 frames + * for the screen share. + * + * yt-dlp does the fetching (so cookies and the proxy apply as everywhere else) + * and hands ffmpeg a progressive stream. `-re` paces the decode at playback + * speed: it keeps picture and sound together and stops the video from racing + * ahead — a decoded 720p frame is 1.4 MB, so an unpaced stream would eat memory + * in seconds. + */ +export function openVideoPipeline(pageUrl: string, seekSeconds = 0): VideoPipeline { + const { VIDEO_WIDTH: width, VIDEO_HEIGHT: height, VIDEO_FPS: fps } = config; + + const ytdlp = spawn( + config.YTDLP_PATH, + [ + ...downloadArgs(), + "-f", + progressiveFormat(height), + "--no-playlist", + "--quiet", + "-o", + "-", + pageUrl, + ], + { stdio: ["ignore", "pipe", "pipe"], windowsHide: true }, + ); + + const filters = [ + `scale=${width}:${height}:force_original_aspect_ratio=decrease`, + `pad=${width}:${height}:(ow-iw)/2:(oh-ih)/2`, + `fps=${fps}`, + "format=yuv420p", + ].join(","); + + const ffmpeg = spawn( + FFMPEG_PATH, + [ + "-hide_banner", + "-loglevel", + "error", + ...(seekSeconds > 0 ? ["-ss", seekSeconds.toFixed(2)] : []), + "-re", + "-i", + "pipe:0", + // Audio branch → stdout. + "-map", + "0:a:0?", + "-f", + "s16le", + "-ar", + "48000", + "-ac", + "2", + "pipe:1", + // Video branch → the extra pipe on fd 3. + "-map", + "0:v:0", + "-vf", + filters, + "-f", + "rawvideo", + "pipe:3", + ], + { stdio: ["pipe", "pipe", "pipe", "pipe"], windowsHide: true }, + ) as ChildProcessWithoutNullStreams & { stdio: Readable[] }; + + ytdlp.stdout.pipe(ffmpeg.stdin); + + ytdlp.stderr.setEncoding("utf8"); + ytdlp.stderr.on("data", (chunk: string) => log.warn({ ytdlp: chunk.trim() }, "video download")); + ffmpeg.stderr.setEncoding("utf8"); + ffmpeg.stderr.on("data", (chunk: string) => log.warn({ ffmpeg: chunk.trim() }, "ffmpeg")); + + // A closed pipe when the other side goes away is expected, not an error. + for (const stream of [ytdlp.stdout, ffmpeg.stdin, ffmpeg.stdout]) { + stream.on("error", (err: NodeJS.ErrnoException) => { + if (err.code !== "EPIPE") log.debug({ err }, "pipeline stream error"); + }); + } + + let killed = false; + return { + audio: ffmpeg.stdout, + audioInputOptions: ["-f", "s16le", "-ar", "48000", "-ac", "2"], + video: ffmpeg.stdio[3] as Readable, + width, + height, + kill: () => { + if (killed) return; + killed = true; + if (ytdlp.exitCode === null) ytdlp.kill("SIGKILL"); + if (ffmpeg.exitCode === null) ffmpeg.kill("SIGKILL"); + }, + }; +} diff --git a/src/sources/ytdlp.ts b/src/sources/ytdlp.ts index 7820a9f..a7eae19 100644 --- a/src/sources/ytdlp.ts +++ b/src/sources/ytdlp.ts @@ -341,19 +341,38 @@ export async function resolveUrl( return { tracks: [toTrack(root, requestedBy)], playlist: null }; } -/** Resolves a direct, time-limited media URL for a page URL (used when seeking). */ -export async function resolveStreamUrl(pageUrl: string): Promise { - const { stdout } = await runYtDlp([ - ...baseArgs({ download: true }), - "-f", - "bestaudio/best", - "--no-playlist", - "-g", - pageUrl, - ]); - const url = stdout.split(/\r?\n/).map((l) => l.trim()).find(Boolean); - if (!url) throw new UserFacingError("Не удалось получить ссылку на аудиопоток"); - return url; +/** yt-dlp arguments for a download, exposed for the video pipeline. */ +export function downloadArgs(): string[] { + return baseArgs({ download: true }); +} + +/** + * Checks whether YouTube still offers a progressive format (video and audio in + * one file) at this size. Only those can be streamed: anything that needs + * merging makes yt-dlp download both streams in full before writing a byte, and + * direct CDN URLs are no longer fetchable by ffmpeg — YouTube stalls them. + */ +export async function hasProgressiveVideo(pageUrl: string, maxHeight: number): Promise { + try { + const { stdout } = await runYtDlp([ + ...baseArgs({ download: true }), + "-f", + progressiveFormat(maxHeight), + "--no-playlist", + "--print", + "%(format_id)s", + pageUrl, + ]); + return stdout.trim().length > 0; + } catch (err) { + log.debug({ err, pageUrl }, "no progressive format for video playback"); + return false; + } +} + +/** Best single-file format within the height budget. */ +export function progressiveFormat(maxHeight: number): string { + return `b[height<=${maxHeight}]/b`; } /** Private, disposable copy of the cookie jar for concurrent reads. */