navidrome/plugins/examples/nowplaying-py/plugin/__init__.py

# Now Playing Logger Plugin for Navidrome
#
# This plugin demonstrates the Scheduler and SubsonicAPI host services by
# periodically logging what is currently playing in Navidrome.
#
# Build with:
#   extism-py plugin/__init__.py -o nowplaying-py.wasm
#
# Test manifest with:
#   extism call nowplaying-py.wasm nd_manifest --wasi
#
# Configuration:
#   [PluginConfig.nowplaying-py]
#   cron = "*/1 * * * *"  # Every minute (default)
#   user = "admin"     # User to query getNowPlaying (default)

import extism
import json

# Schedule ID for our recurring task
SCHEDULE_ID = "nowplaying-check"


# =============================================================================
# Host Function Imports
# =============================================================================
# These are custom host functions provided by Navidrome.
# We import them using the extism:host/user namespace.


@extism.import_fn("extism:host/user", "scheduler_schedulerecurring")
def _scheduler_schedulerecurring(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "subsonicapi_call")
def _subsonicapi_call(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


# =============================================================================
# Host Function Wrappers
# =============================================================================
# These wrappers handle JSON marshalling/unmarshalling and memory management.


def scheduler_schedule_recurring(cron_expression: str, payload: str, schedule_id: str) -> str:
    """Schedule a recurring task using a cron expression.
    
    Args:
        cron_expression: Cron format (e.g., "*/1 * * * *" for every minute)
        payload: Data to pass to the callback
        schedule_id: Unique identifier for the schedule
        
    Returns:
        The schedule ID (same as input or auto-generated)
    """
    request = {
        "cronExpression": cron_expression,
        "payload": payload,
        "scheduleID": schedule_id
    }
    request_bytes = json.dumps(request).encode('utf-8')
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _scheduler_schedulerecurring(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))
    
    if response.get("error"):
        raise Exception(response["error"])
    
    return response.get("newScheduleID", schedule_id)


def subsonicapi_call(uri: str) -> dict:
    """Call a Subsonic API endpoint.
    
    Args:
        uri: API path (e.g., "getNowPlaying")
        
    Returns:
        Parsed JSON response from the API
    """
    request = {"uri": uri}
    request_bytes = json.dumps(request).encode('utf-8')
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _subsonicapi_call(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))
    
    if response.get("error"):
        raise Exception(response["error"])
    
    # Parse the nested JSON response
    response_json = response.get("responseJSON", "{}")
    return json.loads(response_json)


# =============================================================================
# Plugin Exports
# =============================================================================


@extism.plugin_fn
def nd_manifest():
    """Return the plugin manifest with metadata and permissions."""
    manifest = {
        "name": "Now Playing Logger (Python)",
        "author": "Navidrome",
        "version": "1.0.0",
        "description": "Periodically logs currently playing tracks - Python example demonstrating Scheduler and SubsonicAPI host services",
        "website": "https://github.com/navidrome/navidrome/tree/master/plugins/examples/nowplaying-py",
        "permissions": {
            "scheduler": {
                "reason": "Schedule periodic checks for now playing status"
            },
            "subsonicapi": {
                "reason": "Query the getNowPlaying API endpoint",
                "allowAdmins": True
            }
        }
    }
    extism.output_str(json.dumps(manifest))


@extism.plugin_fn
def nd_on_init():
    """Initialize the plugin by scheduling the recurring task."""
    # Read cron expression from config, default to every minute
    cron = extism.Config.get_str("cron")
    if not cron:
        cron = "*/1 * * * *"
    
    extism.log(extism.LogLevel.Info, f"Now Playing Logger initializing with cron: {cron}")
    
    try:
        schedule_id = scheduler_schedule_recurring(cron, "check", SCHEDULE_ID)
        extism.log(extism.LogLevel.Info, f"Scheduled recurring task with ID: {schedule_id}")
    except Exception as e:
        extism.log(extism.LogLevel.Error, f"Failed to schedule task: {e}")
        raise
    
    # Return empty success response
    extism.output_str(json.dumps({}))


@extism.plugin_fn
def nd_scheduler_callback():
    """Handle scheduler callback - check and log now playing tracks."""
    input_data = extism.input_json()
    schedule_id = input_data.get("schedule_id", "")
    
    # Only handle our schedule
    if schedule_id != SCHEDULE_ID:
        extism.output_str(json.dumps({}))
        return
    
    try:
        # Read user from config, default to admin
        user = extism.Config.get_str("user")
        if not user:
            user = "admin"
        
        # Call the getNowPlaying API
        response = subsonicapi_call(f"getNowPlaying?u={user}")
        
        # Extract the subsonic-response
        subsonic_response = response.get("subsonic-response", {})
        now_playing = subsonic_response.get("nowPlaying", {})
        entries = now_playing.get("entry", [])
        
        if not entries:
            extism.log(extism.LogLevel.Info, "🎵 No users currently playing music")
        else:
            # Handle both single entry and list of entries
            if isinstance(entries, dict):
                entries = [entries]
            
            for entry in entries:
                artist = entry.get("artist", "Unknown Artist")
                title = entry.get("title", "Unknown Title")
                album = entry.get("album", "Unknown Album")
                username = entry.get("username", "Unknown User")
                
                extism.log(
                    extism.LogLevel.Info,
                    f"🎵 {username} is playing: {artist} - {title} ({album})"
                )
        
        extism.output_str(json.dumps({}))
        
    except Exception as e:
        error_msg = str(e)
        extism.log(extism.LogLevel.Error, f"Failed to get now playing: {error_msg}")
        extism.output_str(json.dumps({"error": error_msg}))