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"))