1 from hashlib
import sha256
7 from .ffmpeg
import FFmpegPostProcessor
8 from ..compat
import compat_urllib_parse_urlencode
, compat_HTTPError
9 from ..utils
import PostProcessingError
, network_exceptions
, sanitized_Request
12 class SponsorBlockPP(FFmpegPostProcessor
):
19 'intro': 'Intermission/Intro Animation',
20 'outro': 'Endcards/Credits',
21 'selfpromo': 'Unpaid/Self Promotion',
22 'interaction': 'Interaction Reminder',
23 'preview': 'Preview/Recap',
24 'music_offtopic': 'Non-Music Section'
27 def __init__(self
, downloader
, categories
=None, api
='https://sponsor.ajay.app'):
28 FFmpegPostProcessor
.__init
__(self
, downloader
)
29 self
._categories
= tuple(categories
or self
.CATEGORIES
.keys())
30 self
._API
_URL
= api
if re
.match('^https?://', api
) else 'https://' + api
33 extractor
= info
['extractor_key']
34 if extractor
not in self
.EXTRACTORS
:
35 self
.to_screen(f
'SponsorBlock is not supported for {extractor}')
38 self
.to_screen('Fetching SponsorBlock segments')
39 info
['sponsorblock_chapters'] = self
._get
_sponsor
_chapters
(info
, info
['duration'])
42 def _get_sponsor_chapters(self
, info
, duration
):
43 segments
= self
._get
_sponsor
_segments
(info
['id'], self
.EXTRACTORS
[info
['extractor_key']])
45 def duration_filter(s
):
46 start_end
= s
['segment']
47 # Ignore milliseconds difference at the start.
50 # Ignore milliseconds difference at the end.
51 # Never allow the segment to exceed the video.
52 if duration
and duration
- start_end
[1] <= 1:
53 start_end
[1] = duration
54 # SponsorBlock duration may be absent or it may deviate from the real one.
55 return s
['videoDuration'] == 0 or not duration
or abs(duration
- s
['videoDuration']) <= 1
57 duration_match
= [s
for s
in segments
if duration_filter(s
)]
58 if len(duration_match
) != len(segments
):
59 self
.report_warning('Some SponsorBlock segments are from a video of different duration, maybe from an old version of this video')
62 (start
, end
), cat
= s
['segment'], s
['category']
67 'title': self
.CATEGORIES
[cat
],
68 '_categories': [(cat
, start
, end
)]
71 sponsor_chapters
= [to_chapter(s
) for s
in duration_match
]
72 if not sponsor_chapters
:
73 self
.to_screen('No segments were found in the SponsorBlock database')
75 self
.to_screen(f
'Found {len(sponsor_chapters)} segments in the SponsorBlock database')
76 return sponsor_chapters
78 def _get_sponsor_segments(self
, video_id
, service
):
79 hash = sha256(video_id
.encode('ascii')).hexdigest()
80 # SponsorBlock API recommends using first 4 hash characters.
81 url
= f
'{self._API_URL}/api/skipSegments/{hash[:4]}?' + compat_urllib_parse_urlencode({
83 'categories': json
.dumps(self
._categories
),
85 self
.write_debug(f
'SponsorBlock query: {url}')
86 for d
in self
._get
_json
(url
):
87 if d
['videoID'] == video_id
:
91 def _get_json(self
, url
):
92 # While this is not an extractor, it behaves similar to one and
93 # so obey extractor_retries and sleep_interval_requests
94 max_retries
= self
.get_param('extractor_retries', 3)
95 sleep_interval
= self
.get_param('sleep_interval_requests') or 0
96 for retries
in itertools
.count():
98 rsp
= self
._downloader
.urlopen(sanitized_Request(url
))
99 return json
.loads(rsp
.read().decode(rsp
.info().get_param('charset') or 'utf-8'))
100 except network_exceptions
as e
:
101 if isinstance(e
, compat_HTTPError
) and e
.code
== 404:
103 if retries
< max_retries
:
104 self
.report_warning(f
'{e}. Retrying...')
105 if sleep_interval
> 0:
106 self
.to_screen(f
'Sleeping {sleep_interval} seconds ...')
107 time
.sleep(sleep_interval
)
109 raise PostProcessingError(f
'Unable to communicate with SponsorBlock API: {e}')