This repository has been archived by the owner on May 23, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 9
/
utils.py
53 lines (39 loc) · 1.53 KB
/
utils.py
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
import os
from asyncio import sleep
from datetime import datetime
from urllib.parse import urlencode
from aiohttp import ClientSession
from pytube import YouTube
video_links = dict()
metric_chart = {
'k': 3, # kilo
'M': 6, # Mega
'G': 9, # Giga
'T': 12 # Tera
}
def parametrize(url, params):
return url + '?' + urlencode(params)
async def get(url, params=None, loop=None):
async with ClientSession(loop=loop) as client:
async with client.get(parametrize(url, params)) as response:
return await response.read()
def get_resolution(yt_video):
return int(''.join(filter(str.isdigit, yt_video.resolution[:-1])))
def get_youtube_url(video_id):
if video_id in video_links and video_links[video_id]['expire'] > datetime.now():
return video_links[video_id]['url']
yt_video = YouTube(f'http://www.youtube.com/watch?v={video_id}')
video_url = yt_video.streams.get_highest_resolution().url
parts = {part.split('=')[0]: part.split('=')[1] for part in video_url.split('?')[-1].split('&')}
link = {'url': video_url, 'expire': datetime.fromtimestamp(int(parts['expire']))}
video_links[video_id] = link
return link['url']
def metric_to_base(metric):
return int(metric[:-1]) * (10 ** metric_chart[metric[-1]])
async def get_total_storage(directory='.'):
total_storage = 0
for root, directories, files in os.walk(directory):
for file in files:
total_storage += os.path.getsize(os.path.join(root, file))
await sleep(0)
return total_storage