From 4e185b2e1c7893978ee6db11cda7be4abbca12ae Mon Sep 17 00:00:00 2001 From: Media Date: Thu, 6 Aug 2026 18:58:05 -0500 Subject: [PATCH] refactor: wire controller v2 into web runtime --- app.py | 157 ++++++-- controller_core.py | 920 ++++----------------------------------------- gunicorn.conf.py | 25 ++ tests/test_app.py | 316 ++++++++++++++++ wsgi.py | 22 ++ 5 files changed, 557 insertions(+), 883 deletions(-) create mode 100644 gunicorn.conf.py create mode 100644 tests/test_app.py create mode 100644 wsgi.py diff --git a/app.py b/app.py index 6ff9232..7926c57 100644 --- a/app.py +++ b/app.py @@ -1,10 +1,17 @@ -from flask import Flask, render_template_string, request, redirect, url_for, flash +from __future__ import annotations + +import logging import os +from urllib.parse import urlsplit -from controller_core import enqueue_download, start_controller, stop_controller +from flask import Flask, flash, jsonify, redirect, render_template_string, request, url_for -app = Flask(__name__) -app.secret_key = os.environ.get("FLASK_SECRET_KEY", "dev-secret") +from controller.config import ControllerConfig, load_config +from controller.models import StoreError +from controller.service import ControllerService, ControllerUnavailableError + + +LOG = logging.getLogger(__name__) HTML_FORM = """ @@ -217,41 +224,125 @@ HTML_FORM = """ """ -@app.route("/", methods=["GET"]) -def index(): - return render_template_string(HTML_FORM) +def _valid_media_url(value: str) -> bool: + try: + parsed = urlsplit(value) + except ValueError: + return False + return parsed.scheme.lower() in {"http", "https"} and bool(parsed.netloc) -@app.route("/submit", methods=["POST"]) -def submit(): - url = request.form.get("url", "").strip() - submitter = request.form.get("submitter", "").strip() or None - note = request.form.get("note", "").strip() or None +def _submission_message(result: dict[str, object]) -> str: + status = result.get("status") + if status == "queue_full": + seconds = max(0, int(result.get("oldest_expires_in_seconds") or 0)) + hours = max(1, (seconds + 3599) // 3600) + return f"Queue is full. Oldest playlist expires in about {hours} hour(s)." + if status == "slot_busy": + return f"Slot {result.get('slot')} is already being downloaded." + if status == "slot_protected": + return ( + f"Slot {result.get('slot')} is protected by active or uncertain playback; " + "the playlist was not replaced." + ) + if status == "shutting_down": + return "The Controller is shutting down; please try again shortly." + if status == "worker_error": + return "The playlist was reserved, but its download worker could not start." + if status != "queued": + return "The Controller could not queue that playlist; please try again." - if not url.startswith("http"): - flash("Invalid URL.") + slot = result.get("slot") + message = f"Playlist queued in slot {slot}." + if result.get("reused_slot"): + message += " (Existing slot – only new videos will be added.)" + else: + message += " It will start downloading shortly." + return message + + +def create_app( + config: ControllerConfig, + service: ControllerService, + *, + testing: bool = False, +) -> Flask: + """Create the HTTP surface without starting MQTT, threads, or persistence.""" + + secret_key = os.environ.get("FLASK_SECRET_KEY") + if not secret_key and not testing: + raise RuntimeError("FLASK_SECRET_KEY must be set for the Controller web service") + + web_app = Flask(__name__) + web_app.config.update( + TESTING=testing, + SECRET_KEY=secret_key or "controller-test-only-secret", + ) + web_app.extensions["controller_service"] = service + web_app.extensions["controller_config"] = config + + @web_app.get("/") + def index(): + return render_template_string(HTML_FORM) + + @web_app.post("/submit") + def submit(): + url = request.form.get("url", "").strip() + submitter = request.form.get("submitter", "").strip() or None + note = request.form.get("note", "").strip() or None + + if not _valid_media_url(url): + flash("Invalid URL. Use a complete http:// or https:// address.") + return redirect(url_for("index")) + + try: + result = service.enqueue_download(url, submitter, note) + except ControllerUnavailableError as exc: + LOG.warning("playlist submission was rejected: %s", exc) + flash("The Controller is unavailable; please try again shortly.") + except StoreError as exc: + LOG.error("playlist submission hit a storage error: %s", exc) + flash("Controller storage is unavailable; the playlist was not queued.") + except ValueError as exc: + LOG.warning("playlist submission was rejected: %s", exc) + flash(f"The playlist request was rejected: {exc}") + except Exception: + LOG.exception("playlist submission failed unexpectedly") + flash("The Controller could not queue that playlist; please try again.") + else: + flash(_submission_message(result)) return redirect(url_for("index")) - result = enqueue_download(url, submitter, note) + @web_app.get("/healthz") + def healthz(): + try: + health = service.health() + except Exception as exc: + LOG.exception("Controller health check failed") + return jsonify( + { + "status": "unavailable", + "started": False, + "coordinator_alive": False, + "error": str(exc), + } + ), 503 + serving = bool(health.get("started") and health.get("coordinator_alive")) + return jsonify(health), 200 if serving else 503 - if result.get("status") == "queue_full": - secs = result.get("oldest_expires_in_seconds", 0) - hours = max(1, secs // 3600) - flash(f"Queue is full. Oldest playlist expires in about {hours} hour(s).") - else: - slot = result.get("slot") - msg = f"Playlist queued in slot {slot}." - if result.get("reused_slot"): - msg += " (Existing slot – only new videos will be added.)" - else: - msg += " It will start downloading shortly." - flash(msg) + return web_app + + +def _run_development_server() -> None: + config = load_config() + service = ControllerService(config) + web_app = create_app(config, service) + service.start() + try: + web_app.run(host=config.web.host, port=config.web.port) + finally: + service.stop() - return redirect(url_for("index")) if __name__ == "__main__": - start_controller() - try: - app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7531))) - finally: - stop_controller() + _run_development_server() diff --git a/controller_core.py b/controller_core.py index a20c944..8b66407 100644 --- a/controller_core.py +++ b/controller_core.py @@ -1,864 +1,84 @@ -import os -import time -import json +"""Deprecated, lazy compatibility facade for the Controller v2 service. + +New deployments use :mod:`wsgi`. These functions remain for local callers +that imported the original module, but importing this module never opens the +database, creates directories, starts threads, or connects to MQTT. +""" + +from __future__ import annotations + import threading -import subprocess -from pathlib import Path -from dataclasses import dataclass, asdict -from typing import Dict, List, Optional +from typing import Any -import yaml -import paho.mqtt.client as mqtt +from controller.config import ControllerConfig, load_config +from controller.service import ControllerService, ControllerUnavailableError -from controller.config import load_config -# --- Config / paths --- +_lifecycle_lock = threading.Lock() +_service: ControllerService | None = None +_config: ControllerConfig | None = None -CONFIG = load_config() - -MEDIA_BASE = CONFIG.storage.media_base -DATA_BASE = CONFIG.storage.data_base -STATE_PATH = DATA_BASE / "playlists.yaml" - -MAX_SLOTS = len(CONFIG.slot_ids) -RETENTION_SECONDS = CONFIG.storage.retention_seconds - -MQTT_HOST = CONFIG.mqtt.host -MQTT_PORT = CONFIG.mqtt.port -MQTT_CLIENT_ID = CONFIG.mqtt.client_id - -TOPIC_INGEST_PROGRESS = CONFIG.topics.ingest_progress -TOPIC_PLAYBACK_BASE = CONFIG.topics.playback_base -TOPIC_PLAYER_STATUS = CONFIG.topics.player_status -TOPIC_PLAYER_CMD = CONFIG.topics.player_command -TOPIC_CONTROLLER_CMD = CONFIG.topics.controller_command - -# --- MQTT client --- - -import logging, sys -logging.basicConfig(stream=sys.stdout, level=logging.INFO) - -def _on_connect(client, userdata, flags, rc, properties=None): - print(f"[mqtt] connected rc={rc}") - if rc == 0: - client.subscribe(TOPIC_PLAYER_STATUS, qos=1) - client.subscribe(TOPIC_CONTROLLER_CMD, qos=1) - print(f"[controller] subscribed to {TOPIC_PLAYER_STATUS} and {TOPIC_CONTROLLER_CMD}") - if CONFIG.homeassistant.enabled: - ha_discovery_publish() - reannounce_slots() - else: - print(f"[mqtt] connect failed rc={rc}") - -def _on_message(client, userdata, msg): - if msg.topic == TOPIC_PLAYER_STATUS: - try: - data = json.loads(msg.payload.decode("utf-8") or "{}") - except Exception: - return - - state_str = data.get("state") - slot_val = data.get("slot") - - if not state_str or slot_val is None: - return - - try: - slot_num = int(slot_val) - except (TypeError, ValueError): - return - - print(f"[controller] player status: slot={slot_num} state={state_str}") - _update_slot_state_from_player(slot_num, state_str) - - elif msg.topic == TOPIC_CONTROLLER_CMD: - _handle_control_cmd(msg.payload.decode("utf-8", errors="ignore")) - -def _handle_control_cmd(payload: str): - """ - Handle messages on TOPIC_CONTROLLER_CMD, e.g.: - {"action": "delete_slot", "slot": 2, "url": "..."} - {"action": "play", "slot": 1} - {"action": "stop"} - """ - try: - data = json.loads(payload or "{}") - except json.JSONDecodeError: - print(f"[controller] bad JSON on control cmd: {payload!r}") - return - - action = data.get("action") - if not action: - print("[controller] control cmd missing 'action'") - return - - # delete slot - if action == "delete_slot": - slot = data.get("slot") - try: - slot_num = int(slot) - except (TypeError, ValueError): - print(f"[controller] delete_slot: bad slot {slot!r}") - return - - expected_url = data.get("url") # optional safety - delete_slot(slot_num, expected_url=expected_url) - return - - # play / stop → hand off to the orchestrator - if action in ("play", "stop"): - handle_play_request(data) - return - - print(f"[controller] unknown control action: {action!r}") - -def _on_disconnect(client, userdata, rc, properties=None): - print(f"[mqtt] disconnected rc={rc}") - -mqtt_client = None +# Kept as transition indicators for integrations that inspected the old module. +mqtt_client: Any | None = None mqtt_started = False -mqtt_lifecycle_lock = threading.Lock() -def mqtt_publish(topic: str, payload, **kwargs): - """ - Small wrapper that JSON-encodes non-strings and prints what we publish. - Keeps logs useful and avoids repeating code. - """ - if not isinstance(payload, str): - payload = json.dumps(payload) - print(f"[mqtt] publish {topic} -> {payload}") - if mqtt_client is None: - print(f"[mqtt] publish skipped before Controller startup: {topic}") - return None - return mqtt_client.publish(topic, payload, **kwargs) - - -# --- State model --- - -@dataclass -class PlaylistSlot: - slot: int - url: str - yt_id: Optional[str] - title: str - path: str - m3u: str - created_at: float - expires_at: float - item_count: int - state: str # "downloading" | "ready" | "error" | "playing" | "busy" - - def to_dict(self): - d = asdict(self) - # timestamps as ISO-ish strings - d["created_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.created_at)) - d["expires_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.expires_at)) - return d - - @staticmethod - def from_dict(d: Dict): - def parse_ts(s: str) -> float: - try: - # naive ISO Z - return time.mktime(time.strptime(s, "%Y-%m-%dT%H:%M:%SZ")) - except Exception: - return float(d.get("created_at_ts", time.time())) - - created_at = parse_ts(d.get("created_at", "1970-01-01T00:00:00Z")) - expires_at = parse_ts(d.get("expires_at", "1970-01-01T00:00:00Z")) - - return PlaylistSlot( - slot=int(d["slot"]), - url=d["url"], - yt_id=d.get("yt_id"), - title=d.get("title", ""), - path=d["path"], - m3u=d["m3u"], - created_at=created_at, - expires_at=expires_at, - item_count=int(d.get("item_count", 0)), - state=d.get("state", "ready"), - ) - - -class ControllerState: - def __init__(self): - self.slots: Dict[int, PlaylistSlot] = {} - - def load(self): - if not STATE_PATH.exists(): - return - data = yaml.safe_load(STATE_PATH.read_text()) or {} - slots = data.get("slots", {}) - self.slots.clear() - for k, v in slots.items(): - try: - slot = PlaylistSlot.from_dict(v) - self.slots[slot.slot] = slot - except Exception: - continue - - def save(self): - data = {"slots": {str(s): slot.to_dict() for s, slot in self.slots.items()}} - STATE_PATH.write_text(yaml.safe_dump(data, sort_keys=True)) - - def prune_expired(self): - """ - Remove any slots whose expires_at is in the past, delete their folders, - compact remaining slots, save, and reannounce MQTT so ghosts disappear. - """ - now = time.time() - changed = False - to_delete: List[int] = [] - - for s, slot in list(self.slots.items()): - if now >= slot.expires_at: - to_delete.append(s) - - for s in to_delete: - slot = self.slots.pop(s, None) - if not slot: - continue - - try: - p = Path(slot.path) - if p.exists(): - for child in p.iterdir(): - try: - child.unlink() - except IsADirectoryError: - # ignore unexpected subdirs - pass - p.rmdir() - except Exception: - pass - - changed = True - - if changed: - # compact slot numbers (1..N), save, then resync MQTT topics - self.compact_slots() - self.save() - # this is the helper you already have below - reannounce_slots() - - def compact_slots(self): - """Shift slots down so there are no gaps: 1..N""" - if not self.slots: - return - new_slots: Dict[int, PlaylistSlot] = {} - for new_index, slot in enumerate(sorted(self.slots.values(), key=lambda sl: sl.slot), start=1): - if slot.slot != new_index: - old_dir = Path(slot.path) - new_dir = MEDIA_BASE / f"slot{new_index}" - # rename folder if it exists - if old_dir.exists() and old_dir != new_dir: - new_dir.parent.mkdir(parents=True, exist_ok=True) - old_dir.rename(new_dir) - slot.slot = new_index - slot.path = str(new_dir) - slot.m3u = str(new_dir / "playlist.m3u") - new_slots[new_index] = slot - self.slots = new_slots - - def get_next_free_slot(self) -> Optional[int]: - self.prune_expired() - used = set(self.slots.keys()) - for i in range(1, MAX_SLOTS + 1): - if i not in used: - return i - return None - - -state = ControllerState() - - -# --- Helpers --- - -def delete_slot(slot_num: int, expected_url: Optional[str] = None): - """ - Delete a slot from state + filesystem, then compact and republish. - - expected_url is optional: if provided and doesn't match, we ignore - the delete (safety against stale retained messages if someone ever - accidentally turns retain on). - """ - state.load() - - slot = state.slots.get(slot_num) - if not slot: - print(f"[controller] delete_slot: slot {slot_num} not found, ignoring") - return - - if expected_url and slot.url != expected_url: - print(f"[controller] delete_slot: url mismatch for slot {slot_num}, ignoring") - return - - print(f"[controller] deleting slot {slot_num}: {slot.title!r}") - - # Remove files on disk - slot_dir = Path(slot.path) - if slot_dir.exists(): - for child in slot_dir.iterdir(): - try: - child.unlink() - except IsADirectoryError: - # no subdirs expected, but don't crash - pass - try: - slot_dir.rmdir() - except OSError: - pass - - # Remove from state - state.slots.pop(slot_num, None) - - # Compact and save - state.compact_slots() - state.save() - - # Re-publish MQTT for all remaining slots, so HA/Deck are in sync - reannounce_slots() - - -def reannounce_slots(): - """ - Re-publish MQTT state/title/count for all existing slots, - and mark any unused slots as empty. - """ - state.load() - - # Re-announce real slots - for s, slot in sorted(state.slots.items()): - publish_slot_topics(slot) - - # Clear any slots above the current highest used index - highest = max(state.slots.keys(), default=0) - for s in range(highest + 1, MAX_SLOTS + 1): - base = f"{TOPIC_PLAYBACK_BASE}/{s}" - mqtt_publish(f"{base}/title", "", retain=True) - mqtt_publish(f"{base}/state", "empty", retain=True) - mqtt_publish(f"{base}/count", "0", retain=True) - # also clear item names - for i in range(1, 100): - mqtt_publish(f"{base}/items/{i}", "", retain=True) - - -def publish_slot_topics(slot: PlaylistSlot): - """Publish the title/state/count for one slot to its MQTT topics.""" - base = f"{TOPIC_PLAYBACK_BASE}/{slot.slot}" - mqtt_publish(f"{base}/title", slot.title, retain=True) - mqtt_publish(f"{base}/state", slot.state, retain=True) - mqtt_publish(f"{base}/count", slot.item_count, retain=True) - -def set_slot_busy_temporarily(slot: PlaylistSlot, seconds: int = 5): - """ - Briefly mark a slot as 'busy', then restore after a delay. - """ - base = f"{TOPIC_PLAYBACK_BASE}/{slot.slot}" - mqtt_publish(f"{base}/state", "busy", retain=True) - - def _clear_busy_later(slot_number: int, delay: int): - time.sleep(delay) - - state.load() - sl = state.slots.get(slot_number) - - if sl is None: - base2 = f"{TOPIC_PLAYBACK_BASE}/{slot_number}" - mqtt_publish(f"{base2}/title", "", retain=True) - mqtt_publish(f"{base2}/state", "empty", retain=True) - mqtt_publish(f"{base2}/count", 0, retain=True) - return - - publish_slot_topics(sl) - - t = threading.Thread( - target=_clear_busy_later, - args=(slot.slot, seconds), - daemon=True, - ) - t.start() - - - -def _parse_track_title_from_filename(path: Path) -> str: - stem = path.stem # "000 - Song Name [id]" - # strip leading "000 - " - if " - " in stem: - stem = stem.split(" - ", 1)[1] - # strip trailing " [id]" - if " [" in stem: - stem = stem.rsplit(" [", 1)[0] - return stem - - -def _update_slot_state_from_player(slot_num: int, new_state: str): - """ - Mirror player state ("playing", "busy", "idle", "error", etc.) - into our per-slot state + MQTT topics. - """ - slot_obj = state.slots.get(slot_num) - if not slot_obj: - return - - # Map "idle" / "finished" from player back to "ready" for the slot UI - if new_state in ("idle", "finished"): - slot_obj.state = "ready" - mqtt_state = "ready" - else: - slot_obj.state = new_state - mqtt_state = new_state - - state.save() - publish_slot_topics(slot_obj) - - -def _get_title_for_url(url: str) -> str: - """ - Try to detect a human-friendly title for either: - - a playlist (use playlist_title) - - a single video (fall back to video title) - - If we can't figure it out, just return the URL. - """ - try: - # One line: playlist_title|||video_title - out = subprocess.check_output( - [ - "yt-dlp", - "--flat-playlist", - "--print", "%(playlist_title)s|||%(title)s", - "--playlist-end", "1", - url, - ], - stderr=subprocess.DEVNULL, - text=True, - ).strip() - - if not out: - return url - - line = out.splitlines()[0] - playlist_title, _, video_title = line.partition("|||") - - playlist_title = (playlist_title or "").strip() - video_title = (video_title or "").strip() - - # If playlist_title looks real (not empty/NA), use it - if playlist_title and playlist_title.upper() != "NA": - return playlist_title - - # Otherwise fall back to video title - if video_title: - return video_title - - return url - except Exception: - return url - - -def _download_playlist_to_slot(url: str, slot_num: int): - """ - Runs yt-dlp into /mnt/Media/YTPlayer/slotN, builds playlist.m3u, - updates YAML, and publishes MQTT topics. - """ - slot_dir = MEDIA_BASE / f"slot{slot_num}" - slot_dir.mkdir(parents=True, exist_ok=True) - - title = _get_title_for_url(url) - now = time.time() - expires_at = now + RETENTION_SECONDS - - # mark slot as downloading - slot = PlaylistSlot( - slot=slot_num, - url=url, - yt_id=None, - title=title, - path=str(slot_dir), - m3u=str(slot_dir / "playlist.m3u"), - created_at=now, - expires_at=expires_at, - item_count=0, - state="downloading", - ) - state.slots[slot_num] = slot - state.save() - - # Tell HA/Deck: this slot is downloading now - publish_slot_topics(slot) - - mqtt_publish( - TOPIC_INGEST_PROGRESS, - {"slot": slot_num, "state": "downloading", "title": title}, - ) - - # run yt-dlp - try: - subprocess.run( - [ - "yt-dlp", - "--cookies", - "/cookies.txt", # inside the container (mount your exported cookies here) - "-o", - str(slot_dir / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"), - "--yes-playlist", - "--no-overwrites", # keep existing files, only fetch new ones - "-f", - "bv*[height<=1080]+ba/b[height<=1080]/b[height<=1080]/b", - url, - ], - check=True, - ) - except subprocess.CalledProcessError as e: - slot.state = "error" - state.save() - publish_slot_topics(slot) - mqtt_publish( - TOPIC_INGEST_PROGRESS, - {"slot": slot_num, "state": "error", "message": f"yt-dlp failed: {e}"}, - ) - return - - # --- build .m3u in sorted order --- - media_files = sorted( - [ - p - for p in slot_dir.iterdir() - if p.is_file() and p.suffix.lower() in {".mp4", ".mkv", ".webm"} - ] - ) - - if not media_files: - slot.state = "error" - state.save() - publish_slot_topics(slot) - mqtt_publish( - TOPIC_INGEST_PROGRESS, - {"slot": slot_num, "state": "error", "message": "no media files downloaded"}, - ) - return - - item_titles: List[str] = [] - m3u_path = slot_dir / "playlist.m3u" - with m3u_path.open("w", encoding="utf-8") as f: - for p in media_files: - # Use a path relative to the slot directory so that if the - # whole slotN folder gets renamed (compaction), the playlist - # still works. - rel = os.path.relpath(p, slot_dir) - f.write(rel + "\n") - item_titles.append(_parse_track_title_from_filename(p)) - - - slot.item_count = len(media_files) - slot.state = "ready" - state.slots[slot_num] = slot - state.save() - - # publish MQTT info for HA + StreamDeck - publish_slot_topics(slot) - - # items: 1-based index - base = f"{TOPIC_PLAYBACK_BASE}/{slot_num}" - for i, name in enumerate(item_titles, start=1): - mqtt_publish(f"{base}/items/{i}", name, retain=True) - - mqtt_publish( - TOPIC_INGEST_PROGRESS, - {"slot": slot_num, "state": "ready", "title": title, "count": slot.item_count}, - ) - -def publish_slot_topics(slot: PlaylistSlot): - """Publish the title/state/count for one slot to its MQTT topics.""" - base = f"{TOPIC_PLAYBACK_BASE}/{slot.slot}" - mqtt_publish(f"{base}/title", slot.title, retain=True) - mqtt_publish(f"{base}/state", slot.state, retain=True) - mqtt_publish(f"{base}/count", slot.item_count, retain=True) - -# --- Controller → Player orchestration --- - -def handle_play_request(cmd: Dict) -> None: - """ - Handle commands from HA/Companion on TOPIC_CONTROLLER_CMD. - - Supported payloads: - - {"action": "play", "slot": N} - - {"action": "play", "playlist": "/some/path.m3u"} # future perm playlist - - {"action": "stop"} - """ - action = cmd.get("action") - - # Which slot is currently playing? - playing_slot: Optional[int] = None - for s, slot in state.slots.items(): - if slot.state == "playing": - playing_slot = s - break - - # --- STOP command --- - if action == "stop": - # Forward stop to player - mqtt_publish(TOPIC_PLAYER_CMD, {"action": "stop"}) - - # Any slot marked playing goes back to ready - changed = False - for s, slot in state.slots.items(): - if slot.state == "playing": - slot.state = "ready" - base = f"{TOPIC_PLAYBACK_BASE}/{s}" - mqtt_publish(f"{base}/state", "ready", retain=True) - changed = True - if changed: - state.save() - return - - # --- PLAY command --- - if action != "play": - # ignore unknown for now - return - - slot_num_raw = cmd.get("slot") - playlist = cmd.get("playlist") or cmd.get("path") - label = cmd.get("label") - - slot_obj: Optional[PlaylistSlot] = None - - if slot_num_raw is not None: - try: - slot_num = int(slot_num_raw) - except (TypeError, ValueError): - print(f"[controller] bad slot in play cmd: {slot_num_raw!r}") - return - slot_obj = state.slots.get(slot_num) - if not slot_obj: - print(f"[controller] no such slot {slot_num}") - return - - # If we have a slot but no playlist, derive path from the slot. - if slot_obj is not None and not playlist: - playlist = slot_obj.m3u - label = label or slot_obj.title - - # If we still have no playlist/path, bail. - if not playlist: - print("[controller] play cmd missing playlist/path and no valid slot provided") - return - - # Enforce only one slot at a time for now - if playing_slot is not None and slot_obj is not None and playing_slot != slot_obj.slot: - # briefly mark this requested slot as busy, then restore it - set_slot_busy_temporarily(slot_obj, seconds=5) - - mqtt_publish( - TOPIC_INGEST_PROGRESS, - { - "slot": slot_obj.slot, - "state": "busy", - "message": f"slot {playing_slot} already playing; stop first", - }, - ) - return - - - # If we're starting a slot-backed playlist, mark that slot as playing - if slot_obj is not None: - slot_obj.state = "playing" - state.save() - base = f"{TOPIC_PLAYBACK_BASE}/{slot_obj.slot}" - mqtt_publish(f"{base}/state", "playing", retain=True) - - # Build command for the player - play_cmd = { - "action": "play", - "playlist": playlist, - } - if slot_obj is not None: - play_cmd["slot"] = slot_obj.slot - if label: - play_cmd["label"] = label - - mqtt_publish(TOPIC_PLAYER_CMD, play_cmd) - print(f"[controller] sent play cmd to player: {play_cmd}") - - -# --- HA Discovery --- - -HA_DISCOVERY_PREFIX = CONFIG.homeassistant.discovery_prefix -HA_DEVICE = { - "identifiers": ["ytplayer-controller"], - "name": "YouTube Controller", - "manufacturer": "River of Glory", - "model": "YT Controller", -} - -def ha_discovery_publish(): - """ - Publish MQTT Discovery configs for: - - sensor.yt_slot__title - - sensor.yt_slot__state - - sensor.yt_slot__count - """ - for n in range(1, MAX_SLOTS + 1): - base = f"{TOPIC_PLAYBACK_BASE}/{n}" - obj_id = f"yt_slot_{n}" - - # Title - cfg_title = { - "name": f"YT Slot {n} Title", - "state_topic": f"{base}/title", - "unique_id": f"{obj_id}_title", - "icon": "mdi:playlist-music", - "device": HA_DEVICE, - } - mqtt_publish( - f"{HA_DISCOVERY_PREFIX}/sensor/{obj_id}_title/config", - cfg_title, - retain=True, - ) - - # State (ready/downloading/playing/busy/error/idle) - cfg_state = { - "name": f"YT Slot {n} State", - "state_topic": f"{base}/state", - "unique_id": f"{obj_id}_state", - "icon": "mdi:circle-small", - "device": HA_DEVICE, - } - mqtt_publish( - f"{HA_DISCOVERY_PREFIX}/sensor/{obj_id}_state/config", - cfg_state, - retain=True, - ) - - # Count - cfg_count = { - "name": f"YT Slot {n} Count", - "state_topic": f"{base}/count", - "unique_id": f"{obj_id}_count", - "icon": "mdi:numeric", - "device": HA_DEVICE, - } - mqtt_publish( - f"{HA_DISCOVERY_PREFIX}/sensor/{obj_id}_count/config", - cfg_count, - retain=True, - ) - - -# --- Public API used by Flask app --- - -def enqueue_download(url: str, submitter: Optional[str], note: Optional[str]) -> Dict: - """ - Called by Flask when a new playlist is submitted. - - If this URL is already in a slot, reuse that slot and only grab new videos. - - Otherwise, allocate a new slot. - Returns a small dict summarizing what happened, for UI or HA. - """ - state.load() - state.prune_expired() - - # 1) Check if this URL already exists in a slot - existing_slot_obj = None - for sl in state.slots.values(): - if sl.url == url: - existing_slot_obj = sl - break - - if existing_slot_obj: - slot_num = existing_slot_obj.slot - print(f"[controller] enqueue_download: reusing existing slot {slot_num} for URL {url!r}") - reused = True - else: - # 2) Allocate a new slot - slot_num = state.get_next_free_slot() - reused = False - - if slot_num is None: - # queue full - expires = [sl.expires_at for sl in state.slots.values()] - oldest = min(expires) if expires else time.time() - return { - "status": "queue_full", - "oldest_expires_at": oldest, - "oldest_expires_in_seconds": int(oldest - time.time()), - } - - # 3) Spawn background thread so Flask doesn't block - t = threading.Thread( - target=_download_playlist_to_slot, - args=(url, slot_num), - daemon=True, - ) - t.start() - - # 4) Simple MQTT log of submission - mqtt_publish( - TOPIC_INGEST_PROGRESS, - { - "slot": slot_num, - "state": "queued", - "url": url, - "submitter": submitter, - "note": note, - "reused_slot": reused, - }, - ) - - return { - "status": "queued", - "slot": slot_num, - "reused_slot": reused, - } - - -# --- Explicit lifecycle (imports are intentionally side-effect free) --- - def start_controller() -> None: - """Initialize storage and start non-blocking MQTT retry behavior.""" - global mqtt_client, mqtt_started - with mqtt_lifecycle_lock: - if mqtt_started: + """Construct and start the singleton compatibility service explicitly.""" + + global _config, _service, mqtt_client, mqtt_started + with _lifecycle_lock: + if _service is not None and _service.started: return - - MEDIA_BASE.mkdir(parents=True, exist_ok=True) - DATA_BASE.mkdir(parents=True, exist_ok=True) - state.load() - - mqtt_client = mqtt.Client(client_id=MQTT_CLIENT_ID, clean_session=True) - if CONFIG.mqtt.username is not None: - mqtt_client.username_pw_set(CONFIG.mqtt.username, CONFIG.mqtt.password) - mqtt_client.enable_logger() - mqtt_client.on_connect = _on_connect - mqtt_client.on_disconnect = _on_disconnect - mqtt_client.on_message = _on_message - mqtt_client.reconnect_delay_set( - min_delay=CONFIG.mqtt.reconnect_min_seconds, - max_delay=CONFIG.mqtt.reconnect_max_seconds, - ) - - print(f"[controller] attempting asynchronous MQTT connect to {MQTT_HOST}:{MQTT_PORT} …") - mqtt_client.connect_async( - MQTT_HOST, - MQTT_PORT, - keepalive=CONFIG.mqtt.keepalive_seconds, - ) - mqtt_client.loop_start() + config = load_config() + service = ControllerService(config) + service.start() + _config = config + _service = service + mqtt_client = service.mqtt mqtt_started = True - print("[controller] MQTT loop started") def stop_controller() -> None: - """Stop the MQTT client if explicit startup occurred.""" - global mqtt_client, mqtt_started - with mqtt_lifecycle_lock: - if not mqtt_started or mqtt_client is None: - return - try: - mqtt_client.disconnect() - finally: - mqtt_client.loop_stop() - mqtt_client = None - mqtt_started = False + """Stop and discard the singleton compatibility service.""" + + global _config, _service, mqtt_client, mqtt_started + with _lifecycle_lock: + service = _service + if service is not None: + service.stop() + _service = None + _config = None + mqtt_client = None + mqtt_started = False + + +def get_service() -> ControllerService: + """Return the explicitly started compatibility service.""" + + with _lifecycle_lock: + if _service is None or not _service.started: + raise ControllerUnavailableError( + "Controller is not started; call start_controller() first" + ) + return _service + + +def enqueue_download( + url: str, + submitter: str | None = None, + note: str | None = None, +) -> dict[str, Any]: + """Queue a download through Controller v2.""" + + return get_service().enqueue_download(url, submitter, note) + + +__all__ = [ + "enqueue_download", + "get_service", + "mqtt_client", + "mqtt_started", + "start_controller", + "stop_controller", +] diff --git a/gunicorn.conf.py b/gunicorn.conf.py new file mode 100644 index 0000000..6592bf4 --- /dev/null +++ b/gunicorn.conf.py @@ -0,0 +1,25 @@ +"""Single-process runtime: ControllerService owns one local coordinator.""" + +from __future__ import annotations + +import os + + +bind = f"0.0.0.0:{int(os.environ.get('PORT', '7531'))}" +workers = 1 +worker_class = "gthread" +threads = 4 +preload_app = False +accesslog = "-" +errorlog = "-" +capture_output = True +timeout = 30 +graceful_timeout = 15 + + +def worker_exit(server, worker): # pragma: no cover - exercised by Gunicorn + try: + from wsgi import service + except Exception: + return + service.stop() diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..a3f7a38 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,316 @@ +import importlib +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest import mock + +from controller.config import ControllerConfig +from controller.models import StoreError +from controller.mqtt_adapter import MqttAdapter +from controller.service import ControllerService, ControllerUnavailableError + + +class FakeControllerService: + def __init__(self, *, enqueue_result=None, enqueue_error=None, health=None): + self.enqueue_result = enqueue_result or { + "status": "queued", + "slot": 3, + "reused_slot": False, + } + self.enqueue_error = enqueue_error + self.health_result = health or { + "status": "ok", + "started": True, + "coordinator_alive": True, + "mqtt_connected": True, + "router_state": "video", + "degraded_reasons": [], + } + self.enqueue_calls = [] + + def enqueue_download(self, url, submitter, note): + self.enqueue_calls.append((url, submitter, note)) + if self.enqueue_error is not None: + raise self.enqueue_error + return dict(self.enqueue_result) + + def health(self): + return dict(self.health_result) + + +class FlaskFactoryTests(unittest.TestCase): + def setUp(self): + self.config = ControllerConfig() + + @staticmethod + def _factory(): + module = importlib.import_module("app") + return module.create_app + + def _app(self, service): + return self._factory()(self.config, service, testing=True) + + def _post(self, service, *, url="https://example.com/playlist", **fields): + app = self._app(service) + return app.test_client().post( + "/submit", + data={"url": url, **fields}, + follow_redirects=True, + ) + + def test_get_index_preserves_submission_form(self): + response = self._app(FakeControllerService()).test_client().get("/") + + self.assertEqual(response.status_code, 200) + self.assertIn(b"Submit YouTube Playlist", response.data) + self.assertIn(b'action="/submit"', response.data) + self.assertIn(b'name="url"', response.data) + self.assertIn(b'name="submitter"', response.data) + self.assertIn(b'name="note"', response.data) + + def test_submit_accepts_only_http_or_https_urls_with_a_network_location(self): + invalid_urls = ( + "", + "http", + "https:///missing-host", + "httpfoo://example.com/video", + "ftp://example.com/video", + "javascript:alert(1)", + "/relative/path", + ) + for url in invalid_urls: + with self.subTest(url=url): + service = FakeControllerService() + response = self._post(service, url=url) + self.assertEqual(response.status_code, 200) + self.assertIn(b"Invalid URL", response.data) + self.assertEqual(service.enqueue_calls, []) + + for url in ("http://example.com/video", "https://example.com/video"): + with self.subTest(url=url): + service = FakeControllerService() + response = self._post(service, url=url) + self.assertEqual(response.status_code, 200) + self.assertEqual(service.enqueue_calls, [(url, None, None)]) + + def test_queued_submission_preserves_message_and_trims_optional_fields(self): + service = FakeControllerService( + enqueue_result={"status": "queued", "slot": 3, "reused_slot": False} + ) + + response = self._post( + service, + url="https://example.com/playlist", + submitter=" Alice ", + note=" Before service ", + ) + + self.assertEqual(response.status_code, 200) + self.assertIn(b"Playlist queued in slot 3", response.data) + self.assertIn(b"start downloading shortly", response.data) + self.assertEqual( + service.enqueue_calls, + [("https://example.com/playlist", "Alice", "Before service")], + ) + + def test_reused_submission_preserves_existing_slot_message(self): + service = FakeControllerService( + enqueue_result={"status": "queued", "slot": 7, "reused_slot": True} + ) + + response = self._post(service) + + self.assertEqual(response.status_code, 200) + self.assertIn(b"Playlist queued in slot 7", response.data) + self.assertIn(b"Existing slot", response.data) + self.assertIn(b"only new videos will be added", response.data) + + def test_queue_full_has_a_useful_capacity_message(self): + service = FakeControllerService( + enqueue_result={ + "status": "queue_full", + "slot": None, + "reused_slot": False, + "oldest_expires_in_seconds": 7200, + } + ) + + response = self._post(service) + + self.assertEqual(response.status_code, 200) + self.assertIn(b"Queue is full", response.data) + self.assertIn(b"about 2 hour(s)", response.data) + self.assertNotIn(b"slot None", response.data) + + def test_shutting_down_has_a_useful_retry_message(self): + service = FakeControllerService( + enqueue_result={ + "status": "shutting_down", + "slot": None, + "reused_slot": False, + } + ) + + response = self._post(service) + + self.assertEqual(response.status_code, 200) + self.assertIn(b"shutting down", response.data.lower()) + self.assertIn(b"try again", response.data.lower()) + self.assertNotIn(b"slot None", response.data) + + def test_worker_error_has_a_useful_failure_message(self): + service = FakeControllerService( + enqueue_result={ + "status": "worker_error", + "slot": 4, + "reused_slot": False, + } + ) + + response = self._post(service) + + self.assertEqual(response.status_code, 200) + self.assertIn(b"download worker", response.data.lower()) + self.assertTrue( + b"could not" in response.data.lower() + or b"failed" in response.data.lower() + ) + self.assertNotIn(b"Playlist queued", response.data) + + def test_controller_unavailable_is_handled_without_a_500(self): + service = FakeControllerService( + enqueue_error=ControllerUnavailableError( + "Controller coordinator did not respond" + ) + ) + + response = self._post(service) + + self.assertEqual(response.status_code, 200) + self.assertNotIn(b"Internal Server Error", response.data) + self.assertIn(b"Controller", response.data) + self.assertIn(b"unavailable", response.data.lower()) + + def test_store_error_is_handled_without_a_500(self): + service = FakeControllerService( + enqueue_error=StoreError("Controller database is unavailable") + ) + + response = self._post(service) + + self.assertEqual(response.status_code, 200) + self.assertNotIn(b"Internal Server Error", response.data) + self.assertIn(b"Controller", response.data) + self.assertIn(b"unavailable", response.data.lower()) + + def test_health_is_healthy_when_ok_or_degraded(self): + cases = ( + { + "status": "ok", + "started": True, + "coordinator_alive": True, + "mqtt_connected": True, + "degraded_reasons": [], + }, + { + "status": "degraded", + "started": True, + "coordinator_alive": True, + "mqtt_connected": False, + "degraded_reasons": ["mqtt_disconnected"], + }, + ) + for health in cases: + with self.subTest(status=health["status"]): + app = self._app(FakeControllerService(health=health)) + response = app.test_client().get("/healthz") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.get_json(), health) + + def test_health_is_unhealthy_when_stopped_or_coordinator_is_dead(self): + cases = ( + { + "status": "stopped", + "started": False, + "coordinator_alive": False, + "mqtt_connected": False, + "degraded_reasons": [], + }, + { + "status": "degraded", + "started": True, + "coordinator_alive": False, + "mqtt_connected": False, + "degraded_reasons": ["coordinator_dead"], + }, + ) + for health in cases: + with self.subTest(status=health["status"]): + app = self._app(FakeControllerService(health=health)) + response = app.test_client().get("/healthz") + self.assertEqual(response.status_code, 503) + self.assertEqual(response.get_json(), health) + + def test_production_factory_requires_flask_secret_key(self): + with mock.patch.dict(os.environ, {}, clear=True): + with self.assertRaisesRegex(RuntimeError, "FLASK_SECRET_KEY"): + self._factory()(self.config, FakeControllerService()) + + def test_testing_factory_can_use_a_test_only_secret(self): + with mock.patch.dict(os.environ, {}, clear=True): + app = self._factory()( + self.config, + FakeControllerService(), + testing=True, + ) + + self.assertIs(app.testing, True) + self.assertTrue(app.secret_key) + self.assertNotEqual(app.secret_key, "dev-secret") + + +class AppImportLifecycleTests(unittest.TestCase): + def test_import_does_not_construct_or_start_controller_runtime(self): + with tempfile.TemporaryDirectory() as directory: + media = Path(directory) / "media" + data = Path(directory) / "data" + environment = { + **os.environ, + "YTPLAYER_MEDIA_BASE": str(media), + "YTPLAYER_DATA_BASE": str(data), + } + environment.pop("FLASK_SECRET_KEY", None) + before_threads = {thread.ident for thread in threading.enumerate()} + sys.modules.pop("app", None) + + with mock.patch.dict(os.environ, environment, clear=True): + with mock.patch.object( + ControllerService, + "__init__", + side_effect=AssertionError("import constructed ControllerService"), + ), mock.patch.object( + ControllerService, + "start", + side_effect=AssertionError("import started ControllerService"), + ), mock.patch.object( + MqttAdapter, + "start", + side_effect=AssertionError("import started MQTT"), + ): + imported = importlib.import_module("app") + + self.assertTrue(callable(imported.create_app)) + self.assertFalse(media.exists()) + self.assertFalse(data.exists()) + self.assertEqual( + {thread.ident for thread in threading.enumerate()}, + before_threads, + ) + sys.modules.pop("app", None) + + +if __name__ == "__main__": + unittest.main() diff --git a/wsgi.py b/wsgi.py new file mode 100644 index 0000000..2199220 --- /dev/null +++ b/wsgi.py @@ -0,0 +1,22 @@ +"""Production WSGI entry point and Controller service lifecycle owner.""" + +from __future__ import annotations + +import atexit +import logging + +from app import create_app +from controller.config import load_config +from controller.service import ControllerService + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) + +config = load_config() +service = ControllerService(config) +application = create_app(config, service) +service.start() +atexit.register(service.stop)