import os import re import json import logging from typing import List, Dict, Any from models.router import ModelRouter log = logging.getLogger("core") class TaskPlanner: def __init__(self, router: ModelRouter): self.router = router async def generate_plan(self, task_description: str) -> List[Dict[str, Any]]: system_prompt = ( "You are an expert software architect. Break down the user request into a strict sequential " "execution plan. Respond ONLY with a valid JSON array of step objects. Do not wrap in markdown tags.\n" "Each step object must have:\n" "- step_id: integer\n" "- task_type: text (one of: 'code', 'debug', 'review', 'explain', 'fix', 'test', 'generic')\n" "- target_file: text (the path to the file being touched)\n" "- instruction: text (detailed objective for this step)\n" "Example format:\n" '[{"step_id": 1, "task_type": "code", "target_file": "math.py", "instruction": "Write sum function"}]' ) user_prompt = f"Task Description: {task_description}" raw_response = await self.router.call( prompt=user_prompt, task_type="plan", system=system_prompt, estimated_tokens=1000 ) return self._parse_json_plan(raw_response) def _parse_json_plan(self, text: str) -> List[Dict[str, Any]]: # TO'G'RILANDI: regex qatorlari yopildi va xatolik bartaraf etildi clean = re.sub(r"```json\s*", "", text, flags=re.IGNORECASE) clean = re.sub(r"```\s*$", "", clean, flags=re.IGNORECASE) clean = clean.strip() try: plan = json.loads(clean) if isinstance(plan, list): return plan elif isinstance(plan, dict) and "steps" in plan: return plan["steps"] raise ValueError("JSON is not a list structure.") except Exception as e: log.error(f"Failed to parse plan JSON. Raw output: {text}. Error: {e}") return [{ "step_id": 1, "task_type": "generic", "target_file": "unknown_file.py", "instruction": f"Execute raw request due to planner parsing error. Task: {text}" }]