marriott-box-image-video-tagging/scheduler.py

"""
Long-running scheduler entrypoint for the Marriott Box tagger Docker container.

Replaces the Ubuntu systemd timer when running under Docker. Fires main.main()
on the schedule in $SCHEDULE_CRON (default: daily 02:00). If $RUN_AT_STARTUP=1,
also fires an immediate one-off pass on boot.

DB schema is bootstrapped once at startup if Postgres is reachable; main() also
re-checks per run, so a temporary DB outage during startup self-heals.
"""

import os
import signal
import sys
import time

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from dotenv import load_dotenv

import db
import main as tagger


def _bootstrap_db():
    """Best-effort: open a connection, apply schema. Failures are logged, not fatal."""
    try:
        conn = db.get_conn()
        db.ensure_schema(conn)
        conn.close()
        print("[scheduler] Postgres schema ensured.")
    except Exception as e:
        print(f"[scheduler] WARN: could not bootstrap Postgres ({type(e).__name__}: {e}).")


def _run_job():
    print(f"\n[scheduler] Firing tagging pass at {time.strftime('%Y-%m-%d %H:%M:%S %Z')}")
    try:
        tagger.main()
    except SystemExit as e:
        # main() calls sys.exit() on missing credentials; let the scheduler keep running.
        print(f"[scheduler] tagging pass exited with code {e.code} — scheduler stays up.")
    except Exception as e:
        print(f"[scheduler] tagging pass raised {type(e).__name__}: {e} — scheduler stays up.")


def main():
    load_dotenv()
    _bootstrap_db()

    cron_expr = os.getenv("SCHEDULE_CRON", "0 2 * * *").strip()
    tz = os.getenv("TZ")  # apscheduler reads tzinfo; if unset uses system local
    print(f"[scheduler] Cron schedule: '{cron_expr}' (TZ={tz or 'system'})")

    scheduler = BlockingScheduler(timezone=tz) if tz else BlockingScheduler()
    scheduler.add_job(
        _run_job,
        CronTrigger.from_crontab(cron_expr, timezone=tz) if tz else CronTrigger.from_crontab(cron_expr),
        id="tagging_pass",
        max_instances=1,
        coalesce=True,
        misfire_grace_time=3600,
    )

    # SIGTERM/SIGINT → graceful shutdown
    def _shutdown(signum, frame):
        print(f"[scheduler] Received signal {signum} — shutting down.")
        scheduler.shutdown(wait=False)
        sys.exit(0)
    signal.signal(signal.SIGTERM, _shutdown)
    signal.signal(signal.SIGINT, _shutdown)

    if os.getenv("RUN_AT_STARTUP", "").strip() in ("1", "true", "yes"):
        print("[scheduler] RUN_AT_STARTUP set — firing one pass now.")
        _run_job()

    print("[scheduler] Entering scheduler loop. Ctrl-C / SIGTERM to exit.")
    scheduler.start()


if __name__ == "__main__":
    main()