diff --git a/YouTubeMDBot/__init__.py b/YouTubeMDBot/__init__.py index 8a858f1..986bc7c 100755 --- a/YouTubeMDBot/__init__.py +++ b/YouTubeMDBot/__init__.py @@ -13,3 +13,31 @@ # # You should have received a copy of the GNU General Public License # along with this program. If not, see . +from .api import YouTubeAPI +from .api import YouTubeVideoData + +from .audio import FPCalc +from .audio import FFmpegOGG +from .audio import FFmpegMP3 +from .audio import FFmpegOpener +from .audio import ffmpeg_available + +from .commands import StartHandler + +from .constants import * + +from .decorators import restricted +from .decorators import send_action + +from .downloader import YouTubeDownloader + +from .errors import EmptyBodyError + +from .logging_utils import LoggingHandler +from .logging_utils import setup_logging + +from .metadata import AudioMetadata +from .metadata import MetadataIdentifier +from .metadata import YouTubeMetadataIdentifier + +from .utils import get_yt_video_id diff --git a/YouTubeMDBot/api/youtube_api.py b/YouTubeMDBot/api/youtube_api.py index 1260023..d8ff82f 100755 --- a/YouTubeMDBot/api/youtube_api.py +++ b/YouTubeMDBot/api/youtube_api.py @@ -19,8 +19,29 @@ from ..errors import EmptyBodyError -class YouTubeVideoData(object): +class YouTubeVideoData: + """ + Obtains YouTube video data and wraps it inside this class. All fields are direct + access available, so it is possible to access them directly: + - title + - id + - thumbnail + - artist + - duration + - views + - likes + - dislikes + """ + def __init__(self, data: dict, ignore_errors: bool = False): + """ + By passing a dict with the YouTube data (YouTube API v3), generate and obtain + the information available from the result. + :param data: a dictionary with the information obtained from YouTube API. + :param ignore_errors: whether to ignore or not errors (do not raise exceptions). + :raises EmptyBodyError when there is no information available and ignored + errors is False. + """ if not data.get("items"): raise EmptyBodyError("The data object has no items") self.id: str = "" @@ -77,7 +98,12 @@ def __init__(self, data: dict, ignore_errors: bool = False): self.dislikes = int(statistics["dislikeCount"]) -class YouTubeAPI(object): +class YouTubeAPI: + """ + Wrapper for the YouTube API data. Allows the developer searching for videos and, + with a given video ID, obtain its data. + """ + def __init__(self): from googleapiclient.discovery import build @@ -85,7 +111,12 @@ def __init__(self): version=YOUTUBE["api"]["version"], developerKey=YOUTUBE["key"]) - def search(self, term: str): + def search(self, term: str) -> dict: + """ + Searchs for a video with the specified term. + :param term: the search term. + :return: dict with YouTube data - can be wrapped inside "YouTubeVideoData" class. + """ return self.__youtube.search().list( q=term, type="video", @@ -95,12 +126,18 @@ def search(self, term: str): @staticmethod def video_details(video_id: str) -> YouTubeVideoData: + """ + Generates a "YouTubeVideoData" object wrapper with the video ID information. + :param video_id: YouTube video ID. + :return: YouTubeVideoData object with the available metadata. + """ try: import ujson as json except ImportError: import json from urllib.request import urlopen - api_url = YOUTUBE["endpoint"].format(video_id, YOUTUBE["key"]) + youtube_information = YOUTUBE.copy() + api_url = youtube_information["endpoint"].format(video_id, YOUTUBE["key"]) data = urlopen(url=api_url) - return YouTubeVideoData(data=json.loads(data.read())) + return YouTubeVideoData(data=json.loads(data.read()), ignore_errors=True) diff --git a/YouTubeMDBot/audio/ffmpeg.py b/YouTubeMDBot/audio/ffmpeg.py index 53d199f..f94c5ae 100755 --- a/YouTubeMDBot/audio/ffmpeg.py +++ b/YouTubeMDBot/audio/ffmpeg.py @@ -24,6 +24,10 @@ def ffmpeg_available() -> bool: + """ + Checks if "ffmpeg" is installed or not. + :return: True if installed, else False + """ try: proc = Popen(["ffmpeg", "-version"], stdout=PIPE, @@ -36,46 +40,107 @@ def ffmpeg_available() -> bool: class FFmpeg(ABC): + """ + Base abstract class for the FFmpeg operators. All classes that works with FFmpeg + must inherit from this class in order to maintain readability and code optimization. + + Allows execution of the ffmpeg command by using the subprocess module. Everything + is working with PIPEs, so there is no directly discs operations (everything is + loaded and working with RAM). + """ + def __init__(self, data: bytes, command: List[str] = None): + """ + Creates the class by passing the data which will be processed and the command ( + by default, None). + :param data: audio data that will be processed. + :param command: the ffmpeg command. + """ self._data = data self.__command = command self.__out = None self.__err = None def process(self) -> int: + """ + Runs the ffmpeg command in a separate process and pipes both stdout and stderr. + :return: the return code of the operation ('0' if everything is OK, > 0 if not). + """ proc = Popen(self.__command, stdout=PIPE, stderr=PIPE, stdin=PIPE) self.__out, self.__err = proc.communicate(self._data) return proc.returncode def get_command(self) -> List[str]: + """ + Get the command for editing. + :return: List[str] with the command - as this is a pointer, all editions done + to the list are directly changing the self object. + """ return self.__command def set_command(self, command: List[str]): + """ + Sets the new list, overriding every old implementation. + :param command: the new command. + """ self.__command = command def get_output(self) -> bytes: + """ + Gets the stdout of the process. + :return: bytes with the command output. + """ return self.__out def get_extra(self) -> bytes: + """ + Gets the stderr of the process. + :return: bytes with extra information. + """ return self.__err class FFmpegOpener(FFmpeg): + """ + Opens and produces and audio in PWM mode. + """ + def __init__(self, data: bytes): super().__init__(data=data, command=FFMPEG_OPENER.copy()) class FFmpegExporter(FFmpeg): + """ + Base class for the exporter options available in ffmpeg. + All classes that are developed for converting audio files must inherit from this + class and implement the "convert" method. + """ + def __init__(self, data: bytes, bitrate: str = None): + """ + Generates a new instance of the class. + :param data: the audio data. + :param bitrate: the new bitrate of the audio data, or None for keeping its + default. + """ super().__init__(data=data, command=FFMPEG_CONVERTER.copy()) self._bitrate = bitrate @abstractmethod def convert(self) -> int: + """ + Converts the audio to the desired format. + :return: the operation result code. + :raises NotImplementedError when trying to access this method directly on super + class. + """ raise NotImplementedError class FFmpegMP3(FFmpegExporter): + """ + Exports audio data to MP3 format. + """ def convert(self) -> int: command = super().get_command() if self._bitrate: @@ -90,6 +155,9 @@ def convert(self) -> int: class FFmpegOGG(FFmpegExporter): + """ + Exports audio data to OGG format. + """ def convert(self) -> int: command = super().get_command() if self._bitrate: diff --git a/YouTubeMDBot/audio/fpcalc.py b/YouTubeMDBot/audio/fpcalc.py index d4833fc..5629b12 100755 --- a/YouTubeMDBot/audio/fpcalc.py +++ b/YouTubeMDBot/audio/fpcalc.py @@ -21,6 +21,10 @@ def is_fpcalc_available() -> bool: + """ + Checks if ffmpeg is installed in the system. + :return: True if available, else False. + """ try: proc = Popen(["fpcalc", "-v"], stdout=PIPE, stderr=PIPE) except OSError: @@ -29,8 +33,16 @@ def is_fpcalc_available() -> bool: proc.wait() -class FPCalc(object): +class FPCalc: + """ + Calculates audio fingerprint by passing the audio bytes. + It operates with pipes so no file is created. + """ def __init__(self, audio: bytes): + """ + Creates the FPCalc object. + :param audio: the audio bytes. + """ fpcalc = Popen(FPCALC, stdout=PIPE, stdin=PIPE) out, _ = fpcalc.communicate(audio) res = out.decode("utf-8") @@ -44,7 +56,15 @@ def __init__(self, audio: bytes): self.__fp: str = str(fingerprint.group(0)) def duration(self) -> int: + """ + Obtains the audio duration in seconds. + :return: duration in seconds. + """ return self.__duration def fingerprint(self) -> str: + """ + Obtains the audio fingerprint. + :return: fingerprint in seconds. + """ return self.__fp diff --git a/YouTubeMDBot/constants/__init__.py b/YouTubeMDBot/constants/__init__.py index 60906a2..12af4b1 100755 --- a/YouTubeMDBot/constants/__init__.py +++ b/YouTubeMDBot/constants/__init__.py @@ -17,5 +17,6 @@ from ..constants.app_constants import FPCALC from ..constants.app_constants import YDL_CLI_OPTIONS from ..constants.app_constants import YOUTUBE +from ..constants.app_constants import PROGRAM_ARGS from ..constants.app_constants import FFMPEG_OPENER from ..constants.app_constants import FFMPEG_CONVERTER diff --git a/YouTubeMDBot/constants/app_constants.py b/YouTubeMDBot/constants/app_constants.py index 5564c27..1ef1e70 100755 --- a/YouTubeMDBot/constants/app_constants.py +++ b/YouTubeMDBot/constants/app_constants.py @@ -14,7 +14,9 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . import os +import sys +PROGRAM_ARGS = sys.argv # YouTube DL options YDL_CLI_OPTIONS = ["youtube-dl", "--format", "bestaudio[ext=m4a]", "--quiet", "--output", "-"] diff --git a/YouTubeMDBot/decorators/decorators.py b/YouTubeMDBot/decorators/decorators.py index ab6f59e..b3463c4 100755 --- a/YouTubeMDBot/decorators/decorators.py +++ b/YouTubeMDBot/decorators/decorators.py @@ -15,10 +15,7 @@ # along with this program. If not, see . from functools import wraps -from .. import PROGRAM_ARGS - - -# logging = LoggingHandler() +from ..constants import PROGRAM_ARGS def send_action(action): diff --git a/YouTubeMDBot/downloader/youtube_downloader.py b/YouTubeMDBot/downloader/youtube_downloader.py index 6139205..69c8edb 100755 --- a/YouTubeMDBot/downloader/youtube_downloader.py +++ b/YouTubeMDBot/downloader/youtube_downloader.py @@ -19,13 +19,24 @@ from ..constants.app_constants import YDL_CLI_OPTIONS -class YouTubeDownloader(object): +class YouTubeDownloader: + """ + Download a YouTube video directly into memory. + """ def __init__(self, url: str): + """ + Creates the YouTubeDownloader object. Call "download" for obtaining the video. + :param url: the video URL. + """ self.__url: str = url self.__options: list = YDL_CLI_OPTIONS.copy() self.__options.append(self.__url) def download(self) -> Tuple[BytesIO, bytes]: + """ + Downloads the YouTube video directly into memory by using pipes. + :return: a tuple with "BytesIO" and "bytes". + """ import subprocess proc = subprocess.Popen(self.__options, @@ -40,4 +51,8 @@ def download(self) -> Tuple[BytesIO, bytes]: str(stderr.decode("utf-8"))) def get_url(self) -> str: + """ + Obtains the video URL. + :return: str with the URL. + """ return self.__url diff --git a/YouTubeMDBot/errors/EmptyBodyError.py b/YouTubeMDBot/errors/EmptyBodyError.py index 4706bda..a2e8f81 100755 --- a/YouTubeMDBot/errors/EmptyBodyError.py +++ b/YouTubeMDBot/errors/EmptyBodyError.py @@ -16,4 +16,8 @@ class EmptyBodyError(Exception): + """ + Raises an exception when the body of the json data is empty (e.g.: there is no + video information) + """ pass diff --git a/YouTubeMDBot/errors/InvalidCredentialsError.py b/YouTubeMDBot/errors/InvalidCredentialsError.py deleted file mode 100755 index 9b3d8c3..0000000 --- a/YouTubeMDBot/errors/InvalidCredentialsError.py +++ /dev/null @@ -1,19 +0,0 @@ -# YouTubeMDBot -# Copyright (C) 2019 - Javinator9889 -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - - -class InvalidCredentialsError(Exception): - pass diff --git a/YouTubeMDBot/errors/NoMatchError.py b/YouTubeMDBot/errors/NoMatchError.py deleted file mode 100755 index 69c5c3f..0000000 --- a/YouTubeMDBot/errors/NoMatchError.py +++ /dev/null @@ -1,20 +0,0 @@ -# YouTubeMDBot -# Copyright (C) 2019 - Javinator9889 -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - - -class NoMatchError(Exception): - """Raises an error when there is no match available""" - pass diff --git a/YouTubeMDBot/errors/ProcessorError.py b/YouTubeMDBot/errors/ProcessorError.py deleted file mode 100644 index 703bd49..0000000 --- a/YouTubeMDBot/errors/ProcessorError.py +++ /dev/null @@ -1,20 +0,0 @@ -# YouTubeMDBot -# Copyright (C) 2019 - Javinator9889 -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - - -class ProcessorError(Exception): - """Raises an exception when FFmpeg processing fails""" - pass diff --git a/YouTubeMDBot/errors/__init__.py b/YouTubeMDBot/errors/__init__.py index d9dd503..98ce0a2 100755 --- a/YouTubeMDBot/errors/__init__.py +++ b/YouTubeMDBot/errors/__init__.py @@ -14,5 +14,3 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . from ..errors.EmptyBodyError import EmptyBodyError -from ..errors.NoMatchError import NoMatchError -from ..errors.ProcessorError import ProcessorError diff --git a/YouTubeMDBot/logging_utils/utils.py b/YouTubeMDBot/logging_utils/utils.py index 309a620..3b6ac76 100755 --- a/YouTubeMDBot/logging_utils/utils.py +++ b/YouTubeMDBot/logging_utils/utils.py @@ -17,6 +17,10 @@ def cleanup_old_logs(log_file: str): + """ + Cleans-up the old log files. + :param log_file: log filename that must be cleaned. + """ import tarfile import os @@ -33,6 +37,14 @@ def cleanup_old_logs(log_file: str): def setup_logging(logger_name: str, log_file: str, level=logging.DEBUG, formatter: str = "%(process)d - %(asctime)s | [%(levelname)s]: %(message)s"): + """ + Creates a new logging which can log to stdout or file. + :param logger_name: the logger name. + :param log_file: log filename. + :param level: logging level. + :param formatter: the logging formatter. + :return: the logging file handler. + """ from os import path from os import makedirs @@ -53,8 +65,12 @@ def setup_logging(logger_name: str, log_file: str, level=logging.DEBUG, return logging_file_handler -class LoggingHandler(object): - class __LoggingHandler(object): +class LoggingHandler: + """ + LoggingHandler singleton class that outputs to multiple logging instances. + """ + + class __LoggingHandler: def __init__(self, logs: list): self.__logs = logs @@ -84,10 +100,18 @@ def get_loggers(self) -> list: __instance = None def __new__(cls, *args, **kwargs): + """ + Generates a new instance. + :param args: not used. + :param kwargs: "logs" is a list instance that must be provided the first time + this class is created. + :return: the LoggingHandler instance. + """ if not LoggingHandler.__instance: logs = kwargs.get("logs") if not logs or len(logs) == 0: - raise AttributeError("At least kwarg \"log\" (a list of the loggers) must be provided") + raise AttributeError( + "At least kwarg \"log\" (a list of the loggers) must be provided") LoggingHandler.__instance = LoggingHandler.__LoggingHandler(logs) return LoggingHandler.__instance @@ -98,19 +122,43 @@ def __setattr__(self, key, value): return setattr(self.__instance, key, value) def debug(self, msg): + """ + Debugs to loggers + :param msg: message to debug + """ self.__instance.debug(msg) def info(self, msg): + """ + Info to loggers + :param msg: message to info + """ self.__instance.info(msg) def error(self, msg): + """ + Error to loggers + :param msg: message to error + """ self.__instance.error(msg) def warning(self, msg): + """ + Warns to loggers + :param msg: message to warn + """ self.__instance.warning(msg) def critical(self, msg): + """ + Critical to loggers + :param msg: message to critical + """ self.__instance.critical(msg) def get_loggers(self) -> list: + """ + Obtains the list of loggers. + :return: the list of loggers. + """ return self.__instance.get_loggers() diff --git a/YouTubeMDBot/metadata/AudioMetadata.py b/YouTubeMDBot/metadata/AudioMetadata.py index 179dc29..77754e1 100755 --- a/YouTubeMDBot/metadata/AudioMetadata.py +++ b/YouTubeMDBot/metadata/AudioMetadata.py @@ -19,27 +19,60 @@ class AudioMetadata: + """ + Wrapper class for setting the audio metadata to the downloaded YouTube video + object. By using this class, it is possible to set the required information by + using mutagen without having to remember the metadata keys. + """ def __init__(self, audio: BytesIO): + """ + Generates a new instance. + :param audio: the audio metadata, in BytesIO, in MP4 format. + """ self._audio = MP4(audio) self._data = audio def set_title(self, title: str): + """ + Sets the audio title. + :param title: the audio title. + """ self._audio[u"\xa9nam"] = title def set_artist(self, artist: str): + """ + Sets the audio artist. + :param artist: the audio artist. + """ self._audio[u"\xa9ART"] = artist def set_album(self, album: str): + """ + Sets the audio album. + :param album: the audio album + """ self._audio[u"\xa9alb"] = album def set_extras(self, extras: list): + """ + Sets the audio extras. + :param extras: a list of extras that will be added to the audio information. + """ self._audio[u"\xa9cmt"] = '; '.join(map(str, extras)) def set_cover(self, cover: bytes): + """ + Sets the audio cover. + :param cover: the audio cover. + """ mp4_cover = MP4Cover(cover, MP4Cover.FORMAT_JPEG) self._audio[u"covr"] = [mp4_cover] def save(self) -> BytesIO: + """ + Saves the new metadata into the audio file object. + :return: the audio file object with the new metadata. + """ self._data.seek(0) self._audio.save(self._data) return self._data diff --git a/YouTubeMDBot/metadata/MetadataIdentifier.py b/YouTubeMDBot/metadata/MetadataIdentifier.py index c724232..d3396aa 100755 --- a/YouTubeMDBot/metadata/MetadataIdentifier.py +++ b/YouTubeMDBot/metadata/MetadataIdentifier.py @@ -13,13 +13,12 @@ # # You should have received a copy of the GNU General Public License # along with this program. If not, see . -import acoustid -import musicbrainzngs - try: import ujson as json except ImportError: import json +import acoustid +import musicbrainzngs from ..audio import FPCalc from ..api import YouTubeAPI @@ -28,8 +27,30 @@ from ..downloader import YouTubeDownloader -class MetadataIdentifier(object): +class MetadataIdentifier: + """ + Base identifier class. By using the audio data, calculates and generates a + fingerprint for searching across the MusicBrainz database. + + Once the audio has been identified, the available params and information are: + - audio (bytes) + - result (json) + - artist (str) + - title (str) + - release_id (str) + - recording_id (str) + - score (float) + - cover (bytes) + - album (str) + - duration (int) + - youtube_data (bool) + - youtube_id (str) + """ def __init__(self, audio: bytes): + """ + Generates a new instance of the MetadataIdentifier class. + :param audio: the audio data, in bytes. + """ self.audio = audio self.result: json = None self.artist: str = "" @@ -45,6 +66,12 @@ def __init__(self, audio: bytes): @staticmethod def _is_valid_result(data: json) -> bool: + """ + Checks whether the obtained result, in json, is valid or not, by checking for + certain keys that must exist. + :param data: the result in json. + :return: 'True' if the result is valid, else 'False'. + """ if "results" not in data: return False elif data["status"] != "ok": @@ -58,6 +85,12 @@ def _is_valid_result(data: json) -> bool: return True def identify_audio(self) -> bool: + """ + Tries to identify the audio by using the audio fingerprint. If the audio has + been successfully identified, then obtains all the data related to it. + :return: 'True' if the result is valid (the audio was correctly identified), + else 'False'. + """ fingerprint = FPCalc(self.audio) data: json = acoustid.lookup(apikey=ACOUSTID_KEY, fingerprint=fingerprint.fingerprint(), @@ -91,7 +124,35 @@ def identify_audio(self) -> bool: class YouTubeMetadataIdentifier(MetadataIdentifier): + """ + Identifies YouTube metadata by using MusicBrainz database and YouTube metadata. If + the first identification (MusicBrainz) fails, then fallback to YouTube + identification if the "downloader" was provided. + + Once the audio has been identified, the available params and information are: + - audio (bytes) + - result (json) + - artist (str) + - title (str) + - release_id (str) + - recording_id (str) + - score (float) + - cover (bytes) + - album (str) + - duration (int) + - youtube_data (bool) + - youtube_id (str) + + If "youtube_data" is True, then only audio, title, artist, duration, cover and + youtube_id are available. + """ def __init__(self, audio: bytes, downloader: YouTubeDownloader = None): + """ + Generates a new instance of the MetadataIdentifier class. + :param audio: the audio data, in bytes. + :param downloader: a downloader object, for obtaining the video information if + MusicBrainz fails. + """ super().__init__(audio) self._downloader = downloader @@ -111,5 +172,4 @@ def identify_audio(self) -> bool: self.youtube_data = True valid = True - return valid diff --git a/YouTubeMDBot/utils/__init__.py b/YouTubeMDBot/utils/__init__.py index 1315c64..9fe1f3e 100755 --- a/YouTubeMDBot/utils/__init__.py +++ b/YouTubeMDBot/utils/__init__.py @@ -13,5 +13,4 @@ # # You should have received a copy of the GNU General Public License # along with this program. If not, see . -from ..utils import youtube_utils -from ..utils.timeout import timeout +from ..utils.youtube_utils import get_yt_video_id diff --git a/YouTubeMDBot/utils/timeout.py b/YouTubeMDBot/utils/timeout.py deleted file mode 100755 index 4ed87ae..0000000 --- a/YouTubeMDBot/utils/timeout.py +++ /dev/null @@ -1,56 +0,0 @@ -# YouTubeMDBot -# Copyright (C) 2019 - Javinator9889 -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -import signal - - -class timeout: - def __init__(self, secs: int): - self.__secs = secs - - def _raise_timeout(self, signum, frame): - raise TimeoutError("Function timeout! - {0}s".format(self.__secs)) - - def __enter__(self): - if self.__secs <= 0: - self._raise_timeout(0, 0) - - signal.signal(signal.SIGALRM, self._raise_timeout) - signal.alarm(self.__secs) - yield - - def __exit__(self, exc_type, exc_val, exc_tb): - signal.signal(signal.SIGALRM, signal.SIG_IGN) - return exc_val is not None - - -'''@contextmanager -def timeout(secs: int): - def raise_timeout(signum=None, frame=None): - raise TimeoutError("Function timeout! - {0}s".format(secs)) - - if secs <= 0: - secs = 0 - raise_timeout() - - signal.signal(signalnum=signal.SIGALRM, handler=raise_timeout) - signal.alarm(secs) - - try: - yield - except TimeoutError: - pass - finally: - signal.signal(signalnum=signal.SIGALRM, handler=signal.SIG_IGN)'''