From f3f83bff1d9461ecf7883a2072f9ae4ddc77eb7b Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 31 Jul 2026 18:14:53 -0400 Subject: [PATCH 1/2] [youtubearr] Bump version to 1.30.0 --- plugins/youtubearr/plugin.py | 778 ++++++++++++++++++++++++++--------- 1 file changed, 577 insertions(+), 201 deletions(-) diff --git a/plugins/youtubearr/plugin.py b/plugins/youtubearr/plugin.py index d7942dd..445e2b5 100644 --- a/plugins/youtubearr/plugin.py +++ b/plugins/youtubearr/plugin.py @@ -3,6 +3,7 @@ import re import subprocess import sys +import fcntl import threading import time import urllib.request @@ -274,13 +275,6 @@ class Plugin: "button_label": "Reset All", "button_color": "red", }, - { - "id": "poll", - "label": "Poll Now (Cron Entrypoint)", - "description": "Run a synchronous poll cycle. Recommended for cron/host scheduling.", - "button_label": "Poll", - "button_color": "blue", - }, { "id": "diagnostics", "label": "Diagnostics", @@ -299,10 +293,25 @@ def __init__(self) -> None: self._channel_group_name = "YouTube Live" self._starting_channel_number = 2000 + # Runtime state sidecar (replaces settings-based runtime coordination) + self._runtime_state_path = self._base_dir / "runtime_state.json" + self._lock_path = self._base_dir / "monitor.lock" + self._lock_fd = None # File descriptor for the acquired exclusive lock + + # Dedicated lock serializing runtime_state.json read-modify-write across + # threads (in-process) and workers (cross-process via flock). Deliberately + # a separate file from monitor.lock so writing runtime state never + # contends with monitor-lock acquire/release. + self._runtime_state_lock_path = self._base_dir / "runtime_state.lock" + self._runtime_state_thread_lock = threading.Lock() + + # Monitoring thread + self._monitor_thread: Optional[threading.Thread] = None + self._monitor_stop_event = threading.Event() + self._monitoring_active = False # In-memory flag (authoritative within this process) self._manual_refresh_lock = threading.Lock() - self._last_manual_refresh = {"state": "idle"} - self._legacy_task_cleanup_done = False + # Stream profile cache self._stream_profile_id: Optional[int] = None # Track assigned channel numbers during poll cycle to avoid duplicates @@ -311,6 +320,8 @@ def __init__(self) -> None: # Track video IDs that recently failed metadata extraction to avoid retrying every poll self._extraction_failures: Dict[str, float] = {} # video_id -> unix timestamp of failure + self._legacy_task_cleanup_done = False + # Field defaults self._field_defaults = {field["id"]: field.get("default") for field in self.fields} @@ -344,8 +355,6 @@ def run(self, action: str, params: Dict[str, Any], context: Dict[str, Any]) -> D response = self._handle_cleanup(context) elif action == "reset_all": response = self._handle_reset_all(context) - elif action == "poll": - response = self._handle_poll(context) elif action == "diagnostics": response = self._handle_diagnostics(context) else: @@ -354,96 +363,65 @@ def run(self, action: str, params: Dict[str, Any], context: Dict[str, Any]) -> D return response def stop(self, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - """Called when plugin is disabled/reloaded""" - if not context or "settings" not in context: - try: - cfg = PluginConfig.objects.get(key=self._plugin_key) - settings = dict(cfg.settings or {}) - except PluginConfig.DoesNotExist: - settings = {} - context = {"settings": settings} + """Called when plugin is disabled/reloaded by Dispatcharr lifecycle. - return self._handle_stop_monitoring(context) - - # --- Action Handlers --- + Stops the local in-memory thread only — does NOT write monitoring_active=False to DB. + DB state is preserved so _ensure_monitoring_thread can revive monitoring after reload. + Explicit user-initiated stops go through _handle_stop_monitoring() instead. + """ + self._stop_thread_local() + return {"status": "stopped", "message": "Plugin lifecycle stop (monitoring state preserved)"} + def _stop_thread_local(self) -> None: + """Signal and join the local monitor thread without touching DB state. - def execute_poll_cycle(self, settings: dict, trigger: str = "scheduled") -> dict: - """One-shot execution of a full monitoring cycle.""" - started_at = timezone.now().isoformat() - try: - added, ended = self._poll_monitored_channels(settings) - self._refresh_expiring_urls(settings) - self._refresh_epg_times(settings) - if settings.get("auto_cleanup", True): - cleaned = self._cleanup_ended_streams(settings) - else: - cleaned = 0 - - triggered = added > 0 or cleaned > 0 - if triggered: - self._trigger_webhook(settings) - - result = { - "state": "completed", - "trigger": trigger, - "started_at": started_at, - "completed_at": timezone.now().isoformat(), - "added": added, - "ended": ended, - "cleaned": cleaned, - "webhook_triggered": triggered, - "error": None, - } - except Exception as exc: - self._log_error(f"Poll cycle error: {exc}") - result = { - "state": "failed", - "trigger": trigger, - "started_at": started_at, - "completed_at": timezone.now().isoformat(), - "added": None, - "ended": None, - "cleaned": None, - "webhook_triggered": False, - "error": str(exc)[:200], - } + For lifecycle events (plugin reload/disable). Does not persist any DB changes, + so monitoring_active=True is preserved in DB for auto-restart after reload. + _handle_stop_monitoring() is the explicit user-stop path that writes monitoring_active=False. + """ + self._monitoring_active = False + self._monitor_stop_event.set() + if self._monitor_thread and self._monitor_thread.is_alive(): + self._monitor_thread.join(timeout=5.0) + self._log("Local monitor thread stopped (lifecycle, DB state preserved)") - self._last_poll_result = result - try: - self._persist_settings({"last_poll_result": result, "last_poll_time": timezone.now().isoformat()}) - except Exception: - pass - return result + # --- Action Handlers --- def _handle_status(self, context: Dict[str, Any]) -> Dict[str, Any]: """Return current status""" - try: - cfg = PluginConfig.objects.get(key=self._plugin_key) - settings = dict(cfg.settings or {}) - except PluginConfig.DoesNotExist: - settings = context.get("settings", {}) - - tracked_streams = settings.get("tracked_streams", {}) - monitoring_active = settings.get("monitoring_active", False) - + # Clean up the bogus Celery beat task left by older plugin versions (once per instance) + # Must run before any early return, so it fires even when yt-dlp is missing. self._cleanup_legacy_celery_task() + # Check yt-dlp availability if not self._ytdlp_path: return { "status": "error", "message": "yt-dlp not found (bundled version may not be working). Check logs.", } - message_parts = [] - if monitoring_active: - message_parts.append(f"Monitoring active (cron-fallback mode) ({len(tracked_streams)} streams tracked)") - else: - message_parts.append(f"Monitoring inactive ({len(tracked_streams)} streams tracked)") + try: + cfg = PluginConfig.objects.get(key=self._plugin_key) + settings = dict(cfg.settings or {}) + except PluginConfig.DoesNotExist: + settings = context.get("settings", {}) + tracked_streams = settings.get("tracked_streams", {}) + runtime = self._read_runtime_state() + desired_active = runtime.get("desired_active", False) + + # Self-heal: restart the monitor thread if desired but no live thread + self._ensure_monitoring_thread(settings) + + is_active = desired_active or self._monitoring_active + message = ( + f"Monitoring active ({len(tracked_streams)} streams tracked)" + if is_active + else f"Monitoring inactive ({len(tracked_streams)} streams tracked)" + ) return { - "status": "running" if monitoring_active else "stopped", - "message": " | ".join(message_parts) if message_parts else "Ready", + "status": "running" if is_active else "stopped", + "message": message, } def _handle_add_manual(self, context: Dict[str, Any]) -> Dict[str, Any]: @@ -612,40 +590,105 @@ def _handle_add_manual(self, context: Dict[str, Any]) -> Dict[str, Any]: } def _handle_start_monitoring(self, context: Dict[str, Any]) -> Dict[str, Any]: - """Start background monitoring""" + """Start background monitoring thread""" if not self._ytdlp_path: return { "status": "error", "message": "yt-dlp not found (bundled version may not be working). Check logs.", } - try: - cfg = PluginConfig.objects.get(key=self._plugin_key) - settings = dict(cfg.settings or {}) - except PluginConfig.DoesNotExist: - settings = context.get("settings", {}) + # Fast path: local thread is alive + if self._monitor_thread and self._monitor_thread.is_alive(): + self._log("Monitoring already active (local thread alive)") + return {"status": "running", "message": "Monitoring already active"} + settings = context.get("settings", {}) monitored = settings.get("monitored_channels", "").strip() if not monitored: return {"status": "error", "message": "No channels to monitor. Add channel IDs/URLs in settings."} - self._persist_settings({"monitoring_active": True}) - self._log("Monitoring enabled (cron fallback)") + # Persist user intent before attempting lock acquisition + self._write_runtime_state({"desired_active": True}) + + # Try to acquire the exclusive file lock (non-blocking). + # If another worker process holds it, a monitor is already running there. + if not self._acquire_monitor_lock(): + self._log("Monitor lock held by another worker — monitoring already active") + return {"status": "running", "message": "Monitoring already active"} + + # Lock acquired — start the thread + self._monitoring_active = True + self._monitor_stop_event.clear() + self._extraction_failures.clear() + + self._monitor_thread = threading.Thread( + target=self._monitoring_loop, + args=(self._plugin_key,), + daemon=True, + name="YouTubearr-Monitor" + ) + self._monitor_thread.start() + + self._log("Monitoring started") self._cleanup_legacy_celery_task() return { "status": "running", - "message": "Monitoring enabled. Please schedule the 'poll' action via cron.", + "message": "Monitoring started", } def _handle_stop_monitoring(self, context: Dict[str, Any]) -> Dict[str, Any]: """Stop background monitoring thread""" - updates = { - "monitoring_active": False, - "force_refresh_requested_at": None, - } - self._persist_settings(updates) - self._log("Monitoring disabled") + runtime = self._read_runtime_state() + if not self._monitoring_active and not runtime.get("desired_active"): + return {"status": "stopped", "message": "Monitoring not active"} + + # Persist user intent — survives any Dispatcharr settings-save overwrite + self._write_runtime_state({"desired_active": False}) + + # Signal the local thread + self._monitoring_active = False + self._monitor_stop_event.set() + + if self._monitor_thread and self._monitor_thread.is_alive(): + self._monitor_thread.join(timeout=5.0) + if self._monitor_thread.is_alive(): + # Still mid-cycle after the join timeout — it still owns + # monitor.lock and will release it itself in _monitoring_loop's + # finally block. Releasing it here would free the lock while + # the thread is still running, letting another worker acquire + # it and start a second monitor concurrently. Leave it alone; + # desired_active=False is already persisted so the loop will + # observe it and exit at its next safe boundary. + self._log("Stop Monitoring: thread still shutting down after timeout; lock retained") + return { + "status": "stopping", + "message": ( + "Stop requested; monitor is finishing its current cycle " + "and will stop shortly. Try again if this persists." + ), + } + + # Thread has exited (or was never running locally) — its finally + # block already released the lock; this call is a safety net. + self._release_monitor_lock() + + # This worker never owned the local thread (it was None/not alive + # above), so the lock may still be held by another worker process + # that hasn't observed desired_active=False yet. Report the truthful + # "stopping" state instead of falsely claiming monitoring already + # stopped — the owning worker will exit at its next safe boundary. + if self._is_monitor_lock_held_by_other(): + self._log("Stop Monitoring: lock held by another worker; stop requested") + return { + "status": "stopping", + "message": ( + "Stop requested; monitoring is active on another worker " + "process and will stop at its next safe boundary." + ), + } + + self._log("Monitoring stopped") self._cleanup_legacy_celery_task() return { @@ -657,6 +700,55 @@ def _handle_refresh(self, context: Dict[str, Any]) -> Dict[str, Any]: """Manually trigger a refresh cycle""" self._log(f"!!! REFRESH ACTION TRIGGERED - Plugin version {self.version} !!!") + try: + cfg = PluginConfig.objects.get(key=self._plugin_key) + settings = dict(cfg.settings or {}) + except PluginConfig.DoesNotExist: + settings = context.get("settings", {}) + + runtime = self._read_runtime_state() + + if runtime.get("desired_active"): + thread_alive = bool(self._monitor_thread and self._monitor_thread.is_alive()) + + if thread_alive: + poll_interval = settings.get("poll_interval_minutes", 15) + last_poll = runtime.get("last_poll_time", "") + last_poll_display = last_poll[:19].replace("T", " ") if last_poll else "unknown" + last_hb = runtime.get("last_heartbeat_at", "") + last_hb_display = last_hb[:19].replace("T", " ") if last_hb else "unknown" + return { + "status": "success", + "message": ( + f"Monitoring is active (polling every {poll_interval} min). " + f"Last poll: {last_poll_display}. Heartbeat: {last_hb_display}. " + "No manual refresh needed." + ), + } + + # desired_active=True but local thread dead — attempt restart + restarted = self._ensure_monitoring_thread(settings) + if restarted: + return { + "status": "running", + "message": "Monitoring was marked active but was not running; restarted monitoring.", + } + + # Could not restart locally. If another worker process genuinely holds + # the monitor lock, it is actively running the loop — report truthful + # already-active status instead of falling through to a duplicate + # one-shot poll that would race the owning worker's cycle. + if self._is_monitor_lock_held_by_other(): + poll_interval = settings.get("poll_interval_minutes", 15) + return { + "status": "success", + "message": ( + f"Monitoring is active on another worker process (polling every " + f"{poll_interval} min). No manual refresh needed." + ), + } + # Lock is free (e.g. no channels configured or yt-dlp missing) — fall through to one-shot + if not self._manual_refresh_lock.acquire(blocking=False): return {"status": "info", "message": "A manual refresh is already in progress — check logs for progress."} @@ -664,27 +756,19 @@ def _run(): try: cfg = PluginConfig.objects.get(key=self._plugin_key) s = dict(cfg.settings or {}) - self._last_manual_refresh = self.execute_poll_cycle(s, trigger="manual") + added, ended = self._poll_monitored_channels(s) + if s.get("auto_cleanup", True): + self._cleanup_ended_streams(s) + if added > 0 or ended > 0: + self._trigger_webhook(s) + except Exception as exc: + self._log_error(f"Manual refresh failed: {exc}") finally: self._manual_refresh_lock.release() threading.Thread(target=_run, daemon=True, name="YouTubearr-ManualRefresh").start() return {"status": "success", "message": "Refresh started in background — check logs or wait for the next status update."} - def _handle_poll(self, context: Dict[str, Any]) -> Dict[str, Any]: - """Synchronous poll endpoint for cron triggers.""" - try: - cfg = PluginConfig.objects.get(key=self._plugin_key) - settings = dict(cfg.settings or {}) - except PluginConfig.DoesNotExist: - settings = context.get("settings", {}) - - if not settings.get("monitoring_active"): - return {"status": "info", "message": "Monitoring is disabled; poll skipped."} - - result = self.execute_poll_cycle(settings, trigger="cron") - return {"status": "success", "message": "Poll completed", "details": result} - def _handle_cleanup(self, context: Dict[str, Any]) -> Dict[str, Any]: """Manually cleanup ended streams and orphaned tracked_streams entries""" # Get settings from database to preserve monitoring_active flag @@ -732,29 +816,84 @@ def _handle_cleanup(self, context: Dict[str, Any]) -> Dict[str, Any]: def _handle_reset_all(self, context: Dict[str, Any]) -> Dict[str, Any]: """Reset all YouTubearr channels and tracking data to start fresh.""" try: - # Step 1: Force-stop monitoring by setting DB flag FIRST - # This ensures any running thread (even in another worker) will see the stop signal + # Step 1: Persist stop intent to runtime_state. tracked_streams is + # NOT cleared here — clearing it before the monitor is confirmed + # stopped would race a still-running owner (which reloads settings + # every cycle) into treating everything as new and re-adding + # entries while we go on to delete channels/EPG below. It is + # cleared further down, only once the monitor is confirmed stopped. + self._write_runtime_state({"desired_active": False}) + + # Step 2: Signal in-memory stop — covers the case where this worker + # is itself the monitor owner. + self._monitoring_active = False + self._monitor_stop_event.set() + self._log("Reset All: Set in-memory stop flags") + + # Step 3: Wait for the monitor lock to become free. desired_active=False + # was already persisted in Step 1, so whichever worker holds the lock — + # this one or another — will observe it in _monitoring_loop and + # exit/release it. Join our own thread if we own it, then poll for lock + # release (bounded) instead of a blind sleep, so Reset All doesn't race + # a still-running monitor loop on another worker process. + # + # If our own thread is still alive after the join timeout, it still + # owns monitor.lock — do NOT release it out from under a running + # loop, and do NOT proceed to the destructive channel/EPG deletion + # below. Same if another worker still holds the lock once our own + # bounded wait expires: abort truthfully instead of reporting + # success while an owner may still be mid-poll. + monitor_confirmed_stopped = True + if self._monitor_thread and self._monitor_thread.is_alive(): + self._monitor_thread.join(timeout=5.0) + if self._monitor_thread.is_alive(): + monitor_confirmed_stopped = False + + if monitor_confirmed_stopped: + self._release_monitor_lock() + deadline = time.monotonic() + 8.0 + lock_free = not self._is_monitor_lock_held_by_other() + while not lock_free and time.monotonic() < deadline: + time.sleep(0.5) + lock_free = not self._is_monitor_lock_held_by_other() + if not lock_free: + monitor_confirmed_stopped = False + + if not monitor_confirmed_stopped: + self._log("Reset All: aborted — monitor lock could not be confirmed free") + return { + "status": "error", + "message": ( + "Reset aborted: monitoring is still running or shutting down " + "and the monitor lock could not be confirmed free. Wait a " + "moment and try Reset All again." + ), + } + + self._log("Reset All: Confirmed monitoring thread stopped") + + # Step 3b: Now that the monitor is confirmed stopped, clear + # tracked_streams in the DB — safe to do since no owner is left + # mid-poll to repopulate it out from under this reset. try: cfg = PluginConfig.objects.get(key=self._plugin_key) tracked_count = len(cfg.settings.get("tracked_streams", {})) new_settings = dict(cfg.settings or {}) - new_settings["monitoring_active"] = False - new_settings["force_refresh_requested_at"] = None new_settings["tracked_streams"] = {} cfg.settings = new_settings cfg.save(update_fields=["settings", "updated_at"]) - self._log(f"Reset All: Set monitoring_active=False and cleared {tracked_count} tracked_streams") + self._log(f"Reset All: Cleared {tracked_count} tracked_streams") except PluginConfig.DoesNotExist: tracked_count = 0 - # Step 2: Get the channel group (read from settings, not hardcoded) + # Step 4: Get the channel group (read from settings, not hardcoded) group_name = context.get("settings", {}).get("channel_group_name", self._channel_group_name) try: channel_group = ChannelGroup.objects.get(name=group_name) except ChannelGroup.DoesNotExist: channel_group = None - # Step 3: Delete all channels in the YouTube Live group + # Step 5: Delete all channels in the YouTube Live group channels_deleted = 0 streams_deleted = 0 @@ -771,7 +910,7 @@ def _handle_reset_all(self, context: Dict[str, Any]) -> Dict[str, Any]: self._log(f"Reset All: Deleted {channels_deleted} channel(s) and {streams_deleted} stream(s)") - # Step 4: Clean up EPG data for this plugin's EPG source + # Step 6: Clean up EPG data for this plugin's EPG source epg_source_name = context.get("settings", {}).get("epg_source_name", "YouTube Live").strip() epg_cleaned = 0 if epg_source_name: @@ -801,17 +940,7 @@ def _handle_reset_all(self, context: Dict[str, Any]) -> Dict[str, Any]: def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: """Run a non-destructive health check and return diagnostics details.""" - # Prefer fresh DB settings so fields like last_poll_time reflect the most recent - # poll cycle rather than whatever stale values Dispatcharr cached in the request - # context. Merge strategy: start from context settings so test fixtures work, then - # overlay DB settings — DB wins for every key it has. - context_settings = context.get("settings", {}) - try: - cfg = PluginConfig.objects.get(key=self._plugin_key) - db_settings = dict(cfg.settings or {}) - settings = {**context_settings, **db_settings} - except PluginConfig.DoesNotExist: - settings = context_settings + settings = context.get("settings", {}) issues: List[str] = [] # "error:" or "warning:" details: Dict[str, Any] = {} @@ -819,24 +948,41 @@ def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: details["plugin_version"] = self.version details["plugin_key"] = self._plugin_key - # Monitoring state - details["monitoring_active"] = settings.get("monitoring_active", False) - details["scheduler_mode"] = "cron-fallback" - details["poll_interval_minutes"] = settings.get("poll_interval_minutes", 15) + # Monitoring state — read from runtime_state.json, not settings + runtime = self._read_runtime_state() + monitoring_active_db = runtime.get("desired_active", False) + details["monitoring_active"] = monitoring_active_db + thread_alive = bool(self._monitor_thread and self._monitor_thread.is_alive()) + details["monitor_thread_alive"] = thread_alive + _last_poll = runtime.get("last_poll_time") or "" + details["last_poll_time"] = _last_poll or "unknown" + _lpa = self._age_seconds(_last_poll) if _last_poll else None + details["last_poll_age_seconds"] = int(_lpa) if _lpa is not None else None + _hb = runtime.get("last_heartbeat_at") or "" + details["monitoring_heartbeat"] = _hb or "unknown" + + if monitoring_active_db and not thread_alive: + if _hb: + try: + hb = datetime.fromisoformat(_hb.replace("Z", "+00:00")) + if hb.tzinfo is None: + hb = hb.replace(tzinfo=dt_timezone.utc) + if (datetime.now(tz=dt_timezone.utc) - hb).total_seconds() > 600: + issues.append("warning:monitoring active but heartbeat is stale (>10 min)") + except Exception: + issues.append("warning:monitoring active but heartbeat is unparseable") + else: + issues.append("warning:monitoring active but no heartbeat found") - last_poll = settings.get("last_poll_result", {}) - details["last_poll_started_at"] = last_poll.get("started_at", "never") - details["last_poll_completed_at"] = last_poll.get("completed_at", "never") - details["last_poll_status"] = last_poll.get("state", "unknown") - details["last_poll_added"] = last_poll.get("added", 0) - details["last_poll_ended"] = last_poll.get("ended", 0) - details["last_poll_cleaned"] = last_poll.get("cleaned", 0) - details["last_poll_error"] = last_poll.get("error") + # Stale-poll warning — active but last_poll is beyond the expected cycle window + if monitoring_active_db and not self._is_last_poll_recent(runtime): + _poll_age_str = f"{int(_lpa)}s" if _lpa is not None else "never" + issues.append(f"warning:monitoring active but last poll is stale (age={_poll_age_str})") # EPG window counts — surface current/future program counts for the YouTubearr source _epg_window = self._get_youtubearr_epg_window_counts(settings) details["epg_window_counts"] = _epg_window - if details["monitoring_active"] and _epg_window.get("source_found"): + if monitoring_active_db and _epg_window.get("source_found"): if _epg_window.get("current", 0) == 0 and _epg_window.get("future12", 0) == 0: issues.append("warning:YouTubearr EPG source has no current or future programs — monitor may need refresh") @@ -915,6 +1061,7 @@ def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: # Stale stream URLs (live streams whose URL hasn't been refreshed recently) _url_refresh_interval = settings.get("url_refresh_interval_seconds", 3600) _stale_threshold = 2 * _url_refresh_interval + _now_utc = datetime.now(dt_timezone.utc) stale_count = 0 oldest_stale_age = 0.0 for _vid, _sd in tracked_streams.items(): @@ -924,12 +1071,16 @@ def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: if not _last_str: stale_count += 1 continue - _age = self._age_seconds(_last_str) - if _age is None: - stale_count += 1 - elif _age > _stale_threshold: + try: + _lr = datetime.fromisoformat(_last_str.replace("Z", "+00:00")) + if _lr.tzinfo is None: + _lr = _lr.replace(tzinfo=dt_timezone.utc) + _age = (_now_utc - _lr).total_seconds() + if _age > _stale_threshold: + stale_count += 1 + oldest_stale_age = max(oldest_stale_age, _age) + except (ValueError, TypeError): stale_count += 1 - oldest_stale_age = max(oldest_stale_age, _age) details["stale_tracked_stream_url_count"] = stale_count if oldest_stale_age: details["oldest_url_refresh_age_seconds"] = int(oldest_stale_age) @@ -940,7 +1091,7 @@ def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: orphaned_tracked_count = 0 stale_epg_tracked_count = 0 try: - _diag_now = timezone.now() + _diag_now = datetime.now(dt_timezone.utc) for _vid, _sd in tracked_streams.items(): if not _sd.get("is_live"): continue @@ -954,10 +1105,8 @@ def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: _prog = ProgramData.objects.filter(epg=_ch.epg_data).first() if _prog is not None and _prog.end_time is not None: _end = _prog.end_time - if _diag_now.tzinfo is not None and getattr(_end, "tzinfo", None) is None: + if getattr(_end, "tzinfo", None) is None: _end = _end.replace(tzinfo=dt_timezone.utc) - elif _diag_now.tzinfo is None and getattr(_end, "tzinfo", None) is not None: - _end = _end.astimezone().replace(tzinfo=None) if _end < _diag_now: stale_epg_tracked_count += 1 except Channel.DoesNotExist: @@ -973,17 +1122,32 @@ def _handle_diagnostics(self, context: Dict[str, Any]) -> Dict[str, Any]: if stale_epg_tracked_count > 0: issues.append(f"warning:{stale_epg_tracked_count} tracked-live stream(s) have expired EPG data (may be stale)") - # Log file location - details["log_file_path"] = str(self._log_path) - details["log_file_exists"] = self._log_path.exists() - - # Manual refresh state (visible outcome for operators) - details["last_manual_refresh"] = dict(self._last_manual_refresh) - details["force_refresh_pending"] = bool(settings.get("force_refresh_requested_at")) - - # Recent log summary + # Log file path and recent summary + details["log_path"] = str(self._log_path) details["log_summary"] = self._get_recent_log_summary() + # Next-action hints for operators + next_actions: List[str] = [] + for _issue in issues: + if "yt-dlp binary not found" in _issue: + next_actions.append("Install or update yt-dlp in the plugin directory and reload the plugin") + elif "heartbeat is stale" in _issue or "no heartbeat found" in _issue: + next_actions.append("Click 'Start Monitoring' to restart the monitoring thread") + elif "last poll is stale" in _issue: + next_actions.append("Click 'Refresh Now' to trigger an immediate poll, or restart monitoring") + elif "EPG source has no current or future programs" in _issue: + next_actions.append("Click 'Refresh Now' to trigger an EPG refresh for the YouTubearr source") + elif "point to missing channels" in _issue: + next_actions.append("Run 'Cleanup Ended Streams' to remove orphaned tracked entries") + elif "stale (monitor may not be refreshing URLs)" in _issue: + next_actions.append("Check logs for URL refresh errors; monitoring may be stalled") + elif "legacy Celery beat task present" in _issue: + next_actions.append("Reload the plugin page — the legacy health-check task is removed automatically") + elif "expired EPG data" in _issue: + next_actions.append("Run 'Refresh Now' — EPG program data for live streams appears stale") + if next_actions: + details["next_actions"] = list(dict.fromkeys(next_actions)) # deduplicate, preserve order + # Status errors = [i for i in issues if i.startswith("error:")] warnings = [i for i in issues if i.startswith("warning:")] @@ -1075,7 +1239,12 @@ def _get_epg_counts(self, settings: Dict[str, Any]) -> Dict[str, Any]: return result def _get_recent_log_summary(self) -> Dict[str, Any]: - result: Dict[str, Any] = {"error_count": 0, "recent_errors": [], "status": "ok"} + result: Dict[str, Any] = { + "error_count": 0, "recent_errors": [], + "warning_count": 0, "recent_warnings": [], + "recent_lines": [], + "status": "ok", + } if not self._log_path.exists(): result["status"] = "log file not found" return result @@ -1091,8 +1260,14 @@ def _get_recent_log_summary(self) -> Dict[str, Any]: if size > max_bytes and lines: lines = lines[1:] # drop possibly-truncated first line error_lines = [l for l in lines if "ERROR:" in l] + warn_lines = [l for l in lines if "WARNING:" in l or "WARN:" in l] result["error_count"] = len(error_lines) result["recent_errors"] = [e[-120:] for e in error_lines[-5:]] + result["warning_count"] = len(warn_lines) + result["recent_warnings"] = [w[-120:] for w in warn_lines[-5:]] + # Last 20 lines give operators a live tail of plugin activity. + # Log content is safe: the plugin never writes cookies or auth tokens to the log. + result["recent_lines"] = [l[-200:] for l in lines[-20:]] result["lines_scanned"] = len(lines) except Exception as exc: result["status"] = f"read failed: {type(exc).__name__}" @@ -2088,14 +2263,12 @@ def _poll_monitored_channels(self, settings: Dict[str, Any]) -> tuple[int, int]: if not metadata: self._log_error(f"Failed to extract metadata for {video_id} - yt-dlp returned None") self._extraction_failures[video_id] = time.time() - self._persist_extraction_failures() continue if metadata.get("_members_only"): self._log(f"Skipping {video_id}: members-only content (retry in 7 days)") # Store time 6 days in the future so the 24h check won't clear it for 7 days total self._extraction_failures[video_id] = time.time() + 86400 * 6 - self._persist_extraction_failures() continue self._log(f"Metadata extracted for {video_id}: is_live={metadata.get('is_live')}, title={metadata.get('title')}") @@ -2172,11 +2345,9 @@ def _poll_monitored_channels(self, settings: Dict[str, Any]) -> tuple[int, int]: except Exception as exc: self._log_error(f"Failed to poll channel {channel_id}: {exc}") - # Persist final state - self._persist_settings({ - "tracked_streams": tracked_streams, - "last_poll_time": timezone.now().isoformat(), - }) + # Persist tracked_streams to settings; last_poll_time goes to runtime_state + self._persist_settings({"tracked_streams": tracked_streams}) + self._write_runtime_state({"last_poll_time": timezone.now().isoformat()}) return added_count, ended_count @@ -2753,11 +2924,191 @@ def _cleanup_ended_streams(self, settings: Dict[str, Any], force: bool = False) # --- Monitoring Thread --- - def _persist_extraction_failures(self) -> None: - """Persist non-expired extraction failures to DB so they survive container restarts.""" - now = time.time() - to_save = {vid: t for vid, t in self._extraction_failures.items() if t + 86400 > now} - self._persist_settings({"extraction_failures": to_save}) + def _monitoring_loop(self, plugin_key: str) -> None: + """Background monitoring loop (runs in daemon thread). + + The exclusive file lock is already held by the caller before this starts. + Heartbeat goes to runtime_state.json, not settings, so Dispatcharr form + saves cannot overwrite it. + """ + self._log("Monitoring loop started") + + self._write_runtime_state({ + "started_at": timezone.now().isoformat(), + "last_heartbeat_at": timezone.now().isoformat(), + }) + + try: + while not self._monitor_stop_event.is_set(): + try: + # In-memory flag is the stop signal for this process + if not self._monitoring_active: + self._log("Monitoring disabled (in-memory flag), stopping") + break + + # Cross-worker stop: another worker process may have written + # desired_active=False to the shared runtime_state.json (via + # Stop Monitoring or Reset All). The in-memory flag and + # stop_event above are per-process and invisible to other + # workers, so this owner loop must check the shared state + # itself to notice and relinquish the lock. + if not self._read_runtime_state().get("desired_active", True): + self._log("Monitoring disabled (desired_active=False from another worker), stopping") + break + + # Reload operator config from DB each cycle + try: + cfg = PluginConfig.objects.get(key=plugin_key) + settings = dict(cfg.settings or {}) + except PluginConfig.DoesNotExist: + self._log_error("Plugin config not found, stopping monitoring") + break + + # Update heartbeat in runtime_state (not settings) + self._write_runtime_state({"last_heartbeat_at": timezone.now().isoformat()}) + + # Prune stale extraction failures to keep the dict bounded + try: + _pruned = self._prune_extraction_failures() + if _pruned: + self._log(f"Pruned {_pruned} stale extraction failure(s)") + except Exception: + pass + + # Poll channels + try: + added, ended = self._poll_monitored_channels(settings) + self._refresh_expiring_urls(settings) + self._refresh_epg_times(settings) + + if settings.get("auto_cleanup", True): + cleaned = self._cleanup_ended_streams(settings) + else: + cleaned = 0 + + if added > 0 or cleaned > 0: + self._trigger_webhook(settings) + + except Exception as exc: + self._log_error(f"Poll cycle error: {exc}") + + # Sleep for poll interval in small chunks to respond to stop signal, + # checking shared desired_active each second so a cross-worker Stop + # or Reset All is noticed without waiting out the full poll interval. + poll_interval = settings.get("poll_interval_minutes", 15) + for _ in range(int(poll_interval * 60)): + if self._monitor_stop_event.is_set(): + break + if not self._read_runtime_state().get("desired_active", True): + self._log("Monitoring disabled during sleep (desired_active=False from another worker), stopping") + self._monitor_stop_event.set() + break + time.sleep(1) + + except Exception as exc: + self._log_error(f"Monitoring loop error: {exc}") + time.sleep(60) + + finally: + # Clear in-memory flag and heartbeat; release the lock. + # desired_active in runtime_state is NOT cleared here — lifecycle exits + # (container restart, plugin reload) preserve desired intent so + # _ensure_monitoring_thread can auto-restart. Only _handle_stop_monitoring + # clears desired_active. + self._log("Monitoring loop exiting") + self._monitoring_active = False + self._write_runtime_state({"last_heartbeat_at": None}) + self._release_monitor_lock() + + self._log("Monitoring loop stopped") + + # --- State Management --- + + def _read_runtime_state(self) -> Dict[str, Any]: + """Read the sidecar runtime_state.json file. Returns empty dict on any error.""" + try: + return json.loads(self._runtime_state_path.read_text()) + except Exception: + return {} + + def _write_runtime_state(self, updates: Dict[str, Any]) -> None: + """Merge updates into runtime_state.json, serialized across threads/workers. + + Without serialization, two concurrent read-modify-write cycles (e.g. the + monitor loop's heartbeat write racing a Stop/Reset from another worker) + can lose an update: whichever writer read stale data last overwrites the + other's change. A threading.Lock serializes writers within this process; + a blocking flock on a dedicated lock file (separate from monitor.lock) + serializes writers across worker processes. + """ + with self._runtime_state_thread_lock: + try: + with open(str(self._runtime_state_lock_path), 'w') as lock_fd: + fcntl.flock(lock_fd, fcntl.LOCK_EX) + try: + state = self._read_runtime_state() + state.update(updates) + tmp = self._runtime_state_path.with_suffix(".json.tmp") + tmp.write_text(json.dumps(state)) + tmp.replace(self._runtime_state_path) + finally: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + except Exception as exc: + self._log_error(f"Failed to write runtime state: {exc}") + + def _acquire_monitor_lock(self) -> bool: + """Try to acquire the exclusive monitor file lock (non-blocking). + + Uses fcntl.flock so the OS releases the lock automatically if this + process dies, preventing a permanently stuck state. Returns True if + the lock was acquired and stored in self._lock_fd. + """ + try: + fd = open(str(self._lock_path), 'w') + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + self._lock_fd = fd + return True + except (IOError, OSError): + try: + fd.close() + except Exception: + pass + return False + except Exception: + return False + + def _release_monitor_lock(self) -> None: + """Release the exclusive monitor file lock if held.""" + if self._lock_fd is not None: + try: + fcntl.flock(self._lock_fd, fcntl.LOCK_UN) + self._lock_fd.close() + except Exception: + pass + self._lock_fd = None + + def _is_monitor_lock_held_by_other(self) -> bool: + """Non-blocking probe: True if another process currently holds monitor.lock. + + Returns False if this process already holds it (self._lock_fd is set) or + if the lock is currently free. Used to distinguish "another worker is + genuinely running the monitor loop" from "nothing is monitoring + anywhere" without acquiring or disturbing lock ownership. + """ + if self._lock_fd is not None: + return False + try: + probe_fd = open(str(self._lock_path), 'w') + except Exception: + return False + try: + fcntl.flock(probe_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(probe_fd, fcntl.LOCK_UN) + return False + except (IOError, OSError): + return True + finally: + probe_fd.close() def _persist_settings(self, updates: Dict[str, Any]) -> None: """Persist settings updates to database (thread-safe)""" @@ -3136,33 +3487,14 @@ def _cleanup_legacy_celery_task(self) -> None: except Exception as exc: self._log_error(f"Legacy Celery task cleanup failed: {exc}") - def _current_datetime(self) -> datetime: - result = timezone.now() - if isinstance(result, datetime): - return result - return datetime.now(dt_timezone.utc) - def _parse_iso_datetime(self, value) -> Optional[datetime]: - """Parse an ISO-8601 datetime string in the same timezone mode as timezone.now(). - - When Django USE_TZ=True, timestamps are written as tz-aware UTC; return aware. - When Django USE_TZ=False, timestamps are written as naive local time; return naive. - Forcing all naive timestamps to UTC caused cross-worker health check failures when - the server was not in UTC — ages appeared hours off, triggering false restart loops. - """ + """Parse an ISO-8601 datetime string safely. Returns None on any failure.""" if not value or not isinstance(value, str): return None try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) - _now = self._current_datetime() - if _now.tzinfo is not None: - # Django tz-aware mode: ensure result is also aware (assume UTC if naive) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=dt_timezone.utc) - else: - # Django tz-naive mode: strip tzinfo so comparison with timezone.now() works - if dt.tzinfo is not None: - dt = dt.astimezone().replace(tzinfo=None) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=dt_timezone.utc) return dt except (ValueError, TypeError): return None @@ -3172,7 +3504,16 @@ def _age_seconds(self, value) -> Optional[float]: dt = self._parse_iso_datetime(value) if dt is None: return None - return max(0.0, (timezone.now() - dt).total_seconds()) + return max(0.0, (datetime.now(dt_timezone.utc) - dt).total_seconds()) + + def _is_last_poll_recent(self, settings: Dict[str, Any]) -> bool: + """Return True if last_poll_time is within (poll_interval + 10 min) — same grace as heartbeat.""" + age = self._age_seconds(settings.get("last_poll_time")) + if age is None: + return False # Never polled or unparseable + poll_interval_minutes = settings.get("poll_interval_minutes", 15) + threshold = (poll_interval_minutes + 10) * 60 + return age < threshold def _get_youtubearr_epg_window_counts(self, settings: Dict[str, Any]) -> Dict[str, Any]: """Return current/future-12h program counts for the configured EPG source. @@ -3188,7 +3529,7 @@ def _get_youtubearr_epg_window_counts(self, settings: Dict[str, Any]) -> Dict[st if not source: return result result["source_found"] = True - now = timezone.now() + now = datetime.now(dt_timezone.utc) future12 = now + timedelta(hours=12) result["current"] = ProgramData.objects.filter( epg__epg_source=source, @@ -3204,6 +3545,41 @@ def _get_youtubearr_epg_window_counts(self, settings: Dict[str, Any]) -> Dict[st pass return result + def _ensure_monitoring_thread(self, settings: Dict[str, Any]) -> bool: + """Restart the monitor thread if desired_active but no live thread is running. + + Handles container restarts and crashed threads via the file lock: if the lock + can be acquired, no other process is monitoring, so we start. Returns True if + a new thread was started. + """ + runtime = self._read_runtime_state() + if not runtime.get("desired_active"): + return False + + if self._monitor_thread and self._monitor_thread.is_alive(): + return False + + channels = settings.get("monitored_channels", "").strip() + if not channels or not self._ytdlp_path: + return False + + # Try to acquire the lock — if another process holds it, it's already running + if not self._acquire_monitor_lock(): + self._log("Auto-restart skipped: monitor lock held by another process") + return False + + self._log("Auto-restarting monitoring after service restart") + self._monitoring_active = True + self._monitor_stop_event.clear() + self._monitor_thread = threading.Thread( + target=self._monitoring_loop, + args=(self._plugin_key,), + daemon=True, + name="YouTubearr-Monitor" + ) + self._monitor_thread.start() + return True + def _log(self, message: str) -> None: """Write log message""" timestamp = datetime.now().isoformat() From 325d1993d173def6af556c2a322ee4bc25541430 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 31 Jul 2026 18:23:02 -0400 Subject: [PATCH 2/2] [youtubearr] Bump version to 1.30.1 --- plugins/youtubearr/plugin.json | 2 +- plugins/youtubearr/plugin.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/youtubearr/plugin.json b/plugins/youtubearr/plugin.json index 6a17c84..e5a3527 100644 --- a/plugins/youtubearr/plugin.json +++ b/plugins/youtubearr/plugin.json @@ -1,6 +1,6 @@ { "name": "YouTubearr", - "version": "1.30.0", + "version": "1.30.1", "description": "Zero-dependency YouTube livestream plugin with automatic monitoring and configurable numbering", "author": "jeff-gooch", "license": "Unlicense", diff --git a/plugins/youtubearr/plugin.py b/plugins/youtubearr/plugin.py index 445e2b5..552488e 100644 --- a/plugins/youtubearr/plugin.py +++ b/plugins/youtubearr/plugin.py @@ -25,7 +25,7 @@ class Plugin: name = "YouTubearr" - version = "1.30.0" + version = "1.30.1" description = "Zero-dependency YouTube livestream plugin with automatic monitoring and configurable numbering" author = "Jeff Gooch" help_url = "https://github.com/jeff-gooch/youtubearr"