@jhustle44 / jcr-guide-pro
Code Commits Issues Pulls Cloud Agent CI/CD
copilot/change-app-name-to-jcrguru
copilot/add-firebase copilot/change-app-name-to-jcrguru copilot/create-zip-file copilot/create-zip-file-again copilot/merge-any-updates main master
md 359 lines 5.99 KB
Rendered Raw
1 # RSS Feed & Batch Processing
2
3 How to parse podcast RSS feeds, download episodes, and process multiple episodes in batch.
4
5 ## RSS Feed Parsing
6
7 Podcast RSS feeds follow the iTunes/Apple Podcasts namespace. Use `feedparser` for reliable parsing.
8
9 ### Setup
10
11 ```bash
12
13 pip install feedparser requests
14
15 ```
16
17 ### Parsing Script
18
19 ```python
20
21 import feedparser
22
23 import requests
24
25 import os
26
27 from datetime import datetime
28
29 def parse_podcast_feed(feed_url: str) -> list[dict]:
30
31 """Parse a podcast RSS feed and return episode metadata."""
32
33 feed = feedparser.parse(feed_url)
34
35 if feed.bozo:
36
37 raise ValueError(f"Failed to parse feed: {feed.bozo_exception}")
38
39 podcast_info = {
40
41 "title": feed.feed.get("title", "Unknown Podcast"),
42
43 "description": feed.feed.get("summary", ""),
44
45 "link": feed.feed.get("link", ""),
46
47 "image": feed.feed.get("image", {}).get("href", ""),
48
49 }
50
51 episodes = []
52
53 for entry in feed.entries:
54
55 audio_url = None
56
57 for link in entry.get("links", []):
58
59 if link.get("type", "").startswith("audio/"):
60
61 audio_url = link["href"]
62
63 break
64
65 if not audio_url:
66
67 for enc in entry.get("enclosures", []):
68
69 if enc.get("type", "").startswith("audio/"):
70
71 audio_url = enc["href"]
72
73 break
74
75 episodes.append({
76
77 "title": entry.get("title", "Untitled"),
78
79 "description": entry.get("summary", ""),
80
81 "published": entry.get("published", ""),
82
83 "duration": entry.get("itunes_duration", ""),
84
85 "audio_url": audio_url,
86
87 "episode_number": entry.get("itunes_episode", ""),
88
89 "season": entry.get("itunes_season", ""),
90
91 "link": entry.get("link", ""),
92
93 "guest": extract_guest_name(entry),
94
95 })
96
97 episodes.sort(
98
99 key=lambda e: datetime.strptime(e["published"], "%a, %d %b %Y %H:%M:%S %z")
100
101 if e["published"]
102
103 else datetime.min.replace(tzinfo=None),
104
105 reverse=True,
106
107 )
108
109 return {"podcast": podcast_info, "episodes": episodes}
110
111 def extract_guest_name(entry: dict) -> str | None:
112
113 """Try to extract guest name from episode metadata.
114
115 Checks itunes:author, dc:creator, and common title patterns like
116
117 'Episode 42: Interview with Jane Smith'.
118
119 """
120
121 author = entry.get("author") or entry.get("itunes_author")
122
123 if author:
124
125 return author
126
127 title = entry.get("title", "")
128
129 patterns = ["with ", "featuring ", "ft. ", "feat. "]
130
131 for p in patterns:
132
133 if p in title.lower():
134
135 idx = title.lower().index(p) + len(p)
136
137 return title[idx:].split(" - ")[0].split(" | ")[0].strip()
138
139 return None
140
141 ```
142
143 ## Episode Downloading
144
145 ```python
146
147 def download_episode(
148
149 audio_url: str,
150
151 output_dir: str,
152
153 filename: str | None = None,
154
155 ) -> str:
156
157 """Download a podcast episode audio file.
158
159 Returns the path to the downloaded file.
160
161 """
162
163 if not audio_url:
164
165 raise ValueError("No audio URL provided")
166
167 os.makedirs(output_dir, exist_ok=True)
168
169 if filename is None:
170
171 filename = audio_url.split("/")[-1].split("?")[0]
172
173 if not filename.endswith((".mp3", ".m4a", ".wav", ".ogg")):
174
175 filename += ".mp3"
176
177 output_path = os.path.join(output_dir, filename)
178
179 response = requests.get(audio_url, stream=True, timeout=60)
180
181 response.raise_for_status()
182
183 with open(output_path, "wb") as f:
184
185 for chunk in response.iter_content(chunk_size=8192):
186
187 f.write(chunk)
188
189 return output_path
190
191 ```
192
193 ## Batch Processing
194
195 Process the last N episodes from a feed through the full content pipeline.
196
197 ### Batch Workflow
198
199 ```python
200
201 def batch_process(
202
203 feed_url: str,
204
205 num_episodes: int = 5,
206
207 output_base_dir: str = "podcast-content",
208
209 generate_audiograms: bool = False,
210
211 ) -> list[dict]:
212
213 """Process the last N episodes from a podcast feed.
214
215 Returns a list of results with paths to generated content.
216
217 """
218
219 feed_data = parse_podcast_feed(feed_url)
220
221 episodes = feed_data["episodes"][:num_episodes]
222
223 results = []
224
225 for ep in episodes:
226
227 slug = slugify(ep["title"])
228
229 episode_dir = os.path.join(output_base_dir, slug)
230
231 os.makedirs(episode_dir, exist_ok=True)
232
233 audio_path = download_episode(
234
235 ep["audio_url"],
236
237 os.path.join(episode_dir, "audio"),
238
239 )
240
241 transcript = transcribe_episode(audio_path)
242
243 result = {
244
245 "episode": ep,
246
247 "transcript": transcript,
248
249 "output_dir": episode_dir,
250
251 "audio_path": audio_path,
252
253 }
254
255 results.append(result)
256
257 return results
258
259 def slugify(text: str) -> str:
260
261 """Convert text to a filesystem-safe slug."""
262
263 import re
264
265 text = text.lower().strip()
266
267 text = re.sub(r"[^\w\s-]", "", text)
268
269 text = re.sub(r"[-\s]+", "-", text)
270
271 return text[:80]
272
273 ```
274
275 ### Batch Output Structure
276
277 When processing multiple episodes, organize output by episode:
278
279 ```text
280
281 podcast-content/
282
283 ├── episode-title-1/
284
285 │ ├── audio/
286
287 │ │ └── episode.mp3
288
289 │ ├── transcript.json
290
291 │ ├── summary.md
292
293 │ ├── show-notes.md
294
295 │ ├── blog-post.md
296
297 │ ├── social/
298
299 │ │ ├── twitter-thread.md
300
301 │ │ ├── linkedin-post.md
302
303 │ │ └── instagram-caption.md
304
305 │ ├── newsletter.md
306
307 │ └── pull-quotes.md
308
309 ├── episode-title-2/
310
311 │ └── ...
312
313 └── batch-summary.md (index of all processed episodes)
314
315 ```
316
317 ### Rate Limiting and Costs
318
319 When batch processing, be mindful of API rate limits and costs:
320
321 - **Whisper API**: ~$0.006 per minute of audio. A 60-minute episode costs ~$0.36.
322 - **Rate limits**: OpenAI's Whisper API allows 50 RPM on most tiers. With chunked transcription (10-min chunks), a 60-min episode uses 6 requests.
323
324 - **Download bandwidth**: Podcast audio files are typically 30-80 MB per hour. Batch processing 10 episodes could download 300-800 MB.
325 - **Recommendation**: For large batches (>10 episodes), add a 2-second delay between episodes to avoid rate limits. Provide progress updates to the user during processing.
326
327 ### Resumability
328
329 For large batch jobs, track progress to allow resuming if interrupted:
330
331 ```python
332
333 import json
334
335 def save_progress(output_dir: str, completed: list[str]):
336
337 """Save batch processing progress."""
338
339 progress_file = os.path.join(output_dir, ".batch-progress.json")
340
341 with open(progress_file, "w") as f:
342
343 json.dump({"completed": completed}, f)
344
345 def load_progress(output_dir: str) -> list[str]:
346
347 """Load batch processing progress."""
348
349 progress_file = os.path.join(output_dir, ".batch-progress.json")
350
351 if os.path.exists(progress_file):
352
353 with open(progress_file) as f:
354
355 return json.load(f).get("completed", [])
356
357 return []
358
359 ```