mopidy-webhooks/mopidy_webhooks/frontend.py

import logging
import time
import json

import pykka
import requests
from mopidy.core import CoreListener

logger = logging.getLogger(__name__)


class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
    def __init__(self, config, core):
        super().__init__()
        self.config = config
        self.webhook_urls = []
        self.last_start_time = None

    def on_start(self):
        logger.info("Parsing webhook URLs and tokens")
        self.webhook_urls = self.config["webhooks"]["urls"].split(",")
        self.webhook_tokens = self.config["webhooks"]["tokens"].split(",")

    def _build_post_data(self, track) -> dict:
        primary_artist = track.artists[0]
        duration = track.length and track.length // 1000 or 0
        return {
            "name": track.name,
            "artist": primary_artist.name,
            "album": track.album,
            "track_number": track.track_no,
            "run_time_ticks": track.length,
            "run_time": str(duration),
            "musicbrainz_track_id": track.musicbrainz_id,
            "musicbrainz_album_id": track.album.musicbrainz_id
            if track.album
            else None,
            "musicbrainz_artist_id": track.artist.musicbrainz_id
            if primary_artist
            else None,
        }

    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"] = "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
        self.last_start_time = int(time.time())
        logger.debug(f"Now playing track: {track.artists[0]} - {track.name}")
        post_data = self._build_post_data(tl_track.track)

        # Build post data to send to urls
        if not self.webhook_urls:
            logger.info("No webhook URLS are configured ")
            return

        self._post_update_to_webhooks(post_data, "started")

    def track_playback_ended(self, tl_track, time_position):
        track = tl_track.track
        duration = track.length and track.length // 1000 or 0
        time_position = time_position // 1000

        logger.debug(f"Now playing track: {track.artists[0]} - {track.name}")
        post_data = self._build_post_data(tl_track.track)

        if time_position < duration // 2 and time_position < 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.debug(
            f"Sending scroble to webhooks for track: {track.artists} - {track.name}"
        )

        self._post_update_to_webhooks(post_data, "stopped")