Spaces:
Runtime error
Runtime error
| import os | |
| import subprocess | |
| import logging | |
| from typing import Dict, Any | |
| from models.router import ModelRouter | |
| from .github_sync import GitHubSyncer | |
| log = logging.getLogger("core") | |
| class ExecutionEngine: | |
| def __init__(self, router: ModelRouter, syncer: GitHubSyncer): | |
| self.router = router | |
| self.syncer = syncer | |
| async def execute_step(self, plan: Any, step: Any) -> str: | |
| file_path = step.target_file | |
| log.info(f"Executing step {step.step_id} for file: {file_path}") | |
| existing = "" | |
| if os.path.exists(file_path): | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| existing = f.read() | |
| # TO'G'RILANDI: f-string ichidagi \n xatoligi butunlay tuzatildi | |
| existing_block = f"Existing:\n{existing}\n" if existing else "" | |
| system_prompt = ( | |
| "You are an expert production-grade Python developer. Output ONLY raw, executable source code. " | |
| "Do NOT wrap your response in markdown code blocks like ```python. Do not include any explanations." | |
| ) | |
| prompt = f"Task: {plan.task}\nStep: {step.description}\nFile: {file_path}\n{existing_block}Write complete content." | |
| # Modelni chaqirish | |
| code = await self.router.call( | |
| prompt=prompt, | |
| task_type=step.task_type, | |
| system=system_prompt, | |
| estimated_tokens=2500 | |
| ) | |
| # Kodni tozalash (agar baribir markdown ishlatgan bo'lsa) | |
| if code.startswith("```"): | |
| lines = code.splitlines() | |
| if lines[0].startswith("```"): | |
| lines = lines[1:] | |
| if lines and lines[-1].startswith("```"): | |
| lines = lines[:-1] | |
| code = "\n".join(lines) | |
| # Papka mavjud bo'lmasa yaratish | |
| dir_name = os.path.dirname(file_path) | |
| if dir_name and not os.path.exists(dir_name): | |
| os.makedirs(dir_name, exist_ok=True) | |
| # Faylga yozish | |
| with open(file_path, "w", encoding="utf-8") as f: | |
| f.write(code) | |
| # Sintaksisni tekshirish | |
| try: | |
| subprocess.run(["python", "-m", "py_compile", file_path], check=True, capture_output=True) | |
| log.info(f"File {file_path} compiled successfully.") | |
| except subprocess.CalledProcessError as e: | |
| log.warning(f"Syntax error detected in {file_path}, forcing debug loop...") | |
| return await self._debug_loop(file_path, code, e.stderr.decode(), system_prompt) | |
| # GitHub-ga yuklash | |
| try: | |
| self.syncer.sync_file(file_path, code) | |
| return f"Successfully created/updated {file_path} and synced to GitHub." | |
| except Exception as e: | |
| return f"File saved locally to {file_path}, but GitHub sync failed: {e}" | |
| async def _debug_loop(self, file_path: str, broken_code: str, error_msg: str, system_prompt: str) -> str: | |
| prompt = ( | |
| f"The following code for {file_path} has a syntax error.\n" | |
| f"Error:\n{error_msg}\n" | |
| f"Code:\n{broken_code}\n" | |
| f"Fix the code and return ONLY the completely corrected script." | |
| ) | |
| fixed_code = await self.router.call(prompt=prompt, task_type="fix", system=system_prompt, estimated_tokens=2500) | |
| with open(file_path, "w", encoding="utf-8") as f: | |
| f.write(fixed_code) | |
| try: | |
| subprocess.run(["python", "-m", "py_compile", file_path], check=True, capture_output=True) | |
| self.syncer.sync_file(file_path, fixed_code) | |
| return f"Fixed syntax error automatically and synced {file_path} to GitHub." | |
| except Exception as e: | |
| return f"Attempted auto-debug on {file_path}, but it still fails syntax check. Error: {e}" |