Spaces:
Sleeping
Sleeping
File size: 3,027 Bytes
71b4454 | 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 | import asyncio
import os
import uuid
from typing import Dict, Any, Optional
class BuildQueueManager:
def __init__(self):
self._builds: Dict[str, Dict[str, Any]] = {}
def get_build_status(self, build_id: str) -> Optional[Dict[str, Any]]:
return self._builds.get(build_id)
async def _execute_build_script(self, build_id: str, script_name: str, project_dir: str):
script_path = os.path.join("scripts", script_name)
if not os.path.exists(script_path):
self._builds[build_id]["status"] = "failed"
self._builds[build_id]["error"] = f"Script {script_name} not found"
return
try:
proc = await asyncio.create_subprocess_exec(
script_path, build_id, project_dir,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60.0)
except asyncio.TimeoutError:
proc.kill()
self._builds[build_id]["status"] = "failed"
self._builds[build_id]["error"] = "Build timed out after 60 seconds"
return
if proc.returncode == 0:
self._builds[build_id]["status"] = "success"
self._builds[build_id]["stdout"] = stdout.decode(errors="ignore")
else:
self._builds[build_id]["status"] = "failed"
self._builds[build_id]["stdout"] = stdout.decode(errors="ignore")
self._builds[build_id]["stderr"] = stderr.decode(errors="ignore")
self._builds[build_id]["error"] = (
stderr.decode(errors="ignore")
or stdout.decode(errors="ignore")
or "Build script failed"
)
except Exception as e:
self._builds[build_id]["status"] = "failed"
self._builds[build_id]["error"] = str(e)
self._builds[build_id]["status"] = "failed"
self._builds[build_id]["error"] = str(e)
def trigger_build(self, target: str, project_dir: str = ".") -> str:
build_id = f"bld_{uuid.uuid4().hex[:8]}"
self._builds[build_id] = {
"build_id": build_id,
"target": target,
"status": "queued",
"project_dir": project_dir
}
script_map = {
"apk": "build_apk.sh",
"static": "build_static.sh",
"docker": "build_docker.sh"
}
script_name = script_map.get(target)
if not script_name:
self._builds[build_id]["status"] = "failed"
self._builds[build_id]["error"] = f"Unknown target: {target}"
return build_id
self._builds[build_id]["status"] = "building"
asyncio.create_task(self._execute_build_script(build_id, script_name, project_dir))
return build_id
build_queue = BuildQueueManager()
|