from __future__ import annotations import json import os import shutil import subprocess import sys import urllib.parse import xml.etree.ElementTree as ET from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime, timedelta from email.utils import format_datetime from pathlib import Path from clean_podcast_ads import PodcastCleanerError, clean_podcast, probe_audio_metadata from fetch_latest_long_episode import ( FeedCategory, FeedSnapshot, PodcastFetchError, RSS_URL, download_file, fetch_rss, find_latest_episode_over_2h, parse_pub_date, parse_feed_snapshot, sanitize_filename, ) ROOT_DIR = Path(__file__).resolve().parent LEGACY_PUBLIC_BASE_URL = "http://rss.mongueulin.dns.army" PUBLIC_BASE_URL = "https://rss.mongueulin.dns.army" FEED_OUTPUT_PATH = ROOT_DIR / "podcast.xml" FEED_URL = f"{PUBLIC_BASE_URL}/podcast.xml" CACHE_BUST_FEED_OUTPUT_PATH = ROOT_DIR / "podcast-v2.xml" CACHE_BUST_FEED_URL = f"{PUBLIC_BASE_URL}/podcast-v2.xml" PRIMARY_GUID_SUFFIX = "-sans-pub" CACHE_BUST_GUID_SUFFIX = "-sans-pub-v2" EPISODES_DIR = ROOT_DIR / "episodes" ORIGINALS_DIR = ROOT_DIR / "originals" REPORTS_DIR = ROOT_DIR / "reports" STATE_DIR = ROOT_DIR / "state" STATE_FILE = STATE_DIR / "processed_episodes.json" LOCK_FILE = STATE_DIR / "pipeline.lock" LOCK_STALE_AFTER = timedelta(hours=2) ORIGINAL_RETENTION = timedelta(days=3) PART_RETENTION = timedelta(hours=2) CHANNEL_TITLE = "Les Grandes Gueules sans pub" DESCRIPTION_NOTE = "

Version nettoyee et republiee sans coupures publicitaires.

" SCHEDULE_TASK_NAME = "clean_gg_rss_publish_feed" ATOM_NS = "http://www.w3.org/2005/Atom" ITUNES_NS = "http://www.itunes.com/dtds/podcast-1.0.dtd" CONTENT_NS = "http://purl.org/rss/1.0/modules/content/" class PublishFeedError(RuntimeError): """Raised when the pipeline cannot publish the cleaned RSS feed.""" @dataclass(frozen=True) class LockStatus: is_locked: bool can_replace: bool reason: str def now_utc() -> datetime: return datetime.now(UTC) def ensure_directories() -> None: for directory in (EPISODES_DIR, ORIGINALS_DIR, REPORTS_DIR, STATE_DIR): directory.mkdir(parents=True, exist_ok=True) def load_state() -> dict: if not STATE_FILE.exists(): return {"version": 1, "channel": None, "episodes": []} state = json.loads(STATE_FILE.read_text(encoding="utf-8")) state["episodes"] = [normalize_episode_record(record) for record in state.get("episodes", [])] return state def save_state(state: dict) -> None: STATE_DIR.mkdir(parents=True, exist_ok=True) STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def process_exists(pid: int) -> bool: if pid <= 0: return False try: completed = subprocess.run( ["tasklist", "/FI", f"PID eq {pid}"], capture_output=True, text=True, check=True, ) except (FileNotFoundError, subprocess.CalledProcessError): return False return str(pid) in completed.stdout def evaluate_lock(lock_payload: dict, age_seconds: float, pid_exists: bool, stale_after_seconds: float) -> LockStatus: if not pid_exists: return LockStatus(is_locked=False, can_replace=True, reason="missing_owner") if age_seconds > stale_after_seconds: return LockStatus(is_locked=True, can_replace=False, reason="active_but_old") return LockStatus(is_locked=True, can_replace=False, reason="active") @contextmanager def pipeline_lock(): ensure_directories() if LOCK_FILE.exists(): try: payload = json.loads(LOCK_FILE.read_text(encoding="utf-8")) except json.JSONDecodeError: payload = {} age_seconds = max(0.0, now_utc().timestamp() - LOCK_FILE.stat().st_mtime) pid = int(payload.get("pid", 0) or 0) status = evaluate_lock(payload, age_seconds, process_exists(pid), LOCK_STALE_AFTER.total_seconds()) if status.is_locked and not status.can_replace: raise PublishFeedError(f"Another pipeline run is still active ({status.reason}).") LOCK_FILE.write_text( json.dumps( { "pid": os.getpid(), "started_at": now_utc().isoformat(), "cwd": str(ROOT_DIR), }, indent=2, ) + "\n", encoding="utf-8", ) try: yield finally: try: payload = json.loads(LOCK_FILE.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): payload = {} if int(payload.get("pid", 0) or 0) == os.getpid() and LOCK_FILE.exists(): LOCK_FILE.unlink() def audio_stem_for_title(title: str) -> str: return Path(sanitize_filename(title, extension=".mp3")).stem def build_storage_paths(title: str) -> dict[str, Path]: stem = audio_stem_for_title(title) return { "original": ORIGINALS_DIR / f"{stem}.mp3", "cleaned": EPISODES_DIR / f"{stem}.sans_pub.mp3", "report": REPORTS_DIR / f"{stem}.pubs_audacity.txt", "root_original": ROOT_DIR / f"{stem}.mp3", "root_cleaned": ROOT_DIR / f"{stem}.sans_pub.mp3", "root_report": ROOT_DIR / f"{stem}.pubs_audacity.txt", } def build_public_mp3_url(filename: str) -> str: return f"{PUBLIC_BASE_URL}/episodes/{urllib.parse.quote(filename)}" def canonical_public_mp3_url(record: dict) -> str: cleaned_path_value = record.get("cleaned_mp3_path") if cleaned_path_value: return build_public_mp3_url(Path(cleaned_path_value).name) public_mp3_url = record.get("public_mp3_url", "") if public_mp3_url: parsed = urllib.parse.urlsplit(public_mp3_url) path = parsed.path or f"/episodes/{urllib.parse.quote(Path(public_mp3_url).name)}" return urllib.parse.urlunsplit(("https", "rss.mongueulin.dns.army", path, "", "")) raise PublishFeedError("Episode record is missing both cleaned_mp3_path and public_mp3_url.") def normalize_episode_record(record: dict) -> dict: normalized_record = dict(record) normalized_record["public_mp3_url"] = canonical_public_mp3_url(normalized_record) return normalized_record def format_duration(seconds: float) -> str: total_seconds = int(round(seconds)) hours, remainder = divmod(total_seconds, 3600) minutes, seconds_remainder = divmod(remainder, 60) return f"{hours:02d}:{minutes:02d}:{seconds_remainder:02d}" def serialize_categories(categories: tuple[FeedCategory, ...]) -> list[dict[str, str | None]]: return [{"primary": category.primary, "secondary": category.secondary} for category in categories] def deserialize_categories(raw_categories: list[dict[str, str | None]]) -> tuple[FeedCategory, ...]: return tuple(FeedCategory(primary=item["primary"], secondary=item.get("secondary")) for item in raw_categories) def snapshot_channel_to_state(channel) -> dict: return { "title": channel.title, "description": channel.description, "link": channel.link, "language": channel.language, "copyright": channel.copyright, "image_url": channel.image_url, "author": channel.author, "summary": channel.summary, "explicit": channel.explicit, "owner_name": channel.owner_name, "owner_email": channel.owner_email, "categories": serialize_categories(channel.categories), "source_rss_url": RSS_URL, } def find_record(state: dict, source_guid: str) -> dict | None: for record in state["episodes"]: if record["source_guid"] == source_guid: return record return None def upsert_episode_record(state: dict, record: dict) -> None: existing = find_record(state, record["source_guid"]) if existing is None: state["episodes"].append(record) return index = state["episodes"].index(existing) state["episodes"][index] = record def episode_is_ready(record: dict | None) -> bool: if not record: return False cleaned_path = Path(record["cleaned_mp3_path"]) return cleaned_path.exists() def ensure_file_in_place(source_path: Path | None, destination_path: Path) -> Path | None: if source_path is None or not source_path.exists(): return None if source_path.resolve() == destination_path.resolve(): return destination_path destination_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_path, destination_path) return destination_path def build_episode_record(source_episode, channel_image_url: str, original_path: Path | None, cleaned_path: Path, report_path: Path) -> dict: output_metadata = probe_audio_metadata(cleaned_path) record = { "source_guid": source_episode.guid, "title": source_episode.title, "description": source_episode.description, "pub_date": source_episode.pub_date_text, "duration_source": source_episode.duration_text, "source_mp3_url": source_episode.mp3_url, "channel_image_url": channel_image_url, "original_mp3_path": str(original_path) if original_path else None, "cleaned_mp3_path": str(cleaned_path), "audacity_report_path": str(report_path), "public_mp3_url": build_public_mp3_url(cleaned_path.name), "cleaned_size_bytes": cleaned_path.stat().st_size, "cleaned_duration_seconds": output_metadata.duration_seconds, "processed_at": now_utc().isoformat(), } return normalize_episode_record(record) def try_seed_existing_episode(state: dict, snapshot: FeedSnapshot) -> bool: if state["episodes"]: return False seeded = False for episode in snapshot.episodes: if episode.duration_seconds <= 2 * 60 * 60: continue paths = build_storage_paths(episode.title) if not paths["root_cleaned"].exists(): continue cleaned_path = ensure_file_in_place(paths["root_cleaned"], paths["cleaned"]) report_path = ensure_file_in_place(paths["root_report"], paths["report"]) or paths["report"] original_path = ensure_file_in_place(paths["root_original"], paths["original"]) if not report_path.exists(): report_path.write_text("", encoding="utf-8") record = build_episode_record(episode, snapshot.channel.image_url, original_path, cleaned_path, report_path) upsert_episode_record(state, record) seeded = True return seeded def publish_episode(snapshot: FeedSnapshot, state: dict) -> tuple[bool, str]: try: latest_episode = find_latest_episode_over_2h(snapshot) except PodcastFetchError as exc: if "No episode longer than 2 hours" in str(exc): return False, "no_eligible_episode" raise existing_record = find_record(state, latest_episode.guid) if episode_is_ready(existing_record): return False, "latest_episode_already_published" paths = build_storage_paths(latest_episode.title) cleaned_path = paths["cleaned"] report_path = paths["report"] if cleaned_path.exists(): original_path = paths["original"] if paths["original"].exists() else None if not report_path.exists() and paths["root_report"].exists(): ensure_file_in_place(paths["root_report"], report_path) if not report_path.exists(): report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text("", encoding="utf-8") record = build_episode_record(latest_episode, snapshot.channel.image_url, original_path, cleaned_path, report_path) upsert_episode_record(state, record) return True, "reused_existing_cleaned_file" if paths["root_cleaned"].exists(): cleaned_path = ensure_file_in_place(paths["root_cleaned"], paths["cleaned"]) report_path = ensure_file_in_place(paths["root_report"], paths["report"]) or paths["report"] original_path = ensure_file_in_place(paths["root_original"], paths["original"]) if not report_path.exists(): report_path.write_text("", encoding="utf-8") record = build_episode_record(latest_episode, snapshot.channel.image_url, original_path, cleaned_path, report_path) upsert_episode_record(state, record) return True, "seeded_from_root_cleaned_file" source_input_path = paths["original"] if not source_input_path.exists(): if paths["root_original"].exists(): source_input_path = ensure_file_in_place(paths["root_original"], paths["original"]) else: download_file(latest_episode.mp3_url, paths["original"]) source_input_path = paths["original"] clean_podcast(source_input_path, output_audio_path=cleaned_path, report_path=report_path) record = build_episode_record(latest_episode, snapshot.channel.image_url, source_input_path, cleaned_path, report_path) upsert_episode_record(state, record) return True, "downloaded_and_cleaned_latest_episode" def purge_old_originals(reference_time: datetime, retention: timedelta = ORIGINAL_RETENTION) -> list[Path]: deleted: list[Path] = [] if not ORIGINALS_DIR.exists(): return deleted cutoff = reference_time - retention for original_path in ORIGINALS_DIR.glob("*.mp3"): modified_at = datetime.fromtimestamp(original_path.stat().st_mtime, tz=UTC) if modified_at < cutoff: original_path.unlink() deleted.append(original_path) return deleted def purge_orphan_parts(reference_time: datetime, retention: timedelta = PART_RETENTION) -> list[Path]: deleted: list[Path] = [] cutoff = reference_time - retention for part_path in ROOT_DIR.rglob("*.part*"): modified_at = datetime.fromtimestamp(part_path.stat().st_mtime, tz=UTC) if modified_at < cutoff: part_path.unlink() deleted.append(part_path) return deleted def sort_episode_records(records: list[dict]) -> list[dict]: return sorted(records, key=lambda record: parse_pub_date(record["pub_date"]), reverse=True) def append_category_elements(channel_element: ET.Element, categories: tuple[FeedCategory, ...]) -> None: for category in categories: primary_element = ET.SubElement(channel_element, f"{{{ITUNES_NS}}}category", {"text": category.primary}) if category.secondary: ET.SubElement(primary_element, f"{{{ITUNES_NS}}}category", {"text": category.secondary}) def build_feed_tree( channel_state: dict, episode_records: list[dict], *, feed_url: str = FEED_URL, guid_suffix: str = PRIMARY_GUID_SUFFIX, ) -> ET.ElementTree: ET.register_namespace("atom", ATOM_NS) ET.register_namespace("itunes", ITUNES_NS) ET.register_namespace("content", CONTENT_NS) rss_element = ET.Element("rss", {"version": "2.0"}) channel_element = ET.SubElement(rss_element, "channel") ET.SubElement(channel_element, "title").text = CHANNEL_TITLE ET.SubElement(channel_element, "description").text = channel_state["description"] ET.SubElement(channel_element, "link").text = PUBLIC_BASE_URL ET.SubElement(channel_element, "language").text = channel_state["language"] if channel_state["copyright"]: ET.SubElement(channel_element, "copyright").text = channel_state["copyright"] ET.SubElement(channel_element, "generator").text = "clean_gg_rss" ET.SubElement(channel_element, "lastBuildDate").text = format_datetime(now_utc()) ET.SubElement( channel_element, f"{{{ATOM_NS}}}link", { "href": feed_url, "rel": "self", "title": "MP3 Audio", "type": "application/rss+xml", }, ) ET.SubElement(channel_element, f"{{{ITUNES_NS}}}author").text = channel_state["author"] ET.SubElement(channel_element, f"{{{ITUNES_NS}}}summary").text = channel_state["summary"] or channel_state["description"] ET.SubElement(channel_element, f"{{{ITUNES_NS}}}explicit").text = channel_state["explicit"] or "false" if channel_state["image_url"]: ET.SubElement(channel_element, f"{{{ITUNES_NS}}}image", {"href": channel_state["image_url"]}) image_element = ET.SubElement(channel_element, "image") ET.SubElement(image_element, "url").text = channel_state["image_url"] ET.SubElement(image_element, "title").text = CHANNEL_TITLE ET.SubElement(image_element, "link").text = PUBLIC_BASE_URL if channel_state["owner_name"] or channel_state["owner_email"]: owner_element = ET.SubElement(channel_element, f"{{{ITUNES_NS}}}owner") if channel_state["owner_name"]: ET.SubElement(owner_element, f"{{{ITUNES_NS}}}name").text = channel_state["owner_name"] if channel_state["owner_email"]: ET.SubElement(owner_element, f"{{{ITUNES_NS}}}email").text = channel_state["owner_email"] append_category_elements(channel_element, deserialize_categories(channel_state["categories"])) for record in sort_episode_records(episode_records): public_mp3_url = canonical_public_mp3_url(record) item_element = ET.SubElement(channel_element, "item") ET.SubElement(item_element, "guid", {"isPermaLink": "false"}).text = f"{record['source_guid']}{guid_suffix}" ET.SubElement(item_element, "title").text = record["title"] ET.SubElement(item_element, "pubDate").text = record["pub_date"] ET.SubElement(item_element, "description").text = f"{record['description']}{DESCRIPTION_NOTE}" ET.SubElement(item_element, "link").text = public_mp3_url ET.SubElement( item_element, "enclosure", { "url": public_mp3_url, "length": str(record["cleaned_size_bytes"]), "type": "audio/mpeg", }, ) ET.SubElement(item_element, f"{{{ITUNES_NS}}}duration").text = format_duration(record["cleaned_duration_seconds"]) ET.SubElement(item_element, f"{{{ITUNES_NS}}}explicit").text = channel_state["explicit"] or "false" ET.SubElement(item_element, f"{{{ITUNES_NS}}}author").text = channel_state["author"] ET.SubElement(item_element, f"{{{ITUNES_NS}}}summary").text = record["description"] ET.SubElement(item_element, f"{{{CONTENT_NS}}}encoded").text = f"{record['description']}{DESCRIPTION_NOTE}" return ET.ElementTree(rss_element) def serialize_feed_tree(tree: ET.ElementTree) -> str: feed_xml = ET.tostring(tree.getroot(), encoding="utf-8", xml_declaration=True).decode("utf-8") if LEGACY_PUBLIC_BASE_URL in feed_xml: raise PublishFeedError(f"Generated feed still contains legacy HTTP URLs: {LEGACY_PUBLIC_BASE_URL}") return feed_xml def write_feed(state: dict) -> None: if not state["channel"]: raise PublishFeedError("Cannot generate podcast.xml without channel metadata.") primary_tree = build_feed_tree( state["channel"], state["episodes"], feed_url=FEED_URL, guid_suffix=PRIMARY_GUID_SUFFIX, ) primary_feed_xml = serialize_feed_tree(primary_tree) FEED_OUTPUT_PATH.write_text(primary_feed_xml, encoding="utf-8") cache_bust_tree = build_feed_tree( state["channel"], state["episodes"], feed_url=CACHE_BUST_FEED_URL, guid_suffix=CACHE_BUST_GUID_SUFFIX, ) cache_bust_feed_xml = serialize_feed_tree(cache_bust_tree) CACHE_BUST_FEED_OUTPUT_PATH.write_text(cache_bust_feed_xml, encoding="utf-8") def install_seed_if_needed(state: dict, snapshot: FeedSnapshot) -> bool: changed = try_seed_existing_episode(state, snapshot) if changed: save_state(state) return changed def run_pipeline() -> str: ensure_directories() with pipeline_lock(): state = load_state() rss_xml = fetch_rss() snapshot = parse_feed_snapshot(rss_xml) state["channel"] = snapshot_channel_to_state(snapshot.channel) install_seed_if_needed(state, snapshot) changed, reason = publish_episode(snapshot, state) save_state(state) if state["episodes"]: write_feed(state) purge_old_originals(now_utc()) purge_orphan_parts(now_utc()) return reason def main(argv: list[str] | None = None) -> int: if argv is None: argv = sys.argv[1:] if argv: print("Usage: python publish_clean_feed.py", file=sys.stderr) return 1 try: result = run_pipeline() print(f"Pipeline result: {result}") print(f"Feed URL: {FEED_URL}") print(f"Cache-bust feed URL: {CACHE_BUST_FEED_URL}") return 0 except (PodcastFetchError, PodcastCleanerError, PublishFeedError) as exc: print(f"Error: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())