372 lines
14 KiB
Python
372 lines
14 KiB
Python
import streamlit as st
|
||
from datetime import datetime
|
||
from dateutil import parser
|
||
from collections import defaultdict
|
||
import os
|
||
import csv
|
||
import requests
|
||
import base64
|
||
import re
|
||
import json
|
||
import html
|
||
|
||
# Configuration Gitea
|
||
GITEA_URL = os.getenv("GITEA_URL", "https://fabnum-git.peccini.fr/api/v1")
|
||
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
|
||
ORGANISATION = os.getenv("ORGANISATION", "fabnum")
|
||
DEPOT_FICHES = os.getenv("DEPOT_FICHES", "fiches")
|
||
ENV = os.getenv("ENV")
|
||
|
||
def charger_fiches_et_labels():
|
||
chemin_csv = os.path.join("assets", "fiches_labels.csv")
|
||
dictionnaire_fiches = {}
|
||
|
||
try:
|
||
with open(chemin_csv, mode="r", 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"❌ Le fichier {chemin_csv} est introuvable.")
|
||
except Exception as e:
|
||
st.error(f"❌ Erreur lors du chargement des fiches : {str(e)}")
|
||
|
||
return dictionnaire_fiches
|
||
|
||
def rechercher_tickets_gitea(fiche_selectionnee):
|
||
headers = {"Authorization": f"token " + GITEA_TOKEN}
|
||
params = {"state": "open"}
|
||
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/issues"
|
||
|
||
try:
|
||
reponse = requests.get(url, headers=headers, params=params, timeout=10)
|
||
reponse.raise_for_status()
|
||
issues = reponse.json()
|
||
|
||
correspondances = charger_fiches_et_labels()
|
||
cible = correspondances.get(fiche_selectionnee)
|
||
|
||
if not cible:
|
||
return []
|
||
|
||
labels_cibles = set(cible["operations"] + [cible["item"]])
|
||
|
||
tickets_associes = []
|
||
for issue in issues:
|
||
if issue.get("ref") != f"refs/heads/{ENV}":
|
||
continue
|
||
issue_labels = set()
|
||
for label in issue.get("labels", []):
|
||
if isinstance(label, dict) and "name" in label:
|
||
issue_labels.add(label["name"])
|
||
|
||
if labels_cibles.issubset(issue_labels):
|
||
tickets_associes.append(issue)
|
||
|
||
return tickets_associes
|
||
|
||
except requests.RequestException as e:
|
||
st.error(f"Erreur lors de la récupération des tickets : {e}")
|
||
return []
|
||
|
||
def extraire_statut_par_label(ticket):
|
||
labels = [label.get('name', '') for label in ticket.get('labels', [])]
|
||
for statut in ["Backlog", "En attente de traitement", "En cours", "Terminés", "Non retenus"]:
|
||
if statut in labels:
|
||
return statut
|
||
return "Autres"
|
||
|
||
def afficher_tickets_par_fiche(tickets):
|
||
if not tickets:
|
||
st.info("Aucun ticket lié à cette fiche.")
|
||
return
|
||
|
||
st.markdown("📝 **Tickets associés à cette fiche**")
|
||
|
||
tickets_groupes = defaultdict(list)
|
||
for ticket in tickets:
|
||
statut = extraire_statut_par_label(ticket)
|
||
tickets_groupes[statut].append(ticket)
|
||
|
||
st.info(f" ⤇ {len(tickets_groupes["Backlog"])} ticket(s) en attente de modération ne sont pas affichés.")
|
||
|
||
ordre_statuts = ["En attente de traitement", "En cours", "Terminés", "Non retenus", "Autres"]
|
||
|
||
for statut in ordre_statuts:
|
||
if tickets_groupes[statut]:
|
||
with st.expander(f"{statut} ({len(tickets_groupes[statut])})", expanded=(statut == "En cours")):
|
||
for ticket in tickets_groupes[statut]:
|
||
afficher_carte_ticket(ticket)
|
||
|
||
def recuperer_commentaires_ticket(issue_index):
|
||
headers = {
|
||
"Authorization": f"token {GITEA_TOKEN}"
|
||
}
|
||
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/issues/{issue_index}/comments"
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=10)
|
||
response.raise_for_status()
|
||
commentaires = response.json()
|
||
return commentaires
|
||
except Exception as e:
|
||
st.error(f"Erreur lors de la récupération des commentaires pour le ticket {issue_index} : {e}")
|
||
return []
|
||
|
||
def afficher_carte_ticket(ticket):
|
||
titre = ticket.get("title", "Sans titre")
|
||
url = ticket.get("html_url", "")
|
||
user = ticket.get("user", {}).get("login", "inconnu")
|
||
created = ticket.get("created_at", "")
|
||
updated = ticket.get("updated_at", "")
|
||
body = ticket.get("body", "")
|
||
labels = [l["name"] for l in ticket.get("labels", []) if "name" in l]
|
||
|
||
sujet = ""
|
||
match = re.search(r"## Sujet de la proposition\s+(.+?)(\n|$)", body, re.DOTALL)
|
||
if match:
|
||
sujet = match.group(1).strip()
|
||
|
||
if created:
|
||
try:
|
||
dt_created = parser.isoparse(created)
|
||
date_created_str = dt_created.strftime("%d/%m/%Y")
|
||
except Exception:
|
||
date_created_str = "Date inconnue"
|
||
else:
|
||
date_created_str = "Date inconnue"
|
||
|
||
if updated and updated != created:
|
||
try:
|
||
dt_updated = parser.isoparse(updated)
|
||
date_updated_str = dt_updated.strftime("%d/%m/%Y")
|
||
maj_info = f"(MAJ {date_updated_str})"
|
||
except Exception:
|
||
maj_info = ""
|
||
else:
|
||
maj_info = ""
|
||
|
||
commentaires = recuperer_commentaires_ticket(ticket.get("number"))
|
||
|
||
commentaires_html = ''
|
||
if commentaires:
|
||
for commentaire in commentaires:
|
||
auteur = html.escape(commentaire.get('user', {}).get('login', 'inconnu'))
|
||
contenu = html.escape(commentaire.get('body', ''))
|
||
date_commentaire = commentaire.get('created_at', '')
|
||
if date_commentaire:
|
||
try:
|
||
dt_comment = parser.isoparse(date_commentaire)
|
||
date_commentaire_str = dt_comment.strftime("%d/%m/%Y")
|
||
except Exception:
|
||
date_commentaire_str = ""
|
||
else:
|
||
date_commentaire_str = ""
|
||
|
||
commentaires_html += f"""
|
||
<div style='background-color: #f0f0f0; padding: 0.5rem; border-radius: 8px; margin-bottom: 0.5rem;'>
|
||
<p style='margin: 0;'><strong>{auteur}</strong> <small>({date_commentaire_str})</small></p>
|
||
<p style='margin: 0.5rem 0 0 0;'>{contenu}</p>
|
||
</div>
|
||
"""
|
||
else:
|
||
commentaires_html = '<p style="margin-top: 1rem;">Aucun commentaire.</p>'
|
||
|
||
with st.container():
|
||
st.markdown(f"""
|
||
<div style='border: 1px solid #ccc; border-radius: 12px; padding: 1rem; margin-bottom: 2rem; box-shadow: 0 1px 3px rgba(0,0,0,0.1);'>
|
||
<h4 style='margin-bottom: 0.5rem;'>🎫 <a href='{url}' target='_blank'>{titre}</a></h4>
|
||
<p style='margin: 0.2rem 0;'>Ouvert par <strong>{html.escape(user)}</strong> le {date_created_str} {maj_info}</p>
|
||
<p style='margin: 0.2rem 0;'>Sujet de la proposition : <strong>{html.escape(sujet)}</strong></p>
|
||
<p style='margin: 0.2rem 0;'><span>{' • '.join(html.escape(label) for label in labels) if labels else 'aucun'}</span></p>
|
||
<hr style='margin: 1.5rem 0;'>
|
||
</div>
|
||
""", unsafe_allow_html=True)
|
||
|
||
st.markdown("**Contenu du ticket :**")
|
||
st.markdown(body, unsafe_allow_html=False)
|
||
|
||
st.markdown("---")
|
||
st.markdown("**Commentaires :**")
|
||
st.markdown(commentaires_html, unsafe_allow_html=True)
|
||
|
||
def get_labels_existants():
|
||
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
||
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/labels"
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=10)
|
||
response.raise_for_status()
|
||
labels_data = response.json()
|
||
return {label['name']: label['id'] for label in labels_data}
|
||
except Exception as e:
|
||
st.error(f"Erreur lors de la récupération des labels existants : {e}")
|
||
return {}
|
||
|
||
|
||
def creer_ticket_gitea(titre, corps, labels):
|
||
headers = {
|
||
"Authorization": f"token {GITEA_TOKEN}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/issues"
|
||
|
||
data = {
|
||
"title": titre,
|
||
"body": corps,
|
||
"labels": labels,
|
||
"ref": f"refs/heads/{ENV}"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, headers=headers, data=json.dumps(data), timeout=10)
|
||
response.raise_for_status()
|
||
issue = response.json()
|
||
issue_url = issue.get("html_url", "")
|
||
if issue_url:
|
||
st.success(f"✅ Ticket créé avec succès ! [Voir le ticket]({issue_url})")
|
||
else:
|
||
st.success("✅ Ticket créé avec succès !")
|
||
except Exception as e:
|
||
st.error(f"❌ Erreur lors de la création du ticket : {e}")
|
||
|
||
def charger_modele_ticket():
|
||
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
||
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/contents/.gitea/ISSUE_TEMPLATE/Contenu.md"
|
||
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=10)
|
||
response.raise_for_status()
|
||
contenu_base64 = response.json().get("content", "")
|
||
contenu = base64.b64decode(contenu_base64).decode("utf-8")
|
||
return contenu
|
||
except Exception as e:
|
||
st.error(f"Erreur lors du chargement du modèle de ticket : {e}")
|
||
return ""
|
||
|
||
def gerer_tickets_fiche(fiche_selectionnee):
|
||
st.markdown("""
|
||
<hr style='border: 1px solid #ccc; margin: 2rem 0;' />
|
||
""", unsafe_allow_html=True)
|
||
st.markdown("### 🧾 Gestion des tickets pour cette fiche")
|
||
|
||
tickets = rechercher_tickets_gitea(fiche_selectionnee)
|
||
afficher_tickets_par_fiche(tickets)
|
||
formulaire_creation_ticket_dynamique(fiche_selectionnee)
|
||
|
||
# Modification de formulaire_creation_ticket_dynamique pour ajouter Annuler
|
||
def formulaire_creation_ticket_dynamique(fiche_selectionnee):
|
||
with st.expander("➕ Créer un nouveau ticket lié à cette fiche", expanded=False):
|
||
contenu_modele = charger_modele_ticket()
|
||
|
||
if not contenu_modele:
|
||
st.error("Impossible de charger le modèle de ticket.")
|
||
return
|
||
|
||
sections = {}
|
||
lignes = contenu_modele.splitlines()
|
||
titre_courant = None
|
||
contenu_section = []
|
||
|
||
for ligne in lignes:
|
||
if ligne.startswith("## ") and titre_courant:
|
||
sections[titre_courant] = "\n".join(contenu_section).strip()
|
||
titre_courant = ligne[3:].strip()
|
||
contenu_section = []
|
||
elif ligne.startswith("## "):
|
||
titre_courant = ligne[3:].strip()
|
||
contenu_section = []
|
||
elif titre_courant:
|
||
contenu_section.append(ligne)
|
||
|
||
if titre_courant and contenu_section:
|
||
sections[titre_courant] = "\n".join(contenu_section).strip()
|
||
|
||
reponses = {}
|
||
labels = []
|
||
selected_operations = []
|
||
|
||
correspondances = charger_fiches_et_labels()
|
||
cible = correspondances.get(fiche_selectionnee)
|
||
if cible:
|
||
if len(cible["operations"]) == 1:
|
||
labels.append(cible["operations"][0])
|
||
elif len(cible["operations"]) > 1:
|
||
selected_operations = st.multiselect("Labels opération à associer", cible["operations"], default=cible["operations"])
|
||
|
||
for section, aide in sections.items():
|
||
if "Type de contribution" in section:
|
||
options = re.findall(r"- \[.\] (.+)", aide)
|
||
clean_options = []
|
||
for opt in options:
|
||
base = opt.split(":")[0].strip()
|
||
if base not in clean_options:
|
||
clean_options.append(base)
|
||
if "Autre" not in clean_options:
|
||
clean_options.append("Autre")
|
||
|
||
type_contribution = st.radio("Type de contribution", clean_options)
|
||
if type_contribution == "Autre":
|
||
autre = st.text_input("Précisez le type de contribution")
|
||
reponses[section] = autre
|
||
else:
|
||
reponses[section] = type_contribution
|
||
|
||
elif "Fiche concernée" in section:
|
||
base_url = f"https://fabnum-git.peccini.fr/FabNum/Fiches/src/branch/{ENV}/Documents/"
|
||
url_fiche = f"{base_url}{fiche_selectionnee.replace(' ', '%20')}"
|
||
reponses[section] = url_fiche
|
||
st.text_input("Fiche concernée", value=url_fiche, disabled=True)
|
||
|
||
elif "Sujet de la proposition" in section:
|
||
reponses[section] = st.text_input(section, help=aide)
|
||
|
||
else:
|
||
reponses[section] = st.text_area(section, help=aide)
|
||
|
||
col1, col2 = st.columns(2)
|
||
|
||
with col1:
|
||
if st.button("Prévisualiser le ticket"):
|
||
st.session_state.previsualiser = True
|
||
|
||
with col2:
|
||
if st.button("Annuler"):
|
||
st.session_state.previsualiser = False
|
||
st.rerun()
|
||
|
||
if st.session_state.get("previsualiser", False):
|
||
st.subheader("Prévisualisation du ticket")
|
||
for section, texte in reponses.items():
|
||
st.markdown(f"#### {section}")
|
||
st.markdown(texte)
|
||
|
||
if st.button("Confirmer la création du ticket"):
|
||
if cible:
|
||
labels.append(cible["item"])
|
||
if selected_operations:
|
||
labels.extend(selected_operations)
|
||
|
||
labels = list(set([l.strip() for l in labels if l and l.strip()]))
|
||
titre_ticket = reponses.get("Sujet de la proposition", "").strip() or "Ticket FabNum"
|
||
|
||
labels_existants = get_labels_existants()
|
||
labels_ids = [labels_existants[l] for l in labels if l in labels_existants]
|
||
|
||
if "Backlog" in labels_existants:
|
||
labels_ids.append(labels_existants["Backlog"])
|
||
|
||
corps = ""
|
||
for section, texte in reponses.items():
|
||
corps += f"## {section}\n{texte}\n\n"
|
||
|
||
creer_ticket_gitea(titre=titre_ticket, corps=corps, labels=labels_ids)
|
||
|
||
st.success("Formulaire vidé après création du ticket.")
|