"""Module de génération du flux RSS pour podcast.""" import os import re import shutil from datetime import datetime from pathlib import Path from xml.etree import ElementTree as ET from xml.dom import minidom import hashlib import config class RSSGenerator: """Génère et maintient le flux RSS du podcast.""" def __init__(self): # Configuration du podcast self.podcast_title = os.getenv("PODCAST_TITLE", "Chronique Finance & Tech") self.podcast_description = os.getenv( "PODCAST_DESCRIPTION", "Ta chronique quotidienne des news finance et tech françaises. " "Chaque jour, on décrypte l'essentiel de l'actualité fintech, banque et innovation." ) self.podcast_author = os.getenv("PODCAST_AUTHOR", "Data Workers") self.podcast_email = os.getenv("PODCAST_EMAIL", "podcast@data-workers.com") self.podcast_website = os.getenv("PODCAST_WEBSITE", "https://data-workers.com") self.podcast_image = os.getenv("PODCAST_IMAGE", "https://data-workers.com/podcast/cover.jpg") self.podcast_language = "fr" self.podcast_category = "Business" self.podcast_subcategory = "Investing" # URL de base pour les fichiers audio (à configurer selon ton hébergement) self.audio_base_url = os.getenv("PODCAST_AUDIO_URL", "https://data-workers.com/podcast/episodes") self.feed_url = os.getenv("PODCAST_FEED_URL", "https://data-workers.com/podcast/feed.xml") # Chemins self.public_dir = config.BASE_DIR / "public" self.episodes_dir = self.public_dir / "episodes" self.feed_path = self.public_dir / "feed.xml" # Créer les dossiers self.public_dir.mkdir(exist_ok=True) self.episodes_dir.mkdir(exist_ok=True) def generate_feed(self) -> Path: """Génère le flux RSS complet à partir des fichiers audio.""" # Collecter tous les épisodes episodes = self._collect_episodes() if not episodes: print("Aucun épisode trouvé") return None # Créer le XML RSS rss = self._create_rss_element() channel = self._create_channel_element(rss) # Ajouter les épisodes (triés par date décroissante) episodes.sort(key=lambda x: x['date'], reverse=True) for episode in episodes: self._add_episode_to_channel(channel, episode) # Écrire le fichier XML xml_string = self._prettify_xml(rss) with open(self.feed_path, 'w', encoding='utf-8') as f: f.write(xml_string) print(f"✓ Flux RSS généré: {self.feed_path}") print(f" {len(episodes)} épisode(s) dans le feed") return self.feed_path def _collect_episodes(self) -> list: """Collecte les informations de tous les épisodes.""" episodes = [] # Scanner le dossier audio audio_dir = config.OUTPUT_AUDIO_DIR for audio_file in audio_dir.glob("*.mp3"): episode = self._parse_episode_from_file(audio_file) if episode: # Copier le fichier dans le dossier public public_audio = self.episodes_dir / audio_file.name if not public_audio.exists() or audio_file.stat().st_mtime > public_audio.stat().st_mtime: shutil.copy2(audio_file, public_audio) print(f" Copié: {audio_file.name}") episodes.append(episode) return episodes def _parse_episode_from_file(self, audio_file: Path) -> dict: """Extrait les métadonnées d'un fichier audio.""" filename = audio_file.stem # Sans extension # Format attendu: 2026-09-08_Briefing_Instant_Payment_Funding_Circle_audio # ou: 2026-09-08_Hebdo_Lemonway_BPCE_Wero_audio # Extraire la date date_match = re.match(r'(\d{4})-(\d{2})-(\d{2})_(.+?)_audio$', filename) if not date_match: date_match = re.match(r'(\d{4})_(\d{2})_(\d{2})_(.+?)_audio$', filename) if not date_match: print(f" ⚠ Format non reconnu: {filename}") return None year, month, day, title_part = date_match.groups() try: episode_date = datetime(int(year), int(month), int(day)) except ValueError: print(f" ⚠ Date invalide: {filename}") return None # Nettoyer le titre title = title_part.replace('_', ' ').replace(' ', ' ').strip() # Supprimer "Briefing" et "Hebdo" du titre title = re.sub(r'^Briefing\s+', '', title) title = re.sub(r'^Hebdo\s+', '', title) # Normaliser : remplacer " x " existants par espace, puis tout séparer par " x " # Ex: "OpenAI Nu Greenly x Normative" -> "OpenAI x Nu x Greenly x Normative" title = re.sub(r'\s+x\s+', ' ', title, flags=re.IGNORECASE) # Enlever les "x" existants words = title.split() title = ' x '.join(words) # Générer un ID unique guid = hashlib.md5(filename.encode()).hexdigest() # Taille du fichier file_size = audio_file.stat().st_size # Durée estimée (approximation: 1 Mo ≈ 1 minute pour MP3 128kbps) duration_seconds = int(file_size / (128 * 1024 / 8)) duration_formatted = f"{duration_seconds // 60}:{duration_seconds % 60:02d}" # Lire le script clean pour la description si disponible description = self._get_episode_description(filename, title, episode_date) return { 'title': f"Chronique du {day}/{month}/{year} - {title}", 'description': description, 'date': episode_date, 'filename': audio_file.name, 'url': f"{self.audio_base_url}/{audio_file.name}", 'guid': guid, 'size': file_size, 'duration': duration_formatted, } def _get_episode_description(self, filename: str, title: str, date: datetime) -> str: """Récupère ou génère la description de l'épisode.""" # Chercher le fichier clean correspondant base_name = filename.replace('_audio', '') for clean_file in config.OUTPUT_CLEAN_DIR.glob(f"{base_name}*.txt"): try: with open(clean_file, 'r', encoding='utf-8') as f: content = f.read() # Extraire les premiers 500 caractères comme description # Nettoyer les balises SSML clean_content = re.sub(r'<[^>]+>', '', content) clean_content = clean_content.replace('\n', ' ').strip() if len(clean_content) > 500: clean_content = clean_content[:497] + "..." return clean_content except Exception: pass # Description par défaut return f"Chronique finance et tech du {date.strftime('%d/%m/%Y')}. Au programme : {title}." def _create_rss_element(self) -> ET.Element: """Crée l'élément RSS racine.""" rss = ET.Element('rss') rss.set('version', '2.0') rss.set('xmlns:itunes', 'http://www.itunes.com/dtds/podcast-1.0.dtd') rss.set('xmlns:content', 'http://purl.org/rss/1.0/modules/content/') rss.set('xmlns:atom', 'http://www.w3.org/2005/Atom') return rss def _create_channel_element(self, rss: ET.Element) -> ET.Element: """Crée l'élément channel avec les métadonnées du podcast.""" channel = ET.SubElement(rss, 'channel') # Métadonnées de base ET.SubElement(channel, 'title').text = self.podcast_title ET.SubElement(channel, 'description').text = self.podcast_description ET.SubElement(channel, 'link').text = self.podcast_website ET.SubElement(channel, 'language').text = self.podcast_language ET.SubElement(channel, 'copyright').text = f"© {datetime.now().year} {self.podcast_author}" ET.SubElement(channel, 'lastBuildDate').text = datetime.now().strftime('%a, %d %b %Y %H:%M:%S +0000') # Atom self link (requis par certaines plateformes) atom_link = ET.SubElement(channel, '{http://www.w3.org/2005/Atom}link') atom_link.set('href', self.feed_url) atom_link.set('rel', 'self') atom_link.set('type', 'application/rss+xml') # iTunes specific ET.SubElement(channel, '{http://www.itunes.com/dtds/podcast-1.0.dtd}author').text = self.podcast_author ET.SubElement(channel, '{http://www.itunes.com/dtds/podcast-1.0.dtd}summary').text = self.podcast_description ET.SubElement(channel, '{http://www.itunes.com/dtds/podcast-1.0.dtd}explicit').text = 'no' # Image image = ET.SubElement(channel, 'image') ET.SubElement(image, 'url').text = self.podcast_image ET.SubElement(image, 'title').text = self.podcast_title ET.SubElement(image, 'link').text = self.podcast_website itunes_image = ET.SubElement(channel, '{http://www.itunes.com/dtds/podcast-1.0.dtd}image') itunes_image.set('href', self.podcast_image) # Catégorie category = ET.SubElement(channel, '{http://www.itunes.com/dtds/podcast-1.0.dtd}category') category.set('text', self.podcast_category) if self.podcast_subcategory: subcategory = ET.SubElement(category, '{http://www.itunes.com/dtds/podcast-1.0.dtd}category') subcategory.set('text', self.podcast_subcategory) # Owner owner = ET.SubElement(channel, '{http://www.itunes.com/dtds/podcast-1.0.dtd}owner') ET.SubElement(owner, '{http://www.itunes.com/dtds/podcast-1.0.dtd}name').text = self.podcast_author ET.SubElement(owner, '{http://www.itunes.com/dtds/podcast-1.0.dtd}email').text = self.podcast_email return channel def _add_episode_to_channel(self, channel: ET.Element, episode: dict) -> None: """Ajoute un épisode au channel.""" item = ET.SubElement(channel, 'item') ET.SubElement(item, 'title').text = episode['title'] ET.SubElement(item, 'description').text = episode['description'] ET.SubElement(item, '{http://purl.org/rss/1.0/modules/content/}encoded').text = f"" ET.SubElement(item, 'pubDate').text = episode['date'].strftime('%a, %d %b %Y 06:00:00 +0000') ET.SubElement(item, 'guid').text = episode['guid'] # Enclosure (le fichier audio) enclosure = ET.SubElement(item, 'enclosure') enclosure.set('url', episode['url']) enclosure.set('length', str(episode['size'])) enclosure.set('type', 'audio/mpeg') # iTunes specific ET.SubElement(item, '{http://www.itunes.com/dtds/podcast-1.0.dtd}duration').text = episode['duration'] ET.SubElement(item, '{http://www.itunes.com/dtds/podcast-1.0.dtd}summary').text = episode['description'] ET.SubElement(item, '{http://www.itunes.com/dtds/podcast-1.0.dtd}explicit').text = 'no' def _prettify_xml(self, element: ET.Element) -> str: """Formate le XML pour qu'il soit lisible.""" rough_string = ET.tostring(element, encoding='unicode') reparsed = minidom.parseString(rough_string) return reparsed.toprettyxml(indent=" ", encoding=None) def update_rss_feed() -> Path: """Fonction utilitaire pour mettre à jour le feed RSS.""" generator = RSSGenerator() return generator.generate_feed() if __name__ == "__main__": # Test standalone feed_path = update_rss_feed() if feed_path: print(f"\nFeed généré: {feed_path}")