Spaces:
Sleeping
Sleeping
| import logging | |
| from datetime import datetime, timezone | |
| from typing import Dict, Any, List, Optional | |
| from backend.preview.workspace import preview_workspace | |
| logger = logging.getLogger("build_pipeline") | |
| class BuildPipelineEngine: | |
| def __init__(self): | |
| self.active_builds: Dict[str, Dict[str, Any]] = {} | |
| async def trigger_build( | |
| self, | |
| project_id: str, | |
| build_target: str = "web", | |
| options: Optional[Dict[str, Any]] = None | |
| ) -> Dict[str, Any]: | |
| options = options or {} | |
| build_id = f"build_{project_id}_{int(datetime.now(timezone.utc).timestamp())}" | |
| build_record: Dict[str, Any] = { | |
| "build_id": build_id, | |
| "project_id": project_id, | |
| "target": build_target, | |
| "status": "in_progress", | |
| "stages": [ | |
| {"name": "workspace_validation", "status": "pending"}, | |
| {"name": "dependency_check", "status": "pending"}, | |
| {"name": "compilation", "status": "pending"}, | |
| {"name": "artifact_packaging", "status": "pending"} | |
| ], | |
| "start_time": datetime.now(timezone.utc).isoformat(), | |
| "end_time": None, | |
| "logs": [] | |
| } | |
| self.active_builds[build_id] = build_record | |
| try: | |
| # Stage 1: Workspace Validation | |
| build_record["stages"][0]["status"] = "in_progress" | |
| status = await preview_workspace.get_status(project_id) | |
| build_record["logs"].append(f"Workspace validated. Source files count: {status.get('source_files', 0)}") | |
| build_record["stages"][0]["status"] = "completed" | |
| # Stage 2: Dependency Verification | |
| build_record["stages"][1]["status"] = "in_progress" | |
| project_root = await preview_workspace.ensure_project_dir(project_id) | |
| has_package_json = (project_root / "package.json").exists() or (project_root / "src" / "package.json").exists() | |
| if has_package_json: | |
| build_record["logs"].append("package.json detected. Dependencies verified.") | |
| else: | |
| build_record["logs"].append("No package.json found. Proceeding with static web compilation.") | |
| build_record["stages"][1]["status"] = "completed" | |
| # Stage 3: Compilation & Bundling | |
| build_record["stages"][2]["status"] = "in_progress" | |
| rebuild_res = await preview_workspace.rebuild_project(project_id) | |
| build_record["logs"].append(rebuild_res.get("log", "Compilation succeeded.")) | |
| build_record["stages"][2]["status"] = "completed" | |
| # Stage 4: Artifact Packaging | |
| build_record["stages"][3]["status"] = "in_progress" | |
| source_zip = await preview_workspace.generate_source_zip(project_id) | |
| project_zip = await preview_workspace.generate_project_zip(project_id) | |
| build_record["artifacts"] = { | |
| "source_zip": str(source_zip), | |
| "project_zip": str(project_zip) | |
| } | |
| build_record["stages"][3]["status"] = "completed" | |
| build_record["status"] = "success" | |
| build_record["end_time"] = datetime.now(timezone.utc).isoformat() | |
| build_record["logs"].append("Build pipeline execution completed successfully.") | |
| except Exception as e: | |
| build_record["status"] = "failed" | |
| build_record["end_time"] = datetime.now(timezone.utc).isoformat() | |
| build_record["logs"].append(f"Build failed with error: {str(e)}") | |
| logger.error(f"Build pipeline failed for project {project_id}: {str(e)}") | |
| await preview_workspace._log_activity( | |
| project_id, | |
| f"Build pipeline '{build_id}' finished with status: {build_record['status']}" | |
| ) | |
| return build_record | |
| def get_build_status(self, build_id: str) -> Dict[str, Any]: | |
| if build_id not in self.active_builds: | |
| raise KeyError(f"Build ID '{build_id}' not found.") | |
| return self.active_builds[build_id] | |
| build_pipeline_engine = BuildPipelineEngine() | |