app/Piped.py

from random import choice
from typing import Dict, List, Optional

import httpx

from app.config import Settings
from app.proxy_functions import FASTEST_PROXY, UP_PROXIES
from app.Stream import Stream, download_stream

settings = Settings()


class Piped:
    def __init__(self, video_id: str):
        self.video_id: str = video_id
        self._base_url: Optional[str] = "https://pipedapi.kavin.rocks"
        self.stream_url: str = f"{settings.default_proxy}/streams/"
        self._video_properties: Optional[Dict] = None
        self._audio_streams: Optional[List] = None
        self._video_streams: Optional[List] = None

    @property
    def video_properties(self) -> dict:
        if self._video_properties is not None:
            return self._video_properties

        self._video_properties = dict()

        response = httpx.get(f"{self.stream_url}{self.video_id}?instance={FASTEST_PROXY.url}")

        self._video_properties.update(response.json())

        return self._video_properties

    @property
    def audio_streams(self) -> list:
        if self._audio_streams is not None:
            return self._audio_streams
        elif self._video_properties is not None:
            return self._video_properties["audioStreams"]
        return self.video_properties["audioStreams"]

    @property
    def video_streams(self) -> list:
        if self._video_streams is not None:
            return self._video_streams
        return self.video_properties["videoStreams"]

    @property
    def title(self) -> str:
        if self._video_properties is not None:
            return self._video_properties["title"]
        return self.video_properties["title"]

    @property
    def description(self) -> str:
        if self._video_properties is not None:
            return self._video_properties["description"]
        return self.video_properties["description"]

    def set_base_url(self) -> str:
        if FASTEST_PROXY is not None:
            self._base_url = FASTEST_PROXY.url
        elif UP_PROXIES is not None:
            self._base_url = choice(UP_PROXIES[0]).url
        else:
            self._base_url = settings.default_proxy
        return self._base_url

    def get_best_audio_stream(self) -> Stream:
        if self._audio_streams is not None:
            return Stream(self._audio_streams[0])

        return Stream(self.audio_streams[0])

    def get_best_video_stream(self) -> Stream:
        if self._video_streams is not None:
            return Stream(self._video_streams[0])
        return Stream(self.video_streams[0])

    # TODO: Add support for start/stop times
    def download_audio_stream(self, stream) -> str:
        # starting = None
        # ending = None
        return download_stream(stream=stream, title=self.title, file_type="audio")

    # TODO: Add support for start/stop times
    def download_video_stream(self, stream) -> str:
        # starting = None
        # ending = None
        return download_stream(stream=stream, title=self.title, file_type="video")