Create podcast.py

src/podcast.py

@@ -0,0 +1,101 @@
+# podcast.py
+import requests
+import feedparser
+import yt_dlp
+import tempfile
+import os
+
+def search_podcast_series(query: str):
+    base_url = "https://itunes.apple.com/search"
+    params = {"term": query, "media": "podcast", "entity": "podcast", "limit": 10}
+    try:
+        response = requests.get(base_url, params=params, timeout=15)
+        response.raise_for_status()
+        data = response.json()
+        series_list = []
+        for series in data.get("results", []):
+            series_list.append({
+                "id": series.get("collectionId"),
+                "title": series.get("collectionName", "Untitled"),
+                "artist": series.get("artistName", "Unknown"),
+                "feed_url": series.get("feedUrl", ""),
+                "thumbnail": series.get("artworkUrl600", ""),
+                "episode_count": series.get("trackCount", 0)
+            })
+        return series_list
+    except Exception as e:
+        print(f"Series search error: {str(e)}")
+        return []
+
+def fetch_episodes(feed_url: str):
+    try:
+        feed = feedparser.parse(feed_url)
+        episodes = []
+        for entry in feed.entries:
+            audio_url = None
+            for link in entry.get('links', []):
+                if link.get('type', '').startswith('audio/'):
+                    audio_url = link.href
+                    break
+            if not audio_url:
+                continue
+            episodes.append({
+                "title": entry.get("title", "Untitled Episode"),
+                "published": entry.get("published", ""),
+                "audio_url": audio_url,
+                "description": entry.get("description", ""),
+                "duration": entry.get("itunes_duration", "0")
+            })
+        return episodes
+    except Exception as e:
+        print(f"RSS parse error: {str(e)}")
+        return []
+
+def download_podcast_audio(audio_url: str, title: str, status: str) -> tuple:
+    if not audio_url:
+        return None, f"{status}\n❌ No audio URL available"
+    new_status = status + f"\n⏬ Downloading: {title}"
+    temp_dir = tempfile.mkdtemp()
+    ydl_opts = {
+        "format": "bestaudio/best",
+        "outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"),
+        "postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "mp3"}],
+        "quiet": True,
+        "socket_timeout": 60,
+        "retries": 5
+    }
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([audio_url])
+            mp3_files = [f for f in os.listdir(temp_dir) if f.endswith('.mp3')]
+            if mp3_files:
+                audio_path = os.path.join(temp_dir, mp3_files[0])
+                new_status += f"\n✅ Downloaded: {mp3_files[0]}"
+                return audio_path, new_status
+        raise FileNotFoundError("MP3 file not found")
+    except Exception as e:
+        new_status += f"\n❌ Download error: {str(e)}"
+        return None, new_status
+
+def fetch_audio(youtube_url: str, status: str) -> tuple:
+    try:
+        new_status = status + "\n⏬ Downloading audio from URL..."
+        temp_dir = tempfile.mkdtemp()
+        filename = f"yt_audio_{os.urandom(8).hex()}"
+        temp_path = os.path.join(temp_dir, filename)
+        ydl_opts = {
+            'format': 'bestaudio/best',
+            'outtmpl': temp_path,
+            'postprocessors': [{'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3'}],
+            'quiet': True,
+        }
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([youtube_url])
+        audio_path = temp_path + '.mp3'
+        if not os.path.exists(audio_path) or os.path.getsize(audio_path) == 0:
+            raise ValueError("Downloaded file is empty")
+        new_status += "\n✅ Downloaded audio"
+        return audio_path, new_status
+    except Exception as e:
+        new_status += f"\n❌ Error: {str(e)}"
+        return None, new_status