|
@@ -4,6 +4,7 @@ from typing import Any
|
|
|
from urllib.parse import unquote
|
|
|
|
|
|
import feedparser
|
|
|
+import requests
|
|
|
from dateutil.parser import ParserError, parse
|
|
|
from podcasts.models import PodcastEpisode
|
|
|
|
|
@@ -27,30 +28,40 @@ def fetch_metadata_from_rss(uri: str) -> dict[str, Any]:
|
|
|
log_context = {"mopidy_uri": uri, "media_type": "Podcast"}
|
|
|
podcast_data: dict[str, Any] = {}
|
|
|
|
|
|
+ rss_url = uri.split("#")[0].split("podcast+")[1]
|
|
|
+ target_guid = uri.split("#")[1]
|
|
|
+
|
|
|
+ log_context["rss_url"] = rss_url
|
|
|
+ log_context["target_guid"] = target_guid
|
|
|
+
|
|
|
try:
|
|
|
- feed = feedparser.parse(uri.split("#")[0])
|
|
|
- target_guid = uri.split("#")[1]
|
|
|
+ resp = requests.get(rss_url, timeout=10)
|
|
|
+ feed = feedparser.parse(resp.text)
|
|
|
except IndexError:
|
|
|
logger.warning("Tried to parse uri as RSS feed, but no target found", extra=log_context)
|
|
|
return podcast_data
|
|
|
|
|
|
- podcast_publisher = feed.feed.get("itunes_publisher")
|
|
|
- podcast_owner = feed.feed.itunes_owner.get("name") if isinstance(feed.feed.itunes_owner, dict) else feed.feed.itunes_owner
|
|
|
- podcast_other = feed.feed.get("managingeditor") or feed.feed.get("copyright")
|
|
|
+ podcast_publisher = getattr(feed.feed, "itunes_publisher", "")
|
|
|
+ try:
|
|
|
+ podcast_owner = feed.feed.itunes_owner.get("name") if isinstance(feed.feed.itunes_owner, dict) else feed.feed.itunes_owner
|
|
|
+ podcast_other = feed.feed.get("managingeditor") or feed.feed.get("copyright")
|
|
|
+ except AttributeError:
|
|
|
+ podcast_owner = None
|
|
|
+ podcast_other = None
|
|
|
|
|
|
podcast_data = {
|
|
|
- "podcast_name": feed.feed.get("title", "Unknown Podcast"),
|
|
|
- # "podcast_description": feed.feed.get("description", ""),
|
|
|
- # "podcast_link": feed.feed.get("link", ""),
|
|
|
+ "podcast_name": getattr(feed.feed, "title", ""),
|
|
|
+ # "podcast_description": getattr(feed.feed, "description", ""),
|
|
|
+ # "podcast_link": getattr(feed.feed, "link", ""),
|
|
|
"podcast_producer": podcast_publisher or podcast_owner or podcast_other
|
|
|
}
|
|
|
|
|
|
for entry in feed.entries:
|
|
|
- if target_guid in target_guid:
|
|
|
+ if entry.get("guid") == target_guid:
|
|
|
logger.info("🎧 Episode found in RSS feed", extra=log_context)
|
|
|
podcast_data["title"] = entry.title
|
|
|
- podcast_data["episode_num"] = entry.guid
|
|
|
- podcast_data["pub_date"] = entry.get("published", None)
|
|
|
+ podcast_data["episode_num"] = int(entry.get("itunes_episode", 0))
|
|
|
+ podcast_data["pub_date"] = parse(entry.get("published", None))
|
|
|
podcast_data["run_time_seconds"] = parse_duration(entry.get("itunes_duration", None))
|
|
|
# podcast_data["description"] = entry.get("description", None)
|
|
|
# podcast_data["episode_url"] = entry.enclosures[0].href if entry.get("enclosures") else None
|
|
@@ -63,7 +74,7 @@ def parse_mopidy_uri(uri: str) -> dict[str, Any]:
|
|
|
podcast_data: dict[str, Any] = {}
|
|
|
|
|
|
logger.debug(f"Parsing URI: {uri}")
|
|
|
- if "https://" in uri:
|
|
|
+ if "podcast+https" in uri:
|
|
|
return fetch_metadata_from_rss(uri)
|
|
|
|
|
|
|