120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
# core.py
|
|
|
|
import csv
|
|
import json
|
|
import requests
|
|
import os
|
|
import streamlit as st
|
|
from utils.translations import _
|
|
from config import GITEA_URL, GITEA_TOKEN, ORGANISATION, DEPOT_FICHES, ENV
|
|
|
|
|
|
def gitea_request(method, url, **kwargs):
|
|
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', 'Erreur Gitea'))} ({method.upper()}): {e}")
|
|
return None
|
|
|
|
|
|
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"❌ {str(_('errors.file_not_found', 'Le fichier'))} {chemin_csv} {str(_('errors.is_missing', 'est introuvable.'))}")
|
|
except Exception as e:
|
|
st.error(f"❌ {str(_('errors.file_loading', 'Erreur lors du chargement des fiches :'))} {str(e)}")
|
|
|
|
return dictionnaire_fiches
|
|
|
|
|
|
def rechercher_tickets_gitea(fiche_selectionnee):
|
|
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', 'Erreur de décodage JSON :'))} {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():
|
|
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', 'Erreur de parsing des labels :'))} {e}")
|
|
return {}
|
|
|
|
|
|
def nettoyer_labels(labels):
|
|
return sorted(set(l.strip() for l in labels if isinstance(l, str) and l.strip()))
|
|
|
|
|
|
def construire_corps_ticket_markdown(reponses):
|
|
return "\n\n".join(f"## {section}\n{texte}" for section, texte in reponses.items())
|
|
|
|
|
|
def creer_ticket_gitea(titre, corps, labels):
|
|
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
|
|
|
|
issue_url = reponse.json().get("html_url", "")
|
|
if issue_url:
|
|
st.success(f"{str(_('pages.fiches.tickets.created_success', 'Ticket créé !'))} [Voir le ticket]({issue_url})")
|
|
else:
|
|
st.success(str(_('pages.fiches.tickets.created', 'Ticket créé avec succès.')))
|