podcast_helper package

Submodules

Module contents

Podcast Helper

Universal audio stream consumer — URL-in → PCM-out.

Accepts local files, direct audio URLs (RSS enclosure MP3 / M4A / Opus / WAV / HLS m3u8), RSS feed URLs (auto-picks the latest episode), and any yt-dlp-supported URL (YouTube, Vimeo, SoundCloud, Twitch VOD / live, …).

Refuses Spotify-DRM and Apple Podcasts catalog URLs with clear hints toward the right workaround (RSS feed).

Main entry points

Multi-surface exposure

The same public functions are surfaced as:

Usage Example

>>> import asyncio
>>> import podcast_helper as ph
>>>
>>> async def main():
...     async for frame in ph.extract_audio_stream(
...         "https://feeds.npr.org/510289/podcast.xml",
...         target_sample_rate=16000,
...         to_mono=True,
...         frame_ms=20,
...     ):
...         # frame["pcm"]: np.float32 (320,) for 20ms @ 16kHz
...         pass
>>> asyncio.run(main())

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

class podcast_helper.Episode[source]

Bases: TypedDict

One podcast episode, normalised across feed flavours.

Keys

guidstr

Stable episode identifier (RSS <guid> or yt-dlp-style fallback). Empty string if the feed didn’t ship one.

titlestr

Episode title.

descriptionstr

Long-form description / show notes. May contain HTML.

linkstr

Episode webpage / show notes URL (NOT the audio URL).

published_atstr

ISO 8601 timestamp (YYYY-MM-DDTHH:MM:SS+00:00 UTC). Empty string if not provided.

duration_secondsint

Reported duration. 0 when missing.

enclosure_urlstr

Direct audio URL (the actual media file). This is what you feed to podcast_helper.extract_audio_stream().

enclosure_typestr

MIME type of the enclosure ("audio/mpeg", "audio/x-m4a", …).

enclosure_size_bytesint

Reported file size of the enclosure. 0 when missing.

image_urlstr

Episode-level cover art URL; falls back to the show’s image.

description: str
duration_seconds: int
enclosure_size_bytes: int
enclosure_type: str
enclosure_url: str
guid: str
image_url: str
published_at: str
title: str
class podcast_helper.PcmFrame[source]

Bases: TypedDict

One PCM frame in absolute stream time.

Keys

t_abs_sfloat

Seconds since the source started. Monotonic — does not reset on reconnect (future v0.3 will pad with silence if reconnect lands).

pcmNDArray[np.float32]

Audio samples in [-1.0, 1.0], float32.

  • Shape (n_samples,) when to_mono=True (default).

  • Shape (n_samples, n_channels) when to_mono=False, with samples interleaved by channel ([L0, R0, L1, R1, ...] for stereo, reshaped to (n_samples, 2)).

voicedbool | None

Always None here — VAD downstream fills it in if used.

pcm: ndarray[tuple[Any, ...], dtype[float32]]
t_abs_s: float
voiced: bool | None
async podcast_helper.extract_audio_stream(url, *, target_sample_rate=16000, to_mono=True, resample_quality='default', realtime=True, frame_ms=20, headers=None, cookies_from_browser=None, speed=1.0, record_to=None)[source]

Yield PCM frames from any audio-bearing URL.

Parameters:

  • url (str) – File path, file:// URL, direct audio URL (RSS enclosure MP3 / M4A / Opus / WAV / HLS m3u8), RSS feed URL (auto-picks latest episode), or any yt-dlp-supported URL (YouTube, Vimeo, SoundCloud, Twitch VOD / live, etc.). Spotify-protected and Apple Podcasts catalog URLs raise NotImplementedError early with hints to the right workaround.

  • target_sample_rate (int, default 16000) – Exact output sample rate in Hz. Conversion uses libswresample (default) or libsoxr (resample_quality="high"); both apply a Shannon-correct anti-aliasing low-pass before decimation.

  • to_mono (bool, default True) –

    • True: ffmpeg standard downmix to one channel; pcm.shape == (n_samples,).

    • False: source’s native channel count preserved; pcm.shape == (n_samples, n_channels).

  • resample_quality ("default" | "high", default "default") – "high" triggers aresample=...:resampler=soxr:precision=28 (~10× slower, inaudible artefacts). Default libswresample is more than enough for ASR / VAD / ML.

  • realtime (bool, default True) – Pace decoding at wall-clock (ffmpeg’s -re). Forced OFF for live streams (they pace themselves; -re would only add latency).

  • frame_ms (int, default 20) – Frame duration in milliseconds. 20 matches Silero VAD’s native frame size, avoiding a downstream re-buffer.

  • headers (dict[str, str], optional) – HTTP headers ffmpeg should send. Merged on top of yt-dlp’s per-source headers (user keys win).

  • cookies_from_browser (str, optional) – "firefox" / "chrome" / "safari" / etc. — passed to yt-dlp’s --cookies-from-browser for age-gated or members-only sources.

  • speed (float, default 1.0) – Playback rate for VOD only. Implemented via ffmpeg’s atempo= filter so the pitch is preserved (no chipmunk effect). 1.0 = unchanged; > 1.0 = faster (e.g. 2.0 for 2× ASR throughput on long episodes); 0 < speed < 1.0 = slower (e.g. 0.5 for half-speed transcription proofing). Outside [0.5, 100] the chain wraps multiple atempo filters whose product equals speed. Raises ``ValueError`` on live streams — you can’t fast-forward past the live edge, and slowing down lets the consumer fall behind unboundedly. Implicitly disables the -re realtime pacing when speed != 1.0.

  • record_to (str, optional) –

    If set, ffmpeg writes a parallel compressed archive of the same audio to this path while the live PCM stream is consumed. Implemented via ffmpeg’s native multi-output: one decode, two encoder paths, no extra subprocess. Codec is picked from the extension:

    • .mp3 → libmp3lame (128 kbps)

    • .m4a / .aac → aac (128 kbps)

    • .opus → libopus (96 kbps)

    • .ogg → libvorbis (quality 5)

    • .flac → flac (lossless)

    • .wav → pcm_s16le

    Unknown extensions raise ValueError. The archive is target_sample_rate / to_mono / speed -coherent with the live PCM (same filter chain). For live sources, the archive grows for the duration of the stream.

Yields:

PcmFrame – Successive frames in absolute stream time.

Raises:

  • NotImplementedError – For Spotify-protected URLs (DRM) and Apple Podcasts catalog URLs (point to the show’s RSS feed instead).

  • RuntimeError – If yt-dlp can’t resolve the URL (private / removed / geo-blocked).

  • FileNotFoundError – If ffmpeg is not on PATH.

  • ValueError – If speed != 1.0 and the source is live, if speed <= 0, or if record_to’s extension is not in the supported set.

Return type:

AsyncIterator[PcmFrame]

Examples

>>> import asyncio, podcast_helper as ph
>>> async def main():
...     async for frame in ph.extract_audio_stream(
...         "https://www.youtube.com/watch?v=YE7VzlLtp-4",
...         target_sample_rate=16000, to_mono=True,
...     ):
...         # frame["pcm"]: np.float32, shape (320,) for 20ms @ 16kHz
...         pass
>>> asyncio.run(main())
podcast_helper.feed(url, *, max_episodes=None)[source]

Fetch an RSS / Atom podcast feed and return its episodes.

Episodes are sorted most recent first by published_at (ISO); episodes without a parseable date sink to the bottom but keep their relative order.

Parameters:

  • url (str) – Feed URL (RSS or Atom).

  • max_episodes (int, optional) – Cap the number of episodes returned. None (default) returns everything the feed advertises.

Returns:

Possibly empty if the feed has no episodes / no audio enclosures.

Return type:

list[Episode]

Raises:

  • requests.HTTPError – If the GET request returns a non-2xx status.

  • RuntimeError – If neither podcastparser nor feedparser could extract any episode (genuinely broken feed).

podcast_helper.latest_episode(url)[source]

Return the most recent episode of a podcast feed with a non-empty audio enclosure URL.

Convenience around feed() for the common “I just want the latest” case. Skips items without an audio enclosure (some feeds publish text-only “trailer” items).

Parameters:

url (str) – Feed URL.

Return type:

Episode

Raises:

  • requests.HTTPError – If the GET request returns a non-2xx status.

  • RuntimeError – If the feed parses but no episode has an audio enclosure.