335 lines
13 KiB
Python
Executable File
335 lines
13 KiB
Python
Executable File
#!/usr/bin/python3
|
|
# -*- coding: utf-8 -*-
|
|
import logging
|
|
import traceback
|
|
import asyncio
|
|
import datetime
|
|
import time
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import os
|
|
|
|
SAMPLEBYTES = 2
|
|
|
|
# Playing a googlevideo.com URL directly with ffmpeg turned out unreliable:
|
|
# the signed URL (and the exact headers/User-Agent yt-dlp used to obtain it)
|
|
# is tied to the specific request that generated it, and a *separate* ffmpeg
|
|
# process making its own connection -- even through the same proxy, even with
|
|
# matching headers copied from the --dump-json output -- still got HTTP 403
|
|
# from Google. Matching proxy + headers wasn't enough because the header set
|
|
# also drifts (yt-dlp's impersonated Chrome build number alone changed
|
|
# between two of our test runs a few minutes apart), so hardcoding it is a
|
|
# losing game.
|
|
#
|
|
# The reliable fix is to stop trying to replay yt-dlp's signed URL from a
|
|
# second process at all: let yt-dlp itself fetch the media (same process,
|
|
# same session, same proxy/headers it used to resolve the URL) and stream its
|
|
# output straight into ffmpeg's stdin via a pipe. URIs that should be played
|
|
# this way are prefixed with YTDLP_URI_PREFIX by Commands.py.
|
|
YTDLP_URI_PREFIX = "ytdlp:"
|
|
|
|
class FFmpegAudioPlayerFactory():
|
|
VALID_CALLBACKS = ["Play", "Stop", "Update"]
|
|
|
|
def __init__(self, master):
|
|
self.Logger = logging.getLogger(__class__.__name__)
|
|
self.Master = master
|
|
self.Torchlight = self.Master.Torchlight
|
|
|
|
def __del__(self):
|
|
self.Master.Logger.info("~FFmpegAudioPlayerFactory()")
|
|
self.Quit()
|
|
|
|
def NewPlayer(self):
|
|
self.Logger.debug(sys._getframe().f_code.co_name)
|
|
Player = FFmpegAudioPlayer(self)
|
|
return Player
|
|
|
|
def Quit(self):
|
|
self.Master.Logger.info("FFmpegAudioPlayerFactory->Quit()")
|
|
|
|
|
|
class FFmpegAudioPlayer():
|
|
def __init__(self, master):
|
|
self.Master = master
|
|
self.Torchlight = self.Master.Torchlight
|
|
self.Playing = False
|
|
|
|
self.Host = (
|
|
self.Torchlight().Config["VoiceServer"]["Host"],
|
|
self.Torchlight().Config["VoiceServer"]["Port"]
|
|
)
|
|
self.SampleRate = float(self.Torchlight().Config["VoiceServer"]["SampleRate"])
|
|
|
|
self.StartedPlaying = None
|
|
self.StoppedPlaying = None
|
|
self.Seconds = 0.0
|
|
|
|
self.Writer = None
|
|
self.Process = None
|
|
# Only set in ytdlp-pipe mode (see PlayURI): the upstream yt-dlp
|
|
# process feeding ffmpeg's stdin. Needs to be killed alongside
|
|
# self.Process on Stop(), or it lingers as an orphaned process.
|
|
self.SourceProcess = None
|
|
|
|
self.Callbacks = []
|
|
|
|
def __del__(self):
|
|
self.Master.Logger.debug("~FFmpegAudioPlayer()")
|
|
self.Stop()
|
|
|
|
def PlayURI(self, uri, position, rubberband = None, dec_params = None, bitrate = None,
|
|
backwards = None, *args):
|
|
IsYtdlp = uri.startswith(YTDLP_URI_PREFIX)
|
|
# Commands.py may append a "#t=..." fragment to the URI purely so a
|
|
# downstream handler can re-parse it into a start position (see
|
|
# URLFilter.URLInfo) -- strip it back off here since the actual seek
|
|
# already happens via `position`/"-ss" below, and yt-dlp doesn't need
|
|
# (or want) an arbitrary fragment tacked onto the watch URL.
|
|
SourceUrl = uri[len(YTDLP_URI_PREFIX):].split("#")[0] if IsYtdlp else None
|
|
|
|
if IsYtdlp:
|
|
# Input is a pipe (yt-dlp's stdout), which isn't seekable, so a
|
|
# requested start position has to be an *output*-side "-ss"
|
|
# (placed after "-i") rather than the normal fast input-side
|
|
# seek. Slower (ffmpeg decodes and discards up to that point
|
|
# instead of jumping straight there) but correct.
|
|
Command = ["/usr/bin/ffmpeg", "-i", "pipe:0"]
|
|
if position:
|
|
Command += ["-ss", str(datetime.timedelta(seconds = position))]
|
|
Command += ["-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
|
|
elif position:
|
|
PosStr = str(datetime.timedelta(seconds = position))
|
|
Command = ["/usr/bin/ffmpeg", "-ss", PosStr, "-i", uri, "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
|
|
else:
|
|
Command = ["/usr/bin/ffmpeg", "-i", uri, "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
|
|
|
|
self.Playing = True
|
|
if dec_params:
|
|
Command += dec_params
|
|
|
|
if rubberband and backwards:
|
|
Command += ["-filter:a"]
|
|
rubberCommand = ""
|
|
for rubber in rubberband:
|
|
rubberCommand += rubber + ", "
|
|
rubberCommand = rubberCommand[:-2]
|
|
Command += [rubberCommand + "[reversed];[reversed]areverse"] #[reversed] is intermediate stream label so reverse knows what stream label to reverse
|
|
else:
|
|
if rubberband:
|
|
Command += ["-filter:a"]
|
|
rubberCommand = ""
|
|
for rubber in rubberband:
|
|
rubberCommand += rubber + ", "
|
|
rubberCommand = rubberCommand[:-2]
|
|
Command += [rubberCommand]
|
|
if backwards:
|
|
Command += ["-af"]
|
|
Command += ["areverse"]
|
|
if bitrate:
|
|
Command += ["-ab ", str(bitrate), "k"]
|
|
self.Master.Logger.debug(f"command: {Command}")
|
|
Command += ["-"]
|
|
#self.Master.Logger.debug(f"command: {Command}")
|
|
asyncio.ensure_future(self._stream_subprocess(Command, SourceUrl))
|
|
return True
|
|
|
|
def Stop(self, force = True):
|
|
if not self.Playing:
|
|
return False
|
|
|
|
if self.Process:
|
|
try:
|
|
self.Process.terminate()
|
|
self.Process.kill()
|
|
self.Process = None
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
if self.SourceProcess:
|
|
try:
|
|
self.SourceProcess.terminate()
|
|
self.SourceProcess.kill()
|
|
self.SourceProcess = None
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
if self.Writer:
|
|
if force:
|
|
Socket = self.Writer.transport.get_extra_info("socket")
|
|
if Socket:
|
|
Socket.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
|
|
struct.pack("ii", 1, 0))
|
|
|
|
self.Writer.transport.abort()
|
|
|
|
self.Writer.close()
|
|
|
|
self.Playing = False
|
|
|
|
self.Callback("Stop")
|
|
del self.Callbacks
|
|
|
|
return True
|
|
|
|
def AddCallback(self, cbtype, cbfunc):
|
|
if not cbtype in FFmpegAudioPlayerFactory.VALID_CALLBACKS:
|
|
return False
|
|
|
|
self.Callbacks.append((cbtype, cbfunc))
|
|
return True
|
|
|
|
def Callback(self, cbtype, *args, **kwargs):
|
|
for Callback in self.Callbacks:
|
|
if Callback[0] == cbtype:
|
|
try:
|
|
Callback[1](*args, **kwargs)
|
|
except Exception as e:
|
|
self.Master.Logger.error(traceback.format_exc())
|
|
|
|
async def _updater(self):
|
|
LastSecondsElapsed = 0.0
|
|
|
|
while self.Playing:
|
|
SecondsElapsed = time.time() - self.StartedPlaying
|
|
|
|
if SecondsElapsed > self.Seconds:
|
|
SecondsElapsed = self.Seconds
|
|
|
|
self.Callback("Update", LastSecondsElapsed, SecondsElapsed)
|
|
|
|
if SecondsElapsed >= self.Seconds:
|
|
if not self.StoppedPlaying:
|
|
print("BUFFER UNDERRUN!")
|
|
self.Stop(False)
|
|
return
|
|
|
|
LastSecondsElapsed = SecondsElapsed
|
|
|
|
await asyncio.sleep(0.1)
|
|
|
|
async def _read_stream(self, stream, writer):
|
|
Started = False
|
|
|
|
while stream and self.Playing:
|
|
Data = await stream.read(65536)
|
|
|
|
if Data:
|
|
writer.write(Data)
|
|
await writer.drain()
|
|
|
|
Bytes = len(Data)
|
|
Samples = Bytes / SAMPLEBYTES
|
|
Seconds = Samples / self.SampleRate
|
|
|
|
self.Seconds += Seconds
|
|
|
|
if not Started:
|
|
Started = True
|
|
self.Callback("Play")
|
|
self.StartedPlaying = time.time()
|
|
asyncio.ensure_future(self._updater())
|
|
else:
|
|
self.Process = None
|
|
break
|
|
|
|
self.StoppedPlaying = time.time()
|
|
|
|
async def _stream_subprocess(self, cmd, source_url = None):
|
|
if not self.Playing:
|
|
return
|
|
|
|
_, self.Writer = await asyncio.open_connection(self.Host[0], self.Host[1])
|
|
|
|
# The metadata fetch (talks to youtube.com) and the actual media
|
|
# fetch (talks to googlevideo.com) are separate TCP connections even
|
|
# within one yt-dlp run, and WARP's proxy doesn't guarantee the same
|
|
# apparent exit IP across separate connections -- Google signs the
|
|
# media URL to whichever IP it saw on the first connection, and a
|
|
# mismatched second connection gets 403'd. A retry opens a fresh
|
|
# connection through WARP and has a decent chance of landing on a
|
|
# matching IP, so we give it a few tries before giving up.
|
|
MaxAttempts = 3 if source_url else 1
|
|
|
|
for Attempt in range(1, MaxAttempts + 1):
|
|
if not self.Playing:
|
|
return
|
|
|
|
SourceProcess = None
|
|
SourceStdErrTask = None
|
|
StdinArg = None
|
|
ReadFd = None
|
|
if source_url:
|
|
# asyncio.subprocess.PIPE gives back a StreamReader, which is
|
|
# NOT a real file descriptor and can't be handed to a second
|
|
# subprocess as stdin -- so we make an actual OS pipe and pass
|
|
# the raw fds instead, exactly like a shell "|" would, letting
|
|
# the kernel move the bytes without going through this
|
|
# process.
|
|
ReadFd, WriteFd = os.pipe()
|
|
try:
|
|
SourceProcess = await asyncio.create_subprocess_exec(
|
|
"yt-dlp", "-f", "bestaudio/best", "-o", "-", "--no-playlist", "--quiet", source_url,
|
|
stdout = WriteFd, stderr = asyncio.subprocess.PIPE)
|
|
finally:
|
|
os.close(WriteFd) # our copy; the child inherited its own
|
|
|
|
self.SourceProcess = SourceProcess
|
|
SourceStdErrTask = asyncio.ensure_future(SourceProcess.stderr.read())
|
|
StdinArg = ReadFd
|
|
|
|
Process = await asyncio.create_subprocess_exec(*cmd,
|
|
stdin = StdinArg,
|
|
stdout = asyncio.subprocess.PIPE, stderr = asyncio.subprocess.PIPE)
|
|
self.Process = Process
|
|
|
|
if source_url:
|
|
os.close(ReadFd) # our copy; the child inherited its own
|
|
|
|
# Drain stderr concurrently so it can't fill its pipe buffer and
|
|
# stall ffmpeg; we only care about it if playback produced
|
|
# nothing.
|
|
StdErrTask = asyncio.ensure_future(Process.stderr.read())
|
|
|
|
self.Seconds = 0.0
|
|
await self._read_stream(Process.stdout, self.Writer)
|
|
await Process.wait()
|
|
if SourceProcess:
|
|
await SourceProcess.wait()
|
|
|
|
if self.Seconds > 0.0:
|
|
return # success
|
|
|
|
StdErr = b""
|
|
try:
|
|
StdErr = await asyncio.wait_for(StdErrTask, timeout = 2)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
SourceStdErr = b""
|
|
if SourceStdErrTask:
|
|
try:
|
|
SourceStdErr = await asyncio.wait_for(SourceStdErrTask, timeout = 2)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
Is403 = source_url and b"403" in SourceStdErr
|
|
|
|
if Is403 and Attempt < MaxAttempts and self.Playing:
|
|
self.Master.Logger.warning("yt-dlp download got 403 (attempt %s/%s), retrying: %s",
|
|
Attempt, MaxAttempts, source_url)
|
|
continue
|
|
|
|
self.Master.Logger.warning(
|
|
"no audio produced (ffmpeg rc=%s cmd=%s stderr=%s) (yt-dlp rc=%s url=%s stderr=%s)",
|
|
Process.returncode, cmd, StdErr.decode(errors = "replace")[-1500:],
|
|
SourceProcess.returncode if SourceProcess else None, source_url,
|
|
SourceStdErr.decode(errors = "replace")[-1500:])
|
|
|
|
if Is403:
|
|
self.Torchlight().SayChat("Error: YouTube blocked this stream after {0} attempts, try again.".format(MaxAttempts))
|
|
|
|
self.Stop()
|
|
return
|