Create analyze_bob_hf.py

analyze_bob_hf.py

@@ -0,0 +1,537 @@
+#!/usr/bin/env python3
+"""
+Script d'analyse des transcriptions avec Hugging Face Transformers
+Analyse les fichiers txt dans output/transcriptions et génère un résumé structuré
+"""
+
+import os
+from pathlib import Path
+import torch
+from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
+from datetime import datetime
+import re
+
+# Bootstrap environnement portable
+try:
+    from portable_env import setup_portable_env
+    setup_portable_env()
+except Exception:
+    pass
+
+# Configuration (via env, avec fallback local)
+BASE_DIR = Path(os.environ.get("BOB_BASE_DIR", Path(__file__).parent.parent))
+TRANSCRIPTIONS_DIR = Path(os.environ.get("BOB_TRANSCRIPTIONS_DIR", BASE_DIR / "output" / "transcriptions"))
+OUTPUT_FILE = Path(os.environ.get("BOB_OUTPUT_FILE", BASE_DIR / "output" / "resume_bob.txt"))
+HF_MODEL = os.environ.get("HF_MODEL", "meta-llama/Llama-3.2-1B-Instruct")  # par défaut
+
+def load_hf_model():
+    """Charge un modèle Hugging Face"""
+    try:
+        print(f"Chargement du modèle Hugging Face: {HF_MODEL}")
+        
+        # Utiliser pipeline pour plus de simplicité
+        generator = pipeline(
+            "text-generation",
+            model=HF_MODEL,
+            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+            device_map="auto" if torch.cuda.is_available() else "cpu",
+            token=os.environ.get("HF_TOKEN")  # Pour les modèles privés
+        )
+        
+        print(f"Modèle {HF_MODEL} chargé avec succès")
+        return generator
+    except Exception as e:
+        print(f"Erreur lors du chargement du modèle Hugging Face: {e}")
+        print("Assurez-vous que le modèle est disponible et que vous avez les permissions nécessaires")
+        return None
+
+def create_analysis_prompt():
+    """Crée le prompt d'analyse"""
+    return """RÔLE: Expert en classification de contenu journalistique RTL.
+
+TÂCHE: Extraire 3 informations précises de cette transcription radio :
+
+1. AUTEUR : Nom complet du journaliste/présentateur
+   - Chercher "les précisions pour RTL de [NOM]" ou signature en fin
+   - Si absent : "Inconnu"
+
+2. QUALIFICATION du format (TRÈS IMPORTANT) :
+   - P = PAPIER seul : Lecture continue par le journaliste, pas d'interviews
+     • Phrases à la 3e personne uniquement
+     • Aucune citation directe de témoins
+     • Style narratif/descriptif pur
+   
+   - P+S = PAPIER + SON : Reportage avec interviews/témoignages
+     • Présence de citations directes ("Je...", "Nous...")
+     • Témoignages de personnes citées par leur prénom
+     • Alternance narratif + paroles rapportées
+     • Phrases comme "explique Alexandre", "témoigne Lucas"
+   
+   - QR = QUESTIONS-RÉPONSES : Interview/débat en direct
+     • Format conversationnel
+     • Questions-réponses explicites
+     • Dialogue en temps réel
+
+3. TITRE : Sujet principal en 4-6 mots, MAJUSCULES, style presse
+
+INDICES DE DÉTECTION P+S :
+- Citations à la 1ère personne : "J'ai été hospitalisé", "Nous avons commencé"
+- Prénoms + témoignages : "Alexandre explique", "Lucas raconte"
+- Discours rapporté : "Il dit que", "Ils nous ont dit"
+- Changement de ton narratif
+
+FORMAT OBLIGATOIRE :
+AUTEUR|QUALIFICATION|TITRE"""
+
+def detect_format_indicators(text):
+    """Détecte automatiquement les indicateurs de format P/P+S/QR/MT"""
+    indicators = {
+        'p_plus_s': 0,  # Papier + Son
+        'qr': 0,        # Questions-Réponses  
+        'mt': 0,        # Micro-Trottoir
+        'p_only': 0     # Papier seul
+    }
+    
+    text_lower = text.lower()
+    
+    # Indicateurs Micro-Trottoir (MT)
+    mt_patterns = [
+        r'\bmoi je (?:trouve|pense|crois|dis)',
+        r'\bje trouve (?:que|ça|dommage)',
+        r'\bje pense que',
+        r'\bpour moi',
+        r'\bà mon avis',
+        r'\bfranchement',
+        r'en arrivant',
+        r'je viens (?:de|d\')',
+        r'aujourd\'hui',
+        r'c\'est dommage',
+        r'malheureusement',
+        r'donc je (?:voulais|pense)',
+        r'quand même',
+        r'un petit peu',
+        r'vraiment dommage',
+    ]
+    
+    # Indicateurs P+S (Papier + Son)
+    p_plus_s_patterns = [
+        r'\bje\s+(?:suis|ai|me|pense|crois|vais|veux|dois)',
+        r'\bnous\s+(?:avons|sommes|étions|allons|devons)',
+        r'\bj\'(?:ai|étais|avais|irai|aurais)',
+        r'\bmon\s+(?:père|fils|mari|frère)',
+        r'\bma\s+(?:mère|fille|femme|sœur)',
+        r'\b(?:explique|témoigne|raconte|confie|précise|ajoute|poursuit)\s+\w+',
+        r'\b\w+\s+(?:explique|témoigne|raconte|confie|précise|ajoute|poursuit)',
+        r'selon\s+\w+',
+        r'(?:il|elle|ils|elles)\s+(?:dit|disent|explique|expliquent|affirme|assure)\s+que',
+        r'pour\s+\w+\s*,',
+        r'comme\s+(?:le\s+)?(?:dit|explique|précise)\s+\w+',
+        r'voilà ce à quoi',
+        r'c\'est qu?\'?à? partir',
+        r'certains d\'entre (?:nous|eux)',
+        r'parmi les\s+\d+',
+        r'\b[A-Z][a-z]+\s+qui\s+(?:est|a|était)',
+        r'comme\s+[A-Z][a-z]+',
+        r'fièvre\s+et\s+\w+',
+        r'hospitalisé',
+        r'symptômes',
+        r'malade',
+    ]
+    
+    # Indicateurs QR (Questions-Réponses)
+    qr_patterns = [
+        r'\?.*[A-Z]',  # Question suivie de réponse
+        r'question\s*:',
+        r'réponse\s*:',
+        r'vous\s+(?:pensez|croyez|dites)',
+        r'que\s+pensez-vous',
+        r'interview',
+        r'débat',
+    ]
+    
+    # Compter les patterns
+    for pattern in mt_patterns:
+        indicators['mt'] += len(re.findall(pattern, text_lower, re.IGNORECASE))
+    
+    for pattern in p_plus_s_patterns:
+        indicators['p_plus_s'] += len(re.findall(pattern, text_lower, re.IGNORECASE))
+    
+    for pattern in qr_patterns:
+        indicators['qr'] += len(re.findall(pattern, text_lower, re.IGNORECASE))
+    
+    # Si ni MT ni P+S ni QR détecté fortement, c'est probablement P seul
+    if indicators['mt'] < 3 and indicators['p_plus_s'] < 3 and indicators['qr'] < 2:
+        indicators['p_only'] = 5
+    
+    return indicators
+
+def analyze_transcription(generator, transcription_text, filename):
+    """Analyse une transcription avec Hugging Face"""
+    try:
+        # Analyse automatique des patterns
+        format_indicators = detect_format_indicators(transcription_text)
+        
+        # Créer un hint pour l'IA basé sur l'analyse automatique
+        max_score = max(format_indicators.values())
+        likely_format = [k for k, v in format_indicators.items() if v == max_score][0]
+        
+        hint_map = {
+            'p_plus_s': "ATTENTION: Nombreux témoignages détectés → OBLIGATOIREMENT P+S",
+            'qr': "ATTENTION: Format questions-réponses détecté → OBLIGATOIREMENT QR", 
+            'p_only': "ATTENTION: Aucun témoignage/interview → OBLIGATOIREMENT P"
+        }
+        
+        # Si détection très claire (score élevé), forcer le format
+        force_format = ""
+        if format_indicators['p_plus_s'] >= 5:
+            force_format = "\nFORMAT IMPOSÉ: Utilise OBLIGATOIREMENT 'P+S' pour la qualification."
+        elif format_indicators['qr'] >= 3:
+            force_format = "\nFORMAT IMPOSÉ: Utilise OBLIGATOIREMENT 'QR' pour la qualification."
+        elif format_indicators['p_plus_s'] <= 1 and format_indicators['qr'] <= 1:
+            force_format = "\nFORMAT IMPOSÉ: Utilise OBLIGATOIREMENT 'P' pour la qualification."
+        
+        format_hint = hint_map.get(likely_format, "")
+        
+        prompt = create_analysis_prompt()
+        enhanced_prompt = f"{prompt}\n\n{format_hint}{force_format}"
+        
+        # Format pour Llama
+        full_prompt = f"<|begin_of_text|><|start_header_id|>user<|end_header_id|>\n\n{enhanced_prompt}\n\nTRANSCRIPTION:\n{transcription_text}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n"
+        
+        print(f"Analyse de: {filename}")
+        print(f"  Indices détectés: MT={format_indicators['mt']}, P+S={format_indicators['p_plus_s']}, QR={format_indicators['qr']}, P={format_indicators['p_only']}")
+        
+        # Génération avec le modèle
+        response = generator(
+            full_prompt,
+            max_new_tokens=150,
+            temperature=0.2,
+            top_k=20,
+            top_p=0.8,
+            repetition_penalty=1.15,
+            do_sample=True,
+            pad_token_id=generator.tokenizer.eos_token_id
+        )
+        
+        result = response[0]['generated_text'].replace(full_prompt, '').strip()
+        
+        # Parser le résultat
+        result = result.replace('\n', ' ').strip()
+        
+        if "|" in result:
+            # Prendre la première ligne qui contient des |
+            lines = result.split('\n')
+            for line in lines:
+                if "|" in line and line.count("|") >= 2:
+                    parts = line.split("|")
+                    if len(parts) >= 3:
+                        auteur = parts[0].strip()
+                        qualification = parts[1].strip()
+                        titre = parts[2].strip()
+                        
+                        # Nettoyer et formater
+                        if auteur.lower() in ["inconnu", "non mentionné", "auteur", ""]:
+                            auteur = "Inconnu"
+                        
+                        # Valider la qualification
+                        if qualification.upper() not in ["P", "P+S", "SON", "MT", "QR"]:
+                            qualification = "P"  # Défaut
+                        
+                        return {
+                            "success": True,
+                            "auteur": auteur,
+                            "qualification": qualification.upper(),
+                            "titre": titre.upper(),
+                            "filename": filename
+                        }
+        
+        # Si pas de format avec |, essayer de parser différemment
+        if " - " in result:
+            parts = result.split(" - ")
+            if len(parts) >= 3:
+                auteur = parts[0].strip()
+                qualification = parts[1].strip()
+                titre = " - ".join(parts[2:]).strip()
+                
+                if auteur.lower() in ["inconnu", "non mentionné", "auteur", ""]:
+                    auteur = "Inconnu"
+                
+                # Valider la qualification
+                if qualification.upper() not in ["P", "P+S", "QR"]:
+                    qualification = "P"
+                
+                return {
+                    "success": True,
+                    "auteur": auteur,
+                    "qualification": qualification.upper(),
+                    "titre": titre.upper(),
+                    "filename": filename
+                }
+        
+        # Si le format n'est pas correct
+        return {
+            "success": False,
+            "error": f"Format de réponse incorrect: {result}",
+            "filename": filename,
+            "raw_response": result
+        }
+        
+    except Exception as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "filename": filename
+        }
+
+def read_transcription_file(file_path):
+    """Lit le contenu d'un fichier de transcription et extrait les métadonnées"""
+    try:
+        with open(file_path, 'r', encoding='utf-8') as f:
+            content = f.read()
+        
+        # Extraire les métadonnées
+        metadata = {}
+        lines = content.split('\n')
+        
+        for line in lines[:10]:  # Chercher dans les 10 premières lignes
+            if line.startswith('Fichier source:'):
+                metadata['filename'] = line.replace('Fichier source:', '').strip()
+            elif line.startswith('Durée de traitement:'):
+                metadata['processing_time'] = line.replace('Durée de traitement:', '').strip()
+        
+        # Extraire seulement le texte de transcription (après les métadonnées)
+        if "--------------------------------------------------" in content:
+            parts = content.split("--------------------------------------------------")
+            if len(parts) > 1:
+                text_content = parts[1].strip()
+                return text_content, metadata
+        
+        return content.strip(), metadata
+        
+    except Exception as e:
+        print(f"Erreur lors de la lecture de {file_path}: {e}")
+        return None, {}
+
+def apply_duration_correction(result, duration_seconds, format_indicators=None):
+    """Applique une correction probabiliste basée sur la durée et les patterns détectés"""
+    if not duration_seconds:
+        return result
+    
+    original_qualification = result.get("qualification", "")
+    corrected = False
+    
+    # Priorité 1: Détection Micro-Trottoir basée sur patterns + durée
+    if format_indicators and format_indicators.get('mt', 0) >= 8 and duration_seconds < 60:
+        if original_qualification in ["P", "P+S", "SON"]:
+            result["qualification"] = "MT"
+            corrected = True
+            print(f"    → Correction MT détecté: {original_qualification} → MT (patterns={format_indicators['mt']})")
+    
+    # Priorité 2: Logique probabiliste selon durée (si pas MT)
+    elif not corrected:
+        if duration_seconds < 30:
+            # < 30s = quasi-certainement un SON
+            if original_qualification in ["P", "P+S"]:
+                result["qualification"] = "SON"
+                corrected = True
+                print(f"    → Correction durée < 30s: {original_qualification} → SON")
+        
+        elif 30 <= duration_seconds <= 40:
+            # 30-40s = probablement un SON, mais peut être P
+            if original_qualification == "P+S":
+                result["qualification"] = "SON"
+                corrected = True
+                print(f"    → Correction durée 30-40s: P+S → SON")
+    
+    return result
+
+def extract_author_from_filename(filename):
+    """Extrait le nom du journaliste depuis le nom du fichier"""
+    try:
+        # Nettoyer le nom du fichier
+        clean_name = filename.replace('_transcription.txt', '').replace('.mp3', '').replace('.MP3', '')
+        
+        # Patterns courants pour extraire un nom (prénom + nom)
+        words = clean_name.split()
+        
+        # Chercher une séquence de 2 mots qui commencent par une majuscule
+        for i in range(len(words) - 1):
+            word1 = words[i].strip('()[]{}.,;:!?-_')
+            word2 = words[i + 1].strip('()[]{}.,;:!?-_')
+            
+            # Vérifier si les deux mots ressemblent à un prénom + nom
+            if (len(word1) >= 2 and len(word2) >= 2 and 
+                word1[0].isupper() and word2[0].isupper() and
+                word1.isalpha() and word2.isalpha()):
+                return f"{word1} {word2}"
+        
+        # Si pas trouvé, chercher le premier mot qui commence par une majuscule
+        for word in words:
+            clean_word = word.strip('()[]{}.,;:!?-_')
+            if len(clean_word) >= 2 and clean_word[0].isupper() and clean_word.isalpha():
+                # Essayer de trouver le mot suivant
+                word_index = words.index(word)
+                if word_index + 1 < len(words):
+                    next_word = words[word_index + 1].strip('()[]{}.,;:!?-_')
+                    if len(next_word) >= 2 and next_word[0].isupper() and next_word.isalpha():
+                        return f"{clean_word} {next_word}"
+                return clean_word
+        
+        return "Inconnu"
+        
+    except Exception as e:
+        print(f"Erreur extraction auteur de {filename}: {e}")
+        return "Inconnu"
+
+def get_audio_duration(audio_filename, input_dir):
+    """Calcule la durée d'un fichier audio en secondes totales"""
+    try:
+        from pydub import AudioSegment
+        audio_path = None
+        audio_extensions = ['.mp3', '.wav', '.m4a', '.flac', '.ogg', '.mp4', '.avi', '.mov']
+        # Rechercher le fichier audio correspondant
+        for ext in audio_extensions:
+            potential_path = input_dir / audio_filename.replace('_transcription.txt', ext)
+            if potential_path.exists():
+                audio_path = potential_path
+                break
+            base_name = audio_filename.replace('_transcription.txt', '')
+            potential_path = input_dir / f"{base_name}{ext}"
+            if potential_path.exists():
+                audio_path = potential_path
+                break
+        if audio_path:
+            audio = AudioSegment.from_file(str(audio_path))
+            duration_seconds = len(audio) / 1000  # pydub retourne en millisecondes
+            minutes = int(duration_seconds // 60)
+            seconds = int(duration_seconds % 60)
+            return minutes * 100 + seconds
+        return None
+    except Exception as e:
+        print(f"Erreur calcul durée pour {audio_filename}: {e}")
+        return None
+
+def get_transcription_files(transcriptions_dir):
+    """Récupère tous les fichiers de transcription"""
+    if not transcriptions_dir.exists():
+        print(f"Le dossier {transcriptions_dir} n'existe pas")
+        return []
+    txt_files = list(transcriptions_dir.glob("*_transcription.txt"))
+    return sorted(txt_files)
+
+def main():
+    """Fonction principale"""
+    print("=" * 60)
+    print("ANALYSE DES BOB AVEC HUGGING FACE")
+    print("=" * 60)
+    
+    # Vérification des dossiers
+    script_dir = Path(__file__).parent.absolute()
+    transcriptions_dir = Path(os.environ.get("BOB_TRANSCRIPTIONS_DIR", script_dir.parent / "output" / "transcriptions"))
+    output_file = Path(os.environ.get("BOB_OUTPUT_FILE", script_dir.parent / "output" / "resume_bob.txt"))
+    input_dir = Path(os.environ.get("BOB_INPUT_DIR", script_dir.parent / "input"))
+
+    # Utiliser la fonction factorisée avec print comme log
+    analyze_files_hf(
+        transcriptions_dir=transcriptions_dir,
+        input_dir=input_dir,
+        output_file=output_file,
+        log_fn=print,
+        progress_fn=None,
+        cancel_fn=None,
+    )
+
+if __name__ == "__main__":
+    main()
+
+# --- API factorisée pour le GUI ---
+def analyze_files_hf(transcriptions_dir: Path, input_dir: Path, output_file: Path, log_fn=print, progress_fn=None, cancel_fn=None):
+    """Analyse tous les fichiers de transcription avec Hugging Face"""
+    log = log_fn or (lambda *a, **k: None)
+
+    log("Dossier transcriptions: {}".format(transcriptions_dir))
+    log("Fichier de sortie: {}".format(output_file))
+    log("")
+
+    transcription_files = get_transcription_files(transcriptions_dir)
+    if not transcription_files:
+        log("Aucun fichier de transcription trouvé")
+        log("Assurez-vous d'avoir exécuté le script de transcription d'abord")
+        return {"success": False, "count": 0}
+
+    log(f"Trouvé {len(transcription_files)} fichier(s) de transcription:")
+    for i, file in enumerate(transcription_files, 1):
+        log(f"  {i}. {file.name}")
+    log("")
+
+    # Initialisation du modèle Hugging Face
+    generator = load_hf_model()
+    if not generator:
+        return {"success": False, "error": "Modèle Hugging Face indisponible"}
+
+    results = []
+    success_count = 0
+    total = len(transcription_files)
+
+    for i, file_path in enumerate(transcription_files, 1):
+        if cancel_fn and cancel_fn():
+            log("⏹️ Analyse annulée")
+            break
+
+        log(f"[{i}/{total}] ")
+
+        transcription_text, metadata = read_transcription_file(file_path)
+        if not transcription_text:
+            log(f"✗ Impossible de lire {file_path.name}")
+            if progress_fn:
+                progress_fn(i, total)
+            continue
+
+        duration = get_audio_duration(file_path.name, input_dir)
+        author_from_filename = extract_author_from_filename(file_path.name)
+
+        result = analyze_transcription(generator, transcription_text, file_path.name)
+
+        if result["success"]:
+            format_indicators = detect_format_indicators(transcription_text)
+            result = apply_duration_correction(result, duration, format_indicators)
+            if result["auteur"].lower() in ["inconnu", "non mentionné", "auteur", ""]:
+                result["auteur"] = author_from_filename
+            result["duree"] = duration if duration else "000"
+            result["filename_source"] = metadata.get("filename", file_path.name)
+            log(f"✓ {result['auteur']} - {result['qualification']} - {result['titre']} - {result['duree']}")
+            results.append(result)
+            success_count += 1
+        else:
+            log(f"✗ Erreur: {result['error']}")
+            if "raw_response" in result:
+                log(f"  Réponse brute: {result['raw_response']}")
+
+        if progress_fn:
+            progress_fn(i, total)
+        log("")
+
+    if results:
+        output_file.parent.mkdir(parents=True, exist_ok=True)
+        with open(output_file, 'w', encoding='utf-8') as f:
+            f.write(f"# RÉSUMÉ DES BOB - {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}\n")
+            f.write(f"# Format: Auteur | Qualification | Titre | Durée\n")
+            f.write("# Qualification: P=papier, P+S=papier+son, QR=question-réponse\n")
+            f.write("# Durée: format MMss (ex: 1min04 = 104)\n")
+            f.write("# " + "="*70 + "\n\n")
+            for r in results:
+                line = f"{r['auteur']} | {r['qualification']} | {r['titre']} | {r['duree']}"
+                f.write(line + "\n")
+
+        log("=" * 60)
+        log("RÉSUMÉ GÉNÉRÉ")
+        log("=" * 60)
+        log(f"Fichiers analysés: {total}")
+        log(f"Analyses réussies: {success_count}")
+        log(f"Analyses échouées: {total - success_count}")
+        log(f"Fichier de résumé: {output_file}")
+
+        return {"success": True, "count": total, "ok": success_count, "results": results}
+    else:
+        log("Aucune analyse réussie, pas de fichier de résumé généré")
+        return {"success": False, "count": total, "ok": 0}