Source code for stash_vroom.heresphere
# Copyright 2025 Zyquo Onrel
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
This module provides a convenient way to serve Stash content to HereSphere
and easily handle user events and interactions without writing an entire web service.
Example
-------
Here's a basic example of using the HereSphere API:
.. code-block:: python
from stash_vroom.heresphere import on_event
@on_event('play')
def on_play(scene):
print(f"Playing scene: {scene.title}")
@on_event('favorite')
def on_favorite(scene):
print(f"Marking scene as favorite: {scene.title}")
@on_event('un-favorite')
def on_unfavorite(scene):
print(f"Marking scene as not favorite: {scene.title}")
"""
import logging
from typing import Callable, Dict, List, Optional, Any
log = logging.getLogger(__name__)
[docs]
class Scene:
"""
Represents a video scene in HereSphere.
This is a simplified representation that will be passed to event handlers.
"""
def __init__(self, id: str, title: str, path: str, metadata: Optional[Dict[str, Any]] = None):
"""
Initialize a Scene object.
:param id: Unique identifier for the scene
:param title: Title of the scene
:param path: File path to the scene
:param metadata: Optional dictionary of additional metadata
"""
self.id = id
self.title = title
self.path = path
self.metadata = metadata or {}
def __str__(self) -> str:
return f"Scene({self.id}, {self.title})"
def __repr__(self) -> str:
return self.__str__()
[docs]
class HereSphereApp:
"""
Main class for the HereSphere application.
Similar to Flask's Flask class, this manages event handlers and provides
decorators for registering them.
"""
def __init__(self, name: str):
"""
Initialize a HereSphereApp.
:param name: Name of the application
"""
self.name = name
self.event_handlers: Dict[str, List[Callable[[Scene], Any]]] = {}
log.info(f"Initialized HereSphereApp: {name}")
[docs]
def on_event(self, event_name: str):
"""
Decorator for registering event handlers.
:param event_name: Name of the event to handle (e.g., "delete", "favorite", "play")
:return: Decorator function
"""
def decorator(func: Callable[[Scene], Any]):
if event_name not in self.event_handlers:
self.event_handlers[event_name] = []
self.event_handlers[event_name].append(func)
log.debug(f"Registered handler for event {event_name}: {func.__name__}")
return func
return decorator
[docs]
def trigger_event(self, event_name: str, scene: Scene) -> List[Any]:
"""
Trigger an event and call all registered handlers.
:param event_name: Name of the event to trigger
:param scene: Scene object to pass to handlers
:return: List of results from all handlers
"""
results = []
if event_name in self.event_handlers:
for handler in self.event_handlers[event_name]:
log.debug(f"Calling handler {handler.__name__} for event {event_name}")
try:
result = handler(scene)
results.append(result)
except Exception as e:
log.error(f"Error in handler {handler.__name__}: {e}")
else:
log.warning(f"No handlers registered for event {event_name}")
return results
# Create a global app instance for convenience
app = HereSphereApp("default")
# Convenience functions that use the global app
[docs]
def on_event(event_name: str):
"""
Decorator for registering event handlers on the global app.
:param event_name: Name of the event to handle
:return: Decorator function
"""
return app.on_event(event_name)
[docs]
def trigger_event(event_name: str, scene: Scene) -> List[Any]:
"""
Trigger an event on the global app.
:param event_name: Name of the event to trigger
:param scene: Scene object to pass to handlers
:return: List of results from all handlers
"""
return app.trigger_event(event_name, scene)