nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-15 20:32:34 +02:00
Code Issues Packages Projects Releases Wiki Activity

Prepare recordings

Browse Source
This commit is contained in:
Nirvana
2026-03-20 11:50:53 +01:00
parent 392b159972
commit 6a2e810b9f
7 changed files with 780 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/manager.py
+42
View File
@@ -12,6 +12,7 @@ from .drm_operations import DRMOperations
from .epg_operations import EPGOperations
from .event_operations import EventOperations
from .vod_operations import VodOperations
from .recording_operations import RecordingOperations
from .models import StreamingChannel
from .provider_registry import ProviderRegistry
from .subscription_operations import SubscriptionOperations
@@ -36,6 +37,7 @@ class ProviderManager:
self.subscription_ops = SubscriptionOperations(self.registry)
self.event_ops = EventOperations(self.registry)
self.vod_ops = VodOperations(self.registry)
self.recording_ops = RecordingOperations(self.registry)
# Backward compatibility - expose managers directly
self.drm_plugin_manager = self.drm_ops.drm_plugin_manager
@@ -261,6 +263,46 @@ class ProviderManager:
def get_all_vod_roots(self) -> dict:
return self.vod_ops.get_all_vod_roots()
# ==========================================================================
# RECORDING OPERATIONS (delegate to RecordingOperations)
# ==========================================================================
def get_recordings(
self,
provider_name: str,
include_deleted: bool = False,
):
return self.recording_ops.get_recordings(
provider_name, include_deleted=include_deleted
)
def get_all_recordings(self, include_deleted: bool = False):
return self.recording_ops.get_all_recordings(
include_deleted=include_deleted
)
def get_recording_manifest(
self, provider_name: str, recording_id: str, **kwargs
) -> Optional[str]:
return self.recording_ops.get_recording_manifest(
provider_name, recording_id, **kwargs
)
def get_recording_drm_configs(
self, provider_name: str, recording_id: str, **kwargs
) -> List:
# Reuses the same DRM resolution path as channels, events, and VOD.
return self.drm_ops.get_content_drm_configs(
provider_name, recording_id, **kwargs
)
def delete_recording(
self, provider_name: str, recording_id: str, **kwargs
) -> None:
return self.recording_ops.delete_recording(
provider_name, recording_id, **kwargs
)
# ==========================================================================
# SUBSCRIPTION OPERATIONS (delegate to SubscriptionOperations)
# ==========================================================================
lib/streaming_providers/base/models/recording.py
+325
View File
@@ -0,0 +1,325 @@
# streaming_providers/base/models/recording.py
"""
Recording model.
A Recording represents a captured broadcast — it was recorded from a live
channel (like an Event) but is consumed as on-demand content (like a VodItem).
Because it straddles both concerns without cleanly fitting either, it inherits
directly from Content and adds recording-specific fields.
Mapping to PVR API fields is noted in inline comments where names diverge.
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional
from .content import Content
class RecordingStatus(Enum):
"""Lifecycle state of the recording itself (not the broadcast)."""
PENDING = "PENDING" # scheduled to be recorded
RECORDING = "RECORDING" # currently being captured
COMPLETED = "COMPLETED" # fully captured and available
FAILED = "FAILED" # capture failed or was interrupted
DELETED = "DELETED" # marked deleted; may still be recoverable
class ChannelType(Enum):
"""Maps to PVR_RECORDING_CHANNEL_TYPE."""
UNKNOWN = "UNKNOWN"
TV = "TV"
RADIO = "RADIO"
@dataclass
class Recording(Content):
"""
A captured broadcast recording.
Inherits all streaming/DRM/manifest fields from Content so that playback
is resolved the same way as channels, events, and VOD items:
provider.get_manifest(content_id) # content_id == recording_id
provider.get_drm(content_id)
Fields are a union of:
- PVR recording API fields (see table; PVR name noted in comments)
- Fields shared with Event (timing, channel provenance)
- Fields shared with VodItem (episode/series context, playback state)
- Storage / management fields unique to recordings
"""
# ------------------------------------------------------------------
# Identity / provenance
# ------------------------------------------------------------------
# content_id (inherited) == recording_id → SetRecordingId / GetRecordingId
# The title is already `name` on Content, but PVR distinguishes a top-level
# title from an episode name within a series.
episode_name: Optional[str] = None # SetEpisodeName / GetEpisodeName
# Channel the recording was captured from
channel_name: Optional[str] = None # SetChannelName / GetChannelName
channel_uid: Optional[int] = None # SetChannelUid / GetChannelUid
channel_type: ChannelType = ChannelType.TV # SetChannelType / GetChannelType
# EPG linkage — lets the UI look up the original broadcast metadata
epg_event_id: Optional[int] = None # SetEPGEventId / GetEPGEventId (unsigned int)
# ------------------------------------------------------------------
# Timing (mirrors Event.start_time / end_time)
# ------------------------------------------------------------------
recording_time: Optional[datetime] = None # SetRecordingTime / GetRecordingTime
duration_seconds: Optional[int] = None # SetDuration / GetDuration (seconds)
first_aired: Optional[str] = None # SetFirstAired / GetFirstAired (ISO date string)
# ------------------------------------------------------------------
# Series / episode context (mirrors VodItem)
# ------------------------------------------------------------------
season_number: Optional[int] = None # SetSeriesNumber / GetSeriesNumber
episode_number: Optional[int] = None # SetEpisodeNumber / GetEpisodeNumber
series_title: Optional[str] = None # not in PVR table; added for UI grouping
series_id: Optional[str] = None # not in PVR table; added for back-navigation
release_year: Optional[int] = None # SetYear / GetYear
# ------------------------------------------------------------------
# Descriptions / metadata
# ------------------------------------------------------------------
# description (short) is inherited from Content
plot: Optional[str] = None # SetPlot / GetPlot (long description)
plot_outline: Optional[str] = None # SetPlotOutline / GetPlotOutline (short summary)
genre_description: Optional[str] = None # SetGenreDescription / GetGenreDescription
genre_type: Optional[int] = None # SetGenreType / GetGenreType (DVB genre code)
genre_sub_type: Optional[int] = None # SetGenreSubType / GetGenreSubType
# ------------------------------------------------------------------
# Visual assets
# ------------------------------------------------------------------
# logo_url (inherited from Content) ≈ icon_path
icon_path: Optional[str] = None # SetIconPath / GetIconPath
thumbnail_url: Optional[str] = None # SetThumbnailPath / GetThumbnailPath
fanart_url: Optional[str] = None # SetFanartPath / GetFanartPath
# ------------------------------------------------------------------
# Playback state (unique to recordings; events and VOD don't track this)
# ------------------------------------------------------------------
play_count: int = 0 # SetPlayCount / GetPlayCount
last_played_position: int = 0 # SetLastPlayedPosition / GetLastPlayedPosition (seconds)
# ------------------------------------------------------------------
# Storage / management
# ------------------------------------------------------------------
directory: Optional[str] = None # SetDirectory / GetDirectory
size_in_bytes: Optional[int] = None # SetSizeInBytes / GetSizeInBytes
priority: Optional[int] = None # SetPriority / GetPriority
lifetime: Optional[int] = None # SetLifetime / GetLifetime (days; 0 = keep forever)
flags: Optional[str] = None # SetFlags / GetFlags
# Provider identification (separate from the streaming provider on Content)
client_provider_uid: Optional[int] = None # SetClientProviderUid / GetClientProviderUid
provider_name: Optional[str] = None # SetProviderName / GetProviderName
# ------------------------------------------------------------------
# Status
# ------------------------------------------------------------------
status: RecordingStatus = RecordingStatus.COMPLETED
# is_deleted is derived from status but exposed as a convenience flag to
# match the PVR API; setting it syncs the status enum.
_is_deleted: bool = field(default=False, repr=False)
# ------------------------------------------------------------------
# Post-init
# ------------------------------------------------------------------
def __post_init__(self):
# Recordings are always played on-demand
self.mode = "vod"
self.on_demand = True
# Sync is_deleted → status
if self._is_deleted and self.status != RecordingStatus.DELETED:
self.status = RecordingStatus.DELETED
# ------------------------------------------------------------------
# Properties
# ------------------------------------------------------------------
@property
def recording_id(self) -> str:
"""Semantic alias for content_id — maps to PVR GetRecordingId."""
return self.content_id
@recording_id.setter
def recording_id(self, value: str):
self.content_id = value
@property
def is_deleted(self) -> bool:
"""SetIsDeleted / GetIsDeleted — synced with RecordingStatus.DELETED."""
return self.status == RecordingStatus.DELETED
@is_deleted.setter
def is_deleted(self, value: bool):
if value:
self.status = RecordingStatus.DELETED
elif self.status == RecordingStatus.DELETED:
self.status = RecordingStatus.COMPLETED
@property
def is_episode(self) -> bool:
return self.season_number is not None or self.episode_number is not None
@property
def duration_minutes(self) -> Optional[int]:
if self.duration_seconds is not None:
return self.duration_seconds // 60
return None
@property
def is_watched(self) -> bool:
"""True if the recording has been played at least once."""
return self.play_count > 0
@property
def is_in_progress(self) -> bool:
"""True if the recording is currently being captured."""
return self.status == RecordingStatus.RECORDING
# ------------------------------------------------------------------
# Serialisation
# ------------------------------------------------------------------
def to_dict(self) -> Dict:
result = super().to_dict()
result.update({
# Identity
"EpisodeName": self.episode_name,
"ChannelName": self.channel_name,
"ChannelUid": self.channel_uid,
"ChannelType": self.channel_type.value,
"EpgEventId": self.epg_event_id,
# Timing
"RecordingTime": self.recording_time.isoformat() if self.recording_time else None,
"DurationSeconds": self.duration_seconds,
"DurationMinutes": self.duration_minutes,
"FirstAired": self.first_aired,
# Series / episode
"SeasonNumber": self.season_number,
"EpisodeNumber": self.episode_number,
"SeriesTitle": self.series_title,
"SeriesId": self.series_id,
"ReleaseYear": self.release_year,
# Descriptions
"Plot": self.plot,
"PlotOutline": self.plot_outline,
"GenreDescription": self.genre_description,
"GenreType": self.genre_type,
"GenreSubType": self.genre_sub_type,
# Visual
"IconPath": self.icon_path,
"ThumbnailUrl": self.thumbnail_url,
"FanartUrl": self.fanart_url,
# Playback state
"PlayCount": self.play_count,
"LastPlayedPosition": self.last_played_position,
# Storage
"Directory": self.directory,
"SizeInBytes": self.size_in_bytes,
"Priority": self.priority,
"Lifetime": self.lifetime,
"Flags": self.flags,
"ClientProviderUid": self.client_provider_uid,
"ProviderName": self.provider_name,
# Status
"Status": self.status.value,
"IsDeleted": self.is_deleted,
})
return result
# ------------------------------------------------------------------
# Validation
# ------------------------------------------------------------------
def validate(self) -> List[str]:
warnings = []
if not self.manifest and not self.manifest_script:
warnings.append("No manifest URL or manifest script provided")
if self.license_url and not self.drm_config:
warnings.append("License URL provided but no DRM configuration")
if self.duration_seconds is not None and self.duration_seconds <= 0:
warnings.append("duration_seconds must be positive")
if self.size_in_bytes is not None and self.size_in_bytes < 0:
warnings.append("size_in_bytes must not be negative")
if self.lifetime is not None and self.lifetime < 0:
warnings.append("lifetime must be 0 (keep forever) or a positive number of days")
if self.release_year is not None and not (1888 <= self.release_year <= 2100):
warnings.append(f"Unusual release_year: {self.release_year}")
return warnings
# ------------------------------------------------------------------
# Factory methods
# ------------------------------------------------------------------
@classmethod
def create_completed(
cls,
name: str,
recording_id: str,
provider: str,
recording_time: Optional[datetime] = None,
duration_seconds: Optional[int] = None,
**kwargs,
) -> "Recording":
"""Create a fully captured recording ready for playback."""
return cls(
name=name,
content_id=recording_id,
provider=provider,
status=RecordingStatus.COMPLETED,
recording_time=recording_time,
duration_seconds=duration_seconds,
**kwargs,
)
@classmethod
def create_in_progress(
cls,
name: str,
recording_id: str,
provider: str,
recording_time: Optional[datetime] = None,
**kwargs,
) -> "Recording":
"""Create a recording that is currently being captured (live recording)."""
return cls(
name=name,
content_id=recording_id,
provider=provider,
status=RecordingStatus.RECORDING,
recording_time=recording_time,
**kwargs,
)
@classmethod
def create_episode_recording(
cls,
name: str,
recording_id: str,
provider: str,
season_number: int,
episode_number: int,
series_title: Optional[str] = None,
**kwargs,
) -> "Recording":
"""Create a recording of a specific series episode."""
return cls(
name=name,
content_id=recording_id,
provider=provider,
status=RecordingStatus.COMPLETED,
season_number=season_number,
episode_number=episode_number,
series_title=series_title,
**kwargs,
)
lib/streaming_providers/base/provider.py
+59
View File
@@ -18,6 +18,7 @@ from ..providers.auth import AuthContext, AuthStatus
from .models.proxy_models import ProxyConfig
from .models import DRMConfig, Event, StreamingChannel
from .models.subscription import SubscriptionPackage, UserSubscription
from .models.recording import Recording
from .network import HTTPManager, HTTPManagerFactory
from .utils.logger import logger
@@ -666,6 +667,64 @@ class StreamingProvider(ABC):
"""
return []
# =========================================================================
# RECORDINGS
# =========================================================================
@property
def implements_recordings(self) -> bool:
"""
True if this provider can return recorded content.
Return False (default) for providers that only offer live or VOD content.
RecordingOperations skips providers where this returns False when
aggregating across all providers.
"""
return False
def get_recordings(
self,
include_deleted: bool = False,
**kwargs,
) -> List[Recording]:
"""
Return recordings available for the authenticated user.
Args:
include_deleted: If True, also return recordings marked as deleted
(useful for a trash/recycle-bin view).
**kwargs: Provider-specific filtering options.
Returns:
List of Recording objects, or [] if recordings are not supported.
"""
return []
def delete_recording(self, recording_id: str, **kwargs) -> None:
"""
Permanently delete a recording on the provider's backend.
Args:
recording_id: The recording to delete (== content_id on the model).
**kwargs: Provider-specific arguments.
Returns:
None on success.
Raises:
KeyError: If no recording with this ID exists.
RuntimeError: If the provider refuses the deletion (e.g. the
recording is currently being captured, or the user
lacks permission).
Default implementation raises NotImplementedError so misconfigured
providers fail loudly rather than silently doing nothing.
"""
raise NotImplementedError(
f"{self.__class__.__name__} does not implement delete_recording(). "
"Override this method to support recording deletion."
)
# ============================================================================
# CATCHUP ABSTRACT METHODS
# ============================================================================
lib/streaming_providers/base/recording_operations.py
+148
View File
@@ -0,0 +1,148 @@
# streaming_providers/base/recording_operations.py
"""
Recording-related operations separated from core registry.
Mirrors the structure of EventOperations.
"""
from typing import Dict, List, Optional
from .models.recording import Recording
from .utils.logger import logger
class RecordingOperations:
"""Handles all recording-related operations."""
def __init__(self, registry):
self.registry = registry
logger.debug("RecordingOperations: Initialized")
def get_recordings(
self,
provider_name: str,
include_deleted: bool = False,
) -> List[Recording]:
"""
Get recordings from a specific provider.
Args:
provider_name: Name of the provider to query.
include_deleted: If True, include recordings marked as deleted.
Returns:
List of Recording objects.
Raises:
ValueError: If the provider is not found or disabled.
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
recordings = provider.get_recordings(include_deleted=include_deleted)
if not include_deleted:
recordings = [r for r in recordings if not r.is_deleted]
logger.info(
f"Retrieved {len(recordings)} recordings from '{provider_name}'"
)
return recordings
def get_all_recordings(
self,
include_deleted: bool = False,
) -> Dict[str, List[Recording]]:
"""
Get recordings from all enabled providers.
Args:
include_deleted: If True, include recordings marked as deleted.
Returns:
Dict mapping provider name → list of Recording objects.
"""
enabled = self.registry.get_enabled_providers()
logger.info(f"Fetching recordings from {len(enabled)} providers")
result = {}
total = 0
for name in enabled:
try:
recordings = self.get_recordings(
name, include_deleted=include_deleted
)
result[name] = recordings
total += len(recordings)
except Exception as e:
logger.error(f"Failed to get recordings from '{name}': {e}")
result[name] = []
logger.info(f"Retrieved {total} total recordings")
return result
def get_recording_manifest(
self, provider_name: str, recording_id: str, **kwargs
) -> Optional[str]:
"""
Get manifest URL for a specific recording.
Delegates to provider.get_manifest() — the same method used for
channels and events — because manifest resolution is content-type-agnostic.
Args:
provider_name: Name of the provider.
recording_id: Recording identifier (== content_id on the model).
**kwargs: Additional provider-specific arguments.
Returns:
Manifest URL string, or None if unavailable.
Raises:
ValueError: If the provider is not found or disabled.
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
manifest_url = provider.get_manifest(content_id=recording_id, **kwargs)
if manifest_url:
logger.debug(
f"Retrieved manifest for recording '{recording_id}' "
f"from '{provider_name}'"
)
return manifest_url
def delete_recording(
self, provider_name: str, recording_id: str, **kwargs
) -> None:
"""
Permanently delete a recording on the provider.
Delegates to provider.delete_recording(recording_id). The provider is
responsible for the actual API call; this layer only resolves the
provider instance and handles registry-level errors.
Args:
provider_name: Name of the provider.
recording_id: Recording identifier (== content_id on the model).
**kwargs: Additional provider-specific arguments.
Returns:
None on success.
Raises:
ValueError: If the provider is not found or disabled.
KeyError: If the recording does not exist on the provider.
RuntimeError: If the provider rejects the deletion (e.g. permission
denied, recording currently in progress).
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
provider.delete_recording(recording_id, **kwargs)
logger.info(
f"Deleted recording '{recording_id}' from '{provider_name}'"
)
routes/__init__.py
+4
View File
@@ -19,6 +19,8 @@ from .drm import setup_drm_routes
from .epg import setup_epg_routes
from .m3u import setup_m3u_routes
from .events import setup_events_routes
from .vod import setup_vod_routes
from .recordings import setup_recordings_routes
# Now import the route setup functions
from .providers import setup_provider_routes
@@ -33,4 +35,6 @@ __all__ = [
"setup_config_routes",
"setup_epg_routes",
"setup_events_routes",
"setup_vod_routes",
"setup_recordings_routes",
]
routes/recordings.py
+200
View File
@@ -0,0 +1,200 @@
#!/usr/bin/env python3
"""
Recording-related route handlers.
Mirrors the structure of events.py.
"""
from bottle import request, response
from streaming_providers.base.utils import logger
def setup_recordings_routes(app, manager, service):
"""Setup recording-related routes."""
@app.route("/api/providers/<provider>/recordings", method="GET")
def get_provider_recordings(provider):
"""
Get recordings from a specific provider.
Query parameters:
- include_deleted: Optional bool (true/false) — include deleted recordings.
Defaults to false.
Returns:
{
"provider": "provider_name",
"recordings": [
{ ...Recording fields... }
],
"count": 1
}
"""
try:
# Parse optional query parameters
include_deleted_str = request.params.get("include_deleted", "false").lower()
include_deleted = include_deleted_str in ("1", "true", "yes")
try:
recordings = manager.recording_ops.get_recordings(
provider_name=provider,
include_deleted=include_deleted,
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except Exception as e:
logger.error(f"Failed to get recordings from '{provider}': {e}")
response.status = 500
return {
"error": "Failed to get recordings",
"message": str(e),
"provider": provider,
}
serialized = [recording.to_dict() for recording in recordings]
response.status = 200
return {
"provider": provider,
"recordings": serialized,
"count": len(serialized),
}
except Exception as e:
logger.error(f"Unexpected error in get_provider_recordings: {e}")
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
}
@app.route("/api/providers/<provider>/recordings/<recording_id>", method="GET")
def get_provider_recording(provider, recording_id):
"""
Get a single recording by ID from a specific provider.
Returns:
{ ...Recording fields... }
"""
try:
try:
recordings = manager.recording_ops.get_recordings(
provider_name=provider,
include_deleted=True, # include deleted so a 404 is explicit
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except Exception as e:
logger.error(
f"Failed to get recordings from '{provider}': {e}"
)
response.status = 500
return {
"error": "Failed to get recordings",
"message": str(e),
"provider": provider,
}
match = next(
(r for r in recordings if r.recording_id == recording_id), None
)
if not match:
response.status = 404
return {
"error": "Recording not found",
"message": f"No recording with id '{recording_id}' from '{provider}'",
"provider": provider,
"recording_id": recording_id,
}
response.status = 200
return match.to_dict()
except Exception as e:
logger.error(f"Unexpected error in get_provider_recording: {e}")
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
}
@app.route("/api/providers/<provider>/recordings/<recording_id>", method="DELETE")
def delete_provider_recording(provider, recording_id):
"""
Permanently delete a single recording from a specific provider.
Returns:
204 No Content on success.
404 if the provider or recording is not found.
409 if the provider refuses the deletion (e.g. recording in progress).
500 on unexpected errors.
"""
try:
try:
manager.recording_ops.delete_recording(
provider_name=provider,
recording_id=recording_id,
)
except ValueError as e:
# Raised by RecordingOperations when provider is not found
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except KeyError as e:
# Raised by the provider when the recording ID does not exist
response.status = 404
return {
"error": "Recording not found",
"message": str(e),
"provider": provider,
"recording_id": recording_id,
}
except RuntimeError as e:
# Raised by the provider when deletion is refused
# (e.g. recording currently in progress, insufficient permissions)
response.status = 409
return {
"error": "Deletion refused by provider",
"message": str(e),
"provider": provider,
"recording_id": recording_id,
}
except Exception as e:
logger.error(
f"Failed to delete recording '{recording_id}' "
f"from '{provider}': {e}"
)
response.status = 500
return {
"error": "Failed to delete recording",
"message": str(e),
"provider": provider,
"recording_id": recording_id,
}
# 204 No Content — no body
response.status = 204
return ""
except Exception as e:
logger.error(f"Unexpected error in delete_provider_recording: {e}")
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
}
service.py
+2
View File
@@ -1620,6 +1620,7 @@ class UltimateService:
from routes.epg import setup_epg_routes
from routes.events import setup_events_routes
from routes.vod import setup_vod_routes
from routes.recordings import setup_recordings_routes
# Setup routes from separate modules
setup_provider_routes(self.app, self.manager, self)
@@ -1631,6 +1632,7 @@ class UltimateService:
setup_epg_routes(self.app, self.manager, self)
setup_events_routes(self.app, self.manager, self)
setup_vod_routes(self.app, self.manager)
setup_recordings_routes(self.app, self.manager, self)
# Core UI routes
@self.app.route("/config")