import logging import time from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from sqlalchemy import select from app.db import SessionLocal from app.models import Playlist, ScheduledJob from app.services import dedup_review_service, genre_review_service, pipeline_runner from app.settings import settings log = logging.getLogger("alembic") def _resolve_timezone(name: str) -> str: """Validate the configured timezone, falling back to UTC if it isn't a real IANA zone. Returned as a string because that's what APScheduler and CronTrigger accept directly.""" try: ZoneInfo(name) return name except (ZoneInfoNotFoundError, ValueError): log.warning("Unknown timezone %r, falling back to UTC.", name) return "UTC" TIMEZONE = _resolve_timezone(settings.timezone) # --------------------------------------------------------------------------- # Maintenance jobs, ported 1:1 from /etc/cron.d/sldl-maintenance. Each entry # is (job_key, cron kwargs for CronTrigger, callable). Dedup deliberately has # no --apply here (dry-run only on schedule) -- a deliberate behavior change # from the old blind `--apply` cron, matching the false-negative-biased # dedup preference; applying deletions requires a reviewed confirm through # the dedup UI (services/dedup_review_service.py). # # NOTE on job store choice: APScheduler's SQLAlchemyJobStore pickles job # functions to persist them, and closures returned from a factory (like # _lib()/_beet() below) are NOT picklable -- `add_job` would fail at # runtime. MemoryJobStore (the default) keeps live objects in-process # instead, which is fine here because register_all_jobs() rebuilds the # entire schedule from code + the DB on every startup anyway. Persisted # enable/disable state (so a paused job stays paused across a restart) # lives in the app's own `scheduled_jobs` table, not in APScheduler's store. # --------------------------------------------------------------------------- def _lib( job_key: str, script_name: str, args: list[str] | None = None, timeout: float | None = None, use_lock: bool = True, ): async def _run(triggered_by: str = "schedule"): return await pipeline_runner.run_lib_script( job_key, script_name, args, triggered_by=triggered_by, timeout=timeout, use_lock=use_lock ) return _run def _beet(job_key: str, args: list[str]): async def _run(triggered_by: str = "schedule"): return await pipeline_runner.run_beet(job_key, args, triggered_by=triggered_by) return _run async def _dedup_scan(triggered_by: str = "schedule"): """Dry-run only, ever, on the schedule -- see module docstring. Populates dedup_candidates for review; deletion only ever happens through a confirmed dedup_review_service.confirm_and_apply() call from the UI.""" return await dedup_review_service.scan(triggered_by=triggered_by) async def _genre_run(triggered_by: str = "schedule"): """Keeps its current auto-apply behavior (unlike dedup) -- genre writes are low-risk/reversible and GENRE_LOCK-protected -- but now goes through genre_review_service so every run's diff is persisted for after-the-fact review instead of only living in a log file.""" return await genre_review_service.run(apply=True, force=True, triggered_by=triggered_by) async def _log_rotation(triggered_by: str = "schedule"): """Replaces `find /var/log/sldl -name '*.log' -mtime +30 -delete`, and also prunes old job_runs rows so the database doesn't grow without bound.""" from app.db import prune_old_job_runs cutoff = time.time() - 30 * 86400 removed = 0 if settings.logs_dir.exists(): for path in settings.logs_dir.rglob("*.log"): if path.is_file() and path.stat().st_mtime < cutoff: path.unlink(missing_ok=True) removed += 1 pruned_rows = prune_old_job_runs() return {"logs_removed": removed, "job_run_rows_pruned": pruned_rows} MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = { # ==== Daily ==== "maintenance:sync_bandcamp": ( dict(minute=0, hour=8), _lib("maintenance:sync_bandcamp", "sync-bandcamp.sh"), ), "maintenance:normalize_casing": ( dict(minute=52, hour=8), _lib("maintenance:normalize_casing", "normalize-artist-casing.py", ["--apply"]), ), "maintenance:dedup": ( dict(minute=55, hour=8), _dedup_scan, # no --apply, ever, on schedule -- see _dedup_scan docstring ), "maintenance:gen_djmix_playlist": ( dict(minute=57, hour=8), _lib("maintenance:gen_djmix_playlist", "gen-djmix-playlist.sh"), ), "maintenance:gen_vgm_playlist": ( dict(minute=58, hour=8), _lib("maintenance:gen_vgm_playlist", "gen-vgm-playlist.sh"), ), "maintenance:navidrome_scan": ( dict(minute=0, hour=9), _lib("maintenance:navidrome_scan", "navidrome-scan.sh"), ), "maintenance:export_laptop_playlists": ( dict(minute=5, hour=9), _lib("maintenance:export_laptop_playlists", "export-laptop-playlists.py"), ), "maintenance:build_fingerprint_index": ( dict(minute=25, hour=9), _lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"], timeout=3600), ), "maintenance:enrich_buy_url": ( # Runs LAST in the 9am block: it hits external buy-link APIs per track # and can run long, so scheduling it after the fingerprint index (09:25) # and the status report (09:30) keeps it from starving them via the # shared lock. The 30-min cap is a backstop; enrich-buy-url.py also now # caches "tried, no match" so it stops re-querying the whole backlog. dict(minute=45, hour=9), _lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"], timeout=1800), ), "maintenance:pipeline_status_report": ( dict(minute=30, hour=9), # Read-only (reads logs, pings Navidrome, sends Telegram). Runs WITHOUT # the pipeline lock so the daily digest can never be starved by a long # write job, and a short timeout so a stuck Telegram call can't wedge it. _lib("maintenance:pipeline_status_report", "pipeline-status.sh", timeout=300, use_lock=False), ), "maintenance:log_rotation": ( dict(minute=0, hour=0), _log_rotation, ), # ==== Weekly (Sunday) ==== # strip_mb_tags runs first here at 01:00 -- not 08:30 like the rest of # this chain -- because its runtime isn't stable: 10m19s on 2026-07-12, # 39.6min on 2026-07-19 (MusicBrainz lookup latency scales with library # size and isn't under our control). At 08:30 that variance repeatedly # starved every job behind it: strip_watermark_art waited the full # SCHEDULED_LOCK_WAIT_SECONDS and gave up every week from 07-12 onward, # and on 07-19 the starvation cascaded all the way through genre:run, # normalize_casing, beets_update_sync, dedup:scan, and both gen_*_playlist # jobs. 01:00 sits in the dead zone before the first playlist sync (05:00) # and well after log_rotation (00:00), so even a run several times slower # than 07-19's is guaranteed to release the lock long before 08:30. "maintenance:strip_mb_tags": ( dict(minute=0, hour=1, day_of_week="sun"), _lib("maintenance:strip_mb_tags", "strip-mb-tags.sh"), ), "maintenance:strip_watermark_art": ( dict(minute=35, hour=8, day_of_week="sun"), _lib("maintenance:strip_watermark_art", "strip-watermark-art.py", ["--apply"]), ), "maintenance:scrub_watermark_text": ( dict(minute=40, hour=8, day_of_week="sun"), _lib("maintenance:scrub_watermark_text", "scrub-watermark-text.py", ["--apply"]), ), "maintenance:clean_sldl_index": ( dict(minute=45, hour=8, day_of_week="sun"), _lib("maintenance:clean_sldl_index", "clean-sldl-index.py", ["--apply"]), ), "maintenance:clear_bad_genres": ( dict(minute=48, hour=8, day_of_week="sun"), _lib("maintenance:clear_bad_genres", "clear-bad-genres.py", ["--apply"]), ), "maintenance:spotify_genre": ( dict(minute=50, hour=8, day_of_week="sun"), _genre_run, # goes through genre_review_service -- see its docstring ), "maintenance:beets_update_sync": ( dict(minute=53, hour=8, day_of_week="sun"), _beet("maintenance:beets_update_sync", ["update"]), ), # ==== Monthly ==== "maintenance:upgrade_mp3_to_flac": ( dict(minute=0, hour=10, day=1), _lib("maintenance:upgrade_mp3_to_flac", "upgrade-mp3-to-flac.sh"), ), } # One short line per job for the /jobs page -- what it actually does, not # just its key. Kept separate from MAINTENANCE_JOBS so the schedule/function # mapping above stays easy to scan. MAINTENANCE_JOB_DESCRIPTIONS: dict[str, str] = { "maintenance:sync_bandcamp": "Pull new Bandcamp purchases and import them.", "maintenance:normalize_casing": "Fix artist-name casing to match your canonical list.", "maintenance:dedup": "Scan for duplicate tracks (dry-run only; confirm deletes in Dedup).", "maintenance:gen_djmix_playlist": "Rebuild the DJ-mix M3U from configured albums.", "maintenance:gen_vgm_playlist": "Rebuild the video-game-soundtrack M3U.", "maintenance:navidrome_scan": "Trigger a Navidrome library rescan (share-health gated).", "maintenance:export_laptop_playlists": "Export dj-* Navidrome playlists for the laptop.", "maintenance:enrich_buy_url": "Look up buy links for tracks without one yet.", "maintenance:build_fingerprint_index": "Refresh the acoustic fingerprint index.", "maintenance:pipeline_status_report": "Write the daily health snapshot / Telegram digest.", "maintenance:log_rotation": "Delete job logs older than 30 days.", "maintenance:strip_mb_tags": "Strip stale MusicBrainz tags that fragment Navidrome albums.", "maintenance:strip_watermark_art": "Remove Soulseek-uploader watermark cover art.", "maintenance:scrub_watermark_text": "Remove Soulseek-uploader watermark text frames.", "maintenance:clean_sldl_index": "Prune stale entries from each playlist's download index.", "maintenance:clear_bad_genres": "Blank malformed/junk GENRE tags before Spotify refill.", "maintenance:spotify_genre": "Refill GENRE tags from Spotify per-artist data.", "maintenance:beets_update_sync": "Sync the beets DB to on-disk tag changes.", "maintenance:upgrade_mp3_to_flac": "Look for FLAC replacements for MP3 tracks (monthly).", } # Display-only friendly names -- never used as the actual scheduler/DB # identifier (job_key stays as-is everywhere else), just what the UI shows # instead of a raw "maintenance:snake_case_key". Every job_key not listed # here (e.g. playlist:, one per playlist) falls back to # humanize_job_key()'s generic cleanup below. JOB_LABELS: dict[str, str] = { "maintenance:sync_bandcamp": "Sync Bandcamp", "maintenance:normalize_casing": "Normalize artist casing", "maintenance:dedup": "Dedup scan", "maintenance:gen_djmix_playlist": "Rebuild DJ-mix playlist", "maintenance:gen_vgm_playlist": "Rebuild game-soundtrack playlist", "maintenance:navidrome_scan": "Navidrome rescan", "maintenance:export_laptop_playlists": "Export laptop playlists", "maintenance:enrich_buy_url": "Look up buy links", "maintenance:build_fingerprint_index": "Rebuild fingerprint index", "maintenance:pipeline_status_report": "Daily status report", "maintenance:log_rotation": "Rotate old logs", "maintenance:strip_mb_tags": "Strip MusicBrainz tags", "maintenance:strip_watermark_art": "Remove watermark art", "maintenance:scrub_watermark_text": "Remove watermark text", "maintenance:clean_sldl_index": "Clean download index", "maintenance:clear_bad_genres": "Clear bad genre tags", "maintenance:spotify_genre": "Refill genres from Spotify", "maintenance:beets_update_sync": "Sync library tags", "maintenance:upgrade_mp3_to_flac": "Upgrade MP3s to FLAC", "dedup:scan": "Dedup scan (file naming)", "dedup:scan_fuzzy": "Dedup scan (audio)", "dedup:apply": "Dedup: delete confirmed", "genre:run": "Genre refresh", "manual:import": "Manual import", "manual:fix_genre": "Fix genre", "manual:navidrome_scan_after_edit": "Navidrome rescan (after edit)", "manual:retag_from_url": "Retag from URL", } def humanize_job_key(job_key: str) -> str: """Friendly label for a job_key, for display only. Falls back to a cleaned-up version of the key for anything not in JOB_LABELS -- covers playlist:, one per playlist, and anything added later without needing a JOB_LABELS entry.""" if job_key in JOB_LABELS: return JOB_LABELS[job_key] if job_key.startswith("playlist:"): return f"Playlist sync: {job_key.split(':', 1)[1]}" _prefix, _, rest = job_key.partition(":") rest = rest.replace("_", " ").strip() return (rest[:1].upper() + rest[1:]) if rest else job_key # Module-level singleton so other services (playlist_service, future # routers) can reach the live scheduler without main.py threading it through # every function call. None until main.py's lifespan starts it. _instance: AsyncIOScheduler | None = None def get_scheduler() -> AsyncIOScheduler | None: return _instance def create_scheduler() -> AsyncIOScheduler: global _instance # Default job store (MemoryJobStore) -- see module docstring for why # SQLAlchemyJobStore doesn't work here. _instance = AsyncIOScheduler(timezone=TIMEZONE) return _instance def _ensure_scheduled_job_rows(db) -> None: """Make sure every MAINTENANCE_JOBS entry has a scheduled_jobs row, seeding cron_expr from the hardcoded default and enabled=True the first time. Subsequent runs never overwrite an existing row -- this table is the persisted source of truth for enable/disable once it exists.""" existing = {row.job_key for row in db.execute(select(ScheduledJob)).scalars()} now = time.time() for job_key, (cron_kwargs, _runner) in MAINTENANCE_JOBS.items(): if job_key in existing: continue cron_expr = "{minute} {hour} {day} {month} {day_of_week}".format( minute=cron_kwargs.get("minute", "*"), hour=cron_kwargs.get("hour", "*"), day=cron_kwargs.get("day", "*"), month=cron_kwargs.get("month", "*"), day_of_week=cron_kwargs.get("day_of_week", "*"), ) db.add( ScheduledJob( job_key=job_key, job_type="maintenance", playlist_id=None, cron_expr=cron_expr, enabled=True, ) ) db.commit() _ = now # unused, kept for symmetry with playlists' created_at/updated_at style def _playlist_job_key(name: str) -> str: return f"playlist:{name}" async def _run_playlist_job(name: str, no_m3u: bool, triggered_by: str = "schedule"): return await pipeline_runner.run_playlist(name, no_m3u=no_m3u, triggered_by=triggered_by) def sync_playlist_jobs(scheduler: AsyncIOScheduler) -> None: """Converge the scheduler's registered playlist: jobs with the playlists table. Call after any playlist create/update/delete -- this is what makes add/remove-in-the-UI take effect without a redeploy.""" db = SessionLocal() try: playlists = list(db.execute(select(Playlist)).scalars()) finally: db.close() desired_ids = { _playlist_job_key(p.name) for p in playlists if p.cron_expr and p.active } existing_ids = {j.id for j in scheduler.get_jobs() if j.id.startswith("playlist:")} for job_id in existing_ids - desired_ids: scheduler.remove_job(job_id) for playlist in playlists: job_id = _playlist_job_key(playlist.name) if job_id not in desired_ids: continue trigger = CronTrigger.from_crontab(playlist.cron_expr, timezone=TIMEZONE) scheduler.add_job( _run_playlist_job, trigger=trigger, id=job_id, args=[playlist.name, playlist.no_m3u], replace_existing=True, misfire_grace_time=3600, ) def register_all_jobs(scheduler: AsyncIOScheduler) -> None: """Rebuild the entire in-memory schedule from code (MAINTENANCE_JOBS) + the DB (scheduled_jobs.enabled, playlists). Safe to call at every startup -- MemoryJobStore has nothing to reconcile against.""" db = SessionLocal() try: _ensure_scheduled_job_rows(db) enabled_by_key = { row.job_key: row.enabled for row in db.execute(select(ScheduledJob)).scalars() } finally: db.close() for job_key, (cron_kwargs, runner) in MAINTENANCE_JOBS.items(): if not enabled_by_key.get(job_key, True): continue scheduler.add_job( runner, trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE), id=job_key, replace_existing=True, misfire_grace_time=3600, ) sync_playlist_jobs(scheduler) async def trigger_now(scheduler: AsyncIOScheduler, job_key: str) -> None: """'Run now' support: bypass the schedule and run this job's function immediately, still through pipeline_runner's lock/logging/job_runs recording (triggered_by='manual'). Works even for a currently-disabled job (disabled only means "don't run on schedule").""" job = scheduler.get_job(job_key) if job is None: raise ValueError(f"no such registered job: {job_key}") func = job.func args = list(job.args) kwargs = dict(job.kwargs) kwargs["triggered_by"] = "manual" await func(*args, **kwargs) def set_maintenance_enabled(scheduler: AsyncIOScheduler, job_key: str, enabled: bool) -> None: """Pause/resume a maintenance job AND persist the choice to scheduled_jobs.enabled so it survives a restart -- this is the actual 'pause an individual job without a redeploy' mechanism.""" if job_key not in MAINTENANCE_JOBS: raise ValueError(f"not a maintenance job: {job_key}") db = SessionLocal() try: row = db.execute( select(ScheduledJob).where(ScheduledJob.job_key == job_key) ).scalar_one_or_none() if row is not None: row.enabled = enabled db.commit() finally: db.close() if enabled: cron_kwargs, runner = MAINTENANCE_JOBS[job_key] scheduler.add_job( runner, trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE), id=job_key, replace_existing=True, misfire_grace_time=3600, ) else: if scheduler.get_job(job_key) is not None: scheduler.remove_job(job_key)