mopidy-webhooks/mopidy_webhooks/frontend.py

import logging
import os
import time
import json
from typing import Any, Dict, Optional

import pykka
import requests
from mopidy.core import CoreListener

from . import metadata as metadata_lib

logger = logging.getLogger(__name__)


class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
    def __init__(self, config, core):
        super().__init__()
        self.config = config
        self.webhook_urls = []
        self.webhook_tokens = []
        self.media_dirs = []
        self.podcast_dirs = []
        self._metadata_cache: Dict[str, Dict[str, Any]] = {}
        self.last_start_time = None

    def on_start(self):
        self.webhook_urls = self.config["webhooks"]["urls"].split(",")
        self.webhook_tokens = self.config["webhooks"]["tokens"].split(",")
        self.media_dirs = metadata_lib.parse_dirs(
            self.config["webhooks"].get("media_dirs", "")
        )
        self.podcast_dirs = metadata_lib.parse_dirs(
            self.config["webhooks"].get("podcast_dirs", "")
        )
        logger.info(f"Parsing webhook URLs and tokens: {self.webhook_urls}")

    def _get_track_metadata(self, track) -> tuple:
        """Return ``(metadata, path)`` for a local track, or empty values."""
        path = metadata_lib.resolve_media_path(track.uri, self.media_dirs)
        if not path:
            return {}, None

        try:
            cache_key = f"{path}:{int(os.path.getmtime(path))}"
        except OSError:
            cache_key = path

        if len(self._metadata_cache) > 512:
            self._metadata_cache.clear()

        if cache_key not in self._metadata_cache:
            self._metadata_cache[cache_key] = metadata_lib.read_metadata(path)
        return self._metadata_cache[cache_key], path

    def _build_podcast_data(
        self, track, metadata: Dict[str, Any]
    ) -> Dict[str, Any]:
        artists = ", ".join(sorted([a.name for a in track.artists]))
        feed_url, guid = metadata_lib.split_podcast_uri(track.uri)
        feed_url = metadata.get("feed_url") or feed_url
        guid = metadata.get("guid") or guid
        album_name = track.album.name if track.album else ""

        episode_num = metadata.get("track_number") or track.track_no or 0
        try:
            episode_num = int(str(episode_num).split("/")[0].strip())
        except (TypeError, ValueError):
            episode_num = 0

        description = (
            metadata.get("description") or metadata.get("comment") or ""
        )

        try:
            duration_seconds = int(float(metadata.get("duration") or 0))
        except (TypeError, ValueError):
            duration_seconds = 0
        if not duration_seconds:
            duration_seconds = track.length and track.length // 1000 or 0

        return {
            "podcast_name": metadata.get("album") or album_name,
            "podcast_producer": metadata.get("artist") or artists,
            "podcast_description": description,
            "podcast_feed_url": feed_url or "",
            "podcast_site_link": metadata.get("website") or "",
            "episode_num": episode_num,
            "pub_date": metadata_lib.normalize_date(metadata.get("date")),
            "episode_description": metadata.get("comment") or description,
            "episode_url": metadata.get("episode_url") or "",
            "episode_guid": guid or "",
            "duration_seconds": duration_seconds,
        }

    def _build_post_data(
        self, track, time_position: Optional[int] = None
    ) -> dict:
        artists = ", ".join(sorted([a.name for a in track.artists]))
        artists_list = [a for a in track.artists]
        try:
            musicbrainz_artist_id = artists_list[0].musicbrainz_id
        except IndexError:
            musicbrainz_artist_id = None
        duration = track.length and track.length // 1000 or 0
        album_name = track.album.name if track.album else ""

        metadata, path = self._get_track_metadata(track)
        podcast = metadata_lib.is_podcast(
            track.uri,
            path=path,
            metadata=metadata,
            podcast_dirs=self.podcast_dirs,
        )

        name = track.name
        track_number = track.track_no
        if podcast:
            name = metadata.get("title") or track.name
            artists = metadata.get("artist") or artists
            album_name = metadata.get("album") or album_name
            track_number = metadata.get("track_number") or track.track_no

        post_data = {
            "name": name,
            "artist": artists,
            "album": album_name,
            "track_number": track_number,
            "run_time_ticks": track.length,
            "run_time": str(duration),
            "playback_time_ticks": time_position,
            "musicbrainz_track_id": track.musicbrainz_id if track.album else "",
            "musicbrainz_album_id": (
                track.album.musicbrainz_id if track.album else ""
            ),
            "musicbrainz_artist_id": musicbrainz_artist_id,
            "mopidy_uri": track.uri,
            "primary_artist_mopidy_uri": artists_list[0].uri,
            "media_type": "podcast" if podcast else "track",
        }

        if podcast:
            post_data.update(self._build_podcast_data(track, metadata))

        return post_data

    def _post_update_to_webhooks(self, post_data: dict, status: str):
        post_data["status"] = status

        for index, webhook_url in enumerate(self.webhook_urls):
            token = ""
            headers = {}
            try:
                token = self.webhook_tokens[index]
            except IndexError:
                logger.info(f"No token found for Webhook URL: {webhook_url}")

            if token:
                headers["Authorization"] = f"Token {token}"

            response = requests.post(
                webhook_url, json=json.dumps(post_data), headers=headers
            )
            logger.info(response)

    def track_playback_started(self, tl_track):
        track = tl_track.track
        artists = ", ".join(sorted([a.name for a in track.artists]))
        self.last_start_time = int(time.time())
        logger.debug(f"Now playing track: {artists} - {track.name}")
        post_data = self._build_post_data(tl_track.track, time_position=0)

        # Build post data to send to urls
        if not self.webhook_urls:
            logger.info("No webhook URLS are configured ")
            return
        logger.info(f"Scrobbling via webhooks: {artists} - {track.name}")
        self._post_update_to_webhooks(post_data, "started")

    def track_playback_ended(self, tl_track, time_position):
        track = tl_track.track
        artists = ", ".join(sorted([a.name for a in track.artists]))
        duration = track.length and track.length // 1000 or 0
        time_position_sec = time_position // 1000

        post_data = self._build_post_data(
            tl_track.track, time_position=time_position
        )

        if time_position_sec < duration // 2 and time_position_sec < 240:
            logger.debug(
                "Track not played long enough to scrobble. (50% or 240s)"
            )
            return

        if self.last_start_time is None:
            self.last_start_time = int(time.time()) - duration
        logger.info(
            f"Scrobbling finished via webhooks: {artists} - {track.name}"
        )

        self._post_update_to_webhooks(post_data, "stopped")

    def track_playback_paused(self, tl_track, time_position):
        track = tl_track.track
        artists = ", ".join(sorted([a.name for a in track.artists]))
        duration = track.length and track.length // 1000 or 0

        post_data = self._build_post_data(
            tl_track.track, time_position=time_position
        )

        if self.last_start_time is None:
            self.last_start_time = int(time.time()) - duration
        logger.info(f"Scrobbling paused via webhooks: {artists} - {track.name}")

        self._post_update_to_webhooks(post_data, "paused")

    def track_playback_resumed(self, tl_track, time_position):

        track = tl_track.track
        artists = ", ".join(sorted([a.name for a in track.artists]))
        self.last_start_time = int(time.time())
        logger.debug(f"Now resuming track: {artists} - {track.name}")
        post_data = self._build_post_data(
            tl_track.track, time_position=time_position
        )

        # Build post data to send to urls
        if not self.webhook_urls:
            logger.info("No webhook URLS are configured ")
            return
        logger.info(f"Scrobbling via webhooks: {artists} - {track.name}")
        self._post_update_to_webhooks(post_data, "resumed")