#!/usr/bin/env python3 """ Simple Podcast Server Automatically generates and serves podcast RSS feeds from a folder of audio files. """ import os import sys import time import yaml import mimetypes from datetime import datetime from pathlib import Path from http.server import HTTPServer, SimpleHTTPRequestHandler from threading import Thread from urllib.parse import quote from podgen import Podcast, Episode, Media, Category, Person from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from mutagen import File as MutagenFile from mutagen.easyid3 import EasyID3 from mutagen.mp4 import MP4 class PodcastConfig: """Load and validate podcast configuration""" def __init__(self, config_path="config.yaml"): with open(config_path, 'r') as f: self.config = yaml.safe_load(f) self.podcast = self.config['podcast'] self.server = self.config['server'] def get_podcast_info(self): return self.podcast def get_server_info(self): return self.server class AudioMetadataExtractor: """Extract metadata from audio files""" SUPPORTED_FORMATS = {'.mp3', '.m4a', '.mp4', '.aac', '.ogg', '.wav'} @staticmethod def is_audio_file(filepath): """Check if file is a supported audio format""" return Path(filepath).suffix.lower() in AudioMetadataExtractor.SUPPORTED_FORMATS @staticmethod def get_metadata(filepath): """Extract metadata from audio file""" try: audio = MutagenFile(filepath, easy=True) if audio is None: print(f"Warning: Could not read metadata from {filepath}") return None # Get file stats file_stat = os.stat(filepath) file_size = file_stat.st_size modified_time = datetime.fromtimestamp(file_stat.st_mtime) # Extract common metadata metadata = { 'filepath': filepath, 'filename': os.path.basename(filepath), 'file_size': file_size, 'modified_time': modified_time, 'duration': int(audio.info.length) if hasattr(audio, 'info') else 0, } # Try to get title from tags if hasattr(audio, 'tags') and audio.tags: # Handle different tag formats if isinstance(audio.tags, dict): metadata['title'] = audio.tags.get('title', [metadata['filename']])[0] metadata['artist'] = audio.tags.get('artist', ['Unknown'])[0] metadata['album'] = audio.tags.get('album', [''])[0] else: metadata['title'] = str(audio.tags.get('title', metadata['filename'])) metadata['artist'] = str(audio.tags.get('artist', 'Unknown')) metadata['album'] = str(audio.tags.get('album', '')) else: # No tags, use filename metadata['title'] = os.path.splitext(metadata['filename'])[0] metadata['artist'] = 'Unknown' metadata['album'] = '' return metadata except Exception as e: print(f"Error extracting metadata from {filepath}: {e}") return None class PodcastFeedGenerator: """Generate podcast RSS feed using PodGen""" def __init__(self, config, podcasts_dir="podcasts", output_file="feed.xml"): self.config = config self.podcasts_dir = Path(podcasts_dir) self.output_file = output_file self.podcast_info = config.get_podcast_info() self.server_info = config.get_server_info() def generate_feed(self): """Generate RSS feed from audio files in podcasts directory""" print(f"Generating podcast feed from {self.podcasts_dir}...") # Create podcast object p = Podcast() p.name = self.podcast_info['name'] p.description = self.podcast_info['description'] p.website = self.podcast_info['website'] p.image = self.podcast_info.get('image', '') p.language = self.podcast_info.get('language', 'en') p.explicit = self.podcast_info.get('explicit', False) # Set category (using Category object) category_name = self.podcast_info.get('category', 'Technology') subcategory_name = self.podcast_info.get('subcategory') p.category = Category(category_name, subcategory_name) if subcategory_name else Category(category_name) # Set authors and owner (using Person objects) author_person = Person(self.podcast_info['author'], self.podcast_info['email']) p.authors = [author_person] p.owner = author_person # Optional fields if self.podcast_info.get('copyright'): p.copyright = self.podcast_info['copyright'] if self.podcast_info.get('subtitle'): p.subtitle = self.podcast_info['subtitle'] # Find all audio files audio_files = [] for file in self.podcasts_dir.iterdir(): if file.is_file() and AudioMetadataExtractor.is_audio_file(str(file)): metadata = AudioMetadataExtractor.get_metadata(str(file)) if metadata: audio_files.append(metadata) # Sort by modified time (newest first) audio_files.sort(key=lambda x: x['modified_time'], reverse=True) print(f"Found {len(audio_files)} audio file(s)") # Add episodes for idx, metadata in enumerate(audio_files, 1): episode = Episode() episode.title = metadata['title'] episode.summary = f"Episode: {metadata['title']}" # Set publication date to file modification time episode.publication_date = metadata['modified_time'] # Create media URL filename_encoded = quote(metadata['filename']) media_url = f"{self.server_info['base_url']}/podcasts/{filename_encoded}" # Determine MIME type mime_type, _ = mimetypes.guess_type(metadata['filename']) if mime_type is None: mime_type = 'audio/mpeg' # Default to MP3 # Create media object episode.media = Media( url=media_url, size=metadata['file_size'], type=mime_type, duration=metadata['duration'] ) p.episodes.append(episode) print(f" Added: {metadata['title']}") # Generate RSS feed rss_string = p.rss_str() # Write to file with open(self.output_file, 'w', encoding='utf-8') as f: f.write(rss_string) print(f"Feed generated: {self.output_file}") return True class PodcastFileHandler(FileSystemEventHandler): """Handle file system events for podcast directory""" def __init__(self, feed_generator): self.feed_generator = feed_generator self.last_update = 0 self.update_delay = 2 # seconds to wait before regenerating def on_any_event(self, event): """Regenerate feed when files change""" if event.is_directory: return # Check if it's an audio file if not AudioMetadataExtractor.is_audio_file(event.src_path): return # Debounce: only regenerate if enough time has passed current_time = time.time() if current_time - self.last_update < self.update_delay: return self.last_update = current_time print(f"\nFile change detected: {event.src_path}") print(f"Event type: {event.event_type}") # Wait a bit for file operations to complete time.sleep(0.5) # Regenerate feed try: self.feed_generator.generate_feed() except Exception as e: print(f"Error regenerating feed: {e}") class PodcastHTTPHandler(SimpleHTTPRequestHandler): """Custom HTTP handler for serving podcast files""" def log_message(self, format, *args): """Custom logging""" print(f"[{self.address_string()}] {format % args}") def end_headers(self): """Add CORS headers""" self.send_header('Access-Control-Allow-Origin', '*') SimpleHTTPRequestHandler.end_headers(self) class PodcastServer: """Main podcast server""" def __init__(self, config_path="config.yaml"): self.config = PodcastConfig(config_path) self.feed_generator = PodcastFeedGenerator(self.config) self.observer = None self.http_server = None self.server_thread = None def start_file_watcher(self): """Start watching podcast directory for changes""" event_handler = PodcastFileHandler(self.feed_generator) self.observer = Observer() self.observer.schedule(event_handler, "podcasts", recursive=False) self.observer.start() print("File watcher started") def start_http_server(self): """Start HTTP server to serve feed and audio files""" server_info = self.config.get_server_info() host = server_info['host'] port = server_info['port'] # Change to the podcast directory root os.chdir(Path(__file__).parent) self.http_server = HTTPServer((host, port), PodcastHTTPHandler) # Run server in separate thread self.server_thread = Thread(target=self.http_server.serve_forever, daemon=True) self.server_thread.start() print(f"\nPodcast server running on http://{host}:{port}") print(f"RSS Feed URL: http://{host}:{port}/feed.xml") print(f"\nAdd this URL to your podcast client to subscribe:") print(f" {server_info['base_url']}/feed.xml") print(f"\nPress Ctrl+C to stop the server\n") def run(self): """Run the podcast server""" print("=" * 60) print("Simple Podcast Server") print("=" * 60) # Generate initial feed print("\nGenerating initial feed...") self.feed_generator.generate_feed() # Start file watcher print("\nStarting file watcher...") self.start_file_watcher() # Start HTTP server print("\nStarting HTTP server...") self.start_http_server() # Keep running try: while True: time.sleep(1) except KeyboardInterrupt: print("\n\nShutting down...") if self.observer: self.observer.stop() self.observer.join() if self.http_server: self.http_server.shutdown() print("Server stopped") def main(): """Main entry point""" if not os.path.exists("config.yaml"): print("Error: config.yaml not found!") print("Please create a config.yaml file with your podcast settings.") sys.exit(1) server = PodcastServer() server.run() if __name__ == "__main__": main()