File size: 3,088 Bytes
345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | 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())
|