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())