310 lines
8.6 KiB
Python
310 lines
8.6 KiB
Python
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"))
|