File size: 4,509 Bytes
345855e | 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 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | import asyncio
import uuid
from datetime import datetime
# Optional queue integration (safe fallback if not present)
try:
from utils.job_queue import create_job
except Exception:
create_job = None
# -------------------------------------------------
# CONTEXT NORMALIZER
# -------------------------------------------------
def normalize_context(context):
if isinstance(context, dict):
return {
"items": context.get("items", []),
"webhook": context.get("webhook"),
"mode": context.get("mode", "sequential")
}
return {
"items": getattr(context, "items", []),
"webhook": getattr(context, "webhook", None),
"mode": getattr(context, "mode", "sequential")
}
# -------------------------------------------------
# SINGLE TASK EXECUTOR WRAPPER
# -------------------------------------------------
async def execute_single(task_name, payload):
"""
Uses registry executor if available, otherwise returns structured fallback.
"""
try:
from core.execution.executor import execute_task
return await execute_task(
task_name,
payload
)
except Exception as e:
return {
"task": task_name,
"status": "failed",
"error": str(e)
}
# -------------------------------------------------
# MAIN BATCH RUNNER
# -------------------------------------------------
async def run(context):
ctx = normalize_context(context)
items = ctx["items"]
if not items:
return {
"status": "error",
"message": "Batch requires 'items' list"
}
batch_id = str(uuid.uuid4())
started_at = datetime.utcnow().isoformat()
results = []
failed = 0
# -------------------------------------------------
# MODE: SEQUENTIAL EXECUTION
# -------------------------------------------------
if ctx["mode"] == "sequential":
for i, item in enumerate(items):
task_name = item.get("task")
payload = item.get("payload", {})
if not task_name:
results.append({
"index": i,
"status": "skipped",
"reason": "missing task"
})
continue
result = await execute_single(task_name, payload)
if isinstance(result, dict) and result.get("status") == "failed":
failed += 1
results.append({
"index": i,
"task": task_name,
"result": result
})
# -------------------------------------------------
# MODE: PARALLEL EXECUTION
# -------------------------------------------------
elif ctx["mode"] == "parallel":
async def run_item(i, item):
task_name = item.get("task")
payload = item.get("payload", {})
if not task_name:
return {
"index": i,
"status": "skipped"
}
result = await execute_single(task_name, payload)
return {
"index": i,
"task": task_name,
"result": result
}
results = await asyncio.gather(
*[run_item(i, item) for i, item in enumerate(items)]
)
else:
return {
"status": "error",
"message": f"Unsupported mode: {ctx['mode']}"
}
# -------------------------------------------------
# JOB QUEUE INTEGRATION (OPTIONAL)
# -------------------------------------------------
job_id = None
if create_job:
try:
job_id = create_job(
video_path=None,
webhook=ctx["webhook"],
metadata={
"batch_id": batch_id,
"total": len(items),
"failed": failed
}
)
except Exception:
job_id = None
# -------------------------------------------------
# FINAL RESPONSE
# -------------------------------------------------
return {
"status": "success",
"batch_id": batch_id,
"job_id": job_id,
"started_at": started_at,
"completed_at": datetime.utcnow().isoformat(),
"mode": ctx["mode"],
"total": len(items),
"failed": failed,
"results": results
} |