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, repo_url: str = None, files: 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. S'appuie sur repo_url si on est dans une boucle de correction. """ 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 files: user_content += f"💻 CODE ACTUEL (Version précédente à corriger) :\n{json.dumps(files, 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"), repo_url=repo_url) 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, repo_url=repo_url, files=files) async def _ensure_gitea_repo(base_name: str, repo_url: str = None) -> 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. """ if repo_url: repo_name = repo_url.rstrip("/").split("/")[-1] logger.info(f"[Gitea] Mode correction : réutilisation du dépôt existant '{repo_name}'") return repo_name 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: break elif response.status_code == 200: repo_name = f"{base_name}_{counter}" counter += 1 else: 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, repo_url: str = None) -> 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 == "_") files_list = [f.model_dump() for f in code_data.files] try: repo_name = await _ensure_gitea_repo(base_name, repo_url=repo_url) except Exception as e: return { "status": "partial_success_git_failed", "error": str(e), "spec_title": code_data.spec_title, "tree": code_data.tree, "files": files_list } 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, "files": files_list } 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, "files": files_list } finally: if os.path.exists(work_dir): shutil.rmtree(work_dir) def _generate_fallback_code_output(spec: dict, repo_url: str = None, files: list = None) -> dict: """ Génère un livrable de secours. Récupère et reconstruit l'état du code à partir des fichiers existants s'ils sont fournis, sinon génère un README d'erreur. """ logger.warning("[Dev Agent] Génération du package de secours (Fallback)") title = spec.get("title", "automation_script") fallback_files = [] fallback_tree = [] if files: for f in files: path = "" content = "" if isinstance(f, dict): path = f.get("path") or f.get("filename") or f.get("name") or "" content = f.get("content") or f.get("code") or "" elif hasattr(f, "path") and hasattr(f, "content"): path = getattr(f, "path", "") content = getattr(f, "content", "") if path: fallback_files.append(GeneratedFile(path=path, content=content)) fallback_tree.append(path) if not fallback_files: fallback_files = [ GeneratedFile( path="README.md", content=f"# {title}\nErreur interne : le code n'a pas pu être généré par l'agent Dev lors du premier essai." ) ] fallback_tree = ["README.md"] fallback = ProjectCodeOutput( repo_url=repo_url, spec_title=title, tree=fallback_tree, files=fallback_files ) return fallback.model_dump()