# -*- coding: utf-8 -*- """ Nom du fichier : issues.py Chemin : /var/www/html/gitlab-bridge/app/api/v1/issues.py Description : Endpoint API de création d'issue GitLab compatible OpenWebUI. Cette version utilise un payload simple sans bloc "auth" afin de permettre l'import OpenAPI et l'appel réel depuis OpenWebUI. L'appel à GitLab est effectué via le token technique du bridge. Options éventuelles : - POST /api/v1/issues/create Exemples d'utilisation : - Appel OpenWebUI via serveur d'outils OpenAPI - Appel curl : curl -X POST "https://gitlab-bridge.tarbouriech.tech/api/v1/issues/create" \ -H "Content-Type: application/json" \ -d '{ "project_path": "si/intelligence-artificielle/serveur-ia-interne", "title": "Test OpenWebUI", "description": "Issue de test", "labels": ["test", "ia"] }' Prérequis : - Variables d'environnement : GITLAB_BASE_URL GITLAB_API_TOKEN GITLAB_TIMEOUT (optionnel) - Schémas : app.schemas.issues.IssueCreateResponse app.schemas.issues.IssuePayload Auteur : Sylvain SCATTOLINI Date de création : 2026-03-25 Date de modification : 2026-03-25 Version : 1.3 """ from __future__ import annotations import logging import re 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.utils.helpers import normalize_text as _normalize_text, compute_similarity_score as _score_text from app.schemas.issues import ( BatchIssueResult, BulkCloseResult, IssueAssistantBatchCreateRequest, IssueAssistantBatchCreateResponse, IssueAssistantBulkCloseRequest, IssueAssistantBulkCloseResponse, IssueAssistantCommentRequest, IssueAssistantCommentResponse, IssueAssistantCreateRequest, IssueAssistantListRequest, IssueAssistantListResponse, IssueAssistantSetDueDateRequest, IssueAssistantSetDueDateResponse, IssueAssistantUpdateRequest, IssueAssistantUpdateResponse, IssueAssistantAssignUserRequest, IssueAssistantAssignUserResponse, IssueCreateResponse, IssueListItem, IssuePayload, IssueAssistantCloseRequest, IssueAssistantCloseResponse, IssueCandidateItem, IssueAssistantReopenRequest, IssueAssistantReopenResponse, IssueAssistantCreateResponse, IssueDuplicateItem, IssueAssistantAddLabelRequest, IssueAssistantAddLabelResponse, IssueAssistantListLabelsRequest, IssueAssistantListLabelsResponse, LabelItem, BulkIssueResultItem, IssueAssistantAddLabelBulkRequest, IssueAssistantAddLabelBulkResponse, IssueAssistantAddLabelAllRequest, IssueAssistantAddLabelAllResponse, UserCandidateItem, ) from app.schemas.milestones import MilestoneCandidateItem, MilestoneItem from fastapi import Body router = APIRouter(prefix="/api/v1/issues", tags=["issues"]) logger = logging.getLogger(__name__) STOP_WORDS = { "a", "au", "aux", "ce", "ces", "dans", "de", "des", "du", "en", "et", "la", "le", "les", "l", "mettre", "mise", "place", "pour", "sur", "un", "une", "ajouter", "creer", "creation", "configurer", "configuration", "faire", "implementer", "implementation", } _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 def _get_project_by_path(project_path: str) -> dict[str, Any]: """ Résout un projet GitLab à partir de son path_with_namespace. """ encoded_project_path = quote(project_path, safe="") return _gitlab_request( "GET", f"/projects/{encoded_project_path}", ) 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 s = max( _score_text(project_hint, str(project.get("name", ""))), _score_text(project_hint, path), ) if s < 40: continue candidates.append((s, project)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates def _build_candidate_projects(candidates: list[tuple[int, dict[str, Any]]]) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] for score, project in candidates[:5]: results.append( { "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 _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=404, detail="Projet introuvable") return project def _list_project_issues(project_id: int, state: str, limit: int) -> list[IssueListItem]: query_state = None if state == "all" else state issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": query_state, "per_page": min(limit, 100), "page": 1, "order_by": "created_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] results: list[IssueListItem] = [] for issue in issues: if not isinstance(issue, dict): continue results.append( IssueListItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), ) ) return results def _tokenize_meaningful_words(value: str) -> set[str]: normalized = _normalize_text(value) tokens = re.findall(r"[a-z0-9]+", normalized) cleaned_tokens: set[str] = set() for token in tokens: if token in STOP_WORDS: continue # singularisation très simple if len(token) > 4 and token.endswith("s"): token = token[:-1] cleaned_tokens.add(token) return cleaned_tokens def _score_issue_duplicate(title_hint: str, existing_title: str) -> int: normalized_hint = _normalize_text(title_hint) normalized_title = _normalize_text(existing_title) phrase_score = _score_text(title_hint, existing_title) if normalized_hint in normalized_title or normalized_title in normalized_hint: phrase_score = max(phrase_score, 95) hint_words = _tokenize_meaningful_words(title_hint) title_words = _tokenize_meaningful_words(existing_title) if not hint_words or not title_words: return phrase_score common_words = hint_words & title_words union_words = hint_words | title_words overlap_score = int((len(common_words) / len(union_words)) * 100) if union_words else 0 # bonus si un mot métier fort est partagé if len(common_words) >= 1: overlap_score = max(overlap_score, 60) if len(common_words) >= 2: overlap_score = max(overlap_score, 85) return max(phrase_score, overlap_score) def _search_issues_in_project(project_id: int, issue_hint: str, limit: int) -> list[tuple[int, dict[str, Any]]]: issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": "opened", "per_page": min(limit, 100), "page": 1, "order_by": "updated_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] candidates: list[tuple[int, dict[str, Any]]] = [] for issue in issues: if not isinstance(issue, dict): continue title = str(issue.get("title", "")) score = _score_text(issue_hint, title) if _normalize_text(issue_hint) in _normalize_text(title): score = max(score, 95) if score < 40: continue candidates.append((score, issue)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates[:limit] def _build_issue_candidates(candidates: list[tuple[int, dict[str, Any]]]) -> list[IssueCandidateItem]: results: list[IssueCandidateItem] = [] for score, issue in candidates[:5]: results.append( IssueCandidateItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), score=score, ) ) return results def _close_issue(project_id: int, issue_iid: int) -> IssueListItem: issue = _gitlab_request( "PUT", f"/projects/{project_id}/issues/{issue_iid}", payload={"state_event": "close"}, ) if not isinstance(issue, dict): raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la clôture de l'issue.") return IssueListItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), ) def _search_closed_issues_in_project(project_id: int, issue_hint: str, limit: int) -> list[tuple[int, dict[str, Any]]]: issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": "closed", "per_page": min(limit, 100), "page": 1, "order_by": "updated_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] candidates: list[tuple[int, dict[str, Any]]] = [] for issue in issues: if not isinstance(issue, dict): continue title = str(issue.get("title", "")) score = _score_text(issue_hint, title) if _normalize_text(issue_hint) in _normalize_text(title): score = max(score, 95) if score < 40: continue candidates.append((score, issue)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates[:limit] def _reopen_issue(project_id: int, issue_iid: int) -> IssueListItem: issue = _gitlab_request( "PUT", f"/projects/{project_id}/issues/{issue_iid}", payload={"state_event": "reopen"}, ) if not isinstance(issue, dict): raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la réouverture de l'issue.") return IssueListItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), ) def _search_duplicate_issues_in_project( project_id: int, title_hint: str, limit: int, ) -> list[tuple[int, dict[str, Any]]]: issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": "all", "per_page": min(limit, 100), "page": 1, "order_by": "updated_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] candidates: list[tuple[int, dict[str, Any]]] = [] for issue in issues: if not isinstance(issue, dict): continue title = str(issue.get("title", "")).strip() if not title: continue score = _score_issue_duplicate(title_hint, title) if score < 65: continue candidates.append((score, issue)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates[:limit] def _build_duplicate_issues( candidates: list[tuple[int, dict[str, Any]]], ) -> list[IssueDuplicateItem]: results: list[IssueDuplicateItem] = [] for score, issue in candidates[:5]: results.append( IssueDuplicateItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), score=score, ) ) return results def _list_project_labels(project_id: int, limit: int) -> list[LabelItem]: labels = _gitlab_request( "GET", f"/projects/{project_id}/labels", query_params={ "per_page": min(limit, 100), "page": 1, }, ) if not isinstance(labels, list): return [] results: list[LabelItem] = [] for label in labels: if not isinstance(label, dict): continue results.append( LabelItem( id=int(label["id"]) if label.get("id") is not None else None, name=str(label.get("name", "")), color=str(label.get("color")) if label.get("color") is not None else None, description=str(label.get("description")) if label.get("description") is not None else None, text_color=str(label.get("text_color")) if label.get("text_color") is not None else None, ) ) results.sort(key=lambda item: item.name.lower()) return results def _get_issue_by_iid(project_id: int, issue_iid: int) -> dict[str, Any]: issue = _gitlab_request( "GET", f"/projects/{project_id}/issues/{issue_iid}", ) if not isinstance(issue, dict) or "id" not in issue: raise HTTPException(status_code=404, detail="Issue introuvable") return issue def _ensure_project_label(project_id: int, label_name: str, create_if_missing: bool) -> str: existing_labels = _list_project_labels(project_id, 100) for label in existing_labels: if _normalize_text(label.name) == _normalize_text(label_name): return label.name if not create_if_missing: raise HTTPException( status_code=404, detail=f"Le label '{label_name}' est introuvable dans le projet.", ) created_label = _gitlab_request( "POST", f"/projects/{project_id}/labels", payload={ "name": label_name.strip(), "color": "#428BCA", }, ) if not isinstance(created_label, dict): raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la création du label.") return str(created_label.get("name", label_name.strip())) def _search_open_issues_in_project(project_id: int, issue_hint: str, limit: int) -> list[tuple[int, dict[str, Any]]]: issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": "opened", "per_page": min(limit, 100), "page": 1, "order_by": "updated_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] candidates: list[tuple[int, dict[str, Any]]] = [] for issue in issues: if not isinstance(issue, dict): continue title = str(issue.get("title", "")) score = _score_text(issue_hint, title) if _normalize_text(issue_hint) in _normalize_text(title): score = max(score, 95) if score < 40: continue candidates.append((score, issue)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates[:limit] def _update_issue_labels(project_id: int, issue_iid: int, labels: list[str]) -> IssueListItem: issue = _gitlab_request( "PUT", f"/projects/{project_id}/issues/{issue_iid}", payload={ "labels": ",".join(labels), }, ) if not isinstance(issue, dict): raise HTTPException(status_code=502, detail="Réponse GitLab invalide lors de la mise à jour des labels.") return IssueListItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), ) def _merge_issue_labels(issue_labels: list[Any], new_label: str) -> list[str]: merged: list[str] = [] for label in issue_labels: label_str = str(label).strip() if label_str and label_str not in merged: merged.append(label_str) if new_label not in merged: merged.append(new_label) return merged def _list_raw_project_issues(project_id: int, state: str, limit: int) -> list[dict[str, Any]]: query_state = None if state == "all" else state issues = _gitlab_request( "GET", f"/projects/{project_id}/issues", query_params={ "state": query_state, "per_page": min(limit, 100), "page": 1, "order_by": "created_at", "sort": "desc", }, ) if not isinstance(issues, list): return [] return [issue for issue in issues if isinstance(issue, dict)] def _issue_to_list_item(issue: dict[str, Any]) -> IssueListItem: return IssueListItem( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue.get("title", "")), state=str(issue.get("state", "")), web_url=str(issue.get("web_url", "")), ) # ----------------------------------------------- # Helpers milestone # ----------------------------------------------- def _milestone_to_item(milestone: dict[str, Any]) -> MilestoneItem: return MilestoneItem( id=int(milestone["id"]), iid=int(milestone.get("iid")) if milestone.get("iid") is not None else None, title=str(milestone.get("title", "")), description=str(milestone.get("description", "")) if milestone.get("description") else None, state=str(milestone.get("state", "")), due_date=str(milestone.get("due_date")) if milestone.get("due_date") else None, web_url=str(milestone.get("web_url")) if milestone.get("web_url") else None, ) def _search_milestones_in_project( project_id: int, milestone_hint: str, ) -> list[tuple[int, dict[str, Any]]]: """Recherche floue d'une milestone dans un projet (toutes les milestones actives + closed).""" milestones = _gitlab_request( "GET", f"/projects/{project_id}/milestones", query_params={"per_page": 100, "page": 1}, ) if not isinstance(milestones, list): return [] candidates: list[tuple[int, dict[str, Any]]] = [] normalized_hint = _normalize_text(milestone_hint) for ms in milestones: if not isinstance(ms, dict): continue title = str(ms.get("title", "")) score = _score_text(milestone_hint, title) if normalized_hint in _normalize_text(title): score = max(score, 95) if score < 40: continue candidates.append((score, ms)) candidates.sort(reverse=True, key=lambda item: item[0]) return candidates def _build_milestone_candidates( candidates: list[tuple[int, dict[str, Any]]], ) -> list[MilestoneCandidateItem]: return [ MilestoneCandidateItem( id=int(ms["id"]), iid=int(ms.get("iid")) if ms.get("iid") is not None else None, title=str(ms.get("title", "")), state=str(ms.get("state", "")), due_date=str(ms.get("due_date")) if ms.get("due_date") else None, web_url=str(ms.get("web_url")) if ms.get("web_url") else None, score=score, ) for score, ms in candidates[:5] ] def _resolve_milestone( project_id: int, milestone_hint: str, ) -> tuple[str, dict[str, Any] | None, list[MilestoneCandidateItem]]: """ Résout une milestone par hint flou. Retourne (status, milestone_dict_or_None, candidates). Status : "found" | "not_found" | "ambiguous" """ if not milestone_hint.strip(): return "found", None, [] candidates = _search_milestones_in_project(project_id, milestone_hint) if not candidates: return "not_found", None, [] if len(candidates) == 1: return "found", candidates[0][1], [] # Clair si le meilleur dépasse le second de +10 points if candidates[0][0] >= candidates[1][0] + 10: return "found", candidates[0][1], _build_milestone_candidates(candidates) return "ambiguous", None, _build_milestone_candidates(candidates) # ----------------------------------------------- # Endpoints API # ----------------------------------------------- @router.post( "/assistant-list", summary="Assistant liste des issues", response_model=IssueAssistantListResponse, ) def assistant_list_issues( payload: IssueAssistantListRequest, ) -> IssueAssistantListResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-list : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_state = payload.state.strip().lower() safe_limit = int(payload.limit) 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(payload.resolved_project_path.strip()) issues = _list_project_issues(int(project["id"]), safe_state, safe_limit) return IssueAssistantListResponse( success=True, status="resolved", message=f"Issues récupérées pour {project['path_with_namespace']}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issues=issues, candidate_projects=[], ) candidates = _search_best_project(payload.project_hint, payload.root_hint) if not candidates: return IssueAssistantListResponse( success=True, status="not_found", message=f"Je ne trouve pas de projet correspondant à '{payload.project_hint}'.", project=None, issues=[], candidate_projects=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantListResponse( success=True, status="clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, issues=[], candidate_projects=_build_candidate_projects(candidates), ) project = candidates[0][1] issues = _list_project_issues(int(project["id"]), safe_state, safe_limit) return IssueAssistantListResponse( success=True, status="resolved", message=f"Issues récupérées pour {project['path_with_namespace']}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issues=issues, candidate_projects=_build_candidate_projects(candidates[:1]), ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_list_issues : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la récupération assistée des issues.", ) from exc @router.post( "/assistant-create", summary="Assistant création issue", response_model=IssueAssistantCreateResponse, ) def assistant_create_issue( payload: IssueAssistantCreateRequest, ) -> IssueAssistantCreateResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-create : %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(payload.resolved_project_path.strip()) project_id = int(project["id"]) create_payload: dict[str, Any] = { "title": payload.title.strip(), "description": payload.description.strip(), } if payload.labels: create_payload["labels"] = ",".join(l.strip() for l in payload.labels if l.strip()) if payload.resolved_milestone_id > 0: create_payload["milestone_id"] = payload.resolved_milestone_id issue = _gitlab_request("POST", f"/projects/{project_id}/issues", payload=create_payload) created_issue = IssuePayload( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue["title"]), web_url=str(issue["web_url"]), ) milestone_note = ( f" assignée à la milestone #{payload.resolved_milestone_id}" if payload.resolved_milestone_id > 0 else "" ) return IssueAssistantCreateResponse( success=True, status="created", message=f"Issue créée{milestone_note} : {created_issue.title}", issue=created_issue, candidate_projects=[], duplicate_issues=[], ) candidates = _search_best_project(payload.project_hint, payload.root_hint) if not candidates: return IssueAssistantCreateResponse( success=True, status="project_not_found", message=f"Je ne trouve pas de projet correspondant à ‘{payload.project_hint}’.", issue=None, candidate_projects=[], duplicate_issues=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantCreateResponse( success=True, status="project_clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", issue=None, candidate_projects=_build_candidate_projects(candidates), duplicate_issues=[], ) project = candidates[0][1] project_id = int(project["id"]) # Résolution de la milestone si demandée resolved_milestone_id = payload.resolved_milestone_id if resolved_milestone_id == 0 and payload.milestone_hint.strip(): ms_status, ms_data, ms_candidates = _resolve_milestone(project_id, payload.milestone_hint) if ms_status == "ambiguous": return IssueAssistantCreateResponse( success=True, status="milestone_clarification_needed", message=( f"J’ai trouvé plusieurs milestones correspondant à ‘{payload.milestone_hint}’. " "Merci de préciser." ), issue=None, candidate_projects=[], duplicate_issues=[], ) if ms_status == "found" and ms_data: resolved_milestone_id = int(ms_data["id"]) elif ms_status == "not_found": logger.warning("Milestone introuvable pour hint ‘%s’", payload.milestone_hint) duplicate_candidates = _search_duplicate_issues_in_project(project_id, payload.title, 10) if duplicate_candidates and duplicate_candidates[0][0] >= 80: return IssueAssistantCreateResponse( success=True, status="duplicate_warning", message="Une ou plusieurs issues similaires existent déjà. Merci de confirmer la création.", issue=None, candidate_projects=[], duplicate_issues=_build_duplicate_issues(duplicate_candidates), ) create_payload = { "title": payload.title.strip(), "description": payload.description.strip(), } if payload.labels: create_payload["labels"] = ",".join(l.strip() for l in payload.labels if l.strip()) if resolved_milestone_id > 0: create_payload["milestone_id"] = resolved_milestone_id issue = _gitlab_request("POST", f"/projects/{project_id}/issues", payload=create_payload) created_issue = IssuePayload( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue["title"]), web_url=str(issue["web_url"]), ) milestone_note = f" assignée à la milestone #{resolved_milestone_id}" if resolved_milestone_id > 0 else "" return IssueAssistantCreateResponse( success=True, status="created", message=f"Issue créée{milestone_note} : {created_issue.title}", issue=created_issue, candidate_projects=_build_candidate_projects(candidates[:1]), duplicate_issues=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_create_issue : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la création assistée de l’issue.", ) from exc @router.post( "/create", operation_id="create_issue", summary="Créer une issue GitLab", description="Crée une issue GitLab à partir d'un payload simple compatible OpenWebUI.", response_model=IssueCreateResponse, ) def create_issue( project_path: Annotated[str, Body(min_length=1, max_length=255)], title: Annotated[str, Body(min_length=1, max_length=255)], description: Annotated[str, Body(min_length=1, max_length=20000)], labels: Annotated[list[str], Body(max_items=50)] = [], ) -> IssueCreateResponse: """ Crée une issue dans GitLab via le token technique du bridge. """ try: safe_payload = { "project_path": project_path.strip(), "title": title.strip(), "description": description.strip(), "labels": labels, } logger.info( "Payload reçu sur /api/v1/issues/create : %s", json.dumps(safe_payload, ensure_ascii=False), ) project = _get_project_by_path(safe_payload["project_path"]) create_payload: dict[str, Any] = { "title": safe_payload["title"], "description": safe_payload["description"], } if safe_payload["labels"]: normalized_labels = [ label.strip() for label in safe_payload["labels"] if isinstance(label, str) and label.strip() ] if normalized_labels: create_payload["labels"] = ",".join(normalized_labels) issue = _gitlab_request( "POST", f"/projects/{project['id']}/issues", payload=create_payload, ) logger.info( "Issue GitLab créée avec succès : project_id=%s issue_iid=%s", project.get("id"), issue.get("iid"), ) return IssueCreateResponse( success=True, issue=IssuePayload( id=int(issue["id"]), iid=int(issue["iid"]), title=str(issue["title"]), web_url=str(issue["web_url"]), ), ) except HTTPException: raise except KeyError as exc: logger.exception("Champ GitLab manquant dans create_issue : %s", str(exc)) raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="Réponse GitLab incomplète lors de la création d'issue.", ) from exc except Exception as exc: logger.exception("Erreur inattendue dans create_issue : %s", str(exc)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Erreur interne lors de la création d'issue.", ) from exc @router.post( "/assistant-close", summary="Assistant clôture d'issue", response_model=IssueAssistantCloseResponse, ) def assistant_close_issue( payload: IssueAssistantCloseRequest, ) -> IssueAssistantCloseResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-close : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) 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.", ) if payload.resolved_issue_iid <= 0: raise HTTPException( status_code=400, detail="Le champ resolved_issue_iid est obligatoire quand confirm=true.", ) project = _get_project_by_path(payload.resolved_project_path.strip()) closed_issue = _close_issue(int(project["id"]), int(payload.resolved_issue_iid)) return IssueAssistantCloseResponse( success=True, status="closed", message=f"Issue clôturée : {closed_issue.title}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=closed_issue, candidate_projects=[], candidate_issues=[], ) candidates = _search_best_project(payload.project_hint or payload.issue_hint, payload.root_hint) if not candidates: return IssueAssistantCloseResponse( success=True, status="project_not_found", message="Je ne trouve pas le projet correspondant.", project=None, issue=None, candidate_projects=[], candidate_issues=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantCloseResponse( success=True, status="project_clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, issue=None, candidate_projects=_build_candidate_projects(candidates), candidate_issues=[], ) project = candidates[0][1] issue_candidates = _search_issues_in_project(int(project["id"]), payload.issue_hint, safe_limit) if not issue_candidates: return IssueAssistantCloseResponse( success=True, status="issue_not_found", message=f"Je ne trouve pas d'issue correspondant à '{payload.issue_hint}'.", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=None, candidate_projects=[], candidate_issues=[], ) if len(issue_candidates) > 1 and issue_candidates[1][0] >= issue_candidates[0][0] - 10: return IssueAssistantCloseResponse( success=True, status="issue_clarification_needed", message="J’ai trouvé plusieurs issues possibles. Merci de préciser laquelle clôturer.", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=None, candidate_projects=[], candidate_issues=_build_issue_candidates(issue_candidates), ) issue_iid = int(issue_candidates[0][1]["iid"]) closed_issue = _close_issue(int(project["id"]), issue_iid) return IssueAssistantCloseResponse( success=True, status="closed", message=f"Issue clôturée : {closed_issue.title}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=closed_issue, candidate_projects=[], candidate_issues=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_close_issue : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la clôture assistée de l'issue.", ) from exc @router.post( "/assistant-reopen", summary="Assistant réouverture d'issue", response_model=IssueAssistantReopenResponse, ) def assistant_reopen_issue( payload: IssueAssistantReopenRequest, ) -> IssueAssistantReopenResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-reopen : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) 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.", ) if payload.resolved_issue_iid <= 0: raise HTTPException( status_code=400, detail="Le champ resolved_issue_iid est obligatoire quand confirm=true.", ) project = _get_project_by_path(payload.resolved_project_path.strip()) reopened_issue = _reopen_issue(int(project["id"]), int(payload.resolved_issue_iid)) return IssueAssistantReopenResponse( success=True, status="reopened", message=f"Issue réouverte : {reopened_issue.title}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=reopened_issue, candidate_projects=[], candidate_issues=[], ) candidates = _search_best_project(payload.project_hint or payload.issue_hint, payload.root_hint) if not candidates: return IssueAssistantReopenResponse( success=True, status="project_not_found", message="Je ne trouve pas le projet correspondant.", project=None, issue=None, candidate_projects=[], candidate_issues=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantReopenResponse( success=True, status="project_clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, issue=None, candidate_projects=_build_candidate_projects(candidates), candidate_issues=[], ) project = candidates[0][1] issue_candidates = _search_closed_issues_in_project(int(project["id"]), payload.issue_hint, safe_limit) if not issue_candidates: return IssueAssistantReopenResponse( success=True, status="issue_not_found", message=f"Je ne trouve pas d'issue fermée correspondant à '{payload.issue_hint}'.", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=None, candidate_projects=[], candidate_issues=[], ) if len(issue_candidates) > 1 and issue_candidates[1][0] >= issue_candidates[0][0] - 10: return IssueAssistantReopenResponse( success=True, status="issue_clarification_needed", message="J’ai trouvé plusieurs issues fermées possibles. Merci de préciser laquelle réouvrir.", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=None, candidate_projects=[], candidate_issues=_build_issue_candidates(issue_candidates), ) issue_iid = int(issue_candidates[0][1]["iid"]) reopened_issue = _reopen_issue(int(project["id"]), issue_iid) return IssueAssistantReopenResponse( success=True, status="reopened", message=f"Issue réouverte : {reopened_issue.title}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=reopened_issue, candidate_projects=[], candidate_issues=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_reopen_issue : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la réouverture assistée de l'issue.", ) from exc @router.post( "/assistant-list-labels", summary="Assistant liste des labels", response_model=IssueAssistantListLabelsResponse, ) def assistant_list_labels( payload: IssueAssistantListLabelsRequest, ) -> IssueAssistantListLabelsResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-list-labels : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) 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(payload.resolved_project_path.strip()) labels = _list_project_labels(int(project["id"]), safe_limit) return IssueAssistantListLabelsResponse( success=True, status="resolved", message=f"Labels récupérés pour {project['path_with_namespace']}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, labels=labels, candidate_projects=[], ) candidates = _search_best_project(payload.project_hint, payload.root_hint) if not candidates: return IssueAssistantListLabelsResponse( success=True, status="not_found", message=f"Je ne trouve pas de projet correspondant à '{payload.project_hint}'.", project=None, labels=[], candidate_projects=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantListLabelsResponse( success=True, status="clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, labels=[], candidate_projects=_build_candidate_projects(candidates), ) project = candidates[0][1] labels = _list_project_labels(int(project["id"]), safe_limit) return IssueAssistantListLabelsResponse( success=True, status="resolved", message=f"Labels récupérés pour {project['path_with_namespace']}", project={ "id": int(project["id"]), "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, labels=labels, candidate_projects=_build_candidate_projects(candidates[:1]), ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_list_labels : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la récupération assistée des labels.", ) from exc @router.post( "/assistant-add-label", summary="Assistant ajout de label à une issue", response_model=IssueAssistantAddLabelResponse, ) def assistant_add_label( payload: IssueAssistantAddLabelRequest, ) -> IssueAssistantAddLabelResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-add-label : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) safe_label_name = payload.label_name.strip() 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.", ) if payload.resolved_issue_iid <= 0: raise HTTPException( status_code=400, detail="Le champ resolved_issue_iid est obligatoire quand confirm=true.", ) project = _get_project_by_path(payload.resolved_project_path.strip()) project_id = int(project["id"]) label_name = _ensure_project_label( project_id, safe_label_name, payload.create_label_if_missing, ) issue_before = _get_issue_by_iid(project_id, int(payload.resolved_issue_iid)) existing_labels = [str(label) for label in issue_before.get("labels", [])] merged_labels = existing_labels[:] if label_name not in merged_labels: merged_labels.append(label_name) updated_issue = _update_issue_labels(project_id, int(payload.resolved_issue_iid), merged_labels) return IssueAssistantAddLabelResponse( success=True, status="updated", message=f"Label ajouté à l'issue : {updated_issue.title}", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=updated_issue, labels=merged_labels, candidate_projects=[], candidate_issues=[], ) candidates = _search_best_project(payload.project_hint or payload.issue_hint, payload.root_hint) if not candidates: return IssueAssistantAddLabelResponse( success=True, status="project_not_found", message="Je ne trouve pas le projet correspondant.", project=None, issue=None, labels=[], candidate_projects=[], candidate_issues=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantAddLabelResponse( success=True, status="project_clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, issue=None, labels=[], candidate_projects=_build_candidate_projects(candidates), candidate_issues=[], ) project = candidates[0][1] project_id = int(project["id"]) issue_candidates = _search_open_issues_in_project(project_id, payload.issue_hint, safe_limit) if not issue_candidates: return IssueAssistantAddLabelResponse( success=True, status="issue_not_found", message=f"Je ne trouve pas d'issue ouverte correspondant à '{payload.issue_hint}'.", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=None, labels=[], candidate_projects=[], candidate_issues=[], ) if len(issue_candidates) > 1 and issue_candidates[1][0] >= issue_candidates[0][0] - 10: return IssueAssistantAddLabelResponse( success=True, status="issue_clarification_needed", message="J’ai trouvé plusieurs issues possibles. Merci de préciser laquelle modifier.", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=None, labels=[], candidate_projects=[], candidate_issues=_build_issue_candidates(issue_candidates), ) issue_iid = int(issue_candidates[0][1]["iid"]) label_name = _ensure_project_label(project_id, safe_label_name, payload.create_label_if_missing) issue_before = _get_issue_by_iid(project_id, issue_iid) existing_labels = [str(label) for label in issue_before.get("labels", [])] merged_labels = existing_labels[:] if label_name not in merged_labels: merged_labels.append(label_name) updated_issue = _update_issue_labels(project_id, issue_iid, merged_labels) return IssueAssistantAddLabelResponse( success=True, status="updated", message=f"Label ajouté à l'issue : {updated_issue.title}", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, issue=updated_issue, labels=merged_labels, candidate_projects=[], candidate_issues=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_add_label : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de l'ajout assisté de label.", ) from exc @router.post( "/assistant-add-label-bulk", summary="Assistant ajout de label à plusieurs issues", response_model=IssueAssistantAddLabelBulkResponse, ) def assistant_add_label_bulk( payload: IssueAssistantAddLabelBulkRequest, ) -> IssueAssistantAddLabelBulkResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-add-label-bulk : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) safe_label_name = payload.label_name.strip() clean_issue_hints = [ issue_hint.strip() for issue_hint in payload.issue_hints if isinstance(issue_hint, str) and issue_hint.strip() ] if not clean_issue_hints: raise HTTPException( status_code=400, detail="La liste issue_hints est vide.", ) 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(payload.resolved_project_path.strip()) project_id = int(project["id"]) label_name = _ensure_project_label( project_id, safe_label_name, payload.create_label_if_missing, ) results: list[BulkIssueResultItem] = [] for issue_hint in clean_issue_hints: issue_candidates = _search_open_issues_in_project(project_id, issue_hint, safe_limit) if not issue_candidates: results.append( BulkIssueResultItem( issue_hint=issue_hint, status="issue_not_found", message=f"Issue introuvable pour '{issue_hint}'.", issue=None, labels=[], candidate_issues=[], ) ) continue if len(issue_candidates) > 1 and issue_candidates[1][0] >= issue_candidates[0][0] - 10: results.append( BulkIssueResultItem( issue_hint=issue_hint, status="issue_clarification_needed", message=f"Plusieurs issues possibles pour '{issue_hint}'.", issue=None, labels=[], candidate_issues=_build_issue_candidates(issue_candidates), ) ) continue issue_iid = int(issue_candidates[0][1]["iid"]) issue_before = _get_issue_by_iid(project_id, issue_iid) merged_labels = _merge_issue_labels(issue_before.get("labels", []), label_name) updated_issue = _update_issue_labels(project_id, issue_iid, merged_labels) results.append( BulkIssueResultItem( issue_hint=issue_hint, status="updated", message=f"Label ajouté à '{updated_issue.title}'.", issue=updated_issue, labels=merged_labels, candidate_issues=[], ) ) return IssueAssistantAddLabelBulkResponse( success=True, status="processed", message="Traitement des labels terminé.", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, label_name=label_name, results=results, candidate_projects=[], ) candidates = _search_best_project(payload.project_hint, payload.root_hint) if not candidates: return IssueAssistantAddLabelBulkResponse( success=True, status="project_not_found", message="Je ne trouve pas le projet correspondant.", project=None, label_name=safe_label_name, results=[], candidate_projects=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantAddLabelBulkResponse( success=True, status="project_clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, label_name=safe_label_name, results=[], candidate_projects=_build_candidate_projects(candidates), ) project = candidates[0][1] project_id = int(project["id"]) label_name = _ensure_project_label( project_id, safe_label_name, payload.create_label_if_missing, ) results: list[BulkIssueResultItem] = [] for issue_hint in clean_issue_hints: issue_candidates = _search_open_issues_in_project(project_id, issue_hint, safe_limit) if not issue_candidates: results.append( BulkIssueResultItem( issue_hint=issue_hint, status="issue_not_found", message=f"Issue introuvable pour '{issue_hint}'.", issue=None, labels=[], candidate_issues=[], ) ) continue if len(issue_candidates) > 1 and issue_candidates[1][0] >= issue_candidates[0][0] - 10: results.append( BulkIssueResultItem( issue_hint=issue_hint, status="issue_clarification_needed", message=f"Plusieurs issues possibles pour '{issue_hint}'.", issue=None, labels=[], candidate_issues=_build_issue_candidates(issue_candidates), ) ) continue issue_iid = int(issue_candidates[0][1]["iid"]) issue_before = _get_issue_by_iid(project_id, issue_iid) merged_labels = _merge_issue_labels(issue_before.get("labels", []), label_name) updated_issue = _update_issue_labels(project_id, issue_iid, merged_labels) results.append( BulkIssueResultItem( issue_hint=issue_hint, status="updated", message=f"Label ajouté à '{updated_issue.title}'.", issue=updated_issue, labels=merged_labels, candidate_issues=[], ) ) return IssueAssistantAddLabelBulkResponse( success=True, status="processed", message="Traitement des labels terminé.", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, label_name=label_name, results=results, candidate_projects=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_add_label_bulk : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de l'ajout assisté de label en masse.", ) from exc @router.post( "/assistant-add-label-all", summary="Assistant ajout de label à toutes les issues d'un projet", response_model=IssueAssistantAddLabelAllResponse, ) def assistant_add_label_all( payload: IssueAssistantAddLabelAllRequest, ) -> IssueAssistantAddLabelAllResponse: try: logger.info( "Payload reçu sur /api/v1/issues/assistant-add-label-all : %s", json.dumps(payload.model_dump(), ensure_ascii=False), ) safe_limit = int(payload.limit) safe_label_name = payload.label_name.strip() safe_state = payload.state.strip().lower() 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(payload.resolved_project_path.strip()) project_id = int(project["id"]) label_name = _ensure_project_label( project_id, safe_label_name, payload.create_label_if_missing, ) raw_issues = _list_raw_project_issues(project_id, safe_state, safe_limit) results: list[BulkIssueResultItem] = [] updated_count = 0 for issue in raw_issues: issue_iid = int(issue["iid"]) existing_labels = [str(label) for label in issue.get("labels", [])] merged_labels = _merge_issue_labels(existing_labels, label_name) if merged_labels == existing_labels: results.append( BulkIssueResultItem( issue_hint=str(issue.get("title", "")), status="unchanged", message=f"Label déjà présent sur '{issue.get('title', '')}'.", issue=_issue_to_list_item(issue), labels=merged_labels, candidate_issues=[], ) ) continue updated_issue = _update_issue_labels(project_id, issue_iid, merged_labels) updated_count += 1 results.append( BulkIssueResultItem( issue_hint=str(issue.get("title", "")), status="updated", message=f"Label ajouté à '{updated_issue.title}'.", issue=updated_issue, labels=merged_labels, candidate_issues=[], ) ) return IssueAssistantAddLabelAllResponse( success=True, status="processed", message="Ajout du label terminé sur les issues du projet.", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, label_name=label_name, total_issues=len(raw_issues), updated_count=updated_count, results=results, candidate_projects=[], ) candidates = _search_best_project(payload.project_hint, payload.root_hint) if not candidates: return IssueAssistantAddLabelAllResponse( success=True, status="project_not_found", message=f"Je ne trouve pas de projet correspondant à '{payload.project_hint}'.", project=None, label_name=safe_label_name, total_issues=0, updated_count=0, results=[], candidate_projects=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantAddLabelAllResponse( success=True, status="project_clarification_needed", message="J’ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", project=None, label_name=safe_label_name, total_issues=0, updated_count=0, results=[], candidate_projects=_build_candidate_projects(candidates), ) project = candidates[0][1] project_id = int(project["id"]) label_name = _ensure_project_label( project_id, safe_label_name, payload.create_label_if_missing, ) raw_issues = _list_raw_project_issues(project_id, safe_state, safe_limit) results: list[BulkIssueResultItem] = [] updated_count = 0 for issue in raw_issues: issue_iid = int(issue["iid"]) existing_labels = [str(label) for label in issue.get("labels", [])] merged_labels = _merge_issue_labels(existing_labels, label_name) if merged_labels == existing_labels: results.append( BulkIssueResultItem( issue_hint=str(issue.get("title", "")), status="unchanged", message=f"Label déjà présent sur '{issue.get('title', '')}'.", issue=_issue_to_list_item(issue), labels=merged_labels, candidate_issues=[], ) ) continue updated_issue = _update_issue_labels(project_id, issue_iid, merged_labels) updated_count += 1 results.append( BulkIssueResultItem( issue_hint=str(issue.get("title", "")), status="updated", message=f"Label ajouté à '{updated_issue.title}'.", issue=updated_issue, labels=merged_labels, candidate_issues=[], ) ) return IssueAssistantAddLabelAllResponse( success=True, status="processed", message="Ajout du label terminé sur les issues du projet.", project={ "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), }, label_name=label_name, total_issues=len(raw_issues), updated_count=updated_count, results=results, candidate_projects=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_add_label_all : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de l'ajout du label sur toutes les issues du projet.", ) from exc @router.post( "/assistant-batch-create", operation_id="assistant_batch_create_issues", summary="Créer plusieurs issues GitLab en une seule requête", description=( "Crée N issues d'un seul appel dans un projet GitLab, " "avec assignment optionnel à une milestone. " "Résout automatiquement le projet et la milestone par nom partiel. " "En cas d'ambiguïté, retourne les candidats pour clarification. " "Chaque issue est créée indépendamment : une erreur sur une issue n'interrompt pas le lot." ), response_model=IssueAssistantBatchCreateResponse, ) def assistant_batch_create_issues( payload: IssueAssistantBatchCreateRequest, ) -> IssueAssistantBatchCreateResponse: """ Endpoint batch pour Thalia / OpenWebUI. Flux recommandé pour l'IA : 1. Appel initial (confirm=False) → le bridge résout projet + milestone. Si tout est clair → status="ready", les resolved_* sont fournis. Rappeler avec confirm=True et les resolved_* obtenus. 2. Appel de confirmation (confirm=True, resolved_project_path + resolved_milestone_id) → toutes les issues sont créées. Jamais d'interruption partielle. Note : si le projet et la milestone sont non ambigus dès le premier appel, le bridge crée directement les issues et retourne status="completed". """ try: logger.info( "Payload reçu sur /api/v1/issues/assistant-batch-create : %s issues, project_hint=%r, milestone_hint=%r", len(payload.issues), payload.project_hint, payload.milestone_hint, ) # ---------------------------------------------------------------- # Phase confirm=True : créer toutes les issues # ---------------------------------------------------------------- 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(payload.resolved_project_path.strip()) project_id = int(project["id"]) # Récupère les infos de la milestone si fournie milestone_item: MilestoneItem | None = None if payload.resolved_milestone_id > 0: try: ms_data = _gitlab_request( "GET", f"/projects/{project_id}/milestones/{payload.resolved_milestone_id}", ) if isinstance(ms_data, dict): milestone_item = _milestone_to_item(ms_data) except HTTPException: logger.warning( "Milestone %d introuvable — les issues seront créées sans milestone.", payload.resolved_milestone_id, ) results: list[BatchIssueResult] = [] created_count = 0 error_count = 0 for issue_input in payload.issues: try: create_payload: dict[str, Any] = { "title": issue_input.title.strip(), "description": issue_input.description.strip(), } if issue_input.labels: clean_labels = [l.strip() for l in issue_input.labels if l.strip()] if clean_labels: create_payload["labels"] = ",".join(clean_labels) if payload.resolved_milestone_id > 0: create_payload["milestone_id"] = payload.resolved_milestone_id issue_data = _gitlab_request( "POST", f"/projects/{project_id}/issues", payload=create_payload, ) created_count += 1 results.append(BatchIssueResult( title=issue_input.title, status="created", message=f"Issue #{issue_data['iid']} créée.", issue=IssuePayload( id=int(issue_data["id"]), iid=int(issue_data["iid"]), title=str(issue_data["title"]), web_url=str(issue_data["web_url"]), ), milestone_assigned=payload.resolved_milestone_id > 0, )) except HTTPException as exc: error_count += 1 logger.warning( "Erreur création issue '%s' : %s", issue_input.title, exc.detail, ) results.append(BatchIssueResult( title=issue_input.title, status="error", message=f"Erreur : {exc.detail}", )) project_dict = { "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), } milestone_note = f" dans la milestone '{milestone_item.title}'" if milestone_item else "" return IssueAssistantBatchCreateResponse( success=True, status="completed", message=f"{created_count} issue(s) créée(s){milestone_note}, {error_count} erreur(s).", project=project_dict, milestone=milestone_item, results=results, created_count=created_count, error_count=error_count, candidate_projects=[], candidate_milestones=[], ) # ---------------------------------------------------------------- # Phase confirm=False : résoudre projet + milestone # ---------------------------------------------------------------- candidates = _search_best_project(payload.project_hint, payload.root_hint) if not candidates: return IssueAssistantBatchCreateResponse( success=True, status="project_not_found", message=f"Je ne trouve pas de projet correspondant à '{payload.project_hint}'.", results=[], candidate_projects=[], candidate_milestones=[], ) if len(candidates) > 1 and candidates[1][0] >= candidates[0][0] - 10: return IssueAssistantBatchCreateResponse( success=True, status="project_clarification_needed", message="J'ai trouvé plusieurs projets possibles. Merci de préciser lequel utiliser.", results=[], candidate_projects=_build_candidate_projects(candidates), candidate_milestones=[], ) project = candidates[0][1] project_id = int(project["id"]) project_dict = { "id": project_id, "name": str(project.get("name", "")), "path_with_namespace": str(project.get("path_with_namespace", "")), } # Résoudre la milestone si demandée resolved_milestone_id = 0 milestone_item = None candidate_milestones: list[MilestoneCandidateItem] = [] if payload.milestone_hint.strip(): ms_status, ms_data, candidate_milestones = _resolve_milestone( project_id, payload.milestone_hint ) if ms_status == "ambiguous": return IssueAssistantBatchCreateResponse( success=True, status="milestone_clarification_needed", message=( f"J'ai trouvé plusieurs milestones correspondant à '{payload.milestone_hint}'. " "Merci de préciser laquelle utiliser en fournissant resolved_milestone_id." ), project=project_dict, results=[], candidate_projects=[], candidate_milestones=candidate_milestones, ) if ms_status == "not_found": return IssueAssistantBatchCreateResponse( success=True, status="milestone_not_found", message=( f"Je ne trouve pas de milestone correspondant à '{payload.milestone_hint}' " f"dans le projet '{project_dict['path_with_namespace']}'." ), project=project_dict, results=[], candidate_projects=[], candidate_milestones=[], ) if ms_status == "found" and ms_data: resolved_milestone_id = int(ms_data["id"]) milestone_item = _milestone_to_item(ms_data) # Projet et milestone résolus sans ambiguïté → créer directement results = [] created_count = 0 error_count = 0 for issue_input in payload.issues: try: create_payload = { "title": issue_input.title.strip(), "description": issue_input.description.strip(), } if issue_input.labels: clean_labels = [l.strip() for l in issue_input.labels if l.strip()] if clean_labels: create_payload["labels"] = ",".join(clean_labels) if resolved_milestone_id > 0: create_payload["milestone_id"] = resolved_milestone_id issue_data = _gitlab_request( "POST", f"/projects/{project_id}/issues", payload=create_payload, ) created_count += 1 results.append(BatchIssueResult( title=issue_input.title, status="created", message=f"Issue #{issue_data['iid']} créée.", issue=IssuePayload( id=int(issue_data["id"]), iid=int(issue_data["iid"]), title=str(issue_data["title"]), web_url=str(issue_data["web_url"]), ), milestone_assigned=resolved_milestone_id > 0, )) except HTTPException as exc: error_count += 1 logger.warning("Erreur création issue '%s' : %s", issue_input.title, exc.detail) results.append(BatchIssueResult( title=issue_input.title, status="error", message=f"Erreur : {exc.detail}", )) milestone_note = f" dans la milestone '{milestone_item.title}'" if milestone_item else "" return IssueAssistantBatchCreateResponse( success=True, status="completed", message=f"{created_count} issue(s) créée(s){milestone_note}, {error_count} erreur(s).", project=project_dict, milestone=milestone_item, results=results, created_count=created_count, error_count=error_count, candidate_projects=[], candidate_milestones=[], ) except HTTPException: raise except Exception as exc: logger.exception("Erreur inattendue dans assistant_batch_create_issues : %s", str(exc)) raise HTTPException( status_code=500, detail="Erreur interne lors de la création en lot des issues.", ) from exc