/
/
/
1"""Helpers for parsing Overcast's extended OPML export."""
2
3from __future__ import annotations
4
5from dataclasses import dataclass, field
6from datetime import UTC, datetime
7from urllib.parse import urlsplit
8from xml.parsers.expat import ExpatError
9
10import xmltodict
11from music_assistant_models.errors import InvalidDataError
12
13
14@dataclass
15class OvercastEpisodeState:
16 """Playback state of a single episode as exported by Overcast."""
17
18 overcast_id: str | None
19 title: str | None
20 enclosure_url: str | None
21 pub_date: datetime | None
22 progress_s: int | None
23 played: bool
24 user_updated_at: datetime | None
25
26
27@dataclass
28class OvercastSubscription:
29 """A single subscribed podcast feed from the Overcast export."""
30
31 xml_url: str
32 title: str | None
33 overcast_id: str | None
34 episodes: list[OvercastEpisodeState] = field(default_factory=list)
35 # lazily built enclosure url -> state index, see match_episode_state
36 _state_index: dict[str, OvercastEpisodeState] | None = field(
37 default=None, repr=False, compare=False
38 )
39
40
41def parse_extended_opml(xml_text: str) -> dict[str, OvercastSubscription]:
42 """
43 Parse Overcast's extended OPML export into the subscribed feeds, keyed by feed url.
44
45 :param xml_text: The raw OPML document as returned by the export endpoint.
46 """
47 try:
48 document = xmltodict.parse(xml_text, force_list=("outline",))
49 except ExpatError as err:
50 raise InvalidDataError("Invalid Overcast OPML document") from err
51
52 body = (document.get("opml") or {}).get("body") or {}
53 subscriptions: dict[str, OvercastSubscription] = {}
54 for group in body.get("outline", []):
55 # the export also contains a "playlists" group, only "feeds" is relevant
56 if group.get("@text") != "feeds":
57 continue
58 for feed in group.get("outline", []):
59 if feed.get("@type") != "rss":
60 continue
61 xml_url = feed.get("@xmlUrl")
62 if not xml_url or feed.get("@subscribed") != "1":
63 continue
64 episodes = [
65 _parse_episode_outline(episode)
66 for episode in feed.get("outline", [])
67 if episode.get("@type") == "podcast-episode"
68 ]
69 subscriptions[xml_url] = OvercastSubscription(
70 xml_url=xml_url,
71 title=feed.get("@title"),
72 overcast_id=feed.get("@overcastId"),
73 episodes=episodes,
74 )
75 return subscriptions
76
77
78def match_episode_state(
79 subscription: OvercastSubscription, stream_url: str
80) -> OvercastEpisodeState | None:
81 """
82 Find the Overcast playback state matching an episode's stream url.
83
84 :param subscription: The subscription holding the episode states.
85 :param stream_url: The episode's enclosure/stream url from the RSS feed.
86 """
87 if subscription._state_index is None:
88 subscription._state_index = _build_state_index(subscription.episodes)
89 if (state := subscription._state_index.get(stream_url)) is not None:
90 return state
91 # signed enclosure urls may rotate their query part between exports,
92 # so look up again with query and fragment stripped
93 return subscription._state_index.get(_strip_url(stream_url))
94
95
96def _build_state_index(
97 episodes: list[OvercastEpisodeState],
98) -> dict[str, OvercastEpisodeState]:
99 # exact urls are indexed first so they win from a stripped url of another state,
100 # and the first state wins per url to keep the export's order authoritative
101 index: dict[str, OvercastEpisodeState] = {}
102 for state in episodes:
103 if state.enclosure_url:
104 index.setdefault(state.enclosure_url, state)
105 for state in episodes:
106 if state.enclosure_url:
107 index.setdefault(_strip_url(state.enclosure_url), state)
108 return index
109
110
111def _parse_episode_outline(episode: dict[str, str]) -> OvercastEpisodeState:
112 return OvercastEpisodeState(
113 overcast_id=episode.get("@overcastId"),
114 title=episode.get("@title"),
115 enclosure_url=episode.get("@enclosureUrl"),
116 pub_date=_parse_datetime(episode.get("@pubDate")),
117 progress_s=_parse_int(episode.get("@progress")),
118 played=episode.get("@played") == "1",
119 user_updated_at=_parse_datetime(episode.get("@userUpdatedDate")),
120 )
121
122
123def _parse_datetime(value: str | None) -> datetime | None:
124 if value is None:
125 return None
126 try:
127 parsed = datetime.fromisoformat(value)
128 except ValueError:
129 return None
130 # normalize to an aware datetime so values can safely be compared
131 return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
132
133
134def _parse_int(value: str | None) -> int | None:
135 if value is None:
136 return None
137 try:
138 return int(value)
139 except ValueError:
140 return None
141
142
143def _strip_url(url: str) -> str:
144 return urlsplit(url)._replace(query="", fragment="").geturl()
145