#!/usr/bin/env python3 """ Rebuild a podcast feed.xml by scanning a show directory. Usage: python3 rebuild_feed.py [show_dir] show_dir defaults to the directory containing this script (main show). Pass a subdirectory path for other shows, e.g.: python3 rebuild_feed.py /home/node/.openclaw/workspace/podcast/spanish Each show directory must contain: - show.json — show metadata - episodes/ — folder with *.mp3 files - episodes/*.json — sidecar per episode (same stem as MP3) Sidecar format (episode_num is IGNORED — assigned at rebuild time): { "title": "Episode Title", "description": "Short description.", "topic": "original topic", "date": "2026-03-05", "duration": "00:22:13", "size_bytes": 12345678 } Episode numbering: purely positional by (date, filename) ascending — ep1 = oldest. GUIDs: stable, based on mp3 filename — never change on rebuild. """ import json import hashlib import subprocess import sys import xml.etree.ElementTree as ET from pathlib import Path from datetime import datetime, timezone SCRIPT_DIR = Path(__file__).parent ns_itunes = "http://www.itunes.com/dtds/podcast-1.0.dtd" def get_audio_duration(mp3_path: Path) -> int: result = subprocess.run( ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", str(mp3_path)], capture_output=True, text=True ) if result.returncode == 0: try: data = json.loads(result.stdout) return int(float(data["format"].get("duration", 0))) except Exception: pass return 0 def format_duration(seconds: int) -> str: h = seconds // 3600 m = (seconds % 3600) // 60 s = seconds % 60 return f"{h:02d}:{m:02d}:{s:02d}" def rfc2822_date(date_str: str, mtime: float = None) -> str: """Convert date to RFC 2822. Uses mtime for unique timestamps (avoids same-day collisions).""" if mtime: dt = datetime.fromtimestamp(mtime, tz=timezone.utc) else: dt = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc) return dt.strftime("%a, %d %b %Y %H:%M:%S +0000") def stable_guid(mp3_filename: str) -> str: """Stable GUID based on mp3 filename — never changes on rebuild.""" return "clawd-" + hashlib.md5(mp3_filename.encode()).hexdigest() def build_feed(show: dict, episodes: list) -> ET.ElementTree: """ episodes: list of (mp3_path, meta, episode_num) sorted newest-first for the feed. episode_num is assigned by caller based on chronological (oldest=1) position. """ ET.register_namespace("itunes", ns_itunes) ET.register_namespace("content", "http://purl.org/rss/1.0/modules/content/") base_url = show["base_url"] cover_url = base_url + show["cover_path"] feed_url = base_url + show["feed_path"] ep_base = base_url + show["episodes_path"] rss = ET.Element("rss", {"version": "2.0"}) ch = ET.SubElement(rss, "channel") # Standard RSS ET.SubElement(ch, "title").text = show["name"] ET.SubElement(ch, "link").text = feed_url ET.SubElement(ch, "description").text = show["description"] ET.SubElement(ch, "language").text = show["language"] ET.SubElement(ch, "copyright").text = "© 2026 clawd-bot" ET.SubElement(ch, "lastBuildDate").text = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S +0000") ET.SubElement(ch, "ttl").text = "15" # RSS image block img_rss = ET.SubElement(ch, "image") ET.SubElement(img_rss, "url").text = cover_url ET.SubElement(img_rss, "title").text = show["name"] ET.SubElement(img_rss, "link").text = feed_url # iTunes channel metadata ET.SubElement(ch, f"{{{ns_itunes}}}title").text = show["name"] ET.SubElement(ch, f"{{{ns_itunes}}}author").text = show["author"] ET.SubElement(ch, f"{{{ns_itunes}}}explicit").text = "false" ET.SubElement(ch, f"{{{ns_itunes}}}type").text = "episodic" ET.SubElement(ch, f"{{{ns_itunes}}}summary").text = show["description"] img_it = ET.SubElement(ch, f"{{{ns_itunes}}}image") img_it.set("href", cover_url) cat = ET.SubElement(ch, f"{{{ns_itunes}}}category") cat.set("text", show["category"]) if "subcategory" in show: sub = ET.SubElement(cat, f"{{{ns_itunes}}}category") sub.set("text", show["subcategory"]) owner = ET.SubElement(ch, f"{{{ns_itunes}}}owner") ET.SubElement(owner, f"{{{ns_itunes}}}name").text = show["author"] ET.SubElement(owner, f"{{{ns_itunes}}}email").text = show["email"] # Episodes (newest first in feed) for mp3_path, meta, ep_num in episodes: mp3_url = ep_base + mp3_path.name file_size = mp3_path.stat().st_size # Use sidecar duration if available, else probe with ffprobe if "duration" in meta: # Parse "HH:MM:SS" back to seconds for potential re-formatting dur_str = meta["duration"] else: dur_sec = get_audio_duration(mp3_path) dur_str = format_duration(dur_sec) item = ET.SubElement(ch, "item") ET.SubElement(item, "title").text = meta["title"] ET.SubElement(item, "description").text = meta["description"] ET.SubElement(item, "pubDate").text = rfc2822_date(meta["date"], mp3_path.stat().st_mtime) ET.SubElement(item, "link").text = mp3_url guid = ET.SubElement(item, "guid") guid.text = stable_guid(mp3_path.name) guid.set("isPermaLink", "false") enc = ET.SubElement(item, "enclosure") enc.set("url", mp3_url) enc.set("type", "audio/mpeg") enc.set("length", str(file_size)) ET.SubElement(item, f"{{{ns_itunes}}}title").text = meta["title"] ET.SubElement(item, f"{{{ns_itunes}}}author").text = show["author"] ET.SubElement(item, f"{{{ns_itunes}}}summary").text = meta["description"] ET.SubElement(item, f"{{{ns_itunes}}}explicit").text = "false" ET.SubElement(item, f"{{{ns_itunes}}}duration").text = dur_str ET.SubElement(item, f"{{{ns_itunes}}}episode").text = str(ep_num) ET.SubElement(item, f"{{{ns_itunes}}}episodeType").text = "full" ep_img = ET.SubElement(item, f"{{{ns_itunes}}}image") ep_img.set("href", cover_url) return ET.ElementTree(rss) def main(): # Determine show directory if len(sys.argv) > 1: show_dir = Path(sys.argv[1]) if not show_dir.is_absolute(): show_dir = SCRIPT_DIR / show_dir else: show_dir = SCRIPT_DIR config_path = show_dir / "show.json" if not config_path.exists(): print(f"❌ No show.json in {show_dir}") sys.exit(1) with open(config_path) as f: show = json.load(f) episodes_dir = show_dir / "episodes" if not episodes_dir.exists(): print(f"❌ No episodes/ directory in {show_dir}") sys.exit(1) mp3_files = sorted(episodes_dir.glob("*.mp3")) episodes = [] for mp3 in mp3_files: sidecar = mp3.with_suffix(".json") if not sidecar.exists(): print(f" ⚠️ No sidecar for {mp3.name} — skipping") continue with open(sidecar) as f: meta = json.load(f) episodes.append((mp3, meta)) print(f" ✅ {mp3.name}") # Sort chronologically ascending by mtime — oldest first # Using mtime (not JSON date field) so feed order always matches pubDate order episodes.sort(key=lambda x: x[0].stat().st_mtime) # Assign episode numbers 1..N (oldest=1, newest=N) — positional, no counter file # Build feed list newest-first (reverse for display) numbered_asc = [(mp3, meta, i + 1) for i, (mp3, meta) in enumerate(episodes)] numbered_feed = list(reversed(numbered_asc)) # newest first for RSS tree = build_feed(show, numbered_feed) ET.indent(tree, space=" ") feed_path = show_dir / "feed.xml" tree.write(feed_path, encoding="unicode", xml_declaration=True) print(f"\n✅ Feed: {feed_path}") print(f" Show: {show['name']}") print(f" Episodes: {len(numbered_asc)}") for mp3, meta, ep_num in numbered_feed: print(f" ep{ep_num}: {meta['title']}") if __name__ == "__main__": main()