import os
from typing import Any, Callable, Dict, List, Optional
import requests
import xbmc
import xbmcgui
from lib.clients.subtitle.utils import (
get_language_code,
language_code_to_name,
slugify_title,
)
from lib.utils.general.utils import USER_AGENT_HEADER
from lib.utils.kodi.utils import (
ADDON_PROFILE_PATH,
get_setting,
kodilog,
translation,
)
class OpenSubtitleStremioClient:
def __init__(self, notification: Callable[[str], None]):
self.notification = notification
self.base_url = get_setting("stremio_sub_addon_host")
def _build_url(
self, mode: str, imdb_id: str, season: Optional[int], episode: Optional[int]
) -> str:
if mode == "tv":
return f"{self.base_url}subtitles/series/{imdb_id}:{season}:{episode}.json"
return f"{self.base_url}subtitles/movie/{imdb_id}.json"
def _fetch_subtitles_data(
self,
mode: str,
imdb_id: str,
season: Optional[int] = None,
episode: Optional[int] = None,
) -> Optional[List[Dict[str, Any]]]:
url = self._build_url(mode, imdb_id, season, episode)
try:
response = requests.get(url)
if response.status_code != 200:
self.notification(f"Failed to fetch subtitles, status code {response.status_code}")
return None
data = response.json()
kodilog(f"OpenSubtitles Subtitles Response: {data}", level=xbmc.LOGDEBUG)
return data.get("subtitles", [])
except Exception as e:
self.notification(f"Failed to fetch subtitles: {e}")
return None
def get_subtitles(
self,
mode: str,
imdb_id: str,
season: Optional[int] = None,
episode: Optional[int] = None,
auto_select: bool = False,
) -> Optional[List[Dict[str, Any]]]:
subtitles = self._fetch_subtitles_data(mode, imdb_id, season, episode)
if not subtitles:
return
sub_language = get_setting("subtitle_language")
auto_subtitle_download = get_setting("auto_subtitle_download")
if auto_subtitle_download:
filtered = [s for s in subtitles if s.get("lang") == get_language_code(sub_language)]
if filtered:
return filtered
return []
items = [
xbmcgui.ListItem(label=translation(90665) % i, label2=language_code_to_name(s["lang"]))
for i, s in enumerate(subtitles)
]
dialog = xbmcgui.Dialog()
selected_indices = dialog.multiselect(
translation(90256),
items,
useDetails=True,
)
if selected_indices is None:
return []
return [subtitles[i] for i in selected_indices]
def download_subtitles_batch(
self,
subtitles: List[Dict[str, Any]],
imdb_id: str,
title: str,
season: Optional[int] = None,
episode: Optional[int] = None,
folder_path: Optional[str] = None,
) -> List[str]:
file_paths = []
for idx, subtitle in enumerate(subtitles):
try:
file_path = self.download_subtitle(
subtitle, idx, imdb_id, title, season, episode, folder_path
)
if file_path:
file_paths.append(file_path)
except Exception as e:
kodilog(f"Failed to download subtitle: {e}")
continue
return file_paths
def download_subtitle(
self,
subtitle: Dict[str, Any],
index: int,
imdb_id: str,
title: str,
season: Optional[int] = None,
episode: Optional[int] = None,
folder_path: Optional[str] = None,
) -> Optional[str]:
url = subtitle.get("url", "")
lang = subtitle.get("lang")
lang_name = language_code_to_name(lang)
title = slugify_title(title)
if folder_path:
file_path = os.path.join(folder_path, f"Subtitle No.{index}.{title}.{lang_name}.srt")
else:
base_path = os.path.join(ADDON_PROFILE_PATH, "Subtitles", imdb_id)
if season and episode:
file_path = os.path.join(
base_path,
str(season),
str(episode),
f"Subtitle No.{index}.{title}.S{season}E{episode}.{lang_name}.srt",
)
elif season:
file_path = os.path.join(
base_path,
str(season),
f"Subtitle No.{index}.{title}.S{season}.{lang_name}.srt",
)
else:
file_path = os.path.join(base_path, f"Subtitle No.{index}.{title}.{lang_name}.srt")
try:
response = requests.get(url, stream=True, headers=USER_AGENT_HEADER, timeout=15)
if response.status_code != 200:
self.notification(f"Failed to download {url}, status code {response.status_code}")
raise Exception(f"HTTP {response.status_code}")
with open(file_path, "wb") as file:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
file.write(chunk)
return file_path
except Exception as e:
kodilog(f"Subtitle download error for {url}: {e}")
self.notification(f"Subtitle download error for {url}: {e}")
raise