/
/
/
1"""Typed dataclasses + parsers for ORF Radiothek / ORF Sound provider."""
2
3from __future__ import annotations
4
5from collections.abc import Iterable
6from dataclasses import dataclass
7from typing import Any
8
9
10@dataclass(frozen=True, slots=True)
11class StreamRef:
12 """Stream reference parsed from ORF bundle."""
13
14 url: str
15 format: str | None = None
16
17
18@dataclass(frozen=True, slots=True)
19class OrfStation:
20 """Create an ORF station from a bundle entry."""
21
22 id: str
23 name: str
24 live_stream_url_template: str
25 hide_from_stations: bool = False
26 # optional fields that exist in bundle.json and can be useful later
27 timeshift_hls_url_template: str | None = None
28 timeshift_progressive_url_template: str | None = None
29 podcasts_available: bool | None = None
30
31 @classmethod
32 def from_bundle_item(cls, station_id: str, obj: dict[str, Any]) -> OrfStation | None:
33 """Create an ORF station from a bundle entry."""
34 tmpl = obj.get("liveStreamUrlTemplate")
35 if not isinstance(tmpl, str) or "{quality}" not in tmpl:
36 return None
37 name = obj.get("name")
38 if not isinstance(name, str) or not name:
39 name = station_id
40
41 # optional extras (keep loose; bundle varies)
42 ts = obj.get("timeshift")
43 ts_hls = ts.get("liveStreamUrlTemplateHls") if isinstance(ts, dict) else None
44 ts_prog = ts.get("liveStreamUrlTemplateProgressive") if isinstance(ts, dict) else None
45 if not isinstance(ts_hls, str):
46 ts_hls = None
47 if not isinstance(ts_prog, str):
48 ts_prog = None
49
50 podcasts = obj.get("podcasts")
51 podcasts_avail = podcasts.get("available") if isinstance(podcasts, dict) else None
52 if not isinstance(podcasts_avail, bool):
53 podcasts_avail = None
54
55 return cls(
56 id=station_id,
57 name=name,
58 live_stream_url_template=tmpl,
59 hide_from_stations=bool(obj.get("hideFromStations")),
60 timeshift_hls_url_template=ts_hls,
61 timeshift_progressive_url_template=ts_prog,
62 podcasts_available=podcasts_avail,
63 )
64
65
66@dataclass(frozen=True, slots=True)
67class PrivateStation:
68 """Private (non-ORF) radio station definition."""
69
70 id: str
71 name: str
72 streams: tuple[StreamRef, ...] = ()
73 image_urls: tuple[str, ...] = ()
74
75 @classmethod
76 def from_bundle_item(cls, obj: dict[str, Any]) -> PrivateStation | None:
77 """Create a private station from a bundle entry."""
78 sid = obj.get("station")
79 if not isinstance(sid, str) or not sid:
80 return None
81 name = obj.get("name")
82 if not isinstance(name, str) or not name:
83 name = sid
84
85 # streams
86 streams_in = obj.get("streams")
87 streams: list[StreamRef] = []
88 if isinstance(streams_in, list):
89 for s in streams_in:
90 if not isinstance(s, dict):
91 continue
92 url = s.get("url")
93 if not isinstance(url, str) or not url:
94 continue
95 fmt = s.get("format")
96 if not isinstance(fmt, str):
97 fmt = None
98 streams.append(StreamRef(url=url, format=fmt))
99
100 # images (provider only needs URLs; keep it flat)
101 imgs: list[str] = []
102 image = obj.get("image")
103 if isinstance(image, dict) and isinstance(image.get("src"), str):
104 imgs.append(image["src"])
105 image_large = obj.get("imageLarge")
106 if isinstance(image_large, dict):
107 for mode in ("light", "dark"):
108 v = image_large.get(mode)
109 if isinstance(v, dict) and isinstance(v.get("src"), str):
110 imgs.append(v["src"])
111
112 # dedupe while preserving order
113 seen: set[str] = set()
114 deduped = []
115 for u in imgs:
116 if u in seen:
117 continue
118 seen.add(u)
119 deduped.append(u)
120
121 return cls(id=sid, name=name, streams=tuple(streams), image_urls=tuple(deduped))
122
123
124@dataclass(frozen=True, slots=True)
125class PodcastImage:
126 """Holds ORF image versions (path URLs)."""
127
128 versions: dict[str, str]
129
130 @classmethod
131 def from_obj(cls, obj: Any) -> PodcastImage | None:
132 """Create a podcast image from a raw object."""
133 if not isinstance(obj, dict):
134 return None
135 image = obj.get("image")
136 if not isinstance(image, dict):
137 return None
138 versions = image.get("versions")
139 if not isinstance(versions, dict):
140 return None
141 out: dict[str, str] = {}
142 for k, v in versions.items():
143 if not isinstance(v, dict):
144 continue
145 path = v.get("path")
146 if isinstance(path, str) and path:
147 out[str(k)] = path
148 return cls(out) if out else None
149
150 def best(
151 self, preference: Iterable[str] = ("premium", "standard", "id3art", "thumbnail")
152 ) -> str | None:
153 """Return the best matching image URL by preference."""
154 for key in preference:
155 p = self.versions.get(key)
156 if p:
157 return p
158 # fallback: any
159 for p in self.versions.values():
160 if p:
161 return p
162 return None
163
164
165@dataclass(frozen=True, slots=True)
166class OrfPodcast:
167 """ORF podcast metadata."""
168
169 id: int
170 title: str
171 station: str | None = None
172 channel: str | None = None
173 slug: str | None = None
174 description: str | None = None
175 author: str | None = None
176 image: PodcastImage | None = None
177
178 @classmethod
179 def from_index_item(cls, obj: dict[str, Any]) -> OrfPodcast | None:
180 """Create an ORF podcast from an index entry."""
181 pid = obj.get("id")
182 if not isinstance(pid, int):
183 return None
184 title = obj.get("title")
185 if not isinstance(title, str) or not title:
186 title = str(pid)
187
188 station = obj.get("station")
189 if not isinstance(station, str):
190 station = None
191 channel = obj.get("channel")
192 if not isinstance(channel, str):
193 channel = None
194 slug = obj.get("slug")
195 if not isinstance(slug, str):
196 slug = None
197 desc = obj.get("description")
198 if not isinstance(desc, str):
199 desc = None
200 author = obj.get("author")
201 if not isinstance(author, str):
202 author = None
203
204 img = PodcastImage.from_obj(obj)
205
206 return cls(
207 id=pid,
208 title=title,
209 station=station,
210 channel=channel,
211 slug=slug,
212 description=desc,
213 author=author,
214 image=img,
215 )
216
217
218@dataclass(frozen=True, slots=True)
219class Enclosure:
220 """Podcast episode enclosure."""
221
222 url: str
223 mime_type: str | None = None
224 length_bytes: int | None = None
225
226 @classmethod
227 def from_obj(cls, obj: dict[str, Any]) -> Enclosure | None:
228 """Create an enclosure from a raw object."""
229 url = obj.get("url")
230 if not isinstance(url, str) or not url:
231 return None
232 mt = obj.get("type")
233 if not isinstance(mt, str):
234 mt = None
235 ln = obj.get("length")
236 if not isinstance(ln, int):
237 ln = None
238 return cls(url=url, mime_type=mt, length_bytes=ln)
239
240
241@dataclass(frozen=True, slots=True)
242class OrfPodcastEpisode:
243 """ORF podcast episode metadata."""
244
245 guid: str
246 title: str
247 description: str | None = None
248 published: str | None = None # keep as string; provider already formats timestamps itself
249 expiry: str | None = None
250 duration_ms: int | None = None
251 enclosures: tuple[Enclosure, ...] = ()
252 link_url: str | None = None
253 image: PodcastImage | None = None
254
255 @classmethod
256 def from_detail_item(cls, obj: dict[str, Any]) -> OrfPodcastEpisode | None:
257 """Create a podcast episode from a detail entry."""
258 guid = obj.get("guid")
259 if not isinstance(guid, str) or not guid:
260 return None
261 title = obj.get("title")
262 if not isinstance(title, str) or not title:
263 title = guid
264
265 desc = obj.get("description")
266 if not isinstance(desc, str):
267 desc = None
268
269 published = obj.get("published")
270 if not isinstance(published, str):
271 published = None
272 expiry = obj.get("expiry")
273 if not isinstance(expiry, str):
274 expiry = None
275
276 dur = obj.get("duration")
277 if not isinstance(dur, int) or dur <= 0:
278 dur = None
279
280 link = obj.get("url")
281 if not isinstance(link, str):
282 link = None
283
284 enc_in = obj.get("enclosures")
285 encs: list[Enclosure] = []
286 if isinstance(enc_in, list):
287 for e in enc_in:
288 if isinstance(e, dict):
289 enc = Enclosure.from_obj(e)
290 if enc:
291 encs.append(enc)
292
293 img = PodcastImage.from_obj(obj)
294
295 return cls(
296 guid=guid,
297 title=title,
298 description=desc,
299 published=published,
300 expiry=expiry,
301 duration_ms=dur,
302 enclosures=tuple(encs),
303 link_url=link,
304 image=img,
305 )
306
307
308# ----------------------------
309# Parsers
310# ----------------------------
311
312
313def parse_orf_stations(bundle: dict[str, Any], include_hidden: bool) -> list[OrfStation]:
314 """Parse ORF stations from the bundle payload."""
315 stations = bundle.get("stations")
316 if not isinstance(stations, dict):
317 return []
318 out: list[OrfStation] = []
319 for sid, obj in stations.items():
320 if not isinstance(sid, str) or not isinstance(obj, dict):
321 continue
322 st = OrfStation.from_bundle_item(sid, obj)
323 if not st:
324 continue
325 if st.hide_from_stations and not include_hidden:
326 continue
327 out.append(st)
328 return out
329
330
331def parse_private_stations(bundle: dict[str, Any]) -> list[PrivateStation]:
332 """Parse private stations from the bundle payload."""
333 priv = bundle.get("privates")
334 if not isinstance(priv, list):
335 return []
336 out: list[PrivateStation] = []
337 for obj in priv:
338 if not isinstance(obj, dict):
339 continue
340 st = PrivateStation.from_bundle_item(obj)
341 if st:
342 out.append(st)
343 return out
344
345
346def parse_orf_podcasts_index(payload: Any) -> list[OrfPodcast]:
347 """Parse ORF podcast index payload."""
348 # payload is expected to be dict[station_key -> list[podcast_obj]]
349 if not isinstance(payload, dict):
350 return []
351 out: list[OrfPodcast] = []
352 for arr in payload.values():
353 if not isinstance(arr, list):
354 continue
355 for pod in arr:
356 if not isinstance(pod, dict):
357 continue
358 if pod.get("isOnline") is not True:
359 continue
360 item = OrfPodcast.from_index_item(pod)
361 if item:
362 out.append(item)
363 return out
364
365
366def parse_orf_podcast_episodes(payload: Any) -> list[OrfPodcastEpisode]:
367 """Parse podcast episodes from a detail payload."""
368 if not isinstance(payload, dict):
369 return []
370 eps = payload.get("episodes")
371 if not isinstance(eps, list):
372 return []
373 out: list[OrfPodcastEpisode] = []
374 for ep in eps:
375 if not isinstance(ep, dict):
376 continue
377 item = OrfPodcastEpisode.from_detail_item(ep)
378 if item:
379 out.append(item)
380 return out
381