Update services/queue_manager.py

services/queue_manager.py

@@ -2,13 +2,11 @@ import asyncio
 import uuid
 from typing import Dict, Any, Optional
 from enum import Enum
+from pipeline.pipeline import run_pipeline
 
-# Import pipeline that handles all stages
-from pipeline.pipeline import run_pipeline  # This runs story → images → video → etc.
-
-# -------------------------------
-# ENUMS AND GLOBALS
-# -------------------------------
+# ---------------------------
+# Task status enum
+# ---------------------------
 class TaskStatus(str, Enum):
     PENDING = "pending"
     RUNNING = "running"
@@ -17,104 +15,58 @@ class TaskStatus(str, Enum):
     COMPLETED = "completed"
     FAILED = "failed"
 
+# ---------------------------
+# Globals
+# ---------------------------
 tasks: Dict[str, Dict[str, Any]] = {}
 task_queue = asyncio.Queue()
 pending_confirmations: Dict[str, asyncio.Event] = {}
 
-# -------------------------------
-# ADD TASK
-# -------------------------------
+# ---------------------------
+# Add task
+# ---------------------------
 async def add_task(idea: str) -> str:
     task_id = str(uuid.uuid4())
+    confirmation_event = asyncio.Event()
+
     tasks[task_id] = {
         "id": task_id,
         "idea": idea,
         "status": TaskStatus.PENDING,
-        "result": None,
+        "result": {},
         "confirmation_required": False
     }
+    pending_confirmations[task_id] = confirmation_event
     await task_queue.put(task_id)
-    print(f"🧩 Task added to queue: {task_id}")
-    return task_id
-
-# -------------------------------
-# WAIT FOR SCRIPT
-# -------------------------------
-async def wait_for_script(task_id: str, script_results: dict):
-    task = tasks.get(task_id)
-    if not task:
-        return
-
-    task["status"] = TaskStatus.RUNNING
-
-    # === STEP 1: generate script ===
-    from core.script_gen import generate_script
-    script_result = await generate_script(task["idea"])
-    task["result"] = {"script": script_result}
-    task["status"] = TaskStatus.WAITING_CONFIRMATION
-    task["confirmation_required"] = True
 
-    # Store script for server endpoint
-    script_results[task_id] = script_result
-
-    # Create event for confirmation
-    pending_confirmations[task_id] = asyncio.Event()
-    print(f"✋ Task {task_id} waiting for confirmation. Script ready.")
-
-    # Wait for user confirmation
-    await pending_confirmations[task_id].wait()
-
-    # Once confirmed, run full pipeline
-    task["status"] = TaskStatus.CONFIRMED
-    print(f"🎬 Task {task_id} confirmed. Running full pipeline...")
-
-    # Run pipeline (story, images, video, etc.)
-    final_result = await run_pipeline(task)
-    task["result"].update(final_result)
-    task["status"] = TaskStatus.COMPLETED
-    print(f"✅ Task {task_id} completed. All stages done.")
+    # Start the pipeline immediately in background
+    asyncio.create_task(run_pipeline(tasks[task_id], confirmation_event))
+    print(f"🧩 Task added and pipeline started: {task_id}")
+    return task_id
 
-# -------------------------------
-# CONFIRM TASK
-# -------------------------------
+# ---------------------------
+# Confirm task
+# ---------------------------
 async def confirm_task(task_id: str):
     task = tasks.get(task_id)
     if not task:
-        return {"error": "Invalid task ID."}
-
+        return {"error": "Invalid task ID"}
     if task["status"] != TaskStatus.WAITING_CONFIRMATION:
-        return {"error": "Task is not waiting for confirmation."}
+        return {"error": "Task is not waiting for confirmation"}
 
-    # Set event so the worker can continue
     event = pending_confirmations.get(task_id)
     if event:
         event.set()
+    return {"message": f"Task {task_id} confirmed"}
 
-    return {"message": f"Task {task_id} confirmed."}
-
-# -------------------------------
-# GET STATUS
-# -------------------------------
+# ---------------------------
+# Get task status
+# ---------------------------
 def get_task_status(task_id: str) -> Optional[Dict[str, Any]]:
     return tasks.get(task_id)
 
-# -------------------------------
-# WORKER
-# -------------------------------
-async def worker():
-    while True:
-        task_id = await task_queue.get()
-        task = tasks.get(task_id)
-        if not task:
-            task_queue.task_done()
-            continue
-        # Script is generated in wait_for_script, rest handled by pipeline
-        task_queue.task_done()
-
-# -------------------------------
-# START WORKER
-# -------------------------------
+# ---------------------------
+# Worker (optional)
+# ---------------------------
 def start_worker():
-    loop = asyncio.get_event_loop()
-    loop.create_task(worker())
-    print("⚙️ Worker loop started.")
+    print("⚙️ Worker loop not required, pipeline runs per task")