mirror of
https://github.com/nirvana-7777/script.service.ultimate.git
synced 2026-09-16 14:12:20 +02:00
297 lines
10 KiB
Python
297 lines
10 KiB
Python
# 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) |