diff --git a/.gitignore b/.gitignore index 1a94786..af97cd2 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,5 @@ # Secrets and local configuration config.json -dirs.txt # Runtime data (should live under the storage path, not here) .venv/ state/ diff --git a/check_dirs.sh b/check_dirs.sh deleted file mode 100755 index d75ebd8..0000000 --- a/check_dirs.sh +++ /dev/null @@ -1,12 +0,0 @@ -#!/bin/bash -STORAGE="/mnt/storage/radio" -JINGLES="no" -ANNOUNCEMENTS="no" -if find "$STORAGE/jingles" -type f \( -name "*.mp3" -o -name "*.m4a" \) 2>/dev/null | grep -q .; then - JINGLES="yes" -fi -if find "$STORAGE/announcements" -type f \( -name "*.mp3" -o -name "*.m4a" \) 2>/dev/null | grep -q .; then - ANNOUNCEMENTS="yes" -fi -echo "jingles=$JINGLES" > /srv/radio/dirs.txt -echo "announcements=$ANNOUNCEMENTS" >> /srv/radio/dirs.txt diff --git a/fetch_podcasts.py b/fetch_podcasts.py index d6baa90..e143e96 100755 --- a/fetch_podcasts.py +++ b/fetch_podcasts.py @@ -7,16 +7,17 @@ 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 + - An exclusive, non-blocking lockfile (state/radio.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. - -Archived shows (archived=1) have episodes downloaded to disk; non-archived -shows (archived=0) store only the enclosure URL for live streaming. +Politeness / efficiency: + - The audio-vs-video verdict for each show is cached in subscriptions.db + (column media_class). Repeat runs reuse it instead of re-fetching every + feed just to re-confirm it is audio. + - The gpodder OPML pull uses bounded retry with exponential backoff + jitter + so transient failures or a 429 degrade gracefully rather than hammering. """ import argparse @@ -24,18 +25,31 @@ import fcntl import json import logging import os +import random import re import shutil import socket import sqlite3 import sys +import time import uuid 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). +# --- Force IPv4-only resolution --------------------------------------------- +# Box has no usable IPv6 route; gpodder.net publishes both A and AAAA records, +# so dual-stack getaddrinfo returns AAAA first and stalls. Constrain every +# AF_UNSPEC lookup to AF_INET before it reaches the resolver. +_original_getaddrinfo = socket.getaddrinfo + +def _ipv4_only_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): + if family in (0, socket.AF_UNSPEC): + family = socket.AF_INET + return _original_getaddrinfo(host, port, family, type, proto, flags) + +socket.getaddrinfo = _ipv4_only_getaddrinfo + try: import requests.packages.urllib3.util.connection as _urllib3_conn _urllib3_conn.HAS_IPV6 = False @@ -64,7 +78,6 @@ def load_config(): return json.load(f) 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, LOCK_FILE cfg = load_config() storage = Path(cfg["storage"]).expanduser().resolve() @@ -79,9 +92,7 @@ def init_paths(): logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", - handlers=[ - logging.StreamHandler(sys.stdout), - ], + handlers=[logging.StreamHandler(sys.stdout)], ) log = logging.getLogger("fetch_podcasts") @@ -89,7 +100,6 @@ def log_error(msg): log.error(msg) def _setup_logging(): - """Add file handler once LOGS_DIR is known.""" fh = logging.FileHandler(LOGS_DIR / "fetch.log") fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) log.addHandler(fh) @@ -100,8 +110,6 @@ def _setup_logging(): _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) @@ -110,7 +118,6 @@ def acquire_lock(): 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 @@ -136,6 +143,7 @@ SHOWS_COLUMNS = { "source": "TEXT DEFAULT 'manual'", "opml_import": "INTEGER DEFAULT 0", "archived": "INTEGER DEFAULT 1", + "media_class": "TEXT", "created_at": "TEXT DEFAULT (datetime('now'))", } @@ -159,15 +167,12 @@ def _connect(db_path): 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)" + extra = "\n ,UNIQUE(show_slug, guid)" if table == "episodes" else "" conn.execute(f"CREATE TABLE {table} (\n {defs}{extra}\n )") else: existing = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")} @@ -198,27 +203,44 @@ def slugify(name): def gen_uuid(): return str(uuid.uuid4()) -def is_audio_feed(parsed): - """Check whether the feed's latest episode has an audio enclosure. - Returns True if audio, False if video or unknown. - """ +def classify_feed(parsed): + """Return 'audio', 'video', or 'unknown' based on the latest enclosure.""" if not parsed.entries: - return True # No entries yet; let it through, will fail on fetch + return "unknown" entry = parsed.entries[0] enclosures = entry.get("enclosures") or [] if not enclosures: - return True # No enclosure info; assume audio + return "unknown" mime_type = (enclosures[0].get("type") or "").lower() if mime_type.startswith("audio/"): - return True + return "audio" if mime_type.startswith("video/"): - return False + return "video" url = (enclosures[0].get("href") or "").lower() if any(url.endswith(ext) for ext in AUDIO_EXTS): - return True + return "audio" if any(url.endswith(ext) for ext in VIDEO_EXTS): - return False - return True + return "video" + return "unknown" + +def get_media_class(subs_db, slug): + row = subs_db.execute("SELECT media_class FROM shows WHERE slug = ?", (slug,)).fetchone() + return row["media_class"] if row else None + +def set_media_class(subs_db, slug, cls): + subs_db.execute("UPDATE shows SET media_class = ? WHERE slug = ?", (cls, slug)) + subs_db.commit() + +def fetch_feed(feed_url): + try: + parsed = feedparser.parse(feed_url) + except Exception as e: + log_error(f"Feed parse error for {feed_url}: {e}") + return None + if parsed.bozo and not parsed.entries: + log_error(f"Bozo feed (no entries) for {feed_url}: {parsed.bozo_exception}") + return None + return parsed def gpodder_sync(cfg): g = cfg["gpodder"] @@ -230,12 +252,30 @@ def gpodder_sync(cfg): print(f"--- Syncing subscriptions from {base} ---") print(f"Fetching subscriptions for '{username}'...") - resp = requests.get( - url, - auth=(username, password), - headers={"User-Agent": "radio-automation/1.0"}, - timeout=60, - ) + # Bounded retry with exponential backoff + jitter. Retries on network + # errors and on 429/5xx; does not retry on 401/404 (auth/not-found). + max_attempts = 4 + backoff_base = 2.0 + resp = None + for attempt in range(1, max_attempts + 1): + try: + resp = requests.get( + url, + auth=(username, password), + headers={"User-Agent": "radio-automation/1.0"}, + timeout=60, + ) + break + except requests.RequestException as e: + if attempt == max_attempts: + log_error(f"gPodder sync failed after {attempt} attempts: {e}") + return [] + delay = backoff_base ** attempt + random.uniform(0, 1) + log.warning("gPodder request error (%s); retrying in %.1fs", e.__class__.__name__, delay) + time.sleep(delay) + + if resp is None: + return [] if resp.status_code == 200: body = resp.text @@ -247,8 +287,10 @@ def gpodder_sync(cfg): log_error("gPodder sync failed: 401 Unauthorized. Check username/password in config.json.") elif resp.status_code == 404: log_error("gPodder sync failed: 404 Not Found. User may not exist or has no subscriptions.") - elif resp.status_code == 400: - log_error("gPodder sync failed: 400 Bad Request.") + elif resp.status_code == 429: + log_error("gPodder sync throttled (429). Will retry next cycle.") + elif resp.status_code >= 500: + log_error(f"gPodder sync server error ({resp.status_code}). Will retry next cycle.") else: log_error(f"gPodder sync failed: unexpected response {resp.status_code}: {resp.text[:200]}") return [] @@ -266,11 +308,7 @@ def parse_opml(xml_string): guid = (outline.attrib.get("guid") or "").strip() if not feed_url or not re.match(r"^https?://", feed_url): continue - shows.append({ - "name": name, - "feed_url": feed_url, - "guid": guid if guid else None, - }) + shows.append({"name": name, "feed_url": feed_url, "guid": guid if guid else None}) return shows def register_remote_shows(remote_shows): @@ -282,20 +320,24 @@ def register_remote_shows(remote_shows): existing = db.execute("SELECT slug FROM shows WHERE slug = ?", (slug,)).fetchone() if existing is not None: continue + # Reuse a cached classification if we somehow already know it; otherwise + # fetch once to classify. New shows are fetched here regardless. parsed = fetch_feed(show["feed_url"]) if parsed is None: log.warning("Skipping '%s': could not fetch feed.", show["name"]) continue - if not is_audio_feed(parsed): + cls = classify_feed(parsed) + if cls == "video": log.info("Skipping '%s' (%s): video podcast, not audio.", show["name"], slug) skipped_video += 1 continue guid = show["guid"] or gen_uuid() db.execute( - "INSERT INTO shows (slug, guid, name, feed_url, source, opml_import, archived) VALUES (?, ?, ?, ?, 'gpodder', 0, 1)", - (slug, guid, show["name"], show["feed_url"]), + "INSERT INTO shows (slug, guid, name, feed_url, source, opml_import, archived, media_class) " + "VALUES (?, ?, ?, ?, 'gpodder', 0, 1, ?)", + (slug, guid, show["name"], show["feed_url"], cls), ) - log.info("Registered new show: %s (%s)", show["name"], slug) + log.info("Registered new show: %s (%s) [%s]", show["name"], slug, cls) added += 1 db.commit() db.close() @@ -343,17 +385,6 @@ def extract_duration(entry): pass return None -def fetch_feed(feed_url): - try: - parsed = feedparser.parse(feed_url) - except Exception as e: - log_error(f"Feed parse error for {feed_url}: {e}") - return None - if parsed.bozo and not parsed.entries: - log_error(f"Bozo feed (no entries) for {feed_url}: {parsed.bozo_exception}") - return None - return parsed - def download_episode(url, dest_dir, filename): dest = dest_dir / filename if dest.exists(): @@ -384,15 +415,33 @@ def show_archived(subs_db, slug): def fetch_show_episodes(slug, name, feed_url): dest_dir = PODCASTS_DIR / slug dest_dir.mkdir(parents=True, exist_ok=True) + subs_db = open_subs_db() + + # Respect a cached video classification without re-fetching. + cached_cls = get_media_class(subs_db, slug) + if cached_cls == "video": + subs_db.close() + return 0 + parsed = fetch_feed(feed_url) if parsed is None: + subs_db.close() return 0 + + # Classify and cache if we don't already have a verdict. + if cached_cls is None: + cls = classify_feed(parsed) + set_media_class(subs_db, slug, cls) + if cls == "video": + log.info("'%s' (%s) classified as video; skipping.", name, slug) + subs_db.close() + return 0 + played_db = open_played_db() seen = { row["guid"] for row in played_db.execute("SELECT guid FROM episodes WHERE show_slug = ?", (slug,)) } - subs_db = open_subs_db() archived = show_archived(subs_db, slug) new_count = 0 for entry in parsed.entries: @@ -452,16 +501,17 @@ def fetch_all_episodes(): def list_shows(detail=False): db = open_subs_db() rows = db.execute( - "SELECT slug, name, feed_url, source, opml_import, archived FROM shows ORDER BY name" + "SELECT slug, name, feed_url, source, opml_import, archived, media_class FROM shows ORDER BY name" ).fetchall() db.close() if not rows: print("No shows registered.") return - print(f"{'SLUG':<30} {'ARCHIVED':<10} {'SOURCE':<10} NAME") + print(f"{'SLUG':<30} {'ARCHIVED':<10} {'MEDIA':<8} {'SOURCE':<10} NAME") for r in rows: arch = "yes" if r["archived"] == 1 else "no" - line = f"{r['slug']:<30} {arch:<10} {r['source']:<10} {r['name']}" + media = r["media_class"] or "?" + line = f"{r['slug']:<30} {arch:<10} {media:<8} {r['source']:<10} {r['name']}" if detail: line += f"\n{'':<50} {r['feed_url']}" print(line) @@ -471,7 +521,8 @@ def add_show(feed_url): if parsed is None or not parsed.feed.get("title"): log_error(f"Could not determine show title from {feed_url}") return - if not is_audio_feed(parsed): + cls = classify_feed(parsed) + if cls == "video": log_error(f"Refusing to add '{parsed.feed['title']}': video podcast detected.") return name = parsed.feed["title"] @@ -479,12 +530,13 @@ def add_show(feed_url): guid = gen_uuid() db = open_subs_db() db.execute( - "INSERT OR IGNORE INTO shows (slug, guid, name, feed_url, source, opml_import, archived) VALUES (?, ?, ?, ?, 'manual', 0, 1)", - (slug, guid, name, feed_url), + "INSERT OR IGNORE INTO shows (slug, guid, name, feed_url, source, opml_import, archived, media_class) " + "VALUES (?, ?, ?, ?, 'manual', 0, 1, ?)", + (slug, guid, name, feed_url, cls), ) db.commit() db.close() - log.info("Added show: %s (%s)", name, slug) + log.info("Added show: %s (%s) [%s]", name, slug, cls) fetch_show_episodes(slug, name, feed_url) def set_archive(slug, value): @@ -545,14 +597,16 @@ def import_opml(path): if parsed is None: log.warning("OPML import: skipping '%s', could not fetch feed.", show["name"]) continue - if not is_audio_feed(parsed): + cls = classify_feed(parsed) + if cls == "video": log.info("OPML import: skipping '%s' (%s): video podcast.", show["name"], slug) skipped_video += 1 continue guid = show["guid"] or gen_uuid() db.execute( - "INSERT INTO shows (slug, guid, name, feed_url, source, opml_import, archived) VALUES (?, ?, ?, ?, 'opml', 1, 1)", - (slug, guid, show["name"], show["feed_url"]), + "INSERT INTO shows (slug, guid, name, feed_url, source, opml_import, archived, media_class) " + "VALUES (?, ?, ?, ?, 'opml', 1, 1, ?)", + (slug, guid, show["name"], show["feed_url"], cls), ) added += 1 db.commit() @@ -591,7 +645,6 @@ 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(): @@ -619,4 +672,3 @@ def main(): if __name__ == "__main__": main() - diff --git a/fetch_podcasts.rb b/fetch_podcasts.rb index b8631f1..d86ad8d 100755 --- a/fetch_podcasts.rb +++ b/fetch_podcasts.rb @@ -1,603 +1,714 @@ #!/usr/bin/env jruby # frozen_string_literal: true +# +# fetch_podcasts.rb - Podcast subscription management and episode fetching +# for the liquidsoap radio automation stack (JRuby/Sequel variant). +# +# Concurrency: exclusive non-blocking File.lock (state/radio.lock) shared with +# update_playlists.rb; skips cleanly if the sibling holds it. Databases run in +# WAL mode with a busy timeout as a second safety net. +# +# Politeness: audio/video verdict cached in shows.media_class; gpodder OPML +# pull uses bounded retry with exponential backoff + jitter. -require "net/http" -require "uri" -require "cgi" -require "json" -require "jdbc/sqlite3" -require "rexml/document" -require "digest/md5" -require "securerandom" -require "fileutils" -require "optparse" +require 'sequel' +require 'net/http' +require 'openssl' +require 'json' +require 'logger' +require 'digest/md5' +require 'time' -Jdbc::SQLite3.load_driver +ROOT = File.expand_path(File.dirname(__file__)) +CONFIG_PATH = File.join(ROOT, "config.json") -module RadioAutomation - ROOT = File.expand_path("..", __dir__) - CONFIG_PATH = File.join(ROOT, "config.json") - AUDIO_EXTS = [".mp3", ".m4a"] - VIDEO_EXTS = [".mp4", ".mov", ".avi", ".webm", ".mkv"] +AUDIO_EXTS = [".mp3", ".m4a"].freeze +VIDEO_EXTS = [".mp4", ".mov", ".avi", ".webm", ".mkv"].freeze - STORAGE_DIR = nil - STATE_DIR = nil - SUBS_DB = nil - PLAYED_DB = nil - PODCASTS_DIR = nil - LOGS_DIR = nil - PLAYLISTS_DIR = nil +$state_dir = nil +$subs_db_path = nil +$played_db_path = nil +$podcasts_dir = nil +$logs_dir = nil +$playlists_dir = nil +$lock_file = nil - def self.init_paths - cfg = JSON.parse(File.read(CONFIG_PATH)) - @storage = File.expand_path(cfg["storage"]) - self.STORAGE_DIR = @storage - self.STATE_DIR = File.join(@storage, "state") - self.SUBS_DB = File.join(STATE_DIR, "subscriptions.db") - self.PLAYED_DB = File.join(STATE_DIR, "played.db") - self.PODCASTS_DIR = File.join(@storage, "podcasts") - self.LOGS_DIR = File.join(@storage, "logs") - self.PLAYLISTS_DIR = File.join(@storage, "playlists") +def load_config + JSON.parse(File.read(CONFIG_PATH)) +end + +def init_paths! + cfg = load_config + storage = File.realpath(cfg["storage"]) + $state_dir = File.join(storage, "state") + $subs_db_path = File.join($state_dir, "subscriptions.db") + $played_db_path = File.join($state_dir, "played.db") + $podcasts_dir = File.join(storage, "podcasts") + $logs_dir = File.join(storage, "logs") + $playlists_dir = File.join(storage, "playlists") + $lock_file = File.join($state_dir, "radio.lock") +end + +$log = Logger.new(STDOUT) +$log.formatter = proc { |msg, _severity, _time, _progname| "#{Time.now} [INFO] #{msg}\n" } + +def log_error(msg) + $log.error(msg) +end + +def setup_logging! + fh = Logger.new(File.join($logs_dir, "fetch.log")) + fh.formatter = $log.formatter + $log.instance_variable_set(:@logdev, + Logger::LogDevice.new([STDOUT, File.join($logs_dir, "fetch.log")])) +end + +# --------------------------------------------------------------------------- +# Mutual exclusion via flock on a shared lockfile. +# --------------------------------------------------------------------------- +$lock_fh = nil + +def acquire_lock! + Dir.mkdir($state_dir) unless Dir.exist?($state_dir) + fh = File.open($lock_file, File::RDWR | File::CREAT, 0o644) + begin + fh.flock(File::LOCK_EX | File::LOCK_NB) + rescue Errno::EACCES, Errno::EAGAIN + fh.close + return false end + fh.truncate(0) + fh.write(Process.pid.to_s) + fh.rewind + $lock_fh = fh + true +end - # --- JDBC connection helpers --------------------------------------------- - - def self.jdb_connect(db_file) - java.sql.DriverManager.getConnection("jdbc:sqlite:#{db_file}") +def release_lock! + return if $lock_fh.nil? + begin + $lock_fh.flock(File::LOCK_UN) + $lock_fh.close + ensure + $lock_fh = nil end +end - def self.jdb_query(conn, sql, params = []) - stmt = conn.prepareStatement(sql) - params.each_with_index { |p, i| stmt.setObject(i + 1, p) } - rs = stmt.executeQuery - cols = [] - meta = rs.getMetaData - (1..meta.getColumnCount).each { |i| cols << meta.getColumnName(i) } - rows = [] - while rs.next - row = {} - cols.each { |c| row[c] = rs.getObject(c) } - rows << row - end - rs.close - stmt.close - rows +# --------------------------------------------------------------------------- +# Schema: single source of truth, applied idempotently via Sequel. +# --------------------------------------------------------------------------- +SHOWS_COLUMNS = { + slug: { type: :string, primary_key: true }, + guid: { type: :string, null: false, unique: true }, + name: { type: :string, null: false }, + feed_url: { type: :string, null: false, unique: true }, + source: { type: :string, default: "manual" }, + opml_import: { type: :integer, default: 0 }, + archived: { type: :integer, default: 1 }, + media_class: { type: :string }, + created_at: { type: :string, default: Sequel.function(:datetime, "'now'") } +}.freeze + +EPISODES_COLUMNS = { + id: { type: :integer, primary_key: true, auto_increment: true }, + show_slug: { type: :string, null: false }, + guid: { type: :string, null: false }, + title: { type: :string }, + file_path: { type: :string }, + enclosure_url: { type: :string }, + runlength: { type: :integer }, + played: { type: :integer, default: 0 }, + played_at: { type: :string } +}.freeze + +def connect_subs + db = Sequel.connect("jdbc:sqlite:#{$subs_db_path}") + db.extension :pragma + db.pragma journal_mode: :wal + db.pragma busy_timeout: 5000 + ensure_schema!(db, :shows, SHOWS_COLUMNS) + db +end + +def connect_played + db = Sequel.connect("jdbc:sqlite:#{$played_db_path}") + db.extension :pragma + db.pragma journal_mode: :wal + db.pragma busy_timeout: 5000 + ensure_schema!(db, :episodes, EPISODES_COLUMNS) + unless db.index_exists?(:episodes, [:show_slug, :played]) + db.create_index :episodes, [:show_slug, :played], name: :idx_episodes_show_played end + db +end - def self.jdb_exec(conn, sql, params = []) - stmt = conn.prepareStatement(sql) - params.each_with_index { |p, i| stmt.setObject(i + 1, p) } - stmt.executeUpdate - stmt.close - end - - def self.log_info(msg) - puts "#{Time.now.iso8601} [INFO] #{msg}" - append_log("fetch.log", msg) - end - - def self.log_warning(msg) - puts "#{Time.now.iso8601} [WARN] #{msg}" - append_log("fetch.log", msg) - end - - def self.log_error(msg) - puts "#{Time.now.iso8601} [ERROR] #{msg}" - append_log("fetch.log", msg) - end - - def self.append_log(filename, msg) - FileUtils.mkdir_p(LOGS_DIR) - File.open(File.join(LOGS_DIR, filename), "a") { |f| f.puts msg } - rescue StandardError - nil - end - - def self.load_config - JSON.parse(File.read(CONFIG_PATH)) - end - - def self.slugify(name) - s = name.downcase.gsub(/[^a-z0-9]+/, "_").gsub(/\A_+|_+\z/, "") - s[0, 60] || "show" - end - - def self.gen_uuid - SecureRandom.uuid - end - - def self.open_subs_db - db = jdb_connect(SUBS_DB) - jdb_exec(db, <<-SQL) - 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')) - ) - SQL - db - end - - def self.open_played_db - db = jdb_connect(PLAYED_DB) - jdb_exec(db, <<-SQL) - 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) - ) - SQL - db - end - - # --- Audio-only filtering ------------------------------------------------- - - def self.enclosure_mime_and_url(entry_xml) - enc_m = entry_xml.match(/enclosure[^>]*type="([^"]*)"[^>]*url="([^"]*)"/i) || - entry_xml.match(/enclosure[^>]*url="([^"]*)"[^>]*type="([^"]*)"/i) - if enc_m - if enc_m.pre_match.include?("type=") && enc_m.post_match.empty? - return [enc_m[1], enc_m[2]] - else - return [enc_m[2], enc_m[1]] +def ensure_schema!(db, table, columns) + unless db.table_exists?(table) + db.create_table(table) do |t| + columns.each do |col, opts| + t.column(col, **opts) end + t.unique_constraint %i[show_slug guid] if table == :episodes end - url_only = entry_xml.match(/enclosure[^>]*url="([^"]*)"/i) - [nil, url_only[1]] if url_only + return + end + existing = db.columns(table) + columns.each do |col, opts| + next if existing.include?(col) + db.alter_table(table) { |t| t.add_column(col, **opts) } + end +end + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- +def slugify(name) + s = name.downcase.gsub(/[^a-z0-9]+/, "_").gsub(/\A_+|_+\z/, "") + s = "show" if s.empty? + s[0, 60] +end + +def gen_uuid + require "securerandom" + SecureRandom.uuid +end + +def classify_feed(entries) + return "unknown" if entries.nil? || entries.empty? + entry = entries.first + enclosures = entry[:enclosures] || [] + return "unknown" if enclosures.empty? + mime = (enclosures.first[:type] || "").downcase + return "audio" if mime.start_with?("audio/") + return "video" if mime.start_with?("video/") + url = (enclosures.first[:href] || "").downcase + AUDIO_EXTS.any? { |ext| url.end_with?(ext) } && return "audio" + VIDEO_EXTS.any? { |ext| url.end_with?(ext) } && return "video" + "unknown" +end + +def get_media_class(db, slug) + row = db[:shows].where(slug: slug).first + row ? row[:media_class] : nil +end + +def set_media_class(db, slug, cls) + db[:shows].where(slug: slug).update(media_class: cls) +end + +# --------------------------------------------------------------------------- +# Feed parsing (uses open-uri / rss-lite approach via Net::HTTP + REXML) +# --------------------------------------------------------------------------- +require "rexml/document" + +def fetch_feed(feed_url) + uri = URI.parse(feed_url) + http = Net::HTTP.new(uri.host, uri.port) + http.use_ssl = (uri.scheme == "https") + http.timeout = 60 + resp = http.request(Net::HTTP::Get.new(uri.request_uri)) + return nil unless resp.is_a?(Net::HTTPSuccess) + + doc = REXML::Document.new(resp.body) + channel = doc.elements["//channel"] + return nil if channel.nil? + + title = channel.elements["title"]&.text + entries = [] + channel.get_elements("./item").each do |item| + link_el = item.elements["link"] + title_el = item.elements["title"] + guid_el = item.elements["guid"] + dur_el = item.elements["media:duration"] || item.elements["itunes:duration"] + enc_el = item.elements["enclosure"] + iso_dur_el = item.elements["itunes:duration"] + + enclosures = [] + if enc_el + enclosures << { + href: enc_el.attributes["url"], + type: enc_el.attributes["type"], + length: enc_el.attributes["length"] + } + end + + entries << { + link: link_el&.text, + title: title_el&.text, + guid: guid_el&.text, + enclosures: enclosures, + duration: dur_el&.text, + iso_duration: iso_dur_el&.text + } end - def self.is_audio_entry?(entry_xml) - mime, url = enclosure_mime_and_url(entry_xml) - return true if mime.nil? && url.nil? - mime_l = mime.to_s.downcase - return true if mime_l.start_with?("audio/") - return false if mime_l.start_with?("video/") - url_l = url.to_s.downcase - return true if AUDIO_EXTS.any? { |ext| url_l.end_with?(ext) } - return false if VIDEO_EXTS.any? { |ext| url_l.end_with?(ext) } - true - end + { title: title, entries: entries } +rescue StandardError => e + log_error("Feed parse error for #{feed_url}: #{e.message}") + nil +end - def self.is_audio_feed?(raw_xml) - m = raw_xml.match(/<(?:item|entry)[^>]*>(.*?)<\/(?:item|entry)>/mi) - return true unless m - is_audio_entry?(m[1]) - end +# --------------------------------------------------------------------------- +# gPodder sync with bounded retry + exponential backoff + jitter +# --------------------------------------------------------------------------- +def gpodder_sync(cfg) + g = cfg["gpodder"] + base = g["host"].chomp("/") + username = g["username"] + password = g["password"] + path = "/subscriptions/#{CGI.escape(username)}.opml" - # --- End audio-only filtering --------------------------------------------- + puts "--- Syncing subscriptions from #{base} ---" + puts "Fetching subscriptions for '#{username}'..." - def self.gpodder_sync(config) - g = config["gpodder"] - base = g["host"].chomp("/") - username = g["username"] - password = g["password"] - url = "#{base}/subscriptions/#{CGI.escape(username)}.opml" + max_attempts = 4 + backoff_base = 2.0 + resp = nil - puts "--- Syncing subscriptions from #{base} ---" - puts "Fetching subscriptions for '#{username}'..." - - uri = URI.parse(url) - http = Net::HTTP.new(uri.host, uri.port) - http.use_ssl = (uri.scheme == "https") - http.open_timeout = 30 - http.read_timeout = 60 - - req = Net::HTTP::Get.new(uri) - req.basic_auth(username, password) - req["Accept"] = "application/x-opml, text/xml, */*" - req["User-Agent"] = "radio-automation/1.0" - - resp = http.request(req) - - case resp.code - when "200" - body = resp.body - if body.nil? || body.empty? - log_error("gPodder sync returned an empty body.") + max_attempts.times do |attempt| + begin + uri = URI.parse("#{base}#{path}") + http = Net::HTTP.new(uri.host, uri.port) + http.use_ssl = (uri.scheme == "https") + http.timeout = 60 + req = Net::HTTP::Get.new(uri.request_uri) + credentials = Base64.strict_encode64("#{username}:#{password}") + req["Authorization"] = "Basic #{credentials}" + req["User-Agent"] = "radio-automation/1.0" + resp = http.request(req) + break + rescue StandardError => e + if attempt == max_attempts - 1 + log_error("gPodder sync failed after #{attempt + 1} attempts: #{e.class.name}") return [] end - parse_opml(body) - when "401" - log_error("gPodder sync failed: 401 Unauthorized. Check username/password in config.json.") - [] - when "404" - log_error("gPodder sync failed: 404 Not Found. User may not exist or has no subscriptions.") - [] - when "400" - log_error("gPodder sync failed: 400 Bad Request.") - [] + delay = (backoff_base ** (attempt + 1)) + rand + $log.warn("gPodder request error (#{e.class.name}); retrying in #{delay.round(1)}s") + sleep(delay) + end + end + + return [] if resp.nil? + + case resp.code.to_i + when 200 + body = resp.body + return [] if body.strip.empty? + parse_opml(body) + when 401 + log_error("gPodder sync failed: 401 Unauthorized. Check username/password in config.json.") + when 404 + log_error("gPodder sync failed: 404 Not Found. User may not exist or has no subscriptions.") + when 429 + log_error("gPodder sync throttled (429). Will retry next cycle.") + else + code = resp.code.to_i + if code >= 500 + log_error("gPodder sync server error (#{code}). Will retry next cycle.") else - log_error("gPodder sync failed: unexpected response #{resp.code}: #{resp.body.to_s[0..200]}") - [] + log_error("gPodder sync failed: unexpected response #{code}: #{resp.body[0, 200]}") end end + [] +end - def self.parse_opml(xml_string) - doc = REXML::Document.new(xml_string) - shows = [] - REXML::XPath.each(doc, "//outline[@xmlUrl]") do |node| - feed_url = node.attributes["xmlUrl"].to_s.strip - name = node.attributes["text"].to_s.strip - guid = node.attributes["guid"].to_s.strip - next unless feed_url =~ /\Ahttps?:\/\// - shows << { "name" => name, "feed_url" => feed_url, "guid" => (guid.empty? ? nil : guid) } - end - shows - rescue REXML::ParseException => e - log_error("Failed to parse OPML XML: #{e.message}") - [] +def parse_opml(xml_string) + shows = [] + doc = REXML::Document.new(xml_string) + doc.elements.each("//outline") do |outline| + feed_url = (outline.attributes["xmlUrl"] || "").strip + name = (outline.attributes["text"] || "").strip + guid = (outline.attributes["guid"] || "").strip + next unless feed_url.match?(/\Ahttps?:\/\//) + shows << { name: name, feed_url: feed_url, guid: guid.empty? ? nil : guid } end + shows +rescue REXML::ParseException => e + log_error("Failed to parse OPML XML: #{e.message}") + [] +end - def self.register_remote_shows(remote_shows) - db = open_subs_db - added = 0 - skipped_video = 0 - remote_shows.each do |show| - slug = slugify(show["name"]) - existing = jdb_query(db, "SELECT slug FROM shows WHERE slug = ?", [slug]) - next unless existing.empty? +# --------------------------------------------------------------------------- +# Show registration / pruning +# --------------------------------------------------------------------------- +def register_remote_shows(remote_shows) + db = connect_subs + added = 0 + skipped_video = 0 + remote_shows.each do |show| + slug = slugify(show[:name]) + next if db[:shows].where(slug: slug).count > 0 - raw = fetch_feed(show["feed_url"]) - if raw.nil? - log_warning("Skipping '#{show['name']}': could not fetch feed.") - next - end - unless is_audio_feed?(raw) - log_info("Skipping '#{show['name']}' (#{slug}): video podcast, not audio.") - skipped_video += 1 - next - end - guid = show["guid"] || gen_uuid - jdb_exec( - db, - "INSERT INTO shows (slug, guid, name, feed_url, source, opml_import, archived) VALUES (?, ?, ?, ?, 'gpodder', 0, 1)", - [slug, guid, show["name"], show["feed_url"]] - ) - log_info("Registered new show: #{show['name']} (#{slug})") - added += 1 + parsed = fetch_feed(show[:feed_url]) + if parsed.nil? + $log.warn("Skipping '#{show[:name]}': could not fetch feed.") + next end - db.close - log_info("Filtered out #{skipped_video} video podcast(s).") if skipped_video > 0 - added - end - - def self.prune_stale_shows(remote_shows) - db = open_subs_db - remote_slugs = remote_shows.map { |s| slugify(s["name"]) } - stale = jdb_query(db, "SELECT slug, name FROM shows WHERE source = 'gpodder' AND opml_import = 0") - removed = 0 - stale.each do |row| - next if remote_slugs.include?(row["slug"]) - remove_show_data(row["slug"]) - jdb_exec(db, "DELETE FROM shows WHERE slug = ?", [row["slug"]]) - log_info("Pruned stale show: #{row['name']} (#{row['slug']})") - removed += 1 + cls = classify_feed(parsed[:entries]) + if cls == "video" + $log.info("Skipping '#{show[:name]}' (#{slug}): video podcast, not audio.") + skipped_video += 1 + next end - db.close - removed - end - - def self.extract_duration(entry_xml) - if (m = entry_xml.match(/media:duration[^>]*content="([^"]+)"/)) - val = m[1] - return val.to_i if val =~ /\A\d+\z/ - if (iso = val.match(/\APT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?/)) - h, mn, s = iso.captures.compact.map(&:to_i) - return (h || 0) * 3600 + (mn || 0) * 60 + (s || 0) - end - end - if (m = entry_xml.match(/enclosure[^>]*length="(\d+)"/)) - bytes = m[1].to_i - return (bytes * 8 / 128_000) if bytes > 0 - end - nil - end - - def self.fetch_feed(feed_url) - uri = URI.parse(feed_url) - http = Net::HTTP.new(uri.host, uri.port) - http.use_ssl = (uri.scheme == "https") - http.open_timeout = 30 - http.read_timeout = 60 - req = Net::HTTP::Get.new(uri) - req["User-Agent"] = "radio-automation/1.0" - resp = http.request(req) - raise "Feed HTTP #{resp.code}" unless resp.is_a?(Net::HTTPSuccess) - resp.body - rescue StandardError => e - log_error("Feed fetch error for #{feed_url}: #{e.message}") - nil - end - - def self.download_episode(url, dest_dir, filename) - FileUtils.mkdir_p(dest_dir) - dest = File.join(dest_dir, filename) - return dest if File.exist?(dest) - - uri = URI.parse(url) - http = Net::HTTP.new(uri.host, uri.port) - http.use_ssl = (uri.scheme == "https") - http.open_timeout = 30 - http.read_timeout = 120 - req = Net::HTTP::Get.new(uri) - req["User-Agent"] = "radio-automation/1.0" - - tmp = "#{dest}.part" - begin - http.request(req) do |resp| - raise "Download HTTP #{resp.code}" unless resp.is_a?(Net::HTTPSuccess) - File.open(tmp, "wb") do |f| - resp.read_body { |chunk| f.write(chunk) } - end - end - File.rename(tmp, dest) - dest - rescue StandardError => e - log_error("Download failed for #{url}: #{e.message}") - File.delete(tmp) if File.exist?(tmp) - nil - end - end - - def self.safe_filename(title, fallback) - name = title.to_s.gsub(/[^\w\s.\-]/, "").strip.tr(" ", "_")[0, 120] - "#{name || fallback}.mp3" - end - - def self.show_archived?(subs_db, slug) - rows = jdb_query(subs_db, "SELECT archived FROM shows WHERE slug = ?", [slug]) - rows.empty? ? true : (rows.first["archived"] == 1) - end - - def self.fetch_show_episodes(slug, name, feed_url) - dest_dir = File.join(PODCASTS_DIR, slug) - raw = fetch_feed(feed_url) - return 0 if raw.nil? - - played_db = open_played_db - seen_rows = jdb_query(played_db, "SELECT guid FROM episodes WHERE show_slug = ?", [slug]) - seen = seen_rows.map { |r| r["guid"] } - subs_db = open_subs_db - archived = show_archived?(subs_db, slug) - new_count = 0 - - raw.scan(/<(?:item|entry)[^>]*>(.*?)<\/(?:item|entry)>/mi).each do |(entry_xml)| - guid_m = entry_xml.match(/]*>([^<]*)<\/guid>|([^<]*)<\/id>|]*href="([^"]+)"/i) - guid = guid_m ? (guid_m[1] || guid_m[2] || guid_m[3]).strip : Digest::MD5.hexdigest(entry_xml[0, 200]) - next if seen.include?(guid) - - unless is_audio_entry?(entry_xml) - next - end - - _, audio_url = enclosure_mime_and_url(entry_xml) - next if audio_url.nil? - - title_m = entry_xml.match(/]*>([^<]*)<\/title>/i) - title = title_m ? title_m[1].strip : "untitled" - duration = extract_duration(entry_xml) - - if archived - filename = safe_filename(title, guid[-20..]) - file_path = download_episode(audio_url, dest_dir, filename) - next if file_path.nil? - jdb_exec( - played_db, - "INSERT OR IGNORE INTO episodes (show_slug, guid, title, file_path, enclosure_url, runlength, played) VALUES (?, ?, ?, ?, ?, ?, 0)", - [slug, guid, title, file_path, audio_url, duration] - ) - else - jdb_exec( - played_db, - "INSERT OR IGNORE INTO episodes (show_slug, guid, title, file_path, enclosure_url, runlength, played) VALUES (?, ?, ?, NULL, ?, ?, 0)", - [slug, guid, title, audio_url, duration] - ) - end - new_count += 1 - log_info(" New episode: #{title} [#{archived ? 'downloaded' : 'live'}]") - end - - played_db.close - subs_db.close - new_count - end - - def self.fetch_all_episodes - db = open_subs_db - shows = jdb_query(db, "SELECT slug, name, feed_url FROM shows ORDER BY name") - db.close - total_new = 0 - shows.each do |show| - log_info("--- Fetching: #{show['name']} (#{show['slug']}) ---") - begin - total_new += fetch_show_episodes(show["slug"], show["name"], show["feed_url"]) - rescue StandardError => e - log_error("Unexpected error fetching #{show['slug']}: #{e.message}") - end - end - log_info("=== Fetch complete: #{total_new} new episode(s) ===") - end - - def self.list_shows(detail: false) - db = open_subs_db - rows = jdb_query(db, "SELECT slug, name, feed_url, source, opml_import, archived FROM shows ORDER BY name") - db.close - if rows.empty? - puts "No shows registered." - return - end - puts format("%-30s %-10s %-8s %s", "SLUG", "ARCHIVED", "SOURCE", "NAME") - rows.each do |r| - arch = r["archived"] == 1 ? "yes" : "no" - line = format("%-30s %-10s %-8s %s", r["slug"], arch, r["source"], r["name"]) - line += "\n" + (" " * 50) + r["feed_url"] if detail - puts line - end - end - - def self.add_show(feed_url) - raw = fetch_feed(feed_url) - if raw.nil? - log_error("Could not fetch feed: #{feed_url}") - return - end - unless is_audio_feed?(raw) - log_error("Refusing to add: video podcast detected at #{feed_url}") - return - end - title_m = raw.match(/]*>\s*]*>([^<]*)<\/title>/im) || - raw.match(/]*>\s*]*>([^<]*)<\/title>/im) - if title_m.nil? - log_error("Could not determine show title from #{feed_url}") - return - end - name = title_m[1].strip - slug = slugify(name) - guid = gen_uuid - db = open_subs_db - jdb_exec( - db, - "INSERT OR IGNORE INTO shows (slug, guid, name, feed_url, source, opml_import, archived) VALUES (?, ?, ?, ?, 'manual', 0, 1)", - [slug, guid, name, feed_url] + guid = show[:guid] || gen_uuid + db[:shows].insert( + slug: slug, guid: guid, name: show[:name], feed_url: show[:feed_url], + source: "gpodder", opml_import: 0, archived: 1, media_class: cls ) - db.close - log_info("Added show: #{name} (#{slug})") - fetch_show_episodes(slug, name, feed_url) + $log.info("Registered new show: #{show[:name]} (#{slug}) [#{cls}]") + added += 1 end + db.disconnect + $log.info("Filtered out #{skipped_video} video podcast(s).") if skipped_video > 0 + added +end - def self.set_archive(slug, value) - db = open_subs_db - rows = jdb_query(db, "SELECT name FROM shows WHERE slug = ?", [slug]) - if rows.empty? - log_error("No show found with slug '#{slug}'.") - return +def prune_stale_shows(remote_shows) + db = connect_subs + remote_slugs = remote_shows.map { |s| slugify(s[:name]) }.to_set + stale = db[:shows].where(source: "gpodder", opml_import: 0).all + removed = 0 + stale.each do |row| + next if remote_slugs.include?(row[:slug]) + remove_show_data(row[:slug]) + db[:shows].where(slug: row[:slug]).delete + $log.info("Pruned stale show: #{row[:name]} (#{row[:slug]})") + removed += 1 + end + db.disconnect + removed +end + +# --------------------------------------------------------------------------- +# Episode extraction / download +# --------------------------------------------------------------------------- +def extract_duration(entry) + dur = entry[:duration] + if dur + return dur.to_i if dur.match?(/\A\d+\z/) + m = dur.match(/\APT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?\z/) + if m + h = m[1] ? m[1].to_i : 0 + mn = m[2] ? m[2].to_i : 0 + s = m[3] ? m[3].to_i : 0 + return h * 3600 + mn * 60 + s end - jdb_exec(db, "UPDATE shows SET archived = ? WHERE slug = ?", [value, slug]) - db.close - state = value == 1 ? "archived" : "non-archived (live)" - log_info("Show '#{rows.first['name']}' (#{slug}) is now #{state}.") end - - def self.remove_show_data(slug) - pod_dir = File.join(PODCASTS_DIR, slug) - FileUtils.rm_rf(pod_dir) if Dir.exist?(pod_dir) - txt = File.join(PLAYLISTS_DIR, "#{slug}.txt") - File.delete(txt) if File.exist?(txt) - end - - def self.delete_show(slug) - db = open_subs_db - rows = jdb_query(db, "SELECT name FROM shows WHERE slug = ?", [slug]) - if rows.empty? - log_error("No show found with slug '#{slug}'.") - return + iso = entry[:iso_duration] + if iso + m = iso.match(/\APT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?\z/) + if m + h = m[1] ? m[1].to_i : 0 + mn = m[2] ? m[2].to_i : 0 + s = m[3] ? m[3].to_i : 0 + return h * 3600 + mn * 60 + s end - remove_show_data(slug) - jdb_exec(db, "DELETE FROM shows WHERE slug = ?", [slug]) - db.close - played_db = open_played_db - jdb_exec(played_db, "DELETE FROM episodes WHERE show_slug = ?", [slug]) - played_db.close - log_info("Deleted show: #{rows.first['name']} (#{slug})") + end + enc = entry[:enclosures]&.first + if enc && enc[:length] + bytes = enc[:length].to_i + return (bytes * 8 / 128_000) if bytes > 0 + end + nil +end + +def download_episode(url, dest_dir, filename) + dest = File.join(dest_dir, filename) + return dest if File.exist?(dest) + uri = URI.parse(url) + http = Net::HTTP.new(uri.host, uri.port) + http.use_ssl = (uri.scheme == "https") + http.timeout = 120 + tmp = "#{dest}.part" + begin + http.request_get(uri.request_uri) do |response| + raise "HTTP #{response.code}" unless response.is_a?(Net::HTTPSuccess) + File.open(tmp, "wb") do |f| + response.read_body { |chunk| f.write(chunk) } + end + end + File.rename(tmp, dest) + dest + rescue StandardError => e + log_error("Download failed for #{url}: #{e.message}") + File.delete(tmp) if File.exist?(tmp) + nil + end +end + +def safe_filename(title, fallback) + name = (title || "").gsub(/[^\w\s.\-]/, "").strip.tr(" ", "_") + name = fallback if name.empty? + "#{name[0, 120]}.mp3" +end + +def show_archived?(db, slug) + row = db[:shows].where(slug: slug).first + row.nil? ? true : row[:archived] == 1 +end + +# --------------------------------------------------------------------------- +# Per-show episode fetch +# --------------------------------------------------------------------------- +def fetch_show_episodes(slug, name, feed_url) + dest_dir = File.join($podcasts_dir, slug) + Dir.mkdir(dest_dir) unless Dir.exist?(dest_dir) + + subs_db = connect_subs + + cached_cls = get_media_class(subs_db, slug) + if cached_cls == "video" + subs_db.disconnect + return 0 end - def self.import_opml(path) - content = File.read(path) - shows = parse_opml(content) - db = open_subs_db - added = 0 - skipped_video = 0 - shows.each do |show| - slug = slugify(show["name"]) - existing = jdb_query(db, "SELECT slug FROM shows WHERE slug = ?", [slug]) - next unless existing.empty? + parsed = fetch_feed(feed_url) + if parsed.nil? + subs_db.disconnect + return 0 + end - raw = fetch_feed(show["feed_url"]) - if raw.nil? - log_warning("OPML import: skipping '#{show['name']}', could not fetch feed.") - next - end - unless is_audio_feed?(raw) - log_info("OPML import: skipping '#{show['name']}' (#{slug}): video podcast.") - skipped_video += 1 - next - end - guid = show["guid"] || gen_uuid - jdb_exec( - db, - "INSERT INTO shows (slug, guid, name, feed_url, source, opml_import, archived) VALUES (?, ?, ?, ?, 'opml', 1, 1)", - [slug, guid, show["name"], show["feed_url"]] + if cached_cls.nil? + cls = classify_feed(parsed[:entries]) + set_media_class(subs_db, slug, cls) + if cls == "video" + $log.info("'#{name}' (#{slug}) classified as video; skipping.") + subs_db.disconnect + return 0 + end + end + + played_db = connect_played + seen = played_db[:episodes].where(show_slug: slug).select_map(:guid).to_set + archived = show_archived?(subs_db, slug) + new_count = 0 + + parsed[:entries].each do |entry| + guid = entry[:guid] || entry[:link] || entry[:title] || "" + next if seen.include?(guid) + enclosures = entry[:enclosures] || [] + next if enclosures.empty? + mime = (enclosures.first[:type] || "").downcase + next if mime.start_with?("video/") + audio_url = enclosures.first[:href] + next if audio_url.nil? || audio_url.empty? + + title = entry[:title] || "untitled" + duration = extract_duration(entry) + + if archived + filename = safe_filename(title, guid[-20..]) + file_path = download_episode(audio_url, dest_dir, filename) + next if file_path.nil? + played_db[:episodes].insert_or_ignore( + show_slug: slug, guid: guid, title: title, + file_path: file_path, enclosure_url: audio_url, + runlength: duration, played: 0 ) - added += 1 - end - db.close - log_info("OPML import: #{added} added, #{skipped_video} video shows filtered out.") - end - - def self.run_fetch(config) - g = config["gpodder"] - if g["enable"] == true - remote = gpodder_sync(config) - if remote.empty? - log_info("No subscriptions retrieved from gPodder; using local registry only.") - else - added = register_remote_shows(remote) - pruned = prune_stale_shows(remote) - log_info("Sync: #{added} added, #{pruned} pruned.") - end - end - fetch_all_episodes - end - - def self.main - options = {} - OptionParser.new do |opts| - opts.banner = "Usage: fetch_podcasts.rb [options]" - opts.on("--list-shows", "List registered shows") { options[:list] = true } - opts.on("--detail", "With --list-shows, show feed URLs") { options[:detail] = true } - opts.on("--add-show FEED_URL", "Add a show from a feed URL") { |v| options[:add] = v } - opts.on("--delete-show SLUG", "Delete a show and its data") { |v| options[:delete] = v } - opts.on("--archive SLUG", "Mark a show as archived (download episodes)") { |v| options[:archive] = v } - opts.on("--unarchive SLUG", "Mark a show as non-archived (stream live)") { |v| options[:unarchive] = v } - opts.on("--import-opml FILE", "Import shows from an OPML file") { |v| options[:import] = v } - end.parse! - - init_paths - FileUtils.mkdir_p(STATE_DIR) - FileUtils.mkdir_p(LOGS_DIR) - FileUtils.mkdir_p(PODCASTS_DIR) - FileUtils.mkdir_p(PLAYLISTS_DIR) - - config = load_config - - if options[:archive] - set_archive(options[:archive], 1) - elsif options[:unarchive] - set_archive(options[:unarchive], 0) - elsif options[:list] - list_shows(detail: options[:detail]) - elsif options[:add] - add_show(options[:add]) - elsif options[:delete] - delete_show(options[:delete]) - elsif options[:import] - import_opml(options[:import]) else + played_db[:episodes].insert_or_ignore( + show_slug: slug, guid: guid, title: title, + file_path: nil, enclosure_url: audio_url, + runlength: duration, played: 0 + ) + end + new_count += 1 + kind = archived ? "downloaded" : "live" + $log.info(" New episode: #{title} [#{kind}]") + end + + played_db.disconnect + subs_db.disconnect + new_count +end + +def fetch_all_episodes + db = connect_subs + shows = db[:shows].order(:name).all + db.disconnect + total_new = 0 + shows.each do |show| + $log.info("--- Fetching: #{show[:name]} (#{show[:slug]}) ---") + begin + n = fetch_show_episodes(show[:slug], show[:name], show[:feed_url]) + total_new += n + rescue StandardError => e + log_error("Unexpected error fetching #{show[:slug]}: #{e.message}") + end + end + $log.info("=== Fetch complete: #{total_new} new episode(s) ===") +end + +# --------------------------------------------------------------------------- +# Administrative commands +# --------------------------------------------------------------------------- +def list_shows(detail: false) + db = connect_subs + rows = db[:shows].order(:name).all + db.disconnect + if rows.empty? + puts "No shows registered." + return + end + puts format("%-30s %-10s %-8s %-10s %s", "SLUG", "ARCHIVED", "MEDIA", "SOURCE", "NAME") + rows.each do |r| + arch = r[:archived] == 1 ? "yes" : "no" + media = r[:media_class] || "?" + line = format("%-30s %-10s %-8s %-10s %s", r[:slug], arch, media, r[:source], r[:name]) + line += "\n" + (" " * 50) + r[:feed_url] if detail + puts line + end +end + +def add_show(feed_url) + parsed = fetch_feed(feed_url) + if parsed.nil? || parsed[:title].nil? + log_error("Could not determine show title from #{feed_url}") + return + end + cls = classify_feed(parsed[:entries]) + if cls == "video" + log_error("Refusing to add '#{parsed[:title]}': video podcast detected.") + return + end + name = parsed[:title] + slug = slugify(name) + guid = gen_uuid + db = connect_subs + db[:shows].insert_or_ignore( + slug: slug, guid: guid, name: name, feed_url: feed_url, + source: "manual", opml_import: 0, archived: 1, media_class: cls + ) + db.disconnect + $log.info("Added show: #{name} (#{slug}) [#{cls}]") + fetch_show_episodes(slug, name, feed_url) +end + +def set_archive(slug, value) + db = connect_subs + row = db[:shows].where(slug: slug).first + if row.nil? + log_error("No show found with slug '#{slug}'.") + db.disconnect + return + end + db[:shows].where(slug: slug).update(archived: value) + db.disconnect + state = value == 1 ? "archived" : "non-archived (live)" + $log.info("Show '#{row[:name]}' (#{slug}) is now #{state}.") +end + +def remove_show_data(slug) + pod_dir = File.join($podcasts_dir, slug) + FileUtils.rm_rf(pod_dir) if Dir.exist?(pod_dir) + txt = File.join($playlists_dir, "#{slug}.txt") + File.delete(txt) if File.exist?(txt) +end + +def delete_show(slug) + db = connect_subs + row = db[:shows].where(slug: slug).first + if row.nil? + log_error("No show found with slug '#{slug}'.") + db.disconnect + return + end + remove_show_data(slug) + db[:shows].where(slug: slug).delete + db.disconnect + played_db = connect_played + played_db[:episodes].where(show_slug: slug).delete + played_db.disconnect + $log.info("Deleted show: #{row[:name]} (#{slug})") +end + +def import_opml(path) + content = File.read(path) + shows = parse_opml(content) + db = connect_subs + added = 0 + skipped_video = 0 + shows.each do |show| + slug = slugify(show[:name]) + next if db[:shows].where(slug: slug).count > 0 + parsed = fetch_feed(show[:feed_url]) + if parsed.nil? + $log.warn("OPML import: skipping '#{show[:name]}', could not fetch feed.") + next + end + cls = classify_feed(parsed[:entries]) + if cls == "video" + $log.info("OPML import: skipping '#{show[:name]}' (#{slug}): video podcast.") + skipped_video += 1 + next + end + guid = show[:guid] || gen_uuid + db[:shows].insert( + slug: slug, guid: guid, name: show[:name], feed_url: show[:feed_url], + source: "opml", opml_import: 1, archived: 1, media_class: cls + ) + added += 1 + end + db.disconnect + $log.info("OPML import: #{added} added, #{skipped_video} video shows filtered out.") +end + +def run_fetch(config) + g = config["gpodder"] + if g["enable"] == true + remote = gpodder_sync(config) + if remote.empty? + $log.warn("No subscriptions retrieved from gPodder; using local registry only.") + else + added = register_remote_shows(remote) + pruned = prune_stale_shows(remote) + $log.info("Sync: #{added} added, #{pruned} pruned.") + end + end + fetch_all_episodes +end + +require "fileutils" +require "cgi" +require "base64" + +def main + args = ARGV.dup + option = args.shift + + init_paths! + [$state_dir, $logs_dir, $podcasts_dir, $playlists_dir].each do |dir| + Dir.mkdir(dir) unless Dir.exist?(dir) + end + setup_logging! + + config = load_config + + case option + when "--list-shows" + list_shows(detail: args.include?("--detail")) + when "--add-show" + add_show(args.first) + when "--delete-show" + delete_show(args.first) + when "--archive" + set_archive(args.first, 1) + when "--unarchive" + set_archive(args.first, 0) + when "--import-opml" + import_opml(args.first) + else + needs_lock = true + if needs_lock && !acquire_lock! + $log.info("Another radio process holds the lock; skipping this run.") + return + end + begin run_fetch(config) + ensure + release_lock! if needs_lock end end end -RadioAutomation.main +main if __FILE__ == $PROGRAM_NAME diff --git a/install_for_jruby b/install_for_jruby index 401f4aa..2a031ec 100755 --- a/install_for_jruby +++ b/install_for_jruby @@ -92,7 +92,8 @@ for tool in jruby gem bundle rake irb; do done # --------------------------------------------------------------------------- -# 5. Gems (one at a time, raised heap) +# 5. Gems (one at a time, raised heap). Order matters: jdbc-sqlite3 first so +# it's present when sequel loads its JDBC SQLite subadapter. # --------------------------------------------------------------------------- export GEM_HOME="$GEMS_DIR" export GEM_PATH="$GEMS_DIR" @@ -100,6 +101,7 @@ export JRUBY_OPTS="-J-Xmx1g -J-Xss512k" 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 sequel "$JRUBY_BIN" -S gem install --no-document json # Give the service user ownership of the gem cache so cron runs work. @@ -205,49 +207,61 @@ chown "$LIQUIDSOAP_USER":"$LIQUIDSOAP_USER" "$CONFIG_JSON" # --------------------------------------------------------------------------- # 9. Initialize BOTH SQLite databases with the full current schema. -# Uses the jdbc-sqlite3 gem (SQLite JDBC driver), matching how the Ruby -# scripts access the DB at runtime. Idempotent via IF NOT EXISTS. +# Uses Sequel over jdbc-sqlite3, matching how the Ruby scripts access the +# DB at runtime. Idempotent via IF NOT EXISTS semantics. # --------------------------------------------------------------------------- -echo "==> Initializing SQLite databases via jdbc-sqlite3 ..." -GEM_HOME="$GEMS_DIR" GEM_PATH="$GEMS_DIR" "$JRUBY_BIN" -e " -require 'jdbc/sqlite3' -require 'java' +echo "==> Initializing SQLite databases via Sequel/jdbc-sqlite3 ..." +GEM_HOME="$GEMS_DIR" GEM_PATH="$GEMS_DIR" "$JRUBY_BIN" -e ' +require "sequel" + state_dir = ARGV[0] -JDBC::Database.new(\"jdbc:sqlite:\#{state_dir}/subscriptions.db\") do |db| - db.execute(%Q{ - 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')) - ) - }) +subs_path = File.join(state_dir, "subscriptions.db") +played_path = File.join(state_dir, "played.db") + +db = Sequel.connect("jdbc:sqlite:#{subs_path}") +db.extension :pragma +db.pragma journal_mode: :wal +db.pragma busy_timeout: 5000 +unless db.table_exists?(:shows) + db.create_table(:shows) do |t| + t.primary_key :slug, type: :string + t.string :guid, null: false, unique: true + t.string :name, null: false + t.string :feed_url, null: false, unique: true + t.string :source, default: "manual" + t.integer :opml_import, default: 0 + t.integer :archived, default: 1 + t.string :media_class + t.string :created_at + end end -JDBC::Database.new(\"jdbc:sqlite:\#{state_dir}/played.db\") do |db| - db.execute(%Q{ - 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) - ) - }) - db.execute(%Q{ - CREATE INDEX IF NOT EXISTS idx_episodes_show_played ON episodes (show_slug, played) - }) +db.disconnect + +db = Sequel.connect("jdbc:sqlite:#{played_path}") +db.extension :pragma +db.pragma journal_mode: :wal +db.pragma busy_timeout: 5000 +unless db.table_exists?(:episodes) + db.create_table(:episodes) do |t| + t.primary_key :id + t.string :show_slug, null: false + t.string :guid, null: false + t.string :title + t.string :file_path + t.string :enclosure_url + t.integer :runlength + t.integer :played, default: 0 + t.string :played_at + t.unique_constraint %i[show_slug guid] + end end -puts ' subscriptions.db and played.db initialized.' -" "$STORAGE_PATH/state" +unless db.index_exists?(:episodes, [:show_slug, :played]) + db.create_index(:episodes, [:show_slug, :played], name: :idx_episodes_show_played) +end +db.disconnect + +puts " subscriptions.db and played.db initialized." +' "$STORAGE_PATH/state" chown -R "$LIQUIDSOAP_USER":"$LIQUIDSOAP_USER" "$STORAGE_PATH/state" # --------------------------------------------------------------------------- diff --git a/update_playlists.py b/update_playlists.py index 7fb2047..217bdba 100755 --- a/update_playlists.py +++ b/update_playlists.py @@ -3,11 +3,9 @@ update_playlists.py - Select the next unplayed episode per show and write an annotated URI queue file for station.liq to consume. -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. +Concurrency model mirrors fetch_podcasts.py: shares the same exclusive +lockfile (state/radio.lock); skips cleanly if the fetcher holds it. 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. @@ -22,7 +20,6 @@ import logging import os import sqlite3 import sys -from datetime import datetime from pathlib import Path ROOT = Path(__file__).resolve().parent @@ -55,9 +52,7 @@ def init_paths(): logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", - handlers=[ - logging.StreamHandler(sys.stdout), - ], + handlers=[logging.StreamHandler(sys.stdout)], ) log = logging.getLogger("update_playlists") @@ -69,9 +64,7 @@ def _setup_logging(): fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) log.addHandler(fh) -# --------------------------------------------------------------------------- # Shared schema definitions (must match fetch_podcasts.py). -# --------------------------------------------------------------------------- SHOWS_COLUMNS = { "slug": "TEXT PRIMARY KEY", "guid": "TEXT NOT NULL UNIQUE", @@ -80,6 +73,7 @@ SHOWS_COLUMNS = { "source": "TEXT DEFAULT 'manual'", "opml_import": "INTEGER DEFAULT 0", "archived": "INTEGER DEFAULT 1", + "media_class": "TEXT", "created_at": "TEXT DEFAULT (datetime('now'))", } @@ -108,9 +102,7 @@ def _ensure_columns(conn, table, columns): ).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)" + extra = "\n ,UNIQUE(show_slug, guid)" if table == "episodes" else "" conn.execute(f"CREATE TABLE {table} (\n {defs}{extra}\n )") else: existing = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")} @@ -134,9 +126,6 @@ def open_played_db(): conn.commit() return conn -# --------------------------------------------------------------------------- -# Mutual exclusion (shared with fetch_podcasts.py via the same lockfile). -# --------------------------------------------------------------------------- _lock_fd = None def acquire_lock(): @@ -162,39 +151,28 @@ def release_lock(): 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( + return 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", (slug,), ).fetchone() - return row 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"] + uri = ep["file_path"] if archived else ep["enclosure_url"] if not uri: return None line = _annotate_uri(ep["runlength"], ep["title"], uri) @@ -254,7 +232,6 @@ 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(): diff --git a/update_playlists.rb b/update_playlists.rb index e628050..f4889bb 100755 --- a/update_playlists.rb +++ b/update_playlists.rb @@ -1,219 +1,244 @@ #!/usr/bin/env jruby # frozen_string_literal: true +# +# update_playlists.rb - Select the next unplayed episode per show and write an +# annotated URI queue file for station.liq to consume (JRuby/Sequel variant). +# +# Shares the same exclusive lockfile as fetch_podcasts.rb; skips cleanly if +# the fetcher holds it. Databases run in WAL mode with a busy timeout. +require "sequel" require "json" -require "jdbc/sqlite3" -require "fileutils" -require "optparse" +require "logger" +require "time" -Jdbc::SQLite3.load_driver +ROOT = File.expand_path(File.dirname(__file__)) +CONFIG_PATH = File.join(ROOT, "config.json") -module RadioAutomation - ROOT = File.expand_path("..", __dir__) - CONFIG_PATH = File.join(ROOT, "config.json") +$state_dir = nil +$subs_db_path = nil +$played_db_path = nil +$podcasts_dir = nil +$logs_dir = nil +$playlists_dir = nil +$lock_file = nil - STORAGE_DIR = nil - STATE_DIR = nil - SUBS_DB = nil - PLAYED_DB = nil - PODCASTS_DIR = nil - PLAYLISTS_DIR = nil - LOGS_DIR = nil +def load_config + JSON.parse(File.read(CONFIG_PATH)) +end - def self.init_paths - cfg = JSON.parse(File.read(CONFIG_PATH)) - @storage = File.expand_path(cfg["storage"]) - self.STORAGE_DIR = @storage - self.STATE_DIR = File.join(@storage, "state") - self.SUBS_DB = File.join(STATE_DIR, "subscriptions.db") - self.PLAYED_DB = File.join(STATE_DIR, "played.db") - self.PODCASTS_DIR = File.join(@storage, "podcasts") - self.PLAYLISTS_DIR = File.join(@storage, "playlists") - self.LOGS_DIR = File.join(@storage, "logs") +def init_paths! + cfg = load_config + storage = File.realpath(cfg["storage"]) + $state_dir = File.join(storage, "state") + $subs_db_path = File.join($state_dir, "subscriptions.db") + $played_db_path = File.join($state_dir, "played.db") + $podcasts_dir = File.join(storage, "podcasts") + $logs_dir = File.join(storage, "logs") + $playlists_dir = File.join(storage, "playlists") + $lock_file = File.join($state_dir, "radio.lock") +end + +$log = Logger.new(STDOUT) +$log.formatter = proc { |msg, _sev, _time, _prog| "#{Time.now} [INFO] #{msg}\n" } + +def log_error(msg) + $log.error(msg) +end + +def setup_logging! + $log.instance_variable_set(:@logdev, + Logger::LogDevice.new([STDOUT, File.join($logs_dir, "update.log")])) +end + +SHOWS_COLUMNS = { + slug: { type: :string, primary_key: true }, + guid: { type: :string, null: false, unique: true }, + name: { type: :string, null: false }, + feed_url: { type: :string, null: false, unique: true }, + source: { type: :string, default: "manual" }, + opml_import: { type: :integer, default: 0 }, + archived: { type: :integer, default: 1 }, + media_class: { type: :string }, + created_at: { type: :string, default: Sequel.function(:datetime, "'now'") } +}.freeze + +EPISODES_COLUMNS = { + id: { type: :integer, primary_key: true, auto_increment: true }, + show_slug: { type: :string, null: false }, + guid: { type: :string, null: false }, + title: { type: :string }, + file_path: { type: :string }, + enclosure_url: { type: :string }, + runlength: { type: :integer }, + played: { type: :integer, default: 0 }, + played_at: { type: :string } +}.freeze + +def connect_subs + db = Sequel.connect("jdbc:sqlite:#{$subs_db_path}") + db.extension :pragma + db.pragma journal_mode: :wal + db.pragma busy_timeout: 5000 + ensure_schema!(db, :shows, SHOWS_COLUMNS) + db +end + +def connect_played + db = Sequel.connect("jdbc:sqlite:#{$played_db_path}") + db.extension :pragma + db.pragma journal_mode: :wal + db.pragma busy_timeout: 5000 + ensure_schema!(db, :episodes, EPISODES_COLUMNS) + unless db.index_exists?(:episodes, [:show_slug, :played]) + db.create_index :episodes, [:show_slug, :played], name: :idx_episodes_show_played end + db +end - # --- JDBC connection helpers --------------------------------------------- - - def self.jdb_connect(db_file) - java.sql.DriverManager.getConnection("jdbc:sqlite:#{db_file}") - end - - def self.jdb_query(conn, sql, params = []) - stmt = conn.prepareStatement(sql) - params.each_with_index { |p, i| stmt.setObject(i + 1, p) } - rs = stmt.executeQuery - cols = [] - meta = rs.getMetaData - (1..meta.getColumnCount).each { |i| cols << meta.getColumnName(i) } - rows = [] - while rs.next - row = {} - cols.each { |c| row[c] = rs.getObject(c) } - rows << row +def ensure_schema!(db, table, columns) + unless db.table_exists?(table) + db.create_table(table) do |t| + columns.each { |col, opts| t.column(col, **opts) } + t.unique_constraint %i[show_slug guid] if table == :episodes end - rs.close - stmt.close - rows + return end - - def self.jdb_exec(conn, sql, params = []) - stmt = conn.prepareStatement(sql) - params.each_with_index { |p, i| stmt.setObject(i + 1, p) } - stmt.executeUpdate - stmt.close + existing = db.columns(table) + columns.each do |col, opts| + next if existing.include?(col) + db.alter_table(table) { |t| t.add_column(col, **opts) } end +end - def self.log_info(msg) - puts "#{Time.now.iso8601} [INFO] #{msg}" - append_log("update.log", msg) +$lock_fh = nil + +def acquire_lock! + Dir.mkdir($state_dir) unless Dir.exist?($state_dir) + fh = File.open($lock_file, File::RDWR | File::CREAT, 0o644) + begin + fh.flock(File::LOCK_EX | File::LOCK_NB) + rescue Errno::EACCES, Errno::EAGAIN + fh.close + return false end + fh.truncate(0) + fh.write(Process.pid.to_s) + fh.rewind + $lock_fh = fh + true +end - def self.append_log(filename, msg) - FileUtils.mkdir_p(LOGS_DIR) - File.open(File.join(LOGS_DIR, filename), "a") { |f| f.puts msg } - rescue StandardError - nil +def release_lock! + return if $lock_fh.nil? + begin + $lock_fh.flock(File::LOCK_UN) + $lock_fh.close + ensure + $lock_fh = nil end +end - def self.open_subs_db - db = jdb_connect(SUBS_DB) - jdb_exec(db, <<-SQL) - 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')) - ) - SQL - db +def annotate_uri(runlength, title, uri) + def esc(v) + '"' + v.to_s.gsub('"', '\\"') + '"' end + "annotate:liq_runlength=#{esc(runlength)},liq_title=#{esc(title)}:" + uri +end - def self.open_played_db - db = jdb_connect(PLAYED_DB) - jdb_exec(db, <<-SQL) - 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) - ) - SQL - db - end +def select_next_episode(slug, played_db) + played_db[:episodes] + .where(show_slug: slug, played: 0) + .order(:id.asc) + .limit(1) + .first +end - # Pick the next unplayed episode for a show, keyed by guid. - # Archived: prefer a local file_path; Live: use enclosure_url. - def self.select_unplayed_episode(slug, played_db) - rows = jdb_query( - played_db, - "SELECT guid, title, file_path, enclosure_url, runlength FROM episodes WHERE show_slug = ? AND played = 0 ORDER BY id ASC LIMIT 1", - [slug] - ) - return nil if rows.empty? - rows.first - end +def mark_as_played(slug, guid, played_db) + played_db[:episodes] + .where(show_slug: slug, guid: guid) + .update(played: 1, played_at: Sequel.function(:datetime, "'now'")) +end - # Build the annotated URI line station.liq consumes. - # Archived -> local file path; Live -> remote enclosure URL. - def self.annotated_uri(ep) - uri = ep["file_path"] || ep["enclosure_url"] - return nil if uri.nil? || uri.to_s.empty? - rl = ep["runlength"].to_i - title = ep["title"].to_s.gsub('"', "'") - "annotate:liq_runlength=\"#{rl}\",liq_title=\"#{title}\":#{uri}" - end +def write_queue_file(slug, ep, archived) + uri = archived ? ep[:file_path] : ep[:enclosure_url] + return nil if uri.nil? || uri.empty? + line = annotate_uri(ep[:runlength], ep[:title], uri) + out = File.join($playlists_dir, "#{slug}.txt") + File.write(out, line + "\n") + out +end - def self.write_queue_line(slug, line, out_path) - FileUtils.mkdir_p(File.dirname(out_path)) - File.open(out_path, "w") { |f| f.puts(line) } - end - - def self.mark_as_played(slug, guid, played_db) - jdb_exec( - played_db, - "UPDATE episodes SET played = 1, played_at = datetime('now') WHERE show_slug = ? AND guid = ?", - [slug, guid] - ) - end - - def self.update_all - subs_db = open_subs_db - played_db = open_played_db - shows = jdb_query(subs_db, "SELECT slug, name, archived FROM shows ORDER BY name") - - queued = 0 - skipped = 0 - shows.each do |show| - slug = show["slug"] - ep = select_unplayed_episode(slug, played_db) - if ep.nil? - log_info("#{slug}: no unplayed episodes, skipping.") - skipped += 1 - next - end - line = annotated_uri(ep) - if line.nil? - log_info("#{slug}: episode #{ep['guid']} has no usable URI, skipping.") - skipped += 1 - next - end - out_pls = File.join(PLAYLISTS_DIR, "#{slug}.txt") - write_queue_line(slug, line, out_pls) - mark_as_played(slug, ep["guid"], played_db) - kind = ep["file_path"] ? "downloaded" : "live" - log_info("#{slug}: queued #{ep['title']} [#{kind}] runlength=#{ep['runlength'].to_i}s") - queued += 1 +def update_all + subs_db = connect_subs + played_db = connect_played + shows = subs_db[:shows].order(:name).all + updated = 0 + shows.each do |show| + slug = show[:slug] + archived = show[:archived] == 1 + ep = select_next_episode(slug, played_db) + next if ep.nil? + out = write_queue_file(slug, ep, archived) + if out.nil? + $log.warn("No playable URI for #{show[:name]} (#{slug}); skipping.") + next end - - subs_db.close - played_db.close - log_info("=== Update complete: #{queued} queued, #{skipped} skipped ===") + mark_as_played(slug, ep[:guid], played_db) + updated += 1 + $log.info("Queued #{slug}: #{ep[:title]} -> #{File.basename(out)}") end + subs_db.disconnect + played_db.disconnect + $log.info("=== Update complete: #{updated} show(s) queued ===") +end - def self.json_summary - subs_db = open_subs_db - played_db = open_played_db - shows = jdb_query(subs_db, "SELECT slug FROM shows ORDER BY name") - summary = {} - shows.each do |show| - slug = show["slug"] - total = jdb_query(played_db, "SELECT COUNT(*) AS c FROM episodes WHERE show_slug = ?", [slug]).first["c"].to_i - played = jdb_query(played_db, "SELECT COUNT(*) AS c FROM episodes WHERE show_slug = ? AND played = 1", [slug]).first["c"].to_i - summary[slug] = { "total_episodes" => total, "played_count" => played, "unplayed" => total - played } +def json_summary + subs_db = connect_subs + played_db = connect_played + shows = subs_db[:shows].order(:name).all + result = {} + shows.each do |show| + slug = show[:slug] + counts = played_db[:episodes].where(show_slug: slug).hash_and_count + total = counts.values.sum + unplayed = played_db[:episodes].where(show_slug: slug, played: 0).count + result[slug] = { + name: show[:name], + total: total, + unplayed: unplayed, + played: total - unplayed + } + end + subs_db.disconnect + played_db.disconnect + puts JSON.pretty_generate(result) +end + +def main + args = ARGV.dup + json_mode = args.delete("--json") + + init_paths! + [$state_dir, $logs_dir, $playlists_dir].each do |dir| + Dir.mkdir(dir) unless Dir.exist?(dir) + end + setup_logging! + + if json_mode + json_summary + else + if !acquire_lock! + $log.info("Another radio process holds the lock; skipping this run.") + return end - subs_db.close - played_db.close - puts JSON.pretty_generate(summary) - end - - def self.main - options = {} - OptionParser.new do |opts| - opts.banner = "Usage: update_playlists.rb [options]" - opts.on("--json", "Emit JSON summary and exit") { options[:json] = true } - end.parse! - - init_paths - FileUtils.mkdir_p(STATE_DIR) - FileUtils.mkdir_p(LOGS_DIR) - FileUtils.mkdir_p(PLAYLISTS_DIR) - - if options[:json] - json_summary - else + begin update_all + ensure + release_lock! end end end -RadioAutomation.main +main if __FILE__ == $PROGRAM_NAME +