nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-27 03:22:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

Prepare RTLPlus for new streaming prov

Browse Source
This commit is contained in:
Nirvana
2026-04-29 11:32:44 +02:00
parent 4757714ae3
commit db071cb4cc
2 changed files with 187 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/models/vod.py
+112 -53
View File
@@ -1,15 +1,4 @@
# streaming_providers/base/models/vod.py
"""
VOD (Video on Demand) models.
Two types:
VodCategory — a browsable node (directory, collection, series, season, …).
Never directly playable.
VodItem — a playable leaf (movie, episode, documentary, …).
Inherits full Content machinery; manifest/DRM resolved via
get_manifest(content_id) / get_drm(content_id) like channels
and events.
"""
import re
import unicodedata
@@ -21,39 +10,46 @@ from ..utils.logger import logger
# ---------------------------------------------------------------------------
# Slug utilities
# Slug utilities (defensive)
# ---------------------------------------------------------------------------
def slugify(text: str) -> str:
def slugify(text: Optional[str]) -> str:
"""
Convert an arbitrary string to a URL-safe slug.
Steps:
1. Unicode normalise (NFKD) and drop combining characters
2. Lowercase
3. Replace whitespace and common separators with underscores
4. Strip everything that is not alphanumeric or underscore/hyphen
5. Collapse consecutive underscores/hyphens
6. Strip leading/trailing underscores and hyphens
Args:
text: String to slugify (can be None)
Returns:
Slugified string or empty string if text is None/empty
"""
# 1. Normalise unicode
text = unicodedata.normalize("NFKD", text)
text = "".join(c for c in text if not unicodedata.combining(c))
# 2. Lowercase
text = text.lower()
# 3. Whitespace / separators → underscore
text = re.sub(r"[\s\-–—/\\|]+", "_", text)
# 4. Keep only safe characters
text = re.sub(r"\W", "", text)
# 5. Collapse runs
text = re.sub(r"_+", "_", text)
# 6. Strip edges
return text.strip("_-")
# Defensive: handle None or empty input
if not text:
logger.debug("slugify called with None or empty text, returning empty string")
return ""
try:
# 1. Normalise unicode
text = unicodedata.normalize("NFKD", text)
text = "".join(c for c in text if not unicodedata.combining(c))
# 2. Lowercase
text = text.lower()
# 3. Whitespace / separators → underscore
text = re.sub(r"[\s\-–—/\\|]+", "_", text)
# 4. Keep only safe characters
text = re.sub(r"[^\w\-_]", "", text)
# 5. Collapse runs
text = re.sub(r"_+", "_", text)
# 6. Strip edges
return text.strip("_-")
except Exception as e:
logger.warning(f"Error slugifying text '{text}': {e}, returning empty string")
return ""
def build_slug_map(entries: list) -> Dict[str, str]:
"""
Build a slug → content_id mapping for a list of VodCategory / VodItem
Build a slug -> content_id mapping for a list of VodCategory / VodItem
entries, falling back to the content_id itself as the slug when two
siblings would produce the same slug.
@@ -63,24 +59,32 @@ def build_slug_map(entries: list) -> Dict[str, str]:
# First pass: compute preferred slug for every entry
preferred: List[tuple] = []
for entry in entries:
if not hasattr(entry, 'name') or not entry.name:
logger.warning(f"Entry missing name: {entry}")
continue
preferred.append((slugify(entry.name), entry.content_id))
# Detect collisions
seen_slugs: Dict[str, int] = {}
for slug, _ in preferred:
seen_slugs[slug] = seen_slugs.get(slug, 0) + 1
if slug: # Only count non-empty slugs
seen_slugs[slug] = seen_slugs.get(slug, 0) + 1
slug_map: Dict[str, str] = {}
for slug, content_id in preferred:
if seen_slugs[slug] > 1:
if not slug:
# If slug is empty, use content_id directly
final_slug = slugify(content_id) or content_id
slug_map[final_slug] = content_id
elif seen_slugs.get(slug, 0) > 1:
# Fall back to the raw ID as the slug
final_slug = slugify(content_id) or content_id
logger.debug(
f"VOD slug collision for '{slug}' — using id '{final_slug}' instead"
)
slug_map[final_slug] = content_id
else:
final_slug = slug
slug_map[final_slug] = content_id
slug_map[slug] = content_id
return slug_map
@@ -123,10 +127,31 @@ class VodCategory:
# Cached slug (computed lazily if not set)
_slug: Optional[str] = field(default=None, repr=False)
def __post_init__(self):
"""Validate and clean up required fields."""
# Defensive: ensure name is a string
if self.name is None:
logger.warning(f"VodCategory created with None name for content_id {self.content_id}")
self.name = f"Unbenannt_{self.content_id}"
elif not isinstance(self.name, str):
logger.warning(f"VodCategory created with non-string name '{self.name}' for content_id {self.content_id}")
self.name = str(self.name)
# Ensure content_id is valid
if not self.content_id:
raise ValueError("VodCategory requires content_id")
# Ensure provider is valid
if not self.provider:
raise ValueError("VodCategory requires provider")
@property
def slug(self) -> str:
if not self._slug:
self._slug = slugify(self.name)
# If slugify returns empty (e.g., for non-Latin names), use a fallback
if not self._slug:
self._slug = slugify(self.content_id) or self.content_id
return self._slug
@property
@@ -168,14 +193,14 @@ class VodItem(Content):
release_year: Optional[int] = None
# Classification
rating: Optional[str] = None # e.g. "FSK 12", "PG-13", "TV-MA"
genre: Optional[str] = None # primary genre (mainGenre)
genres: Optional[List[str]] = None # full genre list
rating: Optional[str] = None # e.g. "FSK 12", "PG-13", "TV-MA"
genre: Optional[str] = None # primary genre (mainGenre)
genres: Optional[List[str]] = None # full genre list
# Extended descriptions
# description (short) is inherited from Content
long_description: Optional[str] = None # longDescription from contentInformation
original_title: Optional[str] = None # originalTitle — important for localised content
long_description: Optional[str] = None # longDescription from contentInformation
original_title: Optional[str] = None # originalTitle — important for localised content
# People
cast: Optional[List[str]] = None
@@ -185,8 +210,8 @@ class VodItem(Content):
season_number: Optional[int] = None
episode_number: Optional[int] = None
# Back-references for episodes — lets the UI navigate up without re-parsing IDs
series_id: Optional[str] = None # e.g. "GN_SERIES_184925"
series_title: Optional[str] = None # e.g. "Two and a Half Men"
series_id: Optional[str] = None # e.g. "GN_SERIES_184925"
series_title: Optional[str] = None # e.g. "Two and a Half Men"
# Promotional
trailer_url: Optional[str] = None
@@ -200,16 +225,34 @@ class VodItem(Content):
_slug: Optional[str] = field(default=None, repr=False)
def __post_init__(self):
"""Validate and clean up required fields."""
# Defensive: ensure name is a string
if self.name is None:
logger.warning(f"VodItem created with None name for content_id {self.content_id}")
self.name = f"Unbenanntes Video_{self.content_id}"
elif not isinstance(self.name, str):
logger.warning(f"VodItem created with non-string name '{self.name}' for content_id {self.content_id}")
self.name = str(self.name)
# Ensure mode and content_type are set correctly for on-demand content
if self.mode == "live":
self.mode = "vod"
if self.content_type == "LIVE":
self.content_type = "VOD"
# Defensive: handle None for season/episode numbers
if self.season_number is not None and self.season_number < 0:
self.season_number = None
if self.episode_number is not None and self.episode_number < 0:
self.episode_number = None
@property
def slug(self) -> str:
if not self._slug:
self._slug = slugify(self.name)
# If slugify returns empty, use a fallback
if not self._slug:
self._slug = slugify(self.content_id) or self.content_id
return self._slug
@property
@@ -260,13 +303,20 @@ class VodItem(Content):
warnings.append("duration_seconds must be positive")
if self.release_year is not None and not (1888 <= self.release_year <= 2100):
warnings.append(f"Unusual release_year: {self.release_year}")
if not self.name:
warnings.append("Name is empty or None")
if not self.content_id:
warnings.append("Content ID is empty or None")
return warnings
# Factory methods
@classmethod
def create_movie(
cls, name: str, content_id: str, provider: str, **kwargs
cls, name: str, content_id: str, provider: str, **kwargs
) -> "VodItem":
# Defensive: ensure name is string
if name is None:
name = f"Film_{content_id}"
return cls(
name=name,
content_id=content_id,
@@ -278,14 +328,23 @@ class VodItem(Content):
@classmethod
def create_episode(
cls,
name: str,
content_id: str,
provider: str,
season_number: int,
episode_number: int,
**kwargs,
cls,
name: str,
content_id: str,
provider: str,
season_number: int,
episode_number: int,
**kwargs,
) -> "VodItem":
# Defensive: ensure name is string
if name is None:
name = f"Episode_{content_id}"
# Defensive: handle negative season/episode numbers
if season_number is not None and season_number < 0:
season_number = None
if episode_number is not None and episode_number < 0:
episode_number = None
return cls(
name=name,
content_id=content_id,
lib/streaming_providers/providers/rtlplus/vod_manager.py
+75 -15
View File
@@ -160,29 +160,46 @@ class RTLPlusVodManager:
continue
# Get items safely
items = block.get("content", {}).get("items")
content = block.get("content")
if not content:
continue
items = content.get("items")
if not items:
continue
# Process each block
# Process each item in the block
for item in items:
if not item:
continue
# Try to extract as category first
cat = self._extract_vod_category_from_block_item(item)
if cat:
entries.append(cat)
logger.debug(f"Added category: {cat.name} ({cat.content_id})")
continue
# If not a category, try as VOD item
vod_item = self._extract_vod_item_from_block_item(item)
if vod_item:
entries.append(vod_item)
logger.debug(f"Added VOD item: {vod_item.name} ({vod_item.content_id})")
logger.info(f"Found {len(entries)} entries in root VOD category")
# Filter out any entries with None names just in case
valid_entries = []
for entry in entries:
if hasattr(entry, 'name') and entry.name:
valid_entries.append(entry)
else:
logger.warning(f"Entry has no name: {entry}")
return {
"entries": entries,
"entries": valid_entries,
"next_cursor": None,
"total": len(entries),
"total": len(valid_entries),
}
except Exception as e:
logger.error(f"Error in _get_root_category: {e}", exc_info=True)
@@ -396,9 +413,27 @@ class RTLPlusVodManager:
if not clip_id:
return None
# Get name with fallback
title = item_content.get("title", "")
extra_title = item_content.get("extraTitle", "")
full_title = f"{title} - {extra_title}" if extra_title else title
if title and extra_title:
full_title = f"{title} - {extra_title}"
elif title:
full_title = title
elif extra_title:
full_title = extra_title
else:
# Try to get from highlight or other fields
highlight = item_content.get("highlight", "")
if highlight:
# Clean up highlight (remove date info)
if "•" in highlight:
full_title = highlight.split("•")[0].strip()
else:
full_title = highlight
else:
full_title = f"Unbekanntes Video ({clip_id})"
highlight = item_content.get("highlight", "")
season_number: Optional[int] = None
@@ -406,19 +441,31 @@ class RTLPlusVodManager:
if highlight:
import re
m = re.search(r"Staffel\s*(\d+)", highlight, re.IGNORECASE)
if m:
season_number = int(m.group(1))
m = re.search(r"Folge\s*(\d+)", highlight, re.IGNORECASE)
if m:
episode_number = int(m.group(1))
# Look for season/folge patterns
season_match = re.search(r"Staffel\s*(\d+)", highlight, re.IGNORECASE)
if season_match:
season_number = int(season_match.group(1))
else:
# Try alternative: "S01" pattern
season_match = re.search(r"S(\d+)", highlight, re.IGNORECASE)
if season_match:
season_number = int(season_match.group(1))
episode_match = re.search(r"Folge\s*(\d+)", highlight, re.IGNORECASE)
if episode_match:
episode_number = int(episode_match.group(1))
else:
# Try alternative: "E01" pattern
episode_match = re.search(r"E(\d+)", highlight, re.IGNORECASE)
if episode_match:
episode_number = int(episode_match.group(1))
vod_item = VodItem.create_episode(
name=full_title or title or "Unbekannte Episode",
name=full_title,
content_id=clip_id,
provider=self._provider.provider_name,
season_number=season_number,
episode_number=episode_number,
season_number=season_number if season_number is not None else -1, # Use -1 for unknown
episode_number=episode_number if episode_number is not None else -1,
)
vod_item.description = item_content.get("description")
vod_item.logo_url = self._extract_thumbnail(item_content)
@@ -448,8 +495,21 @@ class RTLPlusVodManager:
if not content_id:
return None
# Get name with fallback
name = item_content.get("title")
if not name:
# Try to extract from other fields
name = item_content.get("extraTitle") or item_content.get("highlight") or f"{layout_type}_{content_id}"
if name:
# Clean up highlight if it contains date info
if "•" in str(name):
name = name.split("•")[0].strip()
if not name:
name = f"Unbekannt {layout_type}"
return VodCategory(
name=item_content.get("title", "Unbekannt"),
name=name,
content_id=f"{layout_type}:{content_id}",
provider=self._provider.provider_name,
logo_url=self._extract_thumbnail(item_content),