235 lines
8.5 KiB
Python
235 lines
8.5 KiB
Python
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")
|