publisher/scheduler_engine.py

import asyncio
import logging
from datetime import datetime, timezone

logger = logging.getLogger("scheduler-engine")

# =========================
# INTERNAL SCHEDULER STATE
# =========================

_scheduler_running = False
_tasks = []  # in-memory fallback (can later swap to Redis/DB)


# =========================
# INIT ENTRYPOINT (FIX)
# =========================

def init_scheduler():
    """
    Called by main.py on startup.
    Safe, idempotent scheduler bootstrap.
    """
    global _scheduler_running

    if _scheduler_running:
        logger.info("[Scheduler] Already running")
        return

    _scheduler_running = True
    logger.info("🚀 Scheduler Engine V10 initialized")


# =========================
# CORE SCHEDULER API
# =========================

def schedule_post(payload: dict):
    """
    Adds a post to the queue.
    Expected payload:
    {
        "video_path": str,
        "platform": str,
        "publish_at": datetime ISO string
    }
    """

    job = {
        "id": f"job_{len(_tasks)+1}",
        "type": payload.get("type", "publish"),
        "payload": payload,
        "status": "queued",
        "created_at": datetime.utcnow().isoformat()
    }

    _tasks.append(job)

    logger.info(f"[Scheduler] Job queued: {job['id']}")

    return job


def _publish_time(payload: dict) -> datetime | None:
    value = payload.get("publish_at")
    if not value:
        return None

    publish_time = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    if publish_time.tzinfo is None:
        publish_time = publish_time.replace(tzinfo=timezone.utc)
    return publish_time


async def get_next_job():
    """Return the next due job and claim it for a publisher worker."""
    now = datetime.now(timezone.utc)

    for job in _tasks:
        if job["status"] not in {"queued", "ready"}:
            continue

        publish_time = _publish_time(job["payload"])
        if publish_time and now < publish_time:
            continue

        job["status"] = "processing"
        return job

    return None


# =========================
# WORKER LOOP
# =========================

async def _worker_loop():
    """
    Background scheduler processor.
    """

    logger.info("[Scheduler] Worker loop started")

    while True:
        try:
            now = datetime.now(timezone.utc)

            for job in _tasks:
                if job["status"] != "queued":
                    continue

                publish_time = _publish_time(job["payload"])

                if publish_time is None or now >= publish_time:
                    logger.info(f"[Scheduler] Executing {job['id']}")

                    # mark as done (actual publish handled elsewhere)
                    job["status"] = "ready"

        except Exception as e:
            logger.error(f"[Scheduler Error] {str(e)}")

        await asyncio.sleep(5)


# =========================
# OPTIONAL START LOOP
# =========================

def start_scheduler_loop():
    """
    Optional explicit background runner.
    """
    asyncio.create_task(_worker_loop())