525 lines
22 KiB
Python
525 lines
22 KiB
Python
import chainlit as cl
|
|
import unicodedata
|
|
import zipfile
|
|
import httpx
|
|
import json
|
|
import os
|
|
import io
|
|
import re
|
|
|
|
# --- CONFIGURATION GITEA ---
|
|
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "ton_token_gitea_ici")
|
|
|
|
async def push_readme_update_to_gitea(repo_url: str, readme_content: str):
|
|
"""
|
|
Met à jour ou crée le fichier README.md directement sur Gitea via son API REST.
|
|
"""
|
|
if not repo_url:
|
|
return
|
|
|
|
import base64
|
|
clean_url = repo_url.replace(".git", "")
|
|
parts = clean_url.rstrip("/").split("/")
|
|
|
|
if len(parts) >= 2:
|
|
repo_name = parts[-1]
|
|
owner = parts[-2]
|
|
|
|
# API Gitea pour récupérer/modifier le contenu d'un fichier
|
|
gitea_file_url = f"http://git-arc:3000/api/v1/repos/{owner}/{repo_name}/contents/README.md"
|
|
|
|
headers = {
|
|
"Content-Type": "application/json"
|
|
}
|
|
if GITEA_TOKEN:
|
|
headers["Authorization"] = f"token {GITEA_TOKEN}"
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
sha = None
|
|
# 1. On tente de récupérer le SHA du README actuel s'il existe (requis par Gitea pour les updates)
|
|
try:
|
|
get_resp = await client.get(gitea_file_url, headers=headers, timeout=5.0)
|
|
if get_resp.status_code == 200:
|
|
sha = get_resp.json().get("sha")
|
|
except Exception as e:
|
|
print(f"⚠️ Impossible de récupérer le README actuel (il n'existe peut-être pas) : {e}")
|
|
|
|
# 2. Encodage du contenu en base64 (requis par l'API Gitea)
|
|
encoded_content = base64.b64encode(readme_content.encode("utf-8")).decode("utf-8")
|
|
|
|
payload = {
|
|
"content": encoded_content,
|
|
"message": "docs: ajouter le statut de conformité QA au README",
|
|
"branch": "main" # ou "master" selon ta configuration par défaut
|
|
}
|
|
if sha:
|
|
payload["sha"] = sha # Requis si le fichier existe déjà
|
|
|
|
# 3. Écriture (PUT crée ou met à jour le fichier)
|
|
try:
|
|
put_resp = await client.put(gitea_file_url, headers=headers, json=payload, timeout=10.0)
|
|
if put_resp.status_code in [200, 201]:
|
|
print("📝 README.md mis à jour avec succès sur Gitea !")
|
|
else:
|
|
print(f"⚠️ Échec de la mise à jour du README sur Gitea : {put_resp.status_code} - {put_resp.text}")
|
|
except Exception as e:
|
|
print(f"❌ Erreur lors de l'écriture du README sur Gitea : {e}")
|
|
|
|
async def delete_gitea_repository(repo_url: str):
|
|
"""
|
|
Supprime le dépôt sur Gitea via son API REST en cas d'échec ou d'abandon.
|
|
"""
|
|
if not repo_url:
|
|
return
|
|
|
|
# Nettoyage de l'URL pour extraire le owner et le repo_name
|
|
clean_url = repo_url.replace(".git", "")
|
|
parts = clean_url.rstrip("/").split("/")
|
|
|
|
if len(parts) >= 2:
|
|
repo_name = parts[-1]
|
|
owner = parts[-2]
|
|
|
|
# URL interne au réseau Docker pour appeler l'API Gitea
|
|
gitea_api_url = f"http://git-arc:3000/api/v1/repos/{owner}/{repo_name}"
|
|
|
|
headers = {}
|
|
if GITEA_TOKEN:
|
|
headers["Authorization"] = f"token {GITEA_TOKEN}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.delete(gitea_api_url, headers=headers, timeout=10.0)
|
|
if response.status_code == 204:
|
|
print(f"🗑️ Dépôt Gitea {owner}/{repo_name} supprimé avec succès.")
|
|
else:
|
|
print(f"⚠️ Échec de la suppression Gitea : {response.status_code} - {response.text}")
|
|
except Exception as e:
|
|
print(f"❌ Erreur lors de la suppression du dépôt Gitea : {e}")
|
|
|
|
|
|
def inject_conforme_badge(state: dict) -> dict:
|
|
"""
|
|
Injecte la mention 'Conforme' dans le README.md du projet
|
|
dans l'état de session (gère le format liste de dict ou dictionnaire plat).
|
|
"""
|
|
dev_data = state.get("generated_code", {}) or {}
|
|
raw_files = dev_data.get("files", {})
|
|
badge_text = "\n\n---\n🛡️ **Statut QA :** ✅ Conforme\n"
|
|
|
|
if isinstance(raw_files, list):
|
|
readme_found = False
|
|
for f in raw_files:
|
|
if isinstance(f, dict):
|
|
path = f.get("path") or f.get("filename") or f.get("name") or ""
|
|
if path.lower() in ["readme.md", "readme"]:
|
|
content = f.get("content") or f.get("code") or ""
|
|
if "✅ Conforme" not in content:
|
|
new_content = content + badge_text
|
|
if "content" in f:
|
|
f["content"] = new_content
|
|
elif "code" in f:
|
|
f["code"] = new_content
|
|
readme_found = True
|
|
break
|
|
if not readme_found:
|
|
raw_files.append({
|
|
"path": "README.md",
|
|
"content": "# Projet\n" + badge_text
|
|
})
|
|
|
|
elif isinstance(raw_files, dict):
|
|
readme_key = None
|
|
for k in raw_files.keys():
|
|
if k.lower() in ["readme.md", "readme"]:
|
|
readme_key = k
|
|
break
|
|
if readme_key:
|
|
content = raw_files[readme_key] or ""
|
|
if "✅ Conforme" not in content:
|
|
raw_files[readme_key] = content + badge_text
|
|
else:
|
|
raw_files["README.md"] = "# Projet" + badge_text
|
|
|
|
return state
|
|
|
|
|
|
async def reset_to_start(message_text: str = "Bonjour 👋 Je suis ARC. Décrivez-moi votre besoin logiciel."):
|
|
"""
|
|
Réinitialise complètement l'état de la session utilisateur.
|
|
"""
|
|
initial_state = {
|
|
"user_input": "",
|
|
"user_feedback": None,
|
|
"chat_history": [],
|
|
"spec": {},
|
|
"status": "start",
|
|
"loop_count": 0,
|
|
"existing_project": None,
|
|
"generated_code": None,
|
|
"qa_result": None,
|
|
"is_completed": False
|
|
}
|
|
cl.user_session.set("graph_state", initial_state)
|
|
cl.user_session.set("current_files", {})
|
|
cl.user_session.set("side_code_viewer", None)
|
|
cl.user_session.set("files_message", None)
|
|
|
|
await cl.Message(content=message_text).send()
|
|
|
|
def sanitize_repo_name(name: str) -> str:
|
|
"""
|
|
Nettoie une chaîne pour la rendre 100% compatible avec un nom de dépôt Git.
|
|
Gère les accents, les espaces et supprime les caractères interdits.
|
|
"""
|
|
if not name:
|
|
return "mon-projet"
|
|
|
|
normalized = unicodedata.normalize('NFKD', name)
|
|
no_accent = normalized.encode('ascii', 'ignore').decode('utf-8')
|
|
lowered = no_accent.lower()
|
|
spaced_to_hyphen = lowered.replace(" ", "_")
|
|
|
|
cleaned = re.sub(r'[^a-z0-9-_]', '', spaced_to_hyphen)
|
|
cleaned = re.sub(r'-+', '-', cleaned)
|
|
cleaned = re.sub(r'_+', '_', cleaned)
|
|
|
|
return cleaned.strip('-_')
|
|
|
|
async def render_workflow_state(new_state: dict):
|
|
"""
|
|
Fonction centrale pour aiguiller l'affichage Chainlit
|
|
en fonction du statut renvoyé par LangGraph.
|
|
"""
|
|
current_status = new_state.get("status")
|
|
|
|
if current_status == "spec_incomplete":
|
|
spec = new_state.get("spec", {})
|
|
question = spec.get("clarifying_question")
|
|
await cl.Message(content=f"**Spécifications incomplètes**\n\n{question}").send()
|
|
|
|
elif current_status == "spec_ready":
|
|
spec = new_state.get("spec", {})
|
|
|
|
summary = "### Éléments importants à retenir de votre projet :\n\n"
|
|
summary += f"- **Nom du projet** : {spec.get('title')}\n"
|
|
summary += f"- **Description** : {spec.get('description')}\n"
|
|
summary += "- **Actions** :\n"
|
|
summary += "\n".join(f" - {req}" for req in spec.get("requirements", [])) + "\n"
|
|
summary += "- **Contraintes** :\n"
|
|
summary += "\n".join(f" - {constraint}" for constraint in spec.get("constraints", [])) + "\n"
|
|
summary += f"- **Langage** : {spec.get('language')}\n\n"
|
|
summary += "💡 *(Le nom du projet indiqué pourra être modifié par la suite)*"
|
|
|
|
await cl.Message(content=summary).send()
|
|
|
|
res = await cl.AskActionMessage(
|
|
content="**Est-ce que cela vous convient ?**",
|
|
actions=[
|
|
cl.Action(name="oui", payload={"value": "oui"}, label="Oui, c'est parfait 👍"),
|
|
cl.Action(name="non", payload={"value": "non"}, label="Non, modifier ❌")
|
|
],
|
|
timeout=3600
|
|
).send()
|
|
|
|
if res is None:
|
|
await cl.Message(
|
|
content="⏰ **Session expirée.** Si tu es toujours là, envoie un message pour relancer l'analyse."
|
|
).send()
|
|
return
|
|
|
|
if res and res.get("name") == "oui":
|
|
proposed_title = spec.get('title', 'mon_projet')
|
|
|
|
name_choice = await cl.AskActionMessage(
|
|
content=f"Le nom proposé pour le projet est : **{proposed_title}**.\nSouhaitez-vous le conserver ou le modifier ?",
|
|
actions=[
|
|
cl.Action(name="garder_nom", payload={"value": "keep"}, label="Conserver ce nom 🏷️"),
|
|
cl.Action(name="modifier_nom", payload={"value": "change"}, label="Choisir un autre nom ✏️")
|
|
],
|
|
timeout=3600
|
|
).send()
|
|
|
|
if name_choice is None:
|
|
await cl.Message(content="⏰ **Session expirée.** Envoie un message pour reprendre.").send()
|
|
return
|
|
|
|
if name_choice.get("name") == "modifier_nom":
|
|
new_name_res = await cl.AskUserMessage(
|
|
content="Saisissez le nouveau nom de votre projet : 👇",
|
|
timeout=600
|
|
).send()
|
|
|
|
if new_name_res and new_name_res.get("output"):
|
|
raw_title = new_name_res["output"].strip()
|
|
custom_title = sanitize_repo_name(raw_title)
|
|
spec['title'] = custom_title
|
|
new_state['spec'] = spec
|
|
await cl.Message(content=f"🏷️ Nom du projet configuré sur : **{custom_title}**").send()
|
|
|
|
await cl.Message(content="🚀 **Spécifications validées !** Lancement de la génération du code...").send()
|
|
|
|
new_state["status"] = "spec_approved"
|
|
cl.user_session.set("graph_state", new_state)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
"http://127.0.0.1:8000/api/workflow/run",
|
|
json=new_state,
|
|
timeout=1200.0
|
|
)
|
|
final_state = response.json()
|
|
cl.user_session.set("graph_state", final_state)
|
|
|
|
await render_workflow_state(final_state)
|
|
|
|
else:
|
|
new_state["status"] = "spec_incomplete"
|
|
cl.user_session.set("graph_state", new_state)
|
|
|
|
await cl.Message(
|
|
content="🔄 **Compris.** Qu'est-ce qui ne convient pas ? S'il vous plaît, précisez les éléments manquants ou à corriger :"
|
|
).send()
|
|
|
|
elif current_status in ["wait_human_review", "approved_by_human"]:
|
|
dev_data = new_state.get("generated_code", {}) or {}
|
|
qa_res = new_state.get("qa_result", {}) or {}
|
|
|
|
gitea_url = dev_data.get("repo_url", "")
|
|
if gitea_url and "git-arc:3000" in gitea_url:
|
|
gitea_url = gitea_url.replace("git-arc:3000", "localhost:3000")
|
|
|
|
is_safe = qa_res.get("is_complete_and_safe", False)
|
|
|
|
# ==================== PARCOURS 1 : LE PROJET EST CONFORME ====================
|
|
if is_safe:
|
|
raw_files = dev_data.get("files", {})
|
|
files = {}
|
|
if isinstance(raw_files, list):
|
|
for f in raw_files:
|
|
if isinstance(f, dict):
|
|
path = f.get("path") or f.get("filename") or f.get("name")
|
|
content = f.get("content") or f.get("code") or ""
|
|
if path:
|
|
files[path] = content
|
|
elif isinstance(raw_files, dict):
|
|
files = raw_files
|
|
|
|
access_md = "**Résultat global QA :** ✅ CONFORME\n\n"
|
|
access_md += "### 🔗 Accès au projet\n\n"
|
|
if gitea_url:
|
|
access_md += f"🔗 **Lien vers le dépôt Gitea :** [Accéder au dépôt]({gitea_url})\n\n"
|
|
|
|
project_title = dev_data.get("spec_title", "projet")
|
|
message_elements = []
|
|
|
|
if files:
|
|
zip_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
|
|
for filepath, file_content in files.items():
|
|
zip_file.writestr(filepath, file_content)
|
|
zip_bytes = zip_buffer.getvalue()
|
|
|
|
message_elements.append(
|
|
cl.File(name=f"{project_title}.zip", content=zip_bytes, display="inline")
|
|
)
|
|
else:
|
|
access_md += "⚠️ Aucun fichier source trouvé pour générer le ZIP.\n"
|
|
|
|
await cl.Message(content=access_md, elements=message_elements).send()
|
|
|
|
cl.user_session.set("current_files", files)
|
|
|
|
file_actions = []
|
|
for filename in files.keys():
|
|
file_actions.append(
|
|
cl.Action(
|
|
name="open_file",
|
|
payload={"filename": filename},
|
|
label=f"📄 {filename}"
|
|
)
|
|
)
|
|
|
|
files_message = cl.Message(
|
|
content="### 📂 Code Source Produit\n\nClique sur un fichier pour l'afficher :",
|
|
actions=file_actions
|
|
)
|
|
await files_message.send()
|
|
cl.user_session.set("files_message", files_message)
|
|
|
|
res = await cl.AskActionMessage(
|
|
content="**Souhaitez-vous valider et packager ce livrable ?**",
|
|
actions=[
|
|
cl.Action(name="valider_projet", payload={"value": "approve"}, label="✅ Valider & Livrer", description="Ajoute le statut conforme au README et livre l'archive"),
|
|
cl.Action(name="refuser_projet", payload={"value": "refuse"}, label="❌ Refuser et corriger", description="Renvoie le projet au PM avec tes commentaires")
|
|
],
|
|
timeout=3600
|
|
).send()
|
|
|
|
if res and res.get("name") == "valider_projet":
|
|
await cl.Message(content="🎉 **Projet validé et marqué comme conforme !**").send()
|
|
|
|
new_state = inject_conforme_badge(new_state)
|
|
|
|
readme_content = "# Projet\n\n---\n🛡️ **Statut QA :** ✅ Conforme\n"
|
|
dev_data = new_state.get("generated_code", {}) or {}
|
|
raw_files = dev_data.get("files", {})
|
|
|
|
if isinstance(raw_files, list):
|
|
for f in raw_files:
|
|
if isinstance(f, dict):
|
|
path = f.get("path") or f.get("filename") or f.get("name") or ""
|
|
if path.lower() in ["readme.md", "readme"]:
|
|
readme_content = f.get("content") or f.get("code") or ""
|
|
break
|
|
elif isinstance(raw_files, dict):
|
|
for k, v in raw_files.items():
|
|
if k.lower() in ["readme.md", "readme"]:
|
|
readme_content = v
|
|
break
|
|
|
|
await push_readme_update_to_gitea(gitea_url, readme_content)
|
|
|
|
new_state["status"] = "human_approved"
|
|
cl.user_session.set("graph_state", new_state)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post("http://127.0.0.1:8000/api/workflow/run", json=new_state, timeout=1200.0)
|
|
|
|
final_state = resp.json()
|
|
cl.user_session.set("graph_state", final_state)
|
|
|
|
elif res and res.get("name") == "refuser_projet":
|
|
feedback_user = await cl.AskUserMessage(
|
|
content="Veuillez décrire les corrections ou les modifications à apporter au projet.",
|
|
timeout=1200
|
|
).send()
|
|
|
|
if feedback_user:
|
|
new_state["status"] = "human_refused"
|
|
new_state["user_feedback"] = feedback_user["output"]
|
|
cl.user_session.set("graph_state", new_state)
|
|
|
|
await cl.Message(content="🔄 **Feedback transmis.** Prise en compte des modifications...").send()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post("http://127.0.0.1:8000/api/workflow/run", json=new_state, timeout=1200.0)
|
|
|
|
loop_state = resp.json()
|
|
cl.user_session.set("graph_state", loop_state)
|
|
|
|
await render_workflow_state(loop_state)
|
|
|
|
# ==================== PARCOURS 2 : LE PROJET EST NON CONFORME ====================
|
|
else:
|
|
qa_feedback = qa_res.get("feedback", "L'agent QA a détecté des anomalies majeures sans description.")
|
|
|
|
non_compliant_md = (
|
|
"🚨 **Projet Non Conforme**\n\n"
|
|
"Malheureusement, le code produit par ARC n'a pas passé les critères de sécurité ou de qualité requis.\n\n"
|
|
f"### 📋 Rapport d'audit QA :\n"
|
|
f"> {qa_feedback}\n\n"
|
|
"Pour des raisons de sécurité et de qualité, le projet ne peut pas être livré en l'état.\n\n"
|
|
)
|
|
|
|
await cl.Message(content=non_compliant_md).send()
|
|
|
|
res = await cl.AskActionMessage(
|
|
content="**Que souhaitez-vous faire ?**",
|
|
actions=[
|
|
cl.Action(name="recommencer_projet", payload={"value": "restart"}, label="🔄 Recommencer de zéro", description="Supprime le dépôt et réinitialise l'analyse"),
|
|
cl.Action(name="abandonner_projet", payload={"value": "abort"}, label="🗑️ Abandonner & Quitter", description="Supprime le dépôt et réinitialise l'application")
|
|
],
|
|
timeout=3600
|
|
).send()
|
|
|
|
if res and res.get("name") in ["recommencer_projet", "abandonner_projet"]:
|
|
await cl.Message(content="🗑️ **Nettoyage en cours... Suppression du dépôt Gitea...**").send()
|
|
|
|
# Suppression du dépôt distant
|
|
await delete_gitea_repository(gitea_url)
|
|
|
|
if res.get("name") == "recommencer_projet":
|
|
reset_msg = (
|
|
"🔄 **Ancien projet supprimé avec succès.**\n\n"
|
|
"Faisons table rase ! Décrivez-moi votre besoin logiciel pour repartir sur de nouvelles bases : 👇"
|
|
)
|
|
await reset_to_start(reset_msg)
|
|
else:
|
|
reset_msg = (
|
|
"❌ **Session fermée et dépôt nettoyé.**\n\n"
|
|
"Si vous avez un nouveau besoin à soumettre plus tard, envoie simplement un message pour démarrer."
|
|
)
|
|
await reset_to_start(reset_msg)
|
|
|
|
elif current_status == "delivered":
|
|
await cl.Message(content="✅ Ce projet a déjà été traité et livré.").send()
|
|
else:
|
|
await cl.Message(
|
|
content=f"Résultat du traitement :\n```json\n{json.dumps(new_state, indent=2, ensure_ascii=False)}\n```"
|
|
).send()
|
|
|
|
|
|
# --- ÉVÉNEMENTS CHAINLIT STANDARDS ---
|
|
@cl.action_callback("open_file")
|
|
async def on_open_file(action: cl.Action):
|
|
filename = action.payload.get("filename")
|
|
files = cl.user_session.get("current_files", {})
|
|
content = files.get(filename, "")
|
|
if content is None:
|
|
content = ""
|
|
if not str(content).strip():
|
|
content = "(Fichier vide)"
|
|
|
|
ext = filename.split(".")[-1] if "." in filename else "text"
|
|
is_md = ext in ("md", "markdown")
|
|
|
|
files_message = cl.user_session.get("files_message")
|
|
message_id = files_message.id if files_message else None
|
|
|
|
old_viewer = cl.user_session.get("side_code_viewer")
|
|
if old_viewer:
|
|
await old_viewer.remove()
|
|
|
|
new_viewer = cl.Text(
|
|
name=filename,
|
|
content=str(content),
|
|
language=None if is_md else ext,
|
|
display="side",
|
|
)
|
|
|
|
await new_viewer.send(for_id=message_id)
|
|
cl.user_session.set("side_code_viewer", new_viewer)
|
|
|
|
|
|
@cl.on_chat_start
|
|
async def on_chat_start():
|
|
await reset_to_start()
|
|
|
|
|
|
@cl.on_message
|
|
async def on_message(message: cl.Message):
|
|
state = cl.user_session.get("graph_state")
|
|
|
|
if "chat_history" not in state:
|
|
state["chat_history"] = []
|
|
if "status" not in state:
|
|
state["status"] = "start"
|
|
|
|
if state.get("status") == "spec_incomplete":
|
|
state["user_feedback"] = message.content
|
|
else:
|
|
state["user_input"] = message.content
|
|
state["user_feedback"] = None
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
"http://127.0.0.1:8000/api/workflow/run",
|
|
json=state,
|
|
timeout=1200.0
|
|
)
|
|
|
|
new_state = response.json()
|
|
cl.user_session.set("graph_state", new_state)
|
|
|
|
await render_workflow_state(new_state) |