music_album_creation.downloading.AbstractYoutubeDL.update_backend() - Code Metrics - Inspection of "Music Album Creator v1.4.0 Release" - boromir674/music-album-creator - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#2)

by Konstantinos

created 2023-04-03 18:15 UTC

() A

↳ Parent: music_album_creation.downloading.CMDYoutubeDownloader._download()

Complexity

Conditions

Size

Total Lines	13
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	9.561

Importance

Changes

Metric	Value
cc	3
eloc	11
nop	1
dl	0
loc	13
ccs	1
cts	10
cp	0.1
crap	9.561
rs	9.85
c	0
b	0
f	0

import json
import logging
import re
from abc import ABC
from pathlib import Path
from time import sleep
from typing import Union
from urllib.error import URLError

from pytube import YouTube

logger = logging.getLogger(__name__)


class CMDYoutubeDownloader:
    __instance = None

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(CMDYoutubeDownloader, cls).__new__(cls)
        return cls.__instance

    def download(self, video_url: str, directory: Union[str, Path], **kwargs) -> str:
        return self._download(video_url, directory)

    @classmethod
    def _download(cls, video_url, output_dir, **kwargs) -> str:
        # output dir where to store the stream
        yt = YouTube(video_url)
        download_parameters = dict(
            {
                'output_path': str(output_dir),
                # 'filename':f'{yt.title}.mp3',  # since this is an audio-only stream, the file will be mp4
                # 'filename': f'{title}.mp3',
                'filename_prefix': None,
                'skip_existing': True,  # Skip existing files, defaults to True
                'timeout': None,  # Request timeout length in seconds. Uses system default
                'max_retries': 0,  # Number of retries to attempt after socket timeout. Defaults to 0
            },
            **kwargs
        )
        try:
            title: str = yt.title
        except Exception as error:
            logger.exception(error)
            title = 'failed-to-get-title'

        # # find highest quality audio stream
        # # we currently judge quality by bitrate (higher is better)
        best_audio_stream = yt.streams.filter(only_audio=True).order_by('bitrate')[-1]

        # Download the audio stream
        try:
            local_file = best_audio_stream.download(**download_parameters)
        # Catch common bug on pytube (which is not too stable yet)
        except URLError as error:
            logger.error(
                "Youtube Download Error: %s",
                json.dumps(
                    {
                        'url': str(video_url),
                        'title': title,
                    },
                    indent=4,
                    sort_keys=True,
                ),
            )
            raise error
        logger.info(
            "Downloaded from Youtube: %s",
            json.dumps(
                {
                    'title': title,
                    'local_file': str(local_file),
                },
                indent=4,
                sort_keys=True,
            ),
        )
        return local_file

    def download_trials(self, video_url, directory, times=10, delay=0.5, **kwargs):
        """Download with retries

        Call this method to download a video with retries.

        Note:
            Designed for retrying when non-deterministic errors occur.

        Args:
            video_url (str): the youtube video url
            directory (str): the directory to store the downloaded file
            times (int, optional): Number of retries for non-deterministic bugs. Defaults to 10.
            delay (float, optional): Delay between retries to no stress youtube server. Defaults to 0.5.

        Raises:
            URLError: if the download fails after all retries

        Returns:
            [type]: [description]
        """
        i = 0
        while i < times:
            try:
                return self._download(video_url, directory, **kwargs)
            except URLError as error:
                if 'Network is unreachable' in str(error):
                    i += 1
                    sleep(delay)
                else:
                    raise error
            except TooManyRequestsError as e:
                i += 1
                sleep(delay)
        raise RetriesFailedError


class RetriesFailedError(Exception):
    pass


class YoutubeDownloaderErrorFactory(object):
    @staticmethod
    def create_with_message(msg):
        return Exception(msg)

    @staticmethod
    def create_from_stderr(stderror, video_url):
        exception_classes = (
            UploaderIDExtractionError,
            TokenParameterNotInVideoInfoError,
            InvalidUrlError,
            UnavailableVideoError,
            TooManyRequestsError,
            CertificateVerificationError,
            HTTPForbiddenError,
        )
        for subclass in exception_classes:
            if subclass.reg.search(stderror):
                return subclass(video_url, stderror)
        s = "NOTE: None of the predesinged exceptions' regexs [{}] matched. Perhaps you want to derive a new subclass from AbstractYoutubeDownloaderError to account for this youtube-dl exception with string to parse <S>{}</S>'".format(
            ', '.join(['"{}"'.format(_.reg) for _ in exception_classes]), stderror
        )
        return Exception(AbstractYoutubeDownloaderError(video_url, stderror)._msg + '\n' + s)


#### EXCEPTIONS


class AbstractYoutubeDownloaderError(ABC):
    def __init__(self, *args, **kwargs):
        super(AbstractYoutubeDownloaderError, self).__init__()
        if len(args) > 1:
            self.video_url = args[0]
            self.stderr = args[1]
        elif len(args) > 0:
            self.video_url = args[0]
        self._msg = "YoutubeDownloader generic error."
        self._short_msg = kwargs.get('msg')
        if args or 'msg' in kwargs:
            self._msg = '\n'.join(
                [_ for _ in [kwargs.get('msg', ''), getattr(self, 'stderr', '')] if _]
            )


class TokenParameterNotInVideoInfoError(Exception, AbstractYoutubeDownloaderError):
    """Token error"""

    reg = re.compile('"token" parameter not in video info for unknown reason')

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(self, video_url, stderror)
        Exception.__init__(self, self._msg)


class InvalidUrlError(Exception, AbstractYoutubeDownloaderError):
    """Invalid url error"""

    reg = re.compile(r'is not a valid URL\.')

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(
            self, video_url, stderror, msg="Invalid url '{}'.".format(video_url)
        )
        Exception.__init__(self, self._short_msg)


class UnavailableVideoError(Exception, AbstractYoutubeDownloaderError):
    """Wrong url error"""

    reg = re.compile(r'ERROR: Video unavailable')

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(
            self, video_url, stderror, msg="Unavailable video at '{}'.".format(video_url)
        )
        Exception.__init__(self, self._msg)


class TooManyRequestsError(Exception, AbstractYoutubeDownloaderError):
    """Too many requests (for youtube) to serve"""

    reg = re.compile(
        r"(?:ERROR: Unable to download webpage: HTTP Error 429: Too Many Requests|WARNING: unable to download video info webpage: HTTP Error 429)"
    )

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(
            self,
            video_url,
            stderror,
            msg="Too many requests for youtube at the moment.".format(video_url),
        )
        Exception.__init__(self, self._msg)


class CertificateVerificationError(Exception, AbstractYoutubeDownloaderError):
    """This can happen when downloading is requested from a server like scrutinizer.io\n
    ERROR: Unable to download webpage: <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get
    local issuer certificate (_ssl.c:1056)> (caused by URLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED]
    certificate verify failed: unable to get local issuer certificate (_ssl.c:1056)')))
    """

    reg = re.compile(
        r"ERROR: Unable to download webpage: <urlopen error \[SSL: CERTIFICATE_VERIFY_FAILED\]"
    )

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(
            self,
            video_url,
            stderror,
            msg="Unable to download webpage because ssl certificate verification failed:\n[SSL: CERTIFICATE_VERIFY_FAILED] certificate "
            "verify failed: unable to get local issuer certificate (_ssl.c:1056)> (caused by "
            "URLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: "
            "unable to get local issuer certificate (_ssl.c:1056)')))",
        )
        Exception.__init__(self, self._msg)


class HTTPForbiddenError(Exception, AbstractYoutubeDownloaderError):
    reg = re.compile(r"ERROR: unable to download video data: HTTP Error 403: Forbidden")

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(
            self,
            video_url,
            stderror,
            msg="HTTP 403 Forbidden for some reason.".format(video_url),
        )
        Exception.__init__(self, self._msg)


class UploaderIDExtractionError(Exception, AbstractYoutubeDownloaderError):
    reg = re.compile(r"ERROR: Unable to extract uploader id")

    def __init__(self, video_url, stderror):
        AbstractYoutubeDownloaderError.__init__(
            self,
            video_url,
            stderror,
            msg="Maybe update the youtube-dl binary/executable.".format(video_url),
        )
        Exception.__init__(self, self._msg)


1	1	import json
2	1	import logging
3	1	import re
4	1	from abc import ABC
5		from pathlib import Path
6	1	from time import sleep
7		from typing import Union
8	1	from urllib.error import URLError
9
10		from pytube import YouTube
11
12		logger = logging.getLogger(__name__)
13
14
15		class CMDYoutubeDownloader:
16		__instance = None
17
18		def __new__(cls, args, *kwargs):
19		if not cls.__instance:
20		cls.__instance = super(CMDYoutubeDownloader, cls).__new__(cls)
21		return cls.__instance
22
23		def download(self, video_url: str, directory: Union[str, Path], **kwargs) -> str:
24		return self._download(video_url, directory)
25
26		@classmethod
27		def _download(cls, video_url, output_dir, **kwargs) -> str:
28	1	# output dir where to store the stream
29	1	yt = YouTube(video_url)
30		download_parameters = dict(
31		{
32		'output_path': str(output_dir),
33		# 'filename':f'{yt.title}.mp3', # since this is an audio-only stream, the file will be mp4
34		# 'filename': f'{title}.mp3',
35	1	'filename_prefix': None,
36	1	'skip_existing': True, # Skip existing files, defaults to True
37	1	'timeout': None, # Request timeout length in seconds. Uses system default
38		'max_retries': 0, # Number of retries to attempt after socket timeout. Defaults to 0
39	1	},
40	1	**kwargs
41		)
42	1	try:
43		title: str = yt.title
44		except Exception as error:
45	1	logger.exception(error)
46		title = 'failed-to-get-title'
47
48		# # find highest quality audio stream
49		# # we currently judge quality by bitrate (higher is better)
50		best_audio_stream = yt.streams.filter(only_audio=True).order_by('bitrate')[-1]
51
52		# Download the audio stream
53		try:
54		local_file = best_audio_stream.download(**download_parameters)
55		# Catch common bug on pytube (which is not too stable yet)
56		except URLError as error:
57		logger.error(
58		"Youtube Download Error: %s",
59		json.dumps(
60	1	{
61	1	'url': str(video_url),
62	1	'title': title,
63		},
64	1	indent=4,
65		sort_keys=True,
66		),
67		)
68		raise error
69	1	logger.info(
70		"Downloaded from Youtube: %s",
71		json.dumps(
72	1	{
73		'title': title,
74		'local_file': str(local_file),
75		},
76		indent=4,
77		sort_keys=True,
78		),
79		)
80		return local_file
81
82		def download_trials(self, video_url, directory, times=10, delay=0.5, **kwargs):
83		"""Download with retries
84
85		Call this method to download a video with retries.
86
87		Note:
88		Designed for retrying when non-deterministic errors occur.
89	1
90		Args:
91		video_url (str): the youtube video url
92		directory (str): the directory to store the downloaded file
93		times (int, optional): Number of retries for non-deterministic bugs. Defaults to 10.
94		delay (float, optional): Delay between retries to no stress youtube server. Defaults to 0.5.
95
96		Raises:
97		URLError: if the download fails after all retries
98
99		Returns:
100		[type]: [description]
101		"""
102	1	i = 0
103	1	while i < times:
104		try:
105		return self._download(video_url, directory, **kwargs)
106		except URLError as error:
107	1	if 'Network is unreachable' in str(error):
108		i += 1
109		sleep(delay)
110		else:
111		raise error
112		except TooManyRequestsError as e:
113		i += 1
114		sleep(delay)
115		raise RetriesFailedError
116
117
118		class RetriesFailedError(Exception):
119	1	pass
120	1
121
122	1	class YoutubeDownloaderErrorFactory(object):
123		@staticmethod
124		def create_with_message(msg):
125		return Exception(msg)
126
127		@staticmethod
128		def create_from_stderr(stderror, video_url):
129		exception_classes = (
130		UploaderIDExtractionError,
131		TokenParameterNotInVideoInfoError,
132		InvalidUrlError,
133		UnavailableVideoError,
134		TooManyRequestsError,
135	1	CertificateVerificationError,
136		HTTPForbiddenError,
137	1	)
138		for subclass in exception_classes:
139	1	if subclass.reg.search(stderror):
140		return subclass(video_url, stderror)
141		s = "NOTE: None of the predesinged exceptions' regexs [{}] matched. Perhaps you want to derive a new subclass from AbstractYoutubeDownloaderError to account for this youtube-dl exception with string to parse <S>{}</S>'".format(
142		', '.join(['"{}"'.format(_.reg) for _ in exception_classes]), stderror
143	1	)
144		return Exception(AbstractYoutubeDownloaderError(video_url, stderror)._msg + '\n' + s)
145	1
146
147	1	#### EXCEPTIONS
148
149
150		class AbstractYoutubeDownloaderError(ABC):
151	1	def __init__(self, args, *kwargs):
152		super(AbstractYoutubeDownloaderError, self).__init__()
153	1	if len(args) > 1:
154		self.video_url = args[0]
155	1	self.stderr = args[1]
156		elif len(args) > 0:
157		self.video_url = args[0]
158		self._msg = "YoutubeDownloader generic error."
159	1	self._short_msg = kwargs.get('msg')
160		if args or 'msg' in kwargs:
161	1	self._msg = '\n'.join(
162		[_ for _ in [kwargs.get('msg', ''), getattr(self, 'stderr', '')] if _]
163	1	)
164
165
166		class TokenParameterNotInVideoInfoError(Exception, AbstractYoutubeDownloaderError):
167	1	"""Token error"""
168
169		reg = re.compile('"token" parameter not in video info for unknown reason')
170
171		def __init__(self, video_url, stderror):
172		AbstractYoutubeDownloaderError.__init__(self, video_url, stderror)
173	1	Exception.__init__(self, self._msg)
174
175	1
176		class InvalidUrlError(Exception, AbstractYoutubeDownloaderError):
177		"""Invalid url error"""
178
179		reg = re.compile(r'is not a valid URL\.')
180
181		def __init__(self, video_url, stderror):
182		AbstractYoutubeDownloaderError.__init__(
183		self, video_url, stderror, msg="Invalid url '{}'.".format(video_url)
184		)
185		Exception.__init__(self, self._short_msg)
186
187
188		class UnavailableVideoError(Exception, AbstractYoutubeDownloaderError):
189		"""Wrong url error"""
190
191		reg = re.compile(r'ERROR: Video unavailable')
192
193		def __init__(self, video_url, stderror):
194		AbstractYoutubeDownloaderError.__init__(
195		self, video_url, stderror, msg="Unavailable video at '{}'.".format(video_url)
196		)
197		Exception.__init__(self, self._msg)
198
199
200		class TooManyRequestsError(Exception, AbstractYoutubeDownloaderError):
201		"""Too many requests (for youtube) to serve"""
202
203		reg = re.compile(
204		r"(?:ERROR: Unable to download webpage: HTTP Error 429: Too Many Requests\|WARNING: unable to download video info webpage: HTTP Error 429)"
205		)
206
207		def __init__(self, video_url, stderror):
208		AbstractYoutubeDownloaderError.__init__(
209		self,
210		video_url,
211		stderror,
212		msg="Too many requests for youtube at the moment.".format(video_url),
213		)
214		Exception.__init__(self, self._msg)
215
216
217		class CertificateVerificationError(Exception, AbstractYoutubeDownloaderError):
218		"""This can happen when downloading is requested from a server like scrutinizer.io\n
219		ERROR: Unable to download webpage: <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get
220		local issuer certificate (_ssl.c:1056)> (caused by URLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED]
221		certificate verify failed: unable to get local issuer certificate (_ssl.c:1056)')))
222		"""
223
224		reg = re.compile(
225		r"ERROR: Unable to download webpage: <urlopen error \[SSL: CERTIFICATE_VERIFY_FAILED\]"
226		)
227
228		def __init__(self, video_url, stderror):
229		AbstractYoutubeDownloaderError.__init__(
230		self,
231		video_url,
232		stderror,
233		msg="Unable to download webpage because ssl certificate verification failed:\n[SSL: CERTIFICATE_VERIFY_FAILED] certificate "
234		"verify failed: unable to get local issuer certificate (_ssl.c:1056)> (caused by "
235		"URLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: "
236		"unable to get local issuer certificate (_ssl.c:1056)')))",
237		)
238		Exception.__init__(self, self._msg)
239
240
241		class HTTPForbiddenError(Exception, AbstractYoutubeDownloaderError):
242		reg = re.compile(r"ERROR: unable to download video data: HTTP Error 403: Forbidden")
243
244		def __init__(self, video_url, stderror):
245		AbstractYoutubeDownloaderError.__init__(
246		self,
247		video_url,
248		stderror,
249		msg="HTTP 403 Forbidden for some reason.".format(video_url),
250		)
251		Exception.__init__(self, self._msg)
252
253
254		class UploaderIDExtractionError(Exception, AbstractYoutubeDownloaderError):
255		reg = re.compile(r"ERROR: Unable to extract uploader id")
256
257		def __init__(self, video_url, stderror):
258		AbstractYoutubeDownloaderError.__init__(
259		self,
260		video_url,
261		stderror,
262		msg="Maybe update the youtube-dl binary/executable.".format(video_url),
263		)
264		Exception.__init__(self, self._msg)
265

boromir674 / music-album-creator

Pull Request — master (#2)

() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like