diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml
new file mode 100644
index 0000000..8c918b3
--- /dev/null
+++ b/.gitea/workflows/ci.yml
@@ -0,0 +1,20 @@
+name: CI
+
+on:
+ push:
+ branches: ["**"]
+ pull_request:
+
+jobs:
+ test:
+ runs-on: ubuntu-latest
+ container: ghcr.io/mopidy/ci:7
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v4
+
+ - name: Install package and test dependencies
+ run: python -m pip install -e ".[test]"
+
+ - name: Run tests
+ run: python -m pytest
diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml
index 32c2174..c673d70 100644
--- a/.gitea/workflows/release.yml
+++ b/.gitea/workflows/release.yml
@@ -30,6 +30,13 @@ jobs:
- name: Build sdist and wheel
run: python -m build
+ - name: Build release zip
+ run: |
+ set -euo pipefail
+ VERSION="${{ gitea.ref_name }}"
+ cd dist
+ zip "Mopidy-Webhooks-${VERSION}.zip" *.tar.gz
+
- name: Publish to Gitea release
env:
GITEA_TOKEN: ${{ secrets.GITEA }}
diff --git a/MANIFEST.in b/MANIFEST.in
index 9d9fd9b..d778be6 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -1,11 +1,13 @@
include *.py
-include *.rst
+include *.md
include .mailmap
include LICENSE
include MANIFEST.in
include pyproject.toml
+include setup.cfg
include tox.ini
+recursive-include .gitea *
recursive-include .github *
include mopidy_*/ext.conf
diff --git a/README.md b/README.md
index 7ff7ebd..3818d48 100644
--- a/README.md
+++ b/README.md
@@ -23,6 +23,8 @@ them with a comma. Tokens match the listing of urls::
enabled = true
urls = https://example.com/api/receiver/,http://127.0.0.1:8000/webhook/
tokens = ,2349080989234089
+ media_dirs = /var/lib/mopidy/media
+ podcast_dirs = /var/lib/mopidy/media/podcasts
The following configuration values are available:
@@ -31,6 +33,42 @@ The following configuration values are available:
- ``webhooks/urls``: Comma-separated list of endpoints to send play data to
- ``webhooks/tokens``: Comma-separated list of tokens to send in the
Authorization header for each URL
+- ``webhooks/media_dirs``: Comma-separated list of local media roots used to
+ resolve ``local:track:``/``local:podcast:`` URIs to files so their tags can
+ be read. Defaults to ``/var/lib/mopidy/media``.
+- ``webhooks/podcast_dirs``: Comma-separated list of directories that should
+ always be treated as podcasts, even when the URI does not mention it.
+
+
+Podcast metadata
+================
+
+When a track is detected as a podcast episode, the webhook payload is enriched
+with metadata read from the audio file itself, so the receiver no longer has to
+guess the podcast, episode number or publication date from the file name.
+
+Sources, in order of preference:
+
+1. Audio tags (ID3, Vorbis, MP4) read with `mutagen
+ `_: title, artist, album, track number,
+ date, genre, comment/description, publisher, language, copyright and
+ website.
+2. A sidecar JSON file written next to the episode (``.json``) or in
+ a ``metadata/`` directory (``metadata/.json``), as produced by
+ common podcatchers. This provides ``feed_url``, ``guid``, ``episode_url``,
+ ``duration``, ``image`` and a fallback ``description``.
+
+Podcast episodes are detected when the URI contains ``podcast``, when the file
+lives under one of the ``webhooks/podcast_dirs``, or when the file has a
+sidecar JSON with a feed URL or guid.
+
+In addition to the regular fields, podcast payloads include:
+
+- ``media_type``: ``"podcast"`` or ``"track"``
+- ``podcast_name``, ``podcast_producer``, ``podcast_description``
+- ``podcast_feed_url``, ``podcast_site_link``
+- ``episode_num``, ``pub_date`` (``YYYY-MM-DD``)
+- ``episode_description``, ``episode_url``, ``episode_guid``
Project resources
diff --git a/mopidy_webhooks/__init__.py b/mopidy_webhooks/__init__.py
index a7b4aa5..ccbae27 100644
--- a/mopidy_webhooks/__init__.py
+++ b/mopidy_webhooks/__init__.py
@@ -20,6 +20,8 @@ class Extension(ext.Extension):
schema = super().get_config_schema()
schema["urls"] = config.String()
schema["tokens"] = config.Secret(optional=True)
+ schema["media_dirs"] = config.String(optional=True)
+ schema["podcast_dirs"] = config.String(optional=True)
return schema
def setup(self, registry):
diff --git a/mopidy_webhooks/ext.conf b/mopidy_webhooks/ext.conf
index f1821ee..f225d3c 100644
--- a/mopidy_webhooks/ext.conf
+++ b/mopidy_webhooks/ext.conf
@@ -2,3 +2,5 @@
enabled = true
urls = ""
tokens = ""
+media_dirs = /var/lib/mopidy/media
+podcast_dirs = ""
diff --git a/mopidy_webhooks/frontend.py b/mopidy_webhooks/frontend.py
index 7163550..52b9944 100644
--- a/mopidy_webhooks/frontend.py
+++ b/mopidy_webhooks/frontend.py
@@ -1,12 +1,15 @@
import logging
+import os
import time
import json
-from typing import Optional
+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__)
@@ -15,14 +18,84 @@ class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
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 _build_post_data(self, track, time_position: Optional[int]=None) -> dict:
+ 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:
@@ -30,25 +103,47 @@ class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
except IndexError:
musicbrainz_artist_id = None
duration = track.length and track.length // 1000 or 0
- album_name = ""
- if track.album:
- album_name = getattr(track.album, "name")
+ album_name = track.album.name if track.album else ""
- return {
- "name": track.name,
+ 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.track_no,
+ "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_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
@@ -88,7 +183,9 @@ class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
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)
+ 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(
@@ -108,15 +205,14 @@ class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
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)
+ 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}"
- )
+ logger.info(f"Scrobbling paused via webhooks: {artists} - {track.name}")
self._post_update_to_webhooks(post_data, "paused")
@@ -126,7 +222,9 @@ class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
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)
+ 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:
diff --git a/mopidy_webhooks/metadata.py b/mopidy_webhooks/metadata.py
new file mode 100644
index 0000000..fde2980
--- /dev/null
+++ b/mopidy_webhooks/metadata.py
@@ -0,0 +1,309 @@
+import json
+import logging
+import os
+import re
+from typing import Any, Dict, Iterable, List, Optional
+from urllib.parse import unquote, urlparse
+from urllib.request import url2pathname
+
+try:
+ import mutagen
+except ImportError: # pragma: no cover
+ mutagen = None
+
+logger = logging.getLogger(__name__)
+
+PODCAST_URI_PREFIX = "podcast+"
+
+_LOCAL_URI_TYPES = frozenset(
+ {"album", "artist", "directory", "image", "podcast", "track"}
+)
+
+_SIMPLE_TAG_KEYS = {
+ "title": ("title",),
+ "artist": ("artist", "albumartist"),
+ "album": ("album",),
+ "track_number": ("tracknumber",),
+ "date": ("date", "year"),
+ "genre": ("genre",),
+ "publisher": ("publisher", "organization"),
+ "language": ("language",),
+ "copyright": ("copyright",),
+}
+
+# Fields that are common in podcast files but not exposed by mutagen's
+# "easy" interface (notably the ID3 COMM frame holding the description).
+_RAW_TAG_KEYS = {
+ "comment": (
+ "comment",
+ "COMM::eng",
+ "COMM",
+ "description",
+ "desc",
+ "\xa9cmt",
+ ),
+ "website": ("website", "WXXX", "WOAR", "url", "\xa9url"),
+}
+
+_SIDECAR_KEYS = (
+ "feed_url",
+ "guid",
+ "episode_url",
+ "description",
+ "duration",
+ "image",
+)
+
+_DATE_RE = re.compile(r"(\d{4})(?:-(\d{2})(?:-(\d{2}))?)?")
+
+
+def parse_dirs(value: Any) -> List[str]:
+ """Split a comma-separated config value into a list of directories."""
+ if not value:
+ return []
+ if isinstance(value, (list, tuple)):
+ values = value
+ else:
+ values = str(value).split(",")
+ return [
+ str(directory).strip() for directory in values if str(directory).strip()
+ ]
+
+
+def split_podcast_uri(uri: Optional[str]) -> tuple:
+ """Split a ``podcast+#`` URI into its feed and guid parts."""
+ if not uri or not uri.startswith(PODCAST_URI_PREFIX):
+ return None, None
+ feed_url, _, guid = uri[len(PODCAST_URI_PREFIX) :].partition("#")
+ return feed_url or None, guid or None
+
+
+def _local_relative_path(uri: str) -> Optional[str]:
+ rest = uri.split(":", 1)[1] if ":" in uri else uri
+ if ":" in rest:
+ kind, relative = rest.split(":", 1)
+ if kind in _LOCAL_URI_TYPES:
+ return relative
+ return rest
+
+
+def resolve_media_path(
+ uri: Optional[str], media_dirs: Optional[Iterable[str]] = None
+) -> Optional[str]:
+ """Resolve a Mopidy track URI to a local filesystem path if possible."""
+ if not uri:
+ return None
+
+ parsed = urlparse(uri)
+ if parsed.scheme == "file":
+ path = url2pathname(parsed.path)
+ return path or None
+
+ if parsed.scheme != "local":
+ return None
+
+ relative = _local_relative_path(uri)
+ if not relative:
+ return None
+
+ relative = unquote(relative)
+ if os.path.isabs(relative):
+ return relative
+
+ directories = list(media_dirs or [])
+ for media_dir in directories:
+ candidate = os.path.join(media_dir, relative)
+ if os.path.exists(candidate):
+ return candidate
+
+ if directories:
+ return os.path.join(directories[0], relative)
+ return None
+
+
+def is_under(path: Optional[str], directories: Iterable[str]) -> bool:
+ """Return True when ``path`` lives under one of ``directories``."""
+ if not path:
+ return False
+ real_path = os.path.realpath(path)
+ for directory in directories:
+ real_dir = os.path.realpath(directory)
+ if real_path == real_dir or real_path.startswith(real_dir + os.sep):
+ return True
+ return False
+
+
+def _frame_text(frame: Any) -> Optional[str]:
+ text = getattr(frame, "text", None)
+ if text:
+ value = text[0] if isinstance(text, (list, tuple)) else text
+ return str(value).strip() or None
+ if isinstance(frame, (list, tuple)):
+ return str(frame[0]).strip() or None
+ return str(frame).strip() or None
+
+
+def _first_raw_tag_value(path: str, keys: Iterable[str]) -> Optional[str]:
+ if mutagen is None:
+ return None
+ try:
+ audio = mutagen.File(path)
+ except Exception: # pragma: no cover - defensive against bad files
+ return None
+ tags = getattr(audio, "tags", None)
+ if not tags:
+ return None
+ for key in keys:
+ frame = None
+ try:
+ frame = tags.get(key)
+ except Exception:
+ frame = None
+ if frame is None and hasattr(tags, "getall"):
+ try:
+ frames = tags.getall(key)
+ except Exception:
+ frames = []
+ frame = frames[0] if frames else None
+ if frame is None:
+ continue
+ value = _frame_text(frame)
+ if value:
+ return value
+ return None
+
+
+def _first_tag_value(audio: Any, keys: Iterable[str]) -> Optional[str]:
+ for key in keys:
+ try:
+ value = audio.get(key)
+ except Exception: # pragma: no cover - defensive against bad tags
+ continue
+ if not value:
+ continue
+ if isinstance(value, (list, tuple)):
+ value = value[0]
+ value = str(value).strip()
+ if value:
+ return value
+ return None
+
+
+def read_tags(path: Optional[str]) -> Dict[str, str]:
+ """Read normalized tags from an audio file with mutagen."""
+ if mutagen is None or not path or not os.path.isfile(path):
+ return {}
+
+ try:
+ audio = mutagen.File(path, easy=True)
+ except Exception:
+ logger.warning("Could not read audio tags from %s", path, exc_info=True)
+ return {}
+
+ if audio is None:
+ return {}
+
+ metadata: Dict[str, str] = {}
+ for field, keys in _SIMPLE_TAG_KEYS.items():
+ value = _first_tag_value(audio, keys)
+ if value:
+ metadata[field] = value
+
+ for field, keys in _RAW_TAG_KEYS.items():
+ if metadata.get(field):
+ continue
+ value = _first_raw_tag_value(path, keys)
+ if value:
+ metadata[field] = value
+
+ track_number = metadata.get("track_number")
+ if track_number:
+ metadata["track_number"] = track_number.split("/")[0].strip()
+
+ return metadata
+
+
+def sidecar_paths(path: str) -> List[str]:
+ """Return the candidate sidecar JSON paths for an audio file."""
+ stem, _ = os.path.splitext(path)
+ return [
+ stem + ".json",
+ os.path.join(
+ os.path.dirname(path), "metadata", os.path.basename(stem) + ".json"
+ ),
+ ]
+
+
+def read_sidecar_json(path: str) -> Dict[str, Any]:
+ """Read a podcatcher/yt-dlp style sidecar JSON file if present."""
+ for candidate in sidecar_paths(path):
+ if not os.path.isfile(candidate):
+ continue
+ try:
+ with open(candidate, encoding="utf-8") as sidecar:
+ data = json.load(sidecar)
+ except (OSError, ValueError):
+ logger.warning("Could not read sidecar metadata from %s", candidate)
+ continue
+ if isinstance(data, dict):
+ return data
+ return {}
+
+
+def normalize_date(value: Any) -> str:
+ """Normalize a tag/sidecar date to a ``YYYY-MM-DD`` string when possible."""
+ if not value:
+ return ""
+ if hasattr(value, "strftime"):
+ return value.strftime("%Y-%m-%d")
+ match = _DATE_RE.search(str(value))
+ if not match or not match.group(2):
+ return ""
+ year, month, day = match.groups()
+ return "-".join((year, month, day or "01"))
+
+
+def read_metadata(path: Optional[str]) -> Dict[str, Any]:
+ """Read tags and sidecar metadata for a local media file.
+
+ Audio tags win for title/artist/album/genre, while the sidecar JSON is the
+ source for podcast specific fields such as the feed URL and episode guid.
+ """
+ if not path:
+ return {}
+
+ metadata: Dict[str, Any] = {}
+ sidecar = read_sidecar_json(path)
+ for key in _SIDECAR_KEYS:
+ value = sidecar.get(key)
+ if value:
+ metadata[key] = value
+ for field in ("title", "artist", "album", "date"):
+ value = sidecar.get(field)
+ if value:
+ metadata.setdefault(field, str(value))
+
+ metadata.update(
+ {key: value for key, value in read_tags(path).items() if value}
+ )
+
+ if sidecar.get("date"):
+ metadata["date"] = str(sidecar["date"])
+
+ return metadata
+
+
+def is_podcast(
+ uri: Optional[str],
+ path: Optional[str] = None,
+ metadata: Optional[Dict[str, Any]] = None,
+ podcast_dirs: Optional[Iterable[str]] = None,
+) -> bool:
+ """Best effort detection of podcast episodes."""
+ uri = uri or ""
+ if PODCAST_URI_PREFIX in uri or "podcast" in uri.lower():
+ return True
+ if path and is_under(path, podcast_dirs or []):
+ return True
+ metadata = metadata or {}
+ return bool(metadata.get("feed_url") or metadata.get("guid"))
diff --git a/setup.cfg b/setup.cfg
index 6a82554..0812eea 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -7,7 +7,8 @@ author_email = colin@unbl.ink
license = Apache License, Version 2.0
license_file = LICENSE
description = Mopidy extension for sending playback data to webhook urls
-long_description = file: README.rst
+long_description = file: README.md
+long_description_content_type = text/markdown
classifiers =
Environment :: No Input/Output (Daemon)
Intended Audience :: End Users/Desktop
@@ -28,6 +29,7 @@ python_requires = >= 3.7
install_requires =
Mopidy >= 3.0.0
Pykka >= 2.0.1
+ mutagen >= 1.45
setuptools
diff --git a/tests/data/silence.mp3 b/tests/data/silence.mp3
new file mode 100644
index 0000000..a90485e
Binary files /dev/null and b/tests/data/silence.mp3 differ
diff --git a/tests/test_extension.py b/tests/test_extension.py
index 24bb5ff..9e88f67 100644
--- a/tests/test_extension.py
+++ b/tests/test_extension.py
@@ -11,8 +11,9 @@ def test_get_default_config():
assert "[webhooks]" in config
assert "enabled = true" in config
- assert "username =" in config
- assert "password =" in config
+ assert "urls =" in config
+ assert "media_dirs =" in config
+ assert "podcast_dirs =" in config
def test_get_config_schema():
@@ -20,8 +21,10 @@ def test_get_config_schema():
schema = ext.get_config_schema()
- assert "username" in schema
- assert "password" in schema
+ assert "urls" in schema
+ assert "tokens" in schema
+ assert "media_dirs" in schema
+ assert "podcast_dirs" in schema
def test_setup():
diff --git a/tests/test_frontend.py b/tests/test_frontend.py
index 156e71b..43c5615 100644
--- a/tests/test_frontend.py
+++ b/tests/test_frontend.py
@@ -4,6 +4,7 @@ import pytest
from mopidy import models
from mopidy_webhooks import frontend as frontend_lib
+from mopidy_webhooks import metadata as metadata_lib
@pytest.fixture
@@ -12,13 +13,41 @@ def frontend():
"webhooks": {
"urls": "http://127.0.0.1/receiver/,http://127.0.0.1/receiver/two/",
"tokens": "secrettoken,anotherone",
+ "media_dirs": "",
+ "podcast_dirs": "",
}
}
core = mock.sentinel.core
return frontend_lib.WebhooksFrontend(config, core)
-def test_on_start_creates_lastfm_network(pylast_mock, frontend):
+@pytest.fixture
+def track():
+ return models.Track(
+ uri="local:track:Sublime%20-%20Sublime/Disc%201%20-%2004%20-%20Same%20in%20the%20End.mp3",
+ name="Same in the End",
+ artists=[models.Artist(name="Sublime", uri="local:artist:md5:abc")],
+ album=models.Album(name="Sublime"),
+ track_no=4,
+ length=156604,
+ )
+
+
+@pytest.fixture
+def podcast_track():
+ return models.Track(
+ uri="local:podcast:TED%20Talks/2026-09-22_Do%20the%20hard%20thing.mp3",
+ name="2026-09-22_Do the hard thing",
+ artists=[models.Artist(name="TED Talks Daily")],
+ album=models.Album(name="TED Talks Daily"),
+ length=1800000,
+ )
+
+
+def test_on_start_parses_urls_tokens_and_dirs(frontend):
+ frontend.config["webhooks"]["media_dirs"] = "/media,/srv/media"
+ frontend.config["webhooks"]["podcast_dirs"] = "/media/podcasts"
+
frontend.on_start()
assert frontend.webhook_urls == [
@@ -26,144 +55,176 @@ def test_on_start_creates_lastfm_network(pylast_mock, frontend):
"http://127.0.0.1/receiver/two/",
]
assert frontend.webhook_tokens == ["secrettoken", "anotherone"]
+ assert frontend.media_dirs == ["/media", "/srv/media"]
+ assert frontend.podcast_dirs == ["/media/podcasts"]
-def test_on_start_stops_actor_on_error(pylast_mock, frontend):
- pylast_mock.NetworkError = pylast.NetworkError
- pylast_mock.LastFMNetwork.side_effect = pylast.NetworkError(None, "foo")
- frontend.stop = mock.Mock()
+def test_build_post_data_for_track(frontend, track):
+ post_data = frontend._build_post_data(track, time_position=0)
+ assert post_data["name"] == "Same in the End"
+ assert post_data["artist"] == "Sublime"
+ assert post_data["album"] == "Sublime"
+ assert post_data["track_number"] == 4
+ assert post_data["run_time"] == "156"
+ assert post_data["media_type"] == "track"
+ assert "podcast_name" not in post_data
+
+
+def test_build_post_data_for_podcast_prefers_tags(
+ frontend, podcast_track, monkeypatch
+):
+ monkeypatch.setattr(
+ metadata_lib,
+ "resolve_media_path",
+ lambda uri, dirs: "/media/podcasts/ep.mp3",
+ )
+ monkeypatch.setattr(
+ metadata_lib,
+ "read_metadata",
+ lambda path: {
+ "title": "#564: EVE Online Departs for Python 3",
+ "artist": "Michael Kennedy",
+ "album": "Talk Python To Me",
+ "track_number": "564",
+ "date": "2026-09-11T15:46:15",
+ "comment": "A great episode",
+ "feed_url": "https://talkpython.fm/episodes/rss",
+ "guid": "talk-python-564",
+ "episode_url": "https://talkpython.fm/564",
+ "duration": 4071,
+ },
+ )
+ monkeypatch.setattr(
+ metadata_lib, "is_podcast", lambda uri, **kwargs: "podcast" in uri
+ )
+
+ post_data = frontend._build_post_data(podcast_track, time_position=0)
+
+ assert post_data["media_type"] == "podcast"
+ assert post_data["name"] == "#564: EVE Online Departs for Python 3"
+ assert post_data["artist"] == "Michael Kennedy"
+ assert post_data["album"] == "Talk Python To Me"
+ assert post_data["track_number"] == "564"
+ assert post_data["podcast_name"] == "Talk Python To Me"
+ assert post_data["podcast_producer"] == "Michael Kennedy"
+ assert post_data["podcast_feed_url"] == "https://talkpython.fm/episodes/rss"
+ assert post_data["podcast_description"] == "A great episode"
+ assert post_data["episode_num"] == 564
+ assert post_data["pub_date"] == "2026-09-11"
+ assert post_data["episode_url"] == "https://talkpython.fm/564"
+ assert post_data["episode_guid"] == "talk-python-564"
+ assert post_data["duration_seconds"] == 4071
+
+
+def test_build_post_data_for_podcast_falls_back_to_track(
+ frontend, podcast_track
+):
+ post_data = frontend._build_post_data(podcast_track, time_position=0)
+
+ assert post_data["media_type"] == "podcast"
+ assert post_data["name"] == podcast_track.name
+ assert post_data["podcast_name"] == "TED Talks Daily"
+ assert post_data["podcast_producer"] == "TED Talks Daily"
+ assert post_data["podcast_feed_url"] == ""
+
+
+def test_build_post_data_for_remote_podcast_uri(frontend):
+ track = models.Track(
+ uri="podcast+https://feeds.npr.org/510318/podcast.xml#85b9c4c4-guid",
+ name="UN General Assembly Week",
+ artists=[models.Artist(name="NPR")],
+ album=models.Album(name="Up First"),
+ length=900000,
+ )
+
+ post_data = frontend._build_post_data(track, time_position=0)
+
+ assert post_data["media_type"] == "podcast"
+ assert post_data["podcast_name"] == "Up First"
+ assert (
+ post_data["podcast_feed_url"]
+ == "https://feeds.npr.org/510318/podcast.xml"
+ )
+ assert post_data["episode_guid"] == "85b9c4c4-guid"
+
+
+@mock.patch("mopidy_webhooks.frontend.requests.post")
+def test_track_playback_started_posts_started(mock_post, frontend, track):
frontend.on_start()
- frontend.stop.assert_called_with()
+ frontend.track_playback_started(models.TlTrack(track=track, tlid=1))
+
+ assert mock_post.call_count == 2
+ first_call = mock_post.call_args_list[0]
+ assert first_call.args[0] == "http://127.0.0.1/receiver/"
+ assert first_call.kwargs["headers"] == {
+ "Authorization": "Token secrettoken"
+ }
+ assert '"status": "started"' in first_call.kwargs["json"]
-def test_track_playback_started_updates_now_playing(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- artists = [models.Artist(name="ABC"), models.Artist(name="XYZ")]
- album = models.Album(name="The Collection")
+@mock.patch("mopidy_webhooks.frontend.requests.post")
+def test_track_playback_ended_scrobbles_played_track(
+ mock_post, frontend, track
+):
+ frontend.on_start()
+
+ frontend.track_playback_ended(models.TlTrack(track=track, tlid=1), 150000)
+
+ assert mock_post.call_count == 2
+ assert '"status": "stopped"' in mock_post.call_args.kwargs["json"]
+
+
+@mock.patch("mopidy_webhooks.frontend.requests.post")
+def test_does_not_scrobble_if_played_less_than_half(mock_post, frontend, track):
+ frontend.on_start()
+
+ frontend.track_playback_ended(models.TlTrack(track=track, tlid=1), 60432)
+
+ assert mock_post.call_count == 0
+
+
+@mock.patch("mopidy_webhooks.frontend.requests.post")
+def test_track_playback_paused_and_resumed_post(mock_post, frontend, track):
+ frontend.on_start()
+
+ frontend.track_playback_paused(models.TlTrack(track=track, tlid=1), 15000)
+ frontend.track_playback_resumed(models.TlTrack(track=track, tlid=1), 15000)
+
+ statuses = [call.kwargs["json"] for call in mock_post.call_args_list]
+ assert any('"status": "paused"' in payload for payload in statuses)
+ assert any('"status": "resumed"' in payload for payload in statuses)
+
+
+@mock.patch("mopidy_webhooks.frontend.requests.post")
+def test_no_urls_configured_posts_nothing(mock_post, frontend, track):
+ frontend.on_start()
+ frontend.webhook_urls = []
+
+ frontend.track_playback_started(models.TlTrack(track=track, tlid=1))
+
+ assert mock_post.call_count == 0
+
+
+def test_metadata_cache_avoids_repeated_reads(frontend, monkeypatch, tmp_path):
+ media_file = tmp_path / "song.mp3"
+ media_file.write_bytes(b"data")
+ frontend.media_dirs = [str(tmp_path)]
track = models.Track(
- name="One Two Three",
- artists=artists,
- album=album,
- track_no=3,
- length=180432,
- musicbrainz_id="123-456",
+ uri=f"local:track:{media_file.name}",
+ name="Song",
+ artists=[models.Artist(name="Someone", uri="local:artist:someone")],
+ length=180000,
)
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_started(tl_track)
-
- frontend.lastfm.update_now_playing.assert_called_with(
- "ABC, XYZ",
- "One Two Three",
- duration="180",
- album="The Collection",
- track_number="3",
- mbid="123-456",
+ reads = []
+ monkeypatch.setattr(
+ metadata_lib,
+ "read_metadata",
+ lambda path: reads.append(path) or {"title": "Tagged"},
)
+ frontend._build_post_data(track, time_position=0)
+ frontend._build_post_data(track, time_position=1000)
-def test_track_playback_started_has_default_values(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- track = models.Track()
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_started(tl_track)
-
- frontend.lastfm.update_now_playing.assert_called_with(
- "", "", duration="0", album="", track_number="0", mbid=""
- )
-
-
-def test_track_playback_started_catches_pylast_error(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- pylast_mock.NetworkError = pylast.NetworkError
- frontend.lastfm.update_now_playing.side_effect = pylast.NetworkError(
- None, "foo"
- )
- track = models.Track()
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_started(tl_track)
-
-
-def test_track_playback_ended_scrobbles_played_track(pylast_mock, frontend):
- frontend.last_start_time = 123
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- artists = [models.Artist(name="ABC"), models.Artist(name="XYZ")]
- album = models.Album(name="The Collection")
- track = models.Track(
- name="One Two Three",
- artists=artists,
- album=album,
- track_no=3,
- length=180432,
- musicbrainz_id="123-456",
- )
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_ended(tl_track, 150000)
-
- frontend.lastfm.scrobble.assert_called_with(
- "ABC, XYZ",
- "One Two Three",
- "123",
- duration="180",
- album="The Collection",
- track_number="3",
- mbid="123-456",
- )
-
-
-def test_track_playback_ended_has_default_values(pylast_mock, frontend):
- frontend.last_start_time = 123
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- track = models.Track(length=180432)
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_ended(tl_track, 150000)
-
- frontend.lastfm.scrobble.assert_called_with(
- "", "", "123", duration="180", album="", track_number="0", mbid=""
- )
-
-
-def test_does_not_scrobble_tracks_shorter_than_30_sec(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- track = models.Track(length=20432)
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_ended(tl_track, 20432)
-
- assert frontend.lastfm.scrobble.call_count == 0
-
-
-def test_does_not_scrobble_if_played_less_than_half(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- track = models.Track(length=180432)
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_ended(tl_track, 60432)
-
- assert frontend.lastfm.scrobble.call_count == 0
-
-
-def test_does_scrobble_if_played_not_half_but_240_sec(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- track = models.Track(length=880432)
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_ended(tl_track, 241432)
-
- assert frontend.lastfm.scrobble.call_count == 1
-
-
-def test_track_playback_ended_catches_pylast_error(pylast_mock, frontend):
- frontend.lastfm = mock.Mock(spec=pylast.LastFMNetwork)
- pylast_mock.NetworkError = pylast.NetworkError
- frontend.lastfm.scrobble.side_effect = pylast.NetworkError(None, "foo")
- track = models.Track(length=180432)
- tl_track = models.TlTrack(track=track, tlid=17)
-
- frontend.track_playback_ended(tl_track, 150000)
+ assert len(reads) == 1
diff --git a/tests/test_metadata.py b/tests/test_metadata.py
new file mode 100644
index 0000000..4b98917
--- /dev/null
+++ b/tests/test_metadata.py
@@ -0,0 +1,186 @@
+import json
+import shutil
+from pathlib import Path
+
+import pytest
+from mutagen import File as MutagenFile
+from mutagen.id3 import COMM, WXXX
+
+from mopidy_webhooks import metadata
+
+DATA_DIR = Path(__file__).parent / "data"
+
+
+@pytest.fixture
+def tagged_episode(tmp_path):
+ episode = tmp_path / "2026-09-22 Episode One.mp3"
+ shutil.copy(DATA_DIR / "silence.mp3", episode)
+ audio = MutagenFile(episode, easy=True)
+ audio["title"] = ["Episode One"]
+ audio["artist"] = ["Jane Host"]
+ audio["album"] = ["The Test Podcast"]
+ audio["tracknumber"] = ["42"]
+ audio["date"] = ["2026-09-22"]
+ audio["genre"] = ["Technology"]
+ audio.save()
+
+ raw = MutagenFile(episode)
+ raw.tags.add(
+ COMM(encoding=3, lang="eng", desc="", text="A tag description")
+ )
+ raw.tags.add(WXXX(encoding=3, desc="", url="https://example.com/episode"))
+ raw.save()
+ return episode
+
+
+def test_resolve_media_path_for_file_uri(tagged_episode):
+ path = metadata.resolve_media_path(tagged_episode.as_uri(), [])
+
+ assert path == str(tagged_episode)
+
+
+def test_resolve_media_path_for_local_uri(tagged_episode, tmp_path):
+ uri = f"local:podcast:{tagged_episode.name}"
+
+ path = metadata.resolve_media_path(uri, [str(tmp_path)])
+
+ assert path == str(tagged_episode)
+
+
+def test_resolve_media_path_for_local_track_uri(tagged_episode, tmp_path):
+ uri = f"local:track:{tagged_episode.name}"
+
+ path = metadata.resolve_media_path(uri, [str(tmp_path)])
+
+ assert path == str(tagged_episode)
+
+
+def test_resolve_media_path_percent_decodes_uri(tagged_episode, tmp_path):
+ uri = "local:podcast:2026-09-22%20Episode%20One.mp3"
+
+ path = metadata.resolve_media_path(uri, [str(tmp_path)])
+
+ assert path == str(tagged_episode)
+
+
+def test_resolve_media_path_returns_none_for_remote_uri():
+ assert metadata.resolve_media_path("spotify:track:123", []) is None
+ assert (
+ metadata.resolve_media_path("podcast+https://example.com/feed#1", [])
+ is None
+ )
+ assert metadata.resolve_media_path("", []) is None
+
+
+def test_read_tags_returns_normalized_metadata(tagged_episode):
+ tags = metadata.read_tags(str(tagged_episode))
+
+ assert tags["title"] == "Episode One"
+ assert tags["artist"] == "Jane Host"
+ assert tags["album"] == "The Test Podcast"
+ assert tags["track_number"] == "42"
+ assert tags["date"] == "2026-09-22"
+ assert tags["genre"] == "Technology"
+ assert tags["comment"] == "A tag description"
+ assert tags["website"] == "https://example.com/episode"
+
+
+def test_read_tags_handles_missing_file():
+ assert metadata.read_tags("/does/not/exist.mp3") == {}
+ assert metadata.read_tags(None) == {}
+
+
+def test_read_sidecar_json_prefers_same_directory(tagged_episode):
+ sidecar = tagged_episode.with_suffix(".json")
+ sidecar.write_text(json.dumps({"feed_url": "https://example.com/feed"}))
+
+ data = metadata.read_sidecar_json(str(tagged_episode))
+
+ assert data["feed_url"] == "https://example.com/feed"
+
+
+def test_read_metadata_merges_sidecar_and_tags(tagged_episode, tmp_path):
+ metadata_dir = tmp_path / "metadata"
+ metadata_dir.mkdir()
+ sidecar = metadata_dir / f"{tagged_episode.stem}.json"
+ sidecar.write_text(
+ json.dumps(
+ {
+ "title": "Sidecar Title",
+ "date": "2026-09-21T10:00:00",
+ "feed_url": "https://example.com/feed",
+ "guid": "episode-guid",
+ "episode_url": "https://example.com/episode.mp3",
+ "description": "A sidecar description",
+ "duration": 1234,
+ }
+ )
+ )
+
+ data = metadata.read_metadata(str(tagged_episode))
+
+ assert data["title"] == "Episode One"
+ assert data["artist"] == "Jane Host"
+ assert data["album"] == "The Test Podcast"
+ assert data["feed_url"] == "https://example.com/feed"
+ assert data["guid"] == "episode-guid"
+ assert data["episode_url"] == "https://example.com/episode.mp3"
+ assert data["description"] == "A sidecar description"
+ assert data["duration"] == 1234
+ assert data["date"] == "2026-09-21T10:00:00"
+
+
+@pytest.mark.parametrize(
+ "value,expected",
+ [
+ ("2026-09-22", "2026-09-22"),
+ ("2026-09-22T10:00:00", "2026-09-22"),
+ ("2026-09", "2026-09-01"),
+ ("2026", ""),
+ ("", ""),
+ ("not a date", ""),
+ ],
+)
+def test_normalize_date(value, expected):
+ assert metadata.normalize_date(value) == expected
+
+
+def test_normalize_date_handles_date_objects():
+ import datetime
+
+ assert metadata.normalize_date(datetime.date(2026, 9, 22)) == "2026-09-22"
+
+
+def test_split_podcast_uri():
+ feed, guid = metadata.split_podcast_uri(
+ "podcast+https://feed.example/rss#abc"
+ )
+
+ assert feed == "https://feed.example/rss"
+ assert guid == "abc"
+ assert metadata.split_podcast_uri("local:track:foo.mp3") == (None, None)
+
+
+def test_parse_dirs():
+ assert metadata.parse_dirs("/a, /b ,, ") == ["/a", "/b"]
+ assert metadata.parse_dirs("") == []
+ assert metadata.parse_dirs(None) == []
+ assert metadata.parse_dirs(["/a", "/b"]) == ["/a", "/b"]
+
+
+def test_is_podcast_detection(tagged_episode, tmp_path):
+ assert metadata.is_podcast("local:podcast:foo.mp3")
+ assert metadata.is_podcast("podcast+https://feed.example/rss#abc")
+ assert metadata.is_podcast("file:///media/podcasts/foo.mp3")
+ assert metadata.is_podcast(
+ "file:///tmp/foo.mp3",
+ path=str(tagged_episode),
+ podcast_dirs=[str(tmp_path)],
+ )
+ assert metadata.is_podcast(
+ "file:///tmp/foo.mp3", path="/tmp/foo.mp3", metadata={"feed_url": "x"}
+ )
+ assert not metadata.is_podcast(
+ "local:track:Artist/Album/01%20Song.mp3",
+ path="/tmp/song.mp3",
+ )
diff --git a/tox.ini b/tox.ini
index 565e27a..69b6a3e 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
[tox]
-envlist = py37, py38, py39, check-manifest, flake8
+envlist = py310, py311, py312, py313, check-manifest, flake8
[testenv]
sitepackages = true
@@ -7,7 +7,7 @@ deps = .[test]
commands =
python -m pytest \
--basetemp={envtmpdir} \
- --cov=mopidy_webhooks--cov-report=term-missing \
+ --cov=mopidy_webhooks --cov-report=term-missing \
{posargs}
[testenv:check-manifest]