alembic/app/services/scheduler_service.py

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:enrich_buy_url": (
        dict(minute=10, hour=9),
        # 30-min cap: this hits external buy-link APIs per track and has hung
        # holding the pipeline lock (2026-07-09/10), starving every job after
        # it. A timeout releases the lock so the rest of the 9am chain runs.
        _lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"], timeout=1800),
    ),
    "maintenance:build_fingerprint_index": (
        dict(minute=25, hour=9),
        _lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"], timeout=3600),
    ),
    "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) ====
    "maintenance:strip_mb_tags": (
        dict(minute=30, hour=8, 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:<name>, 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:<name>, 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:<name> 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)