現状
Pythonのライブラリeelを使用してデスクトップアプリを作成してます。
eelはelectronライクでバックエンドをPython、フロントエンドをhtml, css, JavaScriptで作成出来ます。
フロント側にプログレスバーの機能を実装したく質問させて頂きます。
下記の投稿と(当質問における)コメント欄のアドバイスでPyTubeに実装されているプログレスバーを表示させる事が出来ました。
Is there a progress bar feature for pytube?
ここからさらにプログレスバーの値をフロント側に送信出来るようにしたく、How to combine pytube and tkinter label to show progress? の内容を元にコードを書きました。
その際、上記で使用したPyTubeのメソッドon_progress_callback=on_progress
は用いず、register_on_progress_callback(self.show_progress_bar)
を記述し(show_progress_barは自分で作成した関数)
おそらくYouTubeの動画をダウンロード中に登録した関数を呼び出せる機能だと思う。
self.progress
にファイルのプログレス状態が格納される事になっているのだと思うのですが、どうやって取得すればいいか分らないです。
おそらく非同期処理で一定の間隔で動画のダウンロードが終了するまでアクセスするコードを書く必要があると思うのですが、どうやって組み立てていけば分らずここで躓いています。
作成したコード
from pytube import YouTube
#from pytube.cli import on_progress
import os
class Downloader():
"""docstring for Youtubehq"""
def __init__(self, url):
#yt = YouTube(url, on_progress_callback=on_progress)
yt = YouTube(url)
self.title = yt.title
self.alltag = yt.streams
self.yt = yt
print(self.title)
####省略(画質・音質がよいitagを選択self.vieo_hq, self.audio_hqの取得をする。)#####
def show_progress_bar(self, stream=None, chunk=None, file_handle=None, bytes_remaining=None):
self.progress = str(int(100 - (100*(bytes_remaining/self.MaxfileSize))))
def download_video_audio(self):
yt = self.yt
yt.register_on_progress_callback(self.show_progress_bar)
itag_v = self.video_hq['itag']
itag_a = self.audio_hq['itag']
if not os.path.exists('video/original'):
os.makedirs('video/original')
yt.streams.get_by_itag(itag_v).download('video/original')
if not os.path.exists('audio/webm'):
os.makedirs('audio/webm')
yt.streams.get_by_itag(itag_a).download('audio/webm')
Downloader = Downloader('https://www.youtube.com/watch?v=3hPI3xjsNKE')
###省略(画質・音質選択処理)
Downloader.download_video_audio()
追記 自分なりに非同期化してみた。
イメージはリストに格納されたurlを非同期にしてさらにダウンロード最中に別のタスクでself.progress
にアクセスする事でプログレスバーの進捗を取得出来ると考えた。
プログレスバーは動画が複数の場合は別々で取得して、必要であればそれを足して総合的な進捗を表示出来るようにしたいです。
from pytube import YouTube
import os
import asyncio
class Downloader():
"""docstring for Youtubehq"""
def __init__(self):
self.__gtask = []
self.working = True
"""
self.makelist()
self.select_mobile_v()
self.select_hq_a()
の処理が入る。
ここで高画質・音質を選択する。self.video_hq, self.audio_hq
中身を取得する。
"""
def download_video_audio(self):
yt = self.yt
yt.register_on_progress_callback(self.show_progress_bar)
itag_v = self.video_hq['itag']
itag_a = self.audio_hq['itag']
if not os.path.exists('video/original'):
os.makedirs('video/original')
yt.streams.get_by_itag(itag_v).download('video/original')
if not os.path.exists('audio/webm'):
os.makedirs('audio/webm')
yt.streams.get_by_itag(itag_a).download('audio/webm')
def progress_value(self):
while self.working:
print(self.progress)
def download(self, url):
yt = YouTube(url)
self.title = yt.title
self.alltag = yt.streams
self.yt = yt
print(self.title)
self.makelist()
self.select_mobile_v()
self.select_hq_a()
self.download_video_audio()
self.working = False
def async_dl(self, urls):
async def async_wrap(loop, urls):
sem = asyncio.Semaphore(20)
async def async_ex(i):
async with sem:
return await loop.run_in_executor(None, self.download, url[i])
tasks = [async_ex(i) for i in range(len(urls))]
tasks =+ [await loop.run_in_executor(None, self.progress_value)]
self.__gtask = asyncio.gather(*tasks)
return await self.__gtask
loop.run_until_complete(self.__gtask)
loop = asyncio.get_event_loop()
return async_wrap(loop, urls)
Downloader = Downloader()
Downloader.async_dl(['https://www.youtube.com/watch?v=qZj8lrSZHsE'])
エラー
yt_download.py:203: RuntimeWarning: coroutine 'Downloader.async_dl.<locals>.async_wrap' was never awaited
Downloader.async_dl(['https://www.youtube.com/watch?v=qZj8lrSZHsE'])
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
from pytube.cli import on_progress
を入れないと未定義になりますよ。