thegame_1980/TwinVine
1
0
Fork 0
mirror of https://github.com/vinefeeder/TwinVine.git synced 2026-07-15 18:10:04 +02:00
Code Issues Packages Projects Releases Wiki Activity

services update

Browse Source
This commit is contained in:
VineFeeder
2026-03-31 14:15:17 +01:00
parent 1486c4bd37
commit c678732bee
29 changed files with 6969 additions and 1518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/envied/CONFIG.md
+3 -3
View File
@@ -125,7 +125,7 @@ curl_impersonate:
## directories (dict)
Override the default directories used across envied.
Override the default directories used across envied.
The directories are set to common values by default.
The following directories are available and may be overridden,
@@ -246,7 +246,7 @@ decryption: mp4decrypt
## filenames (dict)
Override the default filenames used across envied.
Override the default filenames used across envied.
The filenames use various variables that are replaced during runtime.
The following filenames are available and may be overridden:
@@ -305,7 +305,7 @@ accessible outside their hosting platform.
### Using an API Vault
API vaults use a specific HTTP request format, therefore API or HTTP Key Vault APIs from other projects or services may
not work in envied. The API format can be seen in the [API Vault Code](envied/vaults/API.py).
not work in envied.The API format can be seen in the [API Vault Code](envied/vaults/API.py).
```yaml
- type: API
packages/envied/README.md
+1 -1
View File
@@ -8,7 +8,7 @@
## What is envied?
Envied is a fork of [Devine](https://github.com/devine-dl/devine/). The name 'envied' is an anagram of Devine, and as such, pays homage to the original author.
Is is based on v 1.4.7 of envied. It is a powerful archival tool for downloading movies, TV shows, and music from streaming services. Built with a focus on modularity and extensibility, it provides a robust framework for content acquisition with support for DRM-protected content.
Is is based on v 1.4.7 of envied.It is a powerful archival tool for downloading movies, TV shows, and music from streaming services. Built with a focus on modularity and extensibility, it provides a robust framework for content acquisition with support for DRM-protected content.
No commands have been changed 'uv run envied' still works as usual.
packages/envied/src/envied/services/ADN/__init__.py
+365 -135
View File
@@ -33,28 +33,28 @@ from envied.core.tracks.subtitle import Subtitle
class VideoNoAudio(Video):
"""
Video track qui enlève automatiquement l'audio après téléchargement.
Nécessaire car ADN fournit des streams HLS avec audio muxé.
Video track that automatically removes audio after recording.
Necessary because ADN provides HLS streams with muxed audio.
"""
def download(self, session, prepare_drm, max_workers=None, progress=None, *, cdm=None):
"""Override : télécharge puis demuxe pour enlever l'audio."""
"""Override: download then demux to remove audio."""
import logging
log = logging.getLogger('ADN.VideoNoAudio')
# Téléchargement normal
# Normal download
super().download(session, prepare_drm, max_workers, progress, cdm=cdm)
# Si pas de path, échec du téléchargement
# If no path, download failed
if not self.path or not self.path.exists():
return
# Vérifier FFmpeg disponible
# Check FFmpeg available
if not binaries.FFMPEG:
log.warning("FFmpeg not found, cannot remove audio from video")
return
# Demuxer : enlever l'audio
# Demux: remove audio
if progress:
progress(downloaded="Removing audio")
@@ -68,8 +68,8 @@ class VideoNoAudio(Video):
[
binaries.FFMPEG,
'-i', str(original_path),
'-vcodec', 'copy', # Copie vidéo sans réencodage
'-an', # Enlève l'audio
'-vcodec', 'copy', # Copy video without re-encoding
'-an', # Remove audio
'-y',
str(noaudio_path)
],
@@ -88,7 +88,7 @@ class VideoNoAudio(Video):
noaudio_path.unlink(missing_ok=True)
return
# Remplacer le fichier original
# Replace original file
log.debug(f"Video demuxed successfully: {noaudio_path.stat().st_size} bytes")
original_path.unlink()
noaudio_path.rename(original_path)
@@ -106,30 +106,30 @@ class VideoNoAudio(Video):
class AudioExtracted(Audio):
"""
Audio track déjà extrait d'un flux HLS muxé.
Override download() pour copier le fichier au lieu de télécharger.
Audio track already extracted from muxed HLS stream.
Override download() to copy the file instead of downloading.
"""
def __init__(self, *args, extracted_path: Path, **kwargs):
# URL vide pour éviter que curl essaie de télécharger
# Empty URL to prevent curl from trying to download
super().__init__(*args, url="", **kwargs)
self.extracted_path = extracted_path
def download(self, session, prepare_drm, max_workers=None, progress=None, *, cdm=None):
"""Override : copie le fichier extrait au lieu de télécharger."""
"""Override: copies the extracted file instead of downloading."""
if not self.extracted_path or not self.extracted_path.exists():
if progress:
progress(downloaded="[red]FAILED")
raise ValueError(f"Extracted audio file not found: {self.extracted_path}")
# Créer le path de destination (même logique que Track.download)
# Create destination path (same logic as Track.download)
track_type = self.__class__.__name__
save_path = config.directories.temp / f"{track_type}_{self.id}.m4a"
if progress:
progress(downloaded="Copying", total=100, completed=0)
# Copier le fichier extrait vers le path final
# Copy the extracted file to the final destination
config.directories.temp.mkdir(parents=True, exist_ok=True)
shutil.copy2(self.extracted_path, save_path)
@@ -141,30 +141,30 @@ class AudioExtracted(Audio):
class SubtitleEmbedded(Subtitle):
"""
Subtitle avec contenu embarqué (data URI).
Override download() pour écrire le contenu directement.
Subtitle with embedded content (data URI).
Override download() to write the content directly.
"""
def __init__(self, *args, embedded_content: str, **kwargs):
# URL vide pour éviter que curl essaie de télécharger
# Empty URL to prevent curl from trying to download
super().__init__(*args, url="", **kwargs)
self.embedded_content = embedded_content
def download(self, session, prepare_drm, max_workers=None, progress=None, *, cdm=None):
"""Override : écrit le contenu embarqué au lieu de télécharger."""
"""Override: writes the embedded content instead of downloading."""
if not self.embedded_content:
if progress:
progress(downloaded="[red]FAILED")
raise ValueError("No embedded content in subtitle")
# Créer le path de destination
# Create destination path
track_type = "Subtitle"
save_path = config.directories.temp / f"{track_type}_{self.id}.{self.codec.extension}"
if progress:
progress(downloaded="Writing", total=100, completed=0)
# Écrire le contenu
# Write content
config.directories.temp.mkdir(parents=True, exist_ok=True)
save_path.write_text(self.embedded_content, encoding='utf-8')
@@ -216,29 +216,29 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
@staticmethod
@click.command(
name="ADN",
short_help="Téléchargement depuis Animation Digital Network",
short_help="https://animationdigitalnetwork.com",
help=(
"Télécharge des séries ou films depuis ADN.\n\n"
"TITLE : L'URL de la série ou son ID (ex: 1125).\n\n"
"SYSTÈME DE SÉLECTION :\n"
" - Simple : '-e 1-5' (épisodes 1 à 5)\n"
" - Saisons : '-e S2' ou '-e S02' (toute la saison 2) ou '-e S2E1-12'\n"
" - Mixte : '-e 1,3,S2E5' ou '-e 1,3,S02E05'\n"
" - Bonus : '-e NC1,OAV1'"
"Downloads series or movies from ADN.\n\n"
"TITLE: Series URL or ID (eg. 1125).\n\n"
"SELECTION SYSTEM:\n"
" - Simple: '-e 1-5' (episodes 1 to 5)\n"
" - Seasons: '-e S2' or '-e S02' (all season 2) or '-e S2E1-12'\n"
" - Mixed: '-e 1,3,S2E5' or '-e 1,3,S02E05'\n"
" - Bonus: '-e NC1,OAV1'"
)
)
@click.argument("title", type=str, required=True)
@click.option(
"-e", "--episode", "select", type=str,
help="Sélection : numéros, plages (5-10), saisons (S1, S2) ou combiné (S1E5)."
help="Selection: numbers, ranges (5-10), seasons (S1, S2) or combined (S1E5)."
)
@click.option(
"--but", is_flag=True,
help="Inverse la sélection : télécharge tout SAUF les épisodes spécifiés avec -e."
help="Invert selection: download everything EXCEPT episodes specified with -e."
)
@click.option(
"--all", "all_eps", is_flag=True,
help="Ignore toutes les restrictions et télécharge l'intégralité de la série."
help="Ignore all restrictions and download the entire series."
)
@click.pass_context
def cli(ctx, **kwargs) -> "ADN":
@@ -277,7 +277,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
}
def ensure_authenticated(self) -> None:
"""Vérifie le token et rafraîchit si nécessaire."""
"""Checks the token and refreshes if necessary."""
current_time = int(time.time())
if self.access_token and self.token_expiration and current_time < (self.token_expiration - 60):
@@ -340,13 +340,13 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.token_expiration = int(time.time()) + expires_in
self.session.headers.update(self.auth_header)
def _parse_select(self, ep_id: str, short_number: str, season_num: int) -> bool:
"""Retourne True si l'épisode doit être inclus."""
def _parse_select(self, ep_id: str, short_number: str, season_num: int, relative_number: Optional[int] = None) -> bool:
"""Returns True if the episode should be included."""
if self.all_eps or not self.select_str:
return True
# Préparation des identifiants possibles pour cet épisode
# On teste : "30353" (id), "1" (numéro), "S02E01" (format complet), "S02" (saison entière)
# Preparing possible identifiers for this episode
# We test: "30353" (id), "1" (number), "S02E01" (full format), "S02" (entire season)
candidates = [
str(ep_id),
str(short_number).lstrip("0"),
@@ -354,25 +354,29 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
f"S{season_num:02d}"
]
# Add the relative match (e.g. S03E06 for episode 30 which is the 6th ep in S3)
if relative_number is not None:
candidates.append(f"S{season_num:02d}E{relative_number:02d}")
parts = re.split(r'[ ,]+', self.select_str.strip().upper())
selection: set[str] = set()
for part in parts:
if '-' in part:
start_p, end_p = part.split('-', 1)
# Gestion des plages S02E01-S02E04
# Handle ranges S02E01-S02E04
m_start = re.match(r'^S(\d+)E(\d+)$', start_p)
m_end = re.match(r'^S(\d+)E(\d+)$', end_p)
if m_start and m_end:
s_start, e_start = map(int, m_start.groups())
s_end, e_end = map(int, m_end.groups())
if s_start == s_end: # Même saison
if s_start == s_end: # Same season
for i in range(e_start, e_end + 1):
selection.add(f"S{s_start:02d}E{i:02d}")
continue
# Plages classiques (1-10)
# Classic ranges (1-10)
nums = re.findall(r'\d+', part)
if len(nums) >= 2:
for i in range(int(nums[0]), int(nums[1]) + 1):
@@ -384,18 +388,18 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
return not included if self.but else included
def get_titles(self) -> Series:
"""Récupère les épisodes avec le titre réel de la série."""
"""Retrieves episodes with the actual title of the series."""
show_id = self.parse_show_id(self.title)
# 1. Récupérer d'abord les infos globales du show pour avoir le titre propre
# 1. Fetch the overall show info first to get the proper title
show_url = self.config["endpoints"]["show"].format(show_id=show_id)
show_res = self.session.get(show_url).json()
# On extrait le titre de la série (ex: "Demon Slave")
# C'est ce titre qui servira de nom au dossier unique
# We extract the series title (e.g. "Demon Slave")
# This title will be used as the unique folder name
series_title = show_res["videos"][0]["show"]["title"] if show_res.get("videos") else "ADN Show"
# 2. Récupérer ensuite la structure par saisons
# 2. Fetch the season structure afterwards
url_seasons = self.config["endpoints"].get("seasons")
if not url_seasons:
url_seasons = "https://gw.api.animationdigitalnetwork.com/video/show/{show_id}/seasons?maxAgeCategory=18&order=asc"
@@ -403,7 +407,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
res = self.session.get(url_seasons.format(show_id=show_id)).json()
if not res.get("seasons"):
self.log.error(f"Aucune saison trouvée pour l'ID {show_id}")
self.log.error(f"No seasons found for ID {show_id}")
return Series([])
episodes = []
@@ -411,40 +415,178 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
s_val = str(season_data.get("season", "1"))
season_num = int(s_val) if s_val.isdigit() else 1
for vid in season_data.get("videos", []):
for idx, vid in enumerate(season_data.get("videos", []), 1):
video_id = str(vid["id"])
# Nettoyage du numéro d'épisode (on ne garde que les chiffres)
# Clean the episode number (keep only digits)
num_match = re.search(r'\d+', str(vid.get("number", "0")))
short_number = num_match.group() if num_match else "0"
# Logique de sélection (SxxEyy)
if not self._parse_select(video_id, short_number, season_num):
# Selection logic (SxxEyy) - Relative and absolute support
if not self._parse_select(video_id, short_number, season_num, relative_number=idx):
continue
# Création de l'épisode
# Create the episode
episodes.append(Episode(
id_=video_id,
service=self.__class__,
title=series_title, # Dossier : "Demon Slave"
season=season_num, # Saison : 2
number=int(short_number),
name=vid.get("name") or "", # Nom : "La grande réunion..."
title=series_title, # Folder: "Demon Slave"
season=season_num, # Season: 2
number=idx, # Force relative number (e.g. 30 becomes 6 in S3)
name=vid.get("name") or "", # Name: "The big reunion..."
data=vid
))
episodes.sort(key=lambda x: (x.season, x.number))
return Series(episodes)
def get_discovery(self, n: int = 12) -> list[dict]:
"""
Fetch the latest releases (Series) via the catalog API.
"""
self.ensure_authenticated()
try:
url = self.config["endpoints"]["search"]
response = self.session.get(
url,
params={
"maxAgeCategory": 18,
"order": "new",
"limit": n
}
)
if response.status_code != 200:
self.log.error(f"Catalog fetch failed: {response.status_code}")
return []
data = response.json()
return data.get("shows", [])
except Exception as e:
self.log.error(f"Error fetching discovery: {e}")
return []
def get_latest_releases_calendar(self, n: int = 12) -> list[dict]:
"""
Fetch the latest releases (Episodes) via the calendar API.
Returns recent episodes with their actual release dates.
"""
from datetime import datetime, timedelta
import requests
# self.ensure_authenticated()
try:
results = []
seen_episodes = set()
# Use a fresh session to avoid auth issues if calendar is public
cal_session = requests.Session()
cal_session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
})
# Fetch calendar for the last 7 days including today
today = datetime.now()
for day_offset in range(15):
date = today - timedelta(days=day_offset)
date_str = date.strftime("%Y-%m-%d")
# Calendar endpoint
calendar_url = f"https://gw.api.animationdigitalnetwork.com/video/calendar?maxAgeCategory=18&date={date_str}"
try:
res = cal_session.get(calendar_url, timeout=5)
if res.status_code != 200: continue
data = res.json()
# Handle dict response (API returns {"videos": [...]})
if isinstance(data, dict):
data = data.get("videos", [])
for video in data:
show = video.get("show", {})
show_id = str(show.get("id", ""))
ep_id = str(video.get("id", ""))
if (show_id, ep_id) in seen_episodes: continue
seen_episodes.add((show_id, ep_id))
results.append(video)
except:
pass
# Sort by releaseDate
results.sort(key=lambda x: x.get("releaseDate", ""), reverse=True)
return results[:n]
except Exception as e:
self.log.error(f"Error fetching calendar discovery: {e}")
return []
def get_latest_episode(self, show_id: str) -> Episode | None:
"""
Retrieves the absolute latest episode of a series without filtering.
Used by the GUI to display 'Latest' info.
"""
try:
# 1. Fetch seasons
url_seasons = self.config["endpoints"].get("seasons")
if not url_seasons:
url_seasons = "https://gw.api.animationdigitalnetwork.com/video/show/{show_id}/seasons?maxAgeCategory=18&order=asc"
res = self.session.get(url_seasons.format(show_id=show_id))
if res.status_code != 200:
return None
data = res.json()
if not data.get("seasons"):
return None
episodes = []
for season_data in data["seasons"]:
s_val = str(season_data.get("season", "1"))
season_num = int(s_val) if s_val.isdigit() else 1
for vid in season_data.get("videos", []):
# Simplified parsing similar to get_titles
num_match = re.search(r'\d+', str(vid.get("number", "0")))
short_number = int(num_match.group()) if num_match else 0
episodes.append(Episode(
id_=str(vid["id"]),
service=self.__class__,
title="", # We don't need the series title here for simple display
season=season_num,
number=short_number,
name=vid.get("name") or "",
data=vid
))
if not episodes:
return None
# Sort: Season desc, Number desc to find absolute latest?
# Or Season asc, Number asc and take [-1]
episodes.sort(key=lambda x: (x.season, x.number))
return episodes[-1]
except Exception as e:
self.log.error(f"Error in get_latest_episode: {e}")
return None
def get_tracks(self, title: Episode) -> Tracks:
"""
Récupère les pistes en pré-extrayant les audios.
Les audios sont extraits maintenant et seront copiés pendant download().
Fetches tracks by pre-extracting audio.
Audio is extracted now and will be copied during download().
"""
self.ensure_authenticated()
vid_id = title.id
# Configuration du lecteur
# Player configuration
config_url = self.config["endpoints"]["player_config"].format(video_id=vid_id)
config_res = self.session.get(config_url).json()
@@ -452,7 +594,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
if not player_opts["user"]["hasAccess"]:
raise PermissionError("No access to this video (Premium required?)")
# Token du lecteur
# Player token
refresh_url = player_opts["user"].get("refreshTokenUrl") or self.config["endpoints"]["player_refresh"]
token_res = self.session.post(
refresh_url,
@@ -462,7 +604,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
player_token = token_res["token"]
links_url = player_opts["video"].get("url") or self.config["endpoints"]["player_links"].format(video_id=vid_id)
# Chiffrement RSA
# RSA Encryption
rand_key = uuid.uuid4().hex[:16]
payload = json.dumps({"k": rand_key, "t": player_token}).encode('utf-8')
@@ -474,7 +616,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
encrypted = public_key.encrypt(payload, padding.PKCS1v15())
auth_header_val = base64.b64encode(encrypted).decode('utf-8')
# Récupération des liens
# Fetching links
links_res = self.session.get(
links_url,
params={"freeWithAds": "true", "adaptive": "true", "withMetadata": "true", "source": "Web"},
@@ -484,7 +626,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
tracks = Tracks()
streaming_links = links_res.get("links", {}).get("streaming", {})
# Map des langues
# Language map
lang_map = {
"vf": "fr",
"vostf": "ja",
@@ -492,7 +634,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
"vostde": "ja",
}
# Priorité: VOSTF (original) pour la vidéo principale
# Priority: VOSTF (original) for the main video
priority_order = ["vostf", "vf", "vde", "vostde"]
available_streams = {k: v for k, v in streaming_links.items() if k in lang_map}
@@ -504,7 +646,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
if not sorted_streams:
raise ValueError("No supported streams found")
# Vidéo principale (VOSTF ou premier disponible)
# Main video (VOSTF or first available)
primary_stream = sorted_streams[0]
primary_lang = lang_map[primary_stream]
@@ -521,7 +663,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
tracks.add(video_track)
self.log.info(f"Video track added: {video_track.width}x{video_track.height}")
# Extraire audios pour toutes les langues disponibles
# Extract audio for all available languages
for stream_type in sorted_streams:
audio_lang = lang_map[stream_type]
is_original = stream_type in ["vostf", "vostde"]
@@ -540,12 +682,12 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
tracks.add(audio_track, warn_only=True)
self.log.info(f"Audio track added: {audio_lang}")
# Stocker les données de chapitres pour get_chapters()
# Store chapter data for get_chapters()
if "video" in links_res:
title.data["chapter_data"] = links_res["video"]
self.log.debug(f"Stored chapter data: intro={links_res['video'].get('tcIntroStart')}, ending={links_res['video'].get('tcEndingStart')}")
# Sous-titres
# Subtitles
self._process_subtitles(links_res, rand_key, title, tracks)
if not tracks.videos:
@@ -554,7 +696,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
return tracks
def _get_video_track(self, stream_data: dict, stream_type: str, lang: str, is_original: bool):
"""Récupère la piste vidéo principale (sans audio)."""
"""Fetches the main video track (without audio)."""
try:
m3u8_url = self._resolve_stream_url(stream_data, stream_type)
if not m3u8_url:
@@ -567,13 +709,13 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.log.warning(f"No video tracks found for {stream_type}")
return None
# Meilleure qualité
# Best quality
best_video = max(
hls_tracks.videos,
key=lambda v: (v.height or 0, v.width or 0, v.bitrate or 0)
)
# Convertir en VideoNoAudio pour demuxer automatiquement
# Convert to VideoNoAudio to demux automatically
video_no_audio = VideoNoAudio(
id_=best_video.id,
url=best_video.url,
@@ -599,36 +741,67 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
def _extract_audio_track(self, stream_data: dict, stream_type: str, lang: str, is_original: bool, title: Episode):
"""
Extrait l'audio et retourne un AudioExtracted.
L'audio est extrait MAINTENANT et sera copié pendant download().
Extracts audio and returns an AudioExtracted.
Audio is extracted NOW and will be copied during download().
"""
if not binaries.FFMPEG:
self.log.warning("FFmpeg not found, cannot extract audio")
return None
try:
m3u8_url = self._resolve_stream_url(stream_data, stream_type)
m3u8_url = self._resolve_stream_url(stream_data, stream_type, prioritize_auto=True)
if not m3u8_url:
return None
# Créer un répertoire temp pour ADN dans le temp d'Unshackle
# Create an ADN temp directory inside envied.s temp
adn_temp = config.directories.temp / "adn_audio_extracts"
adn_temp.mkdir(parents=True, exist_ok=True)
# Nom de fichier unique basé sur video_id + langue
# Unique filename based on video_id + language
audio_filename = f"audio_{title.id}_{stream_type}.m4a"
audio_path = adn_temp / audio_filename
# Si déjà extrait, réutiliser
# If already extracted, reuse it
if audio_path.exists() and audio_path.stat().st_size > 1000:
self.log.debug(f"Reusing existing extracted audio: {audio_path}")
else:
# Extraire avec FFmpeg
# Parse M3U8 to find best audio
best_m3u8_url = m3u8_url
try:
import m3u8
variant_m3u8 = m3u8.load(m3u8_url)
audio_playlists = []
# Check for alternative audio in media
for media in variant_m3u8.media:
if media.type == "AUDIO" and media.uri:
audio_playlists.append(media.uri)
# If no media, check strict playlists (variants)
if not audio_playlists and variant_m3u8.playlists:
# Sort by bandwidth descending
sorted_playlists = sorted(variant_m3u8.playlists, key=lambda x: x.stream_info.bandwidth or 0, reverse=True)
audio_playlists = [p.uri for p in sorted_playlists]
if audio_playlists:
# Construct absolute URL if needed
from urllib.parse import urljoin
best_audio_uri = audio_playlists[0]
if not best_audio_uri.startswith("http"):
best_m3u8_url = urljoin(m3u8_url, best_audio_uri)
else:
best_m3u8_url = best_audio_uri
self.log.info(f"Selected best audio stream: {best_m3u8_url}")
except Exception as e:
self.log.warning(f"Failed to parse M3U8 for best audio, using default: {e}")
# Extract with FFmpeg
result = subprocess.run(
[
binaries.FFMPEG,
'-i', m3u8_url,
'-i', best_m3u8_url,
'-vn',
'-acodec', 'copy',
'-y',
@@ -649,14 +822,35 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
audio_path.unlink(missing_ok=True)
return None
# Créer AudioExtracted avec le fichier pré-extrait
# Detect actual bitrate
detected_bitrate = 128000
try:
from pymediainfo import MediaInfo
media_info = MediaInfo.parse(audio_path)
if media_info.audio_tracks:
track = media_info.audio_tracks[0]
if track.bit_rate:
detected_bitrate = int(track.bit_rate)
elif track.other_bit_rate:
# Fallback for some formats
try:
# other_bit_rate is typically list like ['128 kb/s']
raw = track.other_bit_rate[0]
detected_bitrate = int(re.sub(r'[^\d]', '', raw)) * 1000
except:
pass
self.log.debug(f"Detected audio bitrate: {detected_bitrate}")
except Exception as e:
self.log.warning(f"Failed to detect bitrate: {e}")
# Create AudioExtracted with the pre-extracted file
audio_track = AudioExtracted(
id_=f"audio-{stream_type}-{lang}",
extracted_path=audio_path,
codec=Audio.Codec.AAC,
language=Language.get(lang),
is_original_lang=is_original,
bitrate=128000,
bitrate=detected_bitrate,
channels=2.0,
)
@@ -669,9 +863,12 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.log.error(f"Failed to extract audio for {stream_type}: {e}")
return None
def _resolve_stream_url(self, stream_data: dict, stream_type: str) -> Optional[str]:
"""Résout l'URL du stream."""
preferred_keys = ["fhd", "hd", "auto", "sd", "mobile"]
def _resolve_stream_url(self, stream_data: dict, stream_type: str, prioritize_auto: bool = False) -> Optional[str]:
"""Resolves the stream URL."""
if prioritize_auto:
preferred_keys = ["auto", "fhd", "hd", "sd", "mobile"]
else:
preferred_keys = ["fhd", "hd", "auto", "sd", "mobile"]
m3u8_url = None
for key in preferred_keys:
@@ -706,7 +903,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
return None
def _process_subtitles(self, links_res: dict, rand_key: str, title: Episode, tracks: Tracks):
"""Traite les sous-titres."""
"""Processes subtitles."""
subs_root = links_res.get("links", {}).get("subtitles", {})
if "all" not in subs_root:
self.log.debug("No subtitles available")
@@ -732,13 +929,13 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
decryptor = cipher.decryptor()
decrypted_padded = decryptor.update(ciphertext) + decryptor.finalize()
# TOUJOURS retirer le padding PKCS7 (Python ne le fait pas automatiquement)
# ALWAYS remove PKCS7 padding (Python doesn't do it automatically)
pad_len = decrypted_padded[-1]
if not (1 <= pad_len <= 16):
self.log.error(f"Invalid PKCS7 padding length: {pad_len}")
return
# Vérifier que tous les bytes de padding ont la même valeur
# Ensure all padding bytes have the same value
padding = decrypted_padded[-pad_len:]
if not all(b == pad_len for b in padding):
self.log.error(f"Invalid PKCS7 padding bytes")
@@ -759,7 +956,7 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.log.warning("subs_data is empty!")
return
# Debug chaque clé
# Debug each key
for key in subs_data.keys():
value = subs_data[key]
if isinstance(value, list) and len(value) > 0:
@@ -781,24 +978,32 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.log.debug(f" Cues count: {len(cues)}")
self.log.debug(f" First cue: {cues[0]}")
if "vf" in sub_lang_key.lower() or "vostf" in sub_lang_key.lower():
if "vf" in sub_lang_key.lower():
target_lang = "fr"
elif "vde" in sub_lang_key.lower() or "vostde" in sub_lang_key.lower():
is_forced = True
elif "vostf" in sub_lang_key.lower():
target_lang = "fr"
is_forced = False
elif "vde" in sub_lang_key.lower():
target_lang = "de"
is_forced = True
elif "vostde" in sub_lang_key.lower():
target_lang = "de"
is_forced = False
else:
self.log.debug(f"Skipping subtitle language: {sub_lang_key}")
continue
if target_lang in processed_langs:
self.log.debug(f"Already processed {target_lang}, skipping")
if (target_lang, is_forced) in processed_langs:
self.log.debug(f"Already processed {target_lang} (forced={is_forced}), skipping")
continue
processed_langs.add(target_lang)
processed_langs.add((target_lang, is_forced))
# Convertir en ASS
# Convert to ASS
ass_content = self._json_to_ass(cues, title.title, title.number)
# Vérifier si le fichier ASS a du contenu
# Check if ASS file has content
event_count = ass_content.count("Dialogue:")
self.log.debug(f"Generated ASS with {event_count} dialogue events")
@@ -806,13 +1011,13 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.log.warning(f"ASS file has no dialogue events!")
self.log.warning(f"First cue was: {cues[0] if cues else 'EMPTY LIST'}")
# Créer SubtitleEmbedded avec le contenu ASS directement
# Create SubtitleEmbedded directly with ASS content
subtitle = SubtitleEmbedded(
id_=f"sub-{target_lang}-{sub_lang_key}",
embedded_content=ass_content, # Contenu ASS directement
embedded_content=ass_content, # ASS content directly
codec=Subtitle.Codec.SubStationAlphav4,
language=Language.get(target_lang),
forced=False,
forced=is_forced,
sdh=False,
)
@@ -829,19 +1034,19 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
def get_chapters(self, title: Episode) -> Chapters:
"""
Crée les chapitres à partir des timecodes ADN.
- Si tcIntroStart existe:
- Si tcIntroStart != "00:00:00": ajouter "Prologue" à 00:00:00
- Ajouter "Opening" à tcIntroStart
- Ajouter "Episode" à tcIntroEnd
- Sinon: ajouter "Episode" à 00:00:00
- Si tcEndingStart existe:
- Ajouter "Ending Start" à tcEndingStart
- Ajouter "Ending End" à tcEndingEnd
Creates chapters from ADN timecodes.
- If tcIntroStart exists:
- If tcIntroStart != "00:00:00": add "Prologue" at 00:00:00
- Add "Opening" at tcIntroStart
- Add "Episode" at tcIntroEnd
- Otherwise: add "Episode" at 00:00:00
- If tcEndingStart exists:
- Add "Ending Start" at tcEndingStart
- Add "Ending End" at tcEndingEnd
"""
chapters = Chapters()
# Récupérer les données de chapitres stockées dans get_tracks()
# Retrieve chapter data stored in get_tracks()
chapter_data = title.data.get("chapter_data", {})
if not chapter_data:
self.log.debug("No chapter data available")
@@ -855,51 +1060,73 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
self.log.debug(f"Chapter timecodes: intro={tc_intro_start}->{tc_intro_end}, ending={tc_ending_start}->{tc_ending_end}")
try:
chapter_num = 1
if tc_intro_start:
# Si l'intro ne commence pas à 00:00:00, ajouter un prologue
# If intro does not start at 00:00:00, add a prologue (Chapter 1)
if tc_intro_start != "00:00:00":
chapters.add(Chapter(
timestamp=0,
name="Prologue"
name=f"Chapter {chapter_num}"
))
self.log.debug("Added Prologue chapter at 00:00:00")
self.log.debug(f"Added Chapter {chapter_num} at 00:00:00")
chapter_num += 1
# Opening
chapters.add(Chapter(
timestamp=self._timecode_to_ms(tc_intro_start),
name="Opening"
))
self.log.debug(f"Added Opening chapter at {tc_intro_start}")
self.log.debug(f"Added Opening at {tc_intro_start}")
# Episode (après l'intro)
# Episode (after intro)
if tc_intro_end:
chapters.add(Chapter(
timestamp=self._timecode_to_ms(tc_intro_end),
name="Episode"
name=f"Chapter {chapter_num}"
))
self.log.debug(f"Added Episode chapter at {tc_intro_end}")
self.log.debug(f"Added Chapter {chapter_num} at {tc_intro_end}")
chapter_num += 1
else:
# Pas d'intro, épisode commence à 00:00:00
# No intro, episode starts at 00:00:00
chapters.add(Chapter(
timestamp=0,
name="Episode"
name=f"Chapter {chapter_num}"
))
self.log.debug("Added Episode chapter at 00:00:00 (no intro)")
self.log.debug(f"Added Chapter {chapter_num} at 00:00:00 (no intro)")
chapter_num += 1
# Ending
if tc_ending_start:
chapters.add(Chapter(
timestamp=self._timecode_to_ms(tc_ending_start),
name="Ending Start"
name="Ending"
))
self.log.debug(f"Added Ending Start chapter at {tc_ending_start}")
self.log.debug(f"Added Ending at {tc_ending_start}")
if tc_ending_end:
chapters.add(Chapter(
timestamp=self._timecode_to_ms(tc_ending_end),
name="Ending End"
))
self.log.debug(f"Added Ending End chapter at {tc_ending_end}")
# Check if the remaining chapter has a significant duration (> 10s)
# to avoid micro-chapters of 2s, while keeping actual post-credits scenes.
tc_end_ms = self._timecode_to_ms(tc_ending_end)
total_duration_s = title.data.get("duration", 0)
total_duration_ms = int(total_duration_s * 1000)
# If duration is unknown or > 10 seconds remaining
should_add = True
if total_duration_ms > 0:
remaining_ms = total_duration_ms - tc_end_ms
if remaining_ms < 10000: # Less than 10s
should_add = False
self.log.debug(f"Skipping post-ending chapter (only {remaining_ms}ms remaining)")
if should_add:
chapters.add(Chapter(
timestamp=tc_end_ms,
name=f"Chapter {chapter_num}"
))
self.log.debug(f"Added Chapter {chapter_num} at {tc_ending_end}")
chapter_num += 1
self.log.info(f"✓ Created {len(chapters)} chapters")
@@ -935,17 +1162,20 @@ KhS+IFEqwvZqgbBpKuwIDAQAB
raise ValueError(f"Invalid ADN Show ID/URL: {input_str}")
def _json_to_ass(self, cues: List[dict], title: str, ep_num: Union[int, str]) -> str:
"""Convertit les sous-titres JSON en ASS."""
"""Converts JSON subtitles to ASS."""
header = """[Script Info]
ScriptType: v4.00+
WrapStyle: 0
PlayResX: 1280
PlayResY: 720
Collisions: Normal
PlayResX: 1920
PlayResY: 1080
Timer: 0.0000
WrapStyle: 0
ScaledBorderAndShadow: yes
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Default,Arial,50,&H00FFFFFF,&H00FFFFFF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1.95,0,2,0,0,70,0
Style: Default,Trebuchet MS,66,&H00FFFFFF,&H000000FF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,3,3,2,75,75,75,1
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
@@ -955,7 +1185,7 @@ Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
line_align_map = {"middle": 8, "end": 4}
def format_time(seconds: float) -> str:
"""Format exact d'adn : HH:MM:SS.CC (centisecondes sur 2 chiffres)"""
"""Exact ADN format: HH:MM:SS.CC (2-digit centiseconds)"""
secs = int(seconds)
centiseconds = round((seconds - secs) * 100)
@@ -963,7 +1193,7 @@ Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
minutes = (secs % 3600) // 60
remaining_seconds = secs % 60
# Padding sur 2 chiffres pour TOUT (hours inclus)
# 2-digit padding for EVERYTHING (including hours)
return f"{hours:02d}:{minutes:02d}:{remaining_seconds:02d}.{centiseconds:02d}"
for cue in cues:
@@ -971,11 +1201,11 @@ Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
end_time = cue.get("endTime", 0)
text = cue.get("text", "")
# Skip si texte vide
# Skip if text is empty
if not text or not text.strip():
continue
# Nettoyage EXACT du code adn
# EXACT cleanup corresponding to ADN code
text = text.replace(' \\N', '\\N') # remove space before \\N at end
if text.endswith('\\N'):
text = text[:-2] # remove \\N at end
@@ -993,7 +1223,7 @@ Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
text = text[:-2]
text = text.rstrip() # remove trailing spaces
# Skip après nettoyage si vide
# Skip after cleanup if empty
if not text.strip():
continue
@@ -1012,4 +1242,4 @@ Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
if not events:
self.log.warning(f"No subtitle events generated - all cues were empty or invalid (total cues: {len(cues)})")
return header + "\n".join(events)
return header + "\n".join(events)
packages/envied/src/envied/services/ADN/config.yaml
+7 -7
View File
@@ -1,22 +1,22 @@
# Animation Digital Network API Configuration
# API Configuration
# Endpoints API
# API Endpoints
endpoints:
# Authentification
# Authentication
login: "https://gw.api.animationdigitalnetwork.com/authentication/login"
refresh: "https://gw.api.animationdigitalnetwork.com/authentication/refresh"
# Catalogue
# Catalog
search: "https://gw.api.animationdigitalnetwork.com/show/catalog"
show: "https://gw.api.animationdigitalnetwork.com/video/show/{show_id}?maxAgeCategory=18&limit=-1&order=asc"
seasons: "https://gw.api.animationdigitalnetwork.com/video/show/{show_id}/seasons?maxAgeCategory=18&order=asc"
# Player & Lecture
# Player & Playback
player_config: "https://gw.api.animationdigitalnetwork.com/player/video/{video_id}/configuration"
player_refresh: "https://gw.api.animationdigitalnetwork.com/player/refresh/token"
player_links: "https://gw.api.animationdigitalnetwork.com/player/video/{video_id}/link"
# Headers par défaut
# Default Headers
headers:
User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
Origin: "https://animationdigitalnetwork.com"
@@ -24,6 +24,6 @@ headers:
Content-Type: "application/json"
X-Target-Distribution: "fr"
# Paramètres
# Parameters
params:
locale: "fr"
packages/envied/src/envied/services/AMZN/CHANGELOG_AMZN.docx
BIN
View File
Binary file not shown.
packages/envied/src/envied/services/AMZN/__init__.py
+926 -857
View File
File diff suppressed because it is too large Load Diff
packages/envied/src/envied/services/AMZN/config.yaml
+32 -70
View File
@@ -9,72 +9,29 @@ certificate: |
uGl1ComHxXV9i1AqHwzzY2JY2vFqo73jR3IElr6oChPIwcNokmNc0D4TXtjE0BoYkbWKJfHvJJihzMOvDicWUsemVHvua9/FBtpbHgpbgwijFPjtQF9Ldb
8Swf
# device:
# new:
# domain: Device
# app_name: 'com.amazon.amazonvideo.livingroom' # AIV, com.amazon.amazonvideo.livingroom
# app_version: '1.1' # AIV: 3.12.0, livingroom: 1.1
# device_model: 'Nexus Player' # SHIELD Android TV, Nexus Player
# os_version: '8.0.0' # 25, 8.0.0
# device_type: 'A2SNKIF736WF4T'
# device_name: '%FIRST_NAME%''s%DUPE_STRATEGY_1ST% Nexus Player'
# device_serial: 'a906a7f9bfd6a7ab' # f8eb5625fd608718
device:
# old: # !<< take note that this is done per-profile
# domain: Device
# app_name: AIV
# app_version: '3.12.0'
# device_model: 'SHIELD Android TV'
# os_version: '28'
# device_type: A1KAXIG6VXSG8Y
# device_serial: '13f5b56b4a17de5d136f0e4c28236109' # os.urandom(16).hex()
# device_name: "Build/LMY47D Shield TV"
# software_version: '248'
# old2:
# 'domain': 'DeviceLegacy'
# 'device_type': A1KAXIG6VXSG8Y,
# 'device_serial': '870f53d1b509594c2f8cd5e340a7d374'
# 'app_name': 'com.amazon.avod.thirdpartyclient'
# 'app_version': '296016847'
# 'device_model': 'mdarcy/nvidia/SHIELD Android TV'
# 'os_version': 'NVIDIA/mdarcy/mdarcy:11/RQ1A.210105.003/7094531_2971.7725:user/release-keys'
# old3:
# domain: Device
# app_name: com.amazon.amazonvideo.livingroom
# app_version: '1.4'
# device_model: PadFone
# os_version: '6.2.5'
# device_type: 'A2SNKIF736WF4T'
# device_name: 'T008 Build/JSS15Q PadFone' # "%FIRST_NAME%'s%DUPE_STRATEGY_1ST% PadFone"
# device_serial: 'c1ebbb433da4afdf'
# old4:
# domain: Device
# app_name: com.amazon.amazonvideo.livingroom
# app_version: '1.4'
# device_model: 'Hisense TV'
# os_version: '3.9.5'
# device_type: 'A3T3XXY42KZQNP' # A2B5DGIWVDH8J3, A3GTP8TAF8V3YG, AFTHA001 # https://developer.amazon.com/docs/fire-tv/identify-amazon-fire-tv-devices.html, https://github.com/giofrida/Hisense-Amazon-Enabler
# device_name: "%FIRST_NAME%'s%DUPE_STRATEGY_1ST% Hisense" # KS964, Build/RP1A.201005.001
# device_serial: '8e3ddf49ee384247'
# old5:
# domain: Device
# app_name: com.amazon.amazonvideo.livingroom
# app_version: '1.1'
# device_model: Hisense
# os_version: '6.0.1' #6.10.19
# device_type: 'A3REWRVYBYPKUM'
# device_name: '%FIRST_NAME%''s%DUPE_STRATEGY_1ST% Hisense'
# device_serial: 'cd24294bffb75a46' # os.urandom(8).hex()
default:
domain: Device
app_name: com.amazon.amazonvideo.livingroom
app_version: '1.4'
device_model: 'MTC'
os_version: '6.0.1' #6.10.19
device_type: 'A2HYAJ0FEWP6N3'
device_name: '%FIRST_NAME%''s%DUPE_STRATEGY_1ST% MTC'
device_serial: 'e6eb1ecdc8e34320'
#Hisense_HU32E5600FHWV: A2RGJ95OVLR12U
#Hisense_HU50A6100UW: AAJ692ZPT1X85
#Hisense_HE55A700EUWTS: A3REWRVYBYPKUM
#MTC_ATV: A2HYAJ0FEWP6N3
app_name: 'com.amazon.amazonvideo.livingroom'
app_version: '7.3.4'
device_model: 'SHIELD Android TV'
device_name: "%FIRST_NAME%'s%DUPE_STRATEGY_1ST% Shield TV"
os_version: '28'
device_type: 'A1KAXIG6VXSG8Y'
device_serial: '9a4740e6fa227483'
software_version: '248'
device_types:
browser: 'AOAGZA014O5RE' # all browsers? all platforms?
@@ -104,19 +61,18 @@ endpoints:
browse: '/cdp/catalog/Browse'
details: '/gp/video/api/getDetailPage'
getDetailWidgets: '/gp/video/api/getDetailWidgets'
playback: '/cdp/catalog/GetPlaybackResources'
licence: '/cdp/catalog/GetPlaybackResources'
playback: '/playback/prs/GetVodPlaybackResources'
metadata: '/api/enrichItemMetadata'
licence: '/playback/drm-vod/GetWidevineLicense'
licence_pr: '/playback/drm-vod/GetPlayReadyLicense'
# chapters/scenes
xray: '/swift/page/xray'
# device registration
ontv: '/region/eu/ontv/code?ref_=atv_auth_red_aft' #/gp/video/ontv/code
ontvold: '/gp/video/ontv/code/ref=atv_device_code'
mytv: '/mytv'
ontv: '/gp/video/ontv/code'
devicelink: '/gp/video/api/codeBasedLinking'
codepair: '/auth/create/codepair'
register: '/auth/register'
token: '/auth/token'
#cookies: '/ap/exchangetoken/cookies'
regions:
us:
@@ -131,6 +87,12 @@ regions:
base_manifest: 'atv-ps-eu.amazon.co.uk'
marketplace_id: 'A2IR4J4NTCP2M5' # A1F83G8C2ARO7P is also another marketplace_id
es:
base: 'www.amazon.es'
base_api: 'api.amazon.es'
base_manifest: 'atv-ps-eu.primevideo.com'
marketplace_id: 'A1RKKUPIHCS9HS'
it:
base: 'www.amazon.it'
base_api: 'api.amazon.it'
@@ -159,4 +121,4 @@ regions:
base: 'www.amazon.com'
base_api: 'api.amazon.com'
base_manifest: 'atv-ps-eu.primevideo.com'
marketplace_id: 'A3K6Y4MI8GDYMT'
marketplace_id: 'A3K6Y4MI8GDYMT'
packages/envied/src/envied/services/AMZN/subtitle.py
+1349
View File
File diff suppressed because it is too large Load Diff
packages/envied/src/envied/services/CR/__init__.py
+468 -400
View File
File diff suppressed because it is too large Load Diff
packages/envied/src/envied/services/CR/config.yaml
+7 -10
View File
@@ -1,7 +1,7 @@
# Crunchyroll API Configuration
client:
id: "lkesi7snsy9oojmi2r9h"
secret: "-aGDXFFNTluZMLYXERngNYnEjvgH5odv"
id: "bludpho6aoup2q0lmw99"
secret: "bEmjmOi1INunmi96t98OFAjtMJq0TwER"
# API Endpoints
endpoints:
@@ -15,24 +15,20 @@ endpoints:
# Content Metadata
series: "https://www.crunchyroll.com/content/v2/cms/series/{series_id}"
seasons: "https://www.crunchyroll.com/content/v2/cms/series/{series_id}/seasons"
episodes: "https://www.crunchyroll.com/content/v2/cms/objects/{episode_id}"
season_episodes: "https://www.crunchyroll.com/content/v2/cms/seasons/{season_id}/episodes"
skip_events: "https://static.crunchyroll.com/skip-events/production/{episode_id}.json"
# Playback
playback: "https://www.crunchyroll.com/playback/v2/{episode_id}/tv/android_tv/play"
#playback_download: "https://www.crunchyroll.com/playback/v2/{episode_id}/android/phone/download"
playback: "https://www.crunchyroll.com/playback/v3/{episode_id}/tv/android_tv/play"
playback_download: "https://www.crunchyroll.com/playback/v3/{episode_id}/android/phone/download"
playback_delete: "https://www.crunchyroll.com/playback/v1/token/{episode_id}/{token}"
playback_sessions: "https://www.crunchyroll.com/playback/v1/sessions/streaming"
license_widevine: "https://cr-license-proxy.prd.crunchyrollsvc.com/v1/license/widevine"
license_playready: "https://cr-license-proxy.prd.crunchyrollsvc.com/v1/license/playReady"
# Discovery
search: "https://www.crunchyroll.com/content/v2/discover/search"
# Headers for Android TV client
headers:
user-agent: "Crunchyroll/ANDROIDTV/3.49.1_22281 (Android 11; en-US; SHIELD Android TV)"
user-agent: "Crunchyroll/ANDROIDTV/3.58.0_22336 (Android 11; en-US; SHIELD Android TV)"
accept: "application/json"
accept-charset: "UTF-8"
accept-encoding: "gzip"
@@ -41,7 +37,8 @@ headers:
# Query parameters
params:
locale: "es-419"
locale: "en-US"
preferred_audio_language: "en-US"
# Device parameters for authentication
device:
packages/envied/src/envied/services/DSNP/README.md
+39 -6
View File
@@ -1,10 +1,10 @@
# DisneyPlus(디즈니플러스)
```
uv run unshackle dl -vl all -al orig -sl ko,en,ja -q 2160 -v h.265 -r HDR10 DSNP entity-4d12671a-f0ad-4c3f-8526-09ae6772390b
uv run envied.dl -vl all -al orig -sl ko,en,ja -q 1080,2160 -v h.264,h.265 -r SDR,HDR10,DV DSNP entity-4d12671a-f0ad-4c3f-8526-09ae6772390b
```
## Information
## Information(정보)
- Authorization: Credentials, Web Token
- Security: UHD@L1/SL3000, FHD@L1/SL3000, HD@L3/SL2000
@@ -14,7 +14,7 @@ uv run unshackle dl -vl all -al orig -sl ko,en,ja -q 2160 -v h.265 -r HDR10 DSNP
- Audio: AAC, AC3, ATMOS, DTS:X(P2:IMAX)
- Range: SDR, HDR10, DV
## Support Args
## Support Args(지원하는 명령어 인자)
- `-i`, `--imax`: Prefer IMAX Enhanced version if available.
- `-r`, `--remastered-ar`: Prefer Remastered Aspect Ratio if available.
@@ -23,18 +23,51 @@ uv run unshackle dl -vl all -al orig -sl ko,en,ja -q 2160 -v h.265 -r HDR10 DSNP
## Tips
- To enable the web refresh token-based login method, please comment out or delete the DSNP section under credentials in `envied.yaml`.
웹 리프레시 토큰 기반 로그인 방식을 활성화하려면 `envied.yaml`의 credentials에서 DSNP부분을 주석 처리하거나 삭제하세요.
```
credentials:
...
# DSNP: example@example.com:example
```
- Configure user settings within the `envied.yaml` file.
사용자 설정은 `envied.yaml`에서 다음과 같이 사용하세요.
```
services:
DSNP:
# Specifies the index of the profile to use. (0 = first profile, 1 = second profile, etc.)
# Automatically select a profile or language when commenting.
# Language setting is only available in languages supported by Disney+.
## 사용자 환경설정
## User configuration
# 해당 설정값이 주석처리 되어 있는 경우에는 설정값들이 자동으로 선택됩니다.
# If these settings are commented out, values will be selected automatically.
preferences:
# 사용할 프로필의 인덱스 번호를 지정합니다. (0 = 첫 번째 프로필, 1 = 두 번째 프로필 등)
# Specifies the index of the profile to use. (0 = first profile, 1 = second profile, etc.)
# 값이 설정되지 않은 경우에는 자동으로 PIN이 안 걸려 있고 키즈 모드가 아닌 프로필로 자동 선택됩니다.
# If no value is set, a profile without a PIN and not in Kids Mode will be automatically selected.
profile: 0
# 서비스 내에서 표시되는 메타데이터 언어를 선택합니다.
# Selects the metadata language displayed within the service.
# 언어 설정은 Disney+에서 지원하는 언어 코드(예: "ko", "en")만 사용 가능합니다.
# Language settings are only available for language codes supported by Disney+ (e.g., "ko", "en").
# 값이 설정되지 않은 경우에는 현재 프로필에 설정된 언어 설정을 사용합니다.
# If no value is set, the language settings of the current profile will be used.
# language: "ko"
# 매니페스트 로그 출력 레벨을 설정합니다.
# Sets the manifest log output level.
# 로그를 항상 표시해야 하는 경우 "info"를 사용하고, 그 외의 모든 경우에는 가급적 "debug"를 사용하십시오.
# Use "info" if the log must always be displayed; otherwise, use "debug" whenever possible.
# 값이 설정되지 않은 경우 기본값은 "debug"로 적용됩니다.
# If no value is set, the default level is "debug".
# manifest_log: "info"
```
- To enable the tier_unlimits command by default, add the following to `envied.yaml`.
tier_unlimits 명령을 기본값으로 활성화하려면 `envied.yaml`에 다음을 추가하세요.
```
dl:
...
packages/envied/src/envied/services/DSNP/__init__.py
+7 -4
View File
@@ -32,7 +32,7 @@ from . import queries
class DSNP(Service):
"""
Service code for Disney+ Streaming Service (https://disneyplus.com).\n
Version: 26.02.27
Version: 26.03.28
Author: Made by CodeName393 with Special Thanks to narakama, Sam\n
Authorization: Credentials, Web Token\n
@@ -575,7 +575,10 @@ class DSNP(Service):
if self.tier_unlimits:
parsed_url = urlparse(manifest_url)
manifest_url = urlunparse(parsed_url._replace(query="")) # Delete tier params
self.log.debug(f" + Manifest URL: {manifest_url}")
log_level = self.config.get("preferences", {}).get("manifest_log", "debug").lower()
log_func = getattr(self.log, log_level, self.log.debug)
log_func(f" + Manifest URL: {manifest_url}")
tracks = HLS.from_url(url=manifest_url, session=self.session).to_tracks(title.language)
artwork_type = "background" if isinstance(title, Movie) else "thumbnail"
@@ -610,7 +613,7 @@ class DSNP(Service):
if audio.channels == 6.0:
audio.channels = 5.1
if audio.channels == 10.0: # DTS-UHD
audio.channels = "5.1.4" # Unshackle does not recommend
audio.channels = "5.1.4" # envied.does not recommend
audio.codec = Audio.Codec.DTS
audio.drm = None # It need HW decording
@@ -623,7 +626,7 @@ class DSNP(Service):
try:
editorial = self.playback_data[title.id]["editorial"]
if not editorial:
return Chapters()
return []
LABEL_MAP = {
"intro_start": "intro_start",
packages/envied/src/envied/services/DSNP/config.yaml
+41 -12
View File
@@ -1,3 +1,9 @@
## DO NOT EDIT THIS FILE
# 해당 config 파일은 개발자 외에는 수정하지 마세요.
# This configuration file should not be modified by anyone other than developers.
# 사용자 환경설정은 반드시 "envied.yaml"에서만 사용하세요.
# User configuration must be performed only in "envied.yaml".
certificate: |
CAUSugUKtAIIAxIQbj3s4jO5oUyWjDWqjfr9WRjA2afZBSKOAjCCAQoCggEBALhKWfnyA+FGn5P3tl6ffDjoGq2Oq86hKGl6aZIaGaF7XHPO5mIk7Q35ml
ZIgg1A458Udb4eXRws1n+kJFqtZXCY5S1yElLP0Om1WQsoEY2stpl+PZTGnVv/CsOJGKQ8K4KMr7rKjZem9lA9BrBoxgfXY3tbwlnSf3wTEohyANb5Qfpa
@@ -10,31 +16,34 @@ certificate: |
## config ( {configVersion}/{clientId}/{deviceFamily}/{sdkVersion}/{applicationRuntime}/{deviceProfile}/{environment} ) ##
# Browser (windows, chrome) : /browser/v34.4/windows/chrome/prod.json
# Android Phone : /android/v16.0.0/google/handset/prod.json
# Android TV : /android/v16.0.0/google/tv/prod.json
# Amazon Fire TV : /android/v16.0.0/amazon/tv/prod.json
# Android Phone : /android/v18.0.0/google/handset/prod.json
# Android TV : /android/v18.0.0/google/tv/prod.json
# Amazon Fire TV : /android/v18.0.0/amazon/tv/prod.json
# Apple Iphone(old) : https://bam-sdk-configs.bamgrid.com/bam-sdk/v2.0/disney-svod-3d9324fc/apple/v9.10.0/ios/iphone/prod.json
# Apple Ipad(old) : https://bam-sdk-configs.bamgrid.com/bam-sdk/v2.0/disney-svod-3d9324fc/apple/v9.10.0/ios/ipad/prod.json
endpoints:
config: "https://client-sdk-configs.bamgrid.com/bam-sdk/v7.0/disney-svod-3d9324fc/android/v16.0.0/google/tv/prod.json"
config: "https://client-sdk-configs.bamgrid.com/bam-sdk/v7.0/disney-svod-3d9324fc/android/v18.0.0/google/tv/prod.json"
## user_agent (okhttp/5.0.0-alpha.14) ##
# android-phone : BAMSDK/v16.0.0 (disney-svod-3d9324fc 26.0.2+rc1-2026.01.29.0; v7.0/v16.0.0; android; phone)
# android-tv : BAMSDK/v16.0.0 (disney-svod-3d9324fc 26.0.2+rc1-2026.01.29.0; v7.0/v16.0.0; android; tv)
# android-phone : BAMSDK/v18.0.0 (disney-svod-3d9324fc 26.0.2+rc1-2026.01.29.0; v7.0/v18.0.0; android; phone)
# android-tv : BAMSDK/v18.0.0 (disney-svod-3d9324fc 26.0.2+rc1-2026.01.29.0; v7.0/v18.0.0; android; tv)
## api_key ##
# browser : ZGlzbmV5JmJyb3dzZXImMS4wLjA.Cu56AgSfBTDag5NiRA81oLHkDZfu5L3CKadnefEAY84
# android : ZGlzbmV5JmFuZHJvaWQmMS4wLjA.bkeb0m230uUhv8qrAXuNu39tbE_mD5EEhM_NAcohjyA
# apple : ZGlzbmV5JmFwcGxlJjEuMC4w.H9L7eJvc2oPYwDgmkoar6HzhBJRuUUzt_PcaC3utBI4
## yp_service_id ##
# browser : 63626081279ebe65eb50fb54
# android : 624b805dafc5c73635b1a216
bamsdk:
sdk_version: "16.0.0"
application_version: "26.0.2+rc1-2026.01.29.0"
sdk_version: "18.0.1"
application_version: "26.3.0+rc4-2026.03.12.0"
explore_version: "v1.13"
client: "disney-svod-3d9324fc"
user_agent: "BAMSDK/v16.0.0 (disney-svod-3d9324fc 26.0.2+rc1-2026.01.29.0; v7.0/v16.0.0; android; tv)"
user_agent: "BAMSDK/v18.0.1 (disney-svod-3d9324fc 26.3.0+rc4-2026.03.12.0; v7.0/v18.0.0; android; tv)"
api_key: "ZGlzbmV5JmFuZHJvaWQmMS4wLjA.bkeb0m230uUhv8qrAXuNu39tbE_mD5EEhM_NAcohjyA"
yp_service_id: "624b805dafc5c73635b1a216"
@@ -47,9 +56,29 @@ device:
operatingSystem: "Android"
operatingSystemVersion: "16"
# Specifies the index of the profile to use. (0 = first profile, 1 = second profile, etc.)
# Automatically select a profile or language when commenting.
# Language setting is only available in languages supported by Disney+.
# ## 사용자 환경설정
# ## User configuration
# # 해당 설정값이 주석처리 되어 있는 경우에는 설정값들이 자동으로 선택됩니다.
# # If these settings are commented out, values will be selected automatically.
# preferences:
# # 사용할 프로필의 인덱스 번호를 지정합니다. (0 = 첫 번째 프로필, 1 = 두 번째 프로필 등)
# # Specifies the index of the profile to use. (0 = first profile, 1 = second profile, etc.)
# # 값이 설정되지 않은 경우에는 자동으로 PIN이 안 걸려 있고 키즈 모드가 아닌 프로필로 자동 선택됩니다.
# # If no value is set, a profile without a PIN and not in Kids Mode will be automatically selected.
# profile: 0
# # 서비스 내에서 표시되는 메타데이터 언어를 선택합니다.
# # Selects the metadata language displayed within the service.
# # 언어 설정은 Disney+에서 지원하는 언어 코드(예: "ko", "en")만 사용 가능합니다.
# # Language settings are only available for language codes supported by Disney+ (e.g., "ko", "en").
# # 값이 설정되지 않은 경우에는 현재 프로필에 설정된 언어 설정을 사용합니다.
# # If no value is set, the language settings of the current profile will be used.
# language: "ko"
# # 매니페스트 로그 출력 레벨을 설정합니다.
# # Sets the manifest log output level.
# # 로그를 항상 표시해야 하는 경우 "info"를 사용하고, 그 외의 모든 경우에는 가급적 "debug"를 사용하십시오.
# # Use "info" if the log must always be displayed; otherwise, use "debug" whenever possible.
# # 값이 설정되지 않은 경우 기본값은 "debug"로 적용됩니다.
# # If no value is set, the default level is "debug".
# manifest_log: "debug"
packages/envied/src/envied/services/HMAX/__init__.py
Executable
+714
View File
@@ -0,0 +1,714 @@
import hashlib
import json
import re
import uuid
from datetime import datetime
from hashlib import md5
from typing import Optional, Union, Generator
from http.cookiejar import CookieJar
from collections import defaultdict
from copy import deepcopy
from zlib import crc32
import click
import requests
import xmltodict
from lxml import etree
from langcodes import Language
from envied.core.constants import AnyTrack
from envied.core.credential import Credential
from envied.core.manifests import DASH
from envied.core.search_result import SearchResult
from envied.core.service import Service
from envied.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
from envied.core.tracks import Attachment, Chapter, Chapters, Subtitle, Tracks, Video
from envied.core.utilities import is_close_match
class HMAX(Service):
ALIASES = ("HMAX", "max", "hbomax")
#GEOFENCE = ("US",)
TITLE_RE = r"^(?:https?://(?:www\.|play\.)?hbomax\.com/)?(?P<type>[^/]+)/(?P<id>[^/]+)"
VIDEO_CODEC_MAP = {
"H264": [Video.Codec.AVC],
"H265": [Video.Codec.HEVC]
}
AUDIO_CODEC_MAP = {
"AAC": "mp4a",
"AC3": "ac-3",
"EC3": "ec-3"
}
@staticmethod
@click.command(name="HMAX", short_help="https://max.com")
@click.argument("title", type=str)
@click.option("-vcodec", "--video-codec", default=None, help="Video codec preference")
@click.option("-acodec", "--audio-codec", default=None, help="Audio codec preference")
@click.pass_context
def cli(ctx, **kwargs):
return HMAX(ctx, **kwargs)
def __init__(self, ctx, title, video_codec, audio_codec):
super().__init__(ctx)
self.title = title
self.vcodec = video_codec
self.acodec = audio_codec
range_param = ctx.parent.params.get("range_")
self.range = range_param[0].name if range_param else "SDR"
if self.range == 'HDR10':
self.vcodec = "H265"
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
super().authenticate(cookies, credential)
if not cookies:
raise EnvironmentError("Service requires Cookies for Authentication.")
try:
token = next(cookie.value for cookie in cookies if cookie.name == "st")
session_data = next(cookie.value for cookie in cookies if cookie.name == "session")
device_id = json.loads(session_data)
except (StopIteration, json.JSONDecodeError):
raise EnvironmentError("Required authentication cookies not found.")
self.session.headers.update({
'User-Agent': 'BEAM-Android/1.0.0.104 (SONY/XR-75X95EL)',
'Accept': 'application/json, text/plain, */*',
'Content-Type': 'application/json',
'x-disco-client': 'SAMSUNGTV:124.0.0.0:beam:4.0.0.118',
'x-disco-params': 'realm=bolt,bid=beam,features=ar',
'x-device-info': 'beam/4.0.0.118 (Samsung/Samsung-Unknown; Tizen/124.0.0.0; f198a6c1-c582-4725-9935-64eb6b17c3cd/87a996fa-4917-41ae-9b6d-c7f521f0cb78)',
'traceparent': '00-315ac07a3de9ad1493956cf1dd5d1313-988e057938681391-01',
'tracestate': f'wbd=session:{device_id}',
'Origin': 'https://play.hbomax.com',
'Referer': 'https://play.hbomax.com/',
})
auth_token = self._get_device_token()
self.session.headers.update({
"x-wbd-session-state": auth_token
})
def search(self) -> Generator[SearchResult, None, None]:
search_url = "https://default.prd.api.hbomax.com/search"
try:
response = self.session.get(search_url, params={"q": self.title})
response.raise_for_status()
search_data = response.json()
for result in search_data.get("results", []):
yield SearchResult(
id_=result.get("id"),
title=result.get("title", "Unknown"),
label=result.get("type", "UNKNOWN").upper(),
url=f"https://play.hbomax.com/{result.get('type', 'content')}/{result.get('id')}"
)
except Exception as e:
self.log.warning(f"Search functionality not fully implemented: {e}")
return
yield
def get_titles(self) -> Titles_T:
match = re.match(self.TITLE_RE, self.title)
if not match:
raise ValueError("Invalid title format. Expected format: type/id or full URL")
content_type = match.group('type')
external_id = match.group('id')
response = self.session.get(
self.config['endpoints']['contentRoutes'] % (content_type, external_id)
)
response.raise_for_status()
try:
content_data = [x for x in response.json()["included"] if "attributes" in x and "title" in
x["attributes"] and x["attributes"]["alias"] == "generic-%s-blueprint-page" % (re.sub(r"-", "", content_type))][0]["attributes"]
content_title = content_data["title"]
except:
content_data = [x for x in response.json()["included"] if "attributes" in x and "alternateId" in
x["attributes"] and x["attributes"]["alternateId"] == external_id and x["attributes"].get("originalName")][0]["attributes"]
content_title = content_data["originalName"]
if content_type == "sport" or content_type == "event":
included_dt = response.json()["included"]
for included in included_dt:
for key, data in included.items():
if key == "attributes":
for k, d in data.items():
if d == "VOD":
event_data = included
release_date = event_data["attributes"].get("airDate") or event_data["attributes"].get("firstAvailableDate")
year = datetime.strptime(release_date, '%Y-%m-%dT%H:%M:%SZ').year
return Movies([
Movie(
id_=external_id,
service=self.__class__,
name=content_title.title(),
year=year,
data=event_data,
)
])
if content_type == "movie" or content_type == "standalone":
metadata = self.session.get(
url=self.config['endpoints']['moviePages'] % external_id
).json()['data']
try:
edit_id = metadata['relationships']['edit']['data']['id']
except:
for x in response.json()["included"]:
if x.get("type") == "video" and x.get("relationships", {}).get("show", {}).get("data", {}).get("id") == external_id:
metadata = x
release_date = metadata["attributes"].get("airDate") or metadata["attributes"].get("firstAvailableDate")
year = datetime.strptime(release_date, '%Y-%m-%dT%H:%M:%SZ').year
return Movies([
Movie(
id_=external_id,
service=self.__class__,
name=content_title,
year=year,
data=metadata,
)
])
if content_type in ["show", "mini-series", "topical"]:
episodes = []
if content_type == "mini-series":
alias = "generic-miniseries-page-rail-episodes"
elif content_type == "topical":
alias = "generic-topical-show-page-rail-episodes"
else:
alias = "-%s-page-rail-episodes-tabbed-content" % (content_type)
included_dt = response.json()["included"]
season_data = [data for included in included_dt for key, data in included.items()
if key == "attributes" for k, d in data.items() if alias in str(d).lower()][0]
season_data = season_data["component"]["filters"][0]
seasons = [int(season["value"]) for season in season_data["options"]]
season_parameters = [(int(season["value"]), season["parameter"]) for season in season_data["options"]
for season_number in seasons if int(season["value"]) == int(season_number)]
if not season_parameters:
raise ValueError("No seasons found")
image_map = {} # accumulate images across all seasons
for (value, parameter) in season_parameters:
data = self.session.get(
url=self.config['endpoints']['showPages'] % (external_id, parameter)
).json()
try:
episodes_dt = sorted([dt for dt in data["included"] if "attributes" in dt and "videoType" in
dt["attributes"] and dt["attributes"]["videoType"] == "EPISODE"
and int(dt["attributes"]["seasonNumber"]) == int(parameter.split("=")[-1])],
key=lambda x: x["attributes"]["episodeNumber"])
except KeyError:
raise ValueError("Season episodes were not found")
episodes.extend(episodes_dt)
# Accumulate image objects across all seasons
image_map.update({
obj["id"]: obj
for obj in data["included"]
if obj.get("type") == "image"
})
episode_titles = []
release_date = episodes[0]["attributes"].get("airDate") or episodes[0]["attributes"].get("firstAvailableDate")
year = datetime.strptime(release_date, '%Y-%m-%dT%H:%M:%SZ').year
season_map = {int(item[1].split("=")[-1]): item[0] for item in season_parameters}
for episode in episodes:
# Resolve image objects for this episode using relationship IDs
ep_image_ids = [
img["id"]
for img in episode.get("relationships", {}).get("images", {}).get("data", [])
]
ep_images = [image_map[img_id] for img_id in ep_image_ids if img_id in image_map]
ep_data = dict(episode)
ep_data["_images"] = ep_images
episode_titles.append(
Episode(
id_=episode['id'],
service=self.__class__,
title=content_title,
season=season_map.get(episode['attributes'].get('seasonNumber')),
number=episode['attributes']['episodeNumber'],
name=episode['attributes']['name'],
year=year,
data=ep_data
)
)
return Series(episode_titles)
def get_tracks(self, title: Title_T) -> Tracks:
edit_id = title.data['relationships']['edit']['data']['id']
response = self.session.post(
url=self.config['endpoints']['playbackInfo'],
json={
'appBundle': 'beam',
'consumptionType': 'streaming',
'deviceInfo': {
'deviceId': '2dec6cb0-eb34-45f9-bbc9-a0533597303c',
'browser': {
'name': 'chrome',
'version': '113.0.0.0',
},
'make': 'Microsoft',
'model': 'XBOX-Unknown',
'os': {
'name': 'Windows',
'version': '113.0.0.0',
},
'platform': 'XBOX',
'deviceType': 'xbox',
'player': {
'sdk': {
'name': 'Beam Player Console',
'version': '1.0.2.4',
},
'mediaEngine': {
'name': 'GLUON_BROWSER',
'version': '1.20.1',
},
'playerView': {
'height': 1080,
'width': 1920,
},
},
},
'editId': edit_id,
'capabilities': {
'manifests': {
'formats': {
'dash': {},
},
},
'codecs': {
'video': {
'hdrFormats': [
'hlg',
'hdr10',
'dolbyvision5',
'dolbyvision8',
],
'decoders': [
{
'maxLevel': '6.2',
'codec': 'h265',
'levelConstraints': {
'width': {
'min': 1920,
'max': 3840,
},
'height': {
'min': 1080,
'max': 2160,
},
'framerate': {
'min': 15,
'max': 60,
},
},
'profiles': [
'main',
'main10',
],
},
{
'maxLevel': '4.2',
'codec': 'h264',
'levelConstraints': {
'width': {
'min': 640,
'max': 3840,
},
'height': {
'min': 480,
'max': 2160,
},
'framerate': {
'min': 15,
'max': 60,
},
},
'profiles': [
'high',
'main',
'baseline',
],
},
],
},
'audio': {
'decoders': [
{
'codec': 'aac',
'profiles': [
'lc',
'he',
'hev2',
'xhe',
],
},
],
},
},
'devicePlatform': {
'network': {
'lastKnownStatus': {
'networkTransportType': 'unknown',
},
'capabilities': {
'protocols': {
'http': {
'byteRangeRequests': True,
},
},
},
},
'videoSink': {
'lastKnownStatus': {
'width': 1290,
'height': 2796,
},
'capabilities': {
'colorGamuts': [
'standard',
'wide',
],
'hdrFormats': [
'dolbyvision',
'hdr10plus',
'hdr10',
'hlg',
],
},
},
},
},
'gdpr': False,
'firstPlay': False,
'playbackSessionId': str(uuid.uuid4()),
'applicationSessionId': str(uuid.uuid4()),
'userPreferences': {},
'features': [],
}
)
response.raise_for_status()
playback_data = response.json()
video_info = next(x for x in playback_data['videos'] if x['type'] == 'main')
title.language = Language.get(video_info['defaultAudioSelection']['language'])
fallback_url = playback_data["fallback"]["manifest"]["url"]
fallback_url = fallback_url.replace('fly', 'akm').replace('gcp', 'akm')
try:
self.wv_license_url = playback_data["drm"]["schemes"]["widevine"]["licenseUrl"]
except (KeyError, IndexError):
self.wv_license_url = None
try:
self.pr_license_url = playback_data["drm"]["schemes"]["playready"]["licenseUrl"]
except (KeyError, IndexError):
self.pr_license_url = None
manifest_url = fallback_url.replace('_fallback', '')
self.log.info(f" + Manifest: {manifest_url}")
dash_manifest = DASH.from_url(url=manifest_url, session=self.session)
tracks = dash_manifest.to_tracks(language=title.language)
self.log.debug(tracks)
tracks.videos = self._dedupe(tracks.videos)
tracks.audio = self._dedupe(tracks.audio)
tracks.subtitles.clear()
new_subtitles = self._process_max_subtitles(dash_manifest, title.language)
for subtitle in new_subtitles:
tracks.add(subtitle)
if self.vcodec:
tracks.videos = [x for x in tracks.videos if x.codec in self.VIDEO_CODEC_MAP[self.vcodec]]
if self.acodec:
tracks.audio = [x for x in tracks.audio if (x.codec or "")[:4] == self.AUDIO_CODEC_MAP[self.acodec]]
for track in tracks:
if isinstance(track, Video):
codec = track.data.get("dash", {}).get("representation", {}).get("codecs", "")
track.hdr10 = track.range == Video.Range.HDR10
track.dv = codec[:4] in ("dvh1", "dvhe")
if isinstance(track, Subtitle) and not track.codec:
track.codec = Subtitle.Codec.WebVTT
title.data['info'] = video_info
for track in tracks.audio:
if hasattr(track, 'data') and track.data.get("dash", {}).get("adaptation_set"):
role = track.data["dash"]["adaptation_set"].find("Role")
if role is not None and role.get("value") in ["description", "alternative", "alternate"]:
track.descriptive = True
# Attachment: episode/movie image resolved from relationships._images
# _images is built in get_titles by cross-referencing the included image objects
# HBO Max image attributes use "src" (not "url") and "kind" for type
try:
images = title.data.get("_images", [])
image_url = None
# Prefer landscape (default/wide) images; fallback to first available
PREFERRED_KINDS = ("default", "tile", "episode", "cover", "banner",
"centered-background-small", "background")
kind_map = {}
for img in images:
attrs = img.get("attributes", {})
kind = attrs.get("kind", "").lower()
src = attrs.get("src", "") or attrs.get("url", "")
if src and kind not in kind_map:
kind_map[kind] = src
for k in PREFERRED_KINDS:
if k in kind_map:
image_url = kind_map[k]
break
# Fallback: take first available src
if not image_url and images:
for img in images:
attrs = img.get("attributes", {})
src = attrs.get("src", "") or attrs.get("url", "")
if src:
image_url = src
break
if image_url:
if isinstance(title, Movie):
image_name = title.name
else:
image_name = f"{title.title} - S{title.season:02d}E{title.number:02d}"
tracks.add(Attachment.from_url(
url=image_url,
name=image_name,
mime_type="image/jpeg",
session=self.session,
))
except Exception as e:
self.log.warning(f" - Attachment failed: {e}")
return tracks
def get_chapters(self, title: Title_T) -> Chapters:
chapters = []
video_info = title.data.get('info', {})
if 'annotations' in video_info:
chapters.append(Chapter(timestamp=0.0, name='Chapter 1'))
chapters.append(Chapter(timestamp=self._convert_timecode(video_info['annotations'][0]['start']), name='Credits'))
chapters.append(Chapter(timestamp=self._convert_timecode(video_info['annotations'][0]['end']), name='Chapter 2'))
return Chapters(chapters)
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[Union[bytes, str]]:
if not self.wv_license_url:
return None
response = self.session.post(
url=self.wv_license_url,
data=challenge
)
response.raise_for_status()
return response.content
def get_playready_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[bytes]:
if not self.pr_license_url:
return None
if isinstance(challenge, bytes):
decoded_challenge = challenge.decode('utf-8')
else:
decoded_challenge = str(challenge)
response = self.session.post(
url=self.pr_license_url,
data=decoded_challenge,
headers={
'Content-Type': 'text/xml; charset=utf-8',
'SOAPAction': 'http://schemas.microsoft.com/DRM/2007/03/protocols/AcquireLicense'
}
)
response.raise_for_status()
return response.content
def _get_device_token(self):
response = self.session.post(self.config['endpoints']['bootstrap'])
response.raise_for_status()
return response.headers.get('x-wbd-session-state')
@staticmethod
def _convert_timecode(time_seconds):
return float(time_seconds)
def _process_max_subtitles(self, dash_manifest, language):
subtitle_groups = defaultdict(list)
for period in dash_manifest.manifest.findall("Period"):
for adaptation_set in period.findall("AdaptationSet"):
content_type = adaptation_set.get("contentType")
if content_type != "text":
continue
lang = adaptation_set.get("lang")
if not lang:
continue
role = adaptation_set.find("Role")
role_value = role.get("value") if role is not None else "subtitle"
label = adaptation_set.find("Label")
label_text = label.text if label is not None else ""
key = (lang, role_value, label_text)
subtitle_groups[key].append((period, adaptation_set))
final_tracks = []
for (lang, role_value, label_text), adapt_pairs in subtitle_groups.items():
if not adapt_pairs:
continue
first_period, first_adapt = adapt_pairs[0]
first_rep = first_adapt.find("Representation")
if not first_rep:
continue
combined_adapt = deepcopy(first_adapt)
combined_rep = combined_adapt.find("Representation")
seg_template = combined_rep.find("SegmentTemplate")
if seg_template is None:
seg_template = combined_adapt.find("SegmentTemplate")
if seg_template is None:
continue
combined_adapt.remove(seg_template)
seg_template = deepcopy(seg_template)
combined_rep.append(seg_template)
timeline = etree.Element("SegmentTimeline")
segments_info = []
for period, adapt in adapt_pairs:
rep = adapt.find("Representation")
if not rep:
continue
template = rep.find("SegmentTemplate")
if not template:
template = adapt.find("SegmentTemplate")
if not template:
continue
start_num = int(template.get("startNumber", 1))
existing_timeline = template.find("SegmentTimeline")
if existing_timeline is not None:
for s in existing_timeline.findall("S"):
t = int(s.get("t", 0))
d = int(s.get("d", 0))
r = int(s.get("r", 0))
segments_info.append((start_num, t, d, r))
segments_info.sort(key=lambda x: x[0])
if segments_info:
for _, t, d, r in segments_info:
s_elem = etree.Element("S")
s_elem.set("t", str(t))
s_elem.set("d", str(d))
if r > 0:
s_elem.set("r", str(r))
timeline.append(s_elem)
old_timeline = seg_template.find("SegmentTimeline")
if old_timeline is not None:
seg_template.remove(old_timeline)
seg_template.append(timeline)
seg_template.set("startNumber", "1")
seg_template.set("endNumber", str(len(segments_info)))
track_id = hex(crc32(f"sub-{lang}-{role_value}-{label_text}".encode()))[2:]
is_sdh = role_value == "caption" or "sdh" in label_text.lower()
is_forced = role_value in ["forced-subtitle", "forced_subtitle"] or "forced" in label_text.lower()
lang_obj = Language.get(lang)
display_name = lang_obj.display_name()
subtitle_track = Subtitle(
id_=track_id,
url=dash_manifest.url,
codec=Subtitle.Codec.WebVTT,
language=Language.get(lang),
is_original_lang=bool(language and is_close_match(Language.get(lang), [language])),
descriptor=Video.Descriptor.DASH,
sdh=is_sdh,
forced=is_forced,
name=display_name,
data={
"dash": {
"manifest": dash_manifest.manifest,
"period": first_period,
"adaptation_set": combined_adapt,
"representation": combined_rep,
}
},
)
final_tracks.append(subtitle_track)
return final_tracks
@staticmethod
def _dedupe(items: list) -> list:
if not items:
return items
if isinstance(items[0].url, list):
return items
seen = {}
for item in items:
if hasattr(item, 'width') and hasattr(item, 'height'):
key = f"{item.codec}_{item.width}x{item.height}_{item.bitrate}"
elif hasattr(item, 'channels'):
key = f"{item.codec}_{item.language}_{item.bitrate}_{item.channels}"
else:
key = item.url
if key not in seen:
seen[key] = item
return list(seen.values())
packages/envied/src/envied/services/HMAX/config.yaml
Executable
+6
View File
@@ -0,0 +1,6 @@
endpoints:
contentRoutes: 'https://default.prd.api.hbomax.com/cms/routes/%s/%s?include=default'
moviePages: 'https://default.prd.api.hbomax.com/content/videos/%s/activeVideoForShow?&include=edit'
playbackInfo: 'https://default.prd.api.hbomax.com/any/playback/v1/playbackInfo'
showPages: 'https://default.prd.api.hbomax.com/cms/collections/generic-show-page-rail-episodes-tabbed-content?include=default&pf[show.id]=%s&%s'
bootstrap: 'https://default.prd.api.hbomax.com/session-context/headwaiter/v1/bootstrap'
packages/envied/src/envied/services/MGMP/__init__.py
+1033
View File
File diff suppressed because it is too large Load Diff
packages/envied/src/envied/services/MGMP/config.yaml
+43
View File
@@ -0,0 +1,43 @@
endpoints:
sessions: https://api.epixnow.com/v2/sessions
registration_code: https://api.epixnow.com/v2/devices/registration_code
graphql: https://api.epixnow.com/graphql
search: https://api.epixnow.com/v2/search
headers:
Accept: application/json
User-Agent: okhttp/5.1.0
client:
apikey: 592c205299de839aa3948643addfe0bc
device:
app_version: "225.0.2025225001"
device: android
display_height: 2160
display_width: 3840
dpi: 320
format: tv
manufacturer: NVIDIA
model: SHIELD Android TV
os: fire
os_version: "11"
web:
endpoints:
sessions: https://api.mgmplus.com/v2/sessions
graphql: https://api.mgmplus.com/graphql
apikey: "53e208a9bbaee479903f43b39d7301f7"
device:
format: console
os: web
display_width: 2560
display_height: 1440
app_version: "1.0.2"
model: browser
manufacturer: google
device_interface_type: desktop
amazon:
base: https://q423ce52d0d3f1.na.api.amazonvideo.com
device_type: AOAGZA014O5RE
marketplace: ATVPDKIKX0DER
packages/envied/src/envied/services/MGMP/queries.py
+226
View File
@@ -0,0 +1,226 @@
MOVIE = """
query Movie($id: ID!) {
movie(id: $id) {
id
title
shortName
releaseYear
synopsis
duration
mpaaRating
underlyingId
genres { name }
labels { text }
descriptiveAudio { available }
}
}
"""
SERIES = """
query Series($id: ID!) {
series(id: $id) {
id
title
shortName
shortDescription
mpaaRating
genres { name }
labels { text }
descriptiveAudio { available }
seasons {
nodes {
id
title
number
__typename
}
__typename
}
__typename
}
}
"""
EPISODES = """
query Episodes($seasonId: ID!, $first: Int, $after: String) {
episodes(seasonId: $seasonId, first: $first, after: $after) {
pageInfo {
endCursor
hasNextPage
__typename
}
nodes {
id
title
releaseYear
duration
number
description
seasonNumber
shortTitle
rating { value }
series {
shortName
__typename
}
__typename
}
__typename
}
}
"""
EPISODE = """
query Episode($seriesShortName: String!, $seasonNumber: Int!, $episodeNumber: Int!) {
episode(
seriesShortName: $seriesShortName
seasonNumber: $seasonNumber
episodeNumber: $episodeNumber
) {
id
title
releaseYear
duration
number
description
seasonNumber
shortTitle
rating { value }
series {
shortName
__typename
}
__typename
}
}
"""
PLAY = """
query Play($id: String!, $context: String, $behavior: BehaviorEnum, $supportedActions: [PlayFlowActionEnum!]!) {
playFlow(id: $id, context: $context, behavior: $behavior, supportedActions: $supportedActions) {
__typename
... on PlayContent {
type
continuationContext
playheadPosition
streams(types: [
{ packagingSystem: DASH, encryptionScheme: CBCS },
{ packagingSystem: DASH, encryptionScheme: CENC },
{ packagingSystem: HLS, encryptionScheme: NONE }
]) {
id
internalStreamId
videoQuality { width height }
widevine { authenticationToken licenseServerUrl }
packagingSystem
encryptionScheme
playlistUrl
}
closedCaptions {
vtt { location }
}
currentItem {
content {
__typename
... on Movie {
id
title
underlyingId
duration
}
... on Episode {
id
title
underlyingId
duration
}
}
}
hints {
videoId
videoTitle
resourceId
isLive
}
}
... on ShowNotice {
type
actions {
continuationContext
text
}
}
... on ContinuePlay {
type
}
... on LogIn {
type
}
... on Noop {
type
}
}
}
"""
WEB_PLAY = """
query PlayFlow($id: String!, $supportedActions: [PlayFlowActionEnum!]!, $context: String, $behavior: BehaviorEnum = DEFAULT, $streamTypes: [StreamDefinition!]) {
playFlow(
id: $id
supportedActions: $supportedActions
context: $context
behavior: $behavior
) {
... on ShowNotice {
type
actions { continuationContext text }
description
title
__typename
}
... on PlayContent {
type
continuationContext
heartbeatToken
currentItem {
content {
__typename
... on Movie { id shortName amazonContentId duration }
... on Episode {
id
series { shortName __typename }
seasonNumber
number
amazonContentId
duration
}
}
}
amazonPlayback {
playbackEnvelope
playbackId
__typename
}
playheadPosition
closedCaptions {
vtt { location __typename }
__typename
}
streams(types: $streamTypes) {
id
playlistUrl
packagingSystem
encryptionScheme
videoQuality { height width __typename }
widevine { authenticationToken licenseServerUrl __typename }
__typename
}
__typename
}
... on ContinuePlay { type __typename }
... on LogIn { type __typename }
... on Noop { type __typename }
__typename
}
}
"""
packages/envied/src/envied/services/RKTN/__init__.py
+2 -2
View File
@@ -44,7 +44,7 @@ class RKTN(Service):
\b
Command for Titles with no SDR (if not set range to HDR10 it will fail):
uv run unshackle dl -r HDR10 [OPTIONS] RKTN -m https://www.rakuten.tv/...
uv run envied.dl -r HDR10 [OPTIONS] RKTN -m https://www.rakuten.tv/...
\b
TODO: - TV Shows are not yet supported as there's 0 TV Shows to purchase, rent, or watch in my region
@@ -587,7 +587,7 @@ class RKTN(Service):
audio.channels = audio_info.channel_s or "2.0"
# Actualizar codec
# Para Unshackle, necesitas mantener el formato correcto
# Para envied. necesitas mantener el formato correcto
audio.codec = Audio.Codec.from_codecs(codec_name)
# Agregar el track
packages/envied/src/envied/services/SEVEN/__init__.py
+3 -3
View File
@@ -42,9 +42,9 @@ class SEVEN(Service):
\b
Examples:
- SERIES: unshackle dl -w s01e01 7plus https://7plus.com.au/ncis-los-angeles
- EPISODE: unshackle dl 7plus https://7plus.com.au/ncis-los-angeles?episode-id=NCIL01-001
- MOVIE: unshackle dl 7plus --movie https://7plus.com.au/puss-in-boots-the-last-wish
- SERIES: envied.dl -w s01e01 7plus https://7plus.com.au/ncis-los-angeles
- EPISODE: envied.dl 7plus https://7plus.com.au/ncis-los-angeles?episode-id=NCIL01-001
- MOVIE: envied.dl 7plus --movie https://7plus.com.au/puss-in-boots-the-last-wish
"""
packages/envied/src/envied/services/SKST/__init__.py
+1062
View File
File diff suppressed because it is too large Load Diff
packages/envied/src/envied/services/SKST/changelog_skst.docx
BIN
View File
Binary file not shown.
packages/envied/src/envied/services/SKST/config.yaml
+42
View File
@@ -0,0 +1,42 @@
endpoints:
signin: "https://rango.id.skyshowtime.com/signin/service/international"
tokens: "https://ovp.skyshowtime.com/auth/tokens"
personas: "https://web.clients.skyshowtime.com/bff/personas/v2"
atom_node: "https://atom.skyshowtime.com/adapter-calypso/v3/query/node"
atom_search: "https://atom.skyshowtime.com/adapter-calypso/v3/query/search"
playback: "https://ovp.skyshowtime.com/video/playouts/vod"
params:
provider: "SKYSHOWTIME"
proposition: "SKYSHOWTIME"
platform: "PC"
device: "COMPUTER"
client_version: "6.11.21-gsp"
signature:
app_id: "SHOWMAX-ANDROID-v1"
key: "kC2UFjsH6PHrc5ENGfyTgC5bPA7aBVZ4aJAyqBBP"
version: "1.0"
territories:
- NL
- PL
- ES
- PT
- SE
- NO
- DK
- FI
- CZ
- SK
- HU
- RO
- BG
- HR
- SI
- BA
- RS
- ME
- MK
- AL
- XK
packages/envied/src/envied/services/TOD/__init__.py
+572
View File
@@ -0,0 +1,572 @@
import re
import html
import json
import click
import base64
import unicodedata
from langcodes import Language
from typing import Optional, Union
from http.cookiejar import CookieJar
from urllib.parse import urlparse, parse_qsl, urlencode
from envied.core.service import Service
from envied.core.constants import AnyTrack
from envied.core.manifests import DASH, HLS
from envied.core.credential import Credential
from envied.core.tracks import Audio, Chapters, Tracks, Video
from envied.core.session import session as create_curl_session
from envied.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
class TOD(Service):
"""
Service code for TODTV (TR) [https://todtv.com.tr]
Version: 2.0.0
Author: PREDATOR
Authorization: Cookie
Author Recommendation: Recommend using the "requests" download method.
Security: FHD@L3 [Widevine]
"""
TITLE_RE = r'^(?:https?://(?:www\.)?todtv\.com\.tr/)?(?:(?P<type>diziler|dizi|film|tod-studios|bein-series|belgesel)/)?(?P<slug>[^/\?]+)(?:/(?P<season>[^/\?]+))?(?:/(?P<episode>[^/\?]+))?'
ID_RE = r'^(?P<prefix>PS|PZ|PT)(?P<id>\d+)$'
PLAYER_CONFIG_RE = r'var\s+playerConfig\s*=\s*(\{[\s\S]*?\});'
SEASONS_RE = r'var\s+seasons\s*=\s*(\[[\s\S]*?\]);'
SEASON_HREF_RE = r'href="([^"]*?(\d+)sezon-[^"]*)"'
@staticmethod
@click.command(name="TOD", short_help="https://todtv.com.tr", help=__doc__)
@click.argument("title", type=str)
@click.pass_context
def cli(ctx, **kwargs):
return TOD(ctx, **kwargs)
def __init__(self, ctx, title):
super().__init__(ctx)
self.title = title
self.movie = False
self.content_id = None
self.content_type = None
self.slug = None
self.license_url = None
self.drm_token = None
self.castleblack_token = None
if (id_match := re.match(self.ID_RE, self.title)):
self.content_id = self.title
prefix = id_match.group("prefix")
self.content_type = {"PS": "series", "PZ": "season", "PT": "episode"}.get(prefix)
if prefix == "PT":
self.movie = True
elif (url_match := re.match(self.TITLE_RE, self.title)):
self.content_type = url_match.group("type") or "diziler"
self.slug = url_match.group("slug")
if s := url_match.group("season"):
self.slug += f"/{s}"
if e := url_match.group("episode"):
self.slug += f"/{e}"
if self.content_type == "film":
self.movie = True
else:
self.slug = self.title
self.content_type = "film" if self.movie else "diziler"
# ──────────────────────────── Auth ──────────────────────────── #
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
super().authenticate(cookies, credential)
if not cookies:
raise EnvironmentError("TODTV Requires Cookies for Authentication...")
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://www.todtv.com.tr/",
"X-Requested-With": "XMLHttpRequest",
})
self._sanitize_session()
self.log.info("Session Authenticated using Cookies...")
# ──────────────────────────── Titles ──────────────────────────── #
def get_titles(self) -> Titles_T:
return self._fetch_movie_titles() if self.movie else self._fetch_series_titles()
def _fetch_movie_titles(self) -> Movies:
if self.content_id and self.content_id.startswith("PT"):
return Movies([Movie(
id_=self.content_id, service=self.__class__,
name=self.content_id, language=Language.get("tr"),
data={"contentId": self.content_id},
)])
url = self._build_url(self.slug, "film")
response = self.session.get(url)
response.raise_for_status()
data = self._extract_content_info(response.text)
if not data:
raise ValueError(f"Movie Info not Found: {self.slug}")
return Movies([Movie(
id_=data.get("contentId", self.slug),
service=self.__class__,
name=data.get("title", self.slug),
description=data.get("description", ""),
year=data.get("year"),
language=Language.get("tr"),
data=data,
)])
def _fetch_series_titles(self) -> Series:
base_path = self.content_type
response = None
fallback_types = self.config.get("fallback_types", ["diziler", "dizi", "tod-studios", "bein-series", "belgesel"])
paths_to_try = [base_path] if base_path else []
paths_to_try.extend([t for t in fallback_types if t != base_path])
for path in paths_to_try:
if not path:
continue
try:
url = self._build_url(self.slug, path)
resp = self.session.get(url)
resp.raise_for_status()
response = resp
base_path = path
break
except Exception:
continue
if not response:
raise ValueError(f"Series Page not Found: {self.slug}")
series_data = self._extract_series_data(response.text)
if not series_data["episodes"] and (ep_link := re.search(rf'href="(/[^/]+/{re.escape(self.slug)}/[^"]+)"', response.text)):
self.log.debug("Trying Data from Episode Page...")
base_url = self.config.get("endpoints", {}).get("base_url", "https://www.todtv.com.tr")
if (ep_resp := self.session.get(f"{base_url}{ep_link.group(1)}")) and ep_resp.status_code == 200:
series_data = self._extract_series_data(ep_resp.text)
existing_ids = {ep.get("contentId") for ep in series_data["episodes"]}
sorted_metadata = sorted(
series_data.get("seasons_metadata", []),
key=lambda x: int(x["no"]) if str(x["no"]).isdigit() else 999,
)
for meta in sorted_metadata:
s_no, s_id, s_slug = meta["no"], meta["id"], meta.get("slug")
cached_count = sum(1 for ep in series_data["episodes"] if str(ep.get("season")) == str(s_no))
if cached_count > 0:
self.log.info(f"Fetching Season {s_no} data (ID: {s_id or 'Cached'})...")
self.log.info(f"Season {s_no}: {cached_count} new episodes added.")
continue
if not s_id:
continue
self.log.info(f"Fetching Season {s_no} data (ID: {s_id})...")
if s_slug:
configs_to_try = [self._build_url(s_slug)]
else:
page_v_ids = set(re.findall(r'\b(v\d+)\b', response.text))
series_base = self.slug.split('/')[0]
configs_to_try = []
for vid in page_v_ids:
configs_to_try.append(self._build_url(f"{series_base}/{s_no}-sezon-{vid}", base_path))
configs_to_try.append(self._build_url(f"{series_base}/{s_no}sezon-{vid}", base_path))
configs_to_try.append(self._build_url(f"{series_base}/{s_no}sezon-{s_id}", base_path))
if len(configs_to_try) > 2:
self.log.info(f"Auto-discovering Season {s_no} URL (scanning {len(page_v_ids)} candidates)...")
found_season = False
for url in configs_to_try:
if found_season:
break
try:
if (s_resp := self.session.get(url)) and s_resp.status_code == 200:
s_data = self._extract_series_data(s_resp.text)
added_count = 0
for ep in s_data["episodes"]:
if str(ep.get("season")) != str(s_no):
continue
if ep["contentId"] not in existing_ids:
series_data["episodes"].append(ep)
existing_ids.add(ep["contentId"])
added_count += 1
if added_count:
self.log.info(f"Season {s_no}: {added_count} new episodes added.")
found_season = True
except Exception:
pass
if not found_season:
self.log.warning(f"Could not find valid episodes for Season {s_no}")
if not series_data["episodes"]:
raise ValueError(f"Episode not found: {self.slug}")
show_name = series_data.get("title", self.slug)
episodes = [
Episode(
id_=ep["contentId"],
service=self.__class__,
title=show_name,
season=ep["season"],
number=ep["episode"],
name=ep.get("name") or f"Bölüm {ep['episode']}",
description=ep.get("data", {}).get("description", ""),
year=ep.get("data", {}).get("year"),
language=Language.get("tr"),
data=ep["data"],
) for ep in series_data["episodes"]
]
return Series(episodes)
# ──────────────────────────── Tracks ──────────────────────────── #
def get_tracks(self, title: Title_T) -> Tracks:
if isinstance(title, Movie):
page_url = title.data.get("pageUrl") or self._build_url(title.data.get("slug") or self.slug, "film")
else:
slug = (title.data or {}).get("slug") or ((title.data or {}).get("customData") or {}).get("slug")
page_url = self._build_url(slug) if slug else None
if not page_url:
raise ValueError(f"URL not found: {title.id}")
self.log.info(f"🔗 Content Page: {page_url}")
resp = self.session.get(page_url)
resp.raise_for_status()
info = self._extract_stream_info(resp.text)
manifest_url = info.get("manifestUrl")
license_url = info.get("licenseUrl")
if all(info.get(k) for k in ["contentId", "assetId", "usageSpecId"]):
try:
self.log.info("▶ Sending play request (playRequest)...")
play_data = self._play_request(resp.text, info)
if play_data.get("CdnUrl"):
manifest_url = play_data["CdnUrl"]
self.session.get(manifest_url, allow_redirects=True)
if play_data.get("DrmTicket"):
self.drm_token = play_data["DrmTicket"].replace("ticket=", "")
if play_data.get("manifestUrl"):
manifest_url = play_data["manifestUrl"]
if play_data.get("licenseUrl"):
license_url = play_data["licenseUrl"]
except Exception as e:
self.log.warning(f"Play request failed: {e}")
if not manifest_url:
raise ValueError(f"No Manifest URL: {title.id}")
self.license_url = license_url
self.log.info(f"Manifest URL: {manifest_url}")
manifest = HLS if "HLS" in manifest_url or manifest_url.endswith(".m3u8") else DASH
tracks = manifest.from_url(manifest_url, self.session).to_tracks(language=title.language)
parsed_manifest = urlparse(manifest_url)
manifest_query = dict(parse_qsl(parsed_manifest.query))
if token := manifest_query.get("hdnts"):
for track in tracks:
if isinstance(track, (Video, Audio)):
t_parsed = urlparse(track.url)
t_query = dict(parse_qsl(t_parsed.query))
if "hdnts" not in t_query:
t_query["hdnts"] = token
track.url = t_parsed._replace(query=urlencode(t_query)).geturl()
for track in tracks:
if isinstance(track, (Video, Audio)):
track.license_url = license_url
self._sanitize_session()
return tracks
# ──────────────────────────── DRM ──────────────────────────── #
def get_widevine_service_certificate(self, **_) -> Optional[str]:
return None
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[Union[bytes, str]]:
endpoints = self.config.get("endpoints", {})
if self.castleblack_token:
license_url = endpoints.get("license_castleblack", "https://castleblack.digiturk.com.tr/api/widevine/license?version=1.0")
headers = {
"Authorization": f"Bearer {self.castleblack_token}",
"X-CB-Ticket": self.drm_token,
}
else:
if not self.license_url:
raise ValueError("No License URL.")
license_url = self.license_url
headers = {"Content-Type": "application/octet-stream"}
if self.drm_token:
headers["X-ErDRM-Message"] = self.drm_token
resp = self.session.post(license_url, data=challenge, headers=headers)
if "<License>" in resp.text:
if match := re.search(r"<License>(.*?)</License>", resp.text):
return base64.b64decode(match.group(1))
return resp.content
def get_chapters(self, title: Title_T) -> Chapters:
return Chapters()
# ──────────────────────────── Helpers ──────────────────────────── #
def _sanitize_session(self):
"""Normalize non-latin-1 characters in session headers and cookies.
The requests download worker encodes headers/cookies as latin-1.
Characters like œ (\u0153) in cookie values set by the TOD server cause
UnicodeEncodeError. We strip them here at the source so core
downloaders stay untouched.
"""
for key in list(self.session.headers.keys()):
val = self.session.headers[key]
if isinstance(val, str):
try:
val.encode("latin-1")
except UnicodeEncodeError:
self.session.headers[key] = (
unicodedata.normalize("NFKD", val)
.encode("ascii", "ignore")
.decode("ascii")
)
# curl_cffi.Cookies → .jar / stdlib CookieJar → iterate directly
cookie_jar = getattr(self.session.cookies, "jar", None) or self.session.cookies
for cookie in cookie_jar:
if hasattr(cookie, "value") and isinstance(cookie.value, str):
try:
cookie.value.encode("latin-1")
except UnicodeEncodeError:
cookie.value = (
unicodedata.normalize("NFKD", cookie.value)
.encode("ascii", "ignore")
.decode("ascii")
)
def _play_request(self, html_content: str, info: dict) -> dict:
verify_token = ""
if match := re.search(r'name="__RequestVerificationToken"\s+type="hidden"\s+value="([^"]+)"', html_content):
verify_token = match.group(1)
form_data = {
"__RequestVerificationToken": verify_token,
"contentId": info["contentId"],
"versionId": info.get("versionId", ""),
"assetId": info["assetId"],
"usageSpecId": info["usageSpecId"],
"contentType": info.get("contentType", "Movie"),
"assetType": info.get("assetType", "MUL"),
"videoType": "1",
"updateWatchingOptions": "True",
"restart": "False",
}
play_request_url = self.config.get("endpoints", {}).get("play_request", "https://www.todtv.com.tr/content/playRequest")
resp = self.session.post(
play_request_url,
data=form_data,
headers={"X-Requested-With": "XMLHttpRequest"},
)
resp.raise_for_status()
data = resp.json()
if data.get("MessageCode") != 0 and not data.get("CdnUrl"):
raise ValueError(f"PlayRequest error: {data.get('Message', 'Unknown error')} (Code: {data.get('MessageCode')})")
self.drm_token = (data.get("DrmTicket") or "").replace("ticket=", "")
self.castleblack_token = data.get("CastleBlackToken")
return data
def _extract_stream_info(self, html_content: str) -> dict:
info = {}
patterns = {
"contentId": [r'data-cms-id=["\']([^"\']+)["\']', r'data-content-id=["\']([^"\']+)["\']'],
"assetId": [r'data-play-asset-id=["\']([^"\']+)["\']', r'data-id=["\']([^"\']+)["\']'],
"usageSpecId": [r'data-usage-spec=["\']([^"\']+)["\']', r'data-usage-id=["\']([^"\']+)["\']'],
"versionId": [r'data-version-id=["\']([^"\']+)["\']'],
"assetType": [r'data-play-asset-asset-type=["\']([^"\']+)["\']', r'data-assettype=["\']([^"\']+)["\']'],
}
for key, regexes in patterns.items():
for regex in regexes:
if match := re.search(regex, html_content):
val = match.group(1)
if "{{" not in val:
info[key] = val
break
if not info.get("assetId") and (match := re.search(r'data-asset-list=["\'](\[.*?\])["\']', html_content, re.DOTALL)):
if assets := self._parse_json_safe(match.group(1).replace('&quot;', '"')):
asset = assets[0]
info.setdefault("assetId", asset.get("AssetId"))
info.setdefault("usageSpecId", asset.get("UsageSpecId"))
return info
def _build_url(self, slug: str, content_type: str = None) -> str:
base_url = self.config.get("endpoints", {}).get("base_url", "https://www.todtv.com.tr")
if slug.startswith("http"):
return slug
if slug.startswith("/"):
return f"{base_url}{slug}"
base = content_type or self.content_type or "diziler"
if base in ("series", "season", "episode"):
base = "diziler"
return f"{base_url}/{base}/{slug}"
def _clean_title(self, raw_title: str) -> str:
if not raw_title:
return ""
title = html.unescape(raw_title.split("|")[0].strip())
title = re.sub(r'\s*-\s*\d+\.\s*Sezon.*', '', title, flags=re.IGNORECASE)
title = re.sub(r'\s*-\s*(?:TOD|beIN|Fantastik|Aksiyon|Dram|Komedi|Bilim Kurgu).*', '', title, flags=re.IGNORECASE)
return title.strip()
@staticmethod
def _parse_json_safe(text: str) -> dict:
try:
return json.loads(re.sub(r'//[^\n]*', '', text))
except Exception:
return {}
def _extract_season_slugs(self, html_content: str) -> dict:
slugs = {}
for m in re.finditer(self.SEASON_HREF_RE, html_content):
slugs[int(m.group(2) or m.group(3))] = m.group(1)
return slugs
def _process_seasons_list(self, seasons: list, info: dict, source: str, seen_ids: set) -> None:
if "seasons_metadata" not in info:
info["seasons_metadata"] = []
for s_data in seasons:
s_id = s_data.get("id")
s_no = int(s_data.get("no", 1))
episodes = s_data.get("episodes")
if not any(m["id"] == s_id for m in info["seasons_metadata"]):
info["seasons_metadata"].append({
"id": s_id,
"no": s_no,
"slug": (s_data.get("customData") or {}).get("slug"),
"populated": bool(episodes),
})
for ep in episodes or []:
ep_id = ep.get("id")
if ep_id in seen_ids:
continue
seen_ids.add(ep_id)
info["episodes"].append({
"contentId": ep_id,
"name": ep.get("title"),
"season": s_no,
"episode": int(ep.get("no", len(info["episodes"]) + 1)),
"slug": (ep.get("customData") or {}).get("slug"),
"data": ep,
})
if info["episodes"]:
self.log.debug(f"{len(info['episodes'])} episodes found ({source})")
def _extract_series_data(self, html_content: str) -> dict:
info = {"episodes": []}
seen_ids = set()
turkish_title = None
if title_match := re.search(r'<title>([^<]+)</title>', html_content):
turkish_title = title_match.group(1)
if (match := re.search(self.SEASONS_RE, html_content)) and (seasons := self._parse_json_safe(match.group(1))):
self._process_seasons_list(seasons, info, "window.seasons", seen_ids)
if match := re.search(self.PLAYER_CONFIG_RE, html_content):
config = self._parse_json_safe(match.group(1))
if config.get("title"):
turkish_title = config["title"]
if seasons := config.get("seasons") or config.get("seriesSettings", {}).get("seasons"):
self._process_seasons_list(seasons, info, "playerConfig", seen_ids)
found_slugs = self._extract_season_slugs(html_content)
for meta in info.get("seasons_metadata", []):
if meta["no"] in found_slugs and not meta.get("slug"):
meta["slug"] = found_slugs[meta["no"]]
original_title = self._extract_original_title_from_breadcrumbs(html_content)
info["title"] = self._format_title(turkish_title, original_title)
return info
def _extract_original_title_from_breadcrumbs(self, html_content: str) -> Optional[str]:
if match := re.search(r'(<script type="application/ld\+json">[^<]*"BreadcrumbList"[^<]*</script>)', html_content, re.DOTALL):
try:
json_str = match.group(1).replace('<script type="application/ld+json">', '').replace('</script>', '')
data = json.loads(json_str)
items = data.get("itemListElement", [])
if items:
return items[-1].get("item", {}).get("name")
except Exception:
pass
return None
def _format_title(self, turkish_title: str, original_title: Optional[str]) -> str:
if not turkish_title:
return ""
turkish_title = self._clean_title(turkish_title)
if original_title:
original_title = self._clean_title(original_title)
if original_title.lower() != turkish_title.lower():
safe = unicodedata.normalize("NFKD", original_title).encode("ascii", "ignore").decode("ascii").strip()
return f"{turkish_title} ({safe if safe else original_title})"
return turkish_title
def _extract_content_info(self, html_content: str) -> dict:
info = {}
if match := re.search(self.PLAYER_CONFIG_RE, html_content):
info.update(self._parse_json_safe(match.group(1)))
if match := re.search(r'"contentId"\s*:\s*"([^"]+)"', html_content):
info["contentId"] = match.group(1)
if match := re.search(r'var\s+videoEventObject\s*=\s*(\{[^;]+\});', html_content):
info.update(self._parse_json_safe(match.group(1)))
turkish_title = info.get("title")
if not turkish_title:
if match := re.search(r'<title>([^<]+)</title>', html_content):
turkish_title = match.group(1)
original_title = self._extract_original_title_from_breadcrumbs(html_content)
info["title"] = self._format_title(turkish_title, original_title)
return info
# ──────────────────────────── Session ──────────────────────────── #
@staticmethod
def get_session():
"""Creates a session with retries and browser impersonation."""
return create_curl_session("chrome124", max_retries=3, status_forcelist=[429, 500, 502, 503, 504])
packages/envied/src/envied/services/TOD/config.yaml
+11
View File
@@ -0,0 +1,11 @@
endpoints:
base_url: "https://www.todtv.com.tr"
play_request: "https://www.todtv.com.tr/content/playRequest"
license_castleblack: "https://castleblack.digiturk.com.tr/api/widevine/license?version=1.0"
fallback_types:
- diziler
- dizi
- tod-studios
- bein-series
- belgesel
packages/envied/src/envied/services/TUBI/__init__.py
+7 -5
View File
@@ -25,7 +25,7 @@ class TUBI(Service):
Service code for TubiTV streaming service (https://tubitv.com/)
\b
Version: 1.0.7
Version: 1.0.8
Author: stabbedbybrick
Authorization: Cookies (Optional)
Geofence: Locked to whatever region the user is in (API only)
@@ -206,10 +206,12 @@ class TUBI(Service):
tracks = Tracks()
for codec in ["h264", "h265"]:
resource = next((
x for x in resources
if self.drm_system in x.get("type", "") and codec in x.get("codec", "").lower()
), None)
codec_matches = [x for x in resources if codec in x.get("codec", "").lower()]
resource = next(iter(sorted(codec_matches, key=lambda x: (
self.drm_system not in x.get("type", ""),
"dash" not in x.get("type", "")
))), None)
if not resource:
self.log.warning(f" - Could not find a {codec} video resource for this title")
continue
packages/envied/src/envied/services/UNXT/README.md
+1 -1
View File
@@ -1,7 +1,7 @@
# U-NEXT(유넥스트)
```
uv run unshackle dl --list -vl all -al orig -sl all -q 2160 -v h.265 -r HDR10 UNXT SID0087607
uv run envied.dl --list -vl all -al orig -sl all -q 2160 -v h.265 -r HDR10 UNXT SID0087607
```
## Information