import json import logging import httpx import os import shutil import subprocess from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from app.core.config import settings from app.schemas.code_output import ProjectCodeOutput, GeneratedFile from app.llm.prompts import DEV_AGENT_PROMPT logger = logging.getLogger(__name__) # Clients HTTPX configurés pour ignorer les blocages de certificats/révocation du lab sync_client = httpx.Client(verify=False) async_client = httpx.AsyncClient(verify=False) async def run_dev_agent(spec: dict, qa_feedback: list = None) -> dict: """ Agent Dev : prend un état/spec validé, génère l'arborescence et le code, et valide techniquement la sortie avant de la transmettre à la QA. """ logger.info(f"[Dev Agent] Début de la génération pour : {spec.get('title', 'Sans titre')}") llm = ChatOpenAI( base_url=settings.llm_base_url, api_key=settings.llm_api_key, model=settings.llm_model_dev, temperature=0.2, max_retries=2, http_client=sync_client, http_async_client=async_client, model_kwargs={"response_format": {"type": "json_object"}} ) structured_llm = llm.with_structured_output(ProjectCodeOutput, strict=True) messages = [ SystemMessage(content=DEV_AGENT_PROMPT), ] user_content = f"CAHIER DES CHARGES (ProjectSpec) :\n{json.dumps(spec, indent=2, ensure_ascii=False)}\n\n" if qa_feedback: user_content += f"⚠️ RETOURS DE VALIDATION QA (Corrections à appliquer impérativement) :\n{json.dumps(qa_feedback, indent=2, ensure_ascii=False)}\n\n" user_content += "Génère maintenant le JSON complet contenant l'arborescence ('tree') et tous les fichiers ('files') décrits." messages.append(HumanMessage(content=user_content)) try: code_data = await structured_llm.ainvoke(messages) logger.info(f"[Dev Agent] ✅ Code généré ({len(code_data.files)} fichiers)") return await _deploy_to_gitea(code_data, spec.get("title", "project")) except Exception as e: logger.error(f"[Dev Agent] ❌ Échec de la génération/validation : {type(e).__name__}: {str(e)}") return _generate_fallback_code_output(spec) async def _ensure_gitea_repo(base_name: str) -> str: """ Vérifie si le repo existe. Si oui, incrémente un suffixe (_1, _2...) jusqu'à trouver un nom libre, puis le crée et retourne ce nom unique. """ async with httpx.AsyncClient(verify=False) as client: headers = {"Authorization": f"token {settings.gitea_token}"} repo_name = base_name counter = 1 # 1. Boucle de recherche d'un nom disponible via l'endpoint direct du repo while True: check_url = f"{settings.gitea_api_url}/repos/{settings.gitea_admin_user}/{repo_name}" response = await client.get(check_url, headers=headers) if response.status_code == 404: # Parfait ! Le 404 signifie que ce nom n'existe pas encore, il est libre. break elif response.status_code == 200: # Le nom est déjà pris, on tente avec le compteur suivant repo_name = f"{base_name}_{counter}" counter += 1 else: # Sécurité si l'API renvoie autre chose (ex: 401 Unauthorized) raise Exception(f"Erreur Gitea lors de la vérification ({response.status_code}) : {response.text}") # 2. Si on arrive ici, repo_name est garanti unique et disponible. On le crée. logger.info(f"[Gitea] Création du nouveau dépôt unique '{repo_name}'...") create_url = f"{settings.gitea_api_url}/user/repos" payload = { "name": repo_name, "auto_init": True, "description": "Generated by AI Dev Agent (Unique Instance)" } create_res = await client.post(create_url, headers=headers, json=payload) if create_res.status_code in [201, 200]: logger.info(f"[Gitea] Dépôt '{repo_name}' créé avec succès.") return repo_name else: logger.error(f"[Gitea] Erreur lors de la création : {create_res.text}") raise Exception(f"Impossible de créer le dépôt '{repo_name}' sur Gitea.") async def _deploy_to_gitea(code_data: ProjectCodeOutput, project_title: str) -> dict: """Gère le cycle de vie Git : Clone/Init -> Write -> Commit -> Push.""" base_name = "".join(c for c in project_title.lower().replace(" ", "_") if c.isalnum() or c == "_") # --- MODIFICATION ICI : On récupère le nom unique validé et créé par Gitea --- try: repo_name = await _ensure_gitea_repo(base_name) except Exception as e: return { "status": "partial_success_git_failed", "error": str(e), "spec_title": code_data.spec_title, "tree": code_data.tree } work_dir = os.path.abspath(f"./temp_repos/{repo_name}") if os.path.exists(work_dir): shutil.rmtree(work_dir) os.makedirs(work_dir) gitea_host = settings.gitea_base_url.replace("http://", "").replace("https://", "") remote_url = f"http://{settings.gitea_admin_user}:{settings.gitea_token}@{gitea_host}/{settings.gitea_admin_user}/{repo_name}.git" try: # 2. Vérifier si on doit Cloner ou faire un Init (Gitea ayant auto_init=True, il va cloner) check_remote = subprocess.run(["git", "ls-remote", remote_url], cwd=work_dir, capture_output=True) if check_remote.returncode == 0: logger.info(f"[Git] Dépôt distant initialisé. Clonage de la structure de base...") subprocess.run(["git", "clone", remote_url, "."], cwd=work_dir, check=True, capture_output=True) is_update = True else: logger.info(f"[Git] Initialisation locale...") subprocess.run(["git", "init"], cwd=work_dir, check=True, capture_output=True) subprocess.run(["git", "remote", "add", "origin", remote_url], cwd=work_dir, check=True, capture_output=True) is_update = False subprocess.run(["git", "config", "user.name", "ARC Dev Agent"], cwd=work_dir, check=True, capture_output=True) subprocess.run(["git", "config", "user.email", "dev_agent@arc.local"], cwd=work_dir, check=True, capture_output=True) # 3. Écriture des fichiers générés par le LLM for file_info in code_data.files: file_path = os.path.join(work_dir, file_info.path) os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, "w", encoding="utf-8") as f: f.write(file_info.content) # 4. Commit et Push subprocess.run(["git", "add", "."], cwd=work_dir, check=True, capture_output=True) status = subprocess.run(["git", "status", "--porcelain"], cwd=work_dir, capture_output=True, text=True) if status.stdout.strip(): subprocess.run(["git", "commit", "-m", "Update from Dev Agent (AI Generation)"], cwd=work_dir, check=True, capture_output=True) if is_update: subprocess.run(["git", "pull", "origin", "main", "--rebase"], cwd=work_dir, check=True, capture_output=True) subprocess.run(["git", "push", "origin", "main"], cwd=work_dir, check=True, capture_output=True) logger.info(f"[Git] ✅ Push réussi sur {repo_name}") else: logger.info("[Git] Aucun changement détecté, skip commit/push.") return { "status": "success", "repo_url": f"{settings.gitea_base_url}/{settings.gitea_admin_user}/{repo_name}", "files_count": len(code_data.files), "spec_title": code_data.spec_title, "tree": code_data.tree } except Exception as e: logger.error(f"[Git Deployment Error] {str(e)}") return { "status": "partial_success_git_failed", "error": str(e), "spec_title": code_data.spec_title, "tree": code_data.tree } finally: if os.path.exists(work_dir): shutil.rmtree(work_dir) def _generate_fallback_code_output(spec: dict) -> dict: """Génère un livrable minimal de secours en cas de crash du LLM.""" logger.warning("[Dev Agent] Génération du package de secours (Fallback)") title = spec.get("title", "automation_script") fallback = ProjectCodeOutput( spec_title=title, tree=["main.py", "README.md", "requirements.txt"], files=[ GeneratedFile(path="main.py", content="import logging\nlogging.basicConfig(level=logging.INFO)\n\ndef main():\n logging.error('Le Dev Agent a rencontré une erreur de génération.')\n\nif __name__ == '__main__':\n main()"), GeneratedFile(path="README.md", content=f"# {title}\nGénération en mode fallback suite à une erreur technique."), GeneratedFile(path="requirements.txt", content="# Aucune dépendance externe définie (Fallback)\n") ] ) return fallback.model_dump()