scheduler.py
Rama main · réplica de lectura del repositorio original.
"""
scheduler.py – Manages schedules
"""
import json
import uuid
import random
import threading
from typing import Dict, List, Optional, Any
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.jobstores.base import JobLookupError
import app as core
import app.items as items_api
from app.cache import refresh_all_caches
import app_state
import database
from routers.dependencies import get_auth_data
from app.logger import get_logger
logger = get_logger("MixerBee.Scheduler")
import logging
logging.getLogger('apscheduler').setLevel(logging.INFO if app_state.VERBOSE_LOGGING else logging.WARNING)
from quick_playlist_registry import QUICK_PLAYLIST_MAP
# The UI numbers days with Sunday = 0; APScheduler's crontab parser uses Monday = 0.
# Crontabs are written with day names so neither numbering can be misread.
UI_DAY_NAMES = ("sun", "mon", "tue", "wed", "thu", "fri", "sat")
def build_crontab(details: Dict) -> str:
"""Build the stored schedule string from a ScheduleDetails dict.
Returns ``interval:<minutes>`` for interval schedules, otherwise a five-field
crontab with named weekdays. Raises ValueError for unusable details.
"""
frequency = details.get("frequency")
if frequency == "interval":
minutes = details.get("interval_minutes")
if not minutes or minutes < 1:
raise ValueError("Interval minutes must be at least 1.")
return f"interval:{minutes}"
try:
hour, minute = (int(part) for part in str(details.get("time") or "").split(":"))
except ValueError:
raise ValueError("time must be in HH:MM format.") from None
if not (0 <= hour <= 23 and 0 <= minute <= 59):
raise ValueError("time must be in HH:MM format.")
if frequency == "weekly":
days = details.get("days_of_week")
if not days:
raise ValueError("days_of_week must be provided for weekly frequency.")
if any(not isinstance(d, int) or not 0 <= d <= 6 for d in days):
raise ValueError("days_of_week values must be 0 (Sunday) through 6 (Saturday).")
day_field = ",".join(UI_DAY_NAMES[d] for d in sorted(set(days)))
elif frequency == "daily":
day_field = "*"
else:
raise ValueError(f"Unknown schedule frequency: '{frequency}'.")
crontab = f"{minute} {hour} * * {day_field}"
CronTrigger.from_crontab(crontab)
return crontab
def _legacy_crontab_with_day_names(crontab: str) -> str:
"""Rewrite a legacy numeric weekday field (UI numbering, Sunday = 0) to day names."""
fields = crontab.split()
if len(fields) == 5:
tokens = fields[4].split(",")
if all(t.isdigit() and int(t) <= 6 for t in tokens):
fields[4] = ",".join(UI_DAY_NAMES[int(t)] for t in tokens)
return " ".join(fields)
LEGACY_TYPE_MAP = {"continue_watching": "next_up", "forgotten_favorites": "from_the_vault"}
def run_playlist_job(**schedule_data) -> Dict:
if "schedule_data" in schedule_data and isinstance(schedule_data["schedule_data"], dict):
nested = schedule_data.pop("schedule_data")
schedule_data = {**nested, **schedule_data}
schedule_id = schedule_data.get("id") or schedule_data.get("schedule_id")
user_id = schedule_data.get("user_id")
playlist_name = schedule_data.get("playlist_name")
raw_type = schedule_data.get("job_type", "builder")
job_type = "builder" if raw_type == "preset" else raw_type
log_messages = []
if not all([user_id, playlist_name]):
msg = f"Job '{schedule_id}' is missing required data (User or Playlist Name). Aborting."
logger.error(msg)
return {"status": "error", "log": [msg]}
logger.info(f"Running job '{schedule_id}' for playlist '{playlist_name}' (Type: {job_type}) for user {user_id}")
result = {}
try:
connection_id = schedule_data.get("connection_id")
if not connection_id:
raise ValueError("Schedule has no assigned connection. Recreate it under the intended account.")
# Policy first, before the media server is contacted: a suspended enrichment run
# should cost nothing and must not report a connection error. The schedule and its
# enabled flag are left alone, so no failure notification is produced and the next
# occurrence is normal once AI is turned back on.
if job_type == "enrichment":
from app.ai_policy import generative_available
if not generative_available(connection_id):
msg = (f"Enrichment job skipped for connection {connection_id}: AI features are "
"unavailable for this account. The schedule is suspended, not deleted.")
logger.info(msg)
return {"status": "ok", "skipped": True, "suspended": True, "log": [msg]}
media = schedule_data.get("media")
if not media:
auth_data = get_auth_data(connection_id)
media = auth_data["media"].require_user(user_id)
if job_type == "enrichment":
from app.ai.enrichment_manager import enrichment_guard
enrich_data = schedule_data.get("enrichment_data", {})
batch_size = enrich_data.get("batch_size", 15)
timeout = enrich_data.get("timeout", 120)
with enrichment_guard(connection_id) as acquired:
if not acquired:
msg = f"Enrichment job skipped for connection {connection_id}: another enrichment process is currently running."
logger.warning(msg)
return {"status": "ok", "skipped": True, "log": [msg]}
from app.ai import process_enrichment_queue
result = process_enrichment_queue(batch_size=batch_size, timeout=timeout, media=media)
else:
import preset_manager as pm
from routers.builder import _get_random_movie_block, _get_random_tv_block
if job_type == "builder":
blocks = schedule_data.get("blocks")
preset_id = schedule_data.get("preset_id")
preset_name = schedule_data.get("preset_name")
mix_options = schedule_data.get("mix_options")
if not blocks and preset_id:
record = pm.preset_manager.get_preset_by_id(preset_id, connection_id)
if not record:
msg = f"The preset assigned to job '{schedule_id}' no longer exists. Reassign the schedule."
logger.error(msg)
return {"status": "error", "log": [msg]}
blocks = record['data']
preset_name = record['name']
if not mix_options and record.get('mix_options'):
mix_options = record['mix_options']
logger.info("Resolved blocks for job %s from preset ID %s ('%s')", schedule_id, preset_id, preset_name)
elif not blocks and preset_name:
# Compatibility for jobs created before preset IDs were introduced.
record = pm.preset_manager.get_preset_by_name(preset_name, connection_id)
if not record:
msg = f"Preset '{preset_name}' assigned to job '{schedule_id}' no longer exists. Reassign the schedule."
logger.error(msg)
return {"status": "error", "log": [msg]}
blocks = record['data']
if not mix_options and record.get('mix_options'):
mix_options = record['mix_options']
logger.info(f"Resolved legacy job {schedule_id} from preset '{preset_name}'")
elif preset_id and not mix_options:
record = pm.preset_manager.get_preset_by_id(preset_id, connection_id)
if record and record.get('mix_options'):
mix_options = record['mix_options']
if not blocks:
logger.info(f"No blocks found for job {schedule_id}.")
potential_options = [b for b in [_get_random_movie_block(media), _get_random_tv_block(media)] if b is not None]
if potential_options:
blocks = [random.choice(potential_options)]
else:
msg = "Could not generate a block; library appears to be empty."
logger.error(msg)
return {"status": "error", "log": [msg]}
create_as_collection = schedule_data.get("create_as_collection", False)
if create_as_collection:
if len(blocks) != 1 or blocks[0].get("type") != "movie":
msg = "Scheduled collections must consist of exactly one Movie block."
logger.error(msg)
return {"status": "error", "log": [msg]}
result = items_api.create_movie_collection(
user_id=user_id,
collection_name=playlist_name,
filters=blocks[0].get("filters", {}),
media=media
)
else:
trigger_src = schedule_data.get("trigger_source", "clock")
result = core.create_mixed_playlist(
user_id=user_id,
playlist_name=playlist_name,
blocks=blocks,
media=media,
mix_options=mix_options,
trigger_source=trigger_src,
schedule_id=schedule_id,
preset_id=preset_id
)
elif job_type == "quick_playlist":
quick_playlist_data = schedule_data.get("quick_playlist_data", {})
quick_playlist_type = quick_playlist_data.get("quick_playlist_type")
func_to_call = QUICK_PLAYLIST_MAP.get(quick_playlist_type)
if not func_to_call:
raise ValueError(f"Unknown quick_playlist_type '{quick_playlist_type}'")
options = dict(quick_playlist_data.get("options", {}))
if quick_playlist_type == "album_roulette":
if not options.get("album_id") or options.get("album_id") == "random":
rand_album = core.get_random_album(media)
if rand_album and (rand_album.get("Id") or rand_album.get("id")):
options["album_id"] = rand_album.get("Id") or rand_album.get("id")
if playlist_name in ("New Scheduled Mix", "Scheduled Auto Playlist", "Album Roulette"):
playlist_name = f"Album: {rand_album.get('Name') or rand_album.get('name')}"
else:
msg = "No albums found in library for Album Roulette."
logger.error(msg)
return {"status": "error", "log": [msg]}
result = func_to_call(user_id=user_id, playlist_name=playlist_name, media=media, log=log_messages, **options)
final_log = result.get("log", ["No log messages returned from build process."])
final_status = result.get("status", "error")
logger.info(f"Job '{schedule_id}' for '{playlist_name}' completed with status: {final_status.upper()}.")
except Exception as e:
error_message = f"CRITICAL ERROR running job '{schedule_id}': {e}"
logger.error(error_message, exc_info=True)
result = {"status": "error", "log": [error_message]}
return result
# Cap on back-to-back reruns triggered by requests that arrive while a schedule is
# Cap on back-to-back reruns triggered by requests that arrive while a schedule is
# already running, so a heavy burst of webhook events can't loop indefinitely.
MAX_RERUN_PASSES = 3
def is_automatic_run_allowed(schedule: Dict, source: str, now: Optional[datetime] = None) -> bool:
"""
Unified policy check for whether an execution triggered by `source` is permitted.
Manual runs ("manual") are always permitted even if paused/snoozed.
"""
if source == "manual":
return True
if schedule.get("enabled", True) is False:
return False
snoozed_until_str = schedule.get("snoozed_until")
if snoozed_until_str:
try:
from datetime import timezone
now_dt = now or datetime.now(timezone.utc)
if now_dt.tzinfo is None:
now_dt = now_dt.replace(tzinfo=timezone.utc)
snooze_dt = datetime.fromisoformat(snoozed_until_str.replace("Z", "+00:00"))
if snooze_dt.tzinfo is None:
snooze_dt = snooze_dt.replace(tzinfo=timezone.utc)
if now_dt < snooze_dt:
return False
except Exception:
pass
trigger_sources = schedule.get("trigger_sources")
if trigger_sources is not None and isinstance(trigger_sources, list):
if source not in trigger_sources:
return False
return True
def _run_once_and_record(schedule_data: Dict, schedule_id: Optional[str]):
result = run_playlist_job(**schedule_data)
if schedule_id:
last_run_info = {
"timestamp": datetime.now().isoformat(),
"status": result.get("status", "error"),
"log": result.get("log", ["An unknown error occurred."])
}
scheduler_manager._update_schedule_last_run(schedule_id, last_run_info)
def _schedule_exists_in_db(schedule_id: str) -> bool:
"""Authoritative existence check for a schedule about to run.
APScheduler holds its jobs in memory, and the dispatched kwargs are a snapshot
taken when the job was added. Neither notices a schedule deleted from the database
(directly, or by a backup restore) once a run has already been handed to a pool
thread, so the run must confirm its schedule is still real before building.
"""
try:
with database.get_db_connection() as conn:
row = conn.execute("SELECT 1 FROM schedules WHERE id = ?", (schedule_id,)).fetchone()
return row is not None
except Exception as e:
# A database problem is not evidence the schedule was deleted; let the run proceed.
logger.warning("Could not verify schedule %s before running: %s", schedule_id, e)
return True
def _cancel_orphaned_jobs(schedule_id: str):
"""Drop APScheduler jobs left behind for a schedule that no longer exists."""
for job_id in (schedule_id, f"run_{schedule_id}"):
try:
scheduler_manager.scheduler.remove_job(job_id)
logger.info("Removed orphaned job %s for deleted schedule %s", job_id, schedule_id)
except JobLookupError:
pass
except Exception:
pass
def scheduled_job_wrapper(**schedule_data):
schedule_id = schedule_data.get("id")
source = schedule_data.get("trigger_source", "clock")
# Cron runs, webhook-triggered runs, and manual "Run Now" runs all funnel through
# here, so guarding on the schedule id here is enough to keep any two runs of the
# same schedule from executing (and clobbering create_playlist) at the same time.
if not schedule_id:
logger.warning("scheduled_job_wrapper received schedule data with no 'id'; running unguarded.")
_run_once_and_record(schedule_data, None)
return
if not _schedule_exists_in_db(schedule_id):
logger.warning(
"Skipping run for schedule '%s': it no longer exists in the database.", schedule_id
)
scheduler_manager.schedules.pop(schedule_id, None)
_cancel_orphaned_jobs(schedule_id)
return
# Execution-time policy recheck:
current_schedule = scheduler_manager.schedules.get(schedule_id, schedule_data)
if not is_automatic_run_allowed(current_schedule, source):
logger.info(
f"Execution skipped for schedule '{schedule_id}': automatic run not allowed for source '{source}' "
f"(enabled={current_schedule.get('enabled', True)}, snoozed_until={current_schedule.get('snoozed_until')})."
)
return
lock = scheduler_manager._get_schedule_lock(schedule_id)
if not lock.acquire(blocking=False):
scheduler_manager._mark_rerun_pending(schedule_id)
logger.info(f"Schedule '{schedule_id}' is already running; queued a rerun instead of overlapping.")
return
try:
current_data = schedule_data
for pass_num in range(1, MAX_RERUN_PASSES + 1):
if pass_num > 1:
if not _schedule_exists_in_db(schedule_id):
logger.warning(
"Aborting pending rerun for '%s': schedule was deleted mid-run.", schedule_id
)
break
# Recheck policy before queued rerun pass
current_data = scheduler_manager.schedules.get(schedule_id, current_data)
if not is_automatic_run_allowed(current_data, source):
logger.info(f"Aborting pending rerun for '{schedule_id}': schedule state changed to paused/snoozed.")
break
_run_once_and_record(current_data, schedule_id)
if not scheduler_manager._consume_rerun_pending(schedule_id):
break
if pass_num == MAX_RERUN_PASSES:
logger.warning(
f"Schedule '{schedule_id}' hit the {MAX_RERUN_PASSES}-pass rerun cap; "
"dropping the pending rerun."
)
break
# Pick up any edits saved while this schedule was running instead of
# rerunning with the (possibly now-stale) data captured at dispatch time.
current_data = scheduler_manager.schedules.get(schedule_id, current_data)
finally:
lock.release()
def vibe_index_catchup_job():
"""Retry any vibe index that has not succeeded in this process.
A media server that was unreachable at startup used to leave its vibe index
unbuilt until a restart, because indexing was startup-only work.
This runs as its own APScheduler job rather than inside the cache refresh:
max_instances defaults to 1 per job, so an initial index of a large library --
minutes of embedding -- would otherwise make every cache refresh fire in that
window be rejected as "maximum number of running instances reached".
"""
try:
from app.ai.vector_store import connection_needs_index, ensure_library_indexed
from connections import all_media_clients
except Exception as e:
logger.warning("Skipping semantic index catch-up: %s", e)
return
for media in all_media_clients():
try:
# No provider check: this index serves Echo blocks and similarity search on
# every usable connection, configured for AI or not.
if connection_needs_index(media.connection.id):
logger.info("Retrying vibe index for connection %s.", media.connection.id)
ensure_library_indexed(media.user_id, media)
except Exception as e:
logger.warning("Vibe index retry failed for connection %s: %s", media.connection.id, e)
class Scheduler:
def __init__(self):
# APScheduler's default misfire_grace_time is 1 second, which silently drops
# any fire that had to wait longer than that -- e.g. a cron that came due while
# the container was stopped or the host asleep, or while the executor pool was
# busy with a long build. coalesce collapses a backlog into a single catch-up run.
self.scheduler = BackgroundScheduler(
daemon=True,
job_defaults={'misfire_grace_time': 300, 'coalesce': True}
)
self.schedules: Dict[str, Dict] = {}
# Per-schedule run lock + "a rerun was requested while running" flag, keyed by
# schedule id. _schedule_locks_guard protects both dicts so concurrent first-time
# access for the same schedule id can't create two different Lock objects.
self._schedule_locks: Dict[str, threading.Lock] = {}
self._rerun_pending: Dict[str, bool] = {}
self._schedule_locks_guard = threading.Lock()
def _get_schedule_lock(self, schedule_id: str) -> threading.Lock:
with self._schedule_locks_guard:
lock = self._schedule_locks.get(schedule_id)
if lock is None:
lock = threading.Lock()
self._schedule_locks[schedule_id] = lock
return lock
def _mark_rerun_pending(self, schedule_id: str):
with self._schedule_locks_guard:
self._rerun_pending[schedule_id] = True
def _consume_rerun_pending(self, schedule_id: str) -> bool:
with self._schedule_locks_guard:
pending = self._rerun_pending.get(schedule_id, False)
self._rerun_pending[schedule_id] = False
return pending
def _get_trigger(self, schedule_data: Dict):
details = schedule_data.get("schedule_details") or {}
frequency = details.get("frequency")
if frequency == "interval":
mins = details.get("interval_minutes", 30)
return IntervalTrigger(minutes=mins)
# Rebuild from schedule_details rather than trusting the stored crontab: rows
# saved before day names were used hold UI day numbers (Sunday = 0), which
# APScheduler reads as Monday = 0 and fires a day late.
if frequency in ("daily", "weekly"):
try:
return CronTrigger.from_crontab(build_crontab(details))
except ValueError:
logger.warning("Schedule %s has unusable schedule details; using its stored crontab.",
schedule_data.get("id"))
crontab = schedule_data.get("crontab")
if crontab and not crontab.startswith("interval:"):
return CronTrigger.from_crontab(_legacy_crontab_with_day_names(crontab))
return None
@staticmethod
def _bind_legacy_preset(schedule_data: Dict):
"""Upgrade internal callers that still submit a preset name without its ID."""
if schedule_data.get('job_type') not in ('builder', 'preset') or schedule_data.get('blocks'):
schedule_data['preset_id'] = None
schedule_data['preset_name'] = None
return schedule_data
if (not schedule_data.get('preset_id') and schedule_data.get('preset_name')
and schedule_data.get('connection_id')):
import preset_manager as pm
record = pm.preset_manager.get_preset_by_name(
schedule_data['preset_name'], schedule_data['connection_id']
)
if record:
schedule_data['preset_id'] = record['id']
schedule_data['preset_name'] = record['name']
return schedule_data
def _update_schedule_last_run(self, schedule_id: str, last_run_info: Dict):
"""Safely updates the last_run status in both the DB and in-memory cache."""
try:
with database.get_db_connection() as conn:
last_run_json = json.dumps(last_run_info)
conn.execute("UPDATE schedules SET last_run = ? WHERE id = ?", (last_run_json, schedule_id))
conn.commit()
if schedule_id in self.schedules:
self.schedules[schedule_id]['last_run'] = last_run_info
except Exception as e:
logger.error(f"Error updating last run status for {schedule_id} in DB: {e}", exc_info=True)
def _load_schedules(self) -> Dict[str, Dict]:
schedules = {}
try:
with database.get_db_connection() as conn:
rows = conn.execute('''SELECT s.id, s.playlist_name, s.user_id, s.job_type, s.crontab,
s.config_data, s.last_run, s.connection_id, s.preset_id,
p.name AS current_preset_name
FROM schedules s LEFT JOIN connection_presets p ON p.id=s.preset_id''').fetchall()
for row in rows:
schedule_data = dict(row)
current_preset_name = schedule_data.pop('current_preset_name')
config_data = json.loads(row['config_data']) if row['config_data'] else {}
last_run = json.loads(row['last_run']) if row['last_run'] else None
config_data.pop("connection_id", None)
config_data.pop("preset_id", None)
schedule_data.update(config_data)
if schedule_data.get('preset_id') and current_preset_name:
schedule_data['preset_name'] = current_preset_name
schedule_data['config_data'], schedule_data['last_run'] = config_data, last_run
schedules[row['id']] = schedule_data
for schedule_id, data in schedules.items():
if data.get("job_type") == "quick_playlist":
qpd = data.get("quick_playlist_data", {})
if (old_type := qpd.get("quick_playlist_type")) in LEGACY_TYPE_MAP:
new_type = LEGACY_TYPE_MAP[old_type]
data["quick_playlist_data"]["quick_playlist_type"] = new_type
logger.info(f"Migrated legacy schedule type '{old_type}' to '{new_type}' for job {schedule_id}.")
self._update_schedule_config_in_db(schedule_id, data)
return schedules
except Exception:
# Callers must distinguish "no schedules" from "could not read them":
# treating a failed read as an empty table tears down every live job.
logger.exception("Error loading schedules from database")
raise
def _update_schedule_config_in_db(self, schedule_id, schedule_data):
try:
with database.get_db_connection() as conn:
config_payload = {
"preset_name": schedule_data.get("preset_name"),
"blocks": schedule_data.get("blocks"),
"quick_playlist_data": schedule_data.get("quick_playlist_data"),
"enrichment_data": schedule_data.get("enrichment_data"),
"schedule_details": schedule_data.get("schedule_details"),
"create_as_collection": schedule_data.get("create_as_collection", False),
"mix_options": schedule_data.get("mix_options"),
"enabled": schedule_data.get("enabled", True),
"snoozed_until": schedule_data.get("snoozed_until"),
"trigger_sources": schedule_data.get("trigger_sources", ["clock", "watch", "library"]),
"timezone": schedule_data.get("timezone"),
}
conn.execute("UPDATE schedules SET config_data = ? WHERE id = ?", (json.dumps(config_payload), schedule_id))
conn.commit()
except Exception as e:
logger.error(f"Failed to update schedule config in DB for {schedule_id}: {e}", exc_info=True)
def enqueue_schedule_run(self, schedule_id: str, source: str = "clock") -> Optional[Dict]:
if not (schedule_data := self.schedules.get(schedule_id)):
return None
if not is_automatic_run_allowed(schedule_data, source):
logger.info(f"Skipping enqueue for schedule '{schedule_id}': source '{source}' not allowed.")
return {"status": "skipped", "reason": "not_allowed"}
job_id = f"run_{schedule_id}"
job_data = dict(schedule_data)
job_data["trigger_source"] = source
try:
self.scheduler.add_job(
func=scheduled_job_wrapper,
trigger='date',
run_date=datetime.now(),
kwargs=job_data,
id=job_id,
name=f"{source.capitalize()} Run: {schedule_data.get('playlist_name', 'Unnamed Schedule')}",
replace_existing=True,
misfire_grace_time=300
)
logger.info(f"Successfully queued background run ({source}) for schedule {schedule_id}")
return {
"status": "ok",
"log": [f"Execution started for '{schedule_data.get('playlist_name', 'Unnamed')}'."]
}
except Exception as e:
logger.error(f"Failed to queue run for {schedule_id}: {e}", exc_info=True)
return {
"status": "error",
"log": [f"Failed to queue background job: {str(e)}"]
}
def run_schedule_now(self, schedule_id: str, source: Optional[str] = None) -> Optional[Dict]:
if source is None:
try:
from routers.webhooks import get_current_webhook_category
webhook_cat = get_current_webhook_category()
source = webhook_cat if webhook_cat else "manual"
except Exception:
source = "manual"
return self.enqueue_schedule_run(schedule_id, source=source)
def pause_schedule(self, schedule_id: str) -> bool:
if schedule_id not in self.schedules:
return False
sched = self.schedules[schedule_id]
sched["enabled"] = False
self._update_schedule_config_in_db(schedule_id, sched)
logger.info(f"Paused schedule {schedule_id}")
return True
def resume_schedule(self, schedule_id: str) -> bool:
if schedule_id not in self.schedules:
return False
sched = self.schedules[schedule_id]
sched["enabled"] = True
sched["snoozed_until"] = None
self._update_schedule_config_in_db(schedule_id, sched)
logger.info(f"Resumed schedule {schedule_id}")
return True
def snooze_schedule(self, schedule_id: str, minutes: Optional[int] = 60, until: Optional[str] = None) -> bool:
if schedule_id not in self.schedules:
return False
sched = self.schedules[schedule_id]
from datetime import timezone, timedelta
if until:
snooze_until_iso = until
else:
mins = minutes if (minutes and minutes > 0) else 60
snooze_until_iso = (datetime.now(timezone.utc) + timedelta(minutes=mins)).isoformat()
sched["snoozed_until"] = snooze_until_iso
self._update_schedule_config_in_db(schedule_id, sched)
logger.info(f"Snoozed schedule {schedule_id} until {snooze_until_iso}")
return True
def _install_job(self, schedule_id: str, schedule_data: Dict) -> bool:
"""Add or replace the APScheduler job for one schedule.
Returns True when the schedule is now scheduled. A schedule with no trigger or
no verified connection is left inactive, and any job it previously had is
removed so a stale trigger cannot keep firing.
"""
trigger = self._get_trigger(schedule_data)
if not schedule_data.get("connection_id"):
logger.warning("Schedule %s has no verified connection; leaving it inactive.", schedule_id)
if trigger and schedule_data.get("connection_id"):
self.scheduler.add_job(
func=scheduled_job_wrapper,
trigger=trigger,
kwargs=schedule_data,
id=schedule_id,
name=schedule_data.get('playlist_name', 'Unnamed Schedule'),
replace_existing=True
)
return True
try:
self.scheduler.remove_job(schedule_id)
except JobLookupError:
pass
return False
def reload_schedules(self, strict: bool = False) -> Dict[str, Dict]:
"""Re-read every schedule from the database and resync APScheduler to match.
_load_schedules() is a pure read: on its own it changes neither self.schedules
nor the live jobs. A backup restore rewrites the schedules table underneath a
running scheduler, so without this resync deleted schedules keep their in-memory
jobs and restored ones never get a job until the process restarts.
By default an unreadable table keeps the schedules already loaded. With
``strict`` (backup restore) any read or trigger error is raised instead, before
live state changes, so the caller can roll the database back.
"""
try:
new_schedules = self._load_schedules()
except Exception:
if strict:
raise
logger.error(
"Schedule reload aborted; keeping the %d schedule(s) already loaded.",
len(self.schedules)
)
return self.schedules
if strict:
for schedule_id, schedule_data in new_schedules.items():
try:
self._get_trigger(schedule_data)
except Exception as e:
raise ValueError(f"Schedule {schedule_id} has an invalid trigger: {e}") from e
previous_ids = set(self.schedules)
self.schedules = new_schedules
for stale_id in previous_ids - set(new_schedules):
for job_id in (stale_id, f"run_{stale_id}"):
try:
self.scheduler.remove_job(job_id)
except JobLookupError:
pass
with self._schedule_locks_guard:
self._schedule_locks.pop(stale_id, None)
self._rerun_pending.pop(stale_id, None)
logger.info("Reload dropped schedule %s and its jobs.", stale_id)
active = 0
for schedule_id, schedule_data in new_schedules.items():
if self._install_job(schedule_id, schedule_data):
active += 1
logger.info(
"Reloaded %d schedule(s) from the database (%d active, %d removed).",
len(new_schedules), active, len(previous_ids - set(new_schedules))
)
return new_schedules
def start(self):
self.scheduler.add_job(
func=refresh_all_caches,
trigger='interval',
minutes=app_state.CACHE_REFRESH_MINUTES,
id='cache_refresh_job',
name='Refresh Library Data Cache',
replace_existing=True
)
self.scheduler.add_job(
func=vibe_index_catchup_job,
trigger='interval',
minutes=app_state.CACHE_REFRESH_MINUTES,
id='vibe_index_catchup_job',
name='Retry Unbuilt Vibe Indexes',
replace_existing=True
)
try:
self.schedules = self._load_schedules()
except Exception:
# Boot must not fail on an unreadable schedules table. get_all_schedules()
# retries the read on the next API call.
logger.error("Starting with no schedules loaded: the table could not be read.")
self.schedules = {}
from datetime import timezone
now_utc = datetime.now(timezone.utc)
for schedule_id, schedule_data in self.schedules.items():
# Startup reconciliation: check if snooze has expired
snoozed_until = schedule_data.get("snoozed_until")
if snoozed_until:
try:
snooze_dt = datetime.fromisoformat(snoozed_until.replace("Z", "+00:00"))
if snooze_dt.tzinfo is None:
snooze_dt = snooze_dt.replace(tzinfo=timezone.utc)
if now_utc >= snooze_dt:
schedule_data["snoozed_until"] = None
self._update_schedule_config_in_db(schedule_id, schedule_data)
logger.info(f"Reconciled expired snooze for schedule {schedule_id}.")
except Exception:
pass
self._install_job(schedule_id, schedule_data)
if not self.scheduler.running:
self.scheduler.start()
job_count = sum(bool(s.get("connection_id")) for s in self.schedules.values())
total_jobs = len(self.scheduler.get_jobs())
logger.info(f"Scheduler started with {job_count} user schedule(s) and {total_jobs - job_count} system job(s).")
def add_schedule(self, schedule_data: Dict) -> str:
schedule_id = str(uuid.uuid4())
schedule_data['id'] = schedule_id
if not schedule_data.get('connection_id'):
raise ValueError('A schedule requires a saved connection.')
self._bind_legacy_preset(schedule_data)
try:
with database.get_db_connection() as conn:
config_payload = {
"preset_name": schedule_data.get("preset_name"),
"blocks": schedule_data.get("blocks"),
"quick_playlist_data": schedule_data.get("quick_playlist_data"),
"enrichment_data": schedule_data.get("enrichment_data"),
"schedule_details": schedule_data.get("schedule_details"),
"create_as_collection": schedule_data.get("create_as_collection", False),
"mix_options": schedule_data.get("mix_options"),
"enabled": schedule_data.get("enabled", True),
"snoozed_until": schedule_data.get("snoozed_until"),
"trigger_sources": schedule_data.get("trigger_sources", ["clock", "watch", "library"]),
"timezone": schedule_data.get("timezone"),
}
conn.execute(
"INSERT INTO schedules (id, playlist_name, user_id, job_type, crontab, config_data, connection_id, preset_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(schedule_id, schedule_data.get("playlist_name"), schedule_data.get("user_id"), schedule_data.get("job_type"), schedule_data.get("crontab", ""), json.dumps(config_payload), schedule_data["connection_id"], schedule_data.get("preset_id"))
)
conn.commit()
except Exception as e:
logger.error(f"Failed to save schedule {schedule_id} to database: {e}", exc_info=True)
return None
self.schedules[schedule_id] = schedule_data
trigger = self._get_trigger(schedule_data)
self.scheduler.add_job(
func=scheduled_job_wrapper,
trigger=trigger,
kwargs=schedule_data,
id=schedule_id,
name=schedule_data.get('playlist_name', 'Unnamed Schedule')
)
return schedule_id
def update_schedule(self, schedule_id: str, schedule_data: Dict) -> bool:
if schedule_id not in self.schedules:
logger.warning(f"Attempted to update non-existent schedule {schedule_id}")
return False
if schedule_data.get("connection_id") != self.schedules[schedule_id].get("connection_id"):
return False
self._bind_legacy_preset(schedule_data)
try:
with database.get_db_connection() as conn:
config_payload = {
"preset_name": schedule_data.get("preset_name"),
"blocks": schedule_data.get("blocks"),
"quick_playlist_data": schedule_data.get("quick_playlist_data"),
"enrichment_data": schedule_data.get("enrichment_data"),
"schedule_details": schedule_data.get("schedule_details"),
"create_as_collection": schedule_data.get("create_as_collection", False),
"mix_options": schedule_data.get("mix_options"),
"enabled": schedule_data.get("enabled", True),
"snoozed_until": schedule_data.get("snoozed_until"),
"trigger_sources": schedule_data.get("trigger_sources", ["clock", "watch", "library"]),
"timezone": schedule_data.get("timezone"),
}
conn.execute(
"""
UPDATE schedules
SET playlist_name = ?, user_id = ?, job_type = ?, crontab = ?, config_data = ?, preset_id = ?
WHERE id = ?
""",
(
schedule_data.get("playlist_name"),
schedule_data.get("user_id"),
schedule_data.get("job_type"),
schedule_data.get("crontab", ""),
json.dumps(config_payload),
schedule_data.get("preset_id"),
schedule_id
)
)
conn.commit()
schedule_data['id'] = schedule_id
trigger = self._get_trigger(schedule_data)
self.scheduler.add_job(
func=scheduled_job_wrapper,
trigger=trigger,
kwargs=schedule_data,
id=schedule_id,
name=schedule_data.get('playlist_name', 'Unnamed Schedule'),
replace_existing=True
)
last_run_data = self.schedules[schedule_id].get('last_run')
self.schedules[schedule_id] = schedule_data
if last_run_data:
self.schedules[schedule_id]['last_run'] = last_run_data
return True
except Exception as e:
logger.error(f"Failed to update schedule {schedule_id}: {e}", exc_info=True)
return False
def remove_schedule(self, schedule_id: str):
if schedule_id in self.schedules:
try:
self.scheduler.remove_job(schedule_id)
except JobLookupError:
logger.warning(f"Job {schedule_id} not found, removing from storage anyway.")
try:
# A queued-but-not-yet-run request from run_schedule_now (manual "Run Now"
# or a webhook fan-out) uses this deterministic id; without canceling it
# too, it would fire against a schedule that no longer exists.
self.scheduler.remove_job(f"run_{schedule_id}")
except JobLookupError:
pass
del self.schedules[schedule_id]
try:
with database.get_db_connection() as conn:
conn.execute("DELETE FROM schedules WHERE id = ?", (schedule_id,))
conn.commit()
except Exception as e:
logger.error(f"Failed to delete schedule {schedule_id} from database: {e}", exc_info=True)
with self._schedule_locks_guard:
self._schedule_locks.pop(schedule_id, None)
self._rerun_pending.pop(schedule_id, None)
def get_run_state(self, schedule_id: str) -> Dict:
"""Live runtime facts for one schedule: its next clock fire and whether a run is in progress."""
next_run = None
try:
job = self.scheduler.get_job(schedule_id)
if job and job.next_run_time:
next_run = job.next_run_time.isoformat()
except Exception:
pass
with self._schedule_locks_guard:
lock = self._schedule_locks.get(schedule_id)
is_running = bool(lock and lock.locked())
try:
queued = self.scheduler.get_job(f"run_{schedule_id}") is not None
except Exception:
queued = False
return {"next_run_time": next_run, "is_running": is_running or queued}
def get_all_schedules(self) -> List[Dict]:
if not self.schedules and self.scheduler.running:
try:
self.schedules = self._load_schedules()
except Exception:
logger.error("Could not load schedules for this request.")
return list(self.schedules.values())
scheduler_manager = Scheduler()