Spaces:
Runtime error
Runtime error
File size: 3,804 Bytes
ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 ac772c6 97bdb09 | 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 | 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}" |