Code/app/fiches/utils/tickets/core.py

# core.py

import csv
import json
import os

import requests
import streamlit as st

from config import DEPOT_FICHES, ENV, GITEA_TOKEN, GITEA_URL, ORGANISATION
from utils.translations import _


def gitea_request(method, url, **kwargs):
    """Execute une requete HTTP vers l'API Gitea avec authentification automatique.

    Args:
        method: Methode HTTP (get, post, patch, delete, etc.).
        url: URL complete de l'endpoint API Gitea.
        **kwargs: Arguments passes a requests.request (params, data, json, etc.).

    Returns:
        requests.Response | None: Objet Response si succes, None si erreur.
    """
    headers = kwargs.pop("headers", {})
    headers["Authorization"] = f"token {GITEA_TOKEN}"
    try:
        response = requests.request(method, url, headers=headers, timeout=10, **kwargs)
        response.raise_for_status()
        return response
    except requests.RequestException as e:
        st.error(f"{str(_('errors.gitea_error'))} ({method.upper()}): {e}")
        return None


def charger_fiches_et_labels():
    """Charge la correspondance entre fiches et labels depuis le fichier CSV.

    Lit le fichier assets/fiches_labels.csv et construit un dictionnaire associant
    chaque fiche a ses labels (operations et item).

    Returns:
        dict: Dictionnaire au format {nom_fiche: {"operations": [str], "item": str}}.
              Retourne un dict vide en cas d'erreur.
    """
    chemin_csv = os.path.join("assets", "fiches_labels.csv")
    dictionnaire_fiches = {}

    try:
        with open(chemin_csv, encoding="utf-8") as fichier_csv:
            lecteur = csv.DictReader(fichier_csv)
            for ligne in lecteur:
                fiche = ligne.get("Fiche")
                operations = ligne.get("Label opération")
                item = ligne.get("Label item")

                if fiche and operations and item:
                    dictionnaire_fiches[fiche.strip()] = {
                        "operations": [op.strip() for op in operations.split("/")],
                        "item": item.strip()
                    }
    except FileNotFoundError:
        st.error(f"❌ {str(_('errors.file_not_found'))} {chemin_csv} {str(_('errors.is_missing'))}")
    except Exception as e:
        st.error(f"❌ {str(_('errors.file_loading'))} {str(e)}")

    return dictionnaire_fiches


def rechercher_tickets_gitea(fiche_selectionnee):
    """Recherche les tickets Gitea ouverts associes a une fiche specifique.

    Filtre les issues ouvertes du depot DEPOT_FICHES par branche (ENV) et par
    labels correspondant a la fiche selectionnee.

    Args:
        fiche_selectionnee: Nom de la fiche pour laquelle chercher les tickets.

    Returns:
        list[dict]: Liste des issues Gitea correspondantes (format JSON Gitea).
    """
    params = {"state": "open"}
    url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/issues"

    reponse = gitea_request("get", url, params=params)
    if not reponse:
        return []

    try:
        issues = reponse.json()
    except Exception as e:
        st.error(f"{str(_('errors.json_decode'))} {e}")
        return []

    correspondances = charger_fiches_et_labels()
    cible = correspondances.get(fiche_selectionnee)
    if not cible:
        return []

    labels_cibles = set([cible["item"]])
    tickets_associes = []

    for issue in issues:
        if issue.get("ref") != f"refs/heads/{ENV}":
            continue
        issue_labels = set(label.get("name", "") for label in issue.get("labels", []))
        if labels_cibles.issubset(issue_labels):
            tickets_associes.append(issue)

    return tickets_associes


def get_labels_existants():
    """Recupere tous les labels existants dans le depot Gitea.

    Returns:
        dict: Dictionnaire {nom_label: id_label}. Retourne un dict vide en cas d'erreur.
    """
    url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/labels"
    reponse = gitea_request("get", url)
    if not reponse:
        return {}

    try:
        return {label['name']: label['id'] for label in reponse.json()}
    except Exception as e:
        st.error(f"{str(_('errors.label_parsing'))} {e}")
        return {}


def nettoyer_labels(labels):
    """Nettoie et deduplique une liste de labels.

    Args:
        labels: Liste de labels (peut contenir des non-strings, espaces, doublons).

    Returns:
        list[str]: Liste triee de labels uniques et non vides.
    """
    return sorted(set(l.strip() for l in labels if isinstance(l, str) and l.strip()))


def construire_corps_ticket_markdown(reponses):
    """Construit le corps markdown d'un ticket a partir des reponses utilisateur.

    Args:
        reponses: Dictionnaire {nom_section: texte_reponse}.

    Returns:
        str: Corps du ticket en format markdown avec sections de niveau 2.
    """
    return "\n\n".join(f"## {section}\n{texte}" for section, texte in reponses.items())


def creer_ticket_gitea(titre, corps, labels):
    """Cree un nouveau ticket (issue) dans le depot Gitea.

    Args:
        titre: Titre du ticket.
        corps: Corps du ticket en markdown.
        labels: Liste d'IDs de labels a associer au ticket.

    Returns:
        bool: True si creation reussie, False sinon.
    """
    data = {
        "title": titre,
        "body": corps,
        "labels": labels,
        "ref": f"refs/heads/{ENV}"
    }
    url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/issues"

    reponse = gitea_request("post", url, headers={"Content-Type": "application/json"}, data=json.dumps(data))
    if not reponse:
        return False
    return True