Updated ownership under liquidsoap user

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
 
-import feedparser
+# Force IPv4 resolution so we don't stall on AAAA-first lookups when the box
-import requests
+# 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
 
 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("""
+    conn.execute("PRAGMA journal_mode=WAL")
-        CREATE TABLE IF NOT EXISTS shows (
+    conn.execute("PRAGMA busy_timeout=5000")
-            slug TEXT PRIMARY KEY,
+    return conn
-            guid TEXT NOT NULL UNIQUE,
+
-            name TEXT NOT NULL,
+def _ensure_columns(conn, table, columns):
-            feed_url TEXT NOT NULL UNIQUE,
+    """Create the table if absent, else add any missing columns."""
-            source TEXT DEFAULT 'manual',
+    exists = conn.execute(
-            opml_import INTEGER DEFAULT 0,
+        "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)
-            archived INTEGER DEFAULT 1,
+    ).fetchone()
-            created_at TEXT DEFAULT (datetime('now'))
+    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 = _connect(PLAYED_DB)
-    conn.row_factory = sqlite3.Row
+    _ensure_columns(conn, "episodes", EPISODES_COLUMNS)
-    conn.execute("""
+    conn.execute(
-        CREATE TABLE IF NOT EXISTS episodes (
+        "CREATE INDEX IF NOT EXISTS idx_episodes_show_played ON episodes (show_slug, played)"
-            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.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,6 +591,14 @@ def main():
 
     config = load_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:
@@ -527,6 +613,10 @@ def main():
             import_opml(args.import_opml)
         else:
             run_fetch(config)
+    finally:
+        if needs_lock:
+            release_lock()
 
 if __name__ == "__main__":
     main()
+

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
+# Remove any stale entries from the liquidsoap user's crontab, then add ours.
-echo "$FETCH_CRON"  >> /tmp/cron_radio_rb.$$
+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 "$UPDATE_CRON" >> /tmp/cron_radio_rb.$$
+echo "$FETCH_CRON"  >> /tmp/cron_ls_rb.$$
-crontab /tmp/cron_radio_rb.$$
+echo "$UPDATE_CRON" >> /tmp/cron_ls_rb.$$
-rm -f /tmp/cron_radio_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

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_HOST=$(prompt "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_PORT=$(prompt "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_MOUNT=$(prompt "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_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_ENABLE=$(prompt "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_HOST=$(prompt "gPodder host" "$existing_gp_host");            GP_HOST=${GP_HOST:-https://gpodder.net}
-GP_USER=$(prompt gp_user "gPodder username" "$existing_gp_user")
+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
+# Remove any stale entries from the liquidsoap user's crontab, then add ours.
-echo "$FETCH_CRON"   >> /tmp/cron_radio.$$
+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 "$UPDATE_CRON"  >> /tmp/cron_radio.$$
+echo "$FETCH_CRON"  >> /tmp/cron_ls_py.$$
-crontab /tmp/cron_radio.$$
+echo "$UPDATE_CRON" >> /tmp/cron_ls_py.$$
-rm -f /tmp/cron_radio.$$
+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

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
+Concurrency model mirrors fetch_podcasts.py:
-by scanning the filesystem, so it works identically for archived shows
+  - Shares the same exclusive lockfile (state/radio.lock). If the fetcher is
-(local file_path) and live shows (remote enclosure_url). All data under the
+    still running, this run logs a skip and exits 0 instead of hitting a
-storage path from config.json.
+    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/<slug>.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,52 +61,120 @@ 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("""
+    conn.execute("PRAGMA journal_mode=WAL")
-        CREATE TABLE IF NOT EXISTS shows (
+    conn.execute("PRAGMA busy_timeout=5000")
-            slug TEXT PRIMARY KEY,
+    return conn
-            guid TEXT NOT NULL UNIQUE,
+
-            name TEXT NOT NULL,
+def _ensure_columns(conn, table, columns):
-            feed_url TEXT NOT NULL UNIQUE,
+    exists = conn.execute(
-            source TEXT DEFAULT 'manual',
+        "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)
-            opml_import INTEGER DEFAULT 0,
+    ).fetchone()
-            archived INTEGER DEFAULT 1,
+    if exists is None:
-            created_at TEXT DEFAULT (datetime('now'))
+        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 = _connect(PLAYED_DB)
-    conn.row_factory = sqlite3.Row
+    _ensure_columns(conn, "episodes", EPISODES_COLUMNS)
-    conn.execute("""
+    conn.execute(
-        CREATE TABLE IF NOT EXISTS episodes (
+        "CREATE INDEX IF NOT EXISTS idx_episodes_show_played ON episodes (show_slug, played)"
-            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.commit()
     return conn
 
-def select_unplayed_episode(slug, played_db):
+# ---------------------------------------------------------------------------
-    """Pick the next unplayed episode for a show, keyed by guid.
+# Mutual exclusion (shared with fetch_podcasts.py via the same lockfile).
-    Archived -> local file_path; Live -> remote enclosure_url."""
+# ---------------------------------------------------------------------------
+_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",
@@ -103,72 +182,66 @@ def select_unplayed_episode(slug, played_db):
     ).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
+    updated = 0
-    skipped = 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)
+        out = write_queue_file(slug, ep, archived)
-        if line is None:
+        if out is None:
-            log.info("%s: episode %s has no usable URI, skipping.", slug, ep["guid"])
+            log.warning("No playable URI for %s (%s); skipping.", show["name"], slug)
-            skipped += 1
             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"
+        updated += 1
-        log.info("%s: queued %s [%s] runlength=%ss", slug, ep["title"], kind, int(ep["runlength"] or 0))
+        log.info("Queued %s: %s -> %s", slug, ep["title"], out.name)
-        queued += 1
     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()
+    shows = subs_db.execute("SELECT slug, name FROM shows ORDER BY name").fetchall()
-    summary = {}
+    result = {}
     for show in shows:
         slug = show["slug"]
-        total = played_db.execute(
+        counts = played_db.execute(
-            "SELECT COUNT(*) as c FROM episodes WHERE show_slug = ?", (slug,)
+            "SELECT COUNT(*) AS total, SUM(CASE WHEN played=0 THEN 1 ELSE 0 END) AS unplayed "
-        ).fetchone()["c"]
+            "FROM episodes WHERE show_slug=?",
-        played = played_db.execute(
+            (slug,),
-            "SELECT COUNT(*) as c FROM episodes WHERE show_slug = ? AND played = 1", (slug,)
+        ).fetchone()
-        ).fetchone()["c"]
+        total = counts["total"] or 0
-        summary[slug] = {"total_episodes": total, "played_count": played, "unplayed": total - played}
+        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()
 
+    # --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()