# -*- coding: utf-8 -*- """ Nom du fichier : projects.py Chemin : /var/www/html/gitlab-bridge/app/api/v1/projects.py Description : Endpoints API liés aux projets GitLab compatibles OpenWebUI. Ce fichier expose : - le résumé d'un projet GitLab - la création d'un projet GitLab - la pré-vérification avant création - une orchestration conversationnelle de création Options éventuelles : - POST /api/v1/projects/summary - POST /api/v1/projects/create - POST /api/v1/projects/check-create - POST /api/v1/projects/assistant-create Auteur : Sylvain SCATTOLINI Date de création / modification : 2026-03-26 Version : 1.8 """ from __future__ import annotations import logging import re from datetime import datetime, timedelta, timezone from typing import Annotated, Any from fastapi import APIRouter, Body, HTTPException, status from app.clients.gitlab_client import GitLabClient from app.core.exceptions import GitLabApiError from app.schemas.projects import ( CandidateGroup, ProjectAssistantCreateRequest, ProjectAssistantCreateResponse, ProjectCheckCreateRequest, ProjectCheckCreateResponse, ProjectCreatePayload, ProjectCreateRequest, ProjectCreateResponse, ProjectInfo, ProjectProgressResponse, ProjectProgressRequest, ProjectTasksRequest, ProjectTasksResponse, ProjectSummaryResponse, SimilarProject, SummaryPayload, ProjectAssistantProgressRequest, ProjectAssistantProgressResponse, ProjectAssistantTasksRequest, ProjectAssistantTasksResponse, ProjectAssistantListRequest, ProjectAssistantListResponse, ProjectListItem, ) router = APIRouter(prefix="/api/v1/projects", tags=["projects"]) logger = logging.getLogger(__name__) _gl_client = GitLabClient() def _gitlab_request( method: str, endpoint: str, *, query_params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, ) -> Any: """Adaptateur vers GitLabClient — remplace urllib, bénéficie du pooling et du retry.""" try: return _gl_client._request(method, endpoint, params=query_params, json_body=payload) except GitLabApiError as exc: http_status = exc.http_status or 502 raise HTTPException(status_code=http_status, detail=exc.message) from exc from app.utils.helpers import normalize_text as _normalize_text, compute_similarity_score as _compute_similarity_score def _normalize_project_path(value: str) -> str: cleaned = value.strip().strip("/") if not cleaned: raise HTTPException(status_code=400, detail="Le chemin projet est vide.") return cleaned def _normalize_namespace_path(value: str) -> str: cleaned = value.strip().strip("/") if not cleaned: raise HTTPException(status_code=400, detail="Le namespace_path est vide.") return cleaned def _normalize_project_slug(value: str) -> str: cleaned = value.strip().strip("/") if not cleaned: raise HTTPException(status_code=400, detail="Le champ path est vide.") if not re.fullmatch(r"[a-zA-Z0-9._-]+", cleaned): raise HTTPException(status_code=400, detail="Le champ path contient des caractères non autorisés.") return cleaned def _normalize_visibility(value: str) -> str: visibility = value.strip().lower() if visibility not in {"private", "internal", "public"}: raise HTTPException(status_code=400, detail="La visibilité doit être private, internal ou public.") return visibility def _slugify_project_name(name: str) -> str: from app.utils.helpers import slugify try: slug = slugify(name) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return _normalize_project_slug(slug) def _get_project_by_path(project_path: str) -> dict[str, Any]: encoded_project_path = quote(project_path, safe="") project = _gitlab_request("GET", f"/projects/{encoded_project_path}") if not isinstance(project, dict) or "id" not in project: raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la récupération du projet.") return project def _list_open_issues(project_id: int) -> list[dict[str, Any]]: issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": "opened", "per_page": 100, "page": 1, "order_by": "updated_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] return [issue for issue in issues if isinstance(issue, dict)] def _parse_gitlab_datetime(value: str | None) -> datetime | None: if not value or not isinstance(value, str): return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) except ValueError: return None def _list_recent_closed_issues(project_id: int, since_dt: datetime) -> list[dict[str, Any]]: issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": "closed", "updated_after": since_dt.replace(microsecond=0).isoformat().replace("+00:00", "Z"), "per_page": 100, "page": 1, "order_by": "updated_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] filtered_issues: list[dict[str, Any]] = [] for issue in issues: if not isinstance(issue, dict): continue closed_at = _parse_gitlab_datetime(issue.get("closed_at")) if closed_at is not None and closed_at >= since_dt: filtered_issues.append(issue) return filtered_issues def _list_milestones(project_id: int) -> list[dict[str, Any]]: milestones = _gitlab_request( "GET", f"/projects/{project_id}/milestones", query_params={ "state": "active", "per_page": 20, "page": 1, }, ) if not isinstance(milestones, list): return [] return [milestone for milestone in milestones if isinstance(milestone, dict)] def _build_markdown_summary( project_name: str, project_path: str, period_days: int, open_issues_count: int, closed_issues_count: int, milestones: list[dict[str, Any]], ) -> str: lines: list[str] = [ f"## Résumé du projet {project_name}", "", f"- Projet : `{project_path}`", f"- Période analysée : {period_days} jour(s)", f"- Issues ouvertes : {open_issues_count}", f"- Issues fermées sur la période : {closed_issues_count}", ] if milestones: lines.append("- Milestones actives :") for milestone in milestones[:5]: title = str(milestone.get("title", "Sans titre")) due_date = milestone.get("due_date") lines.append(f" - {title}" + (f" (échéance : {due_date})" if due_date else "")) else: lines.append("- Aucune milestone active") return "\n".join(lines) def _get_namespace_by_path(namespace_path: str) -> dict[str, Any]: encoded_namespace_path = quote(namespace_path, safe="") group = _gitlab_request("GET", f"/groups/{encoded_namespace_path}") if not isinstance(group, dict) or "id" not in group: raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la récupération du namespace.") return group def _delete_project(project_id: int) -> None: _gitlab_request("DELETE", f"/projects/{project_id}") def _list_groups(root_hint: str, limit: int) -> list[dict[str, Any]]: groups = _gitlab_request( "GET", "/groups", query_params={ "all_available": True, "per_page": max(limit * 5, 50), "page": 1, }, ) if not isinstance(groups, list): return [] normalized_root_hint = _normalize_text(root_hint) filtered: list[dict[str, Any]] = [] for group in groups: if not isinstance(group, dict): continue full_path = str(group.get("full_path", "")).strip() if not full_path: continue if normalized_root_hint and not _normalize_text(full_path).startswith(normalized_root_hint): continue filtered.append(group) return filtered def _search_candidate_groups(target_group_hint: str, root_hint: str, limit: int) -> list[CandidateGroup]: groups = _list_groups(root_hint, limit) normalized_hint = _normalize_text(target_group_hint) candidates: list[CandidateGroup] = [] for group in groups: name = str(group.get("name", "")) full_path = str(group.get("full_path", "")) score = max( _compute_similarity_score(normalized_hint, name), _compute_similarity_score(normalized_hint, full_path), ) if normalized_hint in _normalize_text(name): score = max(score, 95) elif normalized_hint in _normalize_text(full_path): score = max(score, 90) if score < 40: continue candidates.append( CandidateGroup( id=int(group["id"]), name=name, full_path=full_path, score=score, ) ) candidates.sort(key=lambda item: (-item.score, item.full_path)) return candidates[:limit] def _search_similar_projects(project_name: str, root_hint: str, limit: int) -> list[SimilarProject]: projects = _gitlab_request( "GET", "/projects", query_params={ "search": project_name, "simple": True, "per_page": max(limit * 5, 50), "page": 1, }, ) if not isinstance(projects, list): return [] normalized_root_hint = _normalize_text(root_hint) candidates: list[SimilarProject] = [] for project in projects: if not isinstance(project, dict): continue path_with_namespace = str(project.get("path_with_namespace", "")) if not path_with_namespace: continue if normalized_root_hint and not _normalize_text(path_with_namespace).startswith(normalized_root_hint): continue score = max( _compute_similarity_score(project_name, str(project.get("name", ""))), _compute_similarity_score(project_name, path_with_namespace), ) if score < 40: continue candidates.append( SimilarProject( id=int(project["id"]), name=str(project.get("name", "")), path_with_namespace=path_with_namespace, web_url=str(project.get("web_url", "")), score=score, ) ) candidates.sort(key=lambda item: (-item.score, item.path_with_namespace)) return candidates[:limit] def _evaluate_project_creation( project_name: str, target_group_hint: str, root_hint: str, limit: int, ) -> ProjectCheckCreateResponse: safe_project_name = project_name.strip() safe_target_group_hint = target_group_hint.strip() safe_root_hint = root_hint.strip().strip("/") safe_limit = int(limit) suggested_project_path = _slugify_project_name(safe_project_name) candidate_groups = _search_candidate_groups(safe_target_group_hint, safe_root_hint, safe_limit) similar_projects = _search_similar_projects(safe_project_name, safe_root_hint, safe_limit) if not candidate_groups: return ProjectCheckCreateResponse( success=True, status="group_not_found", suggested_project_path=suggested_project_path, candidate_groups=[], similar_projects=similar_projects, message=f"Je ne trouve pas de groupe correspondant à '{safe_target_group_hint}'.", ) top_group = candidate_groups[0] ambiguous = len(candidate_groups) > 1 and candidate_groups[1].score >= top_group.score - 10 if ambiguous: return ProjectCheckCreateResponse( success=True, status="ambiguous_group", suggested_project_path=suggested_project_path, candidate_groups=candidate_groups, similar_projects=similar_projects, message="Plusieurs groupes sont possibles.", ) return ProjectCheckCreateResponse( success=True, status="ready", suggested_project_path=suggested_project_path, resolved_group=top_group.full_path, candidate_groups=[top_group], similar_projects=similar_projects, message="Création possible.", ) def _build_similar_projects_from_candidates(candidates: list[tuple[int, dict[str, Any]]]) -> list[SimilarProject]: results: list[SimilarProject] = [] for score, project in candidates[:5]: results.append( SimilarProject( id=int(project["id"]), name=str(project.get("name", "")), path_with_namespace=str(project.get("path_with_namespace", "")), web_url=str(project.get("web_url", "")), score=score, ) ) return results def _resolve_project_candidates(project_hint: str, root_hint: str) -> tuple[str, dict[str, Any] | None, list[SimilarProject]]: candidates = _search_best_project(project_hint, root_hint) if not candidates: return "not_found", None, [] if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return "ambiguous", None, _build_similar_projects_from_candidates(candidates) return "ready", candidates[0][1], _build_similar_projects_from_candidates(candidates) def _get_project_tasks_data(project: dict[str, Any]) -> tuple[ProjectInfo, list[str], list[str]]: project_id = int(project["id"]) open_issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={"state": "opened", "per_page": 100}, ) closed_issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={"state": "closed", "per_page": 100}, ) project_info = ProjectInfo( id=project_id, name=str(project["name"]), path_with_namespace=str(project["path_with_namespace"]), ) open_titles = [str(issue.get("title", "")) for issue in open_issues if isinstance(issue, dict)] closed_titles = [str(issue.get("title", "")) for issue in closed_issues if isinstance(issue, dict)] return project_info, open_titles, closed_titles def _get_project_progress_data(project: dict[str, Any]) -> tuple[ProjectInfo, int, int, int]: project_id = int(project["id"]) open_issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={"state": "opened", "per_page": 100}, ) closed_issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={"state": "closed", "per_page": 100}, ) total = len(open_issues) + len(closed_issues) closed = len(closed_issues) progress_percent = int((closed / total) * 100) if total > 0 else 0 project_info = ProjectInfo( id=project_id, name=str(project["name"]), path_with_namespace=str(project["path_with_namespace"]), ) return project_info, total, closed, progress_percent def _list_projects_in_group(group_path: str, limit: int) -> list[ProjectListItem]: safe_group_path = _normalize_namespace_path(group_path) group = _get_namespace_by_path(safe_group_path) group_id = int(group["id"]) projects = _gitlab_request( "GET", f"/groups/{group_id}/projects", query_params={ "include_subgroups": False, "per_page": min(limit, 100), "page": 1, "simple": True, "order_by": "name", "sort": "asc", }, ) if not isinstance(projects, list): return [] results: list[ProjectListItem] = [] for project in projects: if not isinstance(project, dict): continue results.append( ProjectListItem( id=int(project["id"]), name=str(project.get("name", "")), path_with_namespace=str(project.get("path_with_namespace", "")), web_url=str(project.get("web_url", "")), visibility=str(project.get("visibility", "")), open_issues_count=int(project.get("open_issues_count", 0) or 0), ) ) results.sort(key=lambda item: item.name.lower()) return results def _resolve_group_candidates(group_hint: str, root_hint: str, limit: int) -> tuple[str, str | None, list[CandidateGroup]]: candidates = _search_candidate_groups(group_hint, root_hint, limit) if not candidates: return "not_found", None, [] top_group = candidates[0] ambiguous = len(candidates) > 1 and candidates[1].score >= top_group.score - 10 if ambiguous: return "ambiguous", None, candidates return "ready", top_group.full_path, [top_group] # --------------------------------------------------------- # Endpoints API # --------------------------------------------------------- @router.post( "/check-create", operation_id="check_create_project", summary="Pré-vérifier une création de projet GitLab", description="Recherche les groupes candidats et les projets similaires avant création.", response_model=ProjectCheckCreateResponse, ) def check_create_project(payload: ProjectCheckCreateRequest) -> ProjectCheckCreateResponse: try: logger.info( "Payload reçu sur /api/v1/projects/check-create : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) return _evaluate_project_creation( project_name=payload.project_name, target_group_hint=payload.target_group_hint, root_hint=payload.root_hint, limit=payload.limit, ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans check_create_project : %s", str(exc)) raise HTTPException(status_code=500, detail="Erreur interne lors de la pré-vérification du projet.") from exc @router.post( "/assistant-list", operation_id="assistant_list_projects", summary="Assistant liste des projets GitLab", description="Résout un groupe, demande clarification si besoin, puis retourne la liste des projets.", response_model=ProjectAssistantListResponse, ) def assistant_list_projects( payload: ProjectAssistantListRequest, ) -> ProjectAssistantListResponse: try: logger.info( "Payload reçu sur /api/v1/projects/assistant-list : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) if payload.confirm: if not payload.resolved_group_path.strip(): raise HTTPException( status_code=400, detail="Le champ resolved_group_path est obligatoire quand confirm=true.", ) resolved_group = _normalize_namespace_path(payload.resolved_group_path) projects = _list_projects_in_group(resolved_group, safe_limit) return ProjectAssistantListResponse( success=True, status="resolved", message=f"Liste des projets récupérée pour {resolved_group}", resolved_group=resolved_group, candidate_groups=[], projects=projects, ) status_resolve, resolved_group, candidate_groups = _resolve_group_candidates( payload.group_hint, payload.root_hint, safe_limit, ) if status_resolve == "not_found": return ProjectAssistantListResponse( success=True, status="not_found", message=f"Je ne trouve pas de groupe correspondant à '{payload.group_hint}'.", resolved_group=None, candidate_groups=[], projects=[], ) if status_resolve == "ambiguous": return ProjectAssistantListResponse( success=True, status="clarification_needed", message="J’ai trouvé plusieurs groupes possibles. Merci de préciser lequel utiliser.", resolved_group=None, candidate_groups=candidate_groups, projects=[], ) assert resolved_group is not None projects = _list_projects_in_group(resolved_group, safe_limit) return ProjectAssistantListResponse( success=True, status="resolved", message=f"Liste des projets récupérée pour {resolved_group}", resolved_group=resolved_group, candidate_groups=candidate_groups, projects=projects, ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_list_projects : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la récupération assistée des projets.", ) from exc @router.post( "/assistant-create", operation_id="assistant_create_project", summary="Assistant de création de projet GitLab", description="Orchestre la vérification, la clarification et la création d'un projet GitLab.", response_model=ProjectAssistantCreateResponse, ) def assistant_create_project(payload: ProjectAssistantCreateRequest) -> ProjectAssistantCreateResponse: try: logger.info( "Payload reçu sur /api/v1/projects/assistant-create : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_project_name = payload.project_name.strip() safe_root_hint = payload.root_hint.strip().strip("/") safe_description = payload.description.strip() safe_visibility = _normalize_visibility(payload.visibility) safe_initialize_with_readme = bool(payload.initialize_with_readme) safe_limit = int(payload.limit) if payload.confirm: if not payload.resolved_group.strip(): raise HTTPException( status_code=400, detail="Le champ resolved_group est obligatoire quand confirm=true.", ) namespace_path = _normalize_namespace_path(payload.resolved_group) project_path = _slugify_project_name(safe_project_name) create_response = create_project( ProjectCreateRequest( name=safe_project_name, path=project_path, namespace_path=namespace_path, description=safe_description, visibility=safe_visibility, initialize_with_readme=safe_initialize_with_readme, ) ) return ProjectAssistantCreateResponse( success=True, status="created", message=f"Projet créé : {create_response.project.path_with_namespace}", project=create_response.project, suggested_project_path=project_path, resolved_group=namespace_path, candidate_groups=[], similar_projects=[], ) check_response = _evaluate_project_creation( project_name=safe_project_name, target_group_hint=payload.target_group_hint, root_hint=safe_root_hint, limit=safe_limit, ) if check_response.status == "ready" and check_response.resolved_group: create_response = create_project( ProjectCreateRequest( name=safe_project_name, path=check_response.suggested_project_path, namespace_path=check_response.resolved_group, description=safe_description, visibility=safe_visibility, initialize_with_readme=safe_initialize_with_readme, ) ) return ProjectAssistantCreateResponse( success=True, status="created", message=f"Projet créé : {create_response.project.path_with_namespace}", project=create_response.project, suggested_project_path=check_response.suggested_project_path, resolved_group=check_response.resolved_group, candidate_groups=check_response.candidate_groups, similar_projects=check_response.similar_projects, ) if check_response.status == "ambiguous_group": return ProjectAssistantCreateResponse( success=True, status="clarification_needed", message=( "J’ai trouvé plusieurs groupes possibles. " "Merci de préciser lequel utiliser." ), project=None, suggested_project_path=check_response.suggested_project_path, resolved_group=None, candidate_groups=check_response.candidate_groups, similar_projects=check_response.similar_projects, ) return ProjectAssistantCreateResponse( success=True, status="group_not_found", message=check_response.message, project=None, suggested_project_path=check_response.suggested_project_path, resolved_group=None, candidate_groups=check_response.candidate_groups, similar_projects=check_response.similar_projects, ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_create_project : %s", str(exc)) raise HTTPException(status_code=500, detail="Erreur interne lors de l'assistant de création projet.") from exc def _search_best_project(project_hint: str, root_hint: str = ""): projects = _gitlab_request( "GET", "/projects", query_params={ "search": project_hint, "simple": True, "per_page": 50, }, ) if not isinstance(projects, list): return [] candidates = [] for project in projects: if not isinstance(project, dict): continue path = str(project.get("path_with_namespace", "")) if root_hint and not _normalize_text(path).startswith(_normalize_text(root_hint)): continue score = max( _compute_similarity_score(project_hint, str(project.get("name", ""))), _compute_similarity_score(project_hint, path), ) if score < 40: continue candidates.append((score, project)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates @router.post( "/create", operation_id="create_project", summary="Créer un projet GitLab", description="Crée un projet GitLab à partir d'un payload simple compatible OpenWebUI.", response_model=ProjectCreateResponse, ) def create_project(payload: ProjectCreateRequest) -> ProjectCreateResponse: try: safe_name = payload.name.strip() safe_namespace_path = _normalize_namespace_path(payload.namespace_path) safe_description = payload.description.strip() safe_visibility = _normalize_visibility(payload.visibility) safe_initialize_with_readme = bool(payload.initialize_with_readme) raw_path = payload.path.strip() if "/" in raw_path: raw_path = _slugify_project_name(safe_name) else: raw_path = _normalize_text(raw_path) raw_path = re.sub(r"\s+", "-", raw_path) raw_path = re.sub(r"[^a-zA-Z0-9._-]+", "-", raw_path) raw_path = re.sub(r"-{2,}", "-", raw_path).strip("-._") safe_path = _normalize_project_slug(raw_path) logger.info( "Payload reçu sur /api/v1/projects/create : %s", json.dumps( { "name": safe_name, "path": safe_path, "namespace_path": safe_namespace_path, "description": safe_description, "visibility": safe_visibility, "initialize_with_readme": safe_initialize_with_readme, }, ensure_ascii=False, ), ) namespace = _get_namespace_by_path(safe_namespace_path) namespace_id = namespace.get("id") if not isinstance(namespace_id, int): raise HTTPException(status_code=502, detail="Réponse GitLab invalide : namespace_id introuvable.") create_payload: dict[str, Any] = { "name": safe_name, "path": safe_path, "namespace_id": namespace_id, "description": safe_description, "visibility": safe_visibility, "initialize_with_readme": safe_initialize_with_readme, } logger.info( "Payload envoyé à GitLab pour création projet : %s", json.dumps(create_payload, ensure_ascii=False), ) project = _gitlab_request("POST", "/projects", payload=create_payload) if not isinstance(project, dict): raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la création du projet.") created_path_with_namespace = str(project.get("path_with_namespace", "")).strip().lower() expected_prefix = f"{safe_namespace_path.lower()}/" if not created_path_with_namespace.startswith(expected_prefix): project_id = project.get("id") if isinstance(project_id, int): try: _delete_project(project_id) logger.warning( "Projet supprimé car créé dans un mauvais namespace : attendu=%s obtenu=%s project_id=%s", safe_namespace_path, created_path_with_namespace, project_id, ) except Exception: logger.exception( "Impossible de supprimer le projet créé dans le mauvais namespace : project_id=%s", project_id, ) raise HTTPException( status_code=409, detail=( f"Projet créé dans un namespace inattendu : " f"attendu '{safe_namespace_path}', obtenu '{created_path_with_namespace}'." ), ) logger.info( "Projet GitLab créé avec succès : project_id=%s path=%s", project.get("id"), project.get("path_with_namespace"), ) return ProjectCreateResponse( success=True, project=ProjectCreatePayload( id=int(project["id"]), name=str(project["name"]), path_with_namespace=str(project["path_with_namespace"]), web_url=str(project["web_url"]), ), ) except HTTPException: raise except KeyError as exc: logger.exception("Champ GitLab manquant dans create_project : %s", str(exc)) raise HTTPException(status_code=502, detail="Réponse GitLab incomplète lors de la création du projet.") from exc except Exception as exc: logger.exception("Erreur inattendue dans create_project : %s", str(exc)) raise HTTPException(status_code=500, detail="Erreur interne lors de la création du projet.") from exc @router.post( "/assistant-tasks", operation_id="assistant_project_tasks", summary="Assistant tâches projet GitLab", description="Résout un projet, demande clarification si besoin, puis retourne les tâches.", response_model=ProjectAssistantTasksResponse, ) def assistant_project_tasks( payload: ProjectAssistantTasksRequest, ) -> ProjectAssistantTasksResponse: try: logger.info( "Payload reçu sur /api/v1/projects/assistant-tasks : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) if payload.confirm: if not payload.resolved_project_path.strip(): raise HTTPException( status_code=400, detail="Le champ resolved_project_path est obligatoire quand confirm=true.", ) project = _get_project_by_path(_normalize_project_path(payload.resolved_project_path)) project_info, open_titles, closed_titles = _get_project_tasks_data(project) return ProjectAssistantTasksResponse( success=True, status="resolved", message=f"Tâches récupérées pour {project_info.path_with_namespace}", project=project_info, open_issues=open_titles, closed_issues=closed_titles, candidate_projects=[], ) status_resolve, project, candidate_projects = _resolve_project_candidates( payload.project_hint, payload.root_hint, ) if status_resolve == "not_found": return ProjectAssistantTasksResponse( success=True, status="not_found", message=f"Je ne trouve pas de projet correspondant à '{payload.project_hint}'.", project=None, open_issues=[], closed_issues=[], candidate_projects=[], ) if status_resolve == "ambiguous": return ProjectAssistantTasksResponse( success=True, status="clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, open_issues=[], closed_issues=[], candidate_projects=candidate_projects, ) assert project is not None project_info, open_titles, closed_titles = _get_project_tasks_data(project) return ProjectAssistantTasksResponse( success=True, status="resolved", message=f"Tâches récupérées pour {project_info.path_with_namespace}", project=project_info, open_issues=open_titles, closed_issues=closed_titles, candidate_projects=candidate_projects, ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_project_tasks : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la récupération assistée des tâches.", ) from exc @router.post( "/assistant-progress", operation_id="assistant_project_progress", summary="Assistant progression projet GitLab", description="Résout un projet, demande clarification si besoin, puis retourne la progression.", response_model=ProjectAssistantProgressResponse, ) def assistant_project_progress( payload: ProjectAssistantProgressRequest, ) -> ProjectAssistantProgressResponse: try: logger.info( "Payload reçu sur /api/v1/projects/assistant-progress : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) if payload.confirm: if not payload.resolved_project_path.strip(): raise HTTPException( status_code=400, detail="Le champ resolved_project_path est obligatoire quand confirm=true.", ) project = _get_project_by_path(_normalize_project_path(payload.resolved_project_path)) project_info, total, closed, progress_percent = _get_project_progress_data(project) return ProjectAssistantProgressResponse( success=True, status="resolved", message=f"Progression calculée pour {project_info.path_with_namespace}", project=project_info, total=total, closed=closed, progress_percent=progress_percent, candidate_projects=[], ) status_resolve, project, candidate_projects = _resolve_project_candidates( payload.project_hint, payload.root_hint, ) if status_resolve == "not_found": return ProjectAssistantProgressResponse( success=True, status="not_found", message=f"Je ne trouve pas de projet correspondant à '{payload.project_hint}'.", project=None, total=0, closed=0, progress_percent=0, candidate_projects=[], ) if status_resolve == "ambiguous": return ProjectAssistantProgressResponse( success=True, status="clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, total=0, closed=0, progress_percent=0, candidate_projects=candidate_projects, ) assert project is not None project_info, total, closed, progress_percent = _get_project_progress_data(project) return ProjectAssistantProgressResponse( success=True, status="resolved", message=f"Progression calculée pour {project_info.path_with_namespace}", project=project_info, total=total, closed=closed, progress_percent=progress_percent, candidate_projects=candidate_projects, ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_project_progress : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors du calcul assisté de progression.", ) from exc