#!/usr/bin/env python3 """ Script de récupération des commentaires GitLab pour le bot Kuby. Ce script permet de : 1. Récupérer toutes les issues suivies depuis gitlab_issues.json et data/gitlab_reports.json 2. Pour chaque issue, récupérer TOUS les commentaires via l'API GitLab 3. Reconstruire gitlab_issues.json avec les bons last_note_id 4. Afficher un rapport des commentaires manquants Usage: python scripts/recover_gitlab_comments.py Requires: - GITLAB_TOKEN (env var) - GITLAB_PROJECT_ID (env var) - aiohttp """ import os import json import asyncio import sys from datetime import datetime from typing import Dict, List, Any, Optional # Chemin des fichiers de tracking BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) GITLAB_ISSUES_FILE = os.path.join(BASE_DIR, "commandes", "gitlab_issues.json") GITLAB_REPORTS_FILE = os.path.join(BASE_DIR, "data", "gitlab_reports.json") # Configuration GitLab GITLAB_TOKEN = os.getenv("GITLAB_TOKEN") GITLAB_PROJECT_ID = os.getenv("GITLAB_PROJECT_ID") if not GITLAB_TOKEN or not GITLAB_PROJECT_ID: print("❌ ERREUR : Les variables d'environnement GITLAB_TOKEN et GITLAB_PROJECT_ID doivent être définies.") sys.exit(1) GITLAB_API_URL = f"https://gitlab.com/api/v4/projects/{GITLAB_PROJECT_ID}" async def fetch_all_notes(issue_iid: int) -> List[Dict[str, Any]]: """Récupère tous les commentaires d'une issue via l'API GitLab.""" url = f"{GITLAB_API_URL}/issues/{issue_iid}/notes?per_page=100" headers = {"PRIVATE-TOKEN": GITLAB_TOKEN} all_notes = [] page = 1 while True: try: import aiohttp async with aiohttp.ClientSession() as session: async with session.get(f"{url}&page={page}", headers=headers) as resp: if resp.status != 200: print(f" ⚠️ Erreur API pour l'issue #{issue_iid} (page {page}): {resp.status}") break notes = await resp.json() if not notes: break all_notes.extend(notes) # Vérifie s'il y a plus de pages if len(notes) < 100: break page += 1 except Exception as e: print(f" ❌ Erreur lors de la récupération des notes pour l'issue #{issue_iid}: {e}") break return all_notes def load_existing_tracking() -> Dict[str, Any]: """Charge les données de suivi existantes.""" tracking_data = {} # Charge gitlab_issues.json if os.path.exists(GITLAB_ISSUES_FILE): try: with open(GITLAB_ISSUES_FILE, "r", encoding="utf-8") as f: tracking_data = json.load(f) print(f"✅ Chargé {len(tracking_data)} issues depuis gitlab_issues.json") except Exception as e: print(f"⚠️ Erreur lors du chargement de gitlab_issues.json: {e}") # Charge gitlab_reports.json et fusionne if os.path.exists(GITLAB_REPORTS_FILE): try: with open(GITLAB_REPORTS_FILE, "r", encoding="utf-8") as f: reports_data = json.load(f) for issue_iid, user_id in reports_data.items(): if issue_iid not in tracking_data: tracking_data[issue_iid] = { "requester_id": user_id, "last_note_id": 0, "type": "manual-report" } print(f" + Ajout de l'issue #{issue_iid} depuis gitlab_reports.json") print(f"✅ Fusionné {len(reports_data)} issues depuis gitlab_reports.json") except Exception as e: print(f"⚠️ Erreur lors du chargement de gitlab_reports.json: {e}") return tracking_data def save_tracking_data(data: Dict[str, Any]) -> None: """Sauvegarde les données de suivi dans gitlab_issues.json.""" os.makedirs(os.path.dirname(GITLAB_ISSUES_FILE), exist_ok=True) # Sauvegarde dans un fichier temporaire puis renomme temp_file = f"{GITLAB_ISSUES_FILE}.tmp" with open(temp_file, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(temp_file, GITLAB_ISSUES_FILE) print(f"✅ Sauvegardé {len(data)} issues dans gitlab_issues.json") async def recover_comments(): """Fonction principale de récupération.""" print("=" * 60) print("🔍 RÉCUPÉRATION DES COMMENTAIRES GITLAB") print("=" * 60) # Charge les données existantes print("\n📂 Chargement des données de suivi existantes...") tracking_data = load_existing_tracking() if not tracking_data: print("❌ Aucune issue à suivre trouvée. Vérifiez que les fichiers existent.") return print(f"\n🔄 Traitement de {len(tracking_data)} issues...") print("-" * 60) issues_with_new_comments = [] total_notes = 0 total_missing = 0 for issue_iid_str, info in tracking_data.items(): issue_iid = int(issue_iid_str) last_known_id = info.get("last_note_id", 0) requester_id = info.get("requester_id", "Inconnu") print(f"\n📝 Issue #{issue_iid} (demandeur: {requester_id})") print(f" Dernier note_id connu: {last_known_id}") # Récupère tous les commentaires notes = await fetch_all_notes(issue_iid) if not notes: print(f" ⚠️ Aucune note trouvée pour cette issue.") continue # Trie les notes par ID notes.sort(key=lambda n: n["id"]) latest_note_id = max(n["id"] for n in notes) total_notes += len(notes) print(f" ✅ {len(notes)} commentaires trouvés (dernier ID: {latest_note_id})") # Vérifie s'il y a des commentaires manquants if last_known_id == 0: # Premier lancement ou fichier réinitialisé missing_count = len(notes) total_missing += missing_count print(f" ⚠️ {missing_count} commentaires NON SUIVIS (fichier réinitialisé)") issues_with_new_comments.append({ "issue_iid": issue_iid, "missing_notes": notes, "requester_id": requester_id, "new_last_note_id": latest_note_id }) else: # Vérifie quels commentaires sont nouveaux missing_notes = [n for n in notes if n["id"] > last_known_id] if missing_notes: missing_count = len(missing_notes) total_missing += missing_count print(f" ⚠️ {missing_count} NOUVEAUX commentaires détectés") issues_with_new_comments.append({ "issue_iid": issue_iid, "missing_notes": missing_notes, "requester_id": requester_id, "new_last_note_id": latest_note_id }) else: print(f" ✅ Aucun nouveau commentaire") # Met à jour le last_note_id tracking_data[issue_iid_str]["last_note_id"] = latest_note_id # Sauvegarde les données mises à jour print("\n" + "=" * 60) print("💾 SAUVEGARDE DES DONNÉES") print("=" * 60) save_tracking_data(tracking_data) # Génère le rapport print("\n" + "=" * 60) print("📊 RAPPORT DE RÉCUPÉRATION") print("=" * 60) print(f"Total issues traitées: {len(tracking_data)}") print(f"Total commentaires récupérés: {total_notes}") print(f"Total commentaires manquants: {total_missing}") if issues_with_new_comments: print(f"\n📋 Issues avec des commentaires non suivis ({len(issues_with_new_comments)}):") for item in issues_with_new_comments: print(f"\n Issue #{item['issue_iid']}:") print(f" Demandeur: {item['requester_id']}") print(f" Nombre de commentaires manquants: {len(item['missing_notes'])}") print(f" Nouveau last_note_id: {item['new_last_note_id']}") # Affiche les commentaires manquants for note in item["missing_notes"]: author = note.get("author", {}).get("name", "Inconnu") created_at = note.get("created_at", "Inconnu") body = note.get("body", "")[:100] + "..." if len(note.get("body", "")) > 100 else note.get("body", "") print(f" - [{created_at}] {author}: {body}") else: print("\n✅ Aucune issue avec des commentaires manquants !") print("\n" + "=" * 60) print("✅ RÉCUPÉRATION TERMINÉE") print("=" * 60) print("\nProchaines étapes:") print("1. Vérifiez le fichier commandes/gitlab_issues.json") print("2. Redémarrez le bot pour qu'il utilise les nouvelles données") print("3. Les nouveaux commentaires seront désormais suivis correctement") if __name__ == "__main__": print("\n" + "=" * 60) print("SCRIPT DE RÉCUPÉRATION DES COMMENTAIRES GITLAB") print("=" * 60) print(f"Projet GitLab: {GITLAB_PROJECT_ID}") print(f"Fichier de sortie: {GITLAB_ISSUES_FILE}") print("=" * 60 + "\n") asyncio.run(recover_comments())