nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-17 06:32:30 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add timer support

Browse Source
This commit is contained in:
Nirvana
2026-04-02 16:13:45 +02:00
parent 44c1a944a3
commit d5aaecd8e0
8 changed files with 1930 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/manager.py
+44
View File
@@ -13,6 +13,7 @@ from .epg_operations import EPGOperations
from .event_operations import EventOperations
from .vod_operations import VodOperations
from .recording_operations import RecordingOperations
from .timer_operations import TimerOperations
from .models import StreamingChannel
from .provider_registry import ProviderRegistry
from .subscription_operations import SubscriptionOperations
@@ -38,6 +39,7 @@ class ProviderManager:
self.event_ops = EventOperations(self.registry)
self.vod_ops = VodOperations(self.registry)
self.recording_ops = RecordingOperations(self.registry)
self.timer_ops = TimerOperations(self.registry)
# Backward compatibility - expose managers directly
self.drm_plugin_manager = self.drm_ops.drm_plugin_manager
@@ -303,6 +305,48 @@ class ProviderManager:
provider_name, recording_id, **kwargs
)
# ==========================================================================
# TIMER OPERATIONS (delegate to TimerOperations)
# ==========================================================================
def get_timer_types(self, provider_name: str) -> List:
return self.timer_ops.get_timer_types(provider_name)
def get_all_timer_types(self) -> Dict[str, List]:
return self.timer_ops.get_all_timer_types()
def get_timers(
self,
provider_name: str,
include_inactive: bool = False,
) -> List:
return self.timer_ops.get_timers(
provider_name, include_inactive=include_inactive
)
def get_all_timers(self, include_inactive: bool = False) -> Dict[str, List]:
return self.timer_ops.get_all_timers(include_inactive=include_inactive)
def get_timer(self, provider_name: str, client_index: int):
return self.timer_ops.get_timer(provider_name, client_index)
def add_timer(self, provider_name: str, timer, **kwargs):
return self.timer_ops.add_timer(provider_name, timer, **kwargs)
def update_timer(self, provider_name: str, timer, **kwargs):
return self.timer_ops.update_timer(provider_name, timer, **kwargs)
def delete_timer(
self,
provider_name: str,
client_index: int,
force_delete: bool = False,
**kwargs,
) -> None:
return self.timer_ops.delete_timer(
provider_name, client_index, force_delete=force_delete, **kwargs
)
# ==========================================================================
# SUBSCRIPTION OPERATIONS (delegate to SubscriptionOperations)
# ==========================================================================
lib/streaming_providers/base/models/timer.py
+452
View File
@@ -0,0 +1,452 @@
# streaming_providers/base/models/timer.py
"""
Timer model.
A Timer represents a scheduled recording instruction — it tells the backend
*when* and *what* to record. Unlike a Recording (which is captured content
that can be played back), a Timer is purely a management object: it has no
manifest and no DRM. It therefore does NOT inherit from Content.
Mapping to the PVR Timer API is noted in inline comments where names diverge.
Two companion enumerations are provided:
- TimerState — lifecycle state of the timer itself
- TimerWeekday — bitmask constants for recurring-timer day selection
"""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum, IntFlag
from typing import Dict, List, Optional
# ---------------------------------------------------------------------------
# Enumerations
# ---------------------------------------------------------------------------
class TimerState(Enum):
"""
Lifecycle state of the timer.
Maps to PVR_TIMER_STATE_* constants.
"""
NEW = "NEW" # just created, not yet saved to provider
SCHEDULED = "SCHEDULED" # saved, waiting for start time
RECORDING = "RECORDING" # capture is currently in progress
COMPLETED = "COMPLETED" # recording finished successfully
ABORTED = "ABORTED" # recording was manually stopped
CANCELLED = "CANCELLED" # timer was deleted before it fired
CONFLICT = "CONFLICT" # overlaps with another timer / resource issue
ERROR = "ERROR" # provider reported an error
class TimerWeekday(IntFlag):
"""
Bitmask for recurring timer weekday selection.
Maps to PVR_WEEKDAY_* constants.
Combine with bitwise OR: TimerWeekday.MONDAY | TimerWeekday.WEDNESDAY
"""
NONE = 0
MONDAY = 1
TUESDAY = 2
WEDNESDAY = 4
THURSDAY = 8
FRIDAY = 16
SATURDAY = 32
SUNDAY = 64
ALL_DAYS = MONDAY | TUESDAY | WEDNESDAY | THURSDAY | FRIDAY | SATURDAY | SUNDAY
WEEKDAYS = MONDAY | TUESDAY | WEDNESDAY | THURSDAY | FRIDAY
WEEKEND = SATURDAY | SUNDAY
class DuplicateHandling(Enum):
"""
Strategy for preventing duplicate episode recordings.
Maps to PVR_TIMER_DUPLICATE_PREVENTION_* values.
"""
NONE = 0 # record all, even duplicates
SAME_EPG_ID = 1 # skip if same EPG event ID already recorded
SAME_TITLE = 2 # skip if title matches an existing recording
SAME_SUBTITLE = 3 # skip if subtitle/episode name matches
SAME_DESCRIPTION = 4 # skip if description matches
# ---------------------------------------------------------------------------
# Timer dataclass
# ---------------------------------------------------------------------------
@dataclass
class Timer:
"""
A scheduled recording instruction.
Required fields (must be set for every timer):
client_index — unique identifier assigned by the provider
state — current lifecycle state
timer_type_id — references a TimerType.type_id
title — display name / series title
All other fields are optional and mirror PVR API capabilities.
"""
# ------------------------------------------------------------------
# Identity (required)
# ------------------------------------------------------------------
# PVR: SetClientIndex / GetClientIndex (unsigned int)
client_index: int = 0
# PVR: SetState / GetState
state: TimerState = TimerState.NEW
# PVR: SetTimerType / GetTimerType (unsigned int — references TimerType.type_id)
timer_type_id: int = 1
# PVR: SetTitle / GetTitle
title: str = ""
# Provider this timer belongs to (not a PVR field; used for routing)
provider: str = ""
# ------------------------------------------------------------------
# Hierarchy — series / parent timers
# ------------------------------------------------------------------
# PVR: SetParentClientIndex / GetParentClientIndex
# 0 = top-level timer (no parent)
parent_client_index: int = 0
# ------------------------------------------------------------------
# Channel
# ------------------------------------------------------------------
# PVR: SetClientChannelUid / GetClientChannelUid
# -1 = "any channel" (used by EPG-search timers)
client_channel_uid: int = -1
# Human-readable channel name for display (not in PVR table)
channel_name: Optional[str] = None
# ------------------------------------------------------------------
# Timing
# ------------------------------------------------------------------
# PVR: SetStartTime / GetStartTime (Unix timestamp; 0 = ASAP)
start_time: Optional[datetime] = None
# PVR: SetEndTime / GetEndTime (Unix timestamp; 0 = open-ended)
end_time: Optional[datetime] = None
# PVR: SetStartAnyTime / GetStartAnyTime
# True = ignore start_time; begin whenever the event is detected
start_any_time: bool = False
# PVR: SetEndAnyTime / GetEndAnyTime
# True = ignore end_time; record until the event ends naturally
end_any_time: bool = False
# PVR: SetFirstDay / GetFirstDay (Unix timestamp)
# For recurring timers: do not fire before this date
first_day: Optional[datetime] = None
# ------------------------------------------------------------------
# Pre/post padding
# ------------------------------------------------------------------
# PVR: SetMarginStart / GetMarginStart (minutes)
margin_start: int = 0
# PVR: SetMarginEnd / GetMarginEnd (minutes)
margin_end: int = 0
# ------------------------------------------------------------------
# EPG search / linkage
# ------------------------------------------------------------------
# PVR: SetEPGSearchString / GetEPGSearchString
# Used by search-based recurring timers (e.g. "record anything matching 'Tatort'")
epg_search_string: Optional[str] = None
# PVR: SetFullTextEpgSearch / GetFullTextEpgSearch
# True = match anywhere in title/description; False = title-only
full_text_epg_search: bool = False
# PVR: SetEPGUid / GetEPGUid (unsigned int)
# EPG event ID this timer was created from (0 = not linked to an EPG event)
epg_uid: int = 0
# Convenience: the full EPG event ID including provider prefix (not in PVR table)
epg_event_id: Optional[str] = None
# ------------------------------------------------------------------
# Recurring schedule
# ------------------------------------------------------------------
# PVR: SetWeekdays / GetWeekdays (bitmask of TimerWeekday)
weekdays: TimerWeekday = TimerWeekday.NONE
# PVR: SetPreventDuplicateEpisodes / GetPreventDuplicateEpisodes
prevent_duplicate_episodes: DuplicateHandling = DuplicateHandling.NONE
# Series link — used by some providers to group related timers
# PVR: SetSeriesLink / GetSeriesLink
series_link: Optional[str] = None
# ------------------------------------------------------------------
# Recording management
# ------------------------------------------------------------------
# PVR: SetDirectory / GetDirectory
directory: Optional[str] = None
# PVR: SetPriority / GetPriority
# Higher value = higher priority when resources are scarce
priority: int = 50
# PVR: SetLifetime / GetLifetime (days; 0 = keep forever)
lifetime: int = 0
# PVR: SetMaxRecordings / GetMaxRecordings
# 0 = unlimited; positive integer = keep only this many recordings
max_recordings: int = 0
# PVR: SetRecordingGroup / GetRecordingGroup (unsigned int)
recording_group: int = 0
# ------------------------------------------------------------------
# Genre metadata
# ------------------------------------------------------------------
# PVR: SetGenreType / GetGenreType (DVB genre code)
genre_type: Optional[int] = None
# PVR: SetGenreSubType / GetGenreSubType
genre_sub_type: Optional[int] = None
# ------------------------------------------------------------------
# Internal / bookkeeping (not in PVR table)
# ------------------------------------------------------------------
# When this timer record was last fetched or updated locally
last_updated: Optional[datetime] = None
# Free-form summary from the provider (maps to description in UI)
description: Optional[str] = None
# ------------------------------------------------------------------
# Post-init
# ------------------------------------------------------------------
def __post_init__(self):
# Normalise: a recurring timer with explicit weekdays but NONE set
# should fall back to the provider's default — we do not force it here
# but we expose a helper property for convenience.
pass
# ------------------------------------------------------------------
# Properties
# ------------------------------------------------------------------
@property
def is_recurring(self) -> bool:
"""True if this timer fires on multiple days (weekday mask is set)."""
return self.weekdays != TimerWeekday.NONE
@property
def is_epg_based(self) -> bool:
"""True if this timer was created from or linked to an EPG event."""
return bool(self.epg_uid) or bool(self.epg_event_id)
@property
def is_search_based(self) -> bool:
"""True if this timer uses an EPG keyword search."""
return bool(self.epg_search_string)
@property
def is_active(self) -> bool:
"""True if the timer is still expected to produce a recording."""
return self.state in (TimerState.NEW, TimerState.SCHEDULED, TimerState.RECORDING)
@property
def duration_minutes(self) -> Optional[int]:
"""Planned recording duration in minutes (None if open-ended)."""
if self.start_time and self.end_time:
delta = self.end_time - self.start_time
return int(delta.total_seconds() // 60)
return None
# ------------------------------------------------------------------
# Serialisation
# ------------------------------------------------------------------
def to_dict(self) -> Dict:
return {
# Identity
"ClientIndex": self.client_index,
"State": self.state.value,
"TimerTypeId": self.timer_type_id,
"Title": self.title,
"Provider": self.provider,
# Hierarchy
"ParentClientIndex": self.parent_client_index,
# Channel
"ClientChannelUid": self.client_channel_uid,
"ChannelName": self.channel_name,
# Timing
"StartTime": self.start_time.isoformat() if self.start_time else None,
"EndTime": self.end_time.isoformat() if self.end_time else None,
"StartAnyTime": self.start_any_time,
"EndAnyTime": self.end_any_time,
"FirstDay": self.first_day.isoformat() if self.first_day else None,
"DurationMinutes": self.duration_minutes,
# Padding
"MarginStart": self.margin_start,
"MarginEnd": self.margin_end,
# EPG
"EpgSearchString": self.epg_search_string,
"FullTextEpgSearch": self.full_text_epg_search,
"EpgUid": self.epg_uid,
"EpgEventId": self.epg_event_id,
# Recurring
"Weekdays": self.weekdays.value,
"PreventDuplicateEpisodes": self.prevent_duplicate_episodes.value,
"SeriesLink": self.series_link,
# Management
"Directory": self.directory,
"Priority": self.priority,
"Lifetime": self.lifetime,
"MaxRecordings": self.max_recordings,
"RecordingGroup": self.recording_group,
# Genre
"GenreType": self.genre_type,
"GenreSubType": self.genre_sub_type,
# Convenience flags
"IsRecurring": self.is_recurring,
"IsEpgBased": self.is_epg_based,
"IsSearchBased": self.is_search_based,
"IsActive": self.is_active,
# Metadata
"Description": self.description,
"LastUpdated": self.last_updated.isoformat() if self.last_updated else None,
}
# ------------------------------------------------------------------
# Validation
# ------------------------------------------------------------------
def validate(self) -> List[str]:
warnings = []
if not self.title:
warnings.append("title must not be empty")
if self.timer_type_id <= 0:
warnings.append("timer_type_id must be a positive integer")
if self.start_time and self.end_time and self.start_time >= self.end_time:
warnings.append("start_time must be before end_time")
if self.margin_start < 0:
warnings.append("margin_start must not be negative")
if self.margin_end < 0:
warnings.append("margin_end must not be negative")
if self.priority < 0 or self.priority > 100:
warnings.append("priority should be between 0 and 100")
if self.lifetime < 0:
warnings.append("lifetime must be 0 (keep forever) or a positive number of days")
if self.max_recordings < 0:
warnings.append("max_recordings must not be negative")
if self.is_recurring and self.epg_uid:
warnings.append(
"timer has both weekdays and epg_uid set; "
"recurring EPG-based timers should use series_link instead"
)
return warnings
# ------------------------------------------------------------------
# Factory methods
# ------------------------------------------------------------------
@classmethod
def create_one_shot(
cls,
title: str,
provider: str,
client_channel_uid: int,
start_time: datetime,
end_time: datetime,
timer_type_id: int = 1,
**kwargs,
) -> "Timer":
"""Create a single-event timer for a specific channel and time window."""
return cls(
title=title,
provider=provider,
client_channel_uid=client_channel_uid,
start_time=start_time,
end_time=end_time,
timer_type_id=timer_type_id,
state=TimerState.SCHEDULED,
**kwargs,
)
@classmethod
def create_recurring(
cls,
title: str,
provider: str,
client_channel_uid: int,
weekdays: TimerWeekday,
start_time: datetime,
end_time: datetime,
timer_type_id: int = 1,
prevent_duplicate_episodes: DuplicateHandling = DuplicateHandling.SAME_EPG_ID,
**kwargs,
) -> "Timer":
"""Create a recurring weekly timer (e.g. record every Monday at 20:15)."""
return cls(
title=title,
provider=provider,
client_channel_uid=client_channel_uid,
weekdays=weekdays,
start_time=start_time,
end_time=end_time,
timer_type_id=timer_type_id,
prevent_duplicate_episodes=prevent_duplicate_episodes,
state=TimerState.SCHEDULED,
**kwargs,
)
@classmethod
def create_epg_based(
cls,
title: str,
provider: str,
client_channel_uid: int,
epg_uid: int,
start_time: datetime,
end_time: datetime,
timer_type_id: int = 1,
**kwargs,
) -> "Timer":
"""Create a timer tied to a specific EPG event."""
return cls(
title=title,
provider=provider,
client_channel_uid=client_channel_uid,
epg_uid=epg_uid,
start_time=start_time,
end_time=end_time,
timer_type_id=timer_type_id,
state=TimerState.SCHEDULED,
**kwargs,
)
@classmethod
def create_search_based(
cls,
title: str,
provider: str,
epg_search_string: str,
timer_type_id: int = 1,
client_channel_uid: int = -1,
full_text_epg_search: bool = False,
prevent_duplicate_episodes: DuplicateHandling = DuplicateHandling.SAME_SUBTITLE,
**kwargs,
) -> "Timer":
"""Create a keyword-search timer that records any matching EPG event."""
return cls(
title=title,
provider=provider,
client_channel_uid=client_channel_uid,
epg_search_string=epg_search_string,
full_text_epg_search=full_text_epg_search,
prevent_duplicate_episodes=prevent_duplicate_episodes,
timer_type_id=timer_type_id,
state=TimerState.SCHEDULED,
**kwargs,
)
lib/streaming_providers/base/models/timer_type.py
+338
View File
@@ -0,0 +1,338 @@
# streaming_providers/base/models/timer_type.py
"""
Timer type model.
A TimerType describes a *category* of timer a provider supports — not a
single scheduled recording, but the template that governs what options are
available when creating one. Providers expose their supported timer types via
get_timer_types(); clients use this information to build a creation UI and to
know which timer_type_id to pass when calling add_timer().
Mapping to PVRTimerType is noted in inline comments.
"""
from dataclasses import dataclass, field
from typing import Dict, List
# ---------------------------------------------------------------------------
# Attribute bitmask constants
# (mirrors PVR_TIMER_TYPE_ATTRIBUTE_* from the Kodi PVR API)
# ---------------------------------------------------------------------------
class TimerTypeAttribute:
"""
Bitmask constants describing the capabilities of a TimerType.
Combine with bitwise OR to build the `attributes` field.
"""
NONE = 0x00000000
IS_MANUAL = 0x00000001 # timer is set manually (not EPG-based)
IS_REPEATING = 0x00000002 # fires on multiple days / events
IS_EPG_BASED = 0x00000004 # linked to a specific EPG event
SUPPORTS_ENABLE_DISABLE = 0x00000008 # timer can be enabled/disabled without deleting
SUPPORTS_CHANNELS = 0x00000010 # channel selection is meaningful
SUPPORTS_START_TIME = 0x00000020 # start time can be specified
SUPPORTS_START_ANY_TIME = 0x00000040 # "start any time" flag is supported
SUPPORTS_END_TIME = 0x00000080 # end time can be specified
SUPPORTS_END_ANY_TIME = 0x00000100 # "end any time" flag is supported
SUPPORTS_FIRST_DAY = 0x00000200 # first-day constraint is supported
SUPPORTS_WEEKDAYS = 0x00000400 # weekday mask is supported
SUPPORTS_EPG_SEARCH = 0x00000800 # keyword EPG search is supported
SUPPORTS_FULL_TEXT_SEARCH = 0x00001000 # full-text EPG search is supported
SUPPORTS_RECORDING_FOLDERS = 0x00002000 # directory selection is supported
SUPPORTS_PRIORITIES = 0x00004000 # priority selection is supported
SUPPORTS_LIFETIMES = 0x00008000 # lifetime selection is supported
SUPPORTS_PADDING = 0x00010000 # pre/post margin padding is supported
SUPPORTS_RECORDING_GROUP = 0x00020000 # recording group assignment is supported
SUPPORTS_MAX_RECORDINGS = 0x00040000 # max-recordings limit is supported
SUPPORTS_DUPLICATE_CHECK = 0x00080000 # duplicate-episode prevention is supported
REQUIRES_EPG_TAG_ON_CREATE = 0x00100000 # an EPG tag must be provided at creation time
FORBIDS_EPG_TAG_ON_CREATE = 0x00200000 # EPG tag must NOT be provided at creation time
# ---------------------------------------------------------------------------
# Helper: labelled integer option (used for dropdown selections)
# ---------------------------------------------------------------------------
@dataclass
class TimerTypeIntOption:
"""
A single labelled option in a numeric selection list.
Mirrors PVRTypeIntValue used for priority, lifetime, etc.
"""
value: int
description: str
def to_dict(self) -> Dict:
return {"Value": self.value, "Description": self.description}
# ---------------------------------------------------------------------------
# TimerType dataclass
# ---------------------------------------------------------------------------
@dataclass
class TimerType:
"""
Describes a category of timer supported by a provider.
Required fields:
type_id — unique identifier within the provider (PVR: SetId / GetId)
attributes — bitmask of TimerTypeAttribute constants
Optional fields control which UI elements appear when the type is selected
and what default values are pre-filled.
"""
# ------------------------------------------------------------------
# Identity (required)
# ------------------------------------------------------------------
# PVR: SetId / GetId (unsigned int; must be > 0)
type_id: int = 1
# PVR: SetAttributes / GetAttributes (bitmask of TimerTypeAttribute)
attributes: int = TimerTypeAttribute.NONE
# ------------------------------------------------------------------
# Display
# ------------------------------------------------------------------
# PVR: SetDescription / GetDescription
description: str = ""
# ------------------------------------------------------------------
# Priority selection
# ------------------------------------------------------------------
# PVR: SetPriorities / GetPriorities
# List of (value, label) options presented in the priority dropdown.
# Empty list = free-text integer input.
priority_options: List[TimerTypeIntOption] = field(default_factory=list)
# PVR: SetPrioritiesDefault / GetPrioritiesDefault
priority_default: int = 50
# ------------------------------------------------------------------
# Lifetime selection
# ------------------------------------------------------------------
# PVR: SetLifetimes / GetLifetimes
lifetime_options: List[TimerTypeIntOption] = field(default_factory=list)
# PVR: SetLifetimesDefault / GetLifetimesDefault (days; 0 = keep forever)
lifetime_default: int = 0
# ------------------------------------------------------------------
# Duplicate-episode prevention selection
# ------------------------------------------------------------------
# PVR: SetPreventDuplicateEpisodes / GetPreventDuplicateEpisodes
prevent_duplicate_options: List[TimerTypeIntOption] = field(default_factory=list)
# PVR: SetPreventDuplicateEpisodesDefault / GetPreventDuplicateEpisodesDefault
prevent_duplicate_default: int = 0
# ------------------------------------------------------------------
# Recording group selection
# ------------------------------------------------------------------
# PVR: SetRecordingGroups / GetRecordingGroups
recording_group_options: List[TimerTypeIntOption] = field(default_factory=list)
# PVR: SetRecordingGroupDefault / GetRecordingGroupDefault
recording_group_default: int = 0
# ------------------------------------------------------------------
# Max recordings selection
# ------------------------------------------------------------------
# PVR: SetMaxRecordings / GetMaxRecordings
max_recordings_options: List[TimerTypeIntOption] = field(default_factory=list)
# PVR: SetMaxRecordingsDefault / GetMaxRecordingsDefault (0 = unlimited)
max_recordings_default: int = 0
# ------------------------------------------------------------------
# Convenience attribute helpers
# ------------------------------------------------------------------
def supports(self, attribute: int) -> bool:
"""Check whether a specific attribute flag is set."""
return bool(self.attributes & attribute)
@property
def is_manual(self) -> bool:
return self.supports(TimerTypeAttribute.IS_MANUAL)
@property
def is_repeating(self) -> bool:
return self.supports(TimerTypeAttribute.IS_REPEATING)
@property
def is_epg_based(self) -> bool:
return self.supports(TimerTypeAttribute.IS_EPG_BASED)
@property
def supports_channels(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_CHANNELS)
@property
def supports_weekdays(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_WEEKDAYS)
@property
def supports_epg_search(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_EPG_SEARCH)
@property
def supports_padding(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_PADDING)
@property
def supports_priorities(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_PRIORITIES)
@property
def supports_lifetimes(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_LIFETIMES)
@property
def supports_duplicate_check(self) -> bool:
return self.supports(TimerTypeAttribute.SUPPORTS_DUPLICATE_CHECK)
# ------------------------------------------------------------------
# Serialisation
# ------------------------------------------------------------------
def to_dict(self) -> Dict:
return {
"TypeId": self.type_id,
"Attributes": self.attributes,
"Description": self.description,
# Priority
"PriorityOptions": [o.to_dict() for o in self.priority_options],
"PriorityDefault": self.priority_default,
# Lifetime
"LifetimeOptions": [o.to_dict() for o in self.lifetime_options],
"LifetimeDefault": self.lifetime_default,
# Duplicate handling
"PreventDuplicateOptions": [o.to_dict() for o in self.prevent_duplicate_options],
"PreventDuplicateDefault": self.prevent_duplicate_default,
# Recording group
"RecordingGroupOptions": [o.to_dict() for o in self.recording_group_options],
"RecordingGroupDefault": self.recording_group_default,
# Max recordings
"MaxRecordingsOptions": [o.to_dict() for o in self.max_recordings_options],
"MaxRecordingsDefault": self.max_recordings_default,
# Derived flags (convenience for clients that don't parse bitmasks)
"IsManual": self.is_manual,
"IsRepeating": self.is_repeating,
"IsEpgBased": self.is_epg_based,
"SupportsChannels": self.supports_channels,
"SupportsWeekdays": self.supports_weekdays,
"SupportsEpgSearch": self.supports_epg_search,
"SupportsPadding": self.supports_padding,
"SupportsPriorities": self.supports_priorities,
"SupportsLifetimes": self.supports_lifetimes,
"SupportsDuplicateCheck": self.supports_duplicate_check,
}
# ------------------------------------------------------------------
# Factory helpers for the most common timer type patterns
# ------------------------------------------------------------------
@classmethod
def make_manual_one_shot(
cls,
type_id: int = 1,
description: str = "Manual one-time recording",
**kwargs,
) -> "TimerType":
"""Record a specific channel between two explicit timestamps."""
return cls(
type_id=type_id,
description=description,
attributes=(
TimerTypeAttribute.IS_MANUAL
| TimerTypeAttribute.SUPPORTS_CHANNELS
| TimerTypeAttribute.SUPPORTS_START_TIME
| TimerTypeAttribute.SUPPORTS_END_TIME
| TimerTypeAttribute.SUPPORTS_PRIORITIES
| TimerTypeAttribute.SUPPORTS_LIFETIMES
| TimerTypeAttribute.SUPPORTS_PADDING
| TimerTypeAttribute.FORBIDS_EPG_TAG_ON_CREATE
),
**kwargs,
)
@classmethod
def make_epg_one_shot(
cls,
type_id: int = 2,
description: str = "EPG-based one-time recording",
**kwargs,
) -> "TimerType":
"""Record a single EPG event."""
return cls(
type_id=type_id,
description=description,
attributes=(
TimerTypeAttribute.IS_EPG_BASED
| TimerTypeAttribute.SUPPORTS_CHANNELS
| TimerTypeAttribute.SUPPORTS_START_TIME
| TimerTypeAttribute.SUPPORTS_END_TIME
| TimerTypeAttribute.SUPPORTS_PRIORITIES
| TimerTypeAttribute.SUPPORTS_LIFETIMES
| TimerTypeAttribute.SUPPORTS_PADDING
| TimerTypeAttribute.REQUIRES_EPG_TAG_ON_CREATE
),
**kwargs,
)
@classmethod
def make_manual_recurring(
cls,
type_id: int = 3,
description: str = "Manual recurring recording",
**kwargs,
) -> "TimerType":
"""Record the same time slot on selected weekdays."""
return cls(
type_id=type_id,
description=description,
attributes=(
TimerTypeAttribute.IS_MANUAL
| TimerTypeAttribute.IS_REPEATING
| TimerTypeAttribute.SUPPORTS_CHANNELS
| TimerTypeAttribute.SUPPORTS_START_TIME
| TimerTypeAttribute.SUPPORTS_END_TIME
| TimerTypeAttribute.SUPPORTS_WEEKDAYS
| TimerTypeAttribute.SUPPORTS_FIRST_DAY
| TimerTypeAttribute.SUPPORTS_PRIORITIES
| TimerTypeAttribute.SUPPORTS_LIFETIMES
| TimerTypeAttribute.SUPPORTS_PADDING
| TimerTypeAttribute.SUPPORTS_MAX_RECORDINGS
| TimerTypeAttribute.FORBIDS_EPG_TAG_ON_CREATE
),
**kwargs,
)
@classmethod
def make_epg_search(
cls,
type_id: int = 4,
description: str = "EPG keyword search recording",
**kwargs,
) -> "TimerType":
"""Record every EPG event whose title/description matches a keyword."""
return cls(
type_id=type_id,
description=description,
attributes=(
TimerTypeAttribute.IS_EPG_BASED
| TimerTypeAttribute.IS_REPEATING
| TimerTypeAttribute.SUPPORTS_CHANNELS
| TimerTypeAttribute.SUPPORTS_EPG_SEARCH
| TimerTypeAttribute.SUPPORTS_FULL_TEXT_SEARCH
| TimerTypeAttribute.SUPPORTS_PRIORITIES
| TimerTypeAttribute.SUPPORTS_LIFETIMES
| TimerTypeAttribute.SUPPORTS_DUPLICATE_CHECK
| TimerTypeAttribute.SUPPORTS_MAX_RECORDINGS
| TimerTypeAttribute.FORBIDS_EPG_TAG_ON_CREATE
),
**kwargs,
)
lib/streaming_providers/base/provider.py
+114
View File
@@ -19,6 +19,8 @@ from .models.proxy_models import ProxyConfig
from .models import DRMConfig, Event, StreamingChannel
from .models.subscription import SubscriptionPackage, UserSubscription
from .models.recording import Recording
from .models.timer import Timer
from .models.timer_type import TimerType
from .network import HTTPManager, HTTPManagerFactory
from .utils.logger import logger
@@ -749,6 +751,118 @@ class StreamingProvider(ABC):
"Override this method to support recording deletion."
)
# =========================================================================
# TIMERS
# =========================================================================
@property
def implements_timers(self) -> bool:
"""
True if this provider supports scheduled recording timers.
Return False (default) for providers that only offer live or VOD content
and have no PVR/timer backend. TimerOperations skips providers where
this returns False when aggregating across all providers.
"""
return False
def get_timer_types(self) -> List["TimerType"]:
"""
Return the timer types this provider supports.
Providers that support timers MUST override this method and return at
least one TimerType so that clients know what fields to present when
creating a timer.
Returns:
List of TimerType objects, or [] if timers are not supported.
"""
return []
def get_timers(self, **kwargs) -> List["Timer"]:
"""
Return all timers (scheduled recordings) for the authenticated user.
Args:
**kwargs: Provider-specific filtering options.
Returns:
List of Timer objects, or [] if timers are not supported.
"""
return []
def add_timer(self, timer: "Timer", **kwargs) -> "Timer":
"""
Schedule a new timer on the provider's backend.
Args:
timer: Timer to create. timer.client_index is ignored — the
provider allocates and sets it on the returned object.
**kwargs: Provider-specific arguments.
Returns:
The saved Timer with client_index populated by the provider.
Raises:
RuntimeError: If the provider rejects the timer (e.g. scheduling
conflict, unsupported timer type, insufficient
permissions).
Default raises NotImplementedError so misconfigured providers fail
loudly rather than silently doing nothing.
"""
raise NotImplementedError(
f"{self.__class__.__name__} does not implement add_timer(). "
"Override this method to support timer creation."
)
def update_timer(self, timer: "Timer", **kwargs) -> "Timer":
"""
Update an existing timer on the provider's backend.
Args:
timer: Timer with updated fields. timer.client_index identifies
the record to modify.
**kwargs: Provider-specific arguments.
Returns:
The updated Timer as confirmed by the provider.
Raises:
KeyError: If no timer with that client_index exists.
RuntimeError: If the provider refuses the update (e.g. the timer
is currently recording).
"""
raise NotImplementedError(
f"{self.__class__.__name__} does not implement update_timer(). "
"Override this method to support timer updates."
)
def delete_timer(
self, client_index: int, force_delete: bool = False, **kwargs
) -> None:
"""
Delete a timer on the provider's backend.
Args:
client_index: Timer identifier to delete.
force_delete: If True and the timer is currently recording, abort
the ongoing capture before deleting.
**kwargs: Provider-specific arguments.
Returns:
None on success.
Raises:
KeyError: If no timer with that client_index exists.
RuntimeError: If the provider refuses deletion (e.g. recording in
progress and force_delete is False).
"""
raise NotImplementedError(
f"{self.__class__.__name__} does not implement delete_timer(). "
"Override this method to support timer deletion."
)
# ============================================================================
# CATCHUP ABSTRACT METHODS
# ============================================================================
lib/streaming_providers/base/timer_operations.py
+297
View File
@@ -0,0 +1,297 @@
# streaming_providers/base/timer_operations.py
"""
Timer-related operations separated from core registry.
Mirrors the structure of RecordingOperations.
"""
from typing import Dict, List, Optional
from .models.timer import Timer
from .models.timer_type import TimerType
from .utils.logger import logger
class TimerOperations:
"""Handles all timer-related operations."""
def __init__(self, registry):
self.registry = registry
logger.debug("TimerOperations: Initialized")
# ------------------------------------------------------------------
# Timer types
# ------------------------------------------------------------------
def get_timer_types(self, provider_name: str) -> List[TimerType]:
"""
Get the timer types supported by a specific provider.
Args:
provider_name: Name of the provider to query.
Returns:
List of TimerType objects describing what kinds of timers
the provider can accept.
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")
timer_types = provider.get_timer_types()
logger.debug(
f"Retrieved {len(timer_types)} timer types from '{provider_name}'"
)
return timer_types
def get_all_timer_types(self) -> Dict[str, List[TimerType]]:
"""
Get timer types from all enabled providers.
Returns:
Dict mapping provider name → list of TimerType objects.
"""
enabled = self.registry.get_enabled_providers()
result = {}
for name in enabled:
try:
if self._provider_implements_timers(name):
result[name] = self.get_timer_types(name)
else:
result[name] = []
except Exception as e:
logger.error(f"Failed to get timer types from '{name}': {e}")
result[name] = []
return result
# ------------------------------------------------------------------
# Listing timers
# ------------------------------------------------------------------
def get_timers(
self,
provider_name: str,
include_inactive: bool = False,
) -> List[Timer]:
"""
Get timers from a specific provider.
Args:
provider_name: Name of the provider to query.
include_inactive: If True, include completed, cancelled, and
error-state timers in addition to active ones.
Returns:
List of Timer 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")
timers = provider.get_timers()
if not include_inactive:
timers = [t for t in timers if t.is_active]
logger.info(
f"Retrieved {len(timers)} timers from '{provider_name}' "
f"(include_inactive={include_inactive})"
)
return timers
def get_all_timers(
self,
include_inactive: bool = False,
) -> Dict[str, List[Timer]]:
"""
Get timers from all enabled providers.
Args:
include_inactive: If True, include non-active timers.
Returns:
Dict mapping provider name → list of Timer objects.
"""
enabled = self.registry.get_enabled_providers()
logger.info(f"Fetching timers from {len(enabled)} providers")
result = {}
total = 0
for name in enabled:
try:
if self._provider_implements_timers(name):
timers = self.get_timers(name, include_inactive=include_inactive)
else:
timers = []
result[name] = timers
total += len(timers)
except Exception as e:
logger.error(f"Failed to get timers from '{name}': {e}")
result[name] = []
logger.info(f"Retrieved {total} total timers")
return result
def get_timer(
self, provider_name: str, client_index: int
) -> Optional[Timer]:
"""
Get a single timer by its client index.
Args:
provider_name: Name of the provider.
client_index: Timer identifier (== Timer.client_index).
Returns:
Timer object, or None if not found.
Raises:
ValueError: If the provider is not found or disabled.
"""
# Fetch all (including inactive) so we can look up any timer by index
timers = self.get_timers(provider_name, include_inactive=True)
match = next((t for t in timers if t.client_index == client_index), None)
if match:
logger.debug(
f"Found timer {client_index} from '{provider_name}'"
)
else:
logger.debug(
f"Timer {client_index} not found on '{provider_name}'"
)
return match
# ------------------------------------------------------------------
# Creating timers
# ------------------------------------------------------------------
def add_timer(
self, provider_name: str, timer: Timer, **kwargs
) -> Timer:
"""
Schedule a new timer on the provider.
The provider assigns client_index and returns the saved Timer (which
may differ from the input if the provider normalises fields).
Args:
provider_name: Name of the provider.
timer: Timer object to create. client_index is ignored —
the provider allocates one.
**kwargs: Additional provider-specific arguments.
Returns:
The saved Timer as confirmed by the provider.
Raises:
ValueError: If the provider is not found or disabled.
RuntimeError: If the provider rejects the timer (e.g. conflict,
insufficient permissions, unsupported timer type).
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
saved_timer = provider.add_timer(timer, **kwargs)
logger.info(
f"Added timer '{saved_timer.title}' (index={saved_timer.client_index}) "
f"on '{provider_name}'"
)
return saved_timer
# ------------------------------------------------------------------
# Updating timers
# ------------------------------------------------------------------
def update_timer(
self, provider_name: str, timer: Timer, **kwargs
) -> Timer:
"""
Update an existing timer on the provider.
Args:
provider_name: Name of the provider.
timer: Timer object with updated fields.
timer.client_index identifies the record to update.
**kwargs: Additional provider-specific arguments.
Returns:
The updated Timer as confirmed by the provider.
Raises:
ValueError: If the provider is not found or disabled.
KeyError: If no timer with that client_index exists.
RuntimeError: If the provider refuses the update (e.g. the timer
is currently recording and cannot be modified).
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
updated_timer = provider.update_timer(timer, **kwargs)
logger.info(
f"Updated timer {timer.client_index} ('{timer.title}') "
f"on '{provider_name}'"
)
return updated_timer
# ------------------------------------------------------------------
# Deleting timers
# ------------------------------------------------------------------
def delete_timer(
self,
provider_name: str,
client_index: int,
force_delete: bool = False,
**kwargs,
) -> None:
"""
Delete a timer on the provider.
Args:
provider_name: Name of the provider.
client_index: Timer identifier to delete.
force_delete: If True and the timer is currently recording, abort
the ongoing capture and delete. If False and the
timer is recording, the provider should raise
RuntimeError.
**kwargs: Additional provider-specific arguments.
Returns:
None on success.
Raises:
ValueError: If the provider is not found or disabled.
KeyError: If no timer with that client_index exists.
RuntimeError: If the provider refuses deletion (e.g. timer is
recording and force_delete is False).
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
provider.delete_timer(client_index, force_delete=force_delete, **kwargs)
logger.info(
f"Deleted timer {client_index} from '{provider_name}' "
f"(force={force_delete})"
)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _provider_implements_timers(self, provider_name: str) -> bool:
"""Return True if the named provider declares timer support."""
provider = self.registry.get_provider(provider_name)
if not provider:
return False
return getattr(provider, "implements_timers", False)
lib/streaming_providers/providers/movetv/vod_manager.py
+118 -5
View File
@@ -696,13 +696,126 @@ class MoveTvVodManager:
self, path_ids: List[str], **kwargs
) -> List[Union[VodCategory, VodItem]]:
"""
Fetch children of a VOD category node (seasons, episodes, ...).
Fetch children of a VOD category node.
Not yet implemented -- log the browser request when drilling into a
series or category, then implement here.
Path structure:
[] or ["root"] → Return content types (Film, Serija)
["root", "Film"] → Return categories for movies
["root", "Serija"] → Return categories for series
["root", "Film", "123"] → Return VOD items for that category
["root", "Serija", "456"] → Return VOD items for that series category
"""
# Handle root level - return content types
if not path_ids or path_ids[0] == "root":
# Fetch filters to get content types
filters = self.get_vod_filters()
categories = []
for content_type in filters.content_types:
# Map content type ID to a display name
type_name = content_type.name # "Film" or "Serija"
categories.append(
VodCategory(
name=type_name,
content_id=f"content_type_{content_type.content_type_id}",
provider=self._provider.provider_name,
# Store the content_type_id in the path for next level
# The UI will pass this as path_ids[1]
)
)
logger.info(
f"{self._provider.provider_name}: VOD root -> "
f"{len(categories)} content types"
)
return categories
# Handle content type level (e.g., ["root", "Film"] or ["root", "Serija"])
if len(path_ids) == 2 and path_ids[0] == "root":
content_type_name = path_ids[1]
# Fetch filters to get categories for this content type
filters = self.get_vod_filters()
# Find the content type ID
content_type_id = None
for ct in filters.content_types:
if ct.name == content_type_name:
content_type_id = ct.content_type_id
break
if not content_type_id:
logger.warning(
f"{self._provider.provider_name}: Unknown content type '{content_type_name}'"
)
return []
# Return categories/genres for this content type
categories = []
for category in filters.categories:
# Create a category that can be browsed further
categories.append(
VodCategory(
name=category.name,
content_id=str(category.content_id),
provider=self._provider.provider_name,
# The path will be ["root", content_type_name, category_id]
)
)
logger.info(
f"{self._provider.provider_name}: VOD content type '{content_type_name}' -> "
f"{len(categories)} categories"
)
return categories
# Handle category level - return VOD items
if len(path_ids) >= 3 and path_ids[0] == "root":
content_type_name = path_ids[1]
category_id = path_ids[2]
# Fetch filters to get content type ID
filters = self.get_vod_filters()
content_type_id = None
for ct in filters.content_types:
if ct.name == content_type_name:
content_type_id = ct.content_type_id
break
if not content_type_id:
return []
# Map content type name to tag ID (optional filtering)
tag_map = {
"Film": 1, # FILM tag
"Serija": 2, # SERIJA tag
}
tag_id = tag_map.get(content_type_name)
# Fetch VOD items for this category
vod_page = self.get_vod_items(
page=1,
sort="newest",
tag_id=tag_id,
category_id=int(category_id) if category_id != "root" else None,
content_type_id=content_type_id,
)
# Convert VodItems to the expected return format
# (VodOperations likely expects VodItem or similar)
items = []
for item in vod_page.items:
items.append(item)
logger.info(
f"{self._provider.provider_name}: VOD category '{category_id}' -> "
f"{len(items)} items"
)
return items
logger.warning(
f"{self._provider.provider_name}: get_vod_category({path_ids}) -- "
"not yet implemented; log the browse/drill-down endpoint to add support."
f"{self._provider.provider_name}: Unhandled VOD path: {path_ids}"
)
return []
routes/timers.py
+565
View File
@@ -0,0 +1,565 @@
#!/usr/bin/env python3
"""
Timer-related route handlers.
Mirrors the structure of recordings.py.
Endpoints
---------
GET /api/providers/<provider>/timer-types
GET /api/providers/<provider>/timers
GET /api/providers/<provider>/timers/<client_index>
POST /api/providers/<provider>/timers
PUT /api/providers/<provider>/timers/<client_index>
DELETE /api/providers/<provider>/timers/<client_index>
"""
from bottle import request, response
from streaming_providers.base.models.timer import (
DuplicateHandling,
Timer,
TimerState,
TimerWeekday,
)
from streaming_providers.base.utils import logger
from datetime import datetime
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _parse_datetime(value: str, field_name: str) -> datetime:
"""
Parse an ISO 8601 datetime string, raising ValueError with a clear
message if parsing fails.
"""
try:
return datetime.fromisoformat(value)
except (ValueError, TypeError):
raise ValueError(
f"'{field_name}' must be an ISO 8601 datetime string "
f"(e.g. '2025-06-01T20:15:00'), got: {value!r}"
)
def _timer_from_body(body: dict, client_index: int = 0) -> Timer:
"""
Deserialise a request body dictionary into a Timer object.
Required keys: title, timer_type_id, provider
All other keys are optional and map directly to Timer fields.
Raises:
ValueError: If required keys are missing or a field cannot be parsed.
"""
for required in ("title", "timer_type_id"):
if required not in body:
raise ValueError(f"Missing required field: '{required}'")
kwargs = {}
# Timing
if "start_time" in body and body["start_time"] is not None:
kwargs["start_time"] = _parse_datetime(body["start_time"], "start_time")
if "end_time" in body and body["end_time"] is not None:
kwargs["end_time"] = _parse_datetime(body["end_time"], "end_time")
if "first_day" in body and body["first_day"] is not None:
kwargs["first_day"] = _parse_datetime(body["first_day"], "first_day")
# Enumerations
if "state" in body:
try:
kwargs["state"] = TimerState(body["state"])
except ValueError:
raise ValueError(
f"Invalid state '{body['state']}'. "
f"Valid values: {[s.value for s in TimerState]}"
)
if "weekdays" in body:
try:
kwargs["weekdays"] = TimerWeekday(int(body["weekdays"]))
except (ValueError, TypeError):
raise ValueError(
f"'weekdays' must be an integer bitmask "
f"(e.g. {TimerWeekday.MONDAY.value} for Monday). "
f"Got: {body['weekdays']!r}"
)
if "prevent_duplicate_episodes" in body:
try:
kwargs["prevent_duplicate_episodes"] = DuplicateHandling(
int(body["prevent_duplicate_episodes"])
)
except (ValueError, TypeError):
raise ValueError(
f"Invalid prevent_duplicate_episodes value: "
f"{body['prevent_duplicate_episodes']!r}. "
f"Valid values: {[d.value for d in DuplicateHandling]}"
)
# Scalar fields — passed through directly if present
scalar_fields = (
"timer_type_id",
"title",
"parent_client_index",
"client_channel_uid",
"channel_name",
"start_any_time",
"end_any_time",
"margin_start",
"margin_end",
"epg_search_string",
"full_text_epg_search",
"epg_uid",
"epg_event_id",
"series_link",
"directory",
"priority",
"lifetime",
"max_recordings",
"recording_group",
"genre_type",
"genre_sub_type",
"description",
)
for f in scalar_fields:
if f in body:
kwargs[f] = body[f]
return Timer(client_index=client_index, **kwargs)
# ---------------------------------------------------------------------------
# Route setup
# ---------------------------------------------------------------------------
def setup_timers_routes(app, manager, service):
"""Register timer-related routes on the Bottle app."""
# ------------------------------------------------------------------ #
# GET /api/providers/<provider>/timer-types #
# ------------------------------------------------------------------ #
@app.route("/api/providers/<provider>/timer-types", method="GET")
def get_provider_timer_types(provider):
"""
Get the timer types supported by a specific provider.
Returns:
{
"provider": "provider_name",
"timer_types": [ { ...TimerType fields... } ],
"count": 2
}
"""
try:
try:
timer_types = manager.timer_ops.get_timer_types(provider_name=provider)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
serialized = [tt.to_dict() for tt in timer_types]
response.status = 200
return {
"provider": provider,
"timer_types": serialized,
"count": len(serialized),
}
except Exception as e:
logger.error(f"Unexpected error in get_provider_timer_types: {e}")
response.status = 500
return {"error": "Internal server error", "message": str(e), "provider": provider}
# ------------------------------------------------------------------ #
# GET /api/providers/<provider>/timers #
# ------------------------------------------------------------------ #
@app.route("/api/providers/<provider>/timers", method="GET")
def get_provider_timers(provider):
"""
List timers for a specific provider.
Query parameters:
include_inactive: bool (default false) — include completed /
cancelled / error timers.
Returns:
{
"provider": "provider_name",
"timers": [ { ...Timer fields... } ],
"count": 3
}
"""
try:
include_inactive = (
request.params.get("include_inactive", "false").lower()
in ("1", "true", "yes")
)
try:
timers = manager.timer_ops.get_timers(
provider_name=provider,
include_inactive=include_inactive,
)
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 timers from '{provider}': {e}")
response.status = 500
return {
"error": "Failed to get timers",
"message": str(e),
"provider": provider,
}
serialized = [t.to_dict() for t in timers]
response.status = 200
return {
"provider": provider,
"timers": serialized,
"count": len(serialized),
}
except Exception as e:
logger.error(f"Unexpected error in get_provider_timers: {e}")
response.status = 500
return {"error": "Internal server error", "message": str(e), "provider": provider}
# ------------------------------------------------------------------ #
# GET /api/providers/<provider>/timers/<client_index> #
# ------------------------------------------------------------------ #
@app.route("/api/providers/<provider>/timers/<client_index:int>", method="GET")
def get_provider_timer(provider, client_index):
"""
Get a single timer by its client index.
Returns:
{ ...Timer fields... }
"""
try:
try:
timer = manager.timer_ops.get_timer(
provider_name=provider,
client_index=client_index,
)
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 timer {client_index} from '{provider}': {e}"
)
response.status = 500
return {
"error": "Failed to get timer",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
if timer is None:
response.status = 404
return {
"error": "Timer not found",
"message": (
f"No timer with client_index {client_index} "
f"from '{provider}'"
),
"provider": provider,
"client_index": client_index,
}
response.status = 200
return timer.to_dict()
except Exception as e:
logger.error(f"Unexpected error in get_provider_timer: {e}")
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
}
# ------------------------------------------------------------------ #
# POST /api/providers/<provider>/timers #
# ------------------------------------------------------------------ #
@app.route("/api/providers/<provider>/timers", method="POST")
def add_provider_timer(provider):
"""
Schedule a new timer on a specific provider.
Request body (JSON):
{
"title": "Tatort", // required
"timer_type_id": 1, // required
"client_channel_uid": 42,
"start_time": "2025-06-01T20:15:00",
"end_time": "2025-06-01T21:45:00",
"margin_start": 5,
"margin_end": 10,
"priority": 50,
"lifetime": 30,
...
}
Returns:
201 Created with the saved Timer in the body.
400 if the request body is invalid.
404 if the provider is not found.
409 if the provider refuses the timer (e.g. conflict).
500 on unexpected errors.
"""
try:
body = request.json
if not body:
response.status = 400
return {
"error": "Bad request",
"message": "Request body must be valid JSON",
"provider": provider,
}
# Inject provider so the Timer knows where it belongs
body.setdefault("provider", provider)
try:
timer = _timer_from_body(body)
except ValueError as e:
response.status = 400
return {
"error": "Invalid timer data",
"message": str(e),
"provider": provider,
}
try:
saved = manager.timer_ops.add_timer(
provider_name=provider, timer=timer
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except RuntimeError as e:
response.status = 409
return {
"error": "Timer rejected by provider",
"message": str(e),
"provider": provider,
}
except Exception as e:
logger.error(f"Failed to add timer on '{provider}': {e}")
response.status = 500
return {
"error": "Failed to add timer",
"message": str(e),
"provider": provider,
}
response.status = 201
return saved.to_dict()
except Exception as e:
logger.error(f"Unexpected error in add_provider_timer: {e}")
response.status = 500
return {"error": "Internal server error", "message": str(e), "provider": provider}
# ------------------------------------------------------------------ #
# PUT /api/providers/<provider>/timers/<client_index> #
# ------------------------------------------------------------------ #
@app.route("/api/providers/<provider>/timers/<client_index:int>", method="PUT")
def update_provider_timer(provider, client_index):
"""
Update an existing timer.
The client_index from the URL takes precedence over any value in the
body — this prevents accidental cross-timer updates.
Request body (JSON): same shape as POST, all fields optional except
those the provider requires to be present on an update.
Returns:
200 OK with the updated Timer in the body.
400 if the request body is invalid.
404 if the provider or timer is not found.
409 if the provider refuses the update.
500 on unexpected errors.
"""
try:
body = request.json
if not body:
response.status = 400
return {
"error": "Bad request",
"message": "Request body must be valid JSON",
"provider": provider,
"client_index": client_index,
}
body.setdefault("provider", provider)
try:
timer = _timer_from_body(body, client_index=client_index)
except ValueError as e:
response.status = 400
return {
"error": "Invalid timer data",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
try:
updated = manager.timer_ops.update_timer(
provider_name=provider, timer=timer
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except KeyError as e:
response.status = 404
return {
"error": "Timer not found",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
except RuntimeError as e:
response.status = 409
return {
"error": "Update refused by provider",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
except Exception as e:
logger.error(
f"Failed to update timer {client_index} on '{provider}': {e}"
)
response.status = 500
return {
"error": "Failed to update timer",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
response.status = 200
return updated.to_dict()
except Exception as e:
logger.error(f"Unexpected error in update_provider_timer: {e}")
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
}
# ------------------------------------------------------------------ #
# DELETE /api/providers/<provider>/timers/<client_index> #
# ------------------------------------------------------------------ #
@app.route("/api/providers/<provider>/timers/<client_index:int>", method="DELETE")
def delete_provider_timer(provider, client_index):
"""
Delete a timer.
Query parameters:
force: bool (default false) — if the timer is currently recording,
abort the ongoing capture and delete.
Returns:
204 No Content on success.
404 if the provider or timer is not found.
409 if the provider refuses deletion and force was not set.
500 on unexpected errors.
"""
try:
force = (
request.params.get("force", "false").lower()
in ("1", "true", "yes")
)
try:
manager.timer_ops.delete_timer(
provider_name=provider,
client_index=client_index,
force_delete=force,
)
except ValueError as e:
response.status = 404
return {
"error": "Provider not found",
"message": str(e),
"provider": provider,
}
except KeyError as e:
response.status = 404
return {
"error": "Timer not found",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
except RuntimeError as e:
response.status = 409
return {
"error": "Deletion refused by provider",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
except Exception as e:
logger.error(
f"Failed to delete timer {client_index} from '{provider}': {e}"
)
response.status = 500
return {
"error": "Failed to delete timer",
"message": str(e),
"provider": provider,
"client_index": client_index,
}
# 204 No Content — no body
response.status = 204
return ""
except Exception as e:
logger.error(f"Unexpected error in delete_provider_timer: {e}")
response.status = 500
return {
"error": "Internal server error",
"message": str(e),
"provider": provider,
}
service.py
+2
View File
@@ -1587,6 +1587,7 @@ class UltimateService:
from routes.events import setup_events_routes
from routes.vod import setup_vod_routes
from routes.recordings import setup_recordings_routes
from routes.timers import setup_timers_routes
# Setup routes from separate modules
setup_provider_routes(self.app, self.manager, self)
@@ -1599,6 +1600,7 @@ class UltimateService:
setup_events_routes(self.app, self.manager, self)
setup_vod_routes(self.app, self.manager)
setup_recordings_routes(self.app, self.manager, self)
setup_timers_routes(self.app, self.manager, self)
# Core UI routes
@self.app.route("/config")