#!/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