794 lines
36 KiB
Python
794 lines
36 KiB
Python
import disnake
|
||
from disnake.ext import commands
|
||
from utils.forgejo_client import gitlab_client
|
||
from aiohttp import web
|
||
import logging
|
||
import json
|
||
import os
|
||
import asyncio
|
||
import re
|
||
from typing import Dict, Any, Optional
|
||
from utils.premium import check_premium_tier
|
||
|
||
kuby_logger = logging.getLogger("KubyBot")
|
||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
|
||
TRACKING_FILE = os.path.join(PROJECT_ROOT, "commandes", "gitlab_issues.json")
|
||
LEGACY_REPORTS_FILE = os.path.join(PROJECT_ROOT, "data", "gitlab_reports.json")
|
||
|
||
# S'assure que le fichier de tracking est toujours écrit à l'emplacement attendu même si le service systemd démarre depuis un autre répertoire.
|
||
if not os.path.exists(TRACKING_FILE):
|
||
legacy_candidate = os.path.join(BASE_DIR, "commandes", "gitlab_issues.json")
|
||
if os.path.exists(legacy_candidate):
|
||
TRACKING_FILE = legacy_candidate
|
||
|
||
|
||
def classify_webhook_event(payload: Dict[str, Any]) -> Optional[str]:
|
||
"""Détermine si un webhook correspond à une issue ou à un commentaire."""
|
||
object_kind = payload.get("object_kind")
|
||
if object_kind in {"issue", "note"}:
|
||
return object_kind
|
||
|
||
action = str(payload.get("action") or payload.get("event") or "").lower()
|
||
if action in {"comment_created", "comment_updated", "commented", "issue_comment_created", "issue_comment_updated"}:
|
||
return "note"
|
||
|
||
if "comment" in payload or payload.get("comment") or "comment" in action:
|
||
return "note"
|
||
|
||
if payload.get("issue") or payload.get("object_attributes") or action in {"open", "reopen", "close", "closed", "updated", "created", "edited", "label_updated", "label_added", "label_removed", "label_changed"}:
|
||
return "issue"
|
||
|
||
return None
|
||
|
||
|
||
def extract_issue_iid(payload: Dict[str, Any]) -> Optional[str]:
|
||
"""Extrait l’identifiant d’issue depuis plusieurs formats Forgejo/GitLab."""
|
||
for container in (payload.get("object_attributes", {}), payload.get("issue", {}), payload.get("comment", {})):
|
||
if not isinstance(container, dict):
|
||
continue
|
||
for key in ("iid", "number", "id"):
|
||
value = container.get(key)
|
||
if value not in (None, ""):
|
||
return str(value)
|
||
return None
|
||
|
||
|
||
def extract_issue_identifier(issue_payload: Optional[Any]) -> Optional[str]:
|
||
"""Récupère un identifiant stable à stocker dans le tracking, même si le payload est imbriqué."""
|
||
if issue_payload is None:
|
||
return None
|
||
|
||
if isinstance(issue_payload, dict):
|
||
for key in ("iid", "number", "id", "issue_id"):
|
||
value = issue_payload.get(key)
|
||
if value not in (None, ""):
|
||
return str(value)
|
||
|
||
for key in ("html_url", "web_url", "url"):
|
||
value = issue_payload.get(key)
|
||
if isinstance(value, str):
|
||
match = re.search(r"/(?:issues|tickets)/([0-9]+)", value)
|
||
if match:
|
||
return match.group(1)
|
||
|
||
for value in issue_payload.values():
|
||
found = extract_issue_identifier(value)
|
||
if found:
|
||
return found
|
||
|
||
return None
|
||
|
||
if isinstance(issue_payload, list):
|
||
for item in issue_payload:
|
||
found = extract_issue_identifier(item)
|
||
if found:
|
||
return found
|
||
|
||
return None
|
||
|
||
|
||
def normalize_labels(labels: Any) -> list[str]:
|
||
"""Normalise une liste de labels provenant de Forgejo/GitLab."""
|
||
if not labels:
|
||
return []
|
||
if isinstance(labels, str):
|
||
return [labels]
|
||
if isinstance(labels, list):
|
||
normalized = []
|
||
for item in labels:
|
||
if isinstance(item, dict):
|
||
name = item.get("name") or item.get("title") or item.get("label") or item.get("id")
|
||
else:
|
||
name = item
|
||
if name is None:
|
||
continue
|
||
name = str(name).strip()
|
||
if name and name not in normalized:
|
||
normalized.append(name)
|
||
return normalized
|
||
return []
|
||
|
||
|
||
def extract_label_change(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
"""Extrait le changement de labels d’un payload d’issue."""
|
||
action = str(payload.get("action") or payload.get("event") or "").lower()
|
||
if not action.startswith("label"):
|
||
return None
|
||
|
||
issue = payload.get("issue") or {}
|
||
current_labels = normalize_labels(issue.get("labels") or payload.get("labels"))
|
||
if not current_labels:
|
||
return None
|
||
|
||
label_payload = payload.get("label")
|
||
label_name = None
|
||
if isinstance(label_payload, dict):
|
||
label_name = label_payload.get("name") or label_payload.get("title") or label_payload.get("label")
|
||
elif isinstance(label_payload, str):
|
||
label_name = label_payload
|
||
|
||
previous_labels = []
|
||
if isinstance(payload.get("changes"), dict):
|
||
prev = payload["changes"].get("labels") or payload["changes"].get("label")
|
||
if isinstance(prev, dict):
|
||
previous_labels = normalize_labels(prev.get("previous"))
|
||
|
||
if not previous_labels and issue.get("labels"):
|
||
previous_labels = []
|
||
|
||
if action in {"label_removed", "label_deleted"}:
|
||
return {
|
||
"previous": previous_labels or ([label_name] if label_name else []),
|
||
"current": current_labels,
|
||
"added": [],
|
||
"removed": [label_name] if label_name else [],
|
||
}
|
||
|
||
added = [label for label in current_labels if label not in previous_labels]
|
||
removed = [label for label in previous_labels if label not in current_labels]
|
||
if not added and not removed and label_name:
|
||
added = [label_name]
|
||
|
||
return {
|
||
"previous": previous_labels,
|
||
"current": current_labels,
|
||
"added": added,
|
||
"removed": removed,
|
||
}
|
||
|
||
|
||
def calculate_progress_from_labels(labels: list[str]) -> float:
|
||
"""Estime une progression visuelle à partir des labels présents."""
|
||
if not labels:
|
||
return 0.1
|
||
|
||
normalized = [label.lower() for label in labels]
|
||
if any(token in normalized for token in ["done", "resolved", "fixed", "closed", "terminé", "résolu", "ok", "completed"]):
|
||
return 1.0
|
||
if any(token in normalized for token in ["review", "ready", "qa", "testing", "test", "validated", "validation", "verif", "vérif"]):
|
||
return 0.8
|
||
if any(token in normalized for token in ["in progress", "working", "wip", "develop", "dev", "investigating", "progress", "travail"]):
|
||
return 0.6
|
||
if any(token in normalized for token in ["confirmed", "accepted", "triaged", "priority", "urgent", "prioritaire"]):
|
||
return 0.4
|
||
if any(token in normalized for token in ["bug", "enhancement", "feature", "manual", "new", "suggestion"]):
|
||
return 0.2
|
||
return 0.3
|
||
|
||
|
||
def build_status_message_from_labels(labels: list[str]) -> str:
|
||
"""Traduit les labels en un message utilisateur lisible."""
|
||
if not labels:
|
||
return "⏳ La demande évolue et l'équipe suit son avancement."
|
||
|
||
normalized = [str(label).lower() for label in labels if str(label).strip()]
|
||
if any(token in normalized for token in ["done", "resolved", "fixed", "closed", "terminé", "résolu", "completed", "ok"]):
|
||
return "✅ La demande a été traitée et est prête à être déployée."
|
||
if any(token in normalized for token in ["review", "qa", "testing", "test", "validated", "validation", "verif", "vérif"]):
|
||
return "🔎 La demande est en cours de vérification par l'équipe."
|
||
if any(token in normalized for token in ["in progress", "working", "wip", "develop", "dev", "investigating", "progress", "travail"]):
|
||
return "🛠️ L'équipe est en train de travailler dessus."
|
||
if any(token in normalized for token in ["confirmed", "accepted", "triaged", "priority", "urgent", "prioritaire"]):
|
||
return "🗂️ La demande a bien été prise en compte et est en cours d'analyse."
|
||
if any(token in normalized for token in ["bug", "enhancement", "feature", "manual", "new", "suggestion"]):
|
||
return "📝 La demande a été reçue et sera étudiée prochainement."
|
||
return "⏳ La demande évolue et l'équipe suit son avancement."
|
||
|
||
|
||
def build_progress_bar(progress: float, width: int = 18) -> str:
|
||
"""Construit une barre de progression stylée en Unicode."""
|
||
progress = max(0.0, min(1.0, progress))
|
||
filled = int(round(progress * width))
|
||
filled = max(0, min(width, filled))
|
||
empty = width - filled
|
||
percent = int(progress * 100)
|
||
return f"{'▰' * filled}{'▱' * empty} {percent}%"
|
||
|
||
|
||
def color_from_progress(progress: float) -> disnake.Color:
|
||
"""Retourne une couleur violet→bleu selon la progression."""
|
||
progress = max(0.0, min(1.0, progress))
|
||
purple = (117, 82, 255)
|
||
blue = (74, 144, 255)
|
||
r = int(purple[0] + (blue[0] - purple[0]) * progress)
|
||
g = int(purple[1] + (blue[1] - purple[1]) * progress)
|
||
b = int(purple[2] + (blue[2] - purple[2]) * progress)
|
||
return disnake.Color.from_rgb(r, g, b)
|
||
|
||
|
||
def normalize_tracking_issue_id(issue_iid: Any) -> Optional[str]:
|
||
"""Normalise un identifiant d’issue pour le tracking, en ne conservant que les valeurs numériques valides."""
|
||
if issue_iid is None:
|
||
return None
|
||
|
||
if isinstance(issue_iid, int):
|
||
return str(issue_iid)
|
||
|
||
if isinstance(issue_iid, str):
|
||
normalized = issue_iid.strip()
|
||
if not normalized:
|
||
return None
|
||
if normalized.lower() in {"none", "null", "nan", "undefined"}:
|
||
return None
|
||
if re.fullmatch(r"\d+", normalized):
|
||
return normalized
|
||
|
||
return None
|
||
|
||
|
||
def extract_requester_id_from_payload(payload: Dict[str, Any]) -> Optional[int]:
|
||
"""Essaie d'extraire l'ID Discord du demandeur depuis la description du ticket ou le payload webhook."""
|
||
text_candidates = []
|
||
for key in ("description", "body", "content"):
|
||
value = payload.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
text_candidates.append(value)
|
||
|
||
issue_payload = payload.get("issue") or {}
|
||
for key in ("description", "body", "content"):
|
||
value = issue_payload.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
text_candidates.append(value)
|
||
|
||
object_attributes = payload.get("object_attributes") or {}
|
||
for key in ("description", "body", "content"):
|
||
value = object_attributes.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
text_candidates.append(value)
|
||
|
||
for text in text_candidates:
|
||
matches = re.findall(r"\b(\d{15,20})\b", text)
|
||
if matches:
|
||
return int(matches[0])
|
||
|
||
return None
|
||
|
||
|
||
def load_tracking():
|
||
"""Charge le tracking depuis gitlab_issues.json (avec migration automatique)"""
|
||
tracking = {}
|
||
if os.path.exists(TRACKING_FILE):
|
||
try:
|
||
with open(TRACKING_FILE, "r", encoding="utf-8") as f:
|
||
tracking = json.load(f)
|
||
except Exception as e:
|
||
kuby_logger.error(f"Erreur chargement gitlab_issues.json: {e}")
|
||
|
||
# Migration automatique depuis l'ancien fichier
|
||
if os.path.exists(LEGACY_REPORTS_FILE):
|
||
try:
|
||
with open(LEGACY_REPORTS_FILE, "r", encoding="utf-8") as f:
|
||
legacy_data = json.load(f)
|
||
|
||
migrated_count = 0
|
||
for raw_issue_iid, user_id in legacy_data.items():
|
||
issue_iid = normalize_tracking_issue_id(raw_issue_iid)
|
||
if not issue_iid:
|
||
kuby_logger.debug(f"Ignoré pendant la migration de tracking: issue_id={raw_issue_iid}")
|
||
continue
|
||
if issue_iid in tracking:
|
||
continue
|
||
tracking[issue_iid] = {
|
||
"requester_id": user_id,
|
||
"last_note_id": 0,
|
||
"type": "legacy_report"
|
||
}
|
||
migrated_count += 1
|
||
|
||
kuby_logger.info(f"Migration automatique: {migrated_count} issue(s) fusionnée(s) depuis l'historique")
|
||
except Exception as e:
|
||
kuby_logger.error(f"Erreur migration gitlab_reports.json: {e}")
|
||
|
||
return tracking
|
||
|
||
|
||
def save_tracking(data: Dict[str, Any]) -> None:
|
||
"""Sauvegarde dans gitlab_issues.json"""
|
||
os.makedirs(os.path.dirname(TRACKING_FILE), exist_ok=True)
|
||
temp_file = f"{TRACKING_FILE}.tmp"
|
||
with open(temp_file, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2)
|
||
os.replace(temp_file, TRACKING_FILE)
|
||
|
||
|
||
class BugReportModal(disnake.ui.Modal):
|
||
def __init__(self, cog):
|
||
self.cog = cog # ⚡ NOUVEAU: Reçoit le cog pour tracking
|
||
components = [
|
||
disnake.ui.TextInput(
|
||
label="Titre du bug",
|
||
placeholder="Décrivez brièvement le problème...",
|
||
custom_id="bug_title",
|
||
required=True,
|
||
max_length=100,
|
||
),
|
||
disnake.ui.TextInput(
|
||
label="Description détaillée",
|
||
style=disnake.TextInputStyle.paragraph,
|
||
placeholder="Que s'est-il passé ?",
|
||
custom_id="bug_description",
|
||
required=True,
|
||
max_length=1000,
|
||
),
|
||
]
|
||
super().__init__(
|
||
title="Signaler un Bug",
|
||
custom_id="bug_modal",
|
||
components=components,
|
||
)
|
||
|
||
async def callback(self, interaction: disnake.ModalInteraction):
|
||
await interaction.response.send_message("Envoi de votre rapport de bug à GitLab...", ephemeral=True)
|
||
|
||
bug_title = (interaction.text_values.get("bug_title") or "").strip()
|
||
if not bug_title:
|
||
bug_title = "Sans titre"
|
||
|
||
priority_name = "Normale"
|
||
priority_value = "Normal"
|
||
description_value = interaction.text_values.get("bug_description") or ""
|
||
|
||
description_text = f"**Rapporté par:** {interaction.user} ({interaction.user.id})\n"
|
||
description_text += f"**Priorité:** {priority_name}\n\n"
|
||
description_text += f"**Description:**\n{description_value}"
|
||
|
||
labels = ["bug", "manual-report", f"Priority::{priority_value}"]
|
||
|
||
result = await gitlab_client.create_issue(
|
||
title=f"[MANUAL] {bug_title}",
|
||
description=description_text,
|
||
labels=labels,
|
||
)
|
||
|
||
if result:
|
||
issue_iid = extract_issue_identifier(result)
|
||
if not issue_iid:
|
||
kuby_logger.warning("Impossible de déterminer l’ID de l’issue créée depuis le retour Forgejo/GitLab. Tentative de fallback par titre.")
|
||
lookup = await gitlab_client.find_issue_by_title(f"[MANUAL] {bug_title}", state="open")
|
||
issue_iid = extract_issue_identifier(lookup)
|
||
|
||
if not issue_iid:
|
||
kuby_logger.warning("Impossible de déterminer l’ID de l’issue créée après fallback. Payload reçu: %s", result)
|
||
await interaction.edit_original_response(content="✅ Votre rapport a été envoyé, mais l’identifiant de l’issue n’a pas pu être récupéré automatiquement.")
|
||
return
|
||
|
||
self.cog.issue_data[issue_iid] = {
|
||
"requester_id": interaction.user.id,
|
||
"last_note_id": 0,
|
||
"type": "manual-report"
|
||
}
|
||
save_tracking(self.cog.issue_data)
|
||
await interaction.edit_original_response(
|
||
content=(
|
||
"✅ Votre bug a été signalé avec succès ! "
|
||
"Nous avons bien été averti et le bug sera réglé dans la prochaine mise à jour. "
|
||
f"[Voir l'issue]({result.get('web_url') or result.get('html_url')})"
|
||
)
|
||
)
|
||
else:
|
||
await interaction.edit_original_response(
|
||
content="❌ Une erreur est survenue lors de l'envoi du rapport à GitLab. Veuillez contacter un administrateur."
|
||
)
|
||
|
||
|
||
class FeatureSuggestionModal(disnake.ui.Modal):
|
||
def __init__(self, cog):
|
||
self.cog = cog # ⚡ NOUVEAU: Reçoit le cog pour tracking
|
||
components = [
|
||
disnake.ui.TextInput(
|
||
label="Titre de la suggestion",
|
||
placeholder="Que voulez-vous ajouter ?",
|
||
custom_id="suggestion_title",
|
||
required=True,
|
||
max_length=100,
|
||
),
|
||
disnake.ui.TextInput(
|
||
label="Détails de la fonctionnalité",
|
||
style=disnake.TextInputStyle.paragraph,
|
||
placeholder="Décrivez comment cela devrait fonctionner...",
|
||
custom_id="suggestion_description",
|
||
required=True,
|
||
max_length=1000,
|
||
),
|
||
]
|
||
super().__init__(
|
||
title="Suggérer une fonctionnalité",
|
||
custom_id="suggestion_modal",
|
||
components=components,
|
||
)
|
||
|
||
async def callback(self, interaction: disnake.ModalInteraction):
|
||
await interaction.response.send_message("Envoi de votre suggestion à GitLab...", ephemeral=True)
|
||
|
||
suggestion_title = (interaction.text_values.get("suggestion_title") or "").strip() or "Sans titre"
|
||
description_value = interaction.text_values.get("suggestion_description") or ""
|
||
|
||
description_text = f"**Sugéré par:** {interaction.user} ({interaction.user.id})\n\n"
|
||
description_text += f"**Description:**\n{description_value}"
|
||
|
||
labels = ["enhancement", "manual-suggestion"]
|
||
|
||
result = await gitlab_client.create_issue(
|
||
title=f"[SUGGESTION] {suggestion_title}",
|
||
description=description_text,
|
||
labels=labels,
|
||
)
|
||
|
||
if result:
|
||
issue_iid = extract_issue_identifier(result)
|
||
if not issue_iid:
|
||
kuby_logger.warning("Impossible de déterminer l’ID de l’issue créée depuis le retour Forgejo/GitLab. Tentative de fallback par titre.")
|
||
lookup = await gitlab_client.find_issue_by_title(f"[SUGGESTION] {suggestion_title}", state="open")
|
||
issue_iid = extract_issue_identifier(lookup)
|
||
|
||
if not issue_iid:
|
||
kuby_logger.warning("Impossible de déterminer l’ID de l’issue créée après fallback. Payload reçu: %s", result)
|
||
await interaction.edit_original_response(content="✅ Votre suggestion a été envoyée, mais l’identifiant de l’issue n’a pas pu être récupéré automatiquement.")
|
||
return
|
||
|
||
self.cog.issue_data[issue_iid] = {
|
||
"requester_id": interaction.user.id,
|
||
"last_note_id": 0,
|
||
"type": "manual-suggestion"
|
||
}
|
||
save_tracking(self.cog.issue_data)
|
||
await interaction.edit_original_response(
|
||
content=(
|
||
"✅ Votre suggestion a été envoyée avec succès ! "
|
||
"Merci de contribuer à l'amélioration du bot. "
|
||
f"[Voir l'issue]({result.get('web_url') or result.get('html_url')})"
|
||
)
|
||
)
|
||
else:
|
||
await interaction.edit_original_response(
|
||
content="❌ Une erreur est survenue lors de l'envoi de la suggestion à GitLab."
|
||
)
|
||
|
||
|
||
class BugReportView(disnake.ui.View):
|
||
def __init__(self, cog):
|
||
self.cog = cog # ⚡ NOUVEAU: Reçoit le cog
|
||
super().__init__(timeout=None)
|
||
|
||
@disnake.ui.button(label="🎫 Remplir le formulaire", style=disnake.ButtonStyle.primary, custom_id="open_bug_form")
|
||
async def open_bug_form(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
|
||
modal = BugReportModal(self.cog)
|
||
await interaction.response.send_modal(modal)
|
||
|
||
|
||
class FeatureSuggestionView(disnake.ui.View):
|
||
def __init__(self, cog):
|
||
self.cog = cog # ⚡ NOUVEAU: Reçoit le cog
|
||
super().__init__(timeout=None)
|
||
|
||
@disnake.ui.button(label="🎫 Remplir le formulaire", style=disnake.ButtonStyle.green, custom_id="open_suggestion_form")
|
||
async def open_suggestion_form(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
|
||
modal = FeatureSuggestionModal(self.cog)
|
||
await interaction.response.send_modal(modal)
|
||
|
||
|
||
class BugReport(commands.Cog):
|
||
def __init__(self, bot):
|
||
self.bot = bot
|
||
self.issue_data = load_tracking() # ⚡ NOUVEAU: Charge le tracking unifié
|
||
|
||
async def cog_load(self):
|
||
kuby_logger.info("BugReport cog chargé. Démarrage du serveur de relais interne (port 5001)...")
|
||
REPORTS_DIR = os.path.dirname(LEGACY_REPORTS_FILE)
|
||
os.makedirs(REPORTS_DIR, exist_ok=True)
|
||
await self._start_webhook_server()
|
||
|
||
pending_count = await self._send_pending_deployment_notifications()
|
||
if pending_count:
|
||
kuby_logger.info(f"Notifications de déploiement en attente traitées au démarrage: {pending_count}")
|
||
|
||
# ⚡ Enregistrement des vues persistantes (boutons panel)
|
||
# Sans ça, après un redémarrage les boutons "Remplir le formulaire"
|
||
# dans les anciens messages de panel ne fonctionnent plus.
|
||
self.bot.add_view(BugReportView(self))
|
||
self.bot.add_view(FeatureSuggestionView(self))
|
||
kuby_logger.info("✅ Vues persistantes BugReport enregistrées (open_bug_form, open_suggestion_form)")
|
||
|
||
async def _start_webhook_server(self):
|
||
try:
|
||
app = web.Application()
|
||
app.router.add_post('/bot_event', self.handle_bot_event)
|
||
app.router.add_post('/deployment_complete', self.handle_deployment_complete)
|
||
self.runner = web.AppRunner(app)
|
||
await self.runner.setup()
|
||
self.site = web.TCPSite(self.runner, '127.0.0.1', 5001, reuse_address=True)
|
||
await self.site.start()
|
||
kuby_logger.info("Internal Bot Webhook Server started on 127.0.0.1:5001")
|
||
except Exception as e:
|
||
kuby_logger.error(f"Failed to start Internal Bot Webhook Server: {e}")
|
||
|
||
def cog_unload(self):
|
||
if hasattr(self, 'runner'):
|
||
async def cleanup():
|
||
try:
|
||
if hasattr(self, 'site'):
|
||
await self.site.stop()
|
||
await self.runner.cleanup()
|
||
kuby_logger.info("Internal Bot Webhook Server stopped and cleaned up.")
|
||
except Exception as e:
|
||
kuby_logger.error(f"Error during webhook server cleanup: {e}")
|
||
self.bot.loop.create_task(cleanup())
|
||
|
||
async def _send_pending_deployment_notifications(self) -> int:
|
||
"""Envoie les notifications de déploiement en attente depuis le tracking."""
|
||
notified_count = 0
|
||
for issue_iid, issue_data in list(self.issue_data.items()):
|
||
if not issue_data.get("pending_deployment_notification"):
|
||
kuby_logger.debug(f"No pending deployment notification for issue #{issue_iid}")
|
||
continue
|
||
|
||
requester_id = issue_data.get("requester_id")
|
||
if not requester_id:
|
||
kuby_logger.warning(f"Missing requester_id for pending deployment notification on issue #{issue_iid}")
|
||
continue
|
||
|
||
kuby_logger.info(f"Processing deployment notification for issue #{issue_iid} to requester_id={requester_id}")
|
||
user = self.bot.get_user(requester_id)
|
||
if not user:
|
||
try:
|
||
user = await self.bot.fetch_user(requester_id)
|
||
kuby_logger.info(f"Fetched user {user} for issue #{issue_iid}")
|
||
except Exception as fetch_error:
|
||
kuby_logger.warning(f"Unable to fetch user {requester_id} for issue #{issue_iid}: {fetch_error}")
|
||
continue
|
||
else:
|
||
kuby_logger.info(f"Resolved cached user {user} for issue #{issue_iid}")
|
||
|
||
try:
|
||
embed = disnake.Embed(
|
||
title="🚀 Fonctionnalité disponible !",
|
||
description=(
|
||
"✅ Votre demande est désormais fonctionnelle sur le bot.\n\n"
|
||
"Merci encore pour votre contribution et votre patience."
|
||
),
|
||
color=disnake.Color.blurple()
|
||
)
|
||
avatar = getattr(self.bot.user, "display_avatar", None)
|
||
if avatar:
|
||
embed.set_thumbnail(url=avatar.url)
|
||
|
||
kuby_logger.info(f"Attempting to DM deployment notification to {user} for issue #{issue_iid}")
|
||
await user.send(embed=embed)
|
||
issue_data["pending_deployment_notification"] = False
|
||
notified_count += 1
|
||
kuby_logger.info(f"Notification de déploiement envoyée à {user} pour l'issue #{issue_iid}")
|
||
except disnake.Forbidden as forbidden_error:
|
||
kuby_logger.warning(f"DM forbidden for {user} on issue #{issue_iid}: {forbidden_error}")
|
||
except disnake.HTTPException as http_error:
|
||
kuby_logger.warning(f"Discord HTTP error while sending deployment notification to {user} on issue #{issue_iid}: {http_error}")
|
||
except Exception as e:
|
||
kuby_logger.warning(f"Impossible d'envoyer la notification de déploiement à {user} pour l'issue #{issue_iid}: {e}")
|
||
|
||
save_tracking(self.issue_data)
|
||
kuby_logger.info(f"Déploiement terminé: {notified_count} notification(s) envoyée(s)")
|
||
return notified_count
|
||
|
||
async def handle_deployment_complete(self, request):
|
||
"""Notifie les demandeurs dont la fonctionnalité a été intégrée et attend le prochain déploiement."""
|
||
try:
|
||
try:
|
||
await request.json()
|
||
except Exception:
|
||
try:
|
||
raw_body = await request.text()
|
||
if raw_body and raw_body.strip():
|
||
json.loads(raw_body)
|
||
except Exception:
|
||
pass
|
||
|
||
notified_count = await self._send_pending_deployment_notifications()
|
||
return web.json_response({"status": "notified", "sent": notified_count}, status=200)
|
||
except Exception as e:
|
||
kuby_logger.error(f"Error handling deployment complete: {e}")
|
||
return web.json_response({"status": "error", "message": str(e)}, status=500)
|
||
|
||
async def handle_bot_event(self, request):
|
||
"""Gère les webhooks (adapté pour GitLab ET Forgejo)"""
|
||
try:
|
||
if not self.bot.is_ready():
|
||
kuby_logger.info("Webhook received but bot is not ready yet. Waiting up to 10s...")
|
||
try:
|
||
await asyncio.wait_for(self.bot.wait_until_ready(), timeout=10.0)
|
||
except asyncio.TimeoutError:
|
||
kuby_logger.warning("Bot still not ready after 10s. Proceeding anyway.")
|
||
|
||
payload = await request.json()
|
||
event_type = classify_webhook_event(payload)
|
||
|
||
if event_type not in ["issue", "note"]:
|
||
kuby_logger.info(
|
||
"Webhook ignoré : type d’événement non reconnu (keys=%s action=%s object_kind=%s)",
|
||
list(payload.keys()),
|
||
payload.get("action") or payload.get("event"),
|
||
payload.get("object_kind"),
|
||
)
|
||
return web.json_response({"status": "ignored"}, status=200)
|
||
|
||
# Récupère l'IID de l'issue
|
||
issue_iid = extract_issue_iid(payload)
|
||
|
||
if not issue_iid:
|
||
kuby_logger.warning("Webhook ignoré : aucun identifiant d’issue trouvé dans le payload")
|
||
return web.json_response({"status": "no issue id"}, status=200)
|
||
|
||
issue_data = self.issue_data.get(issue_iid)
|
||
if not issue_data:
|
||
inferred_requester_id = extract_requester_id_from_payload(payload)
|
||
if inferred_requester_id is None:
|
||
kuby_logger.warning(f"Webhook issue #{issue_iid} without tracked requester_id and no Discord ID found in payload")
|
||
return web.json_response({"status": "untracked issue"}, status=200)
|
||
|
||
self.issue_data[issue_iid] = {
|
||
"requester_id": inferred_requester_id,
|
||
"last_note_id": 0,
|
||
"type": "webhook-recovered"
|
||
}
|
||
save_tracking(self.issue_data)
|
||
issue_data = self.issue_data[issue_iid]
|
||
kuby_logger.info(f"Tracking entry recreated for issue #{issue_iid} from webhook payload")
|
||
|
||
user_id = issue_data.get("requester_id")
|
||
if not user_id:
|
||
return web.json_response({"status": "no requester"}, status=200)
|
||
|
||
user = self.bot.get_user(user_id)
|
||
if not user:
|
||
try:
|
||
user = await self.bot.fetch_user(user_id)
|
||
except:
|
||
return web.json_response({"status": "user not found"}, status=200)
|
||
|
||
# 🕵️ 3. RÉCUPÉRATION DU TITRE
|
||
issue_title = (
|
||
payload.get("object_attributes", {}).get("title") or
|
||
payload.get("issue", {}).get("title") or
|
||
payload.get("issue", {}).get("name") or
|
||
f"#{issue_iid}"
|
||
)
|
||
|
||
if event_type == "note":
|
||
# ⚡ NOUVEAU: Notifie les commentaires via le webhook (instantané)
|
||
comment_payload = payload.get("comment")
|
||
if isinstance(comment_payload, dict):
|
||
note_body = comment_payload.get("body") or comment_payload.get("content") or ""
|
||
author_name = (
|
||
payload.get("sender", {}).get("username")
|
||
or payload.get("sender", {}).get("login")
|
||
or payload.get("user", {}).get("name")
|
||
or "Développeur"
|
||
)
|
||
is_system = False
|
||
else:
|
||
note_attr = payload.get("object_attributes", {})
|
||
is_system = note_attr.get("system", False)
|
||
note_body = (
|
||
note_attr.get("note") or
|
||
note_attr.get("body") or
|
||
payload.get("note", "") or ""
|
||
)
|
||
author_name = payload.get("user", {}).get("name", "Développeur")
|
||
|
||
if not is_system and note_body:
|
||
try:
|
||
embed = disnake.Embed(
|
||
title="💬 Nouveau commentaire sur votre signalement",
|
||
description=f"Le développeur a répondu à votre rapport **{issue_title}** :\n\n>>> {note_body}\n\n✍️ **Par :** {author_name}",
|
||
color=disnake.Color.orange()
|
||
)
|
||
embed.set_thumbnail(url=self.bot.user.display_avatar.url)
|
||
await user.send(embed=embed)
|
||
kuby_logger.info(f"Commentaire envoyé à {user} pour l'issue #{issue_iid}")
|
||
except (disnake.Forbidden, disnake.HTTPException) as e:
|
||
kuby_logger.warning(f"Impossible d'envoyer le DM à {user}: {e}")
|
||
if dev and dev.id != user.id:
|
||
try:
|
||
await dev.send(f"⚠️ [DM BLOQUÉ] Impossible d'envoyer le commentaire à {user} pour l'issue #{issue_iid}")
|
||
except:
|
||
pass
|
||
|
||
elif event_type == "issue":
|
||
object_attributes = payload.get("object_attributes") or {}
|
||
issue_payload = payload.get("issue") or {}
|
||
action = (object_attributes.get("action") or issue_payload.get("action") or payload.get("action") or "").lower()
|
||
state = (object_attributes.get("state") or issue_payload.get("state") or payload.get("state") or "").lower()
|
||
label_change = extract_label_change(payload)
|
||
|
||
kuby_logger.debug(
|
||
"Issue webhook for #%s: action=%s state=%s labels=%s",
|
||
issue_iid,
|
||
action,
|
||
state,
|
||
label_change["current"] if label_change else [],
|
||
)
|
||
|
||
if label_change:
|
||
current_labels = label_change["current"]
|
||
progress = calculate_progress_from_labels(current_labels)
|
||
bar = build_progress_bar(progress)
|
||
status_message = build_status_message_from_labels(current_labels)
|
||
try:
|
||
embed = disnake.Embed(
|
||
title="📈 Évolution du ticket",
|
||
description=f"Le suivi de **{issue_title}** a été mis à jour.\n\n{status_message}",
|
||
color=color_from_progress(progress),
|
||
)
|
||
embed.add_field(name="Progression", value=bar, inline=False)
|
||
embed.set_thumbnail(url=self.bot.user.display_avatar.url)
|
||
await user.send(embed=embed)
|
||
kuby_logger.info(f"Notification de progression envoyée à {user} pour l'issue #{issue_iid}")
|
||
except Exception as e:
|
||
kuby_logger.error(f"Erreur notification progression: {e}")
|
||
|
||
is_closing_now = (
|
||
action in {"close", "closed", "closing"}
|
||
or state in {"closed", "close", "resolved", "resolved_closed"}
|
||
or payload.get("event_action") in {"close", "closed"}
|
||
or payload.get("object_kind") == "issue" and payload.get("action") in {"close", "closed"}
|
||
)
|
||
if is_closing_now:
|
||
self.issue_data[issue_iid]["pending_deployment_notification"] = True
|
||
save_tracking(self.issue_data)
|
||
kuby_logger.info(f"Marked issue #{issue_iid} as pending deployment notification")
|
||
try:
|
||
embed = disnake.Embed(
|
||
title="🛠️ Demande traitée !",
|
||
description=(
|
||
f"Bonne nouvelle ! Votre demande **{issue_title}** a été intégrée au code du bot.\n\n"
|
||
"✅ Elle n'est pas encore disponible au public, mais elle est maintenant prise en charge côté développement.\n\n"
|
||
"🔜 Lors du prochain déploiement, Kuby vous informera qu'elle est désormais fonctionnelle."
|
||
),
|
||
color=disnake.Color.green()
|
||
)
|
||
if self.bot.user.display_avatar:
|
||
embed.set_thumbnail(url=self.bot.user.display_avatar.url)
|
||
await user.send(embed=embed)
|
||
except Exception as e:
|
||
kuby_logger.error(f"Erreur notification clôture: {e}")
|
||
|
||
return web.json_response({"status": "success"}, status=200)
|
||
|
||
except Exception as e:
|
||
kuby_logger.error(f"Error handling bot event: {e}")
|
||
return web.json_response({"status": "error", "message": str(e)}, status=500)
|
||
|
||
@commands.slash_command(name="signaler_bug", description="Signaler un bug aux développeurs")
|
||
@check_premium_tier()
|
||
async def signifier_bug(self, interaction: disnake.ApplicationCommandInteraction):
|
||
modal = BugReportModal(self)
|
||
await interaction.response.send_modal(modal)
|
||
|
||
@commands.slash_command(name="suggerer_fonctionnalite", description="Proposer une nouvelle fonctionnalité")
|
||
@check_premium_tier()
|
||
async def suggerer_fonctionnalite(self, interaction: disnake.ApplicationCommandInteraction):
|
||
modal = FeatureSuggestionModal(self)
|
||
await interaction.response.send_modal(modal)
|
||
|
||
|
||
def setup(bot):
|
||
bot.add_cog(BugReport(bot))
|