diff --git a/development/playground.py b/development/playground.py index 744c61d..03f08d8 100644 --- a/development/playground.py +++ b/development/playground.py @@ -1,9 +1,9 @@ -from python_sponsorblock import SponsorBlock, _get_video_id +from python_sponsorblock import SponsorBlock if __name__ == "__main__": - sb = SponsorBlock() + sb = SponsorBlock(silent=True) - print(_get_video_id("dksaödfksöldkföslkafdsölfkas")) + print(sb.get_segments("https://yt.artemislena.eu/watch?v=w5GmDRW975g")) diff --git a/python_sponsorblock/__init__.py b/python_sponsorblock/__init__.py index 8388ece..e6bd7eb 100644 --- a/python_sponsorblock/__init__.py +++ b/python_sponsorblock/__init__.py @@ -1,12 +1,38 @@ import requests import logging -from urllib.parse import urlparse, urlunparse, parse_qs +from urllib.parse import urlparse, urlunparse, parse_qs, urlencode import re -from typing import Optional, List +from typing import Optional, List, Dict, Union, Any, Callable +import json +from functools import wraps from .exceptions import SponsorBlockError, SponsorBlockIdNotFoundError from .constants import Segment + +def error_handling(default: Any) -> Callable: + def _decorator(func: Callable) -> Callable: + @wraps(func) + def _wrapper(self, *args, **kwargs) -> Any: + nonlocal default + + try: + return func(self, *args, **kwargs) + except SponsorBlockError as e: + if not self.silent: + raise e + + if self._requests_logging_exists and isinstance(e, SponsorBlockConnectionError): + return default + + self.logger.error(repr(e)) + return default + + return _wrapper + + return _decorator + + class SponsorBlock: def __init__(self, session: requests.Session = None, base_url: str = "https://sponsor.ajay.app", silent: bool = False, _requests_logging_exists: bool = False): self.base_url: str = base_url @@ -36,26 +62,43 @@ class SponsorBlock: else: return query_stuff["v"][0] - def _request(self, method: str, endpoint: str) -> Optional[requests.Response]: + def _request(self, method: str, endpoint: str) -> Union[List, Dict]: error_message = "" url = self.base_url + endpoint r: requests.Response = None try: - r = self.session.request(method="GET", url=url) + r = self.session.request(method=method, url=url) except requests.exceptions.Timeout: error_message = f"Request timed out at \"{url}\"" except requests.exceptions.ConnectionError: error_message = f"Couldn't connect to \"{url}\"" if error_message != "": - if not self._requests_logging_exists: - self.logger.error(error_message) - if not self.silent: - raise exceptions.SponsorBlockConnectionError(error_message) + raise exceptions.SponsorBlockConnectionError(error_message) - return r + if r.status_code == 400: + self.logger.warning(f"{url} returned 400, meaning I did something wrong.") - def get_segments(video: str) -> List[Segment]: - video_id = _get_video_id - r: List[Segment] = [] + data = {} + try: + data = r.json() + except json.JSONDecodeError: + raise exceptions.SponsorBlockConnectionError(f"{r.content} is invalid json.") + + return data + + @error_handling(default=[]) + def get_segments(self, video: str) -> List[Segment]: + video_id = self._get_video_id(video) + if video_id is None: + return [] + + # build query parameters + query = { + "videoID": video_id + } + + print(query) + r = self._request(method="GET", endpoint="/api/skipSegments?" + urlencode(query)) + return [constants.Segment(**d) for d in r] diff --git a/python_sponsorblock/exceptions.py b/python_sponsorblock/exceptions.py index 538bc68..b6a11df 100644 --- a/python_sponsorblock/exceptions.py +++ b/python_sponsorblock/exceptions.py @@ -1,7 +1,6 @@ class SponsorBlockError(Exception): pass - class SponsorBlockIdNotFoundError(SponsorBlockError): pass