diff --git a/fetch_podcasts.py b/fetch_podcasts.py index b9237bf..d6baa90 100755 --- a/fetch_podcasts.py +++ b/fetch_podcasts.py @@ -6,6 +6,12 @@ for the liquidsoap radio automation stack. All data (podcasts, state DBs, logs, playlists) lives under the storage path defined in config.json ("storage" key), keeping the boot drive clean. +Concurrency model: + - An exclusive, non-blocking lockfile (state/fetch.lock) guarantees this + process and update_playlists.py never touch the databases simultaneously. + If the updater holds the lock, this run logs a skip and exits 0. + - Both databases run in WAL mode with a busy timeout as a second safety net. + Filters out video podcasts: only shows whose latest enclosure has an audio/* MIME type are registered. @@ -14,10 +20,13 @@ shows (archived=0) store only the enclosure URL for live streaming. """ import argparse +import fcntl import json import logging +import os import re import shutil +import socket import sqlite3 import sys import uuid @@ -25,10 +34,16 @@ import xml.etree.ElementTree as ET from pathlib import Path from urllib.parse import quote +# Force IPv4 resolution so we don't stall on AAAA-first lookups when the box +# has no usable IPv6 route (gpodder.net publishes both A and AAAA records). +try: + import requests.packages.urllib3.util.connection as _urllib3_conn + _urllib3_conn.HAS_IPV6 = False +except Exception: + pass + import feedparser import requests -import requests.packages.urllib3.util.connection as _urllib3_conn -_urllib3_conn.HAS_IPV6 = False ROOT = Path(__file__).resolve().parent CONFIG_PATH = ROOT / "config.json" @@ -42,6 +57,7 @@ PLAYED_DB = None PODCASTS_DIR = None LOGS_DIR = None PLAYLISTS_DIR = None +LOCK_FILE = None def load_config(): with open(CONFIG_PATH) as f: @@ -49,7 +65,7 @@ def load_config(): def init_paths(): """Resolve all data paths from config.json storage key.""" - global STATE_DIR, SUBS_DB, PLAYED_DB, PODCASTS_DIR, LOGS_DIR, PLAYLISTS_DIR + global STATE_DIR, SUBS_DB, PLAYED_DB, PODCASTS_DIR, LOGS_DIR, PLAYLISTS_DIR, LOCK_FILE cfg = load_config() storage = Path(cfg["storage"]).expanduser().resolve() STATE_DIR = storage / "state" @@ -58,6 +74,7 @@ def init_paths(): PODCASTS_DIR = storage / "podcasts" LOGS_DIR = storage / "logs" PLAYLISTS_DIR = storage / "playlists" + LOCK_FILE = STATE_DIR / "radio.lock" logging.basicConfig( level=logging.INFO, @@ -77,41 +94,100 @@ def _setup_logging(): fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) log.addHandler(fh) -def open_subs_db(): - conn = sqlite3.connect(SUBS_DB) +# --------------------------------------------------------------------------- +# Mutual exclusion: one writer across fetch + update at any moment. +# --------------------------------------------------------------------------- +_lock_fd = None + +def acquire_lock(): + """Acquire an exclusive non-blocking lock. Returns True if acquired, + False if another radio process already holds it.""" + global _lock_fd + STATE_DIR.mkdir(parents=True, exist_ok=True) + fd = os.open(str(LOCK_FILE), os.O_CREAT | os.O_RDWR, 0o644) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + os.close(fd) + return False + # Record our PID for observability. + os.ftruncate(fd, 0) + os.write(fd, str(os.getpid()).encode()) + _lock_fd = fd + return True + +def release_lock(): + global _lock_fd + if _lock_fd is not None: + try: + fcntl.flock(_lock_fd, fcntl.LOCK_UN) + os.close(_lock_fd) + finally: + _lock_fd = None + +# --------------------------------------------------------------------------- +# Schema: single source of truth per database, applied idempotently. +# --------------------------------------------------------------------------- +SHOWS_COLUMNS = { + "slug": "TEXT PRIMARY KEY", + "guid": "TEXT NOT NULL UNIQUE", + "name": "TEXT NOT NULL", + "feed_url": "TEXT NOT NULL UNIQUE", + "source": "TEXT DEFAULT 'manual'", + "opml_import": "INTEGER DEFAULT 0", + "archived": "INTEGER DEFAULT 1", + "created_at": "TEXT DEFAULT (datetime('now'))", +} + +EPISODES_COLUMNS = { + "id": "INTEGER PRIMARY KEY AUTOINCREMENT", + "show_slug": "TEXT NOT NULL", + "guid": "TEXT NOT NULL", + "title": "TEXT", + "file_path": "TEXT", + "enclosure_url": "TEXT", + "runlength": "INTEGER", + "played": "INTEGER DEFAULT 0", + "played_at": "TEXT", +} + +def _connect(db_path): + conn = sqlite3.connect(str(db_path), timeout=5.0) conn.row_factory = sqlite3.Row - conn.execute(""" - CREATE TABLE IF NOT EXISTS shows ( - slug TEXT PRIMARY KEY, - guid TEXT NOT NULL UNIQUE, - name TEXT NOT NULL, - feed_url TEXT NOT NULL UNIQUE, - source TEXT DEFAULT 'manual', - opml_import INTEGER DEFAULT 0, - archived INTEGER DEFAULT 1, - created_at TEXT DEFAULT (datetime('now')) - ) - """) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + return conn + +def _ensure_columns(conn, table, columns): + """Create the table if absent, else add any missing columns.""" + exists = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + if exists is None: + defs = ",\n ".join(f"{name} {spec}" for name, spec in columns.items()) + extra = "" + if table == "episodes": + extra = "\n ,UNIQUE(show_slug, guid)" + conn.execute(f"CREATE TABLE {table} (\n {defs}{extra}\n )") + else: + existing = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")} + for name, spec in columns.items(): + if name not in existing: + col_default = spec.split("DEFAULT", 1)[1].strip() if "DEFAULT" in spec else "NULL" + conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {col_default}") conn.commit() + +def open_subs_db(): + conn = _connect(SUBS_DB) + _ensure_columns(conn, "shows", SHOWS_COLUMNS) return conn def open_played_db(): - conn = sqlite3.connect(PLAYED_DB) - conn.row_factory = sqlite3.Row - conn.execute(""" - CREATE TABLE IF NOT EXISTS episodes ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - show_slug TEXT NOT NULL, - guid TEXT NOT NULL, - title TEXT, - file_path TEXT, - enclosure_url TEXT, - runlength INTEGER, - played INTEGER DEFAULT 0, - played_at TEXT, - UNIQUE(show_slug, guid) - ) - """) + conn = _connect(PLAYED_DB) + _ensure_columns(conn, "episodes", EPISODES_COLUMNS) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_episodes_show_played ON episodes (show_slug, played)" + ) conn.commit() return conn @@ -416,6 +492,7 @@ def set_archive(slug, value): row = db.execute("SELECT name FROM shows WHERE slug = ?", (slug,)).fetchone() if row is None: log_error(f"No show found with slug '{slug}'.") + db.close() return db.execute("UPDATE shows SET archived = ? WHERE slug = ?", (value, slug)) db.commit() @@ -436,6 +513,7 @@ def delete_show(slug): row = db.execute("SELECT name FROM shows WHERE slug = ?", (slug,)).fetchone() if row is None: log_error(f"No show found with slug '{slug}'.") + db.close() return remove_show_data(slug) db.execute("DELETE FROM shows WHERE slug = ?", (slug,)) @@ -513,20 +591,32 @@ def main(): config = load_config() - if args.archive: - set_archive(args.archive, 1) - elif args.unarchive: - set_archive(args.unarchive, 0) - elif args.list_shows: - list_shows(detail=args.detail) - elif args.add_show: - add_show(args.add_show) - elif args.delete_show: - delete_show(args.delete_show) - elif args.import_opml: - import_opml(args.import_opml) - else: - run_fetch(config) + # Read-only administrative commands don't need the write lock. + needs_lock = not args.list_shows + + if needs_lock and not acquire_lock(): + log.info("Another radio process holds the lock; skipping this run.") + return + + try: + if args.archive: + set_archive(args.archive, 1) + elif args.unarchive: + set_archive(args.unarchive, 0) + elif args.list_shows: + list_shows(detail=args.detail) + elif args.add_show: + add_show(args.add_show) + elif args.delete_show: + delete_show(args.delete_show) + elif args.import_opml: + import_opml(args.import_opml) + else: + run_fetch(config) + finally: + if needs_lock: + release_lock() if __name__ == "__main__": main() + diff --git a/install_for_jruby b/install_for_jruby index b4b6e9f..401f4aa 100755 --- a/install_for_jruby +++ b/install_for_jruby @@ -102,6 +102,9 @@ echo "==> Installing gems one at a time into ${GEMS_DIR} ..." "$JRUBY_BIN" -S gem install --no-document jdbc-sqlite3 "$JRUBY_BIN" -S gem install --no-document json +# Give the service user ownership of the gem cache so cron runs work. +chown -R "$LIQUIDSOAP_USER":"$LIQUIDSOAP_USER" "$GEMS_DIR" + # --------------------------------------------------------------------------- # 6. Interactive configuration (defaults from existing config.json) # --------------------------------------------------------------------------- @@ -279,22 +282,32 @@ exec("liquidsoap", File.expand_path("station.liq")) RBRUN chown "$LIQUIDSOAP_USER":"$LIQUIDSOAP_USER" "${INSTALL_DIR}/station.rb" +# Allow the service user to traverse the install dir. +chmod o+x "$INSTALL_DIR" + systemctl daemon-reload # --------------------------------------------------------------------------- -# 11. Cron jobs (deduplicated) +# 11. Cron jobs — installed into the liquidsoap user's crontab so all +# database/media writes happen under the same identity as the service. # --------------------------------------------------------------------------- -echo "==> Setting up cron jobs ..." +echo "==> Setting up cron jobs under user '${LIQUIDSOAP_USER}' ..." FETCH_PREFIX="cd ${INSTALL_DIR} && GEM_HOME=${GEMS_DIR} GEM_PATH=${GEMS_DIR} ${JRUBY_BIN} -S fetch_podcasts.rb" UPDATE_PREFIX="cd ${INSTALL_DIR} && GEM_HOME=${GEMS_DIR} GEM_PATH=${GEMS_DIR} ${JRUBY_BIN} -S update_playlists.rb" FETCH_CRON="0 * * * * ${FETCH_PREFIX} >> ${STORAGE_PATH}/logs/fetch.log 2>&1" UPDATE_CRON="30 * * * * ${UPDATE_PREFIX} >> ${STORAGE_PATH}/logs/update.log 2>&1" -crontab -l 2>/dev/null | grep -vF "fetch_podcasts.rb" | grep -vF "update_playlists.rb" > /tmp/cron_radio_rb.$$ || true -echo "$FETCH_CRON" >> /tmp/cron_radio_rb.$$ -echo "$UPDATE_CRON" >> /tmp/cron_radio_rb.$$ -crontab /tmp/cron_radio_rb.$$ -rm -f /tmp/cron_radio_rb.$$ +# Remove any stale entries from the liquidsoap user's crontab, then add ours. +sudo -u "$LIQUIDSOAP_USER" crontab -l 2>/dev/null | grep -vF "fetch_podcasts.rb" | grep -vF "update_playlists.rb" > /tmp/cron_ls_rb.$$ || true +echo "$FETCH_CRON" >> /tmp/cron_ls_rb.$$ +echo "$UPDATE_CRON" >> /tmp/cron_ls_rb.$$ +sudo -u "$LIQUIDSOAP_USER" crontab /tmp/cron_ls_rb.$$ +rm -f /tmp/cron_ls_rb.$$ + +# Also scrub these from root's crontab in case an earlier install put them there. +crontab -l 2>/dev/null | grep -vF "fetch_podcasts.rb" | grep -vF "update_playlists.rb" > /tmp/cron_root_rb.$$ || true +crontab /tmp/cron_root_rb.$$ +rm -f /tmp/cron_root_rb.$$ # --------------------------------------------------------------------------- # 12. Enable services diff --git a/install_for_python b/install_for_python index 3e08b16..2d4e725 100755 --- a/install_for_python +++ b/install_for_python @@ -40,7 +40,7 @@ fi CONFIG_JSON="${INSTALL_DIR}/config.json" prompt() { - local var_name="$1" prompt_text="$2" default="${3:-}" + local prompt_text="$1" default="${2:-}" local current="" if [[ -n "$default" ]]; then read -rp "${prompt_text} [${default}]: " current || true @@ -71,22 +71,22 @@ fi echo "" echo "--- Storage ---" -STORAGE_PATH=$(prompt storage_path "Storage path (media + state + logs)" "/mnt/storage/radio${existing_storage:+|$existing_storage}") +STORAGE_PATH=$(prompt "Storage path (media + state + logs)" "${existing_storage:-/mnt/storage/radio}") [[ -z "$STORAGE_PATH" ]] && STORAGE_PATH="${existing_storage:-/mnt/storage/radio}" echo "" echo "--- Icecast (source credentials) ---" -IC_HOST=$(prompt ic_host "Icecast host" "$existing_ic_host"); IC_HOST=${IC_HOST:-localhost} -IC_PORT=$(prompt ic_port "Icecast source port" "$existing_ic_port"); IC_PORT=${IC_PORT:-7777} -IC_MOUNT=$(prompt ic_mount "Mount point" "$existing_ic_mount"); IC_MOUNT=${IC_MOUNT:-/data} -IC_USER=$(prompt ic_user "Source username" "$existing_ic_user"); IC_USER=${IC_USER:-source} +IC_HOST=$(prompt "Icecast host" "$existing_ic_host"); IC_HOST=${IC_HOST:-localhost} +IC_PORT=$(prompt "Icecast source port" "$existing_ic_port"); IC_PORT=${IC_PORT:-7777} +IC_MOUNT=$(prompt "Mount point" "$existing_ic_mount"); IC_MOUNT=${IC_MOUNT:-/data} +IC_USER=$(prompt "Source username" "$existing_ic_user"); IC_USER=${IC_USER:-source} read -rsp "Source password: " IC_PASS || true; echo; IC_PASS=${IC_PASS:-$existing_ic_pass} echo "" echo "--- gPodder sync (optional) ---" -GP_ENABLE=$(prompt gp_enable "Enable gPodder sync? (true/false)" "$existing_gp_enable"); GP_ENABLE=${GP_ENABLE:-false} -GP_HOST=$(prompt gp_host "gPodder host" "$existing_gp_host"); GP_HOST=${GP_HOST:-https://gpodder.net} -GP_USER=$(prompt gp_user "gPodder username" "$existing_gp_user") +GP_ENABLE=$(prompt "Enable gPodder sync? (true/false)" "$existing_gp_enable"); GP_ENABLE=${GP_ENABLE:-false} +GP_HOST=$(prompt "gPodder host" "$existing_gp_host"); GP_HOST=${GP_HOST:-https://gpodder.net} +GP_USER=$(prompt "gPodder username" "$existing_gp_user") read -rsp "gPodder password: " GP_PASS || true; echo; GP_PASS=${GP_PASS:-$existing_gp_pass} echo "" @@ -195,6 +195,10 @@ echo "==> Installing Python dependencies (feedparser, requests) ..." "${VENV}/bin/pip" install --quiet --upgrade pip "${VENV}/bin/pip" install --quiet feedparser requests +# Make sure the liquidsoap user can traverse the install dir and use the venv. +chmod o+x "$INSTALL_DIR" +chown -R "$LIQUIDSOAP_USER":"$LIQUIDSOAP_USER" "$VENV" + # --------------------------------------------------------------------------- # 7. systemd unit (named after the directory) # --------------------------------------------------------------------------- @@ -229,17 +233,24 @@ chown "$LIQUIDSOAP_USER":"$LIQUIDSOAP_USER" "${INSTALL_DIR}/station_runner.py" systemctl daemon-reload # --------------------------------------------------------------------------- -# 8. Cron jobs (deduplicated) +# 8. Cron jobs — installed into the liquidsoap user's crontab so all +# database/media writes happen under the same identity as the service. # --------------------------------------------------------------------------- -echo "==> Setting up cron jobs ..." +echo "==> Setting up cron jobs under user '${LIQUIDSOAP_USER}' ..." FETCH_CRON="0 * * * * cd ${INSTALL_DIR} && ./.venv/bin/python fetch_podcasts.py >> ${STORAGE_PATH}/logs/fetch.log 2>&1" UPDATE_CRON="30 * * * * cd ${INSTALL_DIR} && ./.venv/bin/python update_playlists.py >> ${STORAGE_PATH}/logs/update.log 2>&1" -crontab -l 2>/dev/null | grep -vF "fetch_podcasts.py" | grep -vF "update_playlists.py" > /tmp/cron_radio.$$ || true -echo "$FETCH_CRON" >> /tmp/cron_radio.$$ -echo "$UPDATE_CRON" >> /tmp/cron_radio.$$ -crontab /tmp/cron_radio.$$ -rm -f /tmp/cron_radio.$$ +# Remove any stale entries from the liquidsoap user's crontab, then add ours. +sudo -u "$LIQUIDSOAP_USER" crontab -l 2>/dev/null | grep -vF "fetch_podcasts.py" | grep -vF "update_playlists.py" > /tmp/cron_ls_py.$$ || true +echo "$FETCH_CRON" >> /tmp/cron_ls_py.$$ +echo "$UPDATE_CRON" >> /tmp/cron_ls_py.$$ +sudo -u "$LIQUIDSOAP_USER" crontab /tmp/cron_ls_py.$$ +rm -f /tmp/cron_ls_py.$$ + +# Also scrub these from root's crontab in case an earlier install put them there. +crontab -l 2>/dev/null | grep -vF "fetch_podcasts.py" | grep -vF "update_playlists.py" > /tmp/cron_root_py.$$ || true +crontab /tmp/cron_root_py.$$ +rm -f /tmp/cron_root_py.$$ # --------------------------------------------------------------------------- # 9. Enable services diff --git a/update_playlists.py b/update_playlists.py index 545217d..7fb2047 100755 --- a/update_playlists.py +++ b/update_playlists.py @@ -1,19 +1,28 @@ #!/usr/bin/env python3 """ update_playlists.py - Select the next unplayed episode per show and write an -annotated URI line for station.liq to consume. Runs via cron hourly at :30. +annotated URI queue file for station.liq to consume. -Selection is driven by the episodes table (keyed on guid + played flag), not -by scanning the filesystem, so it works identically for archived shows -(local file_path) and live shows (remote enclosure_url). All data under the -storage path from config.json. +Concurrency model mirrors fetch_podcasts.py: + - Shares the same exclusive lockfile (state/radio.lock). If the fetcher is + still running, this run logs a skip and exits 0 instead of hitting a + locked database. + - Databases run in WAL mode with a busy timeout as a second safety net. + +Selection: for each show, pick the earliest episode with played=0, write +playlists/.txt as a single annotated URI line, then mark it played. +Works identically for archived (local file_path) and live (enclosure_url) +shows because both store their episodes in the same table. """ import argparse +import fcntl import json import logging +import os import sqlite3 import sys +from datetime import datetime from pathlib import Path ROOT = Path(__file__).resolve().parent @@ -23,23 +32,25 @@ STATE_DIR = None SUBS_DB = None PLAYED_DB = None PODCASTS_DIR = None -PLAYLISTS_DIR = None LOGS_DIR = None +PLAYLISTS_DIR = None +LOCK_FILE = None def load_config(): with open(CONFIG_PATH) as f: return json.load(f) def init_paths(): - global STATE_DIR, SUBS_DB, PLAYED_DB, PODCASTS_DIR, PLAYLISTS_DIR, LOGS_DIR + global STATE_DIR, SUBS_DB, PLAYED_DB, PODCASTS_DIR, LOGS_DIR, PLAYLISTS_DIR, LOCK_FILE cfg = load_config() storage = Path(cfg["storage"]).expanduser().resolve() STATE_DIR = storage / "state" SUBS_DB = STATE_DIR / "subscriptions.db" PLAYED_DB = STATE_DIR / "played.db" PODCASTS_DIR = storage / "podcasts" - PLAYLISTS_DIR = storage / "playlists" LOGS_DIR = storage / "logs" + PLAYLISTS_DIR = storage / "playlists" + LOCK_FILE = STATE_DIR / "radio.lock" logging.basicConfig( level=logging.INFO, @@ -50,125 +61,187 @@ logging.basicConfig( ) log = logging.getLogger("update_playlists") +def log_error(msg): + log.error(msg) + def _setup_logging(): fh = logging.FileHandler(LOGS_DIR / "update.log") fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) log.addHandler(fh) -def open_subs_db(): - conn = sqlite3.connect(SUBS_DB) +# --------------------------------------------------------------------------- +# Shared schema definitions (must match fetch_podcasts.py). +# --------------------------------------------------------------------------- +SHOWS_COLUMNS = { + "slug": "TEXT PRIMARY KEY", + "guid": "TEXT NOT NULL UNIQUE", + "name": "TEXT NOT NULL", + "feed_url": "TEXT NOT NULL UNIQUE", + "source": "TEXT DEFAULT 'manual'", + "opml_import": "INTEGER DEFAULT 0", + "archived": "INTEGER DEFAULT 1", + "created_at": "TEXT DEFAULT (datetime('now'))", +} + +EPISODES_COLUMNS = { + "id": "INTEGER PRIMARY KEY AUTOINCREMENT", + "show_slug": "TEXT NOT NULL", + "guid": "TEXT NOT NULL", + "title": "TEXT", + "file_path": "TEXT", + "enclosure_url": "TEXT", + "runlength": "INTEGER", + "played": "INTEGER DEFAULT 0", + "played_at": "TEXT", +} + +def _connect(db_path): + conn = sqlite3.connect(str(db_path), timeout=5.0) conn.row_factory = sqlite3.Row - conn.execute(""" - CREATE TABLE IF NOT EXISTS shows ( - slug TEXT PRIMARY KEY, - guid TEXT NOT NULL UNIQUE, - name TEXT NOT NULL, - feed_url TEXT NOT NULL UNIQUE, - source TEXT DEFAULT 'manual', - opml_import INTEGER DEFAULT 0, - archived INTEGER DEFAULT 1, - created_at TEXT DEFAULT (datetime('now')) - ) - """) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + return conn + +def _ensure_columns(conn, table, columns): + exists = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + if exists is None: + defs = ",\n ".join(f"{name} {spec}" for name, spec in columns.items()) + extra = "" + if table == "episodes": + extra = "\n ,UNIQUE(show_slug, guid)" + conn.execute(f"CREATE TABLE {table} (\n {defs}{extra}\n )") + else: + existing = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")} + for name, spec in columns.items(): + if name not in existing: + col_default = spec.split("DEFAULT", 1)[1].strip() if "DEFAULT" in spec else "NULL" + conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {col_default}") conn.commit() + +def open_subs_db(): + conn = _connect(SUBS_DB) + _ensure_columns(conn, "shows", SHOWS_COLUMNS) return conn def open_played_db(): - conn = sqlite3.connect(PLAYED_DB) - conn.row_factory = sqlite3.Row - conn.execute(""" - CREATE TABLE IF NOT EXISTS episodes ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - show_slug TEXT NOT NULL, - guid TEXT NOT NULL, - title TEXT, - file_path TEXT, - enclosure_url TEXT, - runlength INTEGER, - played INTEGER DEFAULT 0, - played_at TEXT, - UNIQUE(show_slug, guid) - ) - """) + conn = _connect(PLAYED_DB) + _ensure_columns(conn, "episodes", EPISODES_COLUMNS) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_episodes_show_played ON episodes (show_slug, played)" + ) conn.commit() return conn -def select_unplayed_episode(slug, played_db): - """Pick the next unplayed episode for a show, keyed by guid. - Archived -> local file_path; Live -> remote enclosure_url.""" +# --------------------------------------------------------------------------- +# Mutual exclusion (shared with fetch_podcasts.py via the same lockfile). +# --------------------------------------------------------------------------- +_lock_fd = None + +def acquire_lock(): + global _lock_fd + STATE_DIR.mkdir(parents=True, exist_ok=True) + fd = os.open(str(LOCK_FILE), os.O_CREAT | os.O_RDWR, 0o644) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + os.close(fd) + return False + os.ftruncate(fd, 0) + os.write(fd, str(os.getpid()).encode()) + _lock_fd = fd + return True + +def release_lock(): + global _lock_fd + if _lock_fd is not None: + try: + fcntl.flock(_lock_fd, fcntl.LOCK_UN) + os.close(_lock_fd) + finally: + _lock_fd = None + +# --------------------------------------------------------------------------- +# Core logic +# --------------------------------------------------------------------------- +def _annotate_uri(runlength, title, uri): + """Build a liquidsoap annotated URI line. Values must be double-quoted + and separated by commas; the whole annotation precedes a colon before + the URI. Escapes embedded double quotes minimally.""" + def q(v): + return '"' + str(v).replace('"', '\\"') + '"' + ann = f"annotate:liq_runlength={q(runlength)},liq_title={q(title)}:" + return ann + uri + +def select_next_episode(slug, played_db): row = played_db.execute( "SELECT guid, title, file_path, enclosure_url, runlength " - "FROM episodes WHERE show_slug = ? AND played = 0 ORDER BY id ASC LIMIT 1", + "FROM episodes WHERE show_slug=? AND played=0 ORDER BY id ASC LIMIT 1", (slug,), ).fetchone() return row -def annotated_uri(ep): - """Build the annotate: URI line station.liq consumes. - Archived -> local file path; Live -> remote enclosure URL.""" - uri = ep["file_path"] or ep["enclosure_url"] - if not uri: - return None - rl = int(ep["runlength"] or 0) - title = (ep["title"] or "").replace('"', "'") - return f'annotate:liq_runlength="{rl}",liq_title="{title}":{uri}' - -def write_queue_line(slug, line, out_path): - out_path.parent.mkdir(parents=True, exist_ok=True) - out_path.write_text(line + "\n") - def mark_as_played(slug, guid, played_db): played_db.execute( - "UPDATE episodes SET played = 1, played_at = datetime('now') WHERE show_slug = ? AND guid = ?", + "UPDATE episodes SET played=1, played_at=datetime('now') " + "WHERE show_slug=? AND guid=?", (slug, guid), ) played_db.commit() +def write_queue_file(slug, ep, archived): + if archived: + uri = ep["file_path"] + else: + uri = ep["enclosure_url"] + if not uri: + return None + line = _annotate_uri(ep["runlength"], ep["title"], uri) + out = PLAYLISTS_DIR / f"{slug}.txt" + out.write_text(line + "\n") + return out + def update_all(): subs_db = open_subs_db() played_db = open_played_db() shows = subs_db.execute("SELECT slug, name, archived FROM shows ORDER BY name").fetchall() - queued = 0 - skipped = 0 + updated = 0 for show in shows: slug = show["slug"] - ep = select_unplayed_episode(slug, played_db) + archived = show["archived"] == 1 + ep = select_next_episode(slug, played_db) if ep is None: - log.info("%s: no unplayed episodes, skipping.", slug) - skipped += 1 continue - line = annotated_uri(ep) - if line is None: - log.info("%s: episode %s has no usable URI, skipping.", slug, ep["guid"]) - skipped += 1 + out = write_queue_file(slug, ep, archived) + if out is None: + log.warning("No playable URI for %s (%s); skipping.", show["name"], slug) continue - out_txt = PLAYLISTS_DIR / f"{slug}.txt" - write_queue_line(slug, line, out_txt) mark_as_played(slug, ep["guid"], played_db) - kind = "downloaded" if ep["file_path"] else "live" - log.info("%s: queued %s [%s] runlength=%ss", slug, ep["title"], kind, int(ep["runlength"] or 0)) - queued += 1 + updated += 1 + log.info("Queued %s: %s -> %s", slug, ep["title"], out.name) subs_db.close() played_db.close() - log.info("=== Update complete: %d queued, %d skipped ===", queued, skipped) + log.info("=== Update complete: %d show(s) queued ===", updated) def json_summary(): subs_db = open_subs_db() played_db = open_played_db() - shows = subs_db.execute("SELECT slug FROM shows ORDER BY name").fetchall() - summary = {} + shows = subs_db.execute("SELECT slug, name FROM shows ORDER BY name").fetchall() + result = {} for show in shows: slug = show["slug"] - total = played_db.execute( - "SELECT COUNT(*) as c FROM episodes WHERE show_slug = ?", (slug,) - ).fetchone()["c"] - played = played_db.execute( - "SELECT COUNT(*) as c FROM episodes WHERE show_slug = ? AND played = 1", (slug,) - ).fetchone()["c"] - summary[slug] = {"total_episodes": total, "played_count": played, "unplayed": total - played} + counts = played_db.execute( + "SELECT COUNT(*) AS total, SUM(CASE WHEN played=0 THEN 1 ELSE 0 END) AS unplayed " + "FROM episodes WHERE show_slug=?", + (slug,), + ).fetchone() + total = counts["total"] or 0 + unplayed = counts["unplayed"] or 0 + result[slug] = {"name": show["name"], "total": total, "unplayed": unplayed, "played": total - unplayed} subs_db.close() played_db.close() - print(json.dumps(summary, indent=2)) + print(json.dumps(result, indent=2)) def main(): parser = argparse.ArgumentParser(description="Playlist updater for radio automation") @@ -181,10 +254,21 @@ def main(): PLAYLISTS_DIR.mkdir(parents=True, exist_ok=True) _setup_logging() - if args.json: - json_summary() - else: - update_all() + # --json is read-only and doesn't need the write lock. + needs_lock = not args.json + + if needs_lock and not acquire_lock(): + log.info("Another radio process holds the lock; skipping this run.") + return + + try: + if args.json: + json_summary() + else: + update_all() + finally: + if needs_lock: + release_lock() if __name__ == "__main__": main()