File size: 3,025 Bytes
81e73d0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
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")
|