Update app.py

app.py

@@ -1,211 +1,346 @@
 # -*- coding: utf-8 -*-
 """
-Antigravity Shield v5.0 — ACoNum / Trusted Sound 2026
+analyze_media.py — Pont Python AI pour MAM SHIELD v5.2
+ACoNum / Trusted Sound 2026
 =====================================================
-Fusion : AASIST neural anti-spoofing + Jitter/Flatness + AcoustID
-Résout :
-  - Faux positifs sur voix off (is.mp3)
-  - Anciennes chansons (vintage/radio)
-  - AI Cover complexes (ElevenLabs, RVC)
-  - Broadcast FM (jitter élevé naturel)
+Usage (appelé par server.js) :
+  python analyze_media.py audio   <chemin_fichier>
+  python analyze_media.py video   <chemin_fichier>
+  python analyze_media.py image   <chemin_fichier>
+  python analyze_media.py pdf     <chemin_fichier>
+
+Retourne un JSON sur stdout :
+  { "summary": "...", "tags": [...], "transcript": "...", "flac_ok": true, "flac_path": "..." }
 """
 
-import numpy as np
-import librosa
-import librosa.display
-import matplotlib.pyplot as plt
-import noisereduce as nr
-import hashlib
+import sys
 import os
 import json
 import subprocess
-import gradio as gr
-
-# ══════════════════════════════════════════════════════════════
-#  CONFIG
-# ══════════════════════════════════════════════════════════════
-
-ACOUSTID_KEY = "TY6HUQsigs"
-AASIST_MODEL_ID = "Mahmoud-Yassen/aasist-antispoof"
-
-# Seuils calibrés pour ACoNum 2026
-JITTER_BROADCAST_MIN = 0.85
-FLATNESS_THRESHOLD   = 0.0012
-FLATNESS_BROADCAST   = 0.0010
-CENTROID_BROADCAST   = 3500.0
-CENTROID_VINTAGE     = 3000.0
-ZCR_SPEECH_THRESHOLD = 0.055  # Détection voix off/parlée
-ZCR_VINTAGE          = 0.06
-
-class AntigravityShield:
-    def __init__(self):
-        print("🚀 Antigravity Shield v5.0 — AASIST + Jitter + AcoustID")
-        self.ast_model  = None
-        self.aasist     = None
-        self.aasist_ok  = False
-
-    def load_ast(self):
-        if self.ast_model is None:
-            try:
-                from transformers import pipeline
-                self.ast_model = pipeline("audio-classification", model="MIT/ast-finetuned-audioset-10-10-0.4593")
-                print("✅ AST chargé")
-            except Exception as e:
-                print(f"⚠️ AST non disponible : {e}")
-
-    def load_aasist(self):
-        if self.aasist is None:
-            try:
-                from transformers import pipeline
-                candidates = ["Mahmoud-Yassen/aasist-antispoof", "m-aliabbas/AASIST"]
-                for model_id in candidates:
-                    try:
-                        self.aasist = pipeline("audio-classification", model=model_id, sampling_rate=16000)
-                        self.aasist_ok = True
-                        print(f"✅ AASIST chargé : {model_id}")
-                        break
-                    except: continue
-            except Exception as e:
-                print(f"⚠️ AASIST erreur : {e}")
-
-    def get_sha256(self, path):
-        with open(path, "rb") as f:
-            return hashlib.sha256(f.read()).hexdigest()
-
-    def acoustid_lookup(self, path):
-        try:
-            import acoustid
-            results = acoustid.match(ACOUSTID_KEY, path)
-            for score, recording_id, title, artist in results:
-                if score > 0.8: return True, f"{artist} — {title}", recording_id
-            return False, "Non identifié", None
-        except:
-            return None, "AcoustID non dispo", None
-
-    def generer_spectrogramme(self, y, sr, jitter_a, jitter_b, aasist_score):
-        fig, axes = plt.subplots(1, 2, figsize=(14, 4))
-        S = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=128)
-        S_dB = librosa.power_to_db(S, ref=np.max)
-        librosa.display.specshow(S_dB, sr=sr, x_axis='time', y_axis='mel', cmap='magma', ax=axes[0])
-        axes[0].set_title('Texture Vocale (Mel)')
-        
-        # Timeline simplifiée
-        axes[1].axhline(y=jitter_b, color='yellow', linestyle='--', label=f'Jitter-B: {jitter_b:.3f}')
-        if aasist_score: axes[1].axhline(y=aasist_score, color='red', linestyle=':', label=f'AASIST: {aasist_score:.2f}')
-        axes[1].set_title('Métriques de Stabilité')
-        axes[1].legend()
+import traceback
+
+# ── Config Ollama ────────────────────────────────────────────
+OLLAMA_HOST  = 'http://localhost:11434'
+OLLAMA_MODEL = 'qwen2.5:3b'   # tags + résumé standard
+
+# ── Nemotron — auto-détection au démarrage ───────────────
+# Modèle téléchargé : nvidia/Nemotron-3-Nano-Omni-30B → version 4B locale
+# Nom Ollama attendu : nemotron-mini ou nemotron-mini:4b
+NEMOTRON_MODEL = None
+NEMOTRON_PRIORITY = [          # ordre de priorité exact
+    'nemotron-mini:4b',
+    'nemotron-mini:latest',
+    'nemotron-mini',
+    'nemotron:4b',
+    'nemotron:mini',
+    'nemotron:latest',
+    'nvidia/nemotron-mini',
+    'nemotron-nano',
+    'nemotron',
+]
+
+def detect_nemotron():
+    """Cherche automatiquement un modèle Nemotron dans Ollama (priorité 4B mini)."""
+    global NEMOTRON_MODEL
+    try:
+        import urllib.request as ur
+        req = ur.Request(OLLAMA_HOST + '/api/tags', method='GET')
+        with ur.urlopen(req, timeout=5) as r:
+            data   = json.loads(r.read().decode('utf-8'))
+            models = [m.get('name','') for m in data.get('models', [])]
+            # Cherche par priorité exacte d'abord
+            for prio in NEMOTRON_PRIORITY:
+                if prio in models:
+                    NEMOTRON_MODEL = prio
+                    return prio
+            # Sinon: correspondance partielle
+            for m in models:
+                if 'nemotron' in m.lower():
+                    NEMOTRON_MODEL = m
+                    return m
+    except Exception:
+        pass
+    NEMOTRON_MODEL = OLLAMA_MODEL   # fallback qwen si absent
+    return NEMOTRON_MODEL
+
+detect_nemotron()
+
+# ── Config FFmpeg ────────────────────────────────────────────
+def find_ffmpeg():
+    candidates = [
+        os.path.join(os.path.expanduser('~'), 'ffmpeg', 'bin', 'ffmpeg.exe'),
+        os.path.join(os.path.expanduser('~'), 'AppData', 'Local', 'Programs', 'ffmpeg', 'bin', 'ffmpeg.exe'),
+        'ffmpeg',
+    ]
+    for c in candidates:
+        if c == 'ffmpeg':
+            return c
+        if os.path.exists(c):
+            return c
+    return 'ffmpeg'
+
+FFMPEG = find_ffmpeg()
+
+# ── Résultat ─────────────────────────────────────────────────
+def ok(**kwargs):
+    print(json.dumps({ 'ok': True, **kwargs }, ensure_ascii=False))
+    sys.exit(0)
+
+def fail(msg):
+    print(json.dumps({ 'ok': False, 'error': str(msg) }, ensure_ascii=False))
+    sys.exit(1)
+
+
+# ════════════════════════════════════════════════════════════
+#  OLLAMA — appel local
+# ════════════════════════════════════════════════════════════
+def ollama_generate(prompt, model=OLLAMA_MODEL):
+    try:
+        import urllib.request
+        body = json.dumps({
+            'model':   model,
+            'prompt':  prompt,
+            'stream':  False,
+            'options': { 'temperature': 0, 'num_predict': 300 }
+        }).encode('utf-8')
+        req = urllib.request.Request(
+            OLLAMA_HOST + '/api/generate',
+            data=body,
+            headers={ 'Content-Type': 'application/json' },
+            method='POST'
+        )
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            data = json.loads(resp.read().decode('utf-8'))
+            return data.get('response', '').strip()
+    except Exception as e:
+        return None
+
+
+def generate_tags(name, media_type, content=''):
+    raw = ollama_generate(
+        f"Tu es un archiviste média. Génère exactement 5 tags pertinents pour cet asset.\n"
+        f"Nom: {name}\nType: {media_type}\nContenu: {content[:300] if content else 'non disponible'}\n"
+        f"Réponds UNIQUEMENT avec un tableau JSON de 5 strings en français. "
+        f"Ex: [\"tag1\",\"tag2\",\"tag3\",\"tag4\",\"tag5\"]"
+    )
+    if not raw:
+        return []
+    try:
+        import re
+        m = re.search(r'\[.*?\]', raw, re.DOTALL)
+        return json.loads(m.group(0)) if m else []
+    except Exception:
+        return []
+
+
+def generate_summary(name, media_type, content=''):
+    return ollama_generate(
+        f"Tu es un archiviste média professionnel. Rédige une description courte (2-3 phrases) pour:\n"
+        f"Nom: {name}\nType: {media_type}\nContenu: {content[:500] if content else 'non disponible'}\n"
+        f"Réponds uniquement avec la description en français."
+    )
+
+
+# ════════════════════════════════════════════════════════════
+#  AUDIO — Whisper + conversion FLAC TC-04
+# ════════════════════════════════════════════════════════════
+def analyze_audio(file_path):
+    transcript = ''
+    flac_ok    = False
+    flac_path  = ''
+
+    # 1. Transcription Whisper
+    try:
+        import whisper
+        model = whisper.load_model('small')
+        result = model.transcribe(file_path, language=None, fp16=False)
+        transcript = result.get('text', '').strip()
+    except ImportError:
+        transcript = '[Whisper non installé — lance INSTALLER_MAM.bat]'
+    except Exception as e:
+        transcript = f'[Erreur Whisper: {str(e)[:100]}]'
+
+    # 2. Conversion FLAC IASA TC-04 (96kHz / 24-bit)
+    try:
+        base      = os.path.splitext(os.path.basename(file_path))[0]
+        flac_dir  = os.path.join(os.path.dirname(file_path), '..', 'flac')
+        os.makedirs(flac_dir, exist_ok=True)
+        flac_path = os.path.abspath(os.path.join(flac_dir, base + '_TC04_96k24b.flac'))
+
+        cmd = [
+            FFMPEG, '-y', '-i', file_path,
+            '-c:a', 'flac',
+            '-ar', '96000',
+            '-sample_fmt', 's32',
+            '-bits_per_raw_sample', '24',
+            '-compression_level', '8',
+            flac_path
+        ]
+        result = subprocess.run(cmd, capture_output=True, timeout=120)
+        flac_ok = result.returncode == 0 and os.path.exists(flac_path)
+    except Exception as e:
+        flac_ok = False
+        flac_path = ''
+
+    # 3. Tags + résumé via Ollama
+    name    = os.path.basename(file_path)
+    content = transcript[:400] if transcript else ''
+    tags    = generate_tags(name, 'audio', content)
+    summary = generate_summary(name, 'audio', content)
+
+    ok(
+        summary    = summary    or f'Fichier audio : {name}',
+        tags       = tags       or ['audio', 'archive', 'patrimoine', 'son', 'ACoNum'],
+        transcript = transcript,
+        flac_ok    = flac_ok,
+        flac_path  = flac_path,
+        standard   = 'IASA TC-04 — FLAC 96kHz/24-bit' if flac_ok else 'Conversion FLAC non effectuée',
+    )
+
+
+# ════════════════════════════════════════════════════════════
+#  VIDEO — FFmpeg + Whisper
+# ════════════════════════════════════════════════════════════
+def analyze_video(file_path):
+    transcript = ''
+    audio_tmp  = file_path + '_audio_tmp.wav'
+
+    # 1. Extraire l'audio en WAV
+    try:
+        cmd = [FFMPEG, '-y', '-i', file_path, '-vn',
+               '-ar', '16000', '-ac', '1', '-f', 'wav', audio_tmp]
+        subprocess.run(cmd, capture_output=True, timeout=120)
+    except Exception:
+        pass
+
+    # 2. Transcrire avec Whisper
+    audio_src = audio_tmp if os.path.exists(audio_tmp) else file_path
+    try:
+        import whisper
+        model  = whisper.load_model('small')
+        result = model.transcribe(audio_src, language=None, fp16=False)
+        transcript = result.get('text', '').strip()
+    except ImportError:
+        transcript = '[Whisper non installé]'
+    except Exception as e:
+        transcript = f'[Erreur Whisper: {str(e)[:100]}]'
+
+    # Nettoyage fichier temporaire
+    if os.path.exists(audio_tmp):
+        try: os.remove(audio_tmp)
+        except: pass
+
+    name    = os.path.basename(file_path)
+    content = transcript[:400]
+    tags    = generate_tags(name, 'video', content)
+    summary = generate_summary(name, 'video', content)
+
+    ok(
+        summary    = summary    or f'Fichier vidéo : {name}',
+        tags       = tags       or ['video', 'archive', 'media', 'ACoNum', 'deepfake'],
+        transcript = transcript,
+        videoshield_url = 'https://huggingface.co/spaces/NOBODY204/VideoShield',
+    )
+
+
+# ════════════════════════════════════════════════════════════
+#  IMAGE — LLaVA via Ollama (si disponible)
+# ════════════════════════════════════════════════════════════
+def analyze_image(file_path):
+    description = ''
+    name        = os.path.basename(file_path)
+
+    # Tenter LLaVA via Ollama (modèle vision)
+    try:
+        import urllib.request, base64
+        with open(file_path, 'rb') as f:
+            img_b64 = base64.b64encode(f.read()).decode('utf-8')
+
+        body = json.dumps({
+            'model':  'llava:7b',
+            'prompt': 'Décris cette image en français en 2-3 phrases pour un archiviste média.',
+            'images': [img_b64],
+            'stream': False,
+            'options': { 'temperature': 0, 'num_predict': 200 }
+        }).encode('utf-8')
+
+        req = urllib.request.Request(
+            OLLAMA_HOST + '/api/generate',
+            data=body,
+            headers={ 'Content-Type': 'application/json' },
+            method='POST'
+        )
+        with urllib.request.urlopen(req, timeout=60) as resp:
+            data = json.loads(resp.read().decode('utf-8'))
+            description = data.get('response', '').strip()
+    except Exception:
+        description = f'[LLaVA non disponible — analyse visuelle désactivée pour {name}]'
+
+    tags    = generate_tags(name, 'image', description)
+    summary = description or generate_summary(name, 'image', '')
+
+    ok(
+        summary         = summary or f'Fichier image : {name}',
+        tags            = tags    or ['image', 'visuel', 'archive', 'media', 'ACoNum'],
+        description     = description,
+        imageshield_url = 'https://huggingface.co/spaces/NOBODY204/ImageShield',
+    )
+
+
+# ════════════════════════════════════════════════════════════
+#  PDF — pdfplumber
+# ════════════════════════════════════════════════════════════
+def analyze_pdf(file_path):
+    text = ''
+    name = os.path.basename(file_path)
+
+    try:
+        import pdfplumber
+        with pdfplumber.open(file_path) as pdf:
+            pages = []
+            for i, page in enumerate(pdf.pages[:5]):  # 5 premières pages max
+                t = page.extract_text()
+                if t:
+                    pages.append(t.strip())
+            text = '\n'.join(pages)
+    except ImportError:
+        text = '[pdfplumber non installé — lance INSTALLER_MAM.bat]'
+    except Exception as e:
+        text = f'[Erreur lecture PDF: {str(e)[:100]}]'
+
+    content = text[:600]
+    tags    = generate_tags(name, 'pdf', content)
+    summary = generate_summary(name, 'pdf', content)
+
+    ok(
+        summary  = summary or f'Document PDF : {name}',
+        tags     = tags    or ['document', 'pdf', 'archive', 'texte', 'ACoNum'],
+        text     = text[:1000] if text else '',
+    )
+
+
+# ════════════════════════════════════════════════════════════
+#  MAIN
+# ════════════════════════════════════════════════════════════
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        fail('Usage: python analyze_media.py <audio|video|image|pdf> <chemin_fichier>')
+
+    media_type = sys.argv[1].lower().strip()
+    file_path  = sys.argv[2].strip()
+
+    if not os.path.exists(file_path):
+        fail(f'Fichier introuvable : {file_path}')
+
+    try:
+        if   media_type == 'audio': analyze_audio(file_path)
+        elif media_type == 'video': analyze_video(file_path)
+        elif media_type == 'image': analyze_image(file_path)
+        elif media_type in ('pdf', 'doc'): analyze_pdf(file_path)
+        else: fail(f'Type non supporté : {media_type}. Utiliser audio/video/image/pdf')
+    except Exception as e:
+        fail(traceback.format_exc())
         
-        plt.tight_layout()
-        plot_path = "spectrum_v5.png"
-        plt.savefig(plot_path, dpi=120)
-        plt.close()
-        return plot_path
-
-    def run_aasist(self, path):
-        if not self.aasist_ok: return None, "AASIST Off"
-        try:
-            results = self.aasist(path)
-            for r in results:
-                if any(k in r['label'].lower() for k in ['spoof', 'fake', 'synthetic']):
-                    return float(r['score']), f"AASIST spoof={r['score']:.1%}"
-            return 1 - float(results[0]['score']), "AASIST Bonafide"
-        except: return None, "AASIST Error"
-
-    def analyser_expert(self, path):
-        if path is None: return None, "En attente...", None
-        try:
-            self.load_ast()
-            self.load_aasist()
-            y, sr = librosa.load(path, sr=44100)
-
-            # Analyse des caractéristiques
-            zcr = float(np.mean(librosa.feature.zero_crossing_rate(y)))
-            centroid = float(np.mean(librosa.feature.spectral_centroid(y=y, sr=sr)))
-            flatness = float(np.mean(librosa.feature.spectral_flatness(y=y)))
-            
-            # --- DÉTECTION VOIX OFF / SPEECH (Correction is.mp3) ---
-            is_speech = zcr < ZCR_SPEECH_THRESHOLD
-            is_vintage = centroid < CENTROID_VINTAGE and zcr < ZCR_VINTAGE
-            
-            # Débruitage adaptatif
-            prop = 0.4 if (is_speech or is_vintage) else 0.7
-            y_denoised = nr.reduce_noise(y=y, sr=sr, prop_decrease=prop)
-
-            def compute_jitter(sig):
-                pitches, mags = librosa.piptrack(y=sig, sr=sr, fmin=60, fmax=4000)
-                mask = mags > np.median(mags)
-                return float(np.std(pitches[mask]) / 1000) if np.any(mask) else 0.0
-
-            jitter_a = compute_jitter(y)
-            jitter_b = compute_jitter(y_denoised)
-            aasist_score, aasist_label = self.run_aasist(path)
-            acoustid_known, acoustid_info, _ = self.acoustid_lookup(path)
-
-            # --- LOGIQUE DE SCORING FUSIONNÉE ---
-            confiance = 50
-            raisons = []
-
-            # Correction Voix Off / Vintage
-            if is_speech:
-                confiance += 20
-                raisons.append("🗣️ Détection Voix Off : Seuil de tolérance augmenté")
-            if is_vintage:
-                confiance += 15
-                raisons.append("🕰️ Signature Vintage/Radio détectée")
-
-            # AASIST
-            if aasist_score and aasist_score > 0.75:
-                if not is_speech: # Moins punitif sur la voix off
-                    confiance -= 40
-                    raisons.append("🤖 AASIST détecte une synthèse neuronale")
-                else:
-                    confiance -= 15
-                    raisons.append("⚠️ AASIST suspect, mais contexte Voix Off")
-
-            # Texture
-            if flatness > FLATNESS_THRESHOLD:
-                confiance += 15
-                raisons.append(f"🌿 Texture organique (Flatness OK)")
-            else:
-                if not is_speech:
-                    confiance -= 20
-                    raisons.append("🔇 Signal trop lisse (Caractéristique IA)")
-
-            # Verdict final
-            confiance = max(0, min(100, confiance))
-            if confiance >= 70: verdict = "🔒 AUTHENTIQUE CERTIFIÉ"
-            elif confiance >= 40: verdict = "⚠️ ANALYSE INCONCLUSIVE"
-            else: verdict = "🛑 AI COVER / DEEPFAKE DÉTECTÉ"
-
-            report = f"📌 RAPPORT v5.0\n🎯 CONFIANCE : {confiance}%\n>>> VERDICT : {verdict}\n\nNOTES :\n" + "\n".join([f" • {r}" for r in raisons])
-
-            return (
-                self.generer_spectrogramme(y, sr, jitter_a, jitter_b, aasist_score),
-                report,
-                { "Score": f"{confiance}%", "Vintage": is_vintage, "VoixOff": is_speech, "Flatness": round(flatness, 6) }
-            )
-        except Exception as e:
-            return None, f"Erreur : {e}", None
-
-# ══════════════════════════════════════════════════════════════
-#  INTERFACE
-# ══════════════════════════════════════════════════════════════
-shield = AntigravityShield()
-with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 🛡️ Antigravity Shield v5.0 PRO")
-    with gr.Row():
-        with gr.Column():
-            audio_input = gr.Audio(type="filepath", label="Fichier Audio")
-            run_btn = gr.Button("⚙️ ANALYSER", variant="primary")
-        with gr.Column():
-            image_output = gr.Image(label="Spectrogramme")
-            report_output = gr.Textbox(label="Rapport d'expertise", lines=12)
-            metrics_output = gr.JSON(label="Métriques")
-
-    run_btn.click(fn=shield.analyser_expert, inputs=audio_input, outputs=[image_output, report_output, metrics_output])
-
-if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)