Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Add Subtitulamos provider integration - Tweaks and add more tests
Browse files Browse the repository at this point in the history
Nyaran committed Oct 7, 2024
1 parent 23d9e5d commit 2314ba1
Showing 6 changed files with 3,150 additions and 4,871 deletions.
109 changes: 63 additions & 46 deletions subliminal/providers/subtitulamos.py
Original file line number Diff line number Diff line change
@@ -76,10 +76,6 @@ def get_matches(self, video: Video) -> set[str]:
},
)

# resolution
if video.resolution and self.release_group and video.resolution in self.release_group.lower():
matches.add('resolution')

# other properties
matches |= guess_matches(video, guessit(self.release_group), partial=True)

@@ -140,24 +136,10 @@ def _read_series(self, series_url: str) -> ParserBeautifulSoup:
r = self._session_request(self.server_url + series_url, headers={'Referer': self.server_url}, timeout=10)
return ParserBeautifulSoup(r.content, ['lxml', 'html.parser'])

def _get_episode_url(self, series_id: str, season: int, episode: int) -> str | None:
"""Provides the URL for a specific episode of the series."""
series_content = self._read_series(f'/shows/{series_id}')

for season_element in series_content.select('#season-choices a.choice'):
if season == int(season_element.get_text()):
if 'selected' not in (list[str], season_element.get('class', [])):
series_content = self._read_series(cast(str, season_element.get('href', '')))
break
return None

for episode_element in series_content.select('#episode-choices a.choice'):
if episode == int(episode_element.get_text()):
return cast(str, episode_element.get('href', ''))
return None

@region.cache_on_arguments(expiration_time=SHOW_EXPIRATION_TIME)
def _search_url_titles(self, series: str, season: int, episode: int, year: int | None = None) -> str | None:
def _read_episode_page(
self, series: str, season: int, episode: int, year: int | None = None
) -> tuple[ParserBeautifulSoup, str]:
"""Search the URL titles by kind for the given `title`, `season` and `episode`.
:param str series: Series to search for.
@@ -174,26 +156,40 @@ def _search_url_titles(self, series: str, season: int, episode: int, year: int |
if len(series_response) == 0:
series_response = self._query_search(series)

episode_url = self._get_episode_url(series_response[0]['show_id'], season, episode)
"""Provides the URL for a specific episode of the series."""
page_content = self._read_series(f'/shows/{series_response[0]['show_id']}')

return self.server_url + episode_url if episode_url else None
# Select season
season_element = next(
(el for el in page_content.select('#season-choices a.choice') if str(season) in el.text), None
)
if season_element is None:
raise SeasonEpisodeNotExists

def query(
if 'selected' not in (list[str], season_element.get('class', [])):
page_content = self._read_series(cast(str, season_element.get('href', '')))

# Select episode
episode_element = next(
(el for el in page_content.select('#episode-choices a.choice') if str(episode) in el.text), None
)
if episode_element is None:
raise SeasonEpisodeNotExists

episode_url = cast(str, episode_element.get('href', ''))
if 'selected' not in (list[str], episode_element.get('class', [])):
page_content = self._read_series(episode_url)

# logger.error('No episode url found for %s, season %d, episode %d', series, season, episode)

return page_content, episode_url

def _query_provider(
self, series: str | None = None, season: int | None = None, episode: int | None = None, year: int | None = None
) -> list[SubtitulamosSubtitle]:
"""Query the provider for subtitles."""
if not self.session:
raise NotInitializedProviderError

# get the episode url
episode_url = self._search_url_titles(series, season, episode, year)
if episode_url is None:
logger.error('No episode url found for %s, season %d, episode %d', series, season, episode)
return []

r = self.session.get(episode_url, headers={'Referer': self.server_url}, timeout=10)
r.raise_for_status()
soup = ParserBeautifulSoup(r.content, ['lxml', 'html.parser'])
# get the episode page content
soup, episode_url = self._read_episode_page(series, season, episode, year)

# get episode title
title = soup.select('#episode-name h3')[0].get_text().strip().lower()
@@ -227,22 +223,31 @@ def query(
# read the subtitle url
subtitle_url = self.server_url + cast(str, sub.parent.get('href', ''))
subtitle = SubtitulamosSubtitle(
language,
hearing_impaired,
episode_url,
series,
season,
episode,
title,
year,
release_group,
subtitle_url,
language=language,
hearing_impaired=hearing_impaired,
page_link=self.server_url + episode_url,
series=series,
season=season,
episode=episode,
title=title,
year=year,
release_group=release_group,
download_link=subtitle_url,
)
logger.debug('Found subtitle %r', subtitle)
subtitles.append(subtitle)

return subtitles

def query(
self, series: str | None = None, season: int | None = None, episode: int | None = None, year: int | None = None
) -> list[SubtitulamosSubtitle]:
"""Query the provider for subtitles."""
try:
return self._query_provider(series, season, episode, year)
except SeasonEpisodeNotExists:
return []

def list_subtitles(self, video: Video, languages: Set[Language]) -> list[SubtitulamosSubtitle]:
"""List all the subtitles for the video."""
if not isinstance(video, Episode):
@@ -263,3 +268,15 @@ def download_subtitle(self, subtitle: SubtitulamosSubtitle) -> None:
r.raise_for_status()

subtitle.content = fix_line_ending(r.content)


class SubtitulamosError(ProviderError):
"""Base class for non-generic :class:`SubtitulamosProvider` exceptions."""

pass


class SeasonEpisodeNotExists(SubtitulamosError):
"""Exception raised when the season and/or the episode does not exist on provider."""

pass
4,880 changes: 57 additions & 4,823 deletions tests/cassettes/subtitulamos/test_download_subtitle.yaml

Large diffs are not rendered by default.

2,354 changes: 2,354 additions & 0 deletions tests/cassettes/subtitulamos/test_download_subtitle_last_season.yaml

Large diffs are not rendered by default.

608 changes: 608 additions & 0 deletions tests/cassettes/subtitulamos/test_download_subtitle_not_exist.yaml

Large diffs are not rendered by default.

8 changes: 7 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
@@ -129,7 +129,6 @@ def episodes() -> dict[str, Episode]:
audio_codec='Dolby Digital',
imdb_id='tt6674448',
size=505152010,
hashes={},
),
'got_s03e10': Episode(
os.path.join(
@@ -454,6 +453,13 @@ def episodes() -> dict[str, Episode]:
1,
1,
),
'dw_s13e03': Episode(
'Doctor.Who.2005.S13E03.Chapter.Three.Once.Upon.Time.1080p.AMZN.WEB-DL.DDP5.1.H.264-NOSIVID.mkv',
'Doctor Who',
13,
3,
year=2005,
),
}


62 changes: 61 additions & 1 deletion tests/providers/test_subtitulamos.py
Original file line number Diff line number Diff line change
@@ -2,7 +2,8 @@

import pytest
from babelfish import Language, language_converters # type: ignore[import-untyped]
from subliminal.providers.subtitulamos import SubtitulamosProvider
from subliminal.exceptions import NotInitializedProviderError
from subliminal.providers.subtitulamos import SubtitulamosProvider, SubtitulamosSubtitle
from vcr import VCR # type: ignore[import-untyped]

vcr = VCR(
@@ -32,6 +33,13 @@ def test_logout():
assert provider.session is None


@pytest.mark.integration()
def test_logout_without_initialization():
provider = SubtitulamosProvider()
with pytest.raises(NotInitializedProviderError):
provider.terminate()


@pytest.mark.integration()
@vcr.use_cassette
def test_download_subtitle(episodes):
@@ -47,6 +55,41 @@ def test_download_subtitle(episodes):
assert subtitle.is_valid() is True


@pytest.mark.integration()
@vcr.use_cassette
def test_download_subtitle_last_season(episodes):
video = episodes['dw_s13e03']
languages = {Language('eng'), Language('spa')}
with SubtitulamosProvider() as provider:
subtitles = provider.list_subtitles(video, languages)
assert len(subtitles) >= 1
subtitle = subtitles[0]

provider.download_subtitle(subtitle)
assert subtitle.content is not None
assert subtitle.is_valid() is True


@pytest.mark.integration()
@vcr.use_cassette
def test_download_subtitle_not_exist(episodes):
video = episodes['bbt_s07e05']
languages = {Language('eng'), Language('spa')}
with SubtitulamosProvider() as provider:
subtitles = provider.list_subtitles(video, languages)
assert len(subtitles) == 0


@pytest.mark.integration()
def test_download_subtitle_without_initialization(episodes):
video = episodes['bbt_s11e16']
languages = {Language('eng'), Language('spa')}

provider = SubtitulamosProvider()
with pytest.raises(NotInitializedProviderError):
provider.list_subtitles(video, languages)


@pytest.mark.converter()
def test_converter_convert_alpha3():
assert language_converters['subtitulamos'].convert('cat') == 'Català'
@@ -82,3 +125,20 @@ def test_converter_reverse_country():
@pytest.mark.converter()
def test_converter_reverse_name_converter():
assert language_converters['subtitulamos'].reverse('French') == ('fra', None, None)


def test_get_matches_episode(episodes):
subtitle = SubtitulamosSubtitle(
language=Language('spa'),
hearing_impaired=True,
page_link=None,
series='The Big Bang Theory',
season=11,
episode=16,
title='the neonatal nomenclature',
year=2007,
release_group='AVS/SVA',
)

matches = subtitle.get_matches(episodes['bbt_s11e16'])
assert matches == {'release_group', 'series', 'year', 'country', 'episode', 'season', 'title'}

0 comments on commit 2314ba1

Please sign in to comment.