n3wk
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
"""Parse The Dick Show RSS feed into data/episodes.jsonl (one episode per line)."""
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import requests
|
|
|
|
from config import FEED_URL, EPISODES_JSONL, DOWNLOAD_TIMEOUT, MAX_FEED_RETRIES
|
|
|
|
NS = {
|
|
"itunes": "http://www.itunes.com/dtds/podcast-1.0.dtd",
|
|
}
|
|
EP_NUM_RE = re.compile(r"Episode\s+(\d+)", re.IGNORECASE)
|
|
|
|
|
|
def fetch_feed(url: str = FEED_URL) -> bytes:
|
|
last_err = None
|
|
for attempt in range(1, MAX_FEED_RETRIES + 1):
|
|
try:
|
|
r = requests.get(url, timeout=DOWNLOAD_TIMEOUT,
|
|
headers={"User-Agent": "dickshow-indexer/1.0"})
|
|
r.raise_for_status()
|
|
return r.content
|
|
except Exception as e: # noqa: BLE001
|
|
last_err = e
|
|
print(f"feed attempt {attempt}/{MAX_FEED_RETRIES} failed: {e}",
|
|
file=sys.stderr)
|
|
time.sleep(3 * attempt)
|
|
raise SystemExit(f"could not download feed: {last_err}")
|
|
|
|
|
|
def itunes_duration_seconds(text: str | None) -> int:
|
|
"""Libsyn uses either HH:MM:SS or plain seconds."""
|
|
if not text:
|
|
return 0
|
|
text = text.strip()
|
|
if ":" in text:
|
|
parts = [int(p) for p in text.split(":")]
|
|
while len(parts) < 3:
|
|
parts.insert(0, 0)
|
|
h, m, s = parts
|
|
return h * 3600 + m * 60 + s
|
|
try:
|
|
return int(float(text))
|
|
except ValueError:
|
|
return 0
|
|
|
|
|
|
def parse_feed(xml_bytes: bytes) -> list[dict]:
|
|
root = ET.fromstring(xml_bytes)
|
|
channel = root.find("channel")
|
|
episodes = []
|
|
for item in channel.findall("item"):
|
|
title_el = item.find("title")
|
|
title = (title_el.text or "").strip() if title_el is not None else ""
|
|
m = EP_NUM_RE.search(title)
|
|
ep_num = int(m.group(1)) if m else None
|
|
|
|
guid_el = item.find("guid")
|
|
guid = (guid_el.text or "").strip() if guid_el is not None else ""
|
|
if not guid:
|
|
continue # unusable without stable id
|
|
|
|
pub_el = item.find("pubDate")
|
|
pub_date = (pub_el.text or "").strip() if pub_el is not None else ""
|
|
|
|
enc = item.find("enclosure")
|
|
audio_url, audio_type, audio_len = "", "", 0
|
|
if enc is not None:
|
|
audio_url = enc.get("url", "").strip()
|
|
audio_type = enc.get("type", "")
|
|
try:
|
|
audio_len = int(enc.get("length", "0") or 0)
|
|
except ValueError:
|
|
audio_len = 0
|
|
|
|
dur_el = item.find("itunes:duration", NS)
|
|
duration_s = itunes_duration_seconds(dur_el.text if dur_el is not None else None)
|
|
|
|
link_el = item.find("link")
|
|
page_link = (link_el.text or "").strip() if link_el is not None else ""
|
|
|
|
episodes.append({
|
|
"ep_num": ep_num,
|
|
"title": title,
|
|
"pub_date": pub_date,
|
|
"guid": guid,
|
|
"audio_url": audio_url,
|
|
"audio_type": audio_type,
|
|
"audio_bytes": audio_len,
|
|
"duration_sec": duration_s,
|
|
"page_link": page_link,
|
|
})
|
|
# newest first from feed; store oldest-first for chronological processing
|
|
episodes.reverse()
|
|
return episodes
|
|
|
|
|
|
def write_episodes(episodes: list[dict]) -> None:
|
|
with open(EPISODES_JSONL, "w", encoding="utf-8") as f:
|
|
for ep in episodes:
|
|
f.write(json.dumps(ep, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def load_episodes() -> list[dict]:
|
|
eps = []
|
|
with open(EPISODES_JSONL, encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
eps.append(json.loads(line))
|
|
return eps
|
|
|
|
|
|
def main() -> None:
|
|
xml = fetch_feed()
|
|
episodes = parse_feed(xml)
|
|
write_episodes(episodes)
|
|
numbered = [e["ep_num"] for e in episodes if e["ep_num"]]
|
|
print(f"parsed {len(episodes)} episodes "
|
|
f"(numbered {min(numbered)}-{max(numbered)}) -> {EPISODES_JSONL}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|