Compare commits
14 Commits
67e74ecde2
...
0.3.0
| Author | SHA1 | Date | |
|---|---|---|---|
| 4cbe143693 | |||
| cbcc42d831 | |||
| 178496582c | |||
| b4c4cec883 | |||
| 0cac2e0a47 | |||
| 2edf49e78e | |||
| ff5f9e0c3e | |||
| 3e33f6b464 | |||
| 2db63ad49c | |||
| a140140322 | |||
| 34a03bed6a | |||
| 53b0abe03a | |||
| 47317081f9 | |||
| b988e4bf0a |
20
.gitea/workflows/ci.yml
Normal file
20
.gitea/workflows/ci.yml
Normal file
@ -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
|
||||
77
.gitea/workflows/release.yml
Normal file
77
.gitea/workflows/release.yml
Normal file
@ -0,0 +1,77 @@
|
||||
name: Release
|
||||
|
||||
on:
|
||||
push:
|
||||
tags: ["*"]
|
||||
|
||||
jobs:
|
||||
release:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Setup Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.13"
|
||||
|
||||
- name: Install build
|
||||
run: python -m pip install --upgrade pip build
|
||||
|
||||
- name: Verify tag matches package version
|
||||
run: |
|
||||
set -euo pipefail
|
||||
VERSION="${{ gitea.ref_name }}"
|
||||
PKG_VERSION=$(python -c "import configparser; c = configparser.ConfigParser(); c.read('setup.cfg'); print(c['metadata']['version'].strip())")
|
||||
echo "Tag: ${VERSION}, setup.cfg: ${PKG_VERSION}"
|
||||
test "${VERSION}" = "${PKG_VERSION}"
|
||||
|
||||
- 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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
VERSION="${{ gitea.ref_name }}"
|
||||
API="http://gitea.service:3000/api/v1/repos/${{ gitea.repository }}"
|
||||
AUTH="Authorization: token ${GITEA_TOKEN}"
|
||||
|
||||
RELEASE_ID=$(curl -sf "${API}/releases/tags/${VERSION}" -H "${AUTH}" \
|
||||
| python3 -c "import sys, json; print(json.load(sys.stdin).get('id', ''))" || true)
|
||||
|
||||
if [ -z "${RELEASE_ID}" ]; then
|
||||
echo "Creating release ${VERSION}"
|
||||
RELEASE_ID=$(curl -f -X POST "${API}/releases" \
|
||||
-H "${AUTH}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "{\"tag_name\": \"${VERSION}\", \"name\": \"${VERSION}\", \"target_commitish\": \"${{ gitea.sha }}\"}" \
|
||||
| python3 -c "import sys, json; print(json.load(sys.stdin)['id'])")
|
||||
else
|
||||
echo "Using existing release ${RELEASE_ID}"
|
||||
fi
|
||||
|
||||
for ASSET in dist/*; do
|
||||
NAME=$(basename "${ASSET}")
|
||||
OLD_ID=$(curl -sf "${API}/releases/${RELEASE_ID}/assets" -H "${AUTH}" \
|
||||
| python3 -c "import sys, json; print(next((a['id'] for a in json.load(sys.stdin) if a['name'] == '${NAME}'), ''))" || true)
|
||||
if [ -n "${OLD_ID}" ]; then
|
||||
echo "Deleting existing asset ${NAME} (id ${OLD_ID})"
|
||||
curl -f -X DELETE "${API}/releases/assets/${OLD_ID}" -H "${AUTH}"
|
||||
fi
|
||||
echo "Uploading ${NAME}"
|
||||
curl -f -X POST "${API}/releases/${RELEASE_ID}/assets?name=${NAME}" \
|
||||
-H "${AUTH}" \
|
||||
-F "attachment=@${ASSET}"
|
||||
done
|
||||
|
||||
echo "Released ${VERSION} with $(ls dist | wc -l) asset(s)"
|
||||
@ -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
|
||||
|
||||
85
README.md
Normal file
85
README.md
Normal file
@ -0,0 +1,85 @@
|
||||
Mopidy-Webhooks
|
||||
===============
|
||||
|
||||
`Mopidy <https://www.mopidy.com/>`_ extension for sending mopidy play status to
|
||||
arbitrary URL endpoints
|
||||
|
||||
Installation
|
||||
============
|
||||
|
||||
Install by running::
|
||||
|
||||
sudo python3 -m pip install Mopidy-Webhooks
|
||||
|
||||
|
||||
Configuration
|
||||
=============
|
||||
|
||||
To enable this extension, make sure to add the following variables to your
|
||||
Mopidy configuration file. Note that multiple URLs can be sent to by separating
|
||||
them with a comma. Tokens match the listing of urls::
|
||||
|
||||
[webhooks]
|
||||
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:
|
||||
|
||||
- ``webhooks/enabled``: If the webhooks extension should be enabled or not.
|
||||
Defaults to disabled.
|
||||
- ``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
|
||||
<https://mutagen.readthedocs.io/>`_: title, artist, album, track number,
|
||||
date, genre, comment/description, publisher, language, copyright and
|
||||
website.
|
||||
2. A sidecar JSON file written next to the episode (``<episode>.json``) or in
|
||||
a ``metadata/`` directory (``metadata/<episode>.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
|
||||
=================
|
||||
|
||||
- `Source code <https://code.unbl.ink/secstate/mopidy-webhooks>`_
|
||||
- `Issue tracker <https://code.unbl.ink/secstate/mopidy-webhooks/issues>`_
|
||||
- `Changelog <https://code.unbl.ink/secstate/mopidy-webhooks/releases>`_
|
||||
|
||||
|
||||
Credits
|
||||
=======
|
||||
|
||||
- Author: `Colin Powell <https://github.com/powellc>`__
|
||||
52
README.rst
52
README.rst
@ -1,52 +0,0 @@
|
||||
****************
|
||||
Mopidy-Webhooks
|
||||
****************
|
||||
|
||||
.. image:: https://img.shields.io/pypi/v/Mopidy-Webhooks
|
||||
:target: https://pypi.org/project/Mopidy-Webhooks/
|
||||
:alt: Latest PyPI version
|
||||
|
||||
`Mopidy <https://www.mopidy.com/>`_ extension for sending mopidy play status to
|
||||
arbitrary URL endpoints
|
||||
|
||||
Installation
|
||||
============
|
||||
|
||||
Install by running::
|
||||
|
||||
sudo python3 -m pip install Mopidy-Webhooks
|
||||
|
||||
|
||||
Configuration
|
||||
=============
|
||||
|
||||
To enable this extension, make sure to add the following variables to your
|
||||
Mopidy configuration file. Note that multiple URLs can be sent to by separating
|
||||
them with a comma. Tokens match the listing of urls::
|
||||
|
||||
[webhooks]
|
||||
enabled = true
|
||||
urls = https://example.com/api/receiver/,http://127.0.0.1:8000/webhook/
|
||||
tokens = ,2349080989234089
|
||||
|
||||
The following configuration values are available:
|
||||
|
||||
- ``webhooks/enabled``: If the webhooks extension should be enabled or not.
|
||||
Defaults to disabled.
|
||||
- ``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
|
||||
|
||||
|
||||
Project resources
|
||||
=================
|
||||
|
||||
- `Source code <https://github.com/powellc/mopidy-webhooks>`_
|
||||
- `Issue tracker <https://github.com/powellc/mopidy-webhooks/issues>`_
|
||||
- `Changelog <https://github.com/powellc/mopidy-webhooks/releases>`_
|
||||
|
||||
|
||||
Credits
|
||||
=======
|
||||
|
||||
- Author: `Colin Powell <https://github.com/powellc>`__
|
||||
@ -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):
|
||||
|
||||
@ -2,3 +2,5 @@
|
||||
enabled = true
|
||||
urls = ""
|
||||
tokens = ""
|
||||
media_dirs = /var/lib/mopidy/media
|
||||
podcast_dirs = ""
|
||||
|
||||
@ -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):
|
||||
logger.info("Parsing webhook URLs and tokens")
|
||||
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,23 +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")
|
||||
return {
|
||||
"name": track.name,
|
||||
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.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
|
||||
|
||||
@ -59,7 +156,7 @@ class WebhooksFrontend(pykka.ThreadingActor, CoreListener):
|
||||
logger.info(f"No token found for Webhook URL: {webhook_url}")
|
||||
|
||||
if token:
|
||||
headers["Authorization"] = "Token {token}"
|
||||
headers["Authorization"] = f"Token {token}"
|
||||
|
||||
response = requests.post(
|
||||
webhook_url, json=json.dumps(post_data), headers=headers
|
||||
@ -71,7 +168,7 @@ 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 playing track: {artists} - {track.name}")
|
||||
post_data = self._build_post_data(tl_track.track)
|
||||
post_data = self._build_post_data(tl_track.track, time_position=0)
|
||||
|
||||
# Build post data to send to urls
|
||||
if not self.webhook_urls:
|
||||
@ -84,11 +181,13 @@ 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 = time_position // 1000
|
||||
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 < duration // 2 and time_position < 240:
|
||||
if time_position_sec < duration // 2 and time_position_sec < 240:
|
||||
logger.debug(
|
||||
"Track not played long enough to scrobble. (50% or 240s)"
|
||||
)
|
||||
@ -106,21 +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 = time_position // 1000
|
||||
|
||||
post_data = self._build_post_data(tl_track.track, time_position=time_position)
|
||||
|
||||
if time_position < duration // 2 and time_position < 240:
|
||||
logger.debug(
|
||||
"Track not played long enough to scrobble. (50% or 240s)"
|
||||
)
|
||||
return
|
||||
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")
|
||||
|
||||
@ -130,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:
|
||||
|
||||
309
mopidy_webhooks/metadata.py
Normal file
309
mopidy_webhooks/metadata.py
Normal file
@ -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+<feed>#<guid>`` 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"))
|
||||
@ -1,13 +1,14 @@
|
||||
[metadata]
|
||||
name = Mopidy-Webhooks
|
||||
version = 0.1.6
|
||||
version = 0.3.0
|
||||
url = https://github.com/powellc/mopidy-webhooks
|
||||
author = Colin Powell
|
||||
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
|
||||
|
||||
|
||||
|
||||
BIN
tests/data/silence.mp3
Normal file
BIN
tests/data/silence.mp3
Normal file
Binary file not shown.
@ -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():
|
||||
|
||||
@ -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
|
||||
|
||||
186
tests/test_metadata.py
Normal file
186
tests/test_metadata.py
Normal file
@ -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",
|
||||
)
|
||||
4
tox.ini
4
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]
|
||||
|
||||
Reference in New Issue
Block a user